diff --git a/changelog.d/1611.fixed.md b/changelog.d/1611.fixed.md new file mode 100644 index 000000000..0de7bc7d3 --- /dev/null +++ b/changelog.d/1611.fixed.md @@ -0,0 +1,28 @@ +- **A row whose re-pend fails after a store fault is recovered without a restart, and no later row + on its lane overtakes it.** A store fault between a committed claim and its handoff leaves the + row `inflight`, and the worker's re-pend of it can fail too. Before, that re-pend was tried once, + and the lane went on claiming past the stranded row until the next start. Now a per-lane worker, + or a pooled lane serializer, whose re-pend fails retries it on a backoff that doubles from 1 s to + 30 s while it holds the lane, claiming nothing newer until the row is pending again, so per-lane + order holds. The first re-pend is always sent; a retry only while this node leads under the lease + it held at the fault, so a node that regains the lead after another node led sends none. The + trade: a re-pend that keeps failing now holds its lane, where before the lane moved on out of + order; the in-flight watch added in 0.5.0 pages such a lane in both claim modes while + `buildup_max_oldest_seconds` is on (300 s by default). At least these end a hold before the + write lands, leaving the row in flight as one failed re-pend did before, after which the lane + goes on as it did then. A stop or pause of an outbound, by an operator, a schedule window or a + DR park, read about once a second, so the purge that waits for the lane to quiesce is not held + until the store heals. The fail-closed halt when the application log cannot be written, which + also ends a router, transform or response lane's hold, in both claim modes. Another node having led since the rows were claimed. In the pooled + claim mode under `infra_fault_policy = stop`, `infra_fault_stop_after` failed re-pends, after + which the lane STOPs with `connection_stopped`; under `retry_forever` the hold goes on and raises + one `lane_stuck` there. A batching outbound's hold on its coalesced rows ends at that count + under `stop` without a STOP of its own, and raises no `lane_stuck` under `retry_forever`. A + per-lane worker never reads `infra_fault_policy`. A stop of the engine + leaves the row in flight for the next start, as before. At least these are not covered. A worker + killed by an error that is not an `Exception` still leaves its row in flight until the next + start, and its replacement claims past it, as before; in the pooled claim mode a serializer + killed that way is not respawned. A claim that committed but whose reply never reached the + worker leaves it no record of the row, which stays in flight until a store recovery path + re-pends it, at least the next start. A node that loses and regains the lead inside one store + write can still send that one retry. (`BACKLOG #1611`) diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index f3cb8ed3c..944d6e0b2 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -1224,7 +1224,7 @@ Because the local diff is cheap and PHI-safe it is **on by default** (zero phone | `ordering` | enum | `fifo` | default queue ordering per outbound: `fifo` (strict in-order, head-of-line on failure) or `unordered` (batch + rotate-past-failures). `unordered` isolates a stuck message and never adds concurrency — a lane sends one message at a time either way. It works in both claim modes. Switching a lane that is already running from `fifo` to `unordered` takes effect at the next engine start, not on reload. Per-outbound `ordering=` overrides. | | `internal_error` | enum | `continue` | what a worker does on an **internal/code error**: on an outbound's delivery worker, at least an `Exception` other than a `DeliveryError` raised by `send` (our bug, not the partner's); on an inbound's router and transform workers, at least a router or handler raise. `continue` (dead-letter the row + advance) or `stop` (halt the worker and raise a `connection_stopped` alert). Under `stop` the row is dead-lettered only when its attempts, counting the one that just failed, have reached the retry cap; a WARNING says when, and names at most 10 of the affected messages. A batched outbound decides for the whole batch from its first row still present, so when that row has reached the cap the dead-letter takes every row of the batch still present, even rows below the cap. For a delivery worker the cap is `retry_max_attempts` above, unless the outbound has its own retry policy (code-first `retry=RetryPolicy(...)`, or an `[outbound.retry]` table in `connections.toml`, whose cap key is `max_attempts`); then that policy's `max_attempts` is the cap. The router and transform workers always take `retry_max_attempts`. How to recover is in [USER-GUIDE.md, *A lane stopped processing*](USER-GUIDE.md#common-problems). Per-outbound `internal_error=` overrides this for that outbound's delivery worker only. Partner NAKs / transport failures are unaffected. | | `buildup_max_depth` | int | _unset_ | raise a `queue_buildup` alert when an outbound lane's pending depth reaches this. Unset = depth dimension off (a healthy ceiling is throughput-specific, so there's no safe default). Per-outbound `buildup=BuildupThreshold(...)` overrides. | -| `buildup_max_oldest_seconds` | num | 300 | raise `queue_buildup` when the lane's **oldest** pending message has waited this long (a stuck head retrying its way toward the cap is the classic cause). On by default — a head stuck >5 min is a problem in any environment. Set to unset/`0`-disable via a per-outbound override. The same age also pages a row held **in flight** that long since its claim, on every stage, read every 30 s (BACKLOG #1611). A store fault between a claim and its handoff can leave such a row in flight until a store recovery path re-pends it, and the pending age cannot see it. | +| `buildup_max_oldest_seconds` | num | 300 | raise `queue_buildup` when the lane's **oldest** pending message has waited this long (a stuck head retrying its way toward the cap is the classic cause). On by default — a head stuck >5 min is a problem in any environment. Set to unset/`0`-disable via a per-outbound override. The same age also pages a row held **in flight** that long since its claim, on every stage, read every 30 s (BACKLOG #1611). A store fault between a claim and its handoff can leave such a row in flight, and the pending age cannot see it. The worker holding the row retries its re-pend while it holds the lane, so no later row overtakes it. While the store stays down the row stays in flight. At least a stop or pause of the outbound, the fail-closed halt when the application log cannot be written (on every stage), another node taking the lead, and the pooled `infra_fault_policy = stop` bound end such a hold, and a worker killed by an error that is not an `Exception` ends it too; each leaves the row in flight until a store recovery path re-pends it, at least a restart. This age pages them, except a row on an outbound that stays stopped or paused: the in-flight watch skips such a lane, so its row is not paged until the lane runs again. | | `stall_max_oldest_seconds` | num | _unset_ | raise a `message_stall` alert (Corepoint "Max Message Stall", [ADR 0014](adr/0014-alerting-rules-engine.md)) when an outbound lane's **oldest undelivered message** has waited this long. **Unset (the default) = the stall alert is OFF** — deny-by-default/opt-in, because it overlaps `buildup_max_oldest_seconds`'s age dimension and would double-page if both fired. Set a threshold to turn it on; a per-outbound `stall=StallThreshold(...)` overrides it. The stall event routes through `[[alerts.rules]]` like any other ([ADR 0014](adr/0014-alerting-rules-engine.md)). When on, it also fires for a row held in flight that long since its claim (BACKLOG #1611). | | `saturation_sustain_samples` | int | _unset_ | raise a `saturation` alert (BACKLOG #93, [ADR 0014 amendment](adr/0014-alerting-rules-engine.md)) when an outbound lane's pending backlog is **rising sustained** over this many consecutive samples — the queue **derivative** (ingest > drain), distinct from the absolute depth/age ceilings above. A bursty-but-**draining** lane (spike then fall) never fires; only a lane whose depth climbs monotonically does. **Unset (the default) = OFF** — deny-by-default/opt-in (it overlaps `buildup_max_oldest_seconds`'s age dimension). Floor of 2 (fewer can't tell a burst from sustained growth). Global-only for now; a per-outbound override is a documented follow-up (a `[[alerts.rules]]` `connection` glob with `transports = []` can suppress it for a known-bursty feed in the interim). | | `priority` | enum | `normal` | **global DR / priority tier default** for every connection (#61, [ADR 0048](adr/0048-third-tier-disaster-recovery-standby.md)). A connection declaring no `priority=` of its own inherits this; resolution order is per-connection override > this global default > the built-in `normal`. The total order is `critical > normal > low`, and the [`[dr]`](#dr--third-tier-disaster-recovery-standby) run-profile starts only connections whose resolved rank is at or above `[dr].priority_threshold`. It governs **when a connection runs**, never what it does. `normal` keeps every connection at the same tier, so a deployment that never enables DR is byte-unchanged. An unknown value fails config load. | @@ -1242,8 +1242,8 @@ Because the local diff is cheap and PHI-safe it is **on by default** (zero phone | `pooled_claim_lane_chunk` | int (1–500) | 256 | Pooled-only: max lanes batch-claimed per claim round-trip (clamped down to the backend store's chunk — SQLite 200, SS/PG 500). | | `pooled_max_processing_lanes` | int (≥1) | 256 | Pooled-only: max concurrently-processing lanes per stage (the decrypted-body / crash-exposure bound). | | `require_rcsi_for_pooled` | | | **Removed, and refused at load** in the file or as `MEFOR_PIPELINE_REQUIRE_RCSI_FOR_POOLED`, at either value (BACKLOG #2090, [ADR 0066 §12](adr/0066-pooled-stage-claimers.md)). A SQL Server store no longer opens with `READ_COMMITTED_SNAPSHOT` off, in any claim mode (BACKLOG #1628). Only `schema_management = "auto"` tries to turn it on first. A pooled start also checks RCSI and always fails closed if it is off, so there is no check left for this key to relax. To run pooled on SQL Server, turn `READ_COMMITTED_SNAPSHOT` on for the database. | -| `infra_fault_policy` | enum | `stop` | **Pooled T17 (infra/machinery-fault) handling** ([ADR 0070](adr/0070-t17-infra-fault-bound.md)). A store/handoff error — or any raise from **outside** the per-item body — is caught by the dispatcher's T17 handler, which always re-pends the faulting head at an exponential-capped backoff (collapsing the ~4×/s sweep spin). This key bounds a **persistent** such fault: `stop` (default) STOPs the head-of-line-blocked lane after `infra_fault_stop_after` consecutive zero-progress faults, reusing the `internal_error = stop` muscle (STOPPED phase + `connection_stopped` alert; reload / new work re-arms) and **never** dead-lettering the good message. `retry_forever` never STOPs — it retries the head at capped backoff forever and emits a throttled `lane_stuck` alert once the horizon is crossed (for a deliberately-unattended flaky-infra site). Reliability-core: **read once at engine construction — a `/config/reload` does NOT re-read it (restart to change).** | -| `infra_fault_stop_after` | int (≥1) | 10 | consecutive zero-progress T17 faults before a `stop`-policy lane transitions to STOPPED — and the stuck horizon at which `retry_forever`'s throttled `lane_stuck` alert first fires. Under the exponential backoff (capped by `infra_fault_backoff_cap`) 10 spans ~4 min of wall clock, so it is really a duration gate. The same count also STOPs a lane whose own dispatch kills its stage's claimer that many times in a row, with a `connection_stopped` alert, under **either** policy: each such death stalls every other lane on that claimer through the respawn backoff (BACKLOG #2074). A reload, a recovery broadcast, or an operator stop and start re-arms it with a fresh count. | +| `infra_fault_policy` | enum | `stop` | **Pooled T17 (infra/machinery-fault) handling** ([ADR 0070](adr/0070-t17-infra-fault-bound.md)). A store/handoff error — or any raise from **outside** the per-item body — is caught by the dispatcher's T17 handler, which always re-pends the faulting head at an exponential-capped backoff (collapsing the ~4×/s sweep spin). This key bounds a **persistent** such fault: `stop` (default) STOPs the head-of-line-blocked lane after `infra_fault_stop_after` consecutive zero-progress faults, reusing the `internal_error = stop` muscle (STOPPED phase + `connection_stopped` alert; reload / new work re-arms) and **never** dead-lettering the good message. `retry_forever` never STOPs — it retries the head at capped backoff forever and emits a throttled `lane_stuck` alert once the horizon is crossed (for a deliberately-unattended flaky-infra site). A T17 re-pend that itself fails holds the lane and retries, on a backoff from 1 s capped at 30 s that `infra_fault_backoff_cap` does not set (BACKLOG #1611). Under `stop` that hold ends at `infra_fault_stop_after` failed re-pends and the lane STOPs with `connection_stopped`, the row left in flight; under `retry_forever` it holds on and raises one `lane_stuck` there. A batching outbound's hold on its coalesced rows differs: under `stop` it ends at that count without a STOP of its own, and under `retry_forever` it raises no `lane_stuck`. At least a stop or pause of the outbound, the fail-closed halt when the application log cannot be written (on every stage), and another node taking the lead, also end a hold under either policy, with the row left in flight. Reliability-core: **read once at engine construction — a `/config/reload` does NOT re-read it (restart to change).** | +| `infra_fault_stop_after` | int (≥1) | 10 | consecutive zero-progress T17 faults before a `stop`-policy lane transitions to STOPPED — and the stuck horizon at which `retry_forever`'s throttled `lane_stuck` alert first fires. Under the exponential backoff (capped by `infra_fault_backoff_cap`) 10 spans ~4 min of wall clock, so it is really a duration gate. It is also the count of consecutive failed re-pends at which a lane held retrying them STOPs under `stop`, or raises its one `lane_stuck` under `retry_forever` (BACKLOG #1611); that hold's own backoff is capped at 30 s, so 10 failures span at least about 2.5 min of backoff, plus each failing write's own time, which a hung store can stretch to its command timeout. The same count also STOPs a lane whose own dispatch kills its stage's claimer that many times in a row, with a `connection_stopped` alert, under **either** policy: each such death stalls every other lane on that claimer through the respawn backoff (BACKLOG #2074). A reload, a recovery broadcast, or an operator stop and start re-arms it with a fresh count. | | `infra_fault_backoff_cap` | float (>0) | 60.0 | cap (seconds) on the T17 head re-pend backoff — base is the dispatcher's 1 s lane-error backoff, doubling per consecutive zero-progress fault. ~60 s picks a recovered dependency back up within about a minute while still collapsing the spin. | | `fuse_thread_hops` | bool | `false` | **Thread-hop fusion** ([ADR 0071](adr/0071-cut-executor-round-trips-b5.md) B5). **Reliability-core, default-OFF, SQL-Server-scoped:** when `true` **and** the store backend is SQL Server **and** `claim_mode = "pooled"`, each fused stage (INGRESS/ROUTED) runs its off-loop CPU stage (`route_only`/`transform_one`) together with its store handoff on a **single** dedicated-executor worker hop, collapsing a multi-statement aioodbc handoff into one executor→loop completion (the profiled per-completion async-marshaling wall). Provably no-op elsewhere: Postgres (asyncpg is loop-native) and SQLite (loop-affine handoff lock) keep the async path by construction and log "ignored", and a sync-handoff-pool open failure downgrades to the async path with a loud warning + a degraded gauge — never a lane outage. **Read once at engine construction (restart to change).** Env/harness A/B: `MEFOR_PIPELINE_FUSE_THREAD_HOPS`. | | `pooled_fusing_workers` | int (≥1) | 8 | worker count for **each** per-stage fusing executor (ADR 0071 B5). Every fused stage gets its own `ThreadPoolExecutor` of this width plus a matching-width dedicated synchronous pyodbc handoff pool (one connection per worker, so a fused hop never blocks acquiring). Small by default — a fused hop holds a worker across DB latency, so this *is* the fused-stage concurrency; it also clamps the fused stages' effective `pooled_max_processing_lanes` to ~2× this value, so the claimer doesn't reserve 256 slots for a handful of workers. Inert unless `fuse_thread_hops` is on. | diff --git a/messagefoundry/config/settings.py b/messagefoundry/config/settings.py index f1884523d..107092699 100644 --- a/messagefoundry/config/settings.py +++ b/messagefoundry/config/settings.py @@ -2259,8 +2259,11 @@ class PipelineSettings(_Section): # InternalErrorPolicy.STOP muscle (STOPPED phase + connection_stopped alert + reload/notify_work # re-arm), never dead-lettering the good message. "retry_forever" never STOPs — it retries the head # at capped backoff forever and emits a throttled lane_stuck alert once the horizon is crossed (for - # a deliberately-unattended flaky-infra site). Reliability-core + read ONCE at construction (a - # /config/reload does NOT re-read it — restart to change, exactly like claim_mode). + # a deliberately-unattended flaky-infra site). A pooled lane whose T17 re-pend itself fails HOLDS + # and retries it (BACKLOG #1611). Under "stop" the hold ends at infra_fault_stop_after failed + # re-pends and the lane STOPs as above; under "retry_forever" it holds on and raises one + # lane_stuck there. Reliability-core + read ONCE at construction (a /config/reload does NOT + # re-read it — restart to change, like claim_mode). infra_fault_policy: Literal["stop", "retry_forever"] = Field(default="stop") # Consecutive zero-progress T17 faults before a "stop"-policy lane transitions to STOPPED. Also the # "retry_forever" stuck horizon at which the throttled lane_stuck alert first fires. Under the @@ -5149,7 +5152,9 @@ class AlertSeverity(str, Enum): # noqa: UP042 "gcm_invocations", "update_available", # #30: a newer MessageFoundry version is pinned than is running (ADR 0026) "backup_failed", # #60 (ADR 0049): a scheduled/on-demand DR backup failed (snapshot/encrypt/verify) - "lane_stuck", # ADR 0070: a pooled lane is retrying a persistent infra fault forever (retry_forever) + # ADR 0070: a pooled lane is retrying a persistent infra fault forever (retry_forever), or is + # held retrying a failed re-pend of its claimed rows under it (BACKLOG #1611). + "lane_stuck", # BACKLOG #305 (ASVS 13.2.2): the store privilege preflight found the store principal # over-granted, or could not read it, at start. "store_privilege_warning", diff --git a/messagefoundry/pipeline/alert_sinks.py b/messagefoundry/pipeline/alert_sinks.py index e9f9746be..38c5aeb50 100644 --- a/messagefoundry/pipeline/alert_sinks.py +++ b/messagefoundry/pipeline/alert_sinks.py @@ -852,7 +852,8 @@ def queue_buildup(self, name: str, *, depth: int, oldest_age_seconds: float) -> def lane_stuck(self, name: str, *, detail: str) -> None: # ADR 0070 retry_forever: a pooled lane is retrying a persistent T17 infra fault at capped - # backoff (never STOPped). The shared _emit throttle keys on (type, connection), so an ongoing + # backoff (never STOPped), or held retrying a failed re-pend of its claimed rows (BACKLOG + # #1611). The shared _emit throttle keys on (type, connection), so an ongoing # stuck lane collapses to one notification per cooldown; detail is a PHI-free stage+streak reason. self._emit({"type": "lane_stuck", "connection": name, "detail": detail}) diff --git a/messagefoundry/pipeline/alerts.py b/messagefoundry/pipeline/alerts.py index f3e8c4fae..a1b769c63 100644 --- a/messagefoundry/pipeline/alerts.py +++ b/messagefoundry/pipeline/alerts.py @@ -229,7 +229,9 @@ def lane_stuck(self, name: str, *, detail: str) -> None: (auto-resolved on the next clean head via ADR-0044 durable state when wired). Distinct from :meth:`connection_stopped` (the ``stop``-policy terminal) so an operator can route a "still retrying, look at the dependency" signal apart from a halted lane. ``name`` is the stage-lane - label; ``detail`` is a PHI-free reason (stage + streak). Emitted by the ``StageDispatcher``.""" + label; ``detail`` is a PHI-free reason (stage + streak). Emitted by the ``StageDispatcher``. + Under ``retry_forever`` it also fires for a serializer HOLDING its lane while the re-pend of + its claimed rows keeps failing (BACKLOG #1611), which the ADR 0070 streak cannot reach.""" ... def message_stall(self, name: str, *, oldest_age_seconds: float) -> None: diff --git a/messagefoundry/pipeline/stage_dispatcher.py b/messagefoundry/pipeline/stage_dispatcher.py index 245bc031d..5d0da6e4e 100644 --- a/messagefoundry/pipeline/stage_dispatcher.py +++ b/messagefoundry/pipeline/stage_dispatcher.py @@ -48,7 +48,7 @@ import time from collections import deque from collections.abc import Awaitable, Callable -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from enum import Enum, auto from typing import Protocol @@ -69,6 +69,10 @@ # backoff (wiring_runner._WORKER_ERROR_BACKOFF_SECONDS). Only the faulting lane pauses; siblings and # the claimer are untouched. _LANE_ERROR_BACKOFF_SECONDS = 1.0 + +# BACKLOG #1611: a serializer whose re-pend of its claimed rows failed retries it on a doubling +# backoff from _LANE_ERROR_BACKOFF_SECONDS, capped here, holding its lane meanwhile. +_REPEND_RETRY_CAP_SECONDS = 30.0 # Claimer store-error backoff: the whole chunk's lanes return to READY and this claimer's partition # pauses ~this long (chunk-scoped; raise K to shrink the blast radius). ADR 0066 §11 item 1. _CLAIM_ERROR_BACKOFF_SECONDS = 1.0 @@ -92,6 +96,106 @@ def _capped_backoff(base: float, exp: int, cap: float) -> float: return min(float(base * (2 ** min(max(exp, 0), 20))), cap) +class HoldResult(Enum): + """How a hold on a failed re-pend of claimed rows ended (BACKLOG #1611): the pooled + :meth:`StageDispatcher._land_while_holding` and the per-lane + ``RegistryRunner._repend_claimed_on_fault``. Only ``STOPPED`` is a teardown, after which the + caller returns. Every other end sends no further retry of that write, leaves its rows + INFLIGHT, and lets the caller go on as it did before step 2 after its one failed write: a + per-lane worker loops, a pooled serializer runs its terminal transition, after at most one + tail release (``_run_lane`` says when).""" + + LANDED = auto() # the write landed + STOPPED = auto() # the engine is stopping: the rows stay INFLIGHT for the next start + ABANDONED = auto() # the lane is paused, at least by an operator or the #122 log halt + SUPERSEDED = auto() # another node may have led since the fault, so the rows may be its own + EXHAUSTED = auto() # infra_fault_policy = stop, at infra_fault_stop_after failures + + +@dataclass(frozen=True) +class RependFence: + """The leadership a held re-pend's RETRIES are gated on (BACKLOG #1611). The first write is + always sent (ADR 0157 C1); a retry only while this node leads under the lease it claimed the + rows under. Callers :meth:`capture` that lease before the claim's work runs, so a flap during + that work, or during a hung first write, is seen. + + The epoch bumps only when the lease changes owner, so a node that regains the lease after + another node led holds a newer one, and that node may have claimed these rows meanwhile: + ``reschedule_claimed`` is fenced only by ``status='inflight'``, so a retry would re-pend the + successor's claim, a duplicate send. :meth:`superseded` reads that. A node that was not the + leader at the capture cannot tell who led since, so it never retries after regaining the lease. + A ``None`` epoch while leading (``NullCoordinator``) keeps no fence. That is safe for a single + node; engine shards also run it, and a lane whose owning shard changes on a reload while a + hold runs is not fenced here either. + + What is left is the window between the check and the write, a flap inside one store round + trip. A store-level owner or epoch fence on ``reschedule_claimed`` is the full fix; it is not + filed yet.""" + + is_leader: Callable[[], bool] + current_epoch: Callable[[], int | None] + leading: bool + epoch: int | None + + @classmethod + def capture( + cls, is_leader: Callable[[], bool], current_epoch: Callable[[], int | None] + ) -> RependFence: + leading = is_leader() + return cls(is_leader, current_epoch, leading, current_epoch() if leading else None) + + def superseded(self) -> bool: + """Whether, leading now, this node may lead under a lease other than the one it held at + the fault.""" + return not self.leading or (self.epoch is not None and self.current_epoch() != self.epoch) + + def moved(self) -> bool: + """Whether another lease than the captured one is held now: proof another node led, where + :meth:`superseded` also counts a demotion that no successor may have followed yet.""" + if self.epoch is None: + return False + now = self.current_epoch() + return now is not None and now != self.epoch + + def may_write(self) -> bool: + """Whether a write sent now, after the first, is fenced as above.""" + return self.is_leader() and not self.superseded() + + +async def wait_to_retry( + delay: float, + *, + poll: float, + sleep: Callable[[float], Awaitable[bool]], + fence: RependFence, + abandon: Callable[[], bool] | None, + not_leader: Callable[[], object], +) -> HoldResult | None: + """Wait out one backoff of a held re-pend, then return None to send the retry, or how the + hold ends (BACKLOG #1611). ``sleep`` returns True on a stop. With an ``abandon`` predicate the + backoff is slept in ``poll`` steps, reading it after each, so a pause ends the hold within + about ``poll`` rather than after a backoff of up to 30 s. Not the leader, the hold waits on + in ``poll`` steps and calls ``not_leader`` once, to log it; leading again, the fence decides.""" + warned = False + while True: + left = delay + while True: + step = min(left, poll) if abandon is not None else left + if await sleep(step): + return HoldResult.STOPPED + if abandon is not None and abandon(): + return HoldResult.ABANDONED + left -= step + if left <= 0: + break + if fence.is_leader(): + return HoldResult.SUPERSEDED if fence.superseded() else None + if not warned: + not_leader() + warned = True + delay = poll + + class LaneResultKind(Enum): """How the injected ``process_item`` body resolved one claimed row — the pooled analog of the per_lane loop's control flow, carrying the park deadline the ``_ItemOutcome`` two-member enum @@ -359,6 +463,8 @@ def __init__( infra_fault_policy: str = "stop", infra_fault_stop_after: int = 10, infra_fault_backoff_cap: float = 60.0, + is_leader: Callable[[], bool] | None = None, + current_epoch: Callable[[], int | None] | None = None, ) -> None: assert claimers_per_stage >= 1 assert max_processing_lanes >= 1 @@ -400,6 +506,10 @@ def __init__( self._infra_fault_policy = infra_fault_policy self._infra_fault_stop_after = infra_fault_stop_after self._infra_fault_backoff_cap = infra_fault_backoff_cap + # Read by _run_lane's RependFence (BACKLOG #1611). None: a single node, always the leader, + # with no epoch to fence on. + self._is_leader: Callable[[], bool] = is_leader or (lambda: True) + self._current_epoch: Callable[[], int | None] = current_epoch or (lambda: None) self._states: dict[str, _LaneState] = {} self._claimers: list[_Claimer] = [ @@ -1289,35 +1399,90 @@ async def _run_lane(self, lane: str, items: list[OutboxItem]) -> None: ``release_claimed``'d (attempts restored, FIFO-neutral) on a park/stop/exception. The final state mutation runs synchronously after the (async) tail release so it is atomic on the loop (no double-dispatch window).""" + # BACKLOG #1611: the lease these rows were claimed under, read before the body runs, so a + # lease change during the body or the first re-pend still fences the retries. + fence = RependFence.capture(self._is_leader, self._current_epoch) outcome, tail_ids, head_reschedule = await self._drain_lane(lane, items) - # Fix A (ADR 0070): re-pend the T17 head with a DURABLE backoff BEFORE the tail release — a - # store write, best-effort (a failure leaves it INFLIGHT for reset_stale_inflight/the sweep). + # Fix A (ADR 0070): re-pend the T17 head with a DURABLE backoff BEFORE the tail release. # This is NOT release_claimed: it dates the head into the future so it reads not-due (collapsing # the ~4×/s spin) rather than past-due (which re-readies it at the sweep cadence). + # + # BACKLOG #1611: both writes are retried while this serializer still HOLDS the lane + # (PROCESSING, slot reserved), so no claim takes a later row past a stranded one. How a hold + # ends is a HoldResult. A stop is the teardown: the rows stay INFLIGHT for + # reset_stale_inflight at the next start, as a cancelled serializer's are, and this returns + # with no terminal transition. Any other end (a pause, an epoch change, the stop policy's + # bound) sends no further retry and runs the terminal transition below, as before step 2 + # after one failed write; the rows stay in flight and the in-flight watch pages them. + # + # After a head re-pend the tail's first write is no longer at fault time, and a successor + # may hold those rows. When the head landed, it is skipped only if another node's lease is + # held now; when the head's hold did not land, it is sent once, as before step 2, and only + # under the fence. _land_while_holding says the rest. + head = tail = HoldResult.LANDED if head_reschedule is not None: head_id, when = head_reschedule - try: - await self._store.reschedule_claimed([head_id], when) - except Exception: # noqa: BLE001 — recovery covers a failed reschedule; don't stall the lane - log.warning( - "StageDispatcher %s reschedule_claimed failed for lane %s head", - self._stage.value, - lane, - exc_info=True, - ) - # Async tail release BEFORE the synchronous transition. Best-effort: a release failure leaves - # the tail INFLIGHT for reset_stale_inflight/the sweep to recover — never blocks the transition. + # The head is dated `when`, and _drain_lane parks its RETRY outcome until that same + # date. Only a retry that runs past it dates the head from now instead, by the same + # escalated span, so it never lands past-due and fix A's backoff is kept; the lane then + # parks until the date actually written. + span = max(when - self._clock(), _LANE_ERROR_BACKOFF_SECONDS) + dated = when + + def reschedule_head() -> Awaitable[object]: + nonlocal dated + now = self._clock() + dated = when if now < when else now + span + return self._store.reschedule_claimed([head_id], dated) + + head = await self._land_while_holding( + lane, "reschedule_claimed of the head", reschedule_head, fence + ) + outcome = replace(outcome, park_until=dated) if tail_ids: - try: - await self._store.release_claimed(tail_ids) - except Exception: # noqa: BLE001 — recovery covers a failed release; do not stall the lane - log.warning( - "StageDispatcher %s release_claimed failed for lane %s (%d rows)", - self._stage.value, - lane, - len(tail_ids), - exc_info=True, - ) + release = functools.partial(self._store.release_claimed, tail_ids) + what = f"release_claimed of {len(tail_ids)} row(s)" + if head is HoldResult.LANDED: + # Not on leader loss alone: the demotion drain relies on this release (zero rows + # INFLIGHT at quiesce), as before step 2. Only a lease another node took skips it. + if head_reschedule is not None and fence.moved(): + tail = HoldResult.SUPERSEDED + else: + tail = await self._land_while_holding(lane, what, release, fence) + elif not fence.may_write(): + tail = HoldResult.SUPERSEDED + else: + tail = await self._write_once(lane, what, release, head) + unlanded = (head is not HoldResult.LANDED) + ( + len(tail_ids) if tail is not HoldResult.LANDED else 0 + ) + if HoldResult.STOPPED in (head, tail): + # No terminal transition, so the dispatcher's quiesce reads this lane as drained; say + # here that its rows did not come back. Counts only, never content. + log.warning( + "StageDispatcher %s lane %s: stopped while holding %d claimed row(s) whose re-pend " + "had not landed; they stay in flight for the next start or a successor's recovery", + self._stage.value, + lane, + unlanded, + ) + return + if unlanded: + log.warning( + "StageDispatcher %s lane %s: stopped holding %d claimed row(s) whose re-pend had " + "not landed (%s); they stay in flight until a store recovery path re-pends them, " + "at least the next start", + self._stage.value, + lane, + unlanded, + (head if head is not HoldResult.LANDED else tail).name.lower(), + ) + # The stop policy's bound on a hold: _apply_retry STOPs a lane only for a zero-progress infra + # fault at the bound, which a re-pend failing that often is. The streak is raised to the + # bound in the RETRY branch below, not here, so a pause that wins skips it untouched. + exhausted = HoldResult.EXHAUSTED in (head, tail) and outcome.kind is LaneResultKind.RETRY + if exhausted: + outcome = replace(outcome, is_infra_fault=True, made_progress=False) # --- synchronous terminal transition (NO await below) ------------------------------------- # S_lane (ARM 1): read the episode clock ONCE, here, before the outcome branch — re-reading it # inside each branch would silently fold the branch's own work into the episode and costs extra @@ -1365,6 +1530,9 @@ async def _run_lane(self, lane: str, items: list[OutboxItem]) -> None: elif outcome.kind is LaneResultKind.RETRY: # T15 # S_lane DROP: a RETRY head was re-pended, not delivered — no completed service. Occupancy. self._book_lane_episode(_episode_start_ns, _episode_end_ns, rows=0, service=False) + if exhausted: + # _apply_retry adds this fault, which reaches infra_fault_stop_after. + st.infra_error_streak = max(st.infra_error_streak, self._infra_fault_stop_after - 1) self._apply_retry(lane, st, outcome) else: # T16 STOP (content-policy STOP — the InternalErrorPolicy path, untouched by ADR 0070) # S_lane DROP: a STOPped lane rendered no completed service either. Occupancy. @@ -1429,6 +1597,151 @@ def _infra_backoff(self, streak: int) -> float: the count is really a duration gate.""" return _capped_backoff(_LANE_ERROR_BACKOFF_SECONDS, streak, self._infra_fault_backoff_cap) + def _page_held_lane(self, lane: str, what: str, failures: int) -> None: + """Raise ``lane_stuck`` once for a lane held by :meth:`_land_while_holding` under + ``retry_forever``: a held lane never reaches :meth:`_apply_retry`, so the ADR 0070 fix B + streak cannot page it. Marks the episode alerted, so :meth:`_maybe_lane_stuck_alert` does + not page it again once the write lands, and an episode already alerted (the head's hold, + then the tail's) is not paged again. A sink that raises is logged and swallowed, as + :meth:`_to_stopped` does, because the serializer still holds the lane and must not die on + an alert.""" + st = self._states.get(lane) + if st is not None: + if st.lane_stuck_alerted: + return + st.lane_stuck_alerted = True + try: + self._alert_sink.lane_stuck( + lane, + detail=( + f"{self._stage.value} lane held: {what} failed {failures} consecutive " + "times; retrying (retry_forever)" + ), + ) + except Exception: # noqa: BLE001 — an alert must not kill the serializer holding the lane + log.warning( + "StageDispatcher %s: lane_stuck alert for lane %s failed", + self._stage.value, + lane, + exc_info=True, + ) + + def _hold_abandoned(self, lane: str) -> bool: + """Whether a ``pause_lane`` waits on this lane, so its hold ends and the terminal transition + reaches PAUSED. On every stage: at least a stop or pause of an outbound (an operator's, a + schedule window's or a DR park's) pauses an OUTBOUND lane, and the #122 fail-closed halt + (application log unwritable) pauses the INGRESS, ROUTED and RESPONSE lanes of each halted + inbound (``RegistryRunner._pause_internal_lanes``).""" + st = self._states.get(lane) + return st is not None and st.pause_pending + + async def _slept_into_stop(self, seconds: float) -> bool: + await self._sleep_or_stop(seconds) + return self._stop.is_set() + + async def _write_once( + self, lane: str, what: str, write: Callable[[], Awaitable[object]], failed: HoldResult + ) -> HoldResult: + """Send ``write`` once, best-effort, as every re-pend was before BACKLOG #1611: LANDED, or + ``failed`` when it raises, logged.""" + try: + await write() + except Exception: # noqa: BLE001 — one best-effort write; the rows stay INFLIGHT + log.warning( + "StageDispatcher %s lane %s: %s failed; not retried", + self._stage.value, + lane, + what, + exc_info=True, + ) + return failed + return HoldResult.LANDED + + async def _land_while_holding( + self, + lane: str, + what: str, + write: Callable[[], Awaitable[object]], + fence: RependFence, + ) -> HoldResult: + """Run ``write``, a re-pend of rows this serializer claimed, until it lands, on a capped + backoff, while the caller holds the lane (BACKLOG #1611). Returns how the hold ended; only + ``STOPPED`` is a teardown, and :meth:`_run_lane` says what each end does. + + The caller still holds the lane, so nothing claims a later row past the ones being + re-pended, and no clock or ownership guess is needed: the serializer is the lane's only + consumer and knows which rows it holds. A failure is logged at WARNING on a backoff (1st, + 2nd, 4th ...) with a line when the write lands; counts and the exception only, never a + row's content. + + **The first write is always sent; a RETRY only under the fence** (:class:`RependFence` + says why), read after each backoff, right before it. Not the leader, the serializer sends + nothing, logs one WARNING, and keeps holding the lane, polling at the base interval: a stop + (the demotion teardown) returns ``STOPPED``, and leading again under the same lease with no + teardown (a flap) resumes the retry. Leading under another lease returns ``SUPERSEDED``. + + **A pause ends the hold** (``ABANDONED``). It is read at the base interval through the + backoff, so a pause, a stop of the outbound, the purge that waits on it and the #122 halt + are not held until the store heals. :meth:`_hold_abandoned` says what pauses a lane. + + **At ``infra_fault_stop_after`` consecutive failures the policy decides.** Under ``stop`` + the hold ends (``EXHAUSTED``) and :meth:`_run_lane` raises the lane's ``infra_error_streak`` + to the bound, so :meth:`_apply_retry` STOPs the lane with ``connection_stopped``, as a + persistent T17 fault does; a pending pause wins over that, as it does over a T17 STOP. Under ``retry_forever`` the hold goes on and raises one ``lane_stuck`` + (:meth:`_page_held_lane`).""" + run = FailureRun() + while True: + try: + await write() + except Exception as exc: # noqa: BLE001 — retried while holding the lane; see the docstring + run = run.record( + log, + exc, + "StageDispatcher %s lane %s: %s failed; holding the lane and retrying, so no " + "later row overtakes it", + self._stage.value, + lane, + what, + level=logging.WARNING, + ) + if run.count >= self._infra_fault_stop_after: + if self._infra_fault_policy == "stop": + return HoldResult.EXHAUSTED + if run.count == self._infra_fault_stop_after: + self._page_held_lane(lane, what, run.count) + else: + run.clear( # logs only after a failure + log, + "StageDispatcher %s lane %s: %s", + self._stage.value, + lane, + what, + level=logging.WARNING, + ) + return HoldResult.LANDED + # One WARNING per leadership loss; a loss after a regain in this retry warns again. + ended = await wait_to_retry( + _capped_backoff( + _LANE_ERROR_BACKOFF_SECONDS, run.count - 1, _REPEND_RETRY_CAP_SECONDS + ), + poll=_LANE_ERROR_BACKOFF_SECONDS, + sleep=self._slept_into_stop, + fence=fence, + abandon=functools.partial(self._hold_abandoned, lane), + not_leader=functools.partial( + log.warning, + "StageDispatcher %s lane %s: %s failed %d time(s) and this node is not the " + "leader, so no retry is sent; the lane stays held with the rows in flight, " + "and the retry resumes only if this node leads again before it stops", + self._stage.value, + lane, + what, + run.count, + ), + ) + if ended is not None: + return ended + async def _drain_lane( self, lane: str, items: list[OutboxItem] ) -> tuple[_LaneOutcome, list[str], tuple[str, float] | None]: @@ -1743,6 +2056,10 @@ def _record_empty(self, *, woken: bool) -> None: self._empty.record_empty(woken=woken) async def _sleep_or_stop(self, seconds: float) -> None: + # A set stop is read first: with ``seconds`` 0, wait_for cancels the wait before it runs and + # raises TimeoutError even on a set Event, so it would wait instead of reading the stop. + if self._stop.is_set(): + return try: # noqa: SIM105 await asyncio.wait_for(self._stop.wait(), timeout=seconds) except TimeoutError: diff --git a/messagefoundry/pipeline/wiring_runner.py b/messagefoundry/pipeline/wiring_runner.py index 09430ceb8..581c1c0a0 100644 --- a/messagefoundry/pipeline/wiring_runner.py +++ b/messagefoundry/pipeline/wiring_runner.py @@ -191,9 +191,14 @@ from messagefoundry.pipeline.saturation import SaturationDetector from messagefoundry.pipeline.sharding import owner_shard_of_destination from messagefoundry.pipeline.stage_dispatcher import ( + _REPEND_RETRY_CAP_SECONDS, + HoldResult, LaneItemResult, LaneResultKind, + RependFence, StageDispatcher, + _capped_backoff, + wait_to_retry, ) from messagefoundry.pipeline.sync_reply import SyncReplyMetrics, SyncReplyResolverImpl from messagefoundry.redaction import safe_exc, safe_text @@ -813,8 +818,8 @@ def _log_inflight_strand(stage: str, name: str, count: int, age: float) -> None: in flight rather than pending, so this line does. Lane name and counts only, never content.""" log.warning( "%d row(s) held in flight on %s lane %r, oldest claimed %.0fs ago with no handoff; " - "a stranded row stays in flight until a store recovery path re-pends it, at least a " - "restart", + "a worker that could not re-pend its row holds the lane and retries, and a row no live " + "worker holds stays in flight until a store recovery path re-pends it, at least a restart", count, stage, name, @@ -6254,6 +6259,8 @@ def _make_dispatcher(self, stage: Stage) -> StageDispatcher: infra_fault_policy=self._infra_fault_policy, infra_fault_stop_after=self._infra_fault_stop_after, infra_fault_backoff_cap=self._infra_fault_backoff_cap, + is_leader=self._coordinator.is_leader, + current_epoch=self._coordinator.current_epoch, ) async def _start_pooled_dispatchers(self) -> None: @@ -8044,6 +8051,11 @@ async def _capture_ack( # --- delivery path ------------------------------------------------------- + def _repend_fence(self) -> RependFence: + """The lease a worker claims under, read before its claim (BACKLOG #1611), so a lease + change during the pass, not only during the re-pend, fences the retries.""" + return RependFence.capture(self._coordinator.is_leader, self._coordinator.current_epoch) + async def _repend_claimed_on_fault( self, worker: str, @@ -8051,49 +8063,122 @@ async def _repend_claimed_on_fault( ids: Sequence[str], *, faults: _WorkerFaultLog | None = None, - ) -> None: - """Best-effort re-pend of the rows a per-lane worker was holding when its loop faulted - (BACKLOG #1611 step 1) — the per-lane analogue of the pooled T17 head re-pend in - :meth:`~messagefoundry.pipeline.stage_dispatcher.StageDispatcher._run_lane`. + fence: RependFence | None = None, + abandon: Callable[[], bool] | None = None, + give_up_after: int | None = None, + ) -> HoldResult: + """Re-pend the rows a per-lane worker was holding when its loop faulted, retrying while it + holds the lane (BACKLOG #1611 steps 1 and 2) — the per-lane analogue of the pooled T17 head + re-pend in :meth:`~messagefoundry.pipeline.stage_dispatcher.StageDispatcher._run_lane`. + Returns how the hold ended. Only ``STOPPED``, the runner stopping with the rows still held, + makes the caller return rather than claim again; every other end leaves the rows INFLIGHT + and the worker loops on, as it did before step 2 after one failed write. The claim is its own committed transaction, so a fault in the handoff that follows leaves the - claimed row INFLIGHT. Every claim path selects ``status='pending'``, so the surviving worker - never reconsiders it, and ``reset_stale_inflight`` runs from ``Engine.start()``, the - cluster promotion path, and a ``reload()`` for a worker that has RETURNED (ADR 0157 Inc 2) - -- which this surviving worker has not. Without this the row waits for a - service restart, overtaken by its successors, while ``pending_depth`` (pending rows only) - reports the lane healthy to the buildup and stall alerts. + claimed row INFLIGHT. Every claim path selects ``status='pending'``, so a worker that went on + claiming would take the row's successor and deliver it first, a per-lane FIFO break, while + ``pending_depth`` (pending rows only) reports the lane healthy. + + **So a failed re-pend is retried, on a capped backoff, while the worker HOLDS the lane**: it + claims nothing new until the stranded rows are pending again. That is what keeps per-lane + FIFO, and it needs no clock and no ownership guess, because the worker retrying is the lane's + only consumer (ADR 0059) and knows which rows it holds. A store that stays down keeps the + rows in flight, and the runner's in-flight watch pages them on their claim age + (:meth:`_check_inflight_strands`), which nothing here touches. + + **How the hold ends.** At least these, each sending no further write: + + * A runner stop returns ``STOPPED`` and leaves the rows for ``reset_stale_inflight`` at the + next start, as a cancelled worker's are; the backoff is :meth:`_stop_or_sleep`, and the + task stays cancellable. + * Leadership and the lease's epoch gate the retries, not the first write, through the + :class:`~messagefoundry.pipeline.stage_dispatcher.RependFence`, which says why; each + caller passes the one it read before its claim (:meth:`_repend_fence`). Not the + leader, the worker keeps holding the lane, so a stop returns ``STOPPED`` and a flap under + the same lease resumes the retry; leading under another lease returns ``SUPERSEDED``. + * ``abandon``, read at the base interval through each backoff, returns ``ABANDONED``. The + delivery worker and the pooled batch guard pass the negation of :meth:`_lane_may_deliver`, + so a stop or pause of the outbound and the #122 halt are not held until the store heals + (that method says how a schedule window or a DR park reaches it). The router, transform + and response workers pass :meth:`_inbound_log_halted`, the #122 gate their loop top + reads, so the halt ends their hold too and the worker returns at that gate. + * ``give_up_after`` failures return ``EXHAUSTED``. Only the pooled batch guard passes it, + under ``infra_fault_policy = stop``, since that hold is the pooled dispatcher's. The + per-lane workers never read the policy and have no STOP path, so they pass none. Dates the rows into the future by the same backoff the caller is about to sleep, exactly as T17 does and for the same reason: a plain release leaves them past-due, so the sweep re-readies them and a broken dependency is re-claimed ~4x/s. ``reschedule_claimed`` is guarded ``status='inflight'`` and FIFO-neutral (``seq`` is never re-minted), so rows the worker already resolved are left untouched and the survivors keep their lane position — which is why a caller - may pass its whole claimed batch without tracking which row died. - - Wrapped in its own ``except`` because a re-pend that raises inside an except arm would kill the - worker, which is strictly worse than the leak it fixes; a failure here simply falls back to the - pre-existing recovery (``reset_stale_inflight`` at the next start). - - A worker loop passes its ``faults`` (BACKLOG #1844), and a failure is then logged on that - worker's backoff rather than once per pass: a busy-timeout that fails the handoff tends to fail - this write on the same passes. + may pass its whole claimed batch without tracking which row died. ``reschedule_claimed`` also + undoes the claim's ``attempts`` increment, as it always has here, so ``max_attempts`` does NOT + bound a row whose handoff faults every pass: the lane stays head-of-line blocked on it, and + the in-flight watch and the worker's fault log are what show it. + + A failure is caught, never raised: a re-pend that raises inside an except arm would kill the + worker with the rows still in flight, which is strictly worse than the retry. It is logged at + WARNING on the worker's ``faults`` backoff (BACKLOG #1844), one line per 1st, 2nd, 4th ... + failure; that run closes, with a line, on the worker's quiet rule rather than when the write + lands. A caller without one gets a backoff of its own, which logs a line when the write lands. + Each line carries counts and the exception, never a row's content. """ if not ids: - return - try: - await self.store.reschedule_claimed( - list(ids), time.time() + _WORKER_ERROR_BACKOFF_SECONDS - ) - except Exception as exc: # noqa: BLE001 — recovery must never kill the worker; see the docstring - message = ( - "%s worker %r: reschedule_claimed failed for %d claimed row(s); they stay INFLIGHT " - "for reset_stale_inflight at the next start" - ) - if faults is None: - log.warning(message, worker, name, len(ids), exc_info=True) + return HoldResult.LANDED + rows = list(ids) + if fence is None: # read before the first write, which can hang across a flap + fence = self._repend_fence() + # Without a worker's `faults` (#1844), this call logs on its own backoff and closes it. + own = FailureRun() + failures = 0 + message = ( + "%s worker %r: reschedule_claimed failed for %d claimed row(s); holding the lane " + "and retrying, so no later row overtakes them" + ) + while True: + try: + await self.store.reschedule_claimed( + rows, time.time() + _WORKER_ERROR_BACKOFF_SECONDS + ) + except Exception as exc: # noqa: BLE001 — retried while holding the lane; see the docstring + failures += 1 + if faults is not None: + faults.repend_failed(exc, message, worker, name, len(rows)) + else: + own = own.record( + log, exc, message, worker, name, len(rows), level=logging.WARNING + ) + if give_up_after is not None and failures >= give_up_after: + return HoldResult.EXHAUSTED else: - faults.repend_failed(exc, message, worker, name, len(ids)) + # A worker's own `faults` closes the re-pend run on quiet, with its main run. + own.clear( + log, "%s worker %r: reschedule_claimed", worker, name, level=logging.WARNING + ) + return HoldResult.LANDED + # One WARNING per leadership loss; a loss after a regain in this retry warns again. + ended = await wait_to_retry( + _capped_backoff( + _WORKER_ERROR_BACKOFF_SECONDS, failures - 1, _REPEND_RETRY_CAP_SECONDS + ), + poll=_WORKER_ERROR_BACKOFF_SECONDS, + sleep=self._stop_or_sleep, + fence=fence, + abandon=abandon, + not_leader=functools.partial( + log.warning, + "%s worker %r: reschedule_claimed failed %d time(s) for %d claimed row(s) " + "and this node is not the leader, so no retry is sent; the lane stays held " + "with the rows in flight, and the retry resumes only if this node leads " + "again before it stops", + worker, + name, + failures, + len(rows), + ), + ) + if ended is not None: + return ended async def _release_tail_on_stop(self, worker: str, name: str, ids: Sequence[str]) -> None: """Best-effort release of the batch tail a per-lane worker still holds when it RETURNS on a @@ -8127,10 +8212,19 @@ async def _release_tail_on_stop(self, worker: str, name: str, ids: Sequence[str] exc_info=True, ) + def _inbound_log_halted(self, name: str) -> bool: + """Whether the #122 halt (ADR 0162) holds inbound ``name``'s internal stages: the gate the + router, transform and response workers read at their loop top, and the ``abandon`` + predicate they pass to a held re-pend (:meth:`_repend_claimed_on_fault`). A new loop-top + gate for those workers belongs here as well, or a held re-pend stays deaf to it.""" + return name in self._log_halted + def _lane_may_deliver(self, name: str) -> bool: """The two conditions :meth:`_delivery_worker`'s loop-top gates refuse a claim on (the #122 - halt and the operator pause), as one predicate for its per-item re-check (#2771). A new - loop-top gate belongs here as well, or a claimed batch keeps sending under it.""" + halt and the pause), as one predicate for its per-item re-check (#2771). A new loop-top + gate belongs here as well, or a claimed batch keeps sending under it. The pause is + :attr:`_outbound_paused`, which a schedule window or a DR park joins through the same + outbound stop an operator's takes; nothing here reads either directly.""" return not self._delivery_halted and name not in self._outbound_paused async def _delivery_worker(self, name: str) -> None: @@ -8148,6 +8242,7 @@ async def _delivery_worker(self, name: str) -> None: claimed: list[str] = [] faults = _WorkerFaultLog("delivery", name) # #1844: backs off the except arm's traceback while not self._stop.is_set(): + fence = self._repend_fence() # #1611: the lease this pass claims under # #122 (ADR 0189) THE CLAIM GATE, per_lane half: the application log is unwritable and this # process has fail-closed, so this lane must not deliver no matter which door started it. # Return BEFORE the claim so no row is left INFLIGHT — the same terminal state @@ -8281,7 +8376,16 @@ async def _delivery_worker(self, name: str) -> None: # #1844: log with a backoff, not a traceback per pass (see _WorkerFaultLog). faults.failed(exc, "delivery worker %r: unexpected error; backing off and retrying") # #1611: surviving is not enough — hand the claimed row back (see the helper). - await self._repend_claimed_on_fault("delivery", name, claimed, faults=faults) + held = await self._repend_claimed_on_fault( + "delivery", + name, + claimed, + faults=faults, + fence=fence, + abandon=lambda: not self._lane_may_deliver(name), + ) + if held is HoldResult.STOPPED: + return # stopped while holding them: they stay INFLIGHT for the next start claimed = [] if await self._stop_or_sleep(_WORKER_ERROR_BACKOFF_SECONDS): return @@ -8751,6 +8855,7 @@ async def _process_delivery_batch( guard appends the extras' ids to it, and the worker's arm re-pends head and extras in ONE write. Two writes for one failed pass would record twice on the worker's backoff, doubling its count and pinning the extras' line to positions the schedule never emits.""" + fence = self._repend_fence() # #1611: before the extras are claimed members: list[OutboxItem] = [head] try: return await self._deliver_coalesced_batch(name, cfg, members) @@ -8763,8 +8868,34 @@ async def _process_delivery_batch( extras = [it.id for it in members[1:]] if extras_out is not None: extras_out.extend(extras) # the caller's re-pend owns them now + # The batch's fault is re-raised however the hold on the extras ended; one that did not + # land leaves them INFLIGHT. That is the stop path too, not only the landed one: the head + # is the pooled dispatcher's, whose T17 arm sends its one permitted first write and then + # reads the same stop, so re-raising ends the pass at once. Returning a result here + # instead would tell the dispatcher the head resolved. This hold is the pooled + # dispatcher's, so it ends as that dispatcher's own holds do: on a pause of the lane, + # and under infra_fault_policy = stop at infra_fault_stop_after failures, after which + # the head's hold counts its own failures toward the lane's STOP. else: - await self._repend_claimed_on_fault("batch delivery", name, extras) + held = await self._repend_claimed_on_fault( + "batch delivery", + name, + extras, + fence=fence, + abandon=lambda: not self._lane_may_deliver(name), + give_up_after=( + self._infra_fault_stop_after if self._infra_fault_policy == "stop" else None + ), + ) + if held is not HoldResult.LANDED: + log.warning( + "delivery lane %r: stopped holding %d coalesced batch row(s) whose " + "re-pend had not landed (%s); they stay in flight until a store recovery " + "path re-pends them, at least the next start", + name, + len(extras), + held.name.lower(), + ) raise async def _deliver_coalesced_batch( @@ -9075,10 +9206,11 @@ async def _router_worker(self, name: str) -> None: claimed: list[str] = [] # #1611: rows this loop claimed, for the except arm's re-pend faults = _WorkerFaultLog("router", name) # #1844: backs off the except arm's traceback while not self._stop.is_set(): + fence = self._repend_fence() # #1611: the lease this pass claims under # #122 (ADR 0162): the application log is unwritable and this process has fail-closed. # Return BEFORE the claim so no row is left INFLIGHT — the same terminal state a # STOP-policy halt leaves, re-armed by restarting this inbound. - if name in self._log_halted: + if self._inbound_log_halted(name): return try: # FIFO per inbound: claim the due head (ingress rows never back off, so this is @@ -9124,7 +9256,16 @@ async def _router_worker(self, name: str) -> None: # back off, and keep going (mirrors the delivery worker). #1844: with a backoff. faults.failed(exc, "router worker %r: unexpected error; backing off and retrying") # #1611: surviving is not enough — hand the claimed row back (see the helper). - await self._repend_claimed_on_fault("router", name, claimed, faults=faults) + held = await self._repend_claimed_on_fault( + "router", + name, + claimed, + faults=faults, + fence=fence, + abandon=functools.partial(self._inbound_log_halted, name), + ) + if held is HoldResult.STOPPED: + return # stopped while holding them: they stay INFLIGHT for the next start claimed = [] if await self._stop_or_sleep(_WORKER_ERROR_BACKOFF_SECONDS): return @@ -9449,7 +9590,8 @@ async def _response_worker(self, name: str) -> None: claimed: list[str] = [] # #1611: the row this loop claimed, for the except arm's re-pend faults = _WorkerFaultLog("response", name) # #1844: backs off the except arm's traceback while not self._stop.is_set(): - if name in self._log_halted: # #122 (ADR 0162) — see _router_worker's gate + fence = self._repend_fence() # #1611: the lease this pass claims under + if self._inbound_log_halted(name): # #122 (ADR 0162) — see _router_worker's gate return try: item = await self.store.claim_next_fifo(name, stage=Stage.RESPONSE.value) @@ -9471,7 +9613,16 @@ async def _response_worker(self, name: str) -> None: # log, back off, keep going (mirrors the router/delivery workers). #1844: with a backoff. faults.failed(exc, "response worker %r: unexpected error; backing off and retrying") # #1611: surviving is not enough — hand the claimed row back (see the helper). - await self._repend_claimed_on_fault("response", name, claimed, faults=faults) + held = await self._repend_claimed_on_fault( + "response", + name, + claimed, + faults=faults, + fence=fence, + abandon=functools.partial(self._inbound_log_halted, name), + ) + if held is HoldResult.STOPPED: + return # stopped while holding them: they stay INFLIGHT for the next start claimed = [] if await self._stop_or_sleep(_WORKER_ERROR_BACKOFF_SECONDS): return @@ -9547,7 +9698,8 @@ async def _transform_worker(self, name: str) -> None: claimed: list[str] = [] # #1611: rows this loop claimed, for the except arm's re-pend faults = _WorkerFaultLog("transform", name) # #1844: backs off the except arm's traceback while not self._stop.is_set(): - if name in self._log_halted: # #122 (ADR 0162) — see _router_worker's gate + fence = self._repend_fence() # #1611: the lease this pass claims under + if self._inbound_log_halted(name): # #122 (ADR 0162) — see _router_worker's gate return try: # FIFO per inbound at the routed stage. Under active-passive HA the graph runs on the @@ -9597,7 +9749,16 @@ async def _transform_worker(self, name: str) -> None: exc, "transform worker %r: unexpected error; backing off and retrying" ) # #1611: surviving is not enough — hand the claimed row back (see the helper). - await self._repend_claimed_on_fault("transform", name, claimed, faults=faults) + held = await self._repend_claimed_on_fault( + "transform", + name, + claimed, + faults=faults, + fence=fence, + abandon=functools.partial(self._inbound_log_halted, name), + ) + if held is HoldResult.STOPPED: + return # stopped while holding them: they stay INFLIGHT for the next start claimed = [] if await self._stop_or_sleep(_WORKER_ERROR_BACKOFF_SECONDS): return @@ -10535,11 +10696,13 @@ async def _check_inflight_strands(self, now: float | None = None) -> None: **Why this exists.** A claim is its own committed transaction, so a store fault between it and the handoff leaves the row ``inflight``. Part A made the per-lane workers re-pend it, and the - pooled dispatcher already did (T17), but both re-pends are store writes that can fail too. - What is left then is recovered by ``reset_stale_inflight`` at the next start, and nothing - before that. It is invisible to ``pending_depth``, so the buildup and stall alerts read the - lane healthy, and those alerts only run from the lane's own processing path, which a strand - has stopped. This check runs on its own clock and reads in-flight rows directly. + pooled dispatcher already did (T17). Both re-pends are store writes that can fail too, so + since step 2 each retries its write while holding the lane. A store that stays down keeps + the row in flight for as long as it stays down, and a worker killed by a ``BaseException`` + leaves its row to ``reset_stale_inflight`` at the next start. Such a row is invisible to + ``pending_depth``, so the buildup and stall alerts read the lane healthy, and those alerts + only run from the lane's own processing path, which a strand has stopped. This check runs on + its own clock and reads in-flight rows directly, so it pages whichever recovery is waiting. **What it pages, on which threshold.** Every lane fires ``queue_buildup`` on the age its pending check uses: an outbound lane's resolved :class:`BuildupThreshold`, and the global @@ -10554,11 +10717,13 @@ async def _check_inflight_strands(self, now: float | None = None) -> None: past the threshold is either stranded or in a send or transform that has run that long, and an operator wants the page in both cases. - **It does not recover the row.** A reclaim would have to know which in-flight rows no live - task still holds, and the pooled dispatcher keeps a claimed prefix inside its own serializer - where this runner cannot see it. Re-pending a held row sends it twice. So this pages, and - the store's existing recovery paths still re-pend. They include at least - ``reset_stale_inflight`` at a single-node start and, on Postgres only, the expired-lease + **It does not recover the row.** A reclaim from here would have to know which in-flight rows + no live task still holds, and the pooled dispatcher keeps a claimed prefix inside its own + serializer where this runner cannot see it. Re-pending a held row sends it twice. So this + pages, and recovery stays with the holder retrying its own re-pend + (:meth:`_repend_claimed_on_fault`, the dispatcher's ``_run_lane``) and with the store's + paths. Those include at least ``reset_stale_inflight`` at a single-node start and, on + Postgres only, the expired-lease reclaim inside the FIFO head claims plus the clustered leader's ``reclaim_expired_leases`` sweep. An ``ordering=unordered`` lane claims through ``claim_ready``, which has no such reclaim. And on a FIFO lane the lease reclaim re-pends the row after its successors have diff --git a/tests/test_backlog1611_inflight_watch.py b/tests/test_backlog1611_inflight_watch.py index 187ae3482..9e28878a3 100644 --- a/tests/test_backlog1611_inflight_watch.py +++ b/tests/test_backlog1611_inflight_watch.py @@ -13,8 +13,14 @@ the two ``pooled`` rows of P3-03's measured table as failure-injection tests: * ``route_handoff`` raises once: T17 re-pends the head and the message routes, no restart. -* ``route_handoff`` AND ``reschedule_claimed`` each raise once: the row strands. It still strands (part - B ships no reclaim; see ``_check_inflight_strands``), and now the in-flight watch pages it. +* ``route_handoff`` AND ``reschedule_claimed`` each raise once: before step 2 the row stranded until + a restart, and the in-flight watch paged it. + +Step 2 part A recovers it with no restart and no clock: a holder whose re-pend fails retries it +while it holds the lane, leader-gated after the first write. The last tests show it in both claim +modes, in FIFO order, with a control that restores the one-shot re-pend and goes red; the leader +gate; and a prompt stop. A worker killed by a ``BaseException`` still strands its row until the next +start, as before; recovering it at the worker's respawn is part B, not here. """ from __future__ import annotations @@ -26,10 +32,16 @@ from collections.abc import AsyncIterator from pathlib import Path from typing import Any +from unittest.mock import ANY import pytest -from messagefoundry.config.models import BuildupThreshold, ConnectorType, StallThreshold +from messagefoundry.config.models import ( + BatchConfig, + BuildupThreshold, + ConnectorType, + StallThreshold, +) from messagefoundry.config.settings import EgressSettings from messagefoundry.config.wiring import ( ConnectionSpec, @@ -40,10 +52,12 @@ from messagefoundry.pipeline import stage_dispatcher, wiring_runner from messagefoundry.pipeline.alerts import LoggingAlertSink from messagefoundry.pipeline.cluster import NullCoordinator +from messagefoundry.pipeline.stage_dispatcher import HoldResult, RependFence from messagefoundry.pipeline.wiring_runner import RegistryRunner from messagefoundry.store import MessageStatus, MessageStore, Stage RAW = "MSH|^~\\&|A|B|C|D|20260101||ADT^A01|MSG1611B|P|2.5.1\r" +RAW_B = "MSH|^~\\&|A|B|C|D|20260101||ADT^A01|MSG1611C|P|2.5.1\r" @pytest.fixture @@ -90,18 +104,22 @@ def _registry(tmp_path: Path) -> Registry: return reg -async def _ingress_inflight(store: MessageStore) -> int: - """Counted with its own SQL, not through ``inflight_by_lane``, so a broken read cannot make a - stranded row look recovered.""" +async def _inflight_at(store: MessageStore, stage: Stage) -> int: + """``IB``'s in-flight rows at ``stage``, counted with its own SQL, not through + ``inflight_by_lane``, so a broken read cannot make a stranded row look recovered.""" cur = await store._db.execute( "SELECT COUNT(*) AS c FROM queue WHERE stage=? AND channel_id=? AND status=?", - (Stage.INGRESS.value, "IB", "inflight"), + (stage.value, "IB", "inflight"), ) row = await cur.fetchone() assert row is not None return int(row["c"]) +async def _ingress_inflight(store: MessageStore) -> int: + return await _inflight_at(store, Stage.INGRESS) + + async def _until(pred: Any, *, timeout: float = 5.0) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: @@ -445,50 +463,1215 @@ async def routed() -> bool: await runner.stop() -async def test_pooled_double_fault_strands_and_the_watch_pages_it( +@pytest.mark.parametrize("claim_mode", ["per_lane", "pooled"]) +async def test_a_store_that_stays_down_keeps_the_row_in_flight_and_the_watch_pages_it( store: MessageStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, + claim_mode: str, ) -> None: - """P3-03's second pooled row: ``route_handoff`` AND ``reschedule_claimed`` each raise once. T17's - own re-pend fails, so the row stays ``inflight``, and ``pending_depth`` reads the lane empty. + """``route_handoff`` raises once and ``reschedule_claimed`` keeps raising. The holder retries its + re-pend (step 2), and nothing it does hides the strand: the row stays in flight, the lane reads + empty to ``pending_depth``, and the in-flight watch pages it. Once the store heals, the retry + lands and the message routes with no restart and no reload. - It STILL strands after part B, by design: no reclaim shipped, because the pooled serializer holds - claimed rows this runner cannot see, and re-pending a held row sends it twice. What changed is that - the strand now pages. If a safe reclaim ever lands, this test is the one to flip: the row should - then route instead of staying in flight.""" - monkeypatch.setattr(stage_dispatcher, "_LANE_ERROR_BACKOFF_SECONDS", 0.01) - monkeypatch.setattr(wiring_runner, "_INFLIGHT_WATCH_INTERVAL_SECONDS", 0.05) + Before step 2 this was P3-03's second pooled row and stayed stranded until a restart.""" + _fast_backoffs(monkeypatch) calls: dict[str, int] = {} _fail_once(monkeypatch, store, "route_handoff", calls) - _fail_once(monkeypatch, store, "reschedule_claimed", calls) + down = _fail_while(monkeypatch, store, "reschedule_claimed", calls) sink = _RecordingSink() runner = RegistryRunner( _registry(tmp_path), store, alert_sink=sink, buildup_default=BuildupThreshold(max_oldest_seconds=0.3), - claim_mode="pooled", + claim_mode=claim_mode, pooled_sweep_interval=0.05, egress=EgressSettings(deny_by_default=False), ) - with caplog.at_level(logging.WARNING, logger=wiring_runner.log.name): + with caplog.at_level(logging.WARNING): await runner.start() try: mid = await store.enqueue_ingress(channel_id="IB", raw=RAW) - runner._dispatchers[Stage.INGRESS].mark_ready("IB") + runner.notify_work() async def paged() -> bool: return "held in flight on ingress lane 'IB'" in caplog.text await _until(paged) - # Both faults fired, once each, and nothing retried the row after them. - assert calls == {"route_handoff": 1, "reschedule_claimed": 1} - msg = await store.get_message(mid) - assert msg is not None and msg["status"] == MessageStatus.RECEIVED.value + assert calls["reschedule_claimed"] >= 2 # retried, not given up after one + assert await _status(store, mid) == MessageStatus.RECEIVED.value assert await _ingress_inflight(store) == 1 assert await store.pending_depth("IB", stage=Stage.INGRESS.value) == (0, None) assert any(name == "IB" and depth == 1 for name, depth, _ in sink.buildup) + assert "holding the lane and retrying" in caplog.text + down.clear() # the store heals + + async def routed() -> bool: + return await _status(store, mid) != MessageStatus.RECEIVED.value + + await _until(routed) + assert await _ingress_inflight(store) == 0 finally: await runner.stop() + + +# --- step 2: recover without a restart, and never re-pend a held row ------------------------------- + + +def _fast_backoffs(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(stage_dispatcher, "_LANE_ERROR_BACKOFF_SECONDS", 0.01) + monkeypatch.setattr(stage_dispatcher, "_REPEND_RETRY_CAP_SECONDS", 0.05) + monkeypatch.setattr(wiring_runner, "_WORKER_ERROR_BACKOFF_SECONDS", 0.01) + monkeypatch.setattr(wiring_runner, "_REPEND_RETRY_CAP_SECONDS", 0.05) + monkeypatch.setattr(wiring_runner, "_INFLIGHT_WATCH_INTERVAL_SECONDS", 0.05) + + +def _fail_while( + monkeypatch: pytest.MonkeyPatch, obj: Any, attr: str, calls: dict[str, int] +) -> asyncio.Event: + """``attr`` raises while the returned Event is set: a store that stays down until the test heals + it.""" + real = getattr(obj, attr) + down = asyncio.Event() + down.set() + + async def flaky(*a: Any, **k: Any) -> Any: + calls[attr] = calls.get(attr, 0) + 1 + if down.is_set(): + raise RuntimeError(f"simulated store outage in {attr}") + return await real(*a, **k) + + monkeypatch.setattr(obj, attr, flaky) + return down + + +async def _status(store: MessageStore, mid: str) -> str | None: + msg = await store.get_message(mid) + return None if msg is None else str(msg["status"]) + + +def _record_handoffs( + monkeypatch: pytest.MonkeyPatch, + store: MessageStore, + handed_off: list[str], + *, + before: Any = None, +) -> None: + """Wrap ``route_handoff``: run ``before(message_id)`` first (it may raise, or hold), then record + the message id of every handoff that COMMITTED, in commit order. Layered over any ``_fail_once`` + already installed.""" + real = store.route_handoff + + async def recording(*a: Any, **k: Any) -> Any: + if before is not None: + await before(k["message_id"]) + result = await real(*a, **k) + handed_off.append(k["message_id"]) + return result + + monkeypatch.setattr(store, "route_handoff", recording) + + +def _one_shot_repend(monkeypatch: pytest.MonkeyPatch) -> None: + """The CONTROL: put back the pre-step-2 behaviour, one re-pend attempt and then move on, in both + claim modes. A test that cannot tell this from the shipped code proves nothing.""" + + async def once_per_lane( + self: RegistryRunner, worker: str, name: str, ids: Any, **_k: Any + ) -> HoldResult: + with contextlib.suppress(Exception): + await self.store.reschedule_claimed( + list(ids), time.time() + wiring_runner._WORKER_ERROR_BACKOFF_SECONDS + ) + return HoldResult.LANDED + + async def once_pooled(self: Any, lane: str, what: str, write: Any, fence: Any) -> HoldResult: + with contextlib.suppress(Exception): + await write() + return HoldResult.LANDED + + monkeypatch.setattr(RegistryRunner, "_repend_claimed_on_fault", once_per_lane) + monkeypatch.setattr(stage_dispatcher.StageDispatcher, "_land_while_holding", once_pooled) + + +@pytest.mark.parametrize("claim_mode", ["per_lane", "pooled"]) +@pytest.mark.parametrize("variant", ["shipped", "control"]) +async def test_a_double_fault_routes_without_a_restart_and_nothing_overtakes_it( + store: MessageStore, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + claim_mode: str, + variant: str, +) -> None: + """``route_handoff`` AND the holder's re-pend each raise once, then succeed. Message A is the one + that faults; B arrives on the same lane while A is stranded. Shipped: the holder retries its + re-pend while it holds the lane, so A routes first, then B, with no restart and no reload. + + The ``control`` variant restores the one-shot re-pend this step replaced: A stays in flight and + B overtakes it, which is the defect.""" + _fast_backoffs(monkeypatch) + if variant == "control": + _one_shot_repend(monkeypatch) + calls: dict[str, int] = {} + _fail_once(monkeypatch, store, "reschedule_claimed", calls) + second: list[str] = [] + + async def fault_a_once(message_id: str) -> None: + calls["route_handoff"] = calls.get("route_handoff", 0) + 1 + if calls["route_handoff"] == 1: + # B lands while A is claimed, so only a lane that is held keeps B behind A. + second.append(await store.enqueue_ingress(channel_id="IB", raw=RAW_B)) + raise RuntimeError("simulated transient store fault in route_handoff") + + handed_off: list[str] = [] + _record_handoffs(monkeypatch, store, handed_off, before=fault_a_once) + runner = RegistryRunner( + _registry(tmp_path), + store, + claim_mode=claim_mode, + pooled_sweep_interval=0.05, + egress=EgressSettings(deny_by_default=False), + ) + await runner.start() + try: + a = await store.enqueue_ingress(channel_id="IB", raw=RAW) + runner.notify_work() + + async def b_routed() -> bool: + return bool(second) and second[0] in handed_off + + await _until(b_routed) + if variant == "control": + assert handed_off == [second[0]] # B overtook A + assert await _ingress_inflight(store) == 1 # and A is stranded + return + + async def a_routed() -> bool: + return a in handed_off + + await _until(a_routed) + assert handed_off == [a, second[0]] # FIFO: A, then B + assert calls["reschedule_claimed"] >= 2 # the first re-pend failed and was retried + assert await _ingress_inflight(store) == 0 + finally: + await runner.stop() + + +@pytest.mark.parametrize("claim_mode", ["per_lane", "pooled"]) +async def test_a_stop_during_the_retry_exits_promptly_and_leaves_the_row_in_flight( + store: MessageStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, claim_mode: str +) -> None: + """The re-pend keeps failing and its backoff is long (30 s). A stop must end the retry at once, + not after the backoff, and leave the row INFLIGHT for the next start's reset, as before.""" + monkeypatch.setattr(stage_dispatcher, "_LANE_ERROR_BACKOFF_SECONDS", 30.0) + monkeypatch.setattr(wiring_runner, "_WORKER_ERROR_BACKOFF_SECONDS", 30.0) + calls: dict[str, int] = {} + _fail_once(monkeypatch, store, "route_handoff", calls) + _fail_while(monkeypatch, store, "reschedule_claimed", calls) + runner = RegistryRunner( + _registry(tmp_path), + store, + claim_mode=claim_mode, + pooled_sweep_interval=0.05, + egress=EgressSettings(deny_by_default=False), + ) + await runner.start() + stopped = False + try: + await store.enqueue_ingress(channel_id="IB", raw=RAW) + runner.notify_work() + + async def retrying() -> bool: + return calls.get("reschedule_claimed", 0) >= 1 + + await _until(retrying) + await asyncio.sleep(0.1) # now inside the 30 s backoff + started = time.monotonic() + await asyncio.wait_for(runner.stop(), timeout=10.0) + stopped = True + assert time.monotonic() - started < 5.0 # far inside the 30 s backoff + assert calls["reschedule_claimed"] == 1 # no retry ran during the stop + assert await _ingress_inflight(store) == 1 + finally: + if not stopped: + await runner.stop() + + +@pytest.mark.parametrize("variant", ["shipped", "control-ignores-stop"]) +async def test_a_stop_during_the_batch_extras_hold_ends_the_pass_promptly( + store: MessageStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, variant: str +) -> None: + """The pooled batch guard re-pends the coalesced extras itself, now holding the lane until that + lands. A stop during its 30 s backoff must end the hold at once: the guard re-raises the batch's + fault, so the dispatcher's T17 arm still owns the head, and the extras stay in flight for the + next start. The control makes the backoff deaf to the stop; the pass then outlives it.""" + monkeypatch.setattr(wiring_runner, "_WORKER_ERROR_BACKOFF_SECONDS", 30.0) + runner = RegistryRunner( + _registry(tmp_path), store, egress=EgressSettings(deny_by_default=False) + ) + for _ in range(3): + await store.enqueue_message(channel_id="IB", raw=RAW, deliveries=[("OB", "p")], now=1.0) + head = await store.claim_next_fifo("OB", now=2.0) + assert head is not None + calls: dict[str, int] = {} + _fail_while(monkeypatch, store, "reschedule_claimed", calls) + + async def coalesce_then_fault(name: str, cfg: Any, items: list[Any]) -> Any: + for _ in range(2): + more = await store.claim_next_fifo(name, now=2.0) + assert more is not None + items.append(more) + raise RuntimeError("simulated store fault after coalescing") + + monkeypatch.setattr(runner, "_deliver_coalesced_batch", coalesce_then_fault) + if variant == "control-ignores-stop": + real_sleep = runner._stop_or_sleep + + async def deaf(delay: float) -> bool: + await asyncio.sleep(delay) + return await real_sleep(0.0) + + monkeypatch.setattr(runner, "_stop_or_sleep", deaf) + cfg = BatchConfig(max_count=3, max_wait_ms=1) + task = asyncio.ensure_future(runner._process_delivery_batch("OB", head, cfg)) + + async def holding() -> bool: + return calls.get("reschedule_claimed", 0) >= 1 + + await _until(holding) + runner._stop.set() + if variant == "control-ignores-stop": + done, _ = await asyncio.wait({task}, timeout=0.5) + assert not done # the hold outlived the stop: the case this test exists to catch + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + return + with pytest.raises(RuntimeError, match="simulated store fault after coalescing"): + await asyncio.wait_for(task, timeout=5.0) + assert calls["reschedule_claimed"] == 1 # no retry ran after the stop + assert await store.inflight_by_lane(stage=Stage.OUTBOUND.value) == {"OB": (3, 2.0)} + + +# --- step 2: a node that is not the leader sends no re-pend RETRY ----------------------------------- +# StageDispatcher._land_while_holding's docstring says why; the first write is always sent. + + +class _Lease(NullCoordinator): + """A coordinator whose leadership the test flips.""" + + def __init__(self) -> None: + super().__init__() + self.held = True + + def is_leader(self) -> bool: + return self.held + + +async def _claimed_ingress_row(store: MessageStore) -> str: + await store.enqueue_ingress(channel_id="IB", raw=RAW, now=1.0) + item = await store.claim_next_fifo("IB", now=2.0, stage=Stage.INGRESS.value) + assert item is not None + return item.id + + +def _bare_dispatcher(store: MessageStore, **kwargs: Any) -> stage_dispatcher.StageDispatcher: + return stage_dispatcher.StageDispatcher( + Stage.INGRESS, + store, + process_item=None, # type: ignore[arg-type] # the retry under test never runs a body + lane_provider=set, + per_lane_limit=1, + **kwargs, + ) + + +# Each case: (leader at the start, lose the lease during the first write, what the 3rd backoff does). +_LEASE_CASES = { + "control-kept": (True, False, None), + "lost-during-the-write-then-stop": (True, True, "stop"), + "not-the-leader-from-the-start-then-stop": (False, False, "stop"), + "lost-then-regained-with-no-teardown": (True, True, "regain"), +} + + +@pytest.mark.parametrize("case", list(_LEASE_CASES)) +async def test_the_pooled_repend_retry_sends_nothing_while_not_the_leader( + store: MessageStore, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + case: str, +) -> None: + """The serializer's first re-pend fails, the store then heals. Kept lease (the control): the + retry lands after one backoff. Not the leader, from the start or from the failing write on: + the first write is still sent (ADR 0157 C1), then no retry while it lasts, one WARNING, and the + stop the demotion teardown sends returns False with the row in flight. Leading again with no + teardown resumes the retry, which lands.""" + _fast_backoffs(monkeypatch) + leader, lose_in_write, third_backoff = _LEASE_CASES[case] + lease = _Lease() + lease.held = leader + dispatcher = _bare_dispatcher(store, is_leader=lease.is_leader) + row = await _claimed_ingress_row(store) + writes = 0 + + async def repend() -> None: + nonlocal writes + writes += 1 + if lose_in_write: + lease.held = False + if writes == 1: + raise RuntimeError("simulated store outage in release_claimed") + await store.release_claimed([row]) + + backoffs = 0 + real_sleep = dispatcher._sleep_or_stop + + async def backoff(seconds: float) -> None: + nonlocal backoffs + backoffs += 1 + if backoffs == 3 and third_backoff == "regain": + lease.held = True + elif backoffs == 3 and third_backoff == "stop": + dispatcher._stop.set() + await real_sleep(seconds) + + monkeypatch.setattr(dispatcher, "_sleep_or_stop", backoff) + fence = RependFence.capture(lease.is_leader, lease.current_epoch) + with caplog.at_level(logging.WARNING): + held = await dispatcher._land_while_holding( + "IB", "release_claimed of 1 row(s)", repend, fence + ) + held_lines = caplog.text.count("this node is not the leader") + landed, stopped = HoldResult.LANDED, HoldResult.STOPPED + if case == "control-kept": + assert (held, writes, backoffs, held_lines) == (landed, 2, 1, 0) + elif third_backoff == "regain": + assert (held, writes, backoffs, held_lines) == (landed, 2, 3, 1) + else: + assert (held, writes, backoffs, held_lines) == (stopped, 1, 3, 1) + assert await _ingress_inflight(store) == (0 if held is landed else 1) + + +@pytest.mark.parametrize( + "case", + [ + "control-kept", + "lost-during-the-backoff-then-stop", + "not-the-leader-from-the-start-then-stop", + "lost-then-regained", + ], +) +async def test_the_per_lane_repend_retry_sends_nothing_while_not_the_leader( + store: MessageStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, case: str +) -> None: + """The per-lane twin. The first re-pend fails, on a non-leader too, and the store heals during + the backoff after it. Kept lease (the control): the retry lands. Not the leader after that + backoff: no retry is sent, and the teardown's stop returns True, so the caller returns rather + than claiming past the held row. Lease regained with no teardown: the retry resumes and lands.""" + _fast_backoffs(monkeypatch) + lease = _Lease() + lease.held = case != "not-the-leader-from-the-start-then-stop" + runner = RegistryRunner( + _registry(tmp_path), + store, + coordinator=lease, + egress=EgressSettings(deny_by_default=False), + ) + row = await _claimed_ingress_row(store) + calls: dict[str, int] = {} + down = _fail_while(monkeypatch, store, "reschedule_claimed", calls) + real_sleep = runner._stop_or_sleep + backoffs = 0 + + async def backoff(delay: float) -> bool: + nonlocal backoffs + backoffs += 1 + if backoffs == 1: + down.clear() # the store heals + lease.held = case == "control-kept" + elif backoffs == 3: + if case == "lost-then-regained": + lease.held = True + else: + runner._stop.set() + return await real_sleep(delay) + + monkeypatch.setattr(runner, "_stop_or_sleep", backoff) + held = await runner._repend_claimed_on_fault("router", "IB", [row]) + if case.endswith("-then-stop"): + assert (held, calls["reschedule_claimed"], backoffs) == (HoldResult.STOPPED, 1, 3) + assert await _ingress_inflight(store) == 1 + else: + expected_backoffs = 1 if case == "control-kept" else 3 + assert (held, calls["reschedule_claimed"], backoffs) == ( + HoldResult.LANDED, + 2, + expected_backoffs, + ) + assert await _ingress_inflight(store) == 0 + + +async def test_the_runner_hands_its_coordinators_leadership_to_every_dispatcher( + store: MessageStore, tmp_path: Path +) -> None: + """``_make_dispatcher`` wires the coordinator's ``is_leader`` and ``current_epoch``, read live, + so a demotion or a new lease reaches a pooled serializer's retry. A dispatcher built without + them is a single node: always the leader, with no epoch.""" + lease = _EpochLease() + runner = RegistryRunner( + _registry(tmp_path), + store, + claim_mode="pooled", + coordinator=lease, + egress=EgressSettings(deny_by_default=False), + ) + for stage in Stage: + dispatcher = runner._make_dispatcher(stage) + lease.held = True + assert dispatcher._is_leader() is True + lease.epoch = 12 + assert dispatcher._current_epoch() == 12 + lease.held = False + assert dispatcher._is_leader() is False + assert dispatcher._current_epoch() is None + assert _bare_dispatcher(store)._is_leader() is True + assert _bare_dispatcher(store)._current_epoch() is None + + +# --- step 2: the stop, the park date, the warning ------------------------------------------ + + +async def _ingress_status(store: MessageStore, mid: str) -> str | None: + cur = await store._db.execute( + "SELECT status FROM queue WHERE stage=? AND message_id=?", (Stage.INGRESS.value, mid) + ) + row = await cur.fetchone() + return None if row is None else str(row["status"]) + + +async def test_a_stop_already_set_is_read_before_a_zero_delay(store: MessageStore) -> None: + """On Python 3.14 ``wait_for(event.wait(), 0)`` raises ``TimeoutError`` even on a SET event, + so a zero-delay sleep would report no stop. The dispatcher's ``_sleep_or_stop`` reads a set + stop first, so a held lane's retry ends at once on a stop.""" + dispatcher = _bare_dispatcher(store) + dispatcher._stop.set() + await asyncio.wait_for(dispatcher._sleep_or_stop(0.0), timeout=1.0) + + +def _lane_run( + store: MessageStore, monkeypatch: pytest.MonkeyPatch, head: str, tail: list[str], when: float +) -> tuple[stage_dispatcher.StageDispatcher, list[Any]]: + """A bare dispatcher whose serializer drained to a T17 RETRY: re-pend ``head`` dated ``when`` + and release ``tail``. Returns it and the list each applied RETRY outcome lands in.""" + dispatcher = _bare_dispatcher(store) + outcome = stage_dispatcher._LaneOutcome( + stage_dispatcher.LaneResultKind.RETRY, when, is_infra_fault=True + ) + + async def drained(lane: str, items: Any) -> Any: + return outcome, tail, (head, when) + + applied: list[Any] = [] + monkeypatch.setattr(dispatcher, "_drain_lane", drained) + monkeypatch.setattr(dispatcher, "_apply_retry", lambda lane, st, out: applied.append(out)) + dispatcher._states["IB"] = stage_dispatcher._LaneState() + return dispatcher, applied + + +@pytest.mark.parametrize("leader", [True, False]) +async def test_a_stop_during_the_head_retry_still_sends_the_tail_release_once( + store: MessageStore, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + leader: bool, +) -> None: + """The head's re-pend keeps failing and a stop lands during its backoff. On the leader the + tail release is still sent once, and lands: the tail is pending again for the next start, the + head stays in flight for its reset. Before, the stop returned first and the tail stayed in + flight too. Not the leader, the late write is not sent, since a successor may hold the rows. + Either way a WARNING names the rows left in flight, which quiesce cannot see.""" + await store.enqueue_ingress(channel_id="IB", raw=RAW, now=1.0) + await store.enqueue_ingress(channel_id="IB", raw=RAW_B, now=1.0) + head, tail = await store.claim_next_fifo_batch( + "IB", stage=Stage.INGRESS.value, limit=2, now=2.0 + ) + calls: dict[str, int] = {} + _fail_while(monkeypatch, store, "reschedule_claimed", calls) + released: list[list[str]] = [] + real_release = store.release_claimed + + async def recording_release(ids: list[str]) -> Any: + released.append(list(ids)) + return await real_release(ids) + + monkeypatch.setattr(store, "release_claimed", recording_release) + dispatcher, applied = _lane_run(store, monkeypatch, head.id, [tail.id], time.time() + 60.0) + + async def stop_now(seconds: float) -> None: + dispatcher._stop.set() + + monkeypatch.setattr(dispatcher, "_sleep_or_stop", stop_now) + monkeypatch.setattr(dispatcher, "_is_leader", lambda: leader) + with caplog.at_level(logging.WARNING): + await dispatcher._run_lane("IB", [head, tail]) + assert calls["reschedule_claimed"] == 1 + assert released == ([[tail.id]] if leader else []) + expected_tail = "pending" if leader else "inflight" + assert await _ingress_status(store, tail.message_id) == expected_tail + assert await _ingress_status(store, head.message_id) == "inflight" + assert applied == [] # stopped: no terminal transition + held = 1 if leader else 2 + assert f"stopped while holding {held} claimed row(s)" in caplog.text + + +async def test_a_held_lane_pages_lane_stuck_once_at_the_infra_fault_bound( + store: MessageStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """Under ``retry_forever`` a held lane never reaches ``_apply_retry``, so the ADR 0070 streak + cannot page it. The hold raises one ``lane_stuck`` at ``infra_fault_stop_after`` consecutive + failures, not one per failure after it, and goes on holding.""" + _fast_backoffs(monkeypatch) + sink = _RecordingSink() + stuck: list[str] = [] + monkeypatch.setattr(sink, "lane_stuck", lambda name, *, detail: stuck.append(detail)) + dispatcher = _bare_dispatcher( + store, alert_sink=sink, infra_fault_stop_after=3, infra_fault_policy="retry_forever" + ) + fence = RependFence.capture(dispatcher._is_leader, dispatcher._current_epoch) + landed = HoldResult.LANDED + failures = 0 + + async def failing() -> None: + nonlocal failures + failures += 1 + if failures <= 5: + raise RuntimeError("simulated store outage") + + dispatcher._states["IB"] = stage_dispatcher._LaneState() + assert await dispatcher._land_while_holding("IB", "release_claimed", failing, fence) is landed + assert failures == 6 + assert len(stuck) == 1 and "failed 3 consecutive times" in stuck[0] + # Marked alerted, so _apply_retry's retry_forever page does not repeat it this episode. + assert dispatcher._states["IB"].lane_stuck_alerted is True + + def broken(name: str, *, detail: str) -> None: + raise RuntimeError("simulated alert sink fault") + + # A second hold in the same episode (the tail's, after the head's) does not page again. + failures = 0 + assert await dispatcher._land_while_holding("IB", "release_claimed", failing, fence) is landed + assert len(stuck) == 1 + + monkeypatch.setattr(sink, "lane_stuck", broken) + failures = 0 + dispatcher._states["IB"].lane_stuck_alerted = False # a new episode + # A raising sink is logged, never raised: the serializer still holds the lane. + assert await dispatcher._land_while_holding("IB", "release_claimed", failing, fence) is landed + + +async def test_a_head_retry_that_overruns_its_date_parks_the_lane_until_the_date_written( + store: MessageStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """The head's re-pend fails once and the retry lands after ``when`` has passed, so the head is + dated from now instead. The lane parks until that date, not the old ``when``, which would wake + it at once for an empty claim.""" + _fast_backoffs(monkeypatch) + await store.enqueue_ingress(channel_id="IB", raw=RAW, now=1.0) + head = await store.claim_next_fifo("IB", stage=Stage.INGRESS.value, now=2.0) + assert head is not None + calls: dict[str, int] = {} + _fail_once(monkeypatch, store, "reschedule_claimed", calls) + when = time.time() + 0.05 + dispatcher, applied = _lane_run(store, monkeypatch, head.id, [], when) + real_sleep = dispatcher._sleep_or_stop + + async def overrun(seconds: float) -> None: + await real_sleep(0.1) # past `when` + + monkeypatch.setattr(dispatcher, "_sleep_or_stop", overrun) + await dispatcher._run_lane("IB", [head]) + assert calls["reschedule_claimed"] == 2 + (outcome,) = applied + cur = await store._db.execute("SELECT next_attempt_at FROM queue WHERE id=?", (head.id,)) + row = await cur.fetchone() + assert row is not None + assert outcome.park_until == pytest.approx(row["next_attempt_at"]) + assert outcome.park_until > when + + +async def test_a_second_leadership_loss_in_one_retry_warns_again( + store: MessageStore, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """Pooled and per-lane: the re-pend fails, the lease is lost (one WARNING), regained (the retry + is sent and fails again), and lost again. The second loss warns too; before, it was silent. + Every backoff is one base step, so each sleep below is one backoff.""" + _fast_backoffs(monkeypatch) + monkeypatch.setattr(stage_dispatcher, "_REPEND_RETRY_CAP_SECONDS", 0.01) + monkeypatch.setattr(wiring_runner, "_REPEND_RETRY_CAP_SECONDS", 0.01) + lease = _Lease() + dispatcher = _bare_dispatcher(store, is_leader=lease.is_leader) + backoffs = 0 + + def flip(stop: asyncio.Event) -> None: + nonlocal backoffs + backoffs += 1 + if backoffs in (1, 3): + lease.held = False + elif backoffs == 2: + lease.held = True + elif backoffs == 4: + stop.set() + + async def failing() -> None: + raise RuntimeError("simulated store outage") + + real_sleep = dispatcher._sleep_or_stop + + async def pooled_backoff(seconds: float) -> None: + flip(dispatcher._stop) + await real_sleep(seconds) + + monkeypatch.setattr(dispatcher, "_sleep_or_stop", pooled_backoff) + with caplog.at_level(logging.WARNING): + fence = RependFence.capture(lease.is_leader, lease.current_epoch) + held = await dispatcher._land_while_holding("IB", "release_claimed", failing, fence) + assert held is HoldResult.STOPPED + assert caplog.text.count("this node is not the leader") == 2 + + caplog.clear() + backoffs = 0 + lease.held = True + runner = RegistryRunner( + _registry(tmp_path), store, coordinator=lease, egress=EgressSettings(deny_by_default=False) + ) + row = await _claimed_ingress_row(store) + _fail_while(monkeypatch, store, "reschedule_claimed", {}) + real_runner_sleep = runner._stop_or_sleep + + async def per_lane_backoff(delay: float) -> bool: + flip(runner._stop) + return await real_runner_sleep(delay) + + monkeypatch.setattr(runner, "_stop_or_sleep", per_lane_backoff) + with caplog.at_level(logging.WARNING): + held = await runner._repend_claimed_on_fault("router", "IB", [row]) + assert held is HoldResult.STOPPED + assert caplog.text.count("this node is not the leader") == 2 + + +# --- step 2: how a hold ends short of landing (the Lander's hold on PR 2574, F1 F3 F12) ----------- + + +class _EpochLease(_Lease): + """A lease with an epoch, bumped by the test when another node takes it. ``None`` while not + held, as the clustered coordinators return after a demotion.""" + + def __init__(self) -> None: + super().__init__() + self.epoch = 7 + + def current_epoch(self) -> int | None: + return self.epoch if self.held else None + + +@pytest.mark.parametrize("variant", ["shipped", "control-ignores-the-epoch"]) +async def test_a_pooled_retry_after_another_node_led_is_not_sent( + store: MessageStore, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + variant: str, +) -> None: + """F1. The head's re-pend fails and this node loses the lease during it. Another node leads + (the epoch moves) and this node leads again with no teardown, under the newer epoch. The + successor may have claimed these rows, so the retry is not sent, the tail release is skipped, + and the serializer runs its terminal transition rather than leave the lane held. The control + ignores the epoch: the retry is sent, the duplicate-send path the fence closes.""" + _fast_backoffs(monkeypatch) + if variant == "control-ignores-the-epoch": + monkeypatch.setattr(RependFence, "superseded", lambda self: False) + monkeypatch.setattr(RependFence, "moved", lambda self: False) + await store.enqueue_ingress(channel_id="IB", raw=RAW, now=1.0) + await store.enqueue_ingress(channel_id="IB", raw=RAW_B, now=1.0) + head, tail = await store.claim_next_fifo_batch( + "IB", stage=Stage.INGRESS.value, limit=2, now=2.0 + ) + lease = _EpochLease() + real_reschedule = store.reschedule_claimed + writes = 0 + + async def flapping_write(ids: list[str], when: float) -> Any: + nonlocal writes + writes += 1 + if writes == 1: + lease.held = False # demoted while the first write hangs, then it fails + raise RuntimeError("simulated store outage in reschedule_claimed") + return await real_reschedule(ids, when) + + monkeypatch.setattr(store, "reschedule_claimed", flapping_write) + released: list[list[str]] = [] + real_release = store.release_claimed + + async def recording_release(ids: list[str]) -> Any: + released.append(list(ids)) + return await real_release(ids) + + monkeypatch.setattr(store, "release_claimed", recording_release) + dispatcher, applied = _lane_run(store, monkeypatch, head.id, [tail.id], time.time() + 60.0) + monkeypatch.setattr(dispatcher, "_is_leader", lease.is_leader) + monkeypatch.setattr(dispatcher, "_current_epoch", lease.current_epoch) + real_sleep = dispatcher._sleep_or_stop + sleeps = 0 + + async def backoff(seconds: float) -> None: + nonlocal sleeps + sleeps += 1 + if sleeps == 2: # another node led meanwhile; this one leads again, under a newer epoch + lease.held, lease.epoch = True, 8 + await real_sleep(seconds) + + monkeypatch.setattr(dispatcher, "_sleep_or_stop", backoff) + with caplog.at_level(logging.WARNING): + await asyncio.wait_for(dispatcher._run_lane("IB", [head, tail]), timeout=5.0) + assert len(applied) == 1 # the terminal transition ran: the lane is not left held + if variant == "control-ignores-the-epoch": + assert writes == 2 # the retry re-pended rows the successor may hold + assert released == [[tail.id]] + return + assert writes == 1 + assert released == [] + assert await _ingress_status(store, head.message_id) == "inflight" + assert await _ingress_status(store, tail.message_id) == "inflight" + assert "stopped holding 2 claimed row(s)" in caplog.text and "(superseded)" in caplog.text + + +@pytest.mark.parametrize("variant", ["shipped", "control-ignores-the-epoch"]) +async def test_a_per_lane_retry_after_another_node_led_is_not_sent( + store: MessageStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, variant: str +) -> None: + """F1, per-lane. As the pooled case: the worker sends no retry after leading again under a + newer epoch, and returns ``SUPERSEDED`` so its loop goes on, as it did after one failed write. + A node that was not the leader at the fault cannot tell who led since, and sends none either.""" + _fast_backoffs(monkeypatch) + if variant == "control-ignores-the-epoch": + monkeypatch.setattr(RependFence, "superseded", lambda self: False) + lease = _EpochLease() + runner = RegistryRunner( + _registry(tmp_path), store, coordinator=lease, egress=EgressSettings(deny_by_default=False) + ) + row = await _claimed_ingress_row(store) + calls: dict[str, int] = {} + down = _fail_while(monkeypatch, store, "reschedule_claimed", calls) + real_sleep = runner._stop_or_sleep + sleeps = 0 + + async def backoff(delay: float) -> bool: + nonlocal sleeps + sleeps += 1 + if sleeps == 1: + down.clear() + lease.held = False + elif sleeps == 2: + lease.held, lease.epoch = True, 8 + return await real_sleep(delay) + + monkeypatch.setattr(runner, "_stop_or_sleep", backoff) + held = await asyncio.wait_for( + runner._repend_claimed_on_fault("router", "IB", [row]), timeout=5.0 + ) + if variant == "control-ignores-the-epoch": + assert (held, calls["reschedule_claimed"]) == (HoldResult.LANDED, 2) + return + assert (held, calls["reschedule_claimed"]) == (HoldResult.SUPERSEDED, 1) + assert await _ingress_inflight(store) == 1 + + # Not the leader at the fault: no epoch was captured, so regaining any lease sends nothing. + lease.held, lease.epoch, sleeps = False, 9, 1 + down.set() + held = await asyncio.wait_for( + runner._repend_claimed_on_fault("router", "IB", [row]), timeout=5.0 + ) + assert (held, calls["reschedule_claimed"]) == (HoldResult.SUPERSEDED, 2) + + +@pytest.mark.parametrize("claim_mode", ["per_lane", "pooled"]) +@pytest.mark.parametrize("variant", ["shipped", "control-deaf-to-the-pause"]) +async def test_a_pause_ends_a_held_delivery_lane_so_purge_is_not_refused( + store: MessageStore, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + claim_mode: str, + variant: str, +) -> None: + """F3. The delivery of an outbound row faults and its re-pend keeps failing, so the lane is + held on a 30 s backoff. An operator stop of the outbound must end the hold within about the + base interval, not when the store heals: the lane quiesces (``outbound_quiesced``, the purge + precondition) within 2 s, PAUSED in the pooled claim mode, and the per-lane worker keeps + running, parked at its pause gate. The row stays in flight, as one failed write left it before + step 2. The control removes the pause check, and the hold outlives the 2 s.""" + _fast_backoffs(monkeypatch) + monkeypatch.setattr(stage_dispatcher, "_LANE_ERROR_BACKOFF_SECONDS", 0.05) + monkeypatch.setattr(wiring_runner, "_WORKER_ERROR_BACKOFF_SECONDS", 0.05) + # Every backoff 30 s, so only the pause can end the hold inside the test. + monkeypatch.setattr(stage_dispatcher, "_capped_backoff", lambda base, exp, cap: 30.0) + monkeypatch.setattr(wiring_runner, "_capped_backoff", lambda base, exp, cap: 30.0) + if variant == "control-deaf-to-the-pause": + monkeypatch.setattr( + stage_dispatcher.StageDispatcher, "_hold_abandoned", lambda self, lane: False + ) + real_repend = RegistryRunner._repend_claimed_on_fault + + async def deaf(self: RegistryRunner, *a: Any, abandon: Any = None, **k: Any) -> Any: + return await real_repend(self, *a, **k) + + monkeypatch.setattr(RegistryRunner, "_repend_claimed_on_fault", deaf) + runner = RegistryRunner( + _registry(tmp_path), + store, + claim_mode=claim_mode, + pooled_sweep_interval=0.05, + egress=EgressSettings(deny_by_default=False), + ) + + async def faulting_delivery(name: str, item: Any) -> Any: + raise RuntimeError("simulated store fault in the delivery handoff") + + monkeypatch.setattr(runner, "_process_delivery_item", faulting_delivery) + calls: dict[str, int] = {} + _fail_while(monkeypatch, store, "reschedule_claimed", calls) + await store.enqueue_message(channel_id="IB", raw=RAW, deliveries=[("OB", "p")], now=1.0) + await runner.start() + try: + runner.notify_work() + + async def holding() -> bool: + return calls.get("reschedule_claimed", 0) >= 1 + + await _until(holding) + await asyncio.sleep(0.2) # inside the 30 s backoff + await runner.stop_outbound("OB") + + async def quiesced() -> bool: + return runner.outbound_quiesced("OB") + + if variant == "control-deaf-to-the-pause": + with pytest.raises(AssertionError, match="timed out"): + await _until(quiesced, timeout=2.0) + return + await _until(quiesced, timeout=2.0) + assert calls["reschedule_claimed"] == 1 # the hold sent no retry after the pause + assert await store.inflight_by_lane(stage=Stage.OUTBOUND.value) == {"OB": (1, ANY)} + if claim_mode == "pooled": + dispatcher = runner._dispatchers[Stage.OUTBOUND] + assert dispatcher._states["OB"].phase is stage_dispatcher._LanePhase.PAUSED + else: + assert not runner._workers["OB"].done() # the worker lives on, at its pause gate + finally: + await runner.stop() + + +@pytest.mark.parametrize("claim_mode", ["per_lane", "pooled"]) +@pytest.mark.parametrize( + ("stage", "handoff", "kind"), + [ + (Stage.INGRESS, "route_handoff", "router"), + (Stage.ROUTED, "transform_handoff", "transform"), + ], +) +@pytest.mark.parametrize("variant", ["shipped", "control-deaf-to-the-halt"]) +async def test_the_log_halt_ends_a_held_internal_lane( + store: MessageStore, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + claim_mode: str, + stage: Stage, + handoff: str, + kind: str, + variant: str, +) -> None: + """The #122 fail-closed halt (application log unwritable) pauses a router or transform lane in + the pooled claim mode and returns the per-lane worker at its loop-top gate. A lane HELD on a + failed re-pend must read that halt too, or it keeps retrying under it until the store heals. + The handoff faults and the re-pend keeps failing on a 30 s backoff; the halt must end the hold + within 2 s: the pooled lane reaches PAUSED, the per-lane worker returns cleanly. The row stays + in flight, as one failed write left it before step 2. The control passes no abandon predicate + to the backoff, which was the shipped code for these lanes, and the hold outlives the 2 s.""" + _fast_backoffs(monkeypatch) + monkeypatch.setattr(stage_dispatcher, "_LANE_ERROR_BACKOFF_SECONDS", 0.05) + monkeypatch.setattr(wiring_runner, "_WORKER_ERROR_BACKOFF_SECONDS", 0.05) + # Every backoff 30 s, so only the halt can end the hold inside the test. + monkeypatch.setattr(stage_dispatcher, "_capped_backoff", lambda base, exp, cap: 30.0) + monkeypatch.setattr(wiring_runner, "_capped_backoff", lambda base, exp, cap: 30.0) + if variant == "control-deaf-to-the-halt": + real_wait = stage_dispatcher.wait_to_retry + + async def deaf(delay: float, *, abandon: Any, **k: Any) -> Any: + return await real_wait(delay, abandon=None, **k) + + monkeypatch.setattr(stage_dispatcher, "wait_to_retry", deaf) + monkeypatch.setattr(wiring_runner, "wait_to_retry", deaf) + calls: dict[str, int] = {} + _fail_while(monkeypatch, store, handoff, calls) + _fail_while(monkeypatch, store, "reschedule_claimed", calls) + runner = RegistryRunner( + _registry(tmp_path), + store, + claim_mode=claim_mode, + pooled_sweep_interval=0.05, + egress=EgressSettings(deny_by_default=False), + ) + await runner.start() + try: + await store.enqueue_ingress(channel_id="IB", raw=RAW) + runner.notify_work() + + async def holding() -> bool: + return calls.get("reschedule_claimed", 0) >= 1 + + await _until(holding) + await asyncio.sleep(0.2) # inside the 30 s backoff + runner._halt_inbound_processing(["IB"]) + if claim_mode == "pooled": + states = runner._dispatchers[stage]._states + + async def ended() -> bool: + return states["IB"].phase is stage_dispatcher._LanePhase.PAUSED + + else: + worker = runner._inbound_worker_dict(kind)["IB"] + + async def ended() -> bool: + return worker.done() + + if variant == "control-deaf-to-the-halt": + with pytest.raises(AssertionError, match="timed out"): + await _until(ended, timeout=2.0) + return + await _until(ended, timeout=2.0) + assert calls["reschedule_claimed"] == 1 # the hold sent no retry after the halt + assert await _inflight_at(store, stage) == 1 + if claim_mode == "per_lane": + assert worker.exception() is None # returned at its halt gate, not crashed + finally: + await runner.stop() + + +class _StopSink(LoggingAlertSink): + def __init__(self) -> None: + self.stopped: list[str] = [] + self.stuck: list[str] = [] + + def connection_stopped(self, name: str, *, detail: str) -> None: + self.stopped.append(detail) + + def lane_stuck(self, name: str, *, detail: str) -> None: + self.stuck.append(detail) + + +@pytest.mark.parametrize("made_progress", [False, True]) +@pytest.mark.parametrize("policy", ["stop", "retry_forever"]) +async def test_the_stop_policy_bounds_a_held_lane_and_retry_forever_pages_it( + store: MessageStore, monkeypatch: pytest.MonkeyPatch, policy: str, made_progress: bool +) -> None: + """F12. The head's re-pend keeps failing. Under ``stop`` (the default) the hold ends at + ``infra_fault_stop_after`` failures and the lane reaches STOPPED with ``connection_stopped``, + as a persistent T17 fault does, and no ``lane_stuck``, whatever progress the prefix made before + the fault. Under ``retry_forever`` it pages ``lane_stuck`` once and holds on until the store + heals, then parks; it never STOPs.""" + _fast_backoffs(monkeypatch) + await store.enqueue_ingress(channel_id="IB", raw=RAW, now=1.0) + head = await store.claim_next_fifo("IB", stage=Stage.INGRESS.value, now=2.0) + assert head is not None + calls: dict[str, int] = {} + down = _fail_while(monkeypatch, store, "reschedule_claimed", calls) + sink = _StopSink() + own_stops: list[str] = [] + dispatcher = _bare_dispatcher( + store, + alert_sink=sink, + infra_fault_policy=policy, + infra_fault_stop_after=3, + on_lane_stopped=own_stops.append, + call_later=asyncio.get_running_loop().call_later, # a park arms its timer + ) + outcome = stage_dispatcher._LaneOutcome( + stage_dispatcher.LaneResultKind.RETRY, + time.time() + 60.0, + is_infra_fault=True, + made_progress=made_progress, + ) + + async def drained(lane: str, items: Any) -> Any: + return outcome, [], (head.id, time.time() + 60.0) + + monkeypatch.setattr(dispatcher, "_drain_lane", drained) + st = stage_dispatcher._LaneState(phase=stage_dispatcher._LanePhase.PROCESSING) + dispatcher._states["IB"] = st + if policy == "retry_forever": + real_sleep = dispatcher._sleep_or_stop + + async def heal_after_five(seconds: float) -> None: + if calls["reschedule_claimed"] >= 5: + down.clear() + await real_sleep(seconds) + + monkeypatch.setattr(dispatcher, "_sleep_or_stop", heal_after_five) + await asyncio.wait_for(dispatcher._run_lane("IB", [head]), timeout=5.0) + if policy == "stop": + assert calls["reschedule_claimed"] == 3 + assert st.phase is stage_dispatcher._LanePhase.STOPPED + assert len(sink.stopped) == 1 and own_stops == ["IB"] + assert sink.stuck == [] + assert await _ingress_status(store, head.message_id) == "inflight" + else: + assert calls["reschedule_claimed"] == 6 + assert st.phase is stage_dispatcher._LanePhase.PARKED + dispatcher._cancel_timer("IB") + assert sink.stopped == [] and own_stops == [] + assert len(sink.stuck) == 1 and "failed 3 consecutive times" in sink.stuck[0] + assert await _ingress_status(store, head.message_id) == "pending" + + +@pytest.mark.parametrize("variant", ["shipped", "control-unfenced-tail"]) +async def test_the_tail_release_after_a_landed_head_is_fenced( + store: MessageStore, monkeypatch: pytest.MonkeyPatch, variant: str +) -> None: + """The head's re-pend hangs while another node takes the lease, then commits. The tail + release that follows is no longer at fault time, and the successor may hold those rows, so it + is not sent. The control lets it through, the duplicate-send path.""" + _fast_backoffs(monkeypatch) + if variant == "control-unfenced-tail": + monkeypatch.setattr(RependFence, "moved", lambda self: False) + await store.enqueue_ingress(channel_id="IB", raw=RAW, now=1.0) + await store.enqueue_ingress(channel_id="IB", raw=RAW_B, now=1.0) + head, tail = await store.claim_next_fifo_batch( + "IB", stage=Stage.INGRESS.value, limit=2, now=2.0 + ) + lease = _EpochLease() + real_reschedule = store.reschedule_claimed + + async def slow_write(ids: list[str], when: float) -> Any: + lease.epoch = 8 # another node took the lease while the write hung; it then commits + return await real_reschedule(ids, when) + + monkeypatch.setattr(store, "reschedule_claimed", slow_write) + released: list[list[str]] = [] + + async def recording_release(ids: list[str]) -> Any: + released.append(list(ids)) + + monkeypatch.setattr(store, "release_claimed", recording_release) + dispatcher, applied = _lane_run(store, monkeypatch, head.id, [tail.id], time.time() + 60.0) + monkeypatch.setattr(dispatcher, "_is_leader", lease.is_leader) + monkeypatch.setattr(dispatcher, "_current_epoch", lease.current_epoch) + await asyncio.wait_for(dispatcher._run_lane("IB", [head, tail]), timeout=5.0) + assert len(applied) == 1 + assert released == ([[tail.id]] if variant == "control-unfenced-tail" else []) + + +async def test_a_lease_change_during_the_body_fences_the_retry( + store: MessageStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """The pooled fence is read before the body runs, not at the fault: a lease change while the + body ran, then a failed first re-pend, sends no retry.""" + _fast_backoffs(monkeypatch) + head_id = await _claimed_ingress_row(store) + lease = _EpochLease() + calls: dict[str, int] = {} + _fail_while(monkeypatch, store, "reschedule_claimed", calls) + dispatcher, applied = _lane_run(store, monkeypatch, head_id, [], time.time() + 60.0) + monkeypatch.setattr(dispatcher, "_is_leader", lease.is_leader) + monkeypatch.setattr(dispatcher, "_current_epoch", lease.current_epoch) + drained = dispatcher._drain_lane + + async def body_spans_a_flap(lane: str, items: Any) -> Any: + lease.epoch = 8 + return await drained(lane, items) + + monkeypatch.setattr(dispatcher, "_drain_lane", body_spans_a_flap) + await asyncio.wait_for(dispatcher._run_lane("IB", []), timeout=5.0) + assert calls["reschedule_claimed"] == 1 + assert len(applied) == 1 + + +async def test_a_pause_that_wins_over_the_stop_bound_leaves_the_streak_alone( + store: MessageStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """Under ``stop`` the hold reaches its bound just as an operator pause lands. The lane goes + PAUSED, not STOPPED, and its infra-fault streak is untouched, so after a resume the next single + T17 fault does not STOP it at once.""" + _fast_backoffs(monkeypatch) + head_id = await _claimed_ingress_row(store) + sink = _StopSink() + paused: list[str] = [] + dispatcher = stage_dispatcher.StageDispatcher( + Stage.OUTBOUND, + store, + process_item=None, # type: ignore[arg-type] # the drain is stubbed + lane_provider=set, + per_lane_limit=1, + alert_sink=sink, + infra_fault_policy="stop", + infra_fault_stop_after=3, + on_lane_paused=paused.append, + ) + st = stage_dispatcher._LaneState(phase=stage_dispatcher._LanePhase.PROCESSING) + dispatcher._states["OB"] = st + outcome = stage_dispatcher._LaneOutcome( + stage_dispatcher.LaneResultKind.RETRY, time.time() + 60.0, is_infra_fault=True + ) + + async def drained(lane: str, items: Any) -> Any: + return outcome, [], (head_id, time.time() + 60.0) + + monkeypatch.setattr(dispatcher, "_drain_lane", drained) + writes = 0 + + async def failing(ids: list[str], when: float) -> Any: + nonlocal writes + writes += 1 + if writes == 3: + dispatcher.pause_lane("OB") # PROCESSING, so it is marked pending + raise RuntimeError("simulated store outage in reschedule_claimed") + + monkeypatch.setattr(store, "reschedule_claimed", failing) + await asyncio.wait_for(dispatcher._run_lane("OB", []), timeout=5.0) + assert writes == 3 + assert st.phase is stage_dispatcher._LanePhase.PAUSED and paused == ["OB"] + assert sink.stopped == [] + assert st.infra_error_streak == 0 + + +async def test_a_demoted_node_still_releases_the_tail_after_its_head_lands( + store: MessageStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """The demotion drain: the lease is lost, and no other node holds one yet, while the head's + re-pend runs; it lands. The tail release is still sent, so the drain leaves zero rows in + flight, as before step 2. Only another node's lease, held now, skips it.""" + _fast_backoffs(monkeypatch) + await store.enqueue_ingress(channel_id="IB", raw=RAW, now=1.0) + await store.enqueue_ingress(channel_id="IB", raw=RAW_B, now=1.0) + head, tail = await store.claim_next_fifo_batch( + "IB", stage=Stage.INGRESS.value, limit=2, now=2.0 + ) + lease = _EpochLease() + real_reschedule = store.reschedule_claimed + + async def demoted_during_the_write(ids: list[str], when: float) -> Any: + lease.held = False # demoted: no lease, and no epoch, while the write runs + return await real_reschedule(ids, when) + + monkeypatch.setattr(store, "reschedule_claimed", demoted_during_the_write) + dispatcher, applied = _lane_run(store, monkeypatch, head.id, [tail.id], time.time() + 60.0) + monkeypatch.setattr(dispatcher, "_is_leader", lease.is_leader) + monkeypatch.setattr(dispatcher, "_current_epoch", lease.current_epoch) + await asyncio.wait_for(dispatcher._run_lane("IB", [head, tail]), timeout=5.0) + assert len(applied) == 1 + assert await _ingress_status(store, tail.message_id) == "pending" + assert await _ingress_status(store, head.message_id) == "pending" diff --git a/tests/test_log_backoff.py b/tests/test_log_backoff.py index ebc83061f..0d68cac13 100644 --- a/tests/test_log_backoff.py +++ b/tests/test_log_backoff.py @@ -490,6 +490,10 @@ async def test_a_failing_re_pend_in_the_same_arm_backs_off_too( """Under a busy timeout the handoff and the re-pend fail on the same passes, so throttling only the first would leave a traceback per pass beside it.""" caplog.set_level(logging.INFO, logger=_RUNNER_LOGGER) + # The held re-pend sleeps its backoff in base-interval steps, reading the #122 halt after each + # (#1611). At this file's 1 ms base a doubling backoff is hundreds of steps, each rounded up to + # the OS timer's tick (about 15 ms on Windows), so cap it to keep ten failures inside the wait. + monkeypatch.setattr(wiring_runner, "_REPEND_RETRY_CAP_SECONDS", 0.004) store = _RepeatFaultNoRependStore() runner = _runner(store) @@ -652,7 +656,8 @@ async def coalesce_then_fail(name: str, cfg: Any, items: list[OutboxItem]) -> An finally: await _stop(runner, task) - # One re-pend per failed pass, naming every row the pass held. + # Every re-pend write, the pass's first and each retry while it holds the lane (#1611), is ONE + # write naming every row the pass held, never head and extras apart. assert all(ids == ["head", "extra-1", "extra-2"] for ids in store.reschedules) repends = [ r