Hold the lane and retry a failed re-pend, no restart needed (BACKLOG #1611 step 2, part A)

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`)

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. |

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",

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})
 

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: