From 207be45bfc735742c379c46f65096f56fb5df04d Mon Sep 17 00:00:00 2001 From: Jared Wein Date: Thu, 6 Aug 2026 15:32:53 -0400 Subject: [PATCH 1/2] Make the action apply path safe to repeat and hard to poison MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two concurrent Pub/Sub deliveries of the same `run.completed` could both read a `pending` row and both post its comment to the bug. Rows are now locked before dispatch and stay locked until the result is committed, so a second delivery — or a manual apply-all racing the automatic one — blocks, then reads `applied` and skips. Holding the lock rather than flagging the row in-progress is what keeps this from needing recovery: a process that dies mid-dispatch rolls back to a row that is exactly as retryable as it was, with no in-flight state for anything to reclaim and no new column to migrate. The cost is a transaction open across the Bugzilla call, which is why the lock is per coalesced group rather than per pass. Along the way, three inputs that could 5xx the push route in a loop. With no dead-letter topic configured the same message returns for the whole retention window, so each now fails a row a human can read instead: a first-delivery race on the row insert (now `ON CONFLICT DO NOTHING`), a recorded action with no type or non-mapping params, and a params shape that raises inside the coalescer. An unresolved `{{actions..}}` also stops being sent. It was left in place deliberately so a human could see it, which is only safe if it never reaches Bugzilla — and it could, once a referenced row failed. --- services/hackbot-api/app/actions_applier.py | 235 ++++++++++--- .../hackbot-api/tests/test_actions_applier.py | 328 +++++++++++++++++- 2 files changed, 503 insertions(+), 60 deletions(-) diff --git a/services/hackbot-api/app/actions_applier.py b/services/hackbot-api/app/actions_applier.py index 49f2f020ce..4699676710 100644 --- a/services/hackbot-api/app/actions_applier.py +++ b/services/hackbot-api/app/actions_applier.py @@ -2,16 +2,21 @@ On run completion the recorded actions from `summary["actions"]` are always upserted as `run_actions` rows (one per entry) so they're visible and -manageable in the UI. Whether they're then applied *automatically* depends on -the agent's `auto_apply_actions` opt-in (see `app/agents.py`); either way they -can be applied on demand (manual apply-all from the UI). Application runs each -pending row through the handler registry in `hackbot_runtime.actions.handlers` -and is idempotent per action — an already-`applied` row is never re-applied, so -Pub/Sub retries and repeated manual applies are safe. +manageable in the UI. Whether they're then applied *automatically* depends on the +agent's `auto_apply_actions` opt-in (see `app/agents.py`); either way they can be +applied on demand (manual apply-all from the UI). Application runs each pending row +through the handler registry in `hackbot_runtime.actions.handlers`. + +Applying is safe to repeat: an already-`applied` row is never re-applied, and a row is +locked before its handler is called and stays locked until the result is committed. A +crash between Bugzilla accepting a write and that commit still rolls back to a retryable +row — Bugzilla offers no idempotency key, so the choice is between a possible duplicate +and a possible silent loss. """ from __future__ import annotations +import json import logging import re from datetime import datetime, timezone @@ -25,6 +30,7 @@ plan_coalesced_groups, ) from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession from app import gcs @@ -86,6 +92,31 @@ def _sub(match: re.Match) -> str: return value +class UnresolvedReference(Exception): + """A placeholder survived substitution, so the action must not be sent.""" + + +def _resolved_params(row: RunAction, results_by_ref: dict[str, dict]) -> Any: + """`row`'s params with placeholders substituted, or raise if any survived. + + `resolve_placeholders` leaves an unresolvable `{{actions..}}` in place + so a human can see what went wrong, which is only safe if it never reaches + Bugzilla — otherwise the literal text lands in a real bug comment with a log line + as the only signal. Raising makes it a failed row a human can retry instead. + """ + resolved = resolve_placeholders(row.params or {}, results_by_ref) + leftover = _PLACEHOLDER_RE.findall(json.dumps(resolved, default=str)) + if leftover: + raise UnresolvedReference( + "refers to " + + ", ".join( + sorted(f"{{{{actions.{ref}.{field}}}}}" for ref, field in leftover) + ) + + ", which has not been applied" + ) + return resolved + + async def ensure_action_rows( db: AsyncSession, run: Run ) -> list[tuple[RunAction, list[dict]]]: @@ -95,27 +126,63 @@ async def ensure_action_rows( summary.json. Idempotent: existing rows are reused, so this can run on every completion and again on each manual apply. """ - actions: list[dict] = (run.summary or {}).get("actions", []) + actions = (run.summary or {}).get("actions", []) + if not isinstance(actions, list) or not actions: + return [] + + # An unusable action is skipped rather than raised on: with no dead-letter topic, a + # raise here would 5xx the push route and the same message would return for the whole + # retention window. Indices are preserved so `ref` placeholders and the coalescing + # order stay meaningful. + usable = [ + (idx, action) + for idx, action in enumerate(actions) + if isinstance(action, dict) + and isinstance(action.get("type"), str) + # An explicit `null` is fine — normalised to `{}` below — but `params` is NOT + # NULL, so any other non-dict would raise on insert. + and isinstance(action.get("params") or {}, dict) + ] + if len(usable) != len(actions): + log.error( + "Run %s recorded %d unusable action(s) (no type, or params that aren't " + "a mapping); skipping them", + run.run_id, + len(actions) - len(usable), + ) + if not usable: + return [] + + # `ON CONFLICT DO NOTHING` rather than select-then-insert: two concurrent first + # deliveries can both find no rows and both insert the same `(run_id, idx)`, and the + # loser would 500 the route — turning the concurrency this path exists to tolerate + # into an error. + await db.execute( + insert(RunAction) + .values( + [ + { + "run_id": run.run_id, + "idx": idx, + "type": action["type"], + "params": action.get("params") or {}, + "ref": action.get("ref"), + "status": "pending", + } + for idx, action in usable + ] + ) + .on_conflict_do_nothing(constraint="uq_run_actions_run_idx") + ) + await db.flush() result = await db.execute(select(RunAction).where(RunAction.run_id == run.run_id)) - existing = {row.idx: row for row in result.scalars()} - - rows: list[tuple[RunAction, list[dict]]] = [] - for idx, action in enumerate(actions): - row = existing.get(idx) - if row is None: - row = RunAction( - run_id=run.run_id, - idx=idx, - type=action["type"], - params=action.get("params", {}), - ref=action.get("ref"), - status="pending", - ) - db.add(row) - rows.append((row, action.get("attachments", []))) - await db.flush() - return rows + by_idx = {row.idx: row for row in result.scalars()} + return [ + (by_idx[idx], action.get("attachments", [])) + for idx, action in usable + if idx in by_idx + ] async def _dispatch( @@ -148,6 +215,43 @@ async def _dispatch( return ActionResult.failed(str(exc)) +async def _lock_unapplied(db: AsyncSession, member_rows: list[RunAction]) -> bool: + """Lock `member_rows` for applying. True if this caller should go on to dispatch. + + The lock is held until the caller commits the result, so a second delivery blocks + here and then reads `applied` instead of posting the same comment again. Holding it + rather than marking the rows and letting go is what makes a crash self-healing: a + process that dies mid-dispatch rolls back, leaving the rows exactly as retryable as + they were, with no in-progress state for anything to reclaim. + + Locking rather than a conditional UPDATE, because a `WHERE ... AND (SELECT count(*) + ...) = n` reads the statement's snapshot: two claimants of a two-row group can each + lock a different member and then each skip the one the other took, leaving the group + unapplied by either. Ordered by id so two overlapping groups can't deadlock. + + `populate_existing` because the session runs with `expire_on_commit=False` and these + rows are already in its identity map, so the SELECT would otherwise hand back the + pre-lock cached copies — judging the predicate against exactly the stale state the + lock exists to rule out. + """ + ids = sorted(member.id for member in member_rows) + result = await db.execute( + select(RunAction) + .where(RunAction.id.in_(ids)) + .order_by(RunAction.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + locked = list(result.scalars()) + + if len(locked) == len(ids) and all(row.status != "applied" for row in locked): + return True + + # Nothing to do, so let the lock go rather than hold it for the rest of the pass. + await db.commit() + return False + + async def _apply_pending_rows( db: AsyncSession, run: Run, rows: list[tuple[RunAction, list[dict]]] ) -> None: @@ -175,13 +279,24 @@ async def _apply_pending_rows( # Drop any group whose rows carry a `ref`: nothing should reference a # coalesced member's result, and this keeps that invariant if a ref is ever # added to a bug action. Everything else applies one row at a time as before. - groups = [ - group - for group in plan_coalesced_groups( - [(row.type, row.params) for row, _ in pending] + # Guarded because it indexes into params the agent wrote, so a surprising shape + # raises here. Coalescing is only an optimisation (one bugmail instead of two), so + # failing to plan it degrades to applying singly — where each bad row fails alone. + try: + groups = [ + group + for group in plan_coalesced_groups( + # `or {}`: a pre-existing row could have null params. + [(row.type, row.params or {}) for row, _ in pending] + ) + if all(pending[i][0].ref is None for i in group) + ] + except Exception: + log.exception( + "Could not plan coalescing for run %s; applying its actions singly", + run.run_id, ) - if all(pending[i][0].ref is None for i in group) - ] + groups = [] # Rows sit in idx order, so a group's last member is its max idx: apply the # whole group there, once every earlier (backward) dependency is resolved. anchor_of = {i: max(group) for group in groups for i in group} @@ -192,19 +307,47 @@ async def _apply_pending_rows( if anchor is not None and pos != anchor: continue # non-anchor member: applied together with its anchor - if anchor is not None: - member_rows = [pending[i][0] for i in group_at[anchor]] - entries = [ - (member.type, resolve_placeholders(member.params, results_by_ref)) - for member in member_rows - ] - outcome = await _dispatch( - run, "bugzilla.update_bug", merge_resolved(entries), [] + member_rows = ( + [pending[i][0] for i in group_at[anchor]] if anchor is not None else [row] + ) + + # Lock before dispatching, or two concurrent deliveries both see `pending` and + # both post the comment to the bug. Also covers a manual apply-all racing the + # automatic one. + if not await _lock_unapplied(db, member_rows): + log.info( + "Rows %s of run %s were already applied; skipping", + [member.idx for member in member_rows], + run.run_id, + ) + continue + + # `merge_resolved` also indexes into agent-written params, so it can raise + # outside `_dispatch`'s guard. A failed row a human can read beats 5xxing the + # push route, which with no dead-letter topic replays for the retention window. + try: + if anchor is not None: + entries = [ + (member.type, _resolved_params(member, results_by_ref)) + for member in member_rows + ] + outcome = await _dispatch( + run, "bugzilla.update_bug", merge_resolved(entries), [] + ) + else: + outcome = await _dispatch( + run, + row.type, + _resolved_params(row, results_by_ref), + attachments, + ) + except Exception as exc: + log.exception( + "Could not build the request for rows %s of run %s", + [member.idx for member in member_rows], + run.run_id, ) - else: - member_rows = [row] - params = resolve_placeholders(row.params, results_by_ref) - outcome = await _dispatch(run, row.type, params, attachments) + outcome = ActionResult.failed(str(exc)) # Only stamp applied_at on a real success, so a failed row isn't # mistaken for one that was applied. @@ -226,9 +369,9 @@ async def _apply_pending_rows( async def on_run_completed(db: AsyncSession, run: Run) -> None: """Record a completed run's actions, and auto-apply them if the agent opts in. - Called from the `apply-run-actions` push route. Actions are always recorded - (so the UI can show/manually apply them); they're applied automatically only - when the run's agent has `auto_apply_actions=True`. + Called from the `apply-run-actions` push route. Actions are always recorded (so the + UI can show/manually apply them); they're applied automatically only when the run's + agent has `auto_apply_actions=True`. """ # Defense-in-depth: only a succeeded run's actions are recorded/applied. A # failed/timed-out run may have recorded actions before erroring, but acting diff --git a/services/hackbot-api/tests/test_actions_applier.py b/services/hackbot-api/tests/test_actions_applier.py index 344c41fd59..5b39bd6c18 100644 --- a/services/hackbot-api/tests/test_actions_applier.py +++ b/services/hackbot-api/tests/test_actions_applier.py @@ -75,19 +75,92 @@ class _FakeRun: agent: str = "bug-fix" run_id: uuid.UUID = field(default_factory=uuid.uuid4) summary: dict | None = None + inputs: dict = field(default_factory=dict) + + +class _FakeResult: + def __init__(self, rows): + self._rows = rows + + def scalars(self): + return iter(self._rows) class _FakeDB: - def __init__(self): + """Enough of `AsyncSession` for the apply path. + + Keeps the row's *database* state separately from the caller's in-memory copy. That + gap is deliberate: the rows are in the real session's identity map, so a lock that + forgets `populate_existing` would judge its own pre-lock copy and re-dispatch an + action another delivery already applied. A fake that returned the caller's own + objects could never catch that — and didn't, until it did. + + `rows=None` means no statement should be reached, which is what the + `on_run_completed` tests want: they stub the apply pass out entirely. + """ + + def __init__(self, rows=None): self.commits = 0 + self.claims = [] + self._identity = {row.id: row for row in rows or []} + self._db_state = {row.id: {"status": row.status} for row in rows or []} + # What each row looked like when we last handed it over, so `commit` can write + # back only what this "transaction" actually changed — a real session flushes + # dirty attributes, not every loaded row. + self._handed_over = dict(self._db_state) + + def set_db_status(self, row_id, status): + """Let a test say "another delivery got here first" without touching `row`.""" + self._db_state[row_id] = {"status": status} async def commit(self): self.commits += 1 - - async def execute(self, *a, **k): - raise AssertionError( - "ensure_action_rows should be monkeypatched in these tests" - ) + for row_id, row in self._identity.items(): + if row_id not in self._db_state: + continue + was = self._handed_over.get(row_id, {}) + current = {"status": row.status} + # Only fields this session changed since it last read the row. + changed = {k: v for k, v in current.items() if was.get(k) != v} + if changed: + self._db_state[row_id] = {**self._db_state[row_id], **changed} + self._handed_over[row_id] = current + + async def execute(self, statement): + compiled = statement.compile() + sql = str(compiled) + locking = "FOR UPDATE" in sql + if locking and not statement.get_execution_options().get("populate_existing"): + raise AssertionError( + "the lock must re-read under it (populate_existing), or it judges the " + "caller's stale copy" + ) + if not locking and not statement.get_execution_options().get( + "populate_existing" + ): + raise AssertionError(f"unexpected statement: {sql}") + + wanted = set() + for key, value in compiled.params.items(): + if key.startswith("id_"): + wanted.update(value if isinstance(value, (list, tuple)) else [value]) + # A re-read selects by run_id, not by row id: hand back everything. + if not wanted: + wanted = set(self._db_state) + elif locking: + self.claims.append(sorted(wanted)) + + # Refreshing the identity-mapped object from the row is what + # `populate_existing` does; the caller then judges current state. + rows = [] + for row_id in sorted(wanted): + if row_id not in self._db_state: + continue + row = self._identity[row_id] + row.status = self._db_state[row_id]["status"] + self._handed_over[row_id] = dict(self._db_state[row_id]) + rows.append(row) + return _FakeResult(rows) def _patch_applier(monkeypatch, *, auto: bool | None): @@ -96,13 +169,16 @@ def _patch_applier(monkeypatch, *, auto: bool | None): `auto=None` means the agent isn't in the registry at all. """ calls = {"ensured": False, "applied": False} + rows = [(_row(0, "pending"), [])] async def fake_ensure(db, run): calls["ensured"] = True - return [("row", [])] + return rows async def fake_apply(db, run, rows): calls["applied"] = True + for row, _ in rows: + row.status = "applied" monkeypatch.setattr(actions_applier, "ensure_action_rows", fake_ensure) monkeypatch.setattr(actions_applier, "_apply_pending_rows", fake_apply) @@ -174,6 +250,7 @@ def _row( applied_at=None, ): return SimpleNamespace( + id=idx + 1, idx=idx, type=action_type, params=params if params is not None else {}, @@ -199,7 +276,9 @@ async def test_apply_pending_rows_retries_failed_and_skips_applied(monkeypatch): rows = [(applied, []), (failed, []), (pending, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # The already-applied row is untouched; its handler never runs. @@ -210,6 +289,215 @@ async def test_apply_pending_rows_retries_failed_and_skips_applied(monkeypatch): assert pending.status == "applied" +async def test_a_row_claimed_elsewhere_is_not_dispatched_again(monkeypatch): + # The failure this claim exists for: Pub/Sub push is at-least-once *and* + # concurrent, so two deliveries can both read the same `pending` row. Without a + # claim both would call the handler and the bug would get two comments. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result={"ok": 1}, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + row = _row(0, "pending") + rows = [(row, [])] + db = _FakeDB([row]) + # Another delivery got there first and committed. Set on the *database* state only: + # the caller's `row` still reads `pending`, which is exactly what the lock has to + # see through — it blocks until that delivery commits, then reads what it wrote. + db.set_db_status(row.id, "applied") + assert row.status == "pending" + + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] + + +async def test_the_lock_is_held_across_the_handler(monkeypatch): + # Ordering is the whole point: the row must still be locked while Bugzilla is being + # written, since that is the window a second delivery would otherwise write in. The + # result is only committed afterwards. + seen = {} + row = _row(0, "pending") + + class _Handler: + async def apply(self, params, ctx): + seen["commits_at_dispatch"] = db.commits + seen["locked"] = db.claims == [[row.id]] + return SimpleNamespace(status="applied", result=None, error=None) + + monkeypatch.setattr(actions_applier, "get_handler", lambda t: _Handler()) + db = _FakeDB([row]) + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), [(row, [])] + ) + + assert seen["locked"], "dispatched without locking the row" + assert seen["commits_at_dispatch"] == 0, "released the lock before dispatching" + assert row.status == "applied" + assert db.commits == 1 + + +async def test_an_unresolved_reference_is_never_posted(monkeypatch): + # The row it referenced failed, so the placeholder can't be substituted. Sending anyway puts a literal `{{actions.patch.url}}` in a real bug + # comment with only a log line as the signal; it must fail visibly instead. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result={"url": "http://x/D1"}, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + patch_row = _row( + 0, "pending", action_type="phabricator.submit_patch", ref="patch", params={} + ) + comment = _row( + 1, + "pending", + params={"bug_id": 5, "text": "Fix: {{actions.patch.url}}"}, + ) + rows = [(patch_row, []), (comment, [])] + db = _FakeDB([patch_row, comment]) + # The referenced row was applied elsewhere, so its result never lands here. + db.set_db_status(patch_row.id, "applied") + + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] # nothing was sent + assert comment.status == "failed" + assert "{{actions.patch.url}}" in comment.error + + +async def test_a_resolvable_reference_is_still_substituted(monkeypatch): + handler = _RecordingHandler( + SimpleNamespace(status="applied", result={"url": "http://x/D1"}, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + patch_row = _row( + 0, "pending", action_type="phabricator.submit_patch", ref="patch", params={} + ) + comment = _row( + 1, "pending", params={"bug_id": 5, "text": "Fix: {{actions.patch.url}}"} + ) + rows = [(patch_row, []), (comment, [])] + + await actions_applier._apply_pending_rows( + _FakeDB([patch_row, comment]), + _FakeRun(status=RunStatus.succeeded.value), + rows, + ) + + assert handler.calls[-1] == {"bug_id": 5, "text": "Fix: http://x/D1"} + assert comment.status == "applied" + + +async def test_an_unhashable_bug_id_does_not_fail_the_whole_route(monkeypatch): + # `plan_coalesced_groups` buckets by `bug_id`, so a list raises while *planning* — + # before any per-action guard. Uncaught that 5xxs the push route, and with no + # dead-letter topic the same message returns for the whole retention window. It + # must degrade to applying rows singly instead. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + bad = _row(0, "pending", params={"bug_id": [5], "text": "hi"}) + good = _row( + 1, + "pending", + action_type="bugzilla.update_bug", + params={"bug_id": 5, "changes": {"a": 1}}, + ) + rows = [(bad, []), (good, [])] + + await actions_applier._apply_pending_rows( + _FakeDB([bad, good]), _FakeRun(status=RunStatus.succeeded.value), rows + ) + + # Both were dispatched individually rather than coalesced, and nothing raised. + assert len(handler.calls) == 2 + + +async def test_a_row_that_cannot_be_merged_fails_by_itself(monkeypatch): + # `merge_resolved` folds a same-bug comment into the update's body, so a non-dict + # `changes` raises there — outside `_dispatch`'s guard, as an argument to it. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + update = _row( + 0, + "pending", + action_type="bugzilla.update_bug", + params={"bug_id": 5, "changes": "oops"}, + ) + comment = _row(1, "pending", params={"bug_id": 5, "text": "hi"}) + rows = [(update, []), (comment, [])] + + await actions_applier._apply_pending_rows( + _FakeDB([update, comment]), _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] # nothing was sent + assert update.status == "failed" and update.error + assert comment.status == "failed" # coalesced with it, so it shares the outcome + + +async def test_a_non_list_actions_summary_records_nothing(monkeypatch): + # `enumerate(5)` would raise straight out of the route. + run = _FakeRun(status=RunStatus.succeeded.value, summary={"actions": 5}) + assert await actions_applier.ensure_action_rows(_FakeDB(), run) == [] + + +async def test_a_row_an_interrupted_apply_left_behind_is_retryable(monkeypatch): + # Holding the lock rather than marking the row is what makes a crash self-healing: + # the transaction rolls back, so the row is still `pending` and nothing has to + # notice it was ever in flight. A `failed` row is retried the same way. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + for status in ("pending", "failed"): + row = _row(0, status) + await actions_applier._apply_pending_rows( + _FakeDB([row]), _FakeRun(status=RunStatus.succeeded.value), [(row, [])] + ) + assert row.status == "applied", status + + +async def test_a_coalesced_group_is_locked_as_one(monkeypatch): + # The group becomes a single Bugzilla PUT, so acting on part of it would mean a + # partial post. Either the whole group is ours or none of it is. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + update = _row( + 0, + "pending", + action_type="bugzilla.update_bug", + params={"bug_id": 5, "changes": {"a": 1}}, + ) + comment = _row(1, "pending", params={"bug_id": 5, "text": "hi"}) + rows = [(update, []), (comment, [])] + db = _FakeDB([update, comment]) + # One member of the group was already applied elsewhere. + db.set_db_status(comment.id, "applied") + + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] + # The member we could have claimed is left alone rather than half-applied. + assert update.status == "pending" + + # --- coalescing same-bug Bugzilla mutations into one PUT ---------------- # @@ -240,7 +528,9 @@ async def test_coalesces_update_and_comment_into_one_put(monkeypatch): rows = [(update, []), (other, []), (comment, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # Two calls: the standalone comment on bug 99, then ONE combined PUT for @@ -284,7 +574,9 @@ async def test_extra_comments_applied_separately(monkeypatch): rows = [(update, []), (near, []), (far, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # Field change rides with the closest comment ("near"); "far" is its own PUT. @@ -319,7 +611,9 @@ async def test_lone_same_type_actions_on_different_bugs_not_merged(monkeypatch): rows = [(u5, []), (u6, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # Different bugs, one update each -> no coalescing, two raw PUTs. assert handler.calls == [ @@ -357,7 +651,9 @@ async def test_coalesced_group_failure_marks_all_then_retries(monkeypatch): rows = [(update, []), (comment, []), (done, [])] run = _FakeRun(status=RunStatus.succeeded.value) - await actions_applier._apply_pending_rows(_FakeDB(), run, rows) + await actions_applier._apply_pending_rows( + _FakeDB([row for row, _ in rows]), run, rows + ) # One combined call; both members failed; the already-applied row untouched. assert len(failing.calls) == 1 assert update.status == "failed" and comment.status == "failed" @@ -368,7 +664,9 @@ async def test_coalesced_group_failure_marks_all_then_retries(monkeypatch): SimpleNamespace(status="applied", result={"bug_id": 5}, error=None) ) monkeypatch.setattr(actions_applier, "get_handler", lambda t: ok) - await actions_applier._apply_pending_rows(_FakeDB(), run, rows) + await actions_applier._apply_pending_rows( + _FakeDB([row for row, _ in rows]), run, rows + ) assert len(ok.calls) == 1 assert update.status == "applied" and comment.status == "applied" @@ -397,7 +695,9 @@ async def test_backward_placeholder_resolves_in_coalesced_comment(monkeypatch): rows = [(patch, []), (update, []), (comment, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # The patch applies first (its own idx), seeding results_by_ref; the From c0b9e03023bdc35f4b3cfb3e4892c87bec146422 Mon Sep 17 00:00:00 2001 From: Jared Wein Date: Thu, 6 Aug 2026 15:33:32 -0400 Subject: [PATCH 2/2] Ack undecodable apply-run-actions messages instead of retrying forever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A malformed envelope, a body that isn't JSON, or a `run_id` that isn't a UUID raised out of the route. Pub/Sub reads the resulting 5xx as a nack, and with no dead-letter topic configured on this subscription the same message comes back for the whole retention window — none of these can ever become valid. Log and ack instead. Only retryable failures should reach the client as an error. --- services/hackbot-api/app/routers/events.py | 18 +++++-- services/hackbot-api/tests/test_events.py | 58 ++++++++++++++++++++++ 2 files changed, 71 insertions(+), 5 deletions(-) diff --git a/services/hackbot-api/app/routers/events.py b/services/hackbot-api/app/routers/events.py index d46da7ab6d..bca46d63fb 100644 --- a/services/hackbot-api/app/routers/events.py +++ b/services/hackbot-api/app/routers/events.py @@ -122,14 +122,22 @@ async def apply_run_actions( webhooks) — each its own route named after its own job. The subscription feeding this one is filtered to succeeded runs (see deploy-events.sh). """ - body = await request.json() - event = _decode_pubsub_push_body(body) - run_id = event.get("run_id") - if not run_id: + # Input that can never become valid is acked, not retried: with no dead-letter topic + # configured, a 5xx here would replay the same unparsable message for the whole + # retention window. Only *retryable* failures should reach the client as an error. + try: + event = _decode_pubsub_push_body(await request.json()) + run_id = event.get("run_id") + run_uuid = uuid.UUID(run_id) if run_id else None + except Exception: + log.exception("Discarding an undecodable apply-run-actions message") + return + + if run_uuid is None: log.warning("apply-run-actions event missing run_id: %s", event) return - run = await db.get(Run, uuid.UUID(run_id)) + run = await db.get(Run, run_uuid) if run is None: log.warning("No run found for run_id %s", run_id) return diff --git a/services/hackbot-api/tests/test_events.py b/services/hackbot-api/tests/test_events.py index 214ca8e912..91dc30f2e2 100644 --- a/services/hackbot-api/tests/test_events.py +++ b/services/hackbot-api/tests/test_events.py @@ -6,11 +6,18 @@ import base64 import json +import uuid +from dataclasses import dataclass, field +from types import SimpleNamespace +from unittest.mock import AsyncMock +import pytest +from app.routers import events from app.routers.events import ( _decode_pubsub_push_body, _execution_name_from_completion_log, ) +from app.schemas import RunStatus def _push_envelope(payload: dict) -> dict: @@ -73,3 +80,54 @@ def test_execution_name_falls_back_to_labels(): def test_execution_name_missing(): assert _execution_name_from_completion_log({"protoPayload": {}}) is None assert _execution_name_from_completion_log({}) is None + + +# --- apply-run-actions: undecodable input is acked, not retried ----------- # + + +@dataclass +class _FakeRun: + run_id: uuid.UUID = field(default_factory=lambda: uuid.UUID(int=7)) + agent: str = "frontend-triage" + status: str = RunStatus.succeeded.value + summary: dict | None = None + inputs: dict = field(default_factory=dict) + + +class _FakeDB: + def __init__(self, run): + self._run = run + + async def get(self, model, run_id): + return self._run + + async def commit(self): + pass + + +def _patch_route(monkeypatch): + """Stub the applier; record that it was called.""" + calls = {"order": []} + + async def fake_on_run_completed(db, run): + calls["order"].append("apply") + + monkeypatch.setattr(events, "on_run_completed", fake_on_run_completed) + return calls + + +@pytest.mark.parametrize( + "body", + [ + {}, # no envelope + {"message": {"data": "bm90LWpzb24="}}, # decodes to "not-json" + {"message": {"data": base64.b64encode(b'{"run_id": "nope"}').decode()}}, + ], + ids=["no-envelope", "not-json", "bad-uuid"], +) +async def test_an_undecodable_message_is_acked_not_retried(monkeypatch, body): + # A message that can never become valid must not 5xx: Pub/Sub would nack, and + # with no dead-letter topic it would come back for the whole retention window. + _patch_route(monkeypatch) + request = SimpleNamespace(json=AsyncMock(return_value=body)) + await events.apply_run_actions(request, db=_FakeDB(_FakeRun()))