From 4e0b1bbdbf511d8f46d5ff54c229c476cb174cb2 Mon Sep 17 00:00:00 2001 From: drew Date: Tue, 19 May 2026 18:43:37 -0400 Subject: [PATCH] =?UTF-8?q?fix(controller):=20second=20review=20pass=20?= =?UTF-8?q?=E2=80=94=20close=20remaining=20MERGING=20stranding=20paths?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first post-commit fix only handled merge-error-* strings and missed that build_train's single-PR failure reasons flow straight into terminal_state via _release. Four were unmapped (no-head-ref, rev-parse-fetched-head-failed, rebase-timeout, push-failed-force-lease-mismatch) — same stranding bug class. - _resolve_bridge_event no longer returns None. Any unrecognised terminal_state falls back to merge_retry_exhausted (-> STUCK): operator-visible and recoverable beats silently orphaned in MERGING. The three static build_train reasons are mapped explicitly; the catch-all logs loudly so an unanticipated state is still visible. - merge-error-* routing refined: 404 -> merge_external_action (reconcile via PR state), 422 -> merge_base_conflict (CONFLICT_- RESOLVING re-rebases) instead of blanket STUCK. - Hard-crash reaper: merge_drive is a singleton and run_one_cycle is serial, so any workflow in MERGING at cycle start is the residue of a crashed prior cycle. run_one_cycle now reclaims them via merge_interrupted (MERGING -> APPROVED) before picking candidates. Adds bridge.list_merging_workflows. - run_one_cycle warns if a claimed PR is absent from pr_terminal_states instead of silently using the aggregate. 3309 tests pass. Co-Authored-By: Claude Opus 4.7 --- .../controller/test_controller_db_bridge.py | 30 ++++ tests/auto_agents/test_merge_drive.py | 100 +++++++++++-- tools/_controller_db_bridge.py | 42 ++++-- tools/merge_drive.py | 136 +++++++++++++++--- 4 files changed, 271 insertions(+), 37 deletions(-) diff --git a/tests/auto_agents/controller/test_controller_db_bridge.py b/tests/auto_agents/controller/test_controller_db_bridge.py index b985c5e9e..343c4f261 100644 --- a/tests/auto_agents/controller/test_controller_db_bridge.py +++ b/tests/auto_agents/controller/test_controller_db_bridge.py @@ -164,6 +164,36 @@ class TestListApprovedWorkflows: assert [w.workflow_id for w in only_alice] == [1] +class TestListMergingWorkflows: + """``list_merging_workflows`` backs the merge_drive hard-crash + reaper: any workflow stranded in MERGING (a prior cycle died + between the APPROVED→MERGING claim and the outcome emit) must be + discoverable so merge_drive can reclaim it.""" + + def test_empty_when_none_merging(self, engine, controller_db_path): + _seed_workflow( + controller_db_path, workflow_id=1, entity_number=10, + current_state="APPROVED", + ) + assert bridge.list_merging_workflows(engine, "owner", "repo") == [] + + def test_lists_only_merging_workflows(self, engine, controller_db_path): + _seed_workflow( + controller_db_path, workflow_id=1, entity_number=10, + current_state="MERGING", + ) + _seed_workflow( + controller_db_path, workflow_id=2, entity_number=11, + current_state="APPROVED", # should NOT be picked + ) + _seed_workflow( + controller_db_path, workflow_id=3, entity_number=12, + current_state="MERGING", + ) + out = bridge.list_merging_workflows(engine, "owner", "repo") + assert sorted(w.workflow_id for w in out) == [1, 3] + + # ─── state transitions ────────────────────────────────────────────── diff --git a/tests/auto_agents/test_merge_drive.py b/tests/auto_agents/test_merge_drive.py index a61a127b3..17e5aaa9b 100644 --- a/tests/auto_agents/test_merge_drive.py +++ b/tests/auto_agents/test_merge_drive.py @@ -1505,6 +1505,49 @@ class TestControllerBridgeIntegration: "merge_start", "merge_interrupted", ] + def _seed_in_state(self, db_path, *, workflow_id, pr_number, state): + from tools.controller.db import Workflow, build_engine, session_scope + eng = build_engine(f"sqlite:///{db_path}") + with session_scope(eng) as s: + s.add(Workflow( + workflow_id=workflow_id, kind="pr", + owner="owner", repo="repo", entity_number=pr_number, + current_state=state, current_tier=1, + )) + eng.dispose() + + def test_run_one_cycle_reclaims_stale_merging_workflow( + self, mod, monkeypatch, controller_db, tmp_path, + ): + """A workflow a crashed prior cycle stranded in MERGING is + reclaimed at the START of run_one_cycle: merge_interrupted fires + (MERGING→APPROVED) so it is no longer orphaned. The controller + has no MERGING handler — without this reclaim the workflow would + sit in MERGING forever (the hard-crash stranding hole).""" + # Seed directly into MERGING — simulates a SIGKILL between the + # APPROVED→MERGING claim and the outcome-event emit. + self._seed_in_state( + controller_db, workflow_id=201, pr_number=77, state="MERGING", + ) + self._stub_eligibility(mod, monkeypatch) + self._stub_forgejo_label_io(mod, monkeypatch) + # No open PRs → no candidates → the cycle does nothing but the + # reclaim, which isolates it. merge_train must not run. + monkeypatch.setattr(mod, "list_open_prs", lambda cfg: []) + monkeypatch.setattr( + mod, "merge_train", + lambda *a, **k: pytest.fail( + "merge_train must not run — no candidates this cycle" + ), + ) + cfg = self._cfg_with_bridge(mod) + cfg.work_dir = tmp_path + mod.run_one_cycle(cfg) + + # Reclaimed: MERGING → APPROVED via merge_interrupted. + assert self._read_state(controller_db, 201) == "APPROVED" + assert "merge_interrupted" in self._read_event_types(controller_db, 201) + # ── T5-7 follow-up: bridge-event resolution + per-PR terminal states ───────── @@ -1537,23 +1580,56 @@ def test_resolve_bridge_event_merge_error_403_is_branch_protection(mod): ) -def test_resolve_bridge_event_merge_error_other_is_retry_exhausted(mod): - """Any non-403 merge-error (5xx, 422, 404) routes to retry-exhausted +def test_resolve_bridge_event_merge_error_404_reconciles(mod): + """404 = PR/branch gone → merge_external_action, which reconciles + via PR state (the controller's designed 404 path).""" + assert ( + mod._resolve_bridge_event("merge-error-404") == "merge_external_action" + ) + + +def test_resolve_bridge_event_merge_error_422_routes_to_conflict(mod): + """422 = PR not mergeable (master moved post-CI) → merge_base_conflict + so CONFLICT_RESOLVING can re-rebase and retry, instead of STUCK.""" + assert ( + mod._resolve_bridge_event("merge-error-422") == "merge_base_conflict" + ) + + +def test_resolve_bridge_event_merge_error_5xx_is_retry_exhausted(mod): + """Any other merge-error status (5xx, ...) routes to retry-exhausted → STUCK, so the workflow is operator-visible, never stranded.""" - assert mod._resolve_bridge_event("merge-error-500") == "merge_retry_exhausted" - assert mod._resolve_bridge_event("merge-error-422") == "merge_retry_exhausted" + assert ( + mod._resolve_bridge_event("merge-error-500") == "merge_retry_exhausted" + ) -def test_resolve_bridge_event_bisected_is_unmapped(mod): - """'bisected' is an aggregate, never a per-PR fate — it must NOT - resolve to an event. Per-PR emission uses pr_terminal_states.""" - assert mod._resolve_bridge_event("bisected") is None +def test_resolve_bridge_event_build_train_reasons_are_mapped(mod): + """build_train single-PR failure reasons flow straight into + terminal_state via _release — every one MUST resolve to an event + (regression: only merge-error-* was handled, these stranded).""" + assert mod._resolve_bridge_event("no-head-ref") == "merge_retry_exhausted" + assert mod._resolve_bridge_event("rebase-timeout") == "merge_retry_exhausted" + assert ( + mod._resolve_bridge_event("push-failed-force-lease-mismatch") + == "merge_retry_exhausted" + ) -def test_resolve_bridge_event_unknown_is_none(mod): - """An unrecognised terminal_state resolves to None; the caller logs - a warning rather than emitting a bogus event.""" - assert mod._resolve_bridge_event("totally-made-up") is None +def test_resolve_bridge_event_never_returns_none(mod): + """The catch-all: ANY unrecognised terminal_state — a dynamic + build_train reason like 'rev-parse-fetched-head-failed: ...', a + future merge_train state, a typo — resolves to merge_retry_exhausted + (STUCK). The controller has no MERGING handler, so 'land in STUCK' + is the only acceptable default; None would strand the workflow.""" + assert ( + mod._resolve_bridge_event("rev-parse-fetched-head-failed: fatal: x") + == "merge_retry_exhausted" + ) + assert mod._resolve_bridge_event("totally-made-up") == "merge_retry_exhausted" + # 'bisected' should never reach here (per-PR emission uses + # pr_terminal_states) but if it did, it must not strand either. + assert mod._resolve_bridge_event("bisected") == "merge_retry_exhausted" def test_release_populates_per_pr_terminal_states(mod, monkeypatch, tmp_path): diff --git a/tools/_controller_db_bridge.py b/tools/_controller_db_bridge.py index 794aa4cf1..7a4646791 100644 --- a/tools/_controller_db_bridge.py +++ b/tools/_controller_db_bridge.py @@ -93,14 +93,11 @@ class ApprovedWorkflow: last_transition_at: str -def list_approved_workflows( - engine: Engine, owner: str, repo: str, +def _list_workflows_in_state( + engine: Engine, owner: str, repo: str, state: str, ) -> list[ApprovedWorkflow]: - """List workflows in ``APPROVED`` state for the given Forgejo repo. - - Only ``kind='pr'`` workflows are returned — issue workflows that - have spawned a PR transition through CREATED_PR, not APPROVED. - """ + """List ``kind='pr'`` workflows in ``state`` for the given repo, + FIFO-ordered by ``entered_state_at``.""" Maker = _session_maker(engine) with Maker() as s: # type: Session rows = s.execute( @@ -111,10 +108,10 @@ def list_approved_workflows( " WHERE owner = :owner " " AND repo = :repo " " AND kind = 'pr' " - " AND current_state = 'APPROVED' " + " AND current_state = :state " " ORDER BY entered_state_at ASC" # FIFO ), - {"owner": owner, "repo": repo}, + {"owner": owner, "repo": repo, "state": state}, ).all() return [ ApprovedWorkflow( @@ -130,6 +127,32 @@ def list_approved_workflows( ] +def list_approved_workflows( + engine: Engine, owner: str, repo: str, +) -> list[ApprovedWorkflow]: + """List workflows in ``APPROVED`` state for the given Forgejo repo. + + Only ``kind='pr'`` workflows are returned — issue workflows that + have spawned a PR transition through CREATED_PR, not APPROVED. + """ + return _list_workflows_in_state(engine, owner, repo, "APPROVED") + + +def list_merging_workflows( + engine: Engine, owner: str, repo: str, +) -> list[ApprovedWorkflow]: + """List workflows stuck in ``MERGING`` for the given Forgejo repo. + + merge_drive is a singleton and ``run_one_cycle`` is serial, so at + the start of a fresh cycle there is no in-flight merge — every + MERGING row is the residue of a previous cycle that died between + the APPROVED→MERGING claim and the outcome-event emit. The + controller has no MERGING handler to reap them, so merge_drive + reclaims them via :data:`EVENT_MERGE_INTERRUPTED` (MERGING→APPROVED). + """ + return _list_workflows_in_state(engine, owner, repo, "MERGING") + + # ─── state-machine transitions ───────────────────────────────────────── @@ -320,6 +343,7 @@ __all__ = [ "EVENT_MERGE_START", "build_optional_engine", "list_approved_workflows", + "list_merging_workflows", "transition_approved_to_merging", "transition_merge_outcome", ] diff --git a/tools/merge_drive.py b/tools/merge_drive.py index 83cfe7228..db03aba65 100755 --- a/tools/merge_drive.py +++ b/tools/merge_drive.py @@ -399,6 +399,14 @@ _TERMINAL_STATE_TO_BRIDGE_EVENT: dict[str, str] = { "bisect-budget-exhausted": "merge_retry_exhausted", "umbrella-pr-creation-failed": "merge_retry_exhausted", "branch-protection-blocked": "merge_branch_protection_blocked", + # build_train single-PR failure reasons (build_train.reason flows + # straight into terminal_state via _release). All genuine + # "operator look" cases → STUCK. rebase-conflict-vs-master is + # mapped above to merge_base_conflict so the conflict_resolver can + # re-rebase; the rest are not auto-recoverable. + "no-head-ref": "merge_retry_exhausted", + "rebase-timeout": "merge_retry_exhausted", + "push-failed-force-lease-mismatch": "merge_retry_exhausted", # Graceful shutdown caught the train mid-merge: return the # workflow to APPROVED so the next run re-picks it up cleanly. "stopped": "merge_interrupted", @@ -409,14 +417,23 @@ _TERMINAL_STATE_TO_BRIDGE_EVENT: dict[str, str] = { # per-PR from that map, never from the aggregate "bisected". -def _resolve_bridge_event(terminal_state: str) -> str | None: +def _resolve_bridge_event(terminal_state: str) -> str: """Map a merge_drive terminal_state to a controller event name. + NEVER returns None. The controller has no MERGING handler — a + workflow whose terminal_state maps to no event would be stranded + in MERGING forever. Any unrecognised state therefore falls back to + ``merge_retry_exhausted`` (→ STUCK): operator-visible and + recoverable beats silently orphaned. + Covers the static table plus the dynamic ``merge-error-{status}`` strings ``merge_train`` emits for any non-2xx/non-409 Forgejo merge - POST. 403 is branch protection (operator must act); every other - status is treated as retry-exhausted so the workflow lands in STUCK - rather than stranded in MERGING with no owner. + POST: + - 403 → branch protection (operator must act) + - 404 → PR gone; reconcile via PR state (merge_external_action) + - 422 → not mergeable; route to CONFLICT_RESOLVING so the + conflict_resolver can re-rebase and retry + - anything else → retry-exhausted → STUCK """ event = _TERMINAL_STATE_TO_BRIDGE_EVENT.get(terminal_state) if event is not None: @@ -425,8 +442,22 @@ def _resolve_bridge_event(terminal_state: str) -> str | None: suffix = terminal_state[len("merge-error-"):] if suffix == "403": return "merge_branch_protection_blocked" + if suffix == "404": + return "merge_external_action" + if suffix == "422": + return "merge_base_conflict" return "merge_retry_exhausted" - return None + # Catch-all: an unmapped terminal_state (an unanticipated + # build_train reason, a future merge_train state, a typo) must + # still resolve to an event so the workflow can never strand in + # MERGING. Land it in STUCK and log loudly. + logger.warning( + "controller-bridge: unrecognised terminal_state=%r — routing to " + "merge_retry_exhausted (STUCK); add an explicit mapping if this " + "is a known merge_drive outcome", + terminal_state, + ) + return "merge_retry_exhausted" def _emit_controller_event_for_outcome( @@ -443,14 +474,6 @@ def _emit_controller_event_for_outcome( if cfg.controller_db_engine is None: return event_name = _resolve_bridge_event(terminal_state) - if event_name is None: - logger.warning( - "controller-bridge: no event mapping for terminal_state=%r " - "(PR#%d wf=%d); workflow may be stranded in MERGING — this " - "is a bug, every terminal_state must map to an event", - terminal_state, pr_number, workflow_id, - ) - return try: import sys as _sys bridge_mod = _sys.modules["_controller_db_bridge"] @@ -472,6 +495,65 @@ def _emit_controller_event_for_outcome( ) +def _reclaim_stale_merging(cfg: DriverConfig) -> int: + """Return any workflow stranded in MERGING to APPROVED. + + merge_drive holds a SingleInstanceLock and ``run_one_cycle`` runs + serially, so at the start of a fresh cycle there is no in-flight + merge — ANY workflow still in MERGING is the residue of a previous + cycle that died (e.g. SIGKILL between the APPROVED→MERGING claim and + the outcome-event emit). The controller has no MERGING handler to + reap it, so without this the workflow is orphaned forever. Firing + merge_interrupted (MERGING→APPROVED) returns it to the handoff + state; it is then re-picked-up cleanly later in this same cycle. + + Best-effort: logs but never raises. Returns the reclaim count.""" + if cfg.controller_db_engine is None: + return 0 + if not cfg.controller_owner or not cfg.controller_repo: + return 0 + import sys as _sys + bridge_mod = _sys.modules.get("_controller_db_bridge") + if bridge_mod is None: + return 0 + try: + stale = bridge_mod.list_merging_workflows( + cfg.controller_db_engine, + cfg.controller_owner, + cfg.controller_repo, + ) + except Exception as e: # noqa: BLE001 — DB blip must not break the cycle + logger.warning( + "controller-bridge: list_merging_workflows failed: %s: %s", + type(e).__name__, e, + ) + return 0 + reclaimed = 0 + for wf in stale: + try: + ok = bridge_mod.transition_merge_outcome( + cfg.controller_db_engine, + wf.workflow_id, + bridge_mod.EVENT_MERGE_INTERRUPTED, + extra_payload={"reclaimed": "stale-merging-at-cycle-start"}, + ) + except Exception as e: # noqa: BLE001 + logger.warning( + "controller-bridge: reclaim of wf=%d failed: %s: %s", + wf.workflow_id, type(e).__name__, e, + ) + continue + if ok: + reclaimed += 1 + logger.info( + "controller-bridge: reclaimed stale MERGING workflow " + "wf=%d (PR#%d) -> APPROVED (a previous merge_drive cycle " + "died mid-merge)", + wf.workflow_id, wf.entity_number, + ) + return reclaimed + + # ─── HTTP / API ──────────────────────────────────────────────────────────── # # Implementation lives in tools/_claim_runtime.py and is shared with @@ -1705,6 +1787,17 @@ def run_one_cycle( """ logger.debug("cycle: starting (heartbeat=%s)", cfg.heartbeat_path) write_heartbeat(cfg.heartbeat_path) + # 0. Reclaim any workflow a prior cycle stranded in MERGING (crash + # between the APPROVED→MERGING claim and the outcome emit). Safe + # because merge_drive is a singleton and run_one_cycle is serial — + # nothing is legitimately in MERGING at cycle start. Reclaimed + # workflows return to APPROVED and are re-picked-up below. + reclaimed_merging = _reclaim_stale_merging(cfg) + if reclaimed_merging: + logger.info( + "cycle: reclaimed %d stale MERGING workflow(s) -> APPROVED", + reclaimed_merging, + ) # 1. Sweep claims abandoned by previously-crashed instances across # every auto/claimed-* label, not just our own. Empty set for # ``session_pr_numbers`` because we haven't claimed anything yet @@ -1824,9 +1917,20 @@ def run_one_cycle( if wf_id is None: continue pr_number = int(pr["number"]) - pr_state = outcome.pr_terminal_states.get( - pr_number, outcome.terminal_state, - ) + if pr_number in outcome.pr_terminal_states: + pr_state = outcome.pr_terminal_states[pr_number] + else: + # Every claimed PR should have a per-PR fate; a miss means + # a CycleOutcome construction path forgot to populate it. + # Fall back to the aggregate (the catch-all in + # _resolve_bridge_event still keeps it out of MERGING) but + # log loudly so the gap is visible. + pr_state = outcome.terminal_state + logger.warning( + "controller-bridge: PR#%d wf=%s absent from " + "pr_terminal_states; falling back to aggregate %r", + pr_number, wf_id, pr_state, + ) _emit_controller_event_for_outcome( cfg, pr_number, int(wf_id), pr_state, )