diff --git a/docs/CLUSTERING.md b/docs/CLUSTERING.md index 1cc7ade1..d9152593 100644 --- a/docs/CLUSTERING.md +++ b/docs/CLUSTERING.md @@ -101,9 +101,14 @@ cluster coordinates the parts that must not double-run or interleave: accepting new inbound work and initiating new processing. (The hard guarantee against *concurrent double-processing of a given row* is the **self-fencing leadership lease** + the leader-gated graph: the graph runs only on the leader, and a partitioned/slow old leader self-fences and lets its - leadership lease **expire** before a standby can acquire leadership — so by the time a promoted node - acts, the old leader has provably stopped. The store's **row leases** are the additional backstop for - the *recurring background* reclaim sweep, which only takes rows whose lease has **expired**.) Clients + leadership lease **expire** before a standby can acquire leadership. Read that as *the old leader has + stopped calling itself leader*, **not** as *the old leader has stopped* — fencing flips an in-memory + flag; the listeners and any in-flight sends wind down on their own schedule, which nothing budgets + against the fence-to-expiry margin. A promoted node can therefore briefly overlap a predecessor that + is still finishing a send. That is bounded to duplicate delivery, which at-least-once permits and + idempotent outbounds absorb — it is not a route to losing or stranding a message. The store's **row + leases** are the additional backstop for the *recurring background* reclaim sweep, which only takes + rows whose lease has **expired**.) Clients reconnect to whichever node is currently primary via a **floating VIP / load-balancer health check** (see the deployment doc). On promotion the new leader recovers the prior leader's stranded in-flight rows immediately — an *owner-scoped, lease-blind* on-promotion recovery (it re-pends only rows owned by @@ -115,9 +120,13 @@ cluster coordinates the parts that must not double-run or interleave: **leader**. The leader renews the lease every `heartbeat_seconds` (to `DB_now + leader_lease_ttl_seconds`, measured on the database's own clock, so node clock skew doesn't affect who may hold it); a standby acquires only once that lease has **expired**. A leader that cannot renew within - `leader_fence_timeout_seconds` (< the TTL) **self-fences** — it stops acting as leader before the lease - can expire and a standby acquire it, so a network-partitioned old leader never double-processes - (the split-brain guard). On a clean stop the leader expires its lease so a standby takes over at once. + `leader_fence_timeout_seconds` (< the TTL) **self-fences** — it stops *reporting itself* leader before + the lease can expire and a standby acquire it (the split-brain guard). Two caveats if you tune these + down from the shipped `10 / 20 / 30`: the usable margin is smaller than `ttl - fence`, because the + fence baseline is taken after the renew round trip returns while the expiry is stamped on the database + clock at statement execution, and detection lands up to one fence tick late; and the config validator + checks the *ordering* `heartbeat < fence < ttl` only, never that any margin survives. On a clean stop + the leader expires its lease so a standby takes over at once. - **Store-checked leader epoch (fencing token).** The self-fence above is *temporal* — it relies on a paused/partitioned old leader noticing it has fallen behind and demoting itself before the lease TTL elapses. As a **second, durable** backstop the `leader_lease` row also carries a monotonic @@ -131,7 +140,10 @@ cluster coordinates the parts that must not double-run or interleave: nothing. The current leader's held epoch equals the lease epoch, so it claims normally; per-lane FIFO is unaffected (the guard only ever *rejects* a stale claim, never reorders a valid one). This is a **server-DB-only** safeguard (Postgres / SQL Server); SQLite is a single active node, so its - `set_leader_epoch` is a no-op and the claim is byte-identical. The migration that adds the column is + `set_leader_epoch` is a no-op and the claim is byte-identical. **Scope:** the guard is attached to the + claim and to nothing else — it stops a superseded ex-leader *claiming*, not *writing*. Disposition + writes (done / failed / dead-letter) resolve their row by id with no epoch check, so an ex-leader that + is still finishing a send can still record that row's outcome. The migration that adds the column is additive (`ADD COLUMN IF NOT EXISTS` / a guarded `ALTER`, run under the DDL lock), so an in-place upgrade of a live cluster is safe; the column back-fills to `0` and the first fresh acquire after the upgrade bumps it to `1`. @@ -182,9 +194,17 @@ false`, `is_leader: true`, `role: "single-node"`, `config_version: 0`: ### `GET /cluster/nodes` — all nodes + the derived leader -`leader_node_id` is the single **live** leader; a crashed ex-leader whose row still carries the leader -flag is filtered out by a freshness check (`last_seen` within `node_timeout_seconds`), so it is never -reported as the leader. +`leader_node_id` is the derived leader: among rows still carrying the leader flag, the freshest one +whose `last_seen` falls within `node_timeout_seconds`. At most one node is ever reported. + +**It is not authoritative, and it is clock-sensitive — use `lease_owner` instead when it matters.** +The freshness test compares the *reading* node's wall clock against a `last_seen` written by the +*beating* node's wall clock, with an upper bound only. A row stamped by a node whose clock runs ahead +has a negative age, passes the test, and — being the largest `last_seen` — wins the pick. So a +crashed ex-leader whose clock ran fast can be reported here as leader while `lease_owner` on the same +response correctly names the live successor. The lease governs who processes; this field does not. +Note also that the web console's "cluster has no leader" health check keys off this field being +absent, so a skew-frozen row can keep it populated and mask a genuinely leaderless cluster. Two-node cluster: @@ -230,8 +250,9 @@ history, so `started_at`/`last_seen` are `null`; permanently leader, so `lease_e A cleanly stopped node leaves a `status: "left"` tombstone (and its leader flag cleared); a crashed node's row goes stale (its `last_seen` stops advancing) and the freshness filter stops counting it as the leader. `leader_node_id` is always **at most one** node — during a failover window (an old leader's -flag not yet cleared while the new leader's flag is already set) the freshest still-beating node wins, -so the array never shows two leaders and never names a dead node. +flag not yet cleared while the new leader's flag is already set) the freshest still-beating node wins. +That the winner is the *live* node holds only while node clocks agree; a fast-clocked crashed node can +win the pick (see `GET /cluster/nodes` above). Never two leaders; not necessarily the right one. `/cluster/status` is the **per-node authoritative** leadership signal (it reads that node's own in-memory lock gate); `/cluster/nodes` derives leadership from the heartbeat flag and so can lag it by @@ -297,7 +318,10 @@ benchmark, don't assume zero-downtime: standby acquires on its next heartbeat — failover is prompt (≈ one `heartbeat_seconds`). - **Crash / partition**: the primary's lease **ages out**, so a standby acquires after up to `leader_lease_ttl_seconds`. A partitioned old primary **self-fences** within - `leader_fence_timeout_seconds` (< the TTL), so it stops processing before the standby takes over. + `leader_fence_timeout_seconds` (< the TTL), so it stops *reporting itself* leader before the standby + takes over. It does not necessarily stop *working* by then: fencing sets a flag, and tearing the graph + down (each inbound stopped in turn, each with its own shutdown grace) is not budgeted against the + remaining margin. Expect a brief overlap in which the old primary finishes sends already in flight. - During the window, in-flight rows are protected by the **row leases** (a standby reclaims only *expired* leases); the new primary runs an owner-scoped recovery **once on promotion** to recover the dead primary's in-flight rows promptly (and the ordinary FIFO claim reclaims a stranded lane head, so @@ -309,13 +333,29 @@ benchmark, don't assume zero-downtime: The defaults (`heartbeat_seconds=10`, `leader_fence_timeout_seconds=20`, `leader_lease_ttl_seconds=30`) trade a ~30 s crash-failover for ample margin. Lower all three proportionally (keeping `heartbeat < fence < ttl`) for faster failover at the cost of less tolerance for a slow DB / GC pause. + +**The validator enforces the ordering, not a margin.** `heartbeat < fence < ttl` is checked at config +load; nothing checks that any usable time survives between the fence firing and the lease expiring. The +real margin is `ttl - fence` *minus* the renew round trip (the fence baseline is taken after the renew +returns, while the expiry is stamped on the database clock at statement execution) *minus* up to one +fence tick of detection lag — and graph teardown then has to fit in what remains. Tightening these +proportionally keeps the ordering legal while shrinking that budget toward zero, and the failure mode +is a longer overlap between an old and a new primary, not a config error. Keep headroom on a +synchronous-commit or cross-AZ database. + Because the **leadership** lease is evaluated on the **database's** clock, node clock skew does not -affect who may hold leadership; the **row** leases, however, use node wall-clock — see below. +affect who may hold leadership. At least two other things *are* node-wall-clock and therefore skew- +sensitive: the **row** leases (see below) and the `nodes.last_seen` heartbeat behind the derived +`leader_node_id`. Do not read that pairing as exhaustive. ## Operational assumptions (honor these) -1. **Clock sync (NTP).** Row leases are wall-clock — keep node clocks reasonably synced so a - lease expiry isn't mistimed across nodes. +1. **Clock sync (NTP).** Keep node clocks synced to well within `[store].lease_ttl_seconds`. Row leases + are wall-clock, so skew mistimes a lease expiry across nodes. Skew also corrupts the derived + `leader_node_id` on `GET /cluster/nodes`, where a fast-clocked node wins the freshness pick even + after it dies — and that field is what the console's "cluster has no leader" check keys off, so + unchecked skew can mask a leaderless cluster. Leadership itself is unaffected (it is evaluated on the + database clock); this is an operability assumption, not a correctness one. 2. **Identical config on every node.** Each node loads the graph (Connections / Routers / Handlers) from its **own** config dir; convergence coordinates the reload *version*, not the files. Deploy the same config dir to all nodes. diff --git a/docs/adr/0008-cluster-observability-api.md b/docs/adr/0008-cluster-observability-api.md index 53fae035..7713dc80 100644 --- a/docs/adr/0008-cluster-observability-api.md +++ b/docs/adr/0008-cluster-observability-api.md @@ -83,6 +83,16 @@ are still fresh, only the one with the largest `last_seen` is reported as leader live leader), so `/cluster/nodes` never shows two leaders and `leader_node_id` is never the dead node. This is the key correctness property: the derived leader is always a single *live* node. + > ⚠️ **CORRECTION (2026-08-01).** The "single leader" half holds unconditionally; the "always *live*" + > half does not. The freshness test is `(now - last_seen) <= node_timeout_seconds` with no lower + > bound, comparing the *reading* node's wall clock against a `last_seen` written by the *beating* + > node's wall clock. A row stamped by a node whose clock runs ahead has a negative age, passes, and + > — being the largest `last_seen` — wins the tiebreak, including after that node hard-crashes. So the + > derived leader can name a dead node while the `lease_owner` field on the same response names the + > live one. Treat `leader_node_id` as observability; the lease is authoritative. This also means the + > console's "cluster has no leader" check, which keys off `leader_node_id` being absent, can be + > masked by a skew-frozen row. + A pre-Step-7 `nodes` table (a cluster upgraded in place) is migrated idempotently under the same DDL advisory lock: `ALTER TABLE nodes ADD COLUMN IF NOT EXISTS is_leader ...`. The `NullCoordinator` synthesizes a single self-entry (single node, always leader, no DB), so diff --git a/docs/adr/0056-engine-managed-vip-failover.md b/docs/adr/0056-engine-managed-vip-failover.md index 06a28459..a74d3118 100644 --- a/docs/adr/0056-engine-managed-vip-failover.md +++ b/docs/adr/0056-engine-managed-vip-failover.md @@ -90,6 +90,13 @@ only. The correctness pivots are closed and **must not be re-decided here**: unreachable). The load-time invariant `heartbeat_seconds < leader_fence_timeout_seconds < leader_lease_ttl_seconds` (defaults 10s/20s/30s; `ClusterSettings._fence_ordering`) **guarantees a partitioned old leader stops processing before a standby can acquire** — the split-brain guard. + + > ⚠️ **CORRECTION (2026-08-01).** That ordering guarantees the old leader stops **reporting itself + > leader**, not that it stops **processing**. `_check_fence` sets an in-memory flag and cancels + > nothing; graph teardown (sequential inbound stops, each with its own shutdown grace) is not budgeted + > against the remaining margin, and the margin itself is smaller than `ttl − fence` because the fence + > baseline is stamped after the renew round trip returns while the expiry is stamped on the DB clock + > at statement execution. The VIP ordering rule below inherits this — see the correction at D4. - **Store-checked leader epoch (H1).** A monotonic `leader_epoch` bumped **only on a fresh acquire** is a durable second backstop: a superseded ex-leader that resumes after a long pause claims **0 rows** (its epoch is stale; the claim `UPDATE` matches nothing). Server-DB-only; a no-op on SQLite. @@ -232,7 +239,15 @@ is wrong by ~2×. The new ordering rule of this ADR is therefore: `_fence_ordering` already enforces `fence_timeout < ttl`, so the budget is always positive; operators who shrink the timings for faster failover **shrink this budget too** and must keep it larger than the -worst-case local release latency (below). The new leader additionally waits `release_grace_seconds` +worst-case local release latency (below). + +> ⚠️ **CORRECTION (2026-08-01).** "Always positive" is true of the *nominal* `ttl − fence_timeout` and +> not of the budget actually available. Two terms are unaccounted: the renew round trip (the fence +> baseline is taken after the renew returns, the expiry is stamped on the DB clock at statement +> execution) and up to one `_fence_tick` of detection lag. On the shipped `10/20/30` the remainder is +> ~8s rather than 10s; proportionally tightened timings shrink it further and `_fence_ordering` will +> still accept them, because it validates ordering only and never a margin. Size the release budget +> against the corrected remainder, not against `ttl − fence_timeout`. The new leader additionally waits `release_grace_seconds` before its gratuitous ARP, so it does not assert the address while a just-fenced old binding might still answer. diff --git a/messagefoundry/config/settings.py b/messagefoundry/config/settings.py index 65f0dae6..49752885 100644 --- a/messagefoundry/config/settings.py +++ b/messagefoundry/config/settings.py @@ -508,12 +508,19 @@ class StoreSettings(_Section): ) application_name: str = "messagefoundry" # Inflight-row lease TTL (seconds) for the multi-node server-DB backends (Track B Step 2). When a - # worker claims a row it stamps owner + a lease_expires_at = now + this; a renew timer extends it - # while processing, and a leader sweep reclaims only rows whose lease has expired (so a crashed - # node's work is recovered without stealing a live sibling's in-flight rows). A shared server-DB - # field — harmless to SQL Server / SQLite, which don't lease and ignore it. The lease is wall-clock - # across nodes, so the no-theft guarantee assumes clocks are NTP-synced to well within this TTL; - # set it comfortably larger than expected clock skew + the renew interval. + # worker claims a row it stamps owner + a lease_expires_at = now + this, and a leader sweep reclaims + # only rows whose lease has expired (so a crashed node's work is recovered without stealing a live + # sibling's in-flight rows). A shared server-DB field — harmless to SQL Server / SQLite, which don't + # lease and ignore it. The lease is wall-clock across nodes, so the no-theft guarantee assumes clocks + # are NTP-synced to well within this TTL. + # + # THE ROW LEASE IS STAMPED ONCE AT CLAIM AND IS NEVER RENEWED — there is no renew timer anywhere in + # the store (earlier text here and in the sweep's own docstrings claimed one; it did not exist). + # So size this against the longest a single row can legitimately stay claimed, not against a renew + # interval: it must comfortably exceed the connector's timeout_seconds + any pacing + batch-gather, + # plus expected clock skew. Set it too low and the leader's own sweep re-pends rows it is still + # processing — the sweep is owner-blind, so "a crashed node's rows" includes this node's own. + lease_ttl_seconds: float = 60.0 # --- Store connection-pool pre-warm (server-DB backends only; no-op on SQLite) ---------- @@ -2896,9 +2903,17 @@ class ClusterSettings(_Section): # (clock_timestamp()), so inter-node clock skew is irrelevant to leadership correctness. Must be > 0. leader_lease_ttl_seconds: float = 30.0 # The SELF-FENCE timeout: a leader that has not renewed its lease within this many seconds (its own - # monotonic clock, with NO DB I/O so a hung/partitioned DB can't block it) halts its leader work. - # MUST be < leader_lease_ttl_seconds so the old leader stops BEFORE the lease can expire and a standby - # acquire — the split-brain guard. MUST be > heartbeat_seconds so a single missed renew doesn't fence. + # monotonic clock, with NO DB I/O so a hung/partitioned DB can't block it) stops reporting itself + # leader. MUST be < leader_lease_ttl_seconds so it does so BEFORE the lease can expire and a standby + # acquire. MUST be > heartbeat_seconds so a single missed renew doesn't fence. _fence_ordering below + # enforces exactly that ordering — and ONLY the ordering. + # + # What the ordering does NOT establish, for anyone sizing these down from the defaults: the usable + # margin is smaller than (ttl - fence), because the fence baseline is stamped after the renew round + # trip returns while the lease expiry is stamped on the DB clock at statement execution, and + # detection lands up to one fence tick late. Nor does fencing stop the graph — it flips a boolean; + # the listeners and in-flight sends wind down on their own schedule, which nothing budgets against + # the remainder. The shipped 10/20/30 leave real slack; tightened values consume it silently. leader_fence_timeout_seconds: float = 20.0 # Leader-PREFERENCE handicap (ADR 0096). Seconds this node waits — MEASURED AGAINST THE LEASE-EXPIRY # TIME on the DB clock — before it may claim an EXPIRED leadership lease. 0.0 (default) = no handicap diff --git a/messagefoundry/pipeline/cluster.py b/messagefoundry/pipeline/cluster.py index 827d09ca..4bbb456a 100644 --- a/messagefoundry/pipeline/cluster.py +++ b/messagefoundry/pipeline/cluster.py @@ -114,9 +114,9 @@ class ClusterMember: dataclass (no API import) so the coordinator stays free of FastAPI/Pydantic — the API maps it to a :class:`~messagefoundry.api.models.ClusterNode` at the boundary. ``is_leader`` is the DERIVED leadership: at most one member carries it — the single freshest node whose durable ``nodes.is_leader`` - heartbeat flag is set AND is fresh (``last_seen`` within ``node_timeout_seconds``) — so a crashed - ex-leader's stale flag is never reported as the live leader and a failover window cannot surface two - leaders. ``last_seen``/``started_at`` are epoch seconds, ``None`` only on the + heartbeat flag is set AND is fresh (``last_seen`` within ``node_timeout_seconds``). At most one + leader is reported; that it is the *live* one holds only while node clocks agree (see + :meth:`cluster_members`). ``last_seen``/``started_at`` are epoch seconds, ``None`` only on the :class:`NullCoordinator` synthetic self-entry (no DB).""" node_id: str @@ -162,10 +162,19 @@ def current_epoch(self) -> int | None: superseded ex-leader holds a strictly *older* epoch than the live leader. The engine reads this synchronously on promotion and pushes it into the store (:meth:`Store.set_leader_epoch`), where the FIFO claim validates ``held_epoch >= leader_lease.leader_epoch`` inside the single claim - transaction so a paused/superseded ex-leader's stale claim affects **0 rows** (Kleppmann fencing - token; store ↔ coordinator import direction is one-way — the engine pushes, the store never + transaction so a paused/superseded ex-leader **claims 0 rows** (Kleppmann fencing token; store ↔ + coordinator import direction is one-way — the engine pushes, the store never imports the coordinator, ARCH-6). Cheap + synchronous (cached state). :class:`NullCoordinator` - returns ``None`` — single-node is unfenced (there is no second writer to fence).""" + returns ``None`` — single-node is unfenced (there is no second writer to fence). + + **Scope of the fence — read this before relying on it.** The guard is appended to the claim + ``UPDATE`` and to nothing else. It does NOT cover (a) the cross-owner stranded-lease reclaim that + runs as the *first* statement of the same claim transaction, which commits even when the guarded + claim matches zero rows, or (b) any post-claim disposition write (``mark_done`` / ``mark_failed`` + / ``dead_letter_now`` / ``complete_with_response``), each of which resolves its row by ``id`` + alone with no epoch, owner, or status precondition. So a demoted ex-leader that is still inside + a send cannot *claim* anything, but can still *write* — including over a row the live leader has + already resolved. Do not read this token as a general write fence.""" ... def lease_key(self) -> str | None: @@ -389,8 +398,14 @@ def __init__( # always waits out the full TTL before taking over. self._lease_ttl = leader_lease_ttl_seconds # The SELF-FENCE timeout: a leader that has not renewed within this many seconds (its own - # monotonic clock, no DB I/O) demotes itself. MUST be < the TTL so the old leader stops before - # the lease can expire and a standby acquire — the split-brain guard. + # monotonic clock, no DB I/O) demotes itself. MUST be < the TTL — but note what that ordering + # does and does not buy. It bounds when this node stops *calling itself* leader; it does NOT + # bound when this node stops *acting*. Two terms sit outside the validator (settings.py + # _fence_ordering, which checks ordering only, never margin): the fence baseline is stamped + # AFTER the renew round trip returns while the lease expiry is stamped on the DB clock at + # statement execution, so the real margin is short by that round trip; and detection lands up to + # one _fence_tick late. Graph teardown is not budgeted against the remainder at all. Treat this + # as the split-brain DETECTION bound, not a proof that the old leader has stopped. self._fence_timeout = leader_fence_timeout_seconds # The fence watchdog polls this often; small relative to the fence timeout so a fence fires # promptly (well before the lease TTL). Pure in-memory check — no DB. @@ -401,8 +416,13 @@ def __init__( self._acquire_delay = acquire_delay_seconds self._promotable = promotable # Monotonic clock for the fence (injectable for deterministic tests). Distinct from the DB clock - # the lease uses: the fence measures a node-local elapsed duration (skew-free by construction), - # the lease compares against the DB's own clock_timestamp() (so inter-node skew is irrelevant). + # the lease uses: the fence measures a node-local elapsed duration (free of INTER-NODE skew — + # that is the property being bought), the lease compares against the DB's own clock_timestamp(). + # Caveat, because "monotonic" is not the same as "always advancing in real time": this measures + # elapsed AWAKE time on platforms whose monotonic source excludes suspend (CLOCK_MONOTONIC on + # Linux does; QueryPerformanceCounter on the Windows/NSSM target does not), while the lease it + # races runs on the DB clock, which never suspends. Where the two differ, the fence is late by + # the suspended interval and the node keeps reporting leader until its next maintenance tick. self._monotonic = monotonic # Namespace the nodes-DDL advisory lock by schema, exactly as PostgresStore._lock_key does: # advisory locks are database-scoped (not schema-scoped), so two deployments sharing one @@ -563,8 +583,24 @@ async def cluster_members(self) -> list[ClusterMember]: (largest ``last_seen``) is reported as leader. During a failover window two rows can briefly both be fresh-and-flagged — the crashed ex-leader whose ``last_seen`` is frozen at the crash instant, and the newly-promoted leader whose ``last_seen`` keeps advancing. Picking the - freshest collapses that overlap to the one node that is actually beating (the live leader), so - ``/cluster/nodes`` never shows two leaders and never names the dead node as leader. + freshest collapses that overlap to a single reported leader. + + **This is a cross-node WALL-CLOCK comparison, and it is one-sided.** ``last_seen`` is stamped by + the beating node's ``time.time()`` and compared against the *reading* node's ``time.time()`` with + an upper bound only — there is no ``now - last_seen >= 0`` guard — so a row stamped by a node + whose clock runs ahead has a NEGATIVE age, trivially passes the freshness test, and (being the + largest ``last_seen``) wins the pick. A fast-clocked node therefore wins the derived-leader pick + for as long as it is alive, and keeps winning after it hard-crashes until the live successor's + advancing ``last_seen`` overtakes the frozen stamp (~the skew), while its row stays *eligible* + for skew + ``node_timeout_seconds``. + + Two consequences a reader must not be surprised by. This field can name a DEAD node as leader + while the sibling ``lease_owner`` on the same ``/cluster/nodes`` response correctly names the + live one — the lease, not this field, is authoritative for who processes. And because the web + console raises engine health to ``down``/"cluster has no leader" only when the derived + ``leader_node_id`` is ``None``, a skew-frozen row keeps that non-``None`` and can mask a + genuinely leaderless cluster for that same interval. Do not gate operational decisions on the + derived leader; gate them on the lease. One DB read, returned ordered by ``node_id`` for a stable listing; off the message hot path (operator-driven).""" @@ -903,8 +939,17 @@ async def _fence_watchdog_loop(self) -> None: """Self-fence watchdog (Workstream A2). Wakes every ``_fence_tick`` and, doing **no DB I/O**, demotes this node if it has not confirmed a lease hold within ``_fence_timeout`` (monotonic). Because it never awaits the pool, a hung/partitioned DB — which would block the maintenance loop - mid-await — cannot stop it from fencing. ``_fence_timeout < lease_ttl`` guarantees a partitioned - old leader stops reporting leader BEFORE its lease can expire and a standby acquire it.""" + mid-await — cannot stop it from fencing. + + ``_fence_timeout < lease_ttl`` is what makes a partitioned old leader stop *reporting* leader + before its lease can expire — and that is the whole of it. Two caveats, both load-bearing for + anyone deriving a safety argument from this ordering. The margin is smaller than + ``lease_ttl - _fence_timeout``: the baseline this compares against is stamped after the renew + round trip RETURNS while the lease expiry is stamped on the DB clock at statement execution, and + detection lands up to one ``_fence_tick`` late. And demotion here flips a boolean — it cancels + no listener, worker, or in-flight send. Nothing budgets graph teardown against the remainder, so + **"fenced" does not imply "stopped processing"**; do not use this as a premise for a write that + assumes the prior leader is quiescent.""" while not self._stop.is_set(): try: await asyncio.wait_for(self._stop.wait(), timeout=self._fence_tick) diff --git a/messagefoundry/pipeline/engine.py b/messagefoundry/pipeline/engine.py index e142ad32..628151ea 100644 --- a/messagefoundry/pipeline/engine.py +++ b/messagefoundry/pipeline/engine.py @@ -1078,8 +1078,18 @@ async def start(self) -> None: ) self._leader_maintenance.start() # else (SQL Server active-passive): no per-row leases, so there is no reclaim sweep — failover - # recovery is the on-promotion reset_stale_inflight in _start_graph (the old leader self-fenced - # before its lease expired, so re-pending its in-flight rows can't steal from a live processor). + # recovery is the on-promotion reset_stale_inflight in _start_graph. Do NOT justify that by "the + # old leader self-fenced before its lease expired, so it can't steal from a live processor": the + # fence only stops the node REPORTING leader, not working (see _check_fence, which flips a bool + # and cancels nothing), so the two can overlap. The reset is correct for a different reason — it + # errs toward a DUPLICATE, which at-least-once permits, rather than a STRAND, which it forbids. + # + # Consequence worth knowing: this is SQL Server's ONLY in-flight recovery. reclaim_expired_leases + # is defined on the Postgres store alone, and the sweep below is gated on hasattr(), so a row left + # INFLIGHT here OUTSIDE a promotion (e.g. the deliberate teardown path in stage_dispatcher, which + # leaves a cancelled prefix INFLIGHT for exactly this recovery) has nothing periodic to reclaim it + # and is stranded until the next promotion or restart. Postgres bounds the same case at roughly + # reclaim_interval + lease_ttl. Tracked in ADR 0157. # Config-reload convergence (Track B Step 6) — only in clustered mode (is_clustered()), so # single-node / SQLite never spawns it. Seed the applied version to the coordinator's CURRENT # shared version BEFORE the loop starts, so a fresh node does not immediately self-reload (it is @@ -1180,9 +1190,16 @@ async def _start_graph(self) -> None: await self._leader_maintenance.recover_on_promotion() elif self._coordinator.is_clustered(): # Active-passive without per-row leases (SQL Server): on promotion, re-pend the prior - # leader's in-flight rows. The prior leader self-fenced and its leadership lease EXPIRED - # before this node could acquire it, so it has stopped processing — and the graph runs ONLY - # on the leader, so there is no live sibling whose rows an unconditional reset could steal. + # leader's in-flight rows. Note what justifies this and what does not. The prior leader + # self-fenced and its leadership lease expired before this node could acquire — but that + # ordering only means it stopped REPORTING itself leader, not that it stopped working: the + # margin is short by the renew round trip plus a fence tick, and graph teardown is not + # budgeted against the remainder. This reset is also owner-BLIND (WHERE status/stage only), + # so unlike the Postgres path it cannot even scope itself to the prior leader's rows. + # It is correct anyway, for the same directional reason as recover_inflight_on_promotion: + # re-pending a row whose sender is still winding down risks a DUPLICATE (permitted by + # at-least-once, absorbed by idempotent outbounds), while not re-pending it risks a STRAND + # (forbidden). Erring toward duplication is the invariant-safe direction. # (Single-node NullCoordinator is_clustered() is False, so this never runs there; its boot # residue was already recovered by the unconditional reset_stale_inflight in start().) await self.store.reset_stale_inflight() diff --git a/messagefoundry/store/postgres.py b/messagefoundry/store/postgres.py index acc53a46..564a6a05 100644 --- a/messagefoundry/store/postgres.py +++ b/messagefoundry/store/postgres.py @@ -2671,13 +2671,19 @@ async def claim_next_fifo( this lane's stranded head — a crashed/fenced prior leader's claimed rows are still ``inflight`` under an expired ROW lease, and the PENDING-only head SELECT would skip them and reorder past the true head N. So before the head SELECT, this lane's expired-lease ``inflight`` rows are - returned to ``pending`` (scoped to ``lease_expires_at < now`` so it never disturbs a live - node's actively-processed rows — their leases are kept in the future by the worker's renew - timer), restoring head-of-line blocking: the recovered N is reconsidered as the (due) head and - blocks the lane until delivered. Without this, after a promotion the new leader would deliver - N+1 before N (a per-lane FIFO break across failover). The wall-clock lease shares Track B - Step 2's NTP assumption: set ``lease_ttl_seconds`` comfortably above clock skew + the claim - cadence.""" + returned to ``pending`` (scoped to ``lease_expires_at < now``), restoring head-of-line blocking: + the recovered N is reconsidered as the (due) head and blocks the lane until delivered. Without + this, after a promotion the new leader would deliver N+1 before N (a per-lane FIFO break across + failover). + + The ``lease_expires_at < now`` scope is what is meant to keep this off a live node's + actively-processed rows. **It is a one-shot claim-time stamp that is never renewed** — there is + no renew timer in this store (earlier text here asserted one). A row therefore becomes eligible + for this reclaim ``lease_ttl_seconds`` after it was CLAIMED, however long it has legitimately + been processing since, so the TTL must exceed the longest legitimate claim-to-terminal-write + hold, not merely the claim cadence. The wall-clock lease also shares Track B Step 2's NTP + assumption. Note this reclaim statement is NOT covered by the H1 epoch guard appended to the + claim below — it is a separate, unguarded statement in the same transaction.""" now = time.time() if now is None else now lease_until = now + self._settings.lease_ttl_seconds # Track B Step 2: stamp the lease lane_col = self._lane_col(stage) # code-controlled literal @@ -4595,9 +4601,17 @@ async def reclaim_expired_leases( Clock assumption: the no-theft guarantee is a wall-clock lease — the reclaiming node compares its own ``now`` against a ``lease_expires_at`` stamped by the (possibly different) holder node's - clock. It holds only when node clocks are synchronized (NTP) to well within ``lease_ttl_seconds``; - set the TTL comfortably larger than expected skew + the renew interval so a skewed reclaimer - can't beat a live holder's lease.""" + clock. It holds only when node clocks are synchronized (NTP) to well within ``lease_ttl_seconds``. + + **The lease is stamped once at claim and never renewed** (there is no renew timer in this store, + despite what earlier text here claimed), and this sweep is OWNER-BLIND. Together those mean the + "live sibling" it must not steal from includes *this node itself*: any row still legitimately + claimed ``lease_ttl_seconds`` after it was claimed is eligible, and in active-passive the leader + that runs this sweep is the only node holding in-flight rows. Size ``lease_ttl_seconds`` above + the longest legitimate claim-to-terminal-write hold (connector ``timeout_seconds`` + pacing + + batch-gather) plus expected skew, not merely above skew. A row this sweep intercepts mid-flight + keeps the claim's ``attempts`` increment: ``release_claimed``/``reschedule_claimed`` are guarded + ``status='inflight'`` and silently no-op once the row is back to ``pending``.""" now = time.time() if now is None else now result = await self._pool.execute( "UPDATE queue SET status=$3, owner=NULL, lease_expires_at=NULL," @@ -4624,11 +4638,22 @@ async def recover_inflight_on_promotion(self, *, now: float | None = None) -> in the prior leader's residue. Re-pending the stranded lane HEAD restores per-lane head-of-line blocking (no N+1-before-N reorder). - **Safe ONLY in active-passive** (the wired graph runs on the leader ONLY): the prior leader - self-fenced and its LEADERSHIP lease expired on the DB clock before this node could acquire it - (``heartbeat < fence < leader_lease_ttl`` is validator-enforced), so there is no live processor - whose rows this could steal — the SAME interlock the shipping SQL Server on-promotion - ``reset_stale_inflight`` relies on. Returns the number of queue rows re-pended.""" + **Safe ONLY in active-passive** (the wired graph runs on the leader ONLY) — but be precise about + WHY, because the obvious argument is not sound. It is tempting to say the prior leader + self-fenced and its leadership lease expired before this node could acquire, so nothing is still + processing. ``heartbeat < fence < leader_lease_ttl`` is validator-enforced, but that ordering + establishes only that the prior leader stopped **reporting** itself leader; it does not + establish that it stopped **working**. The usable margin is short by the renew round trip and up + to one fence tick, and graph teardown — sequential source stops, each with its own grace — is + not budgeted against what remains. So this write CAN land on rows an ex-leader is still + mid-send on. + + The real justification is directional: re-pending a row whose sender is still alive risks a + DUPLICATE delivery, which at-least-once explicitly permits and idempotent outbounds absorb; + NOT re-pending it risks a row stranded ``inflight`` under a one-shot lease with no owner left to + resolve it, which the count-and-log invariant forbids. This errs the way the invariant allows. + The shipping SQL Server on-promotion ``reset_stale_inflight`` makes the same trade, less + precisely — it is owner-blind where this is owner-scoped. Returns the number of rows re-pended.""" now = time.time() if now is None else now async with self._timed_acquire() as conn, conn.transaction(): result = await conn.execute(