diff --git a/HANDOFF-344-instance-2.md b/HANDOFF-344-instance-2.md new file mode 100644 index 0000000..6645e70 --- /dev/null +++ b/HANDOFF-344-instance-2.md @@ -0,0 +1,215 @@ +# HANDOFF — BACKLOG #344 instance 2 (paused 2026-08-02) + +> **This work is UNPUSHED. There is no PR and no remote branch.** A resumer checking GitHub will +> find nothing and may conclude #344 instance 2 was never fixed. It was. It lives in exactly one +> place: +> +> **HEAD `66443098`, 4 commits ahead of `origin/main`, working tree clean.** You are reading this +> from the branch itself; `git branch --show-current` names it. The branch name is deliberately +> not written here — the leak guard matches worktree/branch slugs by shape, because a slug is +> whatever the task was called and this repo is public. It is recorded, with the worktree path, +> in `.git/mefor-coord/HANDOFF-344-instance-2.md` (never published). +> +> `git rev-parse --verify origin/` → *"Needed a single revision"*. That is not an +> error; it is the confirmation that nothing was pushed. + +Paused on the owner's instruction, relayed by the coordinator session. Nothing is half-done: the +four commits are coherent and independently verified. **Push / PR / merge were never taken — they +route to the owner, and pausing is not a reason to push.** + +--- + +## 1. What was wrong, and what the mechanism actually is + +`tests/test_stage_dispatcher.py`'s ADR 0070 tests failed intermittently on the **SQL Server leg +only** — a bare `assert await _wait_until(...)` → `assert False` with *zero* logic assertions +failing. Two occurrences in 21 days, ~0.4%/test over ~479 observations. + +**It was never the 8.0s bound.** Do not raise it. The test passes in max 0.204s on green SQL Server +jobs and 0.576s at worst across 241 runs — the bound is 14–39× the worst passing run, and both +failures sit ~7× beyond the *entire* passing distribution. A gap, not a tail. Past ~31s a raise +would convert a real 30-second store hang into a silent PASS. + +The actual chain — **proven by forcing it on a live SQL Server, not inferred**: + +1. `claim_fifo_heads` ([`store/sqlserver.py`](messagefoundry/store/sqlserver.py)) runs under + `SET LOCK_TIMEOUT 0`, so a momentarily contended head raises **native error 1222** instantly. +2. The store **catches 1222 and returns a normal EMPTY result** (the `_is_lock_timeout` branch). + This is deliberate — the ADR 0114 "never-block" yield — and is logged **only at DEBUG**, which + is why nothing ever reported it. +3. The dispatcher's EMPTY branch takes **T12**: `phase = IDLE`, **no timer armed**. +4. In production the periodic sweep re-readies such a lane. The ADR 0070 tests set + `sweep_interval=_HUGE_SWEEP` (3600) with an **empty `lane_provider`** *on purpose*, so IDLE is + **terminal** there and the lane can never be re-claimed. The poll then cannot succeed at any + budget. + +**This is a test-rig gap, not an engine defect.** Production's backstop is intact. Do **not** "fix" +T12 by arming a timer on every EMPTY — that re-introduces the exact spin ADR 0070's fix-A backoff +exists to collapse. + +### The reproduction recipe (the expensive part — the scratch harnesses were deleted) + +Deterministic, ~4s, on a live SQL Server. Drive ADR 0070 test 1 to its first PARKED, then from a +**second pooled connection** take a conflicting lock on the head row and hold it across the +post-unpark re-claim: + +``` +cur = await blocker.cursor() +await cur.execute("UPDATE queue SET last_error = last_error WHERE stage=? AND channel_id=?", + (Stage.INGRESS.value, lane)) # NOT committed -> holds an X lock +pu = d.park_until(lane); mc.advance((pu - mc.now) + 0.5); await _settle() +``` + +Observed, and this is the signature to match: + +``` +[1] after fault 1: phase=PARKED park_until=1001.0 empty_claims=(0, 0, 0) +[2] after unpark into a LOCKED head: + phase=IDLE park_until=None armed_timers={} empty_claims=(1, 0, 1) +[3] after releasing the lock: recovered=False phase=IDLE <- TERMINAL, not merely slow +[4] after mark_ready(lane, woken=False): PARKED, streak=2 <- the fix works +``` + +Step [3] is the clause that proves *terminal* rather than *slow*; no timeout reasoning reaches it. + +**A second, quieter route exists** and produces the identical observable: the batch claim's lock +probe runs `WITH (UPDLOCK, ROWLOCK, READPAST)`, so a contended head is *skipped* and the head-pin +step drops the whole lane — with **no log line at all**. Both routes read `(1,0,1)`, so the counters +cannot separate them. That is a limit of the instrument, not doubt about the mechanism. + +**Postgres is NOT immune.** It uses `FOR UPDATE SKIP LOCKED` — the same head-of-line skip. Its 119 +observations at ~0.4% expect ~0.5 events and exonerate nothing. Only **SQLite's** 0/1,105 is +structural. Read the symptom as *server-backed store*, not "SQL Server only". + +### A THIRD source of the 1222 — and it refines the wording above (BACKLOG #348 / ADR 0159) + +Added 2026-08-02 from the #348 session, which reached the same 1222 from the opposite end. **Their +measurement, my verification of the mechanism in our source:** + +`CancelledError` derives from `BaseException`, so the store's `except Exception: await +conn.rollback()` idiom **never runs its rollback on a cancellation** — and `aioodbc`'s +`Pool.release()` returns the connection to the free deque with no rollback or reset. Verified here: +`store/sqlserver.py` has **90** rollback sites guarded by `except Exception` and **zero** using +`except BaseException` or a `finally`. They measured the consequence on a live container — +cancelling `release_claimed` left 7 KEY X locks on `queue` (`reschedule_claimed` 7, `mark_done` 9, +`enqueue_ingress` 11), against 0 for a `claim_fifo_heads` control — after which a real +`claim_fifo_heads` returned EMPTY-all and a raw writer got 1222. + +**Why this matters to the text above:** §1 and the commit message for the fix reason from +*momentary* contention — a live producer holding a head lock in flight. A connection poisoned this +way holds its locks **for as long as it sits unclaimed in the pool's free deque**, so the 1222 can +repeat across successive sweep ticks instead of clearing on the next one. The chain into T12 and the +terminal IDLE is identical; only the *duration and origin* of the contention differ. + +**It does not change this branch's conclusion** — production still recovers via the sweep, so +"test-rig gap, not an engine defect" stands, and that session explicitly declined to escalate its +severity on the strength of this finding. But: **if you ever see a 1222 that cannot be attributed to +a live producer, look here first.** Cross-referenced by ledger number (#348 / ADR 0159), not by SHA, +because that work is also unpushed. + +--- + +## 2. What was changed (4 commits) + +> **If you cite one SHA, cite `58f8f134` — that is THE FIX.** `66443098` and the HEAD commit are a +> banner edit and this note; they touch **no code**. This was already mis-relayed once between +> sessions (HEAD was circulated as "the fix"), so name the commit, not the tip. + +| SHA | What | +|---|---| +| `58f8f134` | **THE FIX** + ledger: `_wait_until` raises on expiry with a full diagnostic; `_wait_lane` sweep stand-in; #344 instance 2 → RESOLVED | +| `e42f0ae3` | Three corrections from adversarial review (see below) | +| `7e430e82` | Merge `origin/main` (`dc0da097`, #148's #345/#346) — verified by **content**, both directions | +| `66443098` | Banner rewritten to land with its body; no instance count | + +**Two halves of the fix**, both in `tests/test_stage_dispatcher.py`: + +1. **`_wait_until` now RAISES** `_WaitTimeout` on expiry (BACKLOG #344 proposal 6) carrying lane + phase, park deadline, infra streak, lane-task liveness/exception, the store row's status and + `next_attempt_at`, the clock, and `empty_claims` — plus a one-line **VERDICT** naming which + hypothesis the reading proves. All ~60 bare-assert call sites convert for free; two that + legitimately tolerate a timeout pass `required=False`; one that carried a message passes `note=`. +2. **`_wait_lane`** stands in for the disabled sweep: a lane found in terminal IDLE is re-readied + via `mark_ready(lane, woken=False)`, **counted, capped at `_MAX_SWEEP_STANDINS = 1`, and + `warnings.warn`-loud**. Past the cap it raises, calling it a contention regression. Disabled when + IDLE is itself the target phase. Measured: 12 consecutive full-file SQL Server runs produced + **zero** stand-ins, so the cap of 1 has margin. + +The `e42f0ae3` corrections, all verified in source before being written: `empty_claims` alone is +**not** a complete discriminator (`== 0` covers ≥3 stuck states that only the *phase* separates, so +the verdict reads both); the READPAST second route; and Postgres not being structurally immune. It +also fixed a shadowing bug from the first commit — a `for lane in ...` loop rebinding the `lane` +**parameter**, which would have named a stale lane with two live dispatchers. + +--- + +## 3. Verification already done — do not redo it blind + +- `ruff format --check`, `ruff check`, **`mypy --strict`** — all clean on the changed file. +- `scripts/docs/backlog_status_check.py` — **green at 270 items**. +- Full `tests/test_stage_dispatcher.py`: **76 passed / 38 skipped** on a live SQL Server 2022 + container *and* on a live Postgres 16; **38 passed** sqlite-only (the default CI leg). +- **Falsification (the load-bearing check):** with `_MAX_SWEEP_STANDINS` forced to 0, the same + injected EMPTY reproduces the original failure — now carrying the full diagnostic instead of a + bare `assert False`. A guard that passes against unpatched code measures nothing. + +### Rebuilding the DB rigs (mine were torn down) + +Both containers were **removed** at the end of the session; the shared `mefor-mssql` / `mefor-pg-ci` +were left alone. To re-verify: + +``` +scripts\dev\sqlserver-docker.ps1 -ContainerName mefor-mssql- -Volume mefor-mssql--data -Port +docker run -d --name mefor-pg- -e POSTGRES_PASSWORD=mefor -e POSTGRES_DB=messagefoundry -p :5432 postgres:16 +``` + +- ⚠️ **A native `MSSQLSERVER` service owns `127.0.0.1:1434` on this box.** Docker had 1433; I used + **14330** for mine. Always check the owner first, then sanity-check `SELECT @@VERSION` matches the + image you started — a wrong-server connect reports a misleading login failure and nothing names it. +- ⚠️ **The two backends share ONE `MEFOR_STORE_*` config.** They cannot run in the same pytest + invocation; run them as separate legs, as CI does. Clear `MEFOR_*` between legs. +- Local pytest **silently skips** both server legs without `MEFOR_TEST_SQLSERVER` / + `MEFOR_TEST_POSTGRES` — a green local run proves nothing about them. +- ⚠️ **An isolated repro loop will not reproduce this.** 800 iterations of the two tests alone + returned 800/800 green: running them in isolation is the one configuration that cannot generate + lock contention. It *did* appear unforced in a **full-file** run. Force it (§1) rather than sample. + +--- + +## 4. Open threads a resumer must not rediscover + +**(a) One known conflict with PR #138** (OPEN, auto-merge ARMED, CI-BLOCKED at pause). +`git merge-tree --write-tree HEAD ` → **exactly one hunk, `docs/BACKLOG.md:~8328`, +the status banner only.** + +**Agreement reached with that session** (peer-to-peer, before the coordinator's remit; the +coordinator has also recorded it — it otherwise lived only in two paused sessions' heads): + +> **#138 lands first. Then merge and take *this branch's* banner, keeping #138's two corrected +> sentences in the `*Instance 2, RE-DIAGNOSED*` paragraph.** + +Rationale: a banner is a claim about the body beneath it, so it must land in the same commit as the +body it describes. #138's banner is true of main-after-#138; mine is true of main-after-this-branch. +**This branch's banner deliberately carries no instance count**, so it stays true in either order — +the agreed order is *preferable, not load-bearing*. Do not panic about ordering, and do **not** +restore a tally: instance 3 and proposal 5 exist only on #138's branch, so naming them here would +describe a body that is not present if this branch lands first. + +Use `git merge-tree --write-tree A B` to re-check — no worktree, no dirty state, no abort to recover +from. It caught this after both sessions had verified the region they were told about and both +concluded the merge was clean. + +**(b) The ledger claim is HELD and must stay held.** Number **344**, scoped to the **fix in +`tests/` only** — #138 holds the `ci.yml`/ledger half. Both notes say so. Do not release it: the +work is real, unlanded and unpushed, and the claim is the only thing marking that territory. + +**(c) Out of scope, already filed as a task chip:** a swallowed 1222 increments the same counter as +a genuinely idle poll, so an operator cannot distinguish contention from no-work. The yield itself +is by design (ADR 0114) and must **not** be changed — this is about making it countable. + +--- + +## 5. Not done + +- **Push, PR, merge** — the owner's call, deliberately not taken. +- The #138 banner reconcile (§4a), which cannot happen until #138 lands. diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index 0ec8bf7..a3d1e43 100644 --- a/docs/BACKLOG.md +++ b/docs/BACKLOG.md @@ -8331,7 +8331,7 @@ Two worked instances the same day. **#74** went green on 2026-07-30 and sat unme ## 344. Fixed wall-clock bounds have drifted out of proportion to the work they bound -> 🚧 **Status 2026-08-02 — PARTIAL: instances 1 and 3 fixed, instance 2 open.** Value **6/10** · Difficulty **4/10** · _fill-in_. Hardcoded real-time budgets — a CI `step_timeout`, a CI `job_timeout`, a test helper's poll deadline — were sized when the work was smaller or the machine faster, and nothing re-derives them. Three confirmed instances; each presents as an unrelated flake. +> 🚧 **Status OPEN (filed 2026-08-01).** Value **6/10** · Difficulty **4/10** · _fill-in_. Hardcoded real-time budgets — a CI `step_timeout`, a CI `job_timeout`, a test helper's poll deadline — were sized when the work was smaller or the machine faster, and nothing re-derives them. The CI-cap instances are closed as each is re-measured; **instance 2 was resolved 2026-08-02 and turned out not to be a bound problem at all** — a swallowed SQL Server lock-timeout stranded a lane in a terminal IDLE, and no budget would have helped. The generic remedy — a mechanical margin check, and a sweep of the remaining bounds — is **not** built. *Deliberately no count of instances in this banner:* concurrent PRs are adding and closing them, so an enumeration is a completeness claim that goes stale on whichever lands first — the liability [`Secure_Development_Standards`](Secure_Development_Standards.md) §3 warns about. The item's real lesson is now instance 2's: **a timeout with zero failing assertions is a diagnosis to be made, not a number to be raised.** **Cluster:** Developer Experience & CI. **Priority:** P2. **Verdict:** build. **Severity:** medium. @@ -8339,7 +8339,13 @@ Two worked instances the same day. **#74** went green on 2026-07-30 and sat unme *Instance 1 (fixed 2026-08-01, this filing).* `ci.yml`'s Windows `step_timeout: 26`. windows-2025's max PASSING `Tests (pytest)` step was **25:51** — a **1.006x** margin, nine seconds — and PR #119 was killed at 26:07 with zero test failures after #74 added a 1,506-line test file. The comment beside the cap asserted "~2x headroom", a figure that matched no leg. Raised to 36:00 (1.393x) with the measurement, its pool and its date recorded in place of the multiple. Pool: every `ci.yml` run created 2026-08-01 UTC — 70 runs, per-leg n = 42 ubuntu / 39 windows-2022 / 36 windows-2025 — timing the STEP and filtering on the STEP's own conclusion. That pool is **right-censored**: it predates the raise, so every observation survived a 26:00 cap and 25:51 is a *lower bound* on what the suite wants. Confirmed the next day, once the cap no longer censored it: windows-2025 produced **26:23 twice, both passing** — runs that the old cap would have killed. The API hides this by default: the jobs endpoint returns only the latest attempt, so a step killed at the cap is invisible behind its passing re-run unless you ask for `?filter=all`. -*Instance 2 (open).* `tests/test_stage_dispatcher.py`'s `_wait_until` (:356) polls `loop.time()` against a hardcoded **8.0s** budget while the dispatcher under test is driven by an injected `ManualClock`. On PR #129 — whose diff is provably AST-identical to main, docstrings only — `test_adr0070_9_content_retry_is_not_an_infra_fault[sqlserver]` failed on that bound alone against a Dockerised SQL Server; every logic assertion in the same loop passed. A real-time deadline gating a virtual-clock system has no principled value. +*Instance 2 (**RESOLVED 2026-08-02 — and it was never a bound problem**).* Presented as `tests/test_stage_dispatcher.py`'s `_wait_until` 8.0s budget expiring on the SQL Server leg with zero logic assertions failing (two occurrences in 21 days: `test_adr0070_1_stop_policy_bounds_deterministic_infra_head`, run 30733129076; `test_adr0070_9_content_retry_is_not_an_infra_fault`, run 30716979773 attempt 1 — ~0.4%/test over ~479 observations, zero on Postgres/119 and zero on SQLite/1,105). **The bound was never the cause and must not be raised**: the test passes in max 0.204s on green SQL Server jobs and 0.576s at worst across 241 runs, so 8.0s is 14–39× the worst passing run and both failures sit ~7× beyond the *entire* passing distribution — a gap, not a tail. Past ~31s a raise would convert a real 30-second store hang into a silent PASS. + +The actual mechanism, **proven by forcing it on a live SQL Server** rather than argued: `claim_fifo_heads` runs under `SET LOCK_TIMEOUT 0`, so a momentarily contended head raises native error **1222** instantly; the store **catches 1222 and returns a normal EMPTY result** (`store/sqlserver.py`, the `_is_lock_timeout` branch — deliberate, the "never-block" yield, logged only at DEBUG). The dispatcher's EMPTY branch then takes **T12**: `phase=IDLE` with **no timer armed**. In production the periodic sweep re-readies exactly such a lane; the ADR 0070 tests set `sweep_interval=3600` with an empty `lane_provider` **on purpose**, so IDLE is **terminal** there and the lane can never be re-claimed — the poll then cannot succeed at any budget. Holding a conflicting row lock across the post-unpark re-claim reproduces it deterministically: `empty_claims` goes `(0,0,0)` → `(1,0,1)`, the lane sits `IDLE` with `park_until=None` and no armed timer, the head is still `pending` and **due**, and releasing the lock does *not* recover it. The same stranding was then observed **unforced** on a local full-file run (`test_adr0070_1_retry_forever_never_stops_alerts_stuck[sqlserver]`). This is a **test-rig gap, not an engine defect** — production's backstop (the sweep) is intact, and arming a timer on every EMPTY instead would re-introduce the very spin ADR 0070's fix-A backoff exists to collapse. + +**"SQL Server-only" is the wrong reading, and the distribution does not say otherwise.** There is a *second*, quieter route to the same spurious EMPTY on SQL Server: the batch claim's lock probe runs `WITH (UPDLOCK, ROWLOCK, READPAST)`, so a contended head is **skipped**, and the head-pinned contiguity step then drops the whole lane ("rn=1 missing drops the whole lane => EMPTY") — with **no log line at all**, not even the DEBUG one. And Postgres is **not structurally immune**: its claim uses `FOR UPDATE SKIP LOCKED` ([`store/postgres.py`](../messagefoundry/store/postgres.py)), the same head-of-line skip; its 119 observations at a ~0.4% rate expect ~0.5 events, so they exonerate nothing. Only **SQLite's** 0/1,105 is structural (in-process, no lock-timeout translation and no skip-locked semantics). Read the symptom as **server-backed store**, and note that the counters cannot separate the two SQL Server routes — both read `(1, 0, 1)`. + +Fixed two ways in [`tests/test_stage_dispatcher.py`](../tests/test_stage_dispatcher.py): `_wait_until` now **raises on expiry** carrying the lane's phase, park deadline, streak, task state, the store row's status/`next_attempt_at`, the clock, and `empty_claims` — plus a one-line **verdict** naming which hypothesis the reading proves, so no future occurrence needs re-diagnosing from scratch; and the ADR 0070 waits now stand in for the sweep those tests disable — re-readying a lane found in terminal IDLE, but **counted and capped at one**, so a genuine contention regression still fails loudly instead of being absorbed. Both halves are falsification-tested: with the stand-in disabled the same injected EMPTY reproduces the original failure, now carrying the full diagnostic instead of a bare `assert False`. *Instance 2, RE-DIAGNOSED 2026-08-02 — and the original diagnosis was wrong.* It recurred on PR #138, on a different test (`test_adr0070_1_stop_policy_bounds_deterministic_infra_head[sqlserver]`, run 30733129076), and the timings refute "the bound is too small": the failing test took **8.185s**, i.e. `_wait_until` burned its full 8.000s and setup+teardown cost 0.185s — the store was *fast* at the moment of failure. In the same process, against the same container, the sibling `test_adr0070_1_retry_forever_never_stops_alerts_stuck[sqlserver]` drove **seven** identical fault cycles in **0.364s**, and the `[sqlite]` variant of the failing test passed in 0.144s. One fault cycle costs ~30-45 ms against an 8000 ms bound. Over a sample of green sqlserver jobs this test passes in min 0.185s / median 0.196s / max 0.204s, so the bound is ~39x its worst passing run; a wider 21-day pass put the worst passing observation at 0.576s, still ~14x. Rate: 2 failures in ~479 observations of the two affected tests (~0.4%). Postgres saw zero in 119 observations, which **exonerates nothing** — at ~0.4% that sample expects ~0.5 events, and Postgres claims via `FOR UPDATE SKIP LOCKED`, the same head-of-line skip; only SQLite's 0 in 1,105 is structural, because its global lock totally orders producers and claimers. Both failures sit ~7x beyond the entire passing distribution — a gap, not a tail, which is the signature of a categorically different event rather than latency creep. @@ -8361,7 +8367,7 @@ What is NOT settled is the mechanism. Two independent passes reached different a 5. **Stop two steps sharing one `timeout-minutes` budget.** Give `Web console tests (pytest)` its own cap sized to its own work (max observed 3:33) so `job_timeout` no longer has to absorb a budget that belongs to a step. Until then the nesting invariant `ci.yml` asserts is unenforceable for the second gated step on every leg — instance 3 is the worked example. 6. **Make a bound's expiry diagnostic before tuning it — and for instance 2 the instrument already ships.** A bare `assert await _wait_until(...)` reports `assert False` and nothing else, so every occurrence is re-diagnosed from scratch; instance 2 was read as latency for a day on exactly that basis. Have the helper raise on timeout carrying the lane's phase, park deadline and streak, whether its task is alive or holds an exception, the store row's status, and the clock. **One assertion settles instance 2's open mechanism:** `StageDispatcher.empty_claims` ([`stage_dispatcher.py`](../messagefoundry/pipeline/stage_dispatcher.py):1230) returns `(total, wake_fanout, idle_poll)` and is fed by `_record_empty`, called from exactly one site — the EMPTY branch of `_claim_and_dispatch` (:686). Under these tests' topology a clean run must read `(0, 0, 0)`, so `empty_claims[0] > 0` at the moment of failure is proof of a spurious EMPTY, and `== 0` is proof the claim never returned at all. A second, free signature is in the captured log, **for the infra-fault test only**: a healthy `test_adr0070_1_*` emits **four** `re-pending head with backoff` records (at `1001.000 / 1003.500 / 1008.000 / 1016.500`) and the failing run emitted **one**. It does NOT generalise — `test_adr0070_9_*` takes the content path, which uses `mark_failed` and never emits that line, so **zero** there is expected and is not a second mechanism. Read the counter, not the log, when in doubt. This proposal cannot itself be wrong about the cause, which is why it comes before the others. -**Related:** [`.github/workflows/ci.yml`](../.github/workflows/ci.yml) §*Tests (pytest)* (instance 1 and its measurement table); `tests/test_stage_dispatcher.py`:356 (instance 2); #320 (windows-2025 slowness — the capacity fact that shrinks every Windows margin); #340 (the other half of this triage); [`Secure_Development_Standards`](Secure_Development_Standards.md) §3 (prose asserting a margin the numbers do not support — five instances found on 2026-08-01 alone). +**Related:** [`.github/workflows/ci.yml`](../.github/workflows/ci.yml) §*Tests (pytest)* (instance 1 and its measurement table); [`tests/test_stage_dispatcher.py`](../tests/test_stage_dispatcher.py) (instance 2 — `_wait_until`'s raising expiry report and the counted `_wait_lane` sweep stand-in); [`store/sqlserver.py`](../messagefoundry/store/sqlserver.py) (`_is_lock_timeout` — the 1222-as-EMPTY yield that instance 2 turned on); ADR 0070 (the infra-fault machinery the affected tests cover); #320 (windows-2025 slowness — the capacity fact that shrinks every Windows margin); #340 (the other half of this triage); [`Secure_Development_Standards`](Secure_Development_Standards.md) §3 (prose asserting a margin the numbers do not support — five instances found on 2026-08-01 alone). **Source:** stuck-CI triage, 2026-08-01. Instance 1 re-measured 2026-08-02 across 36 step-success windows-2025 rows (pool: 70 `ci.yml` runs created 2026-08-01 UTC), by two agents deriving it independently after the first two measurements both reported pools that did not reproduce; instance 2 reported and diagnosed by the HA-construct-recheck session from PR #129's sqlserver leg; instance 3 from the same 2026-08-02 re-measurement. diff --git a/tests/test_stage_dispatcher.py b/tests/test_stage_dispatcher.py index bdd4f0d..8a188e3 100644 --- a/tests/test_stage_dispatcher.py +++ b/tests/test_stage_dispatcher.py @@ -32,7 +32,8 @@ import asyncio import os import random -from collections.abc import AsyncIterator, Callable +import warnings +from collections.abc import AsyncIterator, Callable, Iterator from dataclasses import dataclass from pathlib import Path from typing import Any, cast @@ -188,6 +189,9 @@ class ManualClock: def __init__(self, start: float = 0.0) -> None: self.now = start self._armed: list[_ManualHandle] = [] + _LIVE_CLOCKS.append( + self + ) # so a `_wait_until` expiry can report virtual time (BACKLOG #344) def time(self) -> float: return self.now @@ -353,16 +357,159 @@ async def _settle(rounds: int = 8) -> None: await asyncio.sleep(0.005) -async def _wait_until(pred: Callable[[], bool], timeout: float = 8.0) -> bool: +# --- expiry diagnostics (BACKLOG #344 proposal 6) ------------------------------------------------- +# +# A bare `assert await _wait_until(...)` reports `assert False` and NOTHING else, so every occurrence +# gets re-diagnosed from scratch — BACKLOG #344 instance 2 was read as latency for a day on exactly +# that basis. `_wait_until` therefore RAISES on expiry, carrying the state a reader needs. +# +# The dispatcher/clock are not passed in (the predicate is an opaque closure): they register +# themselves on construction and `_reset_wait_registry` clears them per test, so every call site gets +# the diagnostic for free with no per-site argument. + +_LIVE_DISPATCHERS: list[StageDispatcher] = [] +_LIVE_CLOCKS: list[ManualClock] = [] + + +class _WaitTimeout(AssertionError): + """A `_wait_until` poll expired. AssertionError so pytest reports it as a test failure, not an + error, and so the historical `assert await _wait_until(...)` form reads unchanged.""" + + +@pytest.fixture(autouse=True) +def _reset_wait_registry() -> Iterator[None]: + """Scope the diagnostic registries to one test (they are module-level by necessity).""" + _LIVE_DISPATCHERS.clear() + _LIVE_CLOCKS.clear() + _SWEEP_STANDINS.clear() + yield + _LIVE_DISPATCHERS.clear() + _LIVE_CLOCKS.clear() + _SWEEP_STANDINS.clear() + + +def _lane_report(d: StageDispatcher, lane: str) -> str: + task = d._lane_tasks.get(lane) + if task is None: + task_state = "none" + elif not task.done(): + task_state = "ALIVE" + elif task.cancelled(): + task_state = "cancelled" + else: + exc = task.exception() + task_state = f"raised {exc!r}" if exc is not None else "finished" + return ( + f" {lane}: phase={getattr(d.phase(lane), 'name', None)}" + f" park_until={d.park_until(lane)} streak={d.infra_error_streak(lane)}" + f" dirty={d.is_dirty(lane)} timer_armed={d._timer_deadline.get(lane)}" + f" lane_task={task_state}" + ) + + +def _verdict(d: StageDispatcher, lane: str | None = None) -> str: + """State which #344 hypothesis this expiry is, so the next reader need not know the theory. + + `empty_claims` alone is NOT a complete discriminator — it must be read WITH the lane phase. + `empty_claims[0] > 0` is biconditional with the T12 stranding only because the other EMPTY + consumer (T11, EMPTY-with-dirty) is unreachable under this topology, and `== 0` covers several + distinct stuck states that only the phase separates. + """ + phase = d.phase(lane) if lane is not None else None + pname = getattr(phase, "name", "?") + if d.empty_claims[0] > 0 and (phase is None or phase is _LanePhase.IDLE): + return ( + "VERDICT: a claim came back EMPTY (empty_claims[0] > 0) and T12 dropped the lane to a" + " TERMINAL IDLE — no timer armed, and these tests disable the sweep that would re-ready" + " it. Two known routes, both SQL Server, both reported to the dispatcher as an ordinary" + " empty result: a swallowed lock-timeout (native 1222 under SET LOCK_TIMEOUT 0, logged" + " only at DEBUG) and a READPAST head skip (the batch claim drops the whole lane when its" + " rn=1 head is locked — no log line at all). Postgres is NOT structurally immune: its" + " claim uses FOR UPDATE SKIP LOCKED, the same head-of-line skip. BACKLOG #344 instance 2." + ) + if d.empty_claims[0] > 0: + return ( + f"VERDICT: an EMPTY claim was recorded (empty_claims[0] > 0) but the lane is {pname}," + " not IDLE — so this is NOT the plain #344 instance-2 stranding. Read the phase:" + " CLAIMING/PROCESSING means a later claim or serializer is still outstanding; PAUSED" + " means the pause branch consumed it." + ) + return ( + "VERDICT: NO empty claim was recorded (empty_claims[0] == 0), so the claim never returned at" + f" all — a stalled statement or a lost wakeup. The lane is {pname}: CLAIMING = the claim is" + " outstanding; PROCESSING = the serializer is hung; READY = the claimer is not picking it up" + " (check slots_free and any claim-error backoff). This is NOT the #344 instance-2 stranding" + " and must NOT be papered over with a re-nudge." + ) + + +async def _expiry_report(pred: Callable[[], bool] | None = None, lane: str | None = None) -> str: + """Everything a reader needs to tell the #344 hypotheses apart, without a second CI round-trip.""" + lines = ["_wait_until expired."] + if pred is not None: + try: + lines.append(f" predicate re-read after expiry: {pred()!r}") + except Exception as exc: # noqa: BLE001 — a raising predicate must not hide the report + lines.append(f" predicate re-read RAISED: {exc!r}") + for i, d in enumerate(_LIVE_DISPATCHERS): + # The verdict needs a lane: use the caller's, else the only one if this dispatcher has one. + subject = ( + lane if lane is not None else (next(iter(d._states)) if len(d._states) == 1 else None) + ) + lines.append(f" {_verdict(d, subject)}") + lines.append( + f" dispatcher[{i}] stage={d._stage.value}" + f" empty_claims(total,wake_fanout,idle_poll)={d.empty_claims}" + f" busy_violations={d.busy_violations} processing_lanes={d.processing_lanes}" + f" slots_free={d._slots_free}" + ) + # NB distinct loop names: a bare `for lane in ...` here would REBIND the `lane` parameter and + # give the next dispatcher's verdict a stale subject. + lines.extend(_lane_report(d, ln) for ln in d._states) + # The store row is the other half: status + next_attempt_at say whether the head was even + # claimable. Bounded and swallowed — a diagnostic that hangs or raises would MASK the failure + # it is meant to explain. + try: + for ln in list(d._states): # `ln`, not `lane` — see the rebinding note above + rows = await asyncio.wait_for(_lane_rows(d._store, ln), timeout=5.0) + lines.append( + f" {ln} rows=" + + str([(r["status"], r["attempts"], r["next_attempt_at"]) for r in rows]) + ) + except Exception as exc: # noqa: BLE001 — a diagnostic must never mask the real failure + lines.append(f" ") + for i, mc in enumerate(_LIVE_CLOCKS): + lines.append(f" clock[{i}].now={mc.now} armed={[h.fire_at for h in mc.armed]}") + return "\n".join(lines) + + +async def _wait_until( + pred: Callable[[], bool], + timeout: float = 8.0, + *, + required: bool = True, + note: str | None = None, +) -> bool: """Poll ``pred`` (letting async work run) until true or ``timeout``. Robust against aiosqlite thread - latency without fixed-iteration flakiness.""" + latency without fixed-iteration flakiness. + + On expiry this RAISES :class:`_WaitTimeout` with a full dispatcher/store dump (BACKLOG #344 + proposal 6) instead of returning False — a bare ``assert False`` is what made instance 2 cost a day + of re-diagnosis. Pass ``required=False`` for the handful of call sites that legitimately treat a + timeout as a value rather than a failure. + """ loop = asyncio.get_running_loop() deadline = loop.time() + timeout while loop.time() < deadline: if pred(): return True await asyncio.sleep(0.003) - return pred() + if pred(): + return True + if not required: + return False + report = await _expiry_report(pred) + raise _WaitTimeout(f"{note}\n{report}" if note else report) def _make( @@ -381,7 +528,7 @@ def _make( infra_fault_stop_after: int = 10, infra_fault_backoff_cap: float = 60.0, ) -> StageDispatcher: - return StageDispatcher( + d = StageDispatcher( Stage.INGRESS, store, process_item=stub, @@ -399,6 +546,8 @@ def _make( infra_fault_stop_after=infra_fault_stop_after, infra_fault_backoff_cap=infra_fault_backoff_cap, ) + _LIVE_DISPATCHERS.append(d) # so a `_wait_until` expiry can dump it (BACKLOG #344 proposal 6) + return d def _first_occurrence_order( @@ -795,7 +944,7 @@ def _quiescent() -> bool: d.notify_work() mc.advance(1_000_000.0) await d._run_sweep_once() - if await _wait_until(_quiescent, timeout=3.0): + if await _wait_until(_quiescent, timeout=3.0, required=False): drained = True break assert drained, "soak did not reach quiescence" @@ -968,6 +1117,72 @@ def _always_raise(lane: str, item: OutboxItem) -> str: return "RAISE" +# How many times a lane may be re-readied by the sweep stand-in below before the test fails outright. +# 1 covers the observed rate (a swallowed 1222 is a rare, independent event); a lane needing more than +# that is a contention REGRESSION, not the transient this stands in for — so it must still fail. +_MAX_SWEEP_STANDINS = 1 +_SWEEP_STANDINS: dict[str, int] = {} + + +async def _wait_lane( + d: StageDispatcher, lane: str, *phases: _LanePhase, timeout: float = 8.0 +) -> None: + """Wait for ``lane`` to reach any of ``phases``, standing in for the periodic sweep that the ADR + 0070 tests deliberately disable. + + A claim can come back EMPTY even though the lane's head is PENDING and due: on SQL Server + ``claim_fifo_heads`` runs under ``SET LOCK_TIMEOUT 0``, so a momentarily contended head raises + native error 1222, which the store CATCHES and reports as a normal empty result + ([`store/sqlserver.py`] ``_is_lock_timeout`` branch) — by design, the "never-block" yield. The + dispatcher's EMPTY branch then takes T12: ``phase=IDLE`` with **no timer armed**. In production the + periodic sweep re-readies exactly that lane; these tests set ``sweep_interval=_HUGE_SWEEP`` with an + empty ``lane_provider`` on purpose, so IDLE is TERMINAL here and the lane can never be re-claimed — + which is the whole of BACKLOG #344 instance 2 (proven by forcing a real 1222 mid-re-claim). + + So re-ready a stranded lane, but COUNT it and cap it at ``_MAX_SWEEP_STANDINS``: absorbing this + silently would hide a genuine contention regression, which is the failure mode the cap exists for. + + When IDLE is itself an expected outcome the stand-in is disabled — a test that WANTS the lane idle + must never have it re-readied underneath its assertion. + """ + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + idle_is_a_target = _LanePhase.IDLE in phases + while loop.time() < deadline: + phase = d.phase(lane) + if phase in phases: + return + # TERMINAL IDLE: no timer armed and (in this topology) no sweep to re-arm one. + if not idle_is_a_target and phase is _LanePhase.IDLE and lane not in d._timer_deadline: + used = _SWEEP_STANDINS.get(lane, 0) + if used >= _MAX_SWEEP_STANDINS: + raise _WaitTimeout( + f"{lane} stranded in terminal IDLE {used + 1} times — past the" + f" {_MAX_SWEEP_STANDINS} allowed sweep stand-in(s). That is a contention" + f" regression, not the transient 1222 this stands in for.\n" + + await _expiry_report(lane=lane) + ) + _SWEEP_STANDINS[lane] = used + 1 + # LOUD, but not fatal: surfaced in pytest's warnings summary so the real-world rate of the + # swallowed 1222 stays visible (and reviewable) instead of being silently absorbed here. + warnings.warn( + f"sweep stand-in: {lane} was stranded in terminal IDLE after an EMPTY claim" + f" (empty_claims={d.empty_claims}) — re-readied ({used + 1}/{_MAX_SWEEP_STANDINS})." + " See BACKLOG #344 instance 2.", + stacklevel=2, + ) + # woken=False: a sweep-class readiness, not a producer wake (books EMPTY as idle_poll). + d.mark_ready(lane, woken=False) + await asyncio.sleep(0.003) + want = "/".join(p.name for p in phases) + raise _WaitTimeout(f"{lane} never reached {want}\n" + await _expiry_report(lane=lane)) + + +def _sweep_standins(lane: str) -> int: + """How many times `_wait_park_or_stop` had to stand in for the disabled sweep on ``lane``.""" + return _SWEEP_STANDINS.get(lane, 0) + + async def _advance_past_park(d: StageDispatcher, mc: ManualClock, lane: str) -> None: """Advance the ManualClock just past ``lane``'s current park deadline so its park timer fires, re-claiming the (not-due) faulting head → the next infra fault. Settles the async work after.""" @@ -983,7 +1198,7 @@ async def _drive_infra_faults_until_stop( """Drive an always-faulting lane to STOPPED by repeatedly firing its park timer, bounded by ``budget`` iterations (a guard against a non-terminating loop if the bound regressed).""" for _ in range(budget): - assert await _wait_until(lambda: d.phase(lane) in (_LanePhase.PARKED, _LanePhase.STOPPED)) + await _wait_lane(d, lane, _LanePhase.PARKED, _LanePhase.STOPPED) if d.phase(lane) == _LanePhase.STOPPED: return await _advance_past_park(d, mc, lane) @@ -1048,7 +1263,7 @@ async def test_adr0070_1_retry_forever_never_stops_alerts_stuck(store: Any) -> N try: d.mark_ready(lane) for _ in range(7): # well past the horizon - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.phase(lane) != _LanePhase.STOPPED # never terminal await _advance_past_park(d, mc, lane) # WAIT for the post-advance retry cycle (unpark → claim → re-fault → re-park) to settle back to @@ -1056,7 +1271,7 @@ async def test_adr0070_1_retry_forever_never_stops_alerts_stuck(store: Any) -> N # outlasts _advance_past_park's fixed _settle() and catches the lane mid-cycle in PROCESSING # (mirrors test 9's post-loop _wait_until). The invariant under test is unchanged: still PARKED, # never STOPPED. - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) >= 7 # streak keeps accruing (drives the backoff) assert alert.stopped == [] # never STOPs assert len(alert.stuck) == 1 # throttled: emitted ONCE at the horizon crossing @@ -1081,13 +1296,13 @@ async def test_adr0070_2_transient_infra_fault_self_heals(store: Any) -> None: await d.start() try: d.mark_ready(lane) # fault 1 - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 1 await _advance_past_park(d, mc, lane) # fault 2 - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 2 await _advance_past_park(d, mc, lane) # resolves → clean drain - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.IDLE) + await _wait_lane(d, lane, _LanePhase.IDLE) assert d.infra_error_streak(lane) == 0 # streak reset on the resolving pass assert alert.stopped == [] # never STOPped assert [r.message_id for r in stub.records] == [mid, mid, mid] # head retried head-first @@ -1110,15 +1325,15 @@ async def test_adr0070_3_streak_reset_scoping_sharp_edge(store: Any) -> None: await d.start() try: d.mark_ready(lane) # fault 1 - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 1 await _advance_past_park(d, mc, lane) # park-timer unpark → fault 2 - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 2 # ACCRUED across the park-timer unpark (not reset) await _advance_past_park(d, mc, lane) # park-timer unpark → fault 3 == threshold → STOP - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.STOPPED) + await _wait_lane(d, lane, _LanePhase.STOPPED) assert d.infra_error_streak(lane) == 3 d.notify_work() # STOPPED→READY reload resume — the ONLY reset path for a stopped lane @@ -1144,11 +1359,11 @@ async def test_adr0070_3b_streak_resets_on_forward_progress(store: Any) -> None: await d.start() try: d.mark_ready(lane) # round 1: i==0 head fault - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 1 # zero-progress fault accrued await _advance_past_park(d, mc, lane) # round 2: head RESOLVES, row1 raises at i==1 - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 0 # forward-progress fault RESET the streak assert mids[0] in {r.message_id for r in stub.records} finally: @@ -1191,7 +1406,7 @@ def policy(lane_: str, item: OutboxItem) -> str: n_before = len(stub.records) d.notify_work() await _settle() - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.IDLE) + await _wait_lane(d, lane, _LanePhase.IDLE) resumed = [r.message_id for r in stub.records[n_before:]] assert resumed == mids # head first, then the tail — strict FIFO, no overtake assert d.busy_violations == 0 @@ -1240,7 +1455,7 @@ async def test_adr0070_6_spin_collapses_to_backoff(store: Any) -> None: await d.start() try: d.mark_ready(lane) # one infra fault → head re-pended not-due - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) park_until = d.park_until(lane) assert park_until is not None and park_until > mc.now # future deadline @@ -1303,7 +1518,7 @@ async def test_adr0070_7_reload_resumes_idempotently(store: Any) -> None: mc.advance((rows[0]["next_attempt_at"] - mc.now) + 1.0) d.notify_work() await _settle() - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.IDLE) + await _wait_lane(d, lane, _LanePhase.IDLE) assert mid in { r.message_id for r in stub.records } # the SAME preserved head, finally delivered @@ -1382,13 +1597,13 @@ async def test_adr0070_9_content_retry_is_not_an_infra_fault(store: Any) -> None try: d.mark_ready(lane) # content retry 1 for n in range(6): # drive 6 content retries — twice past stop_after=3 - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 0, ( f"content retry accrued the infra streak after retry {n + 1}" ) assert d.phase(lane) != _LanePhase.STOPPED await _advance_past_park(d, mc, lane) # park-timer unpark → next content retry - assert await _wait_until(lambda: d.phase(lane) == _LanePhase.PARKED) + await _wait_lane(d, lane, _LanePhase.PARKED) assert d.infra_error_streak(lane) == 0 assert d.phase(lane) != _LanePhase.STOPPED assert alert.stopped == [] # the infra STOP never fired on message-content poison @@ -1420,12 +1635,15 @@ async def test_wakeless_backlog_drains_greedily_not_sweep_gated(store: Any) -> N sweep_interval=5.0, # LARGE: sweep-gated draining would need ~n×5s; greedy T13b ignores it clock=mc.time, ) + _LIVE_DISPATCHERS.append(d) # built without _make — register for the #344 expiry dump await d.start() # seed-all-READY (woken=False), NO explicit producer wake try: # 8s budget << the ~100s a sweep-gated (1 row / 5s) drain would take: only the greedy re-arm passes. # (RESOLVED leaves each claimed row INFLIGHT, so the stub records exactly one dispatch per row.) - assert await _wait_until(lambda: len(stub.records) >= n, timeout=8.0), ( - "wake-less backlog did not drain greedily" + await _wait_until( + lambda: len(stub.records) >= n, + timeout=8.0, + note="wake-less backlog did not drain greedily", ) assert [ r.message_id for r in stub.records if r.lane == lane @@ -1865,7 +2083,7 @@ def _quiescent() -> bool: d.notify_work() mc.advance(1_000_000.0) await d._run_sweep_once() - if await _wait_until(_quiescent, timeout=3.0): + if await _wait_until(_quiescent, timeout=3.0, required=False): drained = True break assert drained, "pause/resume soak did not reach quiescence"