diff --git a/.woodpecker.yaml b/.woodpecker.yaml index 23e52f6..606873b 100644 --- a/.woodpecker.yaml +++ b/.woodpecker.yaml @@ -34,6 +34,41 @@ when: event: [push, cron] steps: + # ── Fast in-process suites (push) ─────────────────────────────────────────── + # ADDED 2026-08-31 after a survey found that 14 of 23 integration suites were + # run by NO pipeline at all. That is not a theoretical gap: `cluster_routes` + # still asserted a wire fabrication that had been removed hours earlier + # (`applied_events == 0` with a lag derived from it), and nothing caught it + # because nothing ran it. `cluster_sharding` and `vector_search` — which hold + # the dense-rank, /sharded/* opt-in and vector distance-contract gates — were + # in the same position, so the guards written for those fixes would have rotted + # the same way. + # + # These eight need neither the `cluster-e2e` feature nor spawned processes, so + # they take the LIGHT shape. Measured locally: 71s wall for all eight, 51 tests. + # Against a 6.5-minute gate that is free, and it runs FIRST so a cheap failure + # is reported before the expensive one starts. + fast-suites: + image: rust:1-bookworm + when: + event: push + backend_options: &resources-light + kubernetes: + resources: + requests: { cpu: "500m", memory: 1Gi } + limits: { cpu: "2", memory: 4Gi } + commands: + - apt-get update && apt-get install -y --no-install-recommends protobuf-compiler cmake clang + - cargo test -p tidaldb --lib + - cargo test -p tidal-server --test middleware + - cargo test -p tidal-server --test standalone + - cargo test -p tidal-server --test standalone_offload + - cargo test -p tidal-server --test vector_search + - cargo test -p tidal-server --test cluster_routes + - cargo test -p tidal-server --test cluster_region + - cargo test -p tidal-server --test cluster_grpc + - cargo test -p tidal-server --test reseed_install + # ── Release gate (push) ───────────────────────────────────────────────────── # m11p8 release gate: prove a rolling upgrade under load loses no acknowledged # write and never stalls (mp_rolling_upgrade_no_loss_no_stall — a tier-3 test @@ -100,11 +135,7 @@ steps: # Kaniko build: single process, but it OOMKilled at the 2Gi namespace default. # A build does not need the heavy shape's CPU floor, so it is sized separately # to leave headroom for anything co-scheduled on the same node. - backend_options: &resources-light - kubernetes: - resources: - requests: { cpu: "500m", memory: 1Gi } - limits: { cpu: "2", memory: 4Gi } + backend_options: *resources-light settings: repo: tidal/server dockerfile: docker/standalone/Dockerfile @@ -161,6 +192,22 @@ steps: - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_reseed -- --nocapture --test-threads 1 - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_lifecycle -- --nocapture --test-threads 1 - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_runbook -- --nocapture --test-threads 1 + # ADDED 2026-08-31. These six spawn real 3-process clusters and were run by + # NO pipeline before today — including `cluster_sharding`, which holds the + # dense-rank and /sharded/* opt-in gates, and `cluster_poison_embedding`, + # the regression gate for the malformed-embedding outage. Guards nothing + # runs are guards that rot: `cluster_routes` proved it by still asserting a + # wire fabrication that had been removed hours earlier. + # + # They live in the nightly rather than the push path because each boots + # three OS processes; the eight in-process suites moved to `fast-suites` + # above, which costs 71s and runs on every push. + - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_sharding -- --nocapture --test-threads 1 + - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_poison_embedding -- --nocapture --test-threads 1 + - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_cross_shard_reads -- --nocapture --test-threads 1 + - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_graph_persistence -- --nocapture --test-threads 1 + - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_multiproc -- --nocapture --test-threads 1 + - cargo test -p tidal-server --features "cluster-e2e fault-injection" --test cluster_e2e -- --nocapture --test-threads 1 # ── Nightly security + ops correctness (cron) ─────────────────────────────── # The G-Sec and G-Op owner-tests run nightly too (not just on-demand): mTLS / diff --git a/CHANGELOG.md b/CHANGELOG.md index 5cc978f..50228af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,6 +72,51 @@ reseed, or a local shard lag still exit 2. ### Fixed +**Embeddings are validated BEFORE the WAL append; malformed vectors now 400 +(wire-visible)** — *fleet-hardening task 17* + +`POST /embeddings` appended the blob to the leader's WAL **first** and validated +it **second**. A probe posted a 128-dimension vector to a slot the live schema +declares at 1536: the client got a `500`, but the record was already durable, so +it shipped to both followers, neither could apply it, and both **halted their +receivers**. Shard 1 froze (leader `13540698`, followers pinned at +`13540694`/`13540693`, `lag` growing) and writes to the group returned **503** — +a quorum-write outage from one malformed HTTP request. Neither a restart (boot +self-heal re-pulls the same record) nor `POST /cluster/reseed` escaped it: the +snapshot is captured at the leader's applied frontier, which is itself *behind* +the poison, so the reseeded follower replayed straight back into it. Only forcing +a leader election recovered the group. + +Fixed on both sides of the durability boundary: + +- **Write path.** The dimension check now runs before `wal_blob_first`, against + the schema-declared slot width and this node's registered width. A mismatched, + empty, non-finite, or zero-norm vector is rejected **400** + (`TidalError::invalid_input`) and appends **nothing** — it cannot ship, so no + follower can be poisoned by it. The comparison is + `storage::vector::validate_dimensions`, which the apply path's + `normalize_and_store` also calls: one rule, one definition, because it is the + apply-path copy that halts replication when the two disagree. +- **Receive path.** A replicated record that is unapplicable on **every** replica + — its verdict reads only the record's own bytes and the shared schema — is now + **skipped** rather than halted on: dropped before it enters this node's own log, + counted once per record on + `tidaldb_cluster_blobs_apply_failed_total`, logged at ERROR with the seqno + range, and the round's remaining records still apply so the frontier advances + past it. Failures that might apply later still halt, deliberately: a node-local + disk/lock fault, an unknown batch kind a newer binary understands, or a + malformed term/membership record whose loss would diverge the roster. Skipping + those would trade an availability bug for silent data loss. + +`ReplicatedBlobApplier::apply_blobs` now returns `BlobApplyOutcome { rejected }` +instead of `()` (in-crate trait; no HTTP or on-disk surface changes). + +*Migration:* a client that treated a `500` from `/embeddings` as retryable must +treat the `400` as terminal and fix the vector width. Runbook §5 documented the +`500` as "a known wart, tracked post-M8" — it was not cosmetic, and §5 plus the +new §16.6 (halted receiver: recovery by leadership change, and why reseed cannot +escape a poison record) now say so. + **`/feed` and `/search` `rank` is now dense and ascending (wire-visible)** Under full placement — the production shape, every shard group on every node — @@ -136,6 +181,60 @@ diverge unobserved. Co-located groups now render compatibility, so an alert grouped `by (shard)` buckets each replica set separately without double-counting. +**A converged, unquarantined, voting pod no longer sits permanently `503`, and +`/health` no longer names a cause it never checked** — *fleet-hardening task 18* + +During the 2026-08-31 staged roll, `tidaldb-2` logged `latched: 2` reseed markers +(`from_seqno` `13322280` and `13540699`) and `discharged: 1`, then answered +`503 {"cause":"joiner boot not yet converged"}` indefinitely while +`/cluster/status/local` reported `reseed_required=false`, `quarantined=false`, +`membership_role=voter` and `lag=0` on all three shards — every group having +already logged *"first-converged against a KNOWN leader frontier ... readiness is +now sticky-ready"*. It happened on all three pods and each needed a manual +`kubectl delete pod`. + +Three defects, all in `tidal-server/src/cluster/node.rs`: + +- **The latch and the discharge were not the same shape.** A shard group held one + readiness flag plus one durable marker slot that a re-latch **overwrote**, so a + group that latched at two different frontiers (a term join at one, a later + `snapshot-required` refusal at another) let one completed catch-up pull decide + both — clearing readiness with the lower gap still a hole in one latch order, + and clearing nothing in the other. Outstanding gaps are now a set keyed by + `(shard, from_seqno)`, each discharged only by evidence about **itself** (the + unchanged `ReseedMarker::discharged_by_served_range` — never a frontier + comparison), and the durable slot is re-derived from what remains as the + conservative collapse: lowest resume point, least-dischargeable reason. The + latch WARN now fires once per distinct gap and the discharge INFO once per gap + cleared, so `latched` and `discharged` count the same population. +- **A refused `reseed_self_restart` was never re-evaluated.** The §2.4 quorum + refusal is correct and unchanged — it fired during the roll with + `alive_voters_excluding_self=1 < majority=2` and is why the cluster stayed up — + but its only re-drive was *"a future latch re-evaluates"*, and a node whose gap + is real gets no future latch once the catch-up retry stops refusing. It waited + forever on a quorum event nobody polled; `kubectl delete pod` performed by hand + exactly the restart the node had refused. A held refusal now re-evaluates every + 15s until the gap heals, the refusal clears, or the exit fires. +- **`/health` reported a cause it had not tested.** The 503 ladder tested three + gates and printed `"joiner boot not yet converged"` for everything else — + including the latched reseed marker, which was the gate that actually held. + Every gate now has its own arm; the reseed arm names the outstanding gaps + (`reseed_marker_latched: shard group 1 holds an unhealed reseed gap. Outstanding + node-wide: [s1@13540699(snapshot_required)] ...`) and says when a self-restart + is refused; and the terminal arm reports `unready: reason unavailable` instead + of inventing a state. A probe that lies about its reason is worse than one that + admits it has none. + +The readiness gate itself is unchanged in spirit: a node holding stale data it is +about to discard still drains from the client VIP (the property that stopped a +PVC-wiped `tidaldb-0` serving an empty corpus). The bug was that it never stopped +draining. + +*Wire-visible:* `/cluster/status/local` and each `shards[]` row gain +`reseed_gaps` (the outstanding frontiers, lowest first), and the flat/per-group +`reseed_required` now reports the live gap set rather than the marker file — so it +can no longer answer `false` while `/health` answers 503 for a latched marker. + ### Added **Multi-vector user preference modeling + ANN candidate-gen (M12) — a warm user is many interests, not one averaged vector: per-user preference clusters drive a top-M ANN fan-out in `for_you`** diff --git a/docs/runbooks/cluster.md b/docs/runbooks/cluster.md index 11753d3..e0a9e9a 100644 --- a/docs/runbooks/cluster.md +++ b/docs/runbooks/cluster.md @@ -573,11 +573,15 @@ converges from the log when it returns. Embeddings: tidalDB does **not** generate vectors — the caller brings them. The write L2-normalizes and inserts into the HNSW index. Dimensions are **strict**: they must equal the slot's declared dimensions (min 2, max 4096) or the insert is -**rejected with a 500** (the engine surfaces a dimension mismatch as an internal -error, not a 400 — a known wart, tracked post-M8); **zero-norm vectors are also -rejected (500)**. `RETRIEVE` / `SEARCH` route through the **first declared -embedding slot only**; multi-modal apps must fuse offline or use separate entity -kinds. +**rejected with a 400**, and **zero-norm / non-finite vectors are rejected (400)** +too. Both are checked **before the WAL append**, so a malformed vector never +becomes durable and never enters the replication stream — see +[§16.6](#166-a-halted-receiver-a-poison-record-in-the-log) for the outage that +ordering caused when it was the other way round. (These used to surface as `500`; +that was a real defect, not just an imprecise status code — the record had already +been journaled by the time the error was produced.) `RETRIEVE` / `SEARCH` route +through the **first declared embedding slot only**; multi-modal apps must fuse +offline or use separate entity kinds. ### Record signals (cluster `/signals` = global only) @@ -840,6 +844,13 @@ curl -X POST "$BASE/cluster/reseed" # → { "reseed_required": true } ``` +> **Reseed escapes a divergent suffix, not a poison record.** The snapshot is +> captured at the **leader's applied frontier**. If the group is stalled because +> a record in the log cannot be applied by anyone, that frontier is itself +> *behind* the bad record — the reseeded node installs the snapshot and replays +> straight back into it. Do not reach for this verb on a halted receiver; see +> [§16.6](#166-a-halted-receiver-a-poison-record-in-the-log). + **Scale-down order (decommission):** call `/cluster/members/remove` **first** (so the cluster's quorum math shrinks before the node disappears), wait for the record to quorum-commit, then stop / delete the node — lowest ordinal last under @@ -1078,6 +1089,14 @@ report kind-4 capability** — so the first conf-change cannot fire mid-upgrade. node's WAL, downgrading it below p5 requires a reseed.** Complete the binary upgrade across all voters before adding, removing, or replacing a node. +**The same halt was reachable from ordinary malformed input, and that path had +no gate.** A dimension-mismatched embedding was appended to the WAL *before* it +was validated, so it shipped and halted both followers — one client request, one +quorum-write outage (2026-08-31, shard 1). Closed on both sides now: the write +path validates before the append, and the receiver skips a record that is +unapplicable on every replica instead of halting on it. See +[§16.6](#166-a-halted-receiver-a-poison-record-in-the-log). + ## 9. Failover (multi-process) ### 9.1 Automatic failover (m11p4 — the default) @@ -1126,6 +1145,14 @@ strictly leader-ack-only data, within the documented `ack=leader` crash contract (§8). The tier-3 `cluster_reseed.rs` proves the full quarantine → marker → restart → reseeded → gauges-cleared loop. +**Scope of "automatic recovery".** The reseed path above repairs a **divergent +suffix** — records this replica holds that the elected majority does not. It does +**not** repair a record the whole group holds and none of them can apply: the +snapshot is taken at the leader's applied frontier, which in that case is behind +the offending record, so the reseeded node replays back into it. That failure is +[§16.6](#166-a-halted-receiver-a-poison-record-in-the-log), and its remedy is a +leadership change, not a reseed. + **The three-way term-join rule (m11p5).** The divergent-suffix check above is one of three outcomes a node reaches when it joins a new term, comparing its own stream position against the leader's election-time position (`prev_log`, carried @@ -1627,7 +1654,8 @@ apply per follower. Read it per (pod, kind), not as a fleet-wide subtraction. | reading | meaning | next step | | --- | --- | --- | -| `apply_failed` > 0 | the receiver rejected a record and **halted** that stream rather than skipping it | read that pod's logs for the apply error; the stream is stalled, not silently lossy | +| `apply_failed` > 0 **and** that pod's `applied` is flat / `lag` growing | the receiver **halted** that stream (it stopped applying entirely) | §16.6 — read that pod's logs for the apply error; the stream is stalled, not silently lossy | +| `apply_failed` > 0 but `applied` keeps advancing and `lag` returns to 0 | the receiver **skipped** a record no replica could ever apply and continued (logged at ERROR with the seqno range). Expected for a malformed record already in the log; the group stays available | find the ERROR line, fix the producer; nothing to recover on the replica | | peer's `applied` never advances while the writer's `originated` does | the record is not arriving — enqueue or ship | §16.4 | | `applied` advances but that group's `usearch_vector_count` lags | it arrived and was not indexed | index-gap; the store is intact, a restart recovers it | @@ -1693,6 +1721,125 @@ apply normally complete in well under a second on an in-cluster (low-RTT) deploy is the failure in §16.2. The alert's `for: 15m` exists so a rolling deploy's legitimate rebuild window never pages. +### 16.6 A halted receiver: a poison record in the log + +**Symptom.** One shard group's followers stop advancing while its leader keeps +writing: leader at seqno `N`, followers pinned at `N-4` / `N-5`, `lag_events` +growing, `tidaldb_cluster_blobs_apply_failed_total{kind="..."}` non-zero, and +**writes to that group return 503** — the followers cannot ack, so quorum is +gone. The pod log names the record: + +```text +replicated blob batch apply failed (1 records): + [op=write_item_embedding] dimension mismatch: expected 1536, got 128 +replication apply failed; receiver halting (health degraded) +``` + +This is a **quorum-write outage caused by a single record in the WAL**, not by +any node being unhealthy. It happened on 2026-08-31 on shard 1: one probe posted +a 128-dimension vector to a slot the live ConfigMap declares at 1536. + +#### Restarting does NOT fix it + +Boot self-heal re-pulls the same record from the leader and the receiver halts +again at the same seqno. The halt is **permanent across restarts** — the same +shape [§8](#8-write-durability-contract-the-ack-knob-and-its-cost-m11p3)'s +kind-4/pre-p5 capability note describes, reached there from a version skew and +here from ordinary malformed input. + +#### `POST /cluster/reseed` does NOT fix it either — and can make it worse + +This is the opposite of what the reseed narrative in +[§9.1](#91-automatic-failover-m11p4--the-default) and the +[`/cluster/reseed` verb](#membership-verbs-m11p5--online-add--remove--inspect--reseed) +imply, and it is worth understanding before reaching for it: + +> **A snapshot is captured at the leader's APPLIED frontier, and on a halted +> group that frontier is itself BEHIND the poison record.** So the reseeded +> follower installs a snapshot from *before* the bad record, then replays forward +> **into it** and halts again. On 2026-08-31 the reseed left the node reporting 3 +> apply failures instead of 2 — marginally worse than doing nothing. + +Reseed escapes a **divergent suffix** (this replica holds records the elected +majority does not). It cannot escape a record the whole group holds and none of +them can apply. Different failure, different remedy. + +#### What DOES fix it: force a leadership change on the affected group + +Send the promote **to the node you want to lead, naming itself**, not to the +current leader: + +```bash +# $NODE = a surviving FOLLOWER of the stalled group, addressed directly. It is by +# definition not caught up — that is what makes the fenced path below unusable. +curl -sk -X POST "$NODE/cluster/promote?shard=" \ + -H 'content-type: application/json' -H "authorization: Bearer $ADMIN_KEY" \ + -d '{"region":""}' +``` + +On the target-side path the node first asks the current leader for a sanctioned +transfer; that request **fails** here (see below), so the node campaigns directly +and the group elects at term+1. This is the same path the dead-leader failover +drill uses, and a stalled group is the same situation from the survivors' side. + +> **Do NOT reach for `POST /cluster/shards//transfer`, and do not send the +> promote to the LEADER.** Both run the *fenced* transfer, whose drain waits (5s) +> for the target to hold the leader's full flushed prefix before `TimeoutNow`. On +> a stalled group the target can never get there — that is the entire problem — +> so it returns: +> +> ```text +> transfer target '' lags the flushed frontier (13540694 < 13540698); +> heal it first, then retry the promote +> ``` +> +> The message is correct and its advice is a dead end here: `/cluster/heal` +> re-drives the same stream into the same halt. Do not chase it. + +If no verb can be reached, deleting the group's LEADER pod forces the election — +this is what recovered the live cluster on 2026-08-31. Check +`disruptionsAllowed` first (§16.3). This is the one case where restarting a pod +is the remedy rather than evidence destruction (§16's "DO NOT restart a pod +first"), because the group is already stalled and its counters are already +captured: + +```bash +kubectl -n tidaldb-cluster delete pod tidaldb- +``` + +Either way the receivers restart under the new term and re-derive their position, +and the group converges. On 2026-08-31 `tidaldb-1` took shard 1 at term 113, all +three nodes converged to `13540698` with `lag=0`, and writes returned to `201`. + +Verify, per node: + +```bash +curl -sk "$NODE/cluster/status/local" | jq '{applied_events, lag_events, last_seq}' +kubectl -n tidaldb-cluster logs --since=5m | grep -c 'receiver halting' # expect 0 +``` + +Leave the poison records where they are. They sit below the converged frontier +and are harmless once every replica has moved past them. + +#### Why a new malformed write can no longer do this + +The origin now validates an embedding **before** `wal_blob_first` appends +(`tidal/src/db/items.rs`), so a dimension-mismatched vector is rejected **400** +and never becomes durable — it cannot ship, and no follower can be poisoned by +it. And a record that *is* in the log and is unapplicable on **every** replica +(its verdict reads only the record's bytes and the shared schema) is now +**skipped** by the receiver rather than halted on: counted on +`tidaldb_cluster_blobs_apply_failed_total`, logged at ERROR with the seqno range, +stream continues. Failures that might apply later — a node-local disk/lock fault, +an unknown batch kind a newer binary understands, a malformed term/membership +record whose loss would diverge the roster — still halt, deliberately: skipping +those would trade an availability bug for silent data loss. + +Regression gate: `tidal-server/tests/cluster_poison_embedding.rs` +(tier-3, 3 real processes) and the `db::items` unit tests +`a_dimension_mismatched_embedding_is_rejected_before_the_wal_append` and +`a_schema_unapplicable_replicated_embedding_is_skipped_not_halted`. + ## Cross-references - **Kubernetes deployment** — [docs/runbooks/kubernetes.md](kubernetes.md) diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 55ce83a..54db344 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -207,6 +207,36 @@ const REMOVE_DELIVERY_GRACE_DEFAULT_MS: u64 = 30_000; /// circuit-breaker reset window (30 s) without churning cursors every tick. const SELF_HEAL_TICKS: u64 = 60; +/// How often a REFUSED `reseed_self_restart` (§2.4 quorum refusal) re-evaluates +/// while the refusal holds. Deliberately a constant, not a topology knob: it is +/// an internal liveness poll (one `/cluster/status/local` fan-out over the other +/// voters), invisible to clients, and the only requirement is that it be short +/// against an operator's reaction time and long against a rolling deploy's +/// per-pod settle. A refused node is 503 and serving nothing, so re-polling is +/// strictly better than waiting for a latch that may never come. +const REFUSAL_RETRY_INTERVAL: Duration = Duration::from_secs(15); + +/// Whether a REFUSED `reseed_self_restart` should be re-evaluated again. +/// +/// Three exits, all of them terminal for the loop: +/// - the node is draining (the exit it wanted is already happening, or the +/// process is going down for another reason); +/// - the refusal cleared, so ownership of the exit has moved on — either +/// `maybe_self_restart` fired it, or the §2.4 grace deferred it to the +/// co-hosted-shard coordinator; +/// - the gap healed via stream catch-up, so a reboot would reseed nothing and +/// restarting would be a futile loop (the m12 reseed-loop-fix invariant). +/// +/// Split out from the loop body so those exits are asserted, not just read: the +/// bug this retry closes was itself a missing re-drive. +const fn refusal_retry_continues( + shutting_down: bool, + refusal_holds: bool, + gap_outstanding: bool, +) -> bool { + !shutting_down && refusal_holds && gap_outstanding +} + /// The election driver's boot bundle: prepared in [`ShardReplica::new`] /// (where the durable classification runs), consumed by /// [`ShardReplica::start_election_driver`] once the node is in its @@ -219,6 +249,200 @@ struct ElectionBoot { topology_leader: RegionId, } +/// Every reseed gap this NODE is still waiting to have served, keyed by the +/// shard group that latched it and the seqno the stream refused it at. +/// +/// # Why a SET and not one bool cleared by one seqno +/// +/// The latch and the discharge must be the SAME SHAPE. Under full placement a +/// node hosts every shard group, so a rolling deploy reliably opens a gap on +/// more than one of them; a single group can also latch twice at different +/// frontiers (a term join at one, a later `snapshot-required` refusal at +/// another). Before task 18 each replica held ONE `AtomicBool` plus ONE durable +/// marker slot that a re-latch OVERWROTE, and decided the discharge from +/// whichever gap happened to be written last. Both directions were wrong: +/// +/// - `latch(A=13322280)` then `latch(B=13540699)`: a pull completing from any +/// `served_from` in `(A, B]` discharged the surviving marker and cleared +/// readiness while gap A was still unserved — a hole, served to readers. +/// - `latch(B)` then `latch(A)` (the slot now holds A): a pull completing in +/// `(A, B]` discharged NOTHING, so a node whose gap B WAS served stayed +/// drained from the client VIP with every shard at `lag=0`, `quarantined: +/// false`, `membership_role: voter` — and no remaining way to heal. That is +/// the 2026-08-31 outage: `latched: 2, discharged: 1` on `tidaldb-2`, three +/// pods permanently `503`, each cleared only by `kubectl delete pod`. +/// +/// So each gap is tracked individually and discharged by evidence about ITSELF +/// — [`ReseedMarker::discharged_by_served_range`], the one definition of +/// "served", unchanged. Readiness is drained while a group holds ANY outstanding +/// gap (the safety property `note_lag_for_readiness` documents: a node about to +/// discard stale data must not sit in the client VIP), and the durable per-group +/// marker is re-derived from this set on every mutation ([`LatchOutcome::durable`] +/// / [`ServeOutcome::durable`]) — so the operator-visible latch and discharge +/// counts are one per gap and must agree. +pub struct ReseedGapSet { + /// `(shard, from_seqno) -> why`. A `BTreeMap` so one group's range scan is + /// ordered BY `from_seqno`: the first entry of a group's range is its lowest + /// outstanding gap, which is exactly the durable resume point. + outstanding: std::sync::Mutex>, +} + +/// What [`ReseedGapSet::latch`] changed. +pub struct LatchOutcome { + /// Whether this call made a NEW gap outstanding, or strengthened an existing + /// gap's reason. Only a fresh latch is an operator-visible event, so the + /// "reseed marker latched durably" WARN fires once per gap — which is what + /// makes the latched and discharged counts comparable at all. + pub fresh: bool, + /// The marker this group's durable slot must now hold. + pub durable: ReseedMarker, + /// How many gaps this group is waiting on after the latch. + pub outstanding: usize, +} + +/// What [`ReseedGapSet::serve`] changed. +pub struct ServeOutcome { + /// The gaps this completed pull PROVED served, lowest first — one + /// operator-visible "reseed marker discharged" per entry. + pub discharged: Vec<(u64, ReseedReason)>, + /// The marker this group's durable slot must now hold; `None` ⇒ clear it. + pub durable: Option, +} + +impl ReseedGapSet { + /// An empty set (no group is waiting on anything). + pub(crate) const fn new() -> Self { + Self { + outstanding: std::sync::Mutex::new(BTreeMap::new()), + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, BTreeMap<(ShardId, u64), ReseedReason>> { + self.outstanding + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Record a gap on `shard` at `from_seqno`. + /// + /// Idempotent: re-latching the same frontier with the same (or a weaker) + /// reason reports `fresh: false` and changes nothing. A STRENGTHENED reason + /// (structural over `snapshot_required`) does change the set — a gap only a + /// snapshot can heal must never be recorded as one a pull could discharge. + pub(crate) fn latch( + &self, + shard: ShardId, + reason: ReseedReason, + from_seqno: u64, + ) -> LatchOutcome { + let mut gaps = self.lock(); + let key = (shard, from_seqno); + let held = gaps.get(&key).copied(); + let effective = held.map_or(reason, |h| Self::strongest(h, reason)); + let fresh = held != Some(effective); + if fresh { + gaps.insert(key, effective); + } + // Non-empty by construction (the gap above is in the set); the fallback is + // the same value that collapse would return for it. + let durable = Self::collapse(&gaps, shard).unwrap_or(ReseedMarker { + reason: effective, + from_seqno, + }); + let outstanding = Self::range(&gaps, shard).count(); + drop(gaps); + LatchOutcome { + fresh, + durable, + outstanding, + } + } + + /// Discharge every gap on `shard` that a catch-up pull COMPLETED from + /// `served_from` proves served. + /// + /// The per-gap predicate is [`ReseedMarker::discharged_by_served_range`], + /// applied to each gap in its own right: a completed pull streamed + /// everything from `served_from` up to the source's stream end, so it serves + /// exactly the dischargeable gaps at or above `served_from` and says nothing + /// about the ones below. Never a frontier comparison. + pub(crate) fn serve(&self, shard: ShardId, served_from: u64) -> ServeOutcome { + let mut gaps = self.lock(); + let discharged: Vec<(u64, ReseedReason)> = Self::range(&gaps, shard) + .filter(|&(from_seqno, reason)| { + ReseedMarker { reason, from_seqno }.discharged_by_served_range(served_from) + }) + .collect(); + for &(from_seqno, _) in &discharged { + gaps.remove(&(shard, from_seqno)); + } + let durable = Self::collapse(&gaps, shard); + drop(gaps); + ServeOutcome { + discharged, + durable, + } + } + + /// Whether `shard` is still waiting on any gap — the per-group readiness + /// gate ([`ShardReplica::is_ready`]) and the node-level aggregate + /// ([`ClusterNode::is_ready`], which is `all hosted groups ready`). + pub(crate) fn outstanding_for(&self, shard: ShardId) -> bool { + let gaps = self.lock(); + Self::range(&gaps, shard).next().is_some() + } + + /// `shard`'s outstanding gap frontiers, lowest first (the status surface). + pub(crate) fn seqnos_for(&self, shard: ShardId) -> Vec { + let gaps = self.lock(); + Self::range(&gaps, shard).map(|(seq, _)| seq).collect() + } + + /// Every outstanding gap, node-wide, rendered for the `/health` 503 cause + /// and the latch/discharge logs: `s1@13540699(snapshot_required)`. Empty + /// string ⇒ nothing outstanding. + pub(crate) fn summary(&self) -> String { + let gaps = self.lock(); + gaps.iter() + .map(|(&(shard, from_seqno), &reason)| { + format!("s{}@{from_seqno}({})", shard.0, reason.as_str()) + }) + .collect::>() + .join(", ") + } + + /// The least-dischargeable of two reasons for the same frontier: a + /// structural gap (quarantine, divergent suffix, operator request) can never + /// be discharged by a pull, so it always wins over `snapshot_required`. + const fn strongest(held: ReseedReason, incoming: ReseedReason) -> ReseedReason { + if held.stream_dischargeable() { + incoming + } else { + held + } + } + + fn range( + gaps: &BTreeMap<(ShardId, u64), ReseedReason>, + shard: ShardId, + ) -> impl Iterator + '_ { + gaps.range((shard, 0)..=(shard, u64::MAX)) + .map(|(&(_, from_seqno), &reason)| (from_seqno, reason)) + } + + fn collapse( + gaps: &BTreeMap<(ShardId, u64), ReseedReason>, + shard: ShardId, + ) -> Option { + let (from_seqno, lowest_reason) = Self::range(gaps, shard).next()?; + let reason = Self::range(gaps, shard) + .map(|(_, reason)| reason) + .find(|reason| !reason.stream_dischargeable()) + .unwrap_or(lowest_reason); + Some(ReseedMarker { reason, from_seqno }) + } +} + /// One shard group's full replication machinery, hosted inside a [`ClusterNode`]. /// /// A node holds one `ShardReplica` per group it replicates. It owns one region's @@ -405,22 +629,35 @@ pub struct ShardReplica { /// keeps perpetually false). Non-install boots ignore it (ready on today's /// terms). converged: AtomicBool, - /// m12 reseed-loop-fix (readiness gating): mirrors whether a reseed marker - /// (`SnapshotRequired` or `Quarantine`) is currently latched. `is_ready` - /// returns 503 while set, so a node that needs a reseed — INCLUDING a plain - /// restart (`install_boot == false`) that re-latched `SnapshotRequired` while - /// merely behind — is drained from the client VIP until it heals: it reseeds - /// on the next boot, or it catches up via the stream (which clears the marker - /// through `clear_stale_reseed_marker_if_caught_up`). Maintained by - /// `latch_reseed_marker` (set) and the marker-clear paths (cleared); - /// initialized at boot from the durable store. Readiness is thus BOUNDED - /// STALENESS (lag ≤ `learner_promote_lag` once converged, no unhealed reseed - /// marker), not "ready the instant the process is up". - reseed_marker_latched: AtomicBool, + /// m12 reseed-loop-fix (readiness gating), per-gap since task 18: the + /// NODE-WIDE set of outstanding reseed gaps, shared by every co-hosted + /// [`ShardReplica`] and keyed by `(shard, from_seqno)`. `is_ready` returns + /// 503 while THIS group has any outstanding gap, so a node that needs a + /// reseed — INCLUDING a plain restart (`install_boot == false`) that latched + /// `SnapshotRequired` while merely behind — is drained from the client VIP + /// until it heals: it reseeds on the next boot, or a completed catch-up pull + /// proves the stream served the gap (`discharge_reseed_marker_if_served`). + /// Maintained by `latch_reseed_marker` (insert) and + /// `discharge_reseed_marker_if_served` (remove); seeded at boot from this + /// group's durable marker. Readiness is thus BOUNDED STALENESS (lag ≤ + /// `learner_promote_lag` once converged, no unhealed reseed gap), not "ready + /// the instant the process is up". + /// + /// Node-wide rather than per-replica because the gate it feeds is node-wide + /// (one `/health`, one client VIP) and because the 2026-08-31 outage was + /// invisible precisely as long as one group's latch could not be named from + /// the node's own surfaces. See [`ReseedGapSet`]. + reseed_gaps: Arc, /// Whether a `reseed_self_restart` was REFUSED by the §2.4 quorum check — /// surfaced in status so a stuck self-restart is diagnosable. Set when the /// node would have exited but the remaining voters cannot sustain quorum. + /// A refusal is RE-EVALUATED on a standing timer while it holds (see + /// [`Self::spawn_refusal_retry`]) — a refusal that only ever re-evaluated on + /// the next latch strands the node when no further refusal arrives. self_restart_refused: AtomicBool, + /// Whether a refusal-retry loop is already running for this replica, so a + /// re-refusal never stacks a second one. + self_restart_retry_running: AtomicBool, /// The transport's `snapshot-required` refusal sink cell (m11p5 §2.4), /// late-bound by [`Self::start_election_driver`] with a sink that latches /// this node's reseed marker. @@ -576,7 +813,9 @@ impl ShardReplica { // Linear construction sequence (validate → db → sources → transport → // receiver → queue); splitting it would scatter the ordering invariants. #[allow(clippy::too_many_lines, clippy::too_many_arguments)] - pub fn new( + // Only `ClusterNode::new` builds a replica (it owns the group set and the + // shared `ReseedGapSet` every replica latches into). + pub(crate) fn new( topology: &TopologySpec, region_name: &str, schema: Schema, @@ -587,6 +826,7 @@ impl ShardReplica { enable_metrics: bool, multi_shard: bool, creds: Arc, + reseed_gaps: Arc, ) -> Result { super::topology::validate_multiproc(topology, region_name)?; @@ -1188,9 +1428,26 @@ impl ShardReplica { // marker so `is_ready` reflects a still-latched marker at boot — a // degraded install-fallback that kept its marker (the reseed did NOT // complete) must boot 503, not serve stale data. Maintained live by the - // latch/clear paths thereafter. + // latch/discharge paths thereafter. + // + // Task 18: the durable slot is ONE marker per group, so a boot can only + // recover the conservative collapse of whatever set was outstanding when + // the process died — the lowest resume point with the + // least-dischargeable reason (see [`ReseedGapSet`]). That single gap is the + // honest recovery: everything above it is covered by the reseed it drives. let reseed_marker_store = ReseedMarkerStore::new(&data_dir_for_state); - let reseed_marker_latched_init = matches!(reseed_marker_store.load(), Ok(Some(_))); + if let Ok(Some(marker)) = reseed_marker_store.load() { + reseed_gaps.latch(group.shard, marker.reason, marker.from_seqno); + tracing::warn!( + region = region_name, + shard = group.shard.0, + reason = marker.reason.as_str(), + from_seqno = marker.from_seqno, + "boot: this group carries an unhealed reseed gap from the durable marker — \ + readiness stays 503 for it until a catch-up pull serves the gap or the \ + reseed runs" + ); + } Ok(Self { region, @@ -1230,13 +1487,14 @@ impl ShardReplica { quorum_timeout, shutting_down: AtomicBool::new(false), reseed_marker_store, - reseed_marker_latched: AtomicBool::new(reseed_marker_latched_init), + reseed_gaps, reseed_self_restart: topology.replication.reseed_self_restart.unwrap_or(false), learner_promote_lag: topology.replication.learner_promote_lag.unwrap_or(1024), seed_joiner, install_boot, converged: AtomicBool::new(false), self_restart_refused: AtomicBool::new(false), + self_restart_retry_running: AtomicBool::new(false), snapshot_required_cell, catchup_served_cell, membership, @@ -3099,69 +3357,108 @@ impl ShardReplica { ); } - /// Durably latch the reseed marker (m11p5 §2.4), idempotently. + /// Latch a reseed gap for this shard group (m11p5 §2.4), idempotently. /// - /// Sets the `tidaldb_cluster_reseed_required` gauge and persists the marker - /// (reason + the seqno the next-boot fetch resumes from). Idempotent: a - /// re-latch from the same or a different reason simply rewrites the marker. - /// On `reseed_self_restart`, additionally evaluates the §2.4 quorum refusal - /// and either drains+exits or refuses (loudly, in status). + /// Records `(this group, from_seqno)` in the node-wide [`ReseedGapSet`], + /// re-derives this group's durable marker from the resulting set, and sets + /// the `tidaldb_cluster_reseed_required` gauge. On `reseed_self_restart`, + /// additionally evaluates the §2.4 quorum refusal and either drains+exits or + /// refuses (loudly, in status). /// - /// A failed persist is logged at ERROR but never panics — the node keeps - /// serving degraded; the runtime retry timer keeps its standing wake-up, so - /// a transient fsync failure does not strand the node (the next refusal - /// re-latches). + /// Idempotent per GAP, not per call: re-latching the same frontier (the + /// standing catch-up retry does this every `catchup_retry_ms`) changes + /// nothing and logs at DEBUG, so the WARN — and therefore the operator's + /// latched count — is one per distinct gap and comparable with the + /// discharged count. A latch at a DIFFERENT frontier is a DIFFERENT gap and + /// is tracked alongside the first instead of overwriting it; overwriting is + /// what made one discharge clear two latches (see [`ReseedGapSet`]). + /// + /// A failed persist is logged at ERROR but never panics — the gap stays + /// outstanding (so readiness stays drained, the conservative direction) and + /// the standing retry timer re-latches on the next refusal, which re-persists. pub(crate) fn latch_reseed_marker(self: &Arc, reason: ReseedReason, from_seqno: u64) { // Readiness gating (m12 reseed-loop-fix): the node is operationally in the // reseed-required state the instant it latches, regardless of whether the - // durable persist below succeeds — flip readiness to 503 now so it drains - // from the client VIP. Cleared by `clear_stale_reseed_marker_if_caught_up` - // once it catches up via the stream, or consumed by the next-boot reseed. - self.reseed_marker_latched.store(true, Ordering::Release); - let marker = ReseedMarker { reason, from_seqno }; - match self.reseed_marker_store.persist(marker) { - Ok(()) => { - self.cluster_metrics.set_reseed_required(true); - tracing::warn!( - region = %self.region_name, - reason = reason.as_str(), - from_seqno, - "reseed marker latched durably (m11p5 §2.4); the reseed runs on the next \ - boot. Serving stays degraded until then." - ); - } - Err(e) => { - tracing::error!( - region = %self.region_name, - reason = reason.as_str(), - error = %e, - "failed to durably latch the reseed marker; the node keeps serving degraded \ - and the standing retry timer re-latches on the next refusal" - ); - // Reflect the latch intent in the gauge even on a persist miss - // (the node IS in the reseed-required state operationally). - self.cluster_metrics.set_reseed_required(true); - return; - } + // durable persist below succeeds — the gap is in the set now, so + // `is_ready` already answers 503 and the node drains from the client VIP. + // Removed only by `discharge_reseed_marker_if_served` (positive evidence) + // or consumed by the next-boot reseed. + let outcome = self.reseed_gaps.latch(self.group_shard, reason, from_seqno); + self.cluster_metrics.set_reseed_required(true); + // Persist on EVERY latch, not only a fresh one. A prior persist may have + // FAILED, and the standing catch-up retry re-latching the same frontier is + // exactly the re-persist opportunity that failure path relies on — the + // write is idempotent (same bytes for an unchanged set), so a repeat costs + // one fsync per retry tick and buys back the durable marker. + if let Err(e) = self.reseed_marker_store.persist(outcome.durable) { + tracing::error!( + region = %self.region_name, + shard = self.group_shard.0, + reason = reason.as_str(), + from_seqno, + error = %e, + "failed to durably latch the reseed marker; the gap stays outstanding (readiness \ + stays 503) and the standing retry timer re-latches, which re-persists" + ); + // The gauge already reflects the latch intent above (the node IS in + // the reseed-required state operationally, persisted or not). + return; + } + if outcome.fresh { + tracing::warn!( + region = %self.region_name, + shard = self.group_shard.0, + reason = reason.as_str(), + from_seqno, + outstanding = outcome.outstanding, + durable_from_seqno = outcome.durable.from_seqno, + durable_reason = outcome.durable.reason.as_str(), + gaps = %self.reseed_gaps.summary(), + "reseed marker latched durably (m11p5 §2.4); the reseed runs on the next \ + boot. Serving stays degraded until then." + ); + } else { + // Deliberately DEBUG: the retry re-latches the same gap every + // `catchup_retry_ms`, and one WARN per gap is what makes the operator's + // latched count comparable with the discharged count. + tracing::debug!( + region = %self.region_name, + shard = self.group_shard.0, + reason = reason.as_str(), + from_seqno, + outstanding = outcome.outstanding, + "reseed gap already outstanding at this frontier; marker re-persisted, no new gap" + ); } if self.reseed_self_restart { self.spawn_self_restart_eval(); } } - /// Discharge a reseed marker on POSITIVE EVIDENCE that the stream served the - /// range which latched it: a `StreamSegments` pull that began at or below the - /// marker's `from_seqno` and ran to completion. + /// Discharge every outstanding gap on this shard group that a COMPLETED + /// catch-up pull proves served: a `StreamSegments` pull that began at or + /// below the gap's `from_seqno` and ran to completion. /// /// This exists for the genuine false-alarm case (the `own < prev_log → /// ReseedRequired` join arm latches for a node merely BEHIND by a shippable /// tail, which the stream then serves), without the unsoundness of the two /// frontier comparisons that preceded it. /// + /// # Why every gap, and not "the marker" + /// + /// One completed pull is evidence about a RANGE, and a group can hold more + /// than one gap. Deciding from the single durable slot — which a re-latch + /// overwrote — meant one pull either cleared a gap it had not served or + /// cleared none of the gaps it HAD served, depending only on which latch + /// wrote last. So the evidence is applied to each outstanding gap + /// individually, and the durable slot is re-derived from what remains + /// (cleared only when the group's set empties). Latch and discharge are then + /// the same shape and the two operator-visible counts must agree. + /// /// # Why no frontier comparison can work here /// /// The first version asked `applied >= leader_last_seq`, asserting "a node - /// genuinely behind a COMPACTED gap never reaches caught_up". The second asked + /// genuinely behind a COMPACTED gap never reaches `caught_up`". The second asked /// `applied >= marker.from_seqno`. BOTH are unsound for the same reason: the /// applied frontier is a HIGH-WATER-MARK, and a term join re-bases it onto the /// new leader's stream (`replication_state().advance(.., baseline + 1)`), so it @@ -3185,37 +3482,50 @@ impl ShardReplica { { return; } - let Ok(Some(marker)) = self.reseed_marker_store.load() else { - return; - }; - if !marker.discharged_by_served_range(served_from) { - // Refuse quietly. The undischarged state is already observable: the - // latch logged WARN, `tidaldb_cluster_reseed_required` stays 1 (which - // is what makes the 10m alert reachable), and `/cluster/status/local` - // reports `reseed_required: true`. + // Refuses quietly when nothing is proven served. The undischarged state + // is already observable: the latch logged WARN, + // `tidaldb_cluster_reseed_required` stays 1 (which is what makes the 10m + // alert reachable), `/cluster/status/local` reports `reseed_required: + // true` with the outstanding frontiers, and `/health`'s 503 names them. + let outcome = self.reseed_gaps.serve(self.group_shard, served_from); + if outcome.discharged.is_empty() { return; } - match self.reseed_marker_store.clear() { - Ok(()) => { - self.cluster_metrics.set_reseed_required(false); - // Readiness gating (m12 reseed-loop-fix): the marker is healed — - // clear the readiness latch so `is_ready` can return 200 again. - self.reseed_marker_latched.store(false, Ordering::Release); - tracing::info!( - region = %self.region_name, - reason = marker.reason.as_str(), - from_seqno = marker.from_seqno, - served_from, - "reseed marker discharged — a catch-up pull completed from at/below the \ - seqno that latched it, so the stream genuinely closed the gap and no \ - reseed is needed" - ); - } - Err(e) => tracing::warn!( + // The durable slot follows the set: re-persist the conservative collapse + // of what is LEFT, or clear the file when this group is clean. A failure + // here does not un-discharge the gap — the evidence stands, so readiness + // recovers; the stale file only costs a needless reseed on the next boot + // (and that boot re-latches the gap, so it can never serve a hole). + // The gauge tracks the SET, not the file: no gap left ⇒ this group is no + // longer reseed-required, whether or not the file write below lands. + if outcome.durable.is_none() { + self.cluster_metrics.set_reseed_required(false); + } + let durable_result = outcome.durable.map_or_else( + || self.reseed_marker_store.clear(), + |remaining| self.reseed_marker_store.persist(remaining), + ); + if let Err(e) = durable_result { + tracing::warn!( region = %self.region_name, + shard = self.group_shard.0, error = %e, - "failed to clear a discharged reseed marker; retried on the next served pull" - ), + "failed to update the durable reseed marker after a discharge; readiness follows \ + the served evidence and the next boot re-latches whatever the file still names" + ); + } + for (from_seqno, reason) in outcome.discharged { + tracing::info!( + region = %self.region_name, + shard = self.group_shard.0, + reason = reason.as_str(), + from_seqno, + served_from, + remaining = %self.reseed_gaps.summary(), + "reseed marker discharged — a catch-up pull completed from at/below the \ + seqno that latched it, so the stream genuinely closed the gap and no \ + reseed is needed" + ); } } @@ -3236,6 +3546,60 @@ impl ShardReplica { } } + /// Re-evaluate a REFUSED `reseed_self_restart` on a standing timer for as + /// long as the refusal holds. + /// + /// The refusal itself is correct and stays (§2.4: exiting while the + /// remaining voters cannot sustain quorum is total write unavailability — + /// during the 2026-08-31 roll it fired with + /// `alive_voters_excluding_self=1 < majority=2` and is why the cluster stayed + /// up). What was missing is the re-drive: the refusal's own ERROR promised + /// "the reseed runs once quorum is safe (a future latch re-evaluates)", and a + /// node whose gap is real gets no future latch once the catch-up retry stops + /// producing refusals — so it waits for a quorum event nobody polls, staying + /// 503 until an operator deletes the pod by hand. A rolling deploy is exactly + /// the window that produces a refusal, so this is the common case, not the + /// exotic one. + /// + /// One loop per replica (guarded by `self_restart_retry_running`), exiting as + /// soon as the gap heals, the refusal clears, or the node shuts down. + fn spawn_refusal_retry(self: &Arc) { + if self.self_restart_retry_running.swap(true, Ordering::AcqRel) { + return; // a loop is already re-evaluating this replica's refusal + } + let node = Arc::clone(self); + if let Err(e) = std::thread::Builder::new() + .name("tidal-reseed-refusal-retry".into()) + .spawn(move || { + loop { + std::thread::sleep(REFUSAL_RETRY_INTERVAL); + if !refusal_retry_continues( + node.is_shutting_down(), + node.self_restart_refused.load(Ordering::Acquire), + node.reseed_gaps.outstanding_for(node.group_shard), + ) { + break; + } + tracing::info!( + region = %node.region_name, + shard = node.group_shard.0, + gaps = %node.reseed_gaps.summary(), + "reseed_self_restart: re-evaluating the quorum refusal (the gap is still \ + outstanding); the reseed proceeds as soon as the remaining voters can \ + sustain quorum without this node" + ); + node.maybe_self_restart(); + } + node.self_restart_retry_running + .store(false, Ordering::Release); + }) + { + self.self_restart_retry_running + .store(false, Ordering::Release); + tracing::error!(error = %e, "failed to spawn reseed refusal-retry thread"); + } + } + /// Evaluate the §2.4 quorum refusal and, when safe, trigger a graceful /// drain + clean exit(0) so the node restarts and reseeds. /// @@ -3261,11 +3625,14 @@ impl ShardReplica { alive_voters_excluding_self = alive_others, total_voters, majority = reseed::majority(total_voters), + retry_secs = REFUSAL_RETRY_INTERVAL.as_secs(), "reseed_self_restart REFUSED: the remaining voters cannot sustain quorum without \ this node (alive_others < majority). Keeping the marker latched and serving \ degraded — exiting now would be total write unavailability (§2.4). The reseed \ - runs once quorum is safe (a future latch re-evaluates)." + runs once quorum is safe: this refusal is re-evaluated every retry_secs until \ + the gap heals or the exit fires." ); + self.spawn_refusal_retry(); return; } self.self_restart_refused.store(false, Ordering::Release); @@ -3273,14 +3640,14 @@ impl ShardReplica { // A snapshot_required latch from a TRANSIENT classification — the leader's // baseline advanced past this node's persisted frontier while it was down, // so the first heartbeat's decide_join saw it briefly behind — self-heals - // via the stream: `clear_stale_reseed_marker_if_caught_up` clears the marker - // once caught up, journaling the durable term marker on the clean join. The + // via the stream: `discharge_reseed_marker_if_served` removes the gap once a + // completed pull proves the range was served. The // quorum poll above took time (a blocking peer fan-out); if the marker // healed meanwhile, a reboot would reseed NOTHING (the leader answers // needed=false for a caught-up shard), so self-restarting is futile and // loops. Abort. Only a marker that CANNOT self-heal (a genuine compacted // gap, still latched here) proceeds to the restart. - if !self.reseed_marker_latched.load(Ordering::Acquire) { + if !self.reseed_gaps.outstanding_for(self.group_shard) { tracing::info!( region = %self.region_name, "reseed_self_restart: the reseed marker healed via stream catch-up during the \ @@ -3329,7 +3696,7 @@ impl ShardReplica { // healed, a reboot would reseed nothing (caught up), so abort // the futile self-restart. Only a still-latched marker (a // genuine compacted gap) fires the exit. - if !node.reseed_marker_latched.load(Ordering::Acquire) { + if !node.reseed_gaps.outstanding_for(node.group_shard) { tracing::info!( region = %node.region_name, "reseed_self_restart: the reseed marker healed via stream \ @@ -3588,7 +3955,7 @@ impl ShardReplica { // about to discard — drain it from the client VIP until it heals. The latch // clears when a completed catch-up pull proves the stream served the gap // (`discharge_reseed_marker_if_served`) or the next boot reseeds. - if self.reseed_marker_latched.load(Ordering::Acquire) { + if self.reseed_gaps.outstanding_for(self.group_shard) { return false; } // POSITIVE EVIDENCE, every boot. This was @@ -3632,6 +3999,31 @@ impl ShardReplica { true } + /// Why THIS group is unready, for the `/health` 503 body. + /// + /// Reads every gate [`Self::is_ready`] tests, in the same order, so the cause + /// is the gate that actually holds — see [`unready_cause`] for what a + /// fall-through arm cost during the 2026-08-31 incident. + fn unready_cause(&self) -> String { + let gaps = self.reseed_gaps.summary(); + unready_cause( + self.group_shard, + &UnreadyGates { + shutting_down: self.is_shutting_down(), + quarantined: self + .election_runtime + .get() + .is_some_and(|rt| rt.is_quarantined()), + removed: self.membership.self_role() + == Some(tidaldb::wal::format::MemberRole::Removed), + decommissioned: self.decommissioned_by_signal.load(Ordering::Acquire), + reseed_gaps: (!gaps.is_empty()).then_some(gaps.as_str()), + self_restart_refused: self.self_restart_refused.load(Ordering::Acquire), + converged: self.converged.load(Ordering::Acquire), + }, + ) + } + /// Start the election driver (m11p4). Called once the node is in its /// final `Arc` (the driver holds a `Weak` back-reference). pub fn start_election_driver(self: &Arc) { @@ -3845,10 +4237,17 @@ impl ShardReplica { ); // m11p5 §4: feed the sticky readiness latch from the lag we just - // computed (no separate polling thread) and report the durable reseed - // state. + // computed (no separate polling thread) and report the reseed state. self.note_lag_for_readiness(leader_seqno, lag_events); - let reseed_required = self.reseed_marker_store.exists(); + // Task 18: the LIVE outstanding-gap set, not `reseed_marker_store.exists()`. + // The gap set is what gates readiness, is seeded at boot from the durable + // marker, and is what the `tidaldb_cluster_reseed_required` gauge already + // tracked (the gauge is set even when the persist fails). Reporting the + // file instead let status answer `reseed_required: false` while `/health` + // answered 503 — the pair of readings that made the 2026-08-31 outage + // look like a phantom. + let reseed_gaps = self.reseed_gaps.seqnos_for(self.group_shard); + let reseed_required = !reseed_gaps.is_empty(); // `reseeding` = this node has not yet first-converged against a KNOWN // leader frontier. No longer scoped to joiner boots: a plain restart is // equally un-converged until it learns where the leader is, and reporting @@ -3886,6 +4285,7 @@ impl ShardReplica { .map(|(k, v)| (u32::from(k.0), v)) .collect(), reseed_required, + reseed_gaps, reseeding, self_restart_refused: self.self_restart_refused.load(Ordering::Acquire), membership_version: self.membership_version(), @@ -4526,6 +4926,13 @@ impl ClusterNode { // everywhere. let creds = Arc::new(crate::cluster::security::ClusterCreds::from_env()); + // Task 18: ONE outstanding-gap set for the whole process, shared with + // every hosted replica. Node-wide because the gate it feeds is node-wide + // (one `/health`, one client VIP) — and because a per-replica latch could + // not be named from the node's own surfaces, which is what turned a + // two-gap rolling deploy into an unexplained three-pod outage. + let reseed_gaps = Arc::new(ReseedGapSet::new()); + let mut groups: BTreeMap> = BTreeMap::new(); let mut metrics_owner: Option = None; for group in &resolved { @@ -4560,6 +4967,7 @@ impl ClusterNode { enable_metrics, !single, Arc::clone(&creds), + Arc::clone(&reseed_gaps), )?; if enable_metrics { metrics_owner = Some(group.shard); @@ -4842,6 +5250,7 @@ impl ClusterNode { // hosted group was stuck behind a compacted leader, and every // operator (and every diagnosis) read it as converged. reseed_required: s.reseed_required, + reseed_gaps: s.reseed_gaps, reseeding: s.reseeding, }) }) @@ -5159,6 +5568,93 @@ async fn admin_gate( // ── Health ────────────────────────────────────────────────────────────────── +/// Every readiness gate [`ShardReplica::is_ready`] tests, sampled once so the +/// 503 cause can name the one that actually holds. +// One bool per INDEPENDENT gate, mirroring `ShardReplica`'s own allow: these are +// six separately-latched facts, not a state machine, and any two can hold at +// once (a draining node with an outstanding gap). Packing them into flags or an +// enum would force false either/or relations between orthogonal latches and make +// the ladder harder to read against `is_ready`, which tests them one at a time. +#[allow(clippy::struct_excessive_bools)] +pub struct UnreadyGates<'a> { + /// This group is draining (process shutdown or a reseed self-restart). + pub shutting_down: bool, + /// The m11p4 divergence quarantine fenced this group. + pub quarantined: bool, + /// A `Removed` record retired this node (m11p5 §3.3). + pub removed: bool, + /// A peer's typed removed signal reached this node, which never folded the + /// record itself (the §3.3 MISSED-RECORD path). + pub decommissioned: bool, + /// `Some(summary)` ⇒ outstanding reseed gaps, already rendered + /// ([`ReseedGapSet::summary`]). `None` ⇒ none outstanding. + pub reseed_gaps: Option<&'a str>, + /// A `reseed_self_restart` was refused by the §2.4 quorum check, so the + /// pending reseed is deliberately waiting on quorum. + pub self_restart_refused: bool, + /// This group has first-converged against a KNOWN leader frontier (§4). + pub converged: bool, +} + +/// Render why a shard group is unready, testing the gates in +/// [`ShardReplica::is_ready`]'s own order. +/// +/// # Why this is not a fall-through ladder +/// +/// It used to be. Four arms tested three gates and everything else printed +/// `"joiner boot not yet converged"`, so the arm that fired in production was +/// one the ladder never checked: `reseed_marker_latched`. On 2026-08-31 all +/// three pods answered `503 {"cause":"joiner boot not yet converged"}` with +/// `lag=0` on every shard, `quarantined: false`, `membership_role: voter`, and +/// each group having already logged "first-converged against a KNOWN leader +/// frontier ... readiness is now sticky-ready". The cause was false, it was +/// confidently false, and it cost real diagnosis time. +/// +/// So every gate gets its own arm, the reseed arm NAMES the outstanding gaps, +/// and the terminal arm reports that no gate is holding rather than inventing +/// one. "Reason unavailable" is a worse answer to read and a better answer to +/// trust. +pub fn unready_cause(shard: ShardId, gates: &UnreadyGates<'_>) -> String { + if gates.shutting_down { + return "shutting down".to_string(); + } + if gates.quarantined { + return "quarantined (divergent suffix); reseeds on next boot".to_string(); + } + if gates.removed { + return "removed from the cluster (decommissioned)".to_string(); + } + if gates.decommissioned { + return "decommissioned by a peer's removed signal; this node never folded the Removed \ + record. Recovery: delete this node (runbook §8 'decommission')" + .to_string(); + } + if let Some(gaps) = gates.reseed_gaps { + let waiting = if gates.self_restart_refused { + "; reseed_self_restart is REFUSED (the remaining voters cannot sustain quorum \ + without this node) and is re-evaluated on a timer" + } else { + "" + }; + return format!( + "reseed_marker_latched: shard group {} holds an unhealed reseed gap. Outstanding \ + node-wide: [{gaps}]. Drained from the client VIP until a completed catch-up pull \ + serves each gap or the reseed runs{waiting}", + shard.0 + ); + } + if !gates.converged { + return format!( + "shard group {} has not yet first-converged against a KNOWN leader frontier", + shard.0 + ); + } + format!( + "unready: reason unavailable — no readiness gate on shard group {} is holding; re-poll", + shard.0 + ) +} + #[allow(clippy::significant_drop_tightening)] async fn region_health( State(node): State>, @@ -5168,28 +5664,30 @@ async fn region_health( // boot has not yet first-converged (sticky-ready after). m11p6: the node is // ready only when EVERY hosted shard group is ready (aggregate). if !node.is_ready() { - // Inspect the group ACTUALLY keeping the node unready (S>1: not - // necessarily the first hosted group) so the 503 cause is accurate; - // fall back to the default group if only the node-level drain flag is set. - let unready = node - .hosted() - .find(|r| !r.is_ready()) - .cloned() - .unwrap_or_else(|| Arc::clone(&state)); - let cause = if unready.is_shutting_down() { - "shutting down" - } else if unready - .election_runtime - .get() - .is_some_and(|rt| rt.is_quarantined()) - { - "quarantined (divergent suffix); reseeds on next boot" - } else if unready.membership().self_role() - == Some(tidaldb::wal::format::MemberRole::Removed) - { - "removed from the cluster (decommissioned)" - } else { - "joiner boot not yet converged" + // Report the gate that is ACTUALLY holding this node, and nothing else. + // + // The pre-task-18 ladder fell through to "joiner boot not yet converged" + // for every cause it did not test — including a latched reseed marker, + // which is the one that actually fired. On 2026-08-31 all three pods + // answered `503 {"cause":"joiner boot not yet converged"}` while every + // hosted group had already logged "first-converged against a KNOWN leader + // frontier ... readiness is now sticky-ready". A probe that names a state + // it never checked is worse than one that says "reason unavailable": it + // sends the operator somewhere else, with confidence. + let cause = match node.hosted().find(|r| !r.is_ready()) { + Some(unready) => unready.unready_cause(), + // `ClusterNode::is_ready` is `!shutting_down && all hosted ready`, so + // no unready group means the node-level drain is the gate. Test it + // rather than asserting it — a group can flip ready between the two + // reads, and that race must not be reported as a shutdown. + None if node.shutting_down.load(Ordering::Acquire) => { + "node-level drain: the process is shutting down (every hosted shard group \ + reports ready)" + .to_string() + } + None => "unready: reason unavailable — every hosted shard group reported ready \ + after the node-level check; re-poll" + .to_string(), }; return Ok(( StatusCode::SERVICE_UNAVAILABLE, @@ -5316,10 +5814,20 @@ pub struct LocalStatusResponse { /// Whether this node is quarantined with a divergent suffix (m11p4): /// fenced from the data plane until reseeded. quarantined: bool, - /// Whether this node has durably latched a `reseed_required` marker (m11p5 - /// §2.4): a snapshot reseed runs on its next boot; it serves degraded until - /// then. Mirrors the `tidaldb_cluster_reseed_required` gauge. + /// Whether this node still holds an unhealed reseed gap for the group + /// `replica_for(None)` resolves (m11p5 §2.4): a snapshot reseed runs on its + /// next boot; it serves degraded until then. Mirrors the + /// `tidaldb_cluster_reseed_required` gauge and the readiness gate — both read + /// the live outstanding-gap set, so this can no longer answer `false` while + /// `/health` answers 503 for a latched marker. reseed_required: bool, + /// The exact frontiers that group is still waiting to have served, lowest + /// first. `reseed_required: true` with `[13322280, 13540699]` says the group + /// holds TWO gaps and names both; the pre-task-18 surface could only say + /// "true" and could not say how many, which is why `latched: 2, discharged: + /// 1` had to be reconstructed from log counts. + #[serde(default)] + reseed_gaps: Vec, /// Whether this node is mid-reseed: an install boot whose post-snapshot /// catch-up has not yet first-converged (m11p5 §4). Readiness is 503 while /// this is true. @@ -5391,6 +5899,11 @@ pub struct ShardStatusRow { /// pre-fix peer during a mixed-version window. #[serde(default)] reseed_required: bool, + /// THIS GROUP's outstanding gap frontiers, lowest first — the per-group + /// counterpart of `reseed_required`, so an operator can see WHICH frontiers a + /// group is waiting on rather than inferring the count from log greps. + #[serde(default)] + reseed_gaps: Vec, /// Whether THIS GROUP is mid-reseed (install/seed-join not yet converged). #[serde(default)] reseeding: bool, @@ -9713,3 +10226,10 @@ mod forward_retry_tests { ); } } + +/// Task 18 (reseed latch/discharge symmetry + an honest 503 cause). Split out +/// under `#[path]` rather than grown inline: this file is already ~10k lines, +/// and the same convention keeps `tidal/src/wal/segment.rs` readable. +#[cfg(test)] +#[path = "node_reseed_gap_tests.rs"] +mod node_reseed_gap_tests; diff --git a/tidal-server/src/cluster/node_reseed_gap_tests.rs b/tidal-server/src/cluster/node_reseed_gap_tests.rs new file mode 100644 index 0000000..91dbee7 --- /dev/null +++ b/tidal-server/src/cluster/node_reseed_gap_tests.rs @@ -0,0 +1,510 @@ +//! Task 18 regression gates: the reseed latch and its discharge must be the +//! same shape, and the `/health` 503 cause must name the gate that actually +//! holds. +//! +//! # What broke, measured +//! +//! During the 2026-08-31 staged roll, `tidaldb-2` logged: +//! +//! ```text +//! latched: 2 (from_seqno 13322280 and 13540699) +//! discharged: 1 +//! ``` +//! +//! and then, indefinitely: +//! +//! ```text +//! /health -> 503 {"cause":"joiner boot not yet converged"} +//! /cluster/status/local -> reseed_required=false, reseeding=false, quarantined=false, +//! membership_role=voter, shards: s0 lag=0, s1 lag=0, s2 lag=0 +//! ``` +//! +//! Every group had `lag=0` and had already logged "first-converged against a +//! KNOWN leader frontier ... readiness is now sticky-ready". The node was +//! healthy by every measure it exposed and permanently unready; only `kubectl +//! delete pod` cleared it, on all three pods. +//! +//! # What these tests drive +//! +//! The real [`ReseedGapSet`] and the real [`unready_cause`] — the exact +//! functions `latch_reseed_marker`, `discharge_reseed_marker_if_served`, +//! `ShardReplica::is_ready` and `region_health` call. The node-level aggregate is +//! composed here the way [`ClusterNode::is_ready`] composes it (`all hosted +//! groups ready`, each group gated on its own entries in the shared set), because +//! constructing three real `ShardReplica`s means three `TidalDb` opens and three +//! bound gRPC listeners — a multi-process integration concern, not a unit one. +//! Every predicate under test is production code; only the `all(..)` fold is +//! restated. + +use super::*; + +/// Production's own numbers, so a future reader can match them against the +/// incident log. +const GAP_LOW: u64 = 13_322_280; +const GAP_HIGH: u64 = 13_540_699; + +const S0: ShardId = ShardId(0); +const S1: ShardId = ShardId(1); +const S2: ShardId = ShardId(2); + +/// The node-level readiness predicate over reseed gaps, composed exactly as +/// [`ClusterNode::is_ready`] composes it: every hosted group must be clear. +fn node_ready(gaps: &ReseedGapSet, hosted: &[ShardId]) -> bool { + hosted.iter().all(|&shard| !gaps.outstanding_for(shard)) +} + +/// THE regression gate: a node hosting three groups takes TWO simultaneous gaps +/// on DIFFERENT groups, has each served, and becomes ready again — with the +/// latched and discharged counts equal and no restart anywhere in the story. +/// +/// Before task 18 this could not pass: the two latches shared one durable slot +/// per node-visible state, the second overwrote the first, and one discharge +/// either cleared a gap it had not served or cleared none of the gaps it had. +#[test] +fn multi_group_node_recovers_from_two_simultaneous_gaps() { + let gaps = ReseedGapSet::new(); + let hosted = [S0, S1, S2]; + let mut latched = 0usize; + let mut discharged = 0usize; + + // Two groups latch at different frontiers, exactly as a rolling deploy of a + // fully-placed 3-node cluster produces. + for (shard, from_seqno) in [(S0, GAP_LOW), (S1, GAP_HIGH)] { + let outcome = gaps.latch(shard, ReseedReason::SnapshotRequired, from_seqno); + assert!(outcome.fresh, "a new gap on s{} must latch", shard.0); + latched += 1; + } + assert_eq!(latched, 2); + assert!( + !node_ready(&gaps, &hosted), + "a node holding stale data it is about to discard must stay drained from \ + the client VIP (the safety property note_lag_for_readiness records)" + ); + assert!(gaps.outstanding_for(S0) && gaps.outstanding_for(S1)); + assert!( + !gaps.outstanding_for(S2), + "s2 never latched, so it must not be gated by its siblings' gaps" + ); + + // The first catch-up pull serves s1's gap only. s0's gap is untouched: the + // pull says nothing about another group's stream. + let served_s1 = gaps.serve(S1, GAP_HIGH); + discharged += served_s1.discharged.len(); + assert_eq!( + served_s1.discharged, + vec![(GAP_HIGH, ReseedReason::SnapshotRequired)] + ); + assert!( + served_s1.durable.is_none(), + "s1 has nothing left outstanding, so its durable marker file is cleared" + ); + assert!( + !node_ready(&gaps, &hosted), + "ONE discharge must not clear TWO latches — s0's gap is still unserved" + ); + + // The second pull serves s0's gap. Now every hosted group is clear. + let served_s0 = gaps.serve(S0, GAP_LOW); + discharged += served_s0.discharged.len(); + assert_eq!( + served_s0.discharged, + vec![(GAP_LOW, ReseedReason::SnapshotRequired)] + ); + assert!(served_s0.durable.is_none()); + + assert_eq!( + latched, discharged, + "every latched gap must be discharged by evidence about itself \ + (production read latched: 2, discharged: 1)" + ); + assert!( + node_ready(&gaps, &hosted), + "a node with every shard served, unquarantined and not removed must become \ + READY with no manual restart" + ); + assert_eq!( + gaps.summary(), + "", + "nothing outstanding ⇒ nothing to report" + ); +} + +/// Two gaps on ONE group also need two discharges, and a pull that starts +/// BETWEEN them proves only the upper one. +/// +/// This is the unsound direction of the old single-slot design: with the durable +/// slot holding the LOWER frontier, a pull in `(low, high]` discharged the slot +/// and re-admitted the node while the lower gap was still a hole. +#[test] +fn two_gaps_on_one_group_need_two_discharges() { + let gaps = ReseedGapSet::new(); + assert!( + gaps.latch(S1, ReseedReason::SnapshotRequired, GAP_LOW) + .fresh + ); + let second = gaps.latch(S1, ReseedReason::SnapshotRequired, GAP_HIGH); + assert!(second.fresh); + assert_eq!(second.outstanding, 2); + assert_eq!( + second.durable.from_seqno, GAP_LOW, + "the single durable slot must hold the LOWEST resume point: a reseed from \ + there covers every gap above it" + ); + + // A pull that began above the low gap cannot have served it. + let between = gaps.serve(S1, GAP_LOW + 1); + assert_eq!( + between.discharged, + vec![(GAP_HIGH, ReseedReason::SnapshotRequired)], + "only the gap at or above the pull's start is proven served" + ); + assert_eq!( + between.durable.map(|m| m.from_seqno), + Some(GAP_LOW), + "the durable slot follows what REMAINS, it is not cleared" + ); + assert!(gaps.outstanding_for(S1), "the low gap is still a hole"); + + let below = gaps.serve(S1, GAP_LOW); + assert_eq!( + below.discharged, + vec![(GAP_LOW, ReseedReason::SnapshotRequired)] + ); + assert!(below.durable.is_none()); + assert!(!gaps.outstanding_for(S1)); +} + +/// The standing catch-up retry re-latches the same frontier every +/// `catchup_retry_ms`. That is not a new gap: it must not inflate the latched +/// count, or "latched == discharged" is unmeasurable. +#[test] +fn re_latching_the_same_frontier_is_not_a_new_gap() { + let gaps = ReseedGapSet::new(); + assert!( + gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_LOW) + .fresh + ); + for _ in 0..5 { + let repeat = gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_LOW); + assert!(!repeat.fresh, "the same frontier is the same gap"); + assert_eq!(repeat.outstanding, 1); + } + let served = gaps.serve(S0, GAP_LOW); + assert_eq!( + served.discharged.len(), + 1, + "one gap ⇒ exactly one discharge, however many times it was re-latched" + ); +} + +/// Structural gaps are never dischargeable by the stream, per +/// [`ReseedReason::stream_dischargeable`]. A pull from seqno 1 must not clear a +/// quarantine, a divergent suffix, or an operator reseed request. +#[test] +fn structural_gaps_are_never_discharged_by_a_pull() { + for reason in [ + ReseedReason::Quarantine, + ReseedReason::Operator, + ReseedReason::DivergentPostBaseline, + ] { + let gaps = ReseedGapSet::new(); + assert!(gaps.latch(S0, reason, GAP_LOW).fresh); + for served_from in [1, GAP_LOW - 1, GAP_LOW, GAP_HIGH] { + let outcome = gaps.serve(S0, served_from); + assert!( + outcome.discharged.is_empty(), + "{} discharged by a pull from {served_from}", + reason.as_str() + ); + } + assert!( + gaps.outstanding_for(S0), + "{} must keep the node drained until it reseeds", + reason.as_str() + ); + } +} + +/// A structural reason arriving at a frontier already held as +/// `snapshot_required` STRENGTHENS the gap — it is now a state no forward +/// progress resolves, and a reboot (which re-seeds the set from the single +/// durable slot) must not read it back as dischargeable. +#[test] +fn a_structural_reason_wins_at_the_same_frontier() { + let gaps = ReseedGapSet::new(); + assert!( + gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_LOW) + .fresh + ); + let upgraded = gaps.latch(S0, ReseedReason::Quarantine, GAP_LOW); + assert!( + upgraded.fresh, + "a strengthened reason is new information and must be re-persisted" + ); + assert_eq!(upgraded.outstanding, 1, "same frontier ⇒ still one gap"); + assert_eq!(upgraded.durable.reason, ReseedReason::Quarantine); + assert!(gaps.serve(S0, 1).discharged.is_empty()); + + // And the reverse order never downgrades it. + let downgrade = gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_LOW); + assert!(!downgrade.fresh); + assert_eq!(downgrade.durable.reason, ReseedReason::Quarantine); + assert!(gaps.serve(S0, 1).discharged.is_empty()); +} + +/// The durable slot collapses a group's set conservatively: LOWEST resume point, +/// LEAST-dischargeable reason. Either component chosen the other way loses a gap +/// across a reboot. +#[test] +fn the_durable_slot_is_the_conservative_collapse() { + let gaps = ReseedGapSet::new(); + gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_LOW); + let with_structural_above = gaps.latch(S0, ReseedReason::Quarantine, GAP_HIGH); + assert_eq!( + with_structural_above.durable, + ReseedMarker { + reason: ReseedReason::Quarantine, + from_seqno: GAP_LOW + }, + "the lowest resume point covers both gaps; the structural reason keeps the \ + next boot from discharging the quarantine by a pull" + ); + + // Serving the low (dischargeable) gap leaves the structural one, and the slot + // moves up to it. + let served = gaps.serve(S0, GAP_LOW); + assert_eq!( + served.discharged, + vec![(GAP_LOW, ReseedReason::SnapshotRequired)] + ); + assert_eq!( + served.durable, + Some(ReseedMarker { + reason: ReseedReason::Quarantine, + from_seqno: GAP_HIGH + }) + ); +} + +/// Gaps are keyed by `(shard, from_seqno)`: the same frontier on two groups is +/// two gaps, and serving one leaves the other. +#[test] +fn gaps_are_keyed_by_shard_and_frontier() { + let gaps = ReseedGapSet::new(); + assert!( + gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_HIGH) + .fresh + ); + let sibling = gaps.latch(S1, ReseedReason::SnapshotRequired, GAP_HIGH); + assert!( + sibling.fresh, + "the same frontier on another group is another gap" + ); + assert_eq!( + sibling.outstanding, 1, + "`outstanding` counts THIS group's gaps, since readiness is per group" + ); + assert_eq!(gaps.seqnos_for(S0), vec![GAP_HIGH]); + assert_eq!(gaps.seqnos_for(S1), vec![GAP_HIGH]); + + gaps.serve(S0, GAP_HIGH); + assert!(!gaps.outstanding_for(S0)); + assert!(gaps.outstanding_for(S1)); + assert_eq!(gaps.summary(), format!("s1@{GAP_HIGH}(snapshot_required)")); +} + +/// `seqnos_for` and `summary` are the operator-facing readings that were missing: +/// status could only say `reseed_required: true/false`, so `latched: 2, +/// discharged: 1` had to be reconstructed from log greps. +#[test] +fn the_surfaces_name_every_outstanding_gap() { + let gaps = ReseedGapSet::new(); + gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_LOW); + gaps.latch(S0, ReseedReason::SnapshotRequired, GAP_HIGH); + gaps.latch(S1, ReseedReason::Quarantine, 7); + assert_eq!(gaps.seqnos_for(S0), vec![GAP_LOW, GAP_HIGH], "lowest first"); + assert_eq!( + gaps.summary(), + format!( + "s0@{GAP_LOW}(snapshot_required), s0@{GAP_HIGH}(snapshot_required), s1@7(quarantine)" + ) + ); +} + +// ── /health cause ladder ──────────────────────────────────────────────────── + +fn ready_gates<'a>() -> UnreadyGates<'a> { + UnreadyGates { + shutting_down: false, + quarantined: false, + removed: false, + decommissioned: false, + reseed_gaps: None, + self_restart_refused: false, + converged: true, + } +} + +/// The incident's exact shape: every group converged, nothing quarantined, +/// nothing removed — and a latched reseed marker. The old ladder printed +/// "joiner boot not yet converged" here, which was false. +#[test] +fn health_cause_names_the_reseed_latch_not_convergence() { + let summary = format!("s1@{GAP_HIGH}(snapshot_required)"); + let cause = unready_cause( + S1, + &UnreadyGates { + reseed_gaps: Some(&summary), + ..ready_gates() + }, + ); + assert!( + cause.contains("reseed_marker_latched"), + "the 503 cause must name the gate that is holding: {cause}" + ); + assert!( + cause.contains(&GAP_HIGH.to_string()), + "and the outstanding gap the operator is waiting on: {cause}" + ); + assert!( + !cause.contains("not yet first-converged"), + "it must not claim a convergence state it did not test: {cause}" + ); +} + +/// A refused self-restart is why the reseed has not run yet, so the cause says +/// so instead of leaving the operator to guess. +#[test] +fn health_cause_reports_a_refused_self_restart() { + let summary = format!("s0@{GAP_LOW}(snapshot_required)"); + let cause = unready_cause( + S0, + &UnreadyGates { + reseed_gaps: Some(&summary), + self_restart_refused: true, + ..ready_gates() + }, + ); + assert!(cause.contains("reseed_marker_latched"), "{cause}"); + assert!(cause.contains("REFUSED"), "{cause}"); +} + +/// With no gate holding, the honest answer is that there is no reason to give — +/// not a state the ladder never checked. +#[test] +fn health_cause_admits_when_no_gate_is_holding() { + let cause = unready_cause(S0, &ready_gates()); + assert!(cause.contains("reason unavailable"), "{cause}"); + assert!(!cause.contains("converged"), "{cause}"); +} + +/// Every gate has its own arm, tested in `is_ready`'s order. +#[test] +fn health_cause_covers_every_readiness_gate() { + let gaps_summary = format!("s0@{GAP_LOW}(snapshot_required)"); + let cases: [(UnreadyGates<'_>, &str); 6] = [ + ( + UnreadyGates { + shutting_down: true, + ..ready_gates() + }, + "shutting down", + ), + ( + UnreadyGates { + quarantined: true, + ..ready_gates() + }, + "quarantined", + ), + ( + UnreadyGates { + removed: true, + ..ready_gates() + }, + "removed from the cluster", + ), + ( + UnreadyGates { + decommissioned: true, + ..ready_gates() + }, + "decommissioned by a peer's removed signal", + ), + ( + UnreadyGates { + reseed_gaps: Some(&gaps_summary), + ..ready_gates() + }, + "reseed_marker_latched", + ), + ( + UnreadyGates { + converged: false, + ..ready_gates() + }, + "not yet first-converged", + ), + ]; + for (gates, expected) in cases { + let cause = unready_cause(S0, &gates); + assert!( + cause.contains(expected), + "expected {expected:?} in the cause, got {cause:?}" + ); + } +} + +/// The gates are tested in priority order: a draining node says so even while it +/// also holds a gap (its readiness is already 503 for the drain, and the drain is +/// the actionable fact). +#[test] +fn health_cause_prefers_the_earlier_gate() { + let gaps_summary = format!("s0@{GAP_LOW}(snapshot_required)"); + let cause = unready_cause( + S0, + &UnreadyGates { + shutting_down: true, + quarantined: true, + reseed_gaps: Some(&gaps_summary), + converged: false, + ..ready_gates() + }, + ); + assert_eq!(cause, "shutting down"); +} + +// ── refused self-restart re-drive ─────────────────────────────────────────── + +/// The §2.4 quorum refusal is correct and stays; what was missing is the +/// re-drive. Its own ERROR promised "the reseed runs once quorum is safe (a +/// future latch re-evaluates)", and a node with a real gap gets no future latch +/// once the catch-up retry stops producing refusals — so it waited on a quorum +/// event nobody polled. A rolling deploy is exactly the window that produces the +/// refusal (`alive_voters_excluding_self=1 < majority=2` during the incident). +#[test] +fn a_held_refusal_keeps_being_re_evaluated() { + assert!( + refusal_retry_continues(false, true, true), + "a refusal that still holds over a still-outstanding gap must be re-polled \ + until quorum can spare this node" + ); +} + +/// And it stops for each of the three reasons there is nothing left to drive. +#[test] +fn the_refusal_re_drive_stops_when_it_should() { + assert!( + !refusal_retry_continues(true, true, true), + "draining: the exit it wanted is already under way" + ); + assert!( + !refusal_retry_continues(false, false, true), + "refusal cleared: the exit is owned by maybe_self_restart or the Fix 3 grace" + ); + assert!( + !refusal_retry_continues(false, true, false), + "gap healed via stream catch-up: a reboot would reseed nothing, and \ + restarting anyway is the reseed loop m12 already fixed" + ); +} diff --git a/tidal-server/tests/cluster_chaos.rs b/tidal-server/tests/cluster_chaos.rs index e8ded12..dda48be 100644 --- a/tidal-server/tests/cluster_chaos.rs +++ b/tidal-server/tests/cluster_chaos.rs @@ -305,8 +305,8 @@ fn mp_uat_step3_degraded_query_during_partition() { "[step3] lag is real: leader hwm {leader_hwm} > ap-south applied {applied_now} (stalled)" ); - // The leader's aggregated /cluster/status reports ap-south unreachable with - // worst-case lag (the HTTP status probe traverses the severed proxy). + // The leader's aggregated /cluster/status reports ap-south unreachable with an + // UNKNOWN frontier (the HTTP status probe traverses the severed proxy). poll_until( Duration::from_secs(10), "leader status must show ap-south reachable:false", @@ -326,13 +326,30 @@ fn mp_uat_step3_degraded_query_during_partition() { .find(|r| r["name"].as_str() == Some("ap-south")) .unwrap(); assert_eq!(ap_row["reachable"].as_bool(), Some(false)); + // CHANGED 2026-08-31. This asserted `lag_events >= 1` — "an unreachable peer + // must report worst-case lag". That value was a FABRICATION: `aggregate_region_row` + // could not reach the peer, so it invented `applied_events: 0` and derived the + // deficit `leader_last_seq - 0` from it. Both fields are now `null`, which is the + // honest answer to "how far has a peer I cannot reach applied?". + // + // Asserting a number here would re-require the fabrication. The real invariant is + // that the gap is REPRESENTED rather than guessed, so that is what is pinned: an + // unreachable peer reports an explicitly unknown frontier, never a made-up one. + // + // This test only runs in the nightly cron, which had never executed in 216 days — + // so this caller survived the task-04a migration that updated `cluster_runbook.rs`. + // The nightly's first-ever run caught it, which is precisely why it was configured. assert!( - ap_row["lag_events"].as_u64().unwrap_or(0) >= 1, - "unreachable ap-south must report worst-case lag: {ap_row}" + ap_row["lag_events"].is_null(), + "an unreachable peer's lag must be null (unknown), not a fabricated worst case: {ap_row}" + ); + assert!( + ap_row["applied_events"].is_null(), + "an unreachable peer's applied frontier must be null (unknown), not 0: {ap_row}" ); println!( - "[step3] aggregated status: ap-south reachable:false lag_events={}", - ap_row["lag_events"] + "[step3] aggregated status: ap-south reachable:false applied={} lag={} (both unknown, not fabricated)", + ap_row["applied_events"], ap_row["lag_events"] ); // The degraded scatter-gather: GET /sharded/feed on the leader → 200, degraded, diff --git a/tidal-server/tests/cluster_poison_embedding.rs b/tidal-server/tests/cluster_poison_embedding.rs new file mode 100644 index 0000000..ec5027e --- /dev/null +++ b/tidal-server/tests/cluster_poison_embedding.rs @@ -0,0 +1,223 @@ +//! Tier-3 regression gate for the 2026-08-31 shard-1 quorum-write outage +//! (REAL 3-process cluster, RF=3). +//! +//! # The incident this file exists to prevent +//! +//! A probe posted a **128-dimension** vector to `/embeddings` for slot +//! `content_vector`; the live schema declares **1536**. The origin appended the +//! blob to the WAL FIRST (`wal_blob_first` returned `Ok(Some(seq))`), validated +//! SECOND, failed, and answered the client **500**. The record was already +//! durable, so it shipped to both followers, neither could apply it, and both +//! **halted their receivers**: +//! +//! ```text +//! replicated blob batch apply failed (1 records): +//! [op=write_item_embedding] dimension mismatch: expected 1536, got 128 +//! ``` +//! +//! Shard 1 froze — leader at `13540698`, followers pinned at +//! `13540694`/`13540693`, `lag` growing — and writes to the group returned +//! **503**, because the followers could not ack. One malformed HTTP request took +//! out replication for a whole shard group. Neither a restart (boot self-heal +//! re-pulls the same record) nor `POST /cluster/reseed` escaped it: the snapshot +//! is captured at the leader's applied frontier, which is itself BEHIND the +//! poison. Only forcing a leader election recovered the group. +//! +//! # What this test asserts +//! +//! 1. A dimension-mismatched `/embeddings` write is rejected **400** — a +//! malformed vector is a caller error; the old 500 misattributed it to the +//! server while the record was, in fact, more than failed: durable, +//! unapplicable, and blocking. +//! 2. It appends **NOTHING**: the leader's WAL frontier (`last_seq`) does not +//! move across the rejected writes. This is the load-bearing assertion — a +//! fix that only corrected the status code would still poison the stream. +//! 3. Every receiver stays healthy: all three nodes converge to `lag = 0`, and a +//! VALID write issued afterwards still replicates to both followers. A halted +//! receiver freezes flat, so the second half is what distinguishes "alive" +//! from "merely quiet". +//! 4. `/health` is 200 on every node. +//! +//! Node logs are discarded by the harness (`Stdio::null()` — a piped-but-undrained +//! pipe deadlocks a chatty node), so the `grep -c 'receiver halting'` check from +//! the runbook is expressed here as its observable consequence: lag returns to +//! zero and continues to track new writes. +//! +//! Run: +//! ```bash +//! cargo test -p tidal-server --features cluster-e2e --test cluster_poison_embedding -- --nocapture +//! ``` + +#![cfg(feature = "cluster-e2e")] +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +mod support; + +use std::time::Duration; + +use support::multiproc::{MultiProcCluster, convergence_budget, seed_items_and_embeddings}; + +/// Region 0 = `us-east` = the initial leader in every harness topology. +const LEADER: usize = 0; +const NODES: usize = 3; + +/// The harness schema declares `content_vector` at **4** dimensions +/// (`support::multiproc::write_schema`). Two floats is therefore the local +/// analogue of the live cluster's 128-into-1536 probe. +const DECLARED_DIMENSIONS: usize = 4; + +/// How many times the malformed write is replayed. The incident was ONE request; +/// hammering proves the rejection cannot accumulate durable state either. +const POISON_ATTEMPTS: u64 = 5; + +/// Serializes the heavy multi-process tests in THIS target. +/// +/// Each test here spawns 3 OS processes. The harness's `spawn_lock` only +/// serializes the spawn itself and is released as soon as `start` returns, so +/// without this every test in the file can hold a live cluster simultaneously and +/// the resulting contention starves each other's election/convergence budgets +/// (see the identical guard in `cluster_sharding.rs`). Poison is recovered rather +/// than propagated: one failing test must not cascade into "the rest panicked on a +/// poisoned lock", which hides the original failure. +fn heavy_test_guard() -> std::sync::MutexGuard<'static, ()> { + static LOCK: std::sync::LazyLock> = + std::sync::LazyLock::new(|| std::sync::Mutex::new(())); + LOCK.lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +/// The leader's WAL high-water-mark, once it has stopped moving. +/// +/// Returns a frontier observed IDENTICAL twice across a quiet window, so the +/// "did the rejected write append anything?" comparison is against a settled +/// value rather than a racing one. Panics if the frontier never settles, which +/// would mean the cluster is not idle and the test's premise is broken. +fn settled_leader_frontier(cluster: &MultiProcCluster) -> u64 { + let deadline = std::time::Instant::now() + convergence_budget(); + let mut last = None; + loop { + let seq = cluster + .leader_last_seq() + .expect("the leader must report its own frontier"); + if last == Some(seq) { + return seq; + } + assert!( + std::time::Instant::now() <= deadline, + "leader frontier never settled (last two samples {last:?} then {seq}); \ + the cluster is not idle, so an append-nothing assertion would be meaningless" + ); + last = Some(seq); + std::thread::sleep(Duration::from_millis(300)); + } +} + +/// Post a malformed embedding to the leader and return `(status, body)`. +fn post_malformed(cluster: &MultiProcCluster, entity_id: u64) -> (u16, String) { + let resp = cluster.post( + LEADER, + "/embeddings", + &serde_json::json!({ "entity_id": entity_id, "values": [0.1, 0.2] }), + ); + let status = resp.status().as_u16(); + let body = resp.text().unwrap_or_default(); + (status, body) +} + +#[test] +fn mp_malformed_embedding_is_rejected_400_and_never_enters_the_replication_stream() { + let _heavy = heavy_test_guard(); + let cluster = MultiProcCluster::start(NODES); + let followers: Vec = (0..NODES).filter(|&i| i != LEADER).collect(); + + // Steady state: valid 4-dim embeddings replicated to every node. + seed_items_and_embeddings(&cluster, LEADER, 4); + cluster.wait_converged_all(convergence_budget()); + let frontier_before = settled_leader_frontier(&cluster); + println!("[poison] converged at leader frontier {frontier_before}"); + + // ── 1. The malformed write is a CALLER error: 400, not 500 ── + for attempt in 0..POISON_ATTEMPTS { + let entity_id = 1000 + attempt; + let (status, body) = post_malformed(&cluster, entity_id); + assert_eq!( + status, 400, + "a 2-float vector against a {DECLARED_DIMENSIONS}-dim slot is a caller \ + error and must be 400 (the incident returned 500, misattributing it to \ + the server); body: {body}" + ); + assert!( + body.contains("dimension mismatch"), + "the rejection must name the real cause so a caller can fix its model: {body}" + ); + } + + // ── 2. It appended NOTHING ── + // + // The whole defect was ordering: append, then validate. If the record still + // enters the log, it still ships, and the followers still halt — the status + // code is cosmetic next to this. + let frontier_after = settled_leader_frontier(&cluster); + assert_eq!( + frontier_after, frontier_before, + "{POISON_ATTEMPTS} rejected embeddings must append NOTHING to the WAL; the \ + leader frontier moved {frontier_before} -> {frontier_after}, so the poison \ + is durable and will ship to every follower" + ); + + // ── 3. Every receiver is still healthy ── + // + // First: nothing halted, so the group is still converged at lag 0. + cluster.wait_converged_all(convergence_budget()); + + // Then the half that a halted receiver cannot fake — a follower whose + // receiver died reports lag 0 forever while applying nothing, so prove the + // streams still MOVE by shipping a valid write through them. + let resp = cluster.post( + LEADER, + "/embeddings", + &serde_json::json!({ "entity_id": 2000, "values": [1.0, 2.0, 3.0, 4.0] }), + ); + assert_eq!( + resp.status().as_u16(), + 204, + "a correctly-sized embedding must still be accepted after the rejections" + ); + let frontier_live = settled_leader_frontier(&cluster); + assert!( + frontier_live > frontier_after, + "the valid write must advance the frontier the rejected ones left alone \ + ({frontier_after} -> {frontier_live})" + ); + cluster.wait_converged_all(convergence_budget()); + for &idx in &followers { + let status = cluster + .local_status(idx) + .expect("a live follower must serve /cluster/status/local"); + assert_eq!( + status["lag_events"].as_u64(), + Some(0), + "follower {} must be at lag 0 after the post-rejection write; a halted \ + receiver freezes instead of tracking: {status}", + cluster.region_name(idx) + ); + assert!( + status["applied_events"].as_u64().unwrap_or(0) >= frontier_live, + "follower {} must have APPLIED up to the new frontier {frontier_live}, not \ + merely report zero lag: {status}", + cluster.region_name(idx) + ); + } + + // ── 4. And every node still reports healthy ── + for idx in 0..NODES { + let resp = cluster.get(idx, "/health"); + assert_eq!( + resp.status().as_u16(), + 200, + "node {} must be healthy after the rejected writes", + cluster.region_name(idx) + ); + } + println!("[poison] all {NODES} nodes healthy, lag 0, frontier {frontier_live}"); +} diff --git a/tidal-server/tests/cluster_routes.rs b/tidal-server/tests/cluster_routes.rs index c4b50e2..d8290a4 100644 --- a/tidal-server/tests/cluster_routes.rs +++ b/tidal-server/tests/cluster_routes.rs @@ -404,11 +404,24 @@ fn status_aggregates_all_regions() { "down region unreachable: {down_row}" ); assert_eq!(down_row["partitioned"].as_bool(), Some(true)); - assert_eq!(down_row["applied_events"].as_u64(), Some(0)); - assert_eq!( - down_row["lag_events"].as_u64(), - Some(relay_len), - "worst-case lag = leader_last_seq" + // CHANGED 2026-08-31, same reason as cluster_chaos.rs:329. This asserted + // `applied_events == 0` and `lag_events == relay_len` ("worst-case lag = + // leader_last_seq"). Both were FABRICATED by `aggregate_region_row`'s + // unreachable arm: it could not reach the peer, so it invented an applied + // frontier of 0 and derived the deficit from it. A down peer's frontier is + // genuinely unknown and now says so. + // + // Note `relay_len` is no longer compared against: the leader's own frontier tells + // you nothing about how far a peer you cannot reach has applied, which was the + // whole defect. + assert!( + down_row["applied_events"].is_null(), + "a down region's applied frontier must be null (unknown), not 0: {down_row}" + ); + assert!( + down_row["lag_events"].is_null(), + "a down region's lag must be null (unknown), not a worst case derived from a \ + fabricated applied=0: {down_row}" ); } diff --git a/tidal-server/tests/cluster_runbook.rs b/tidal-server/tests/cluster_runbook.rs index b9e0822..d33467b 100644 --- a/tidal-server/tests/cluster_runbook.rs +++ b/tidal-server/tests/cluster_runbook.rs @@ -303,11 +303,17 @@ fn runbook_s5_health_and_openapi() { /// §5 data writes: the documented success contracts (items 201 + report; /// embeddings 200 + report on the leader; signals 204) and the validation -/// rejections (strict-dimension, zero-norm, undeclared-signal). The validation -/// status codes are asserted against the REAL engine→HTTP mapping, not assumed: -/// the engine wraps a vector dimension/zero-norm violation as an internal error -/// (HTTP 500) and an undeclared signal name as a `BadRequest` (HTTP 400); the -/// runbook documents exactly these. +/// rejections (strict-dimension, zero-norm, undeclared-signal), each asserted +/// as the EXACT status the runbook documents. +/// +/// All three are **400**. The dimension and zero-norm cases used to surface as +/// `500` because the engine validated the vector AFTER appending it to the WAL +/// and wrapped the resulting `VectorError` as an internal error — which was not +/// merely an imprecise status code: the record was durable by then and halted +/// every follower that received it (2026-08-31 shard-1 quorum-write outage, +/// runbook §16.6). Both are now rejected before the append, as caller errors. +/// The range assertion this replaced accepted the 500 and so could never have +/// caught the defect. #[test] fn runbook_s5_data_writes_and_validation() { let cluster = MultiProcCluster::start(3); @@ -316,30 +322,29 @@ fn runbook_s5_data_writes_and_validation() { seed_items_and_embeddings(&cluster, LEADER, 4); write_view(&cluster, LEADER, 1, 1.0); - // Strict dimensions: the slot declares 4 dims; a 3-vector is rejected. The - // engine validation surfaces as a server-side error (>= 400), NEVER a silent - // 2xx accept. Capture the exact status so the runbook documents it truthfully. + // Strict dimensions: the slot declares 4 dims; a 3-vector is a CALLER error. let resp = cluster.post( LEADER, "/embeddings", &serde_json::json!({ "entity_id": 1, "values": [0.1, 0.2, 0.3] }), ); let dim_status = resp.status().as_u16(); - assert!( - (400..600).contains(&dim_status), - "strict-dimension embedding must be rejected (>= 400), got {dim_status}" + assert_eq!( + dim_status, 400, + "strict-dimension embedding must be rejected 400 (a 500 here means the \ + record was journaled before it was validated), got {dim_status}" ); - // Zero-norm vector (all zeros) is rejected — same error class as the dim case. + // Zero-norm vector (all zeros) is rejected — same class, same status. let resp = cluster.post( LEADER, "/embeddings", &serde_json::json!({ "entity_id": 1, "values": [0.0, 0.0, 0.0, 0.0] }), ); let zero_status = resp.status().as_u16(); - assert!( - (400..600).contains(&zero_status), - "zero-norm embedding must be rejected (>= 400), got {zero_status}" + assert_eq!( + zero_status, 400, + "zero-norm embedding must be rejected 400, got {zero_status}" ); // Undeclared signal name → 400 (the schema-resolution `BadRequest`). diff --git a/tidal/src/db/items.rs b/tidal/src/db/items.rs index 0936ccc..efb497d 100644 --- a/tidal/src/db/items.rs +++ b/tidal/src/db/items.rs @@ -185,13 +185,18 @@ impl TidalDb { /// /// # Metrics /// - /// Increments `tidaldb_cluster_blobs_applied_total` per kind on success and - /// `..._apply_failed_total` on failure. This is the **live** apply path only; - /// boot-time replay of already-counted records goes through + /// Increments `tidaldb_cluster_blobs_applied_total` per kind for every + /// record that applied, and `..._apply_failed_total` once per record + /// rejected as permanently unapplicable plus once per kind present in a + /// halted round. This is the **live** apply path only; boot-time replay of + /// already-counted records goes through /// [`replay_recovered_blobs`](Self::replay_recovered_blobs) and is deliberately /// NOT counted here — counting it would inflate `applied` past the writer's /// `originated` on every restart and make the comparison worthless. - pub(crate) fn apply_replicated_blobs(&self, records: Vec) -> crate::Result<()> { + pub(crate) fn apply_replicated_blobs( + &self, + records: Vec, + ) -> crate::Result { // Tally before the records move into the apply; kinds are needed on both // the success and the failure edge. #[cfg(feature = "metrics")] @@ -202,7 +207,11 @@ impl TidalDb { } t }; - let outcome = self.apply_replicated_blobs_inner(records); + // Per-kind count of records Phase 1 DROPPED as permanently unapplicable + // (see `apply_replicated_blobs_inner`). Filled even without the metrics + // feature: the receiver's skip-vs-halt log reads the total. + let mut rejected = [0u64; crate::wal::format::batch::BlobKind::COUNT]; + let outcome = self.apply_replicated_blobs_inner(records, &mut rejected); #[cfg(feature = "metrics")] { let cluster = &self.metrics.cluster; @@ -212,7 +221,19 @@ impl TidalDb { continue; } if outcome.is_ok() { - cluster.observe_blobs_applied(kind, n); + // A dropped record was never applied, so it counts on the + // failure series only — once per record, because each is an + // independently rejected write an operator must be able to + // total. + let dropped = rejected[kind.index()]; + let applied = n - dropped; + for _ in 0..dropped { + cluster.observe_blob_apply_failed(kind); + } + if applied == 0 { + continue; + } + cluster.observe_blobs_applied(kind, applied); // The follower side of `wal_blob_first`'s vector count: this // node now holds vectors the originating node counted when // its append went durable, so both ends of one replicated @@ -221,7 +242,7 @@ impl TidalDb { // path — boot replay (`replay_recovered_blobs`) is excluded // here for the same reason `applied` excludes it. if kind == crate::wal::format::batch::BlobKind::Embedding { - self.metrics.observe_replicated_vectors(n); + self.metrics.observe_replicated_vectors(applied); } } else { // One failure per kind present in the halted round. The round is @@ -231,13 +252,19 @@ impl TidalDb { } } } - outcome + outcome.map(|()| crate::replication::receiver::BlobApplyOutcome { + rejected: rejected.iter().sum(), + }) } // One linear three-phase pass (validate -> journal -> apply); splitting it // would scatter the WAL-first ordering invariants across helpers. #[allow(clippy::too_many_lines)] - fn apply_replicated_blobs_inner(&self, records: Vec) -> crate::Result<()> { + fn apply_replicated_blobs_inner( + &self, + records: Vec, + rejected: &mut [u64; crate::wal::format::batch::BlobKind::COUNT], + ) -> crate::Result<()> { /// One validated record, parsed exactly once in Phase 1 and applied /// in Phase 3 (item metadata is deserialized here, never re-parsed). enum BlobApply<'a> { @@ -272,19 +299,101 @@ impl TidalDb { let records: Vec> = records.into_iter().map(std::sync::Arc::new).collect(); - // Phase 1 — validate ALL records before journaling ANY, capturing - // each record's parsed form for the Phase 3 upserts. + // ── Phase 1: the halt-vs-skip decision ──────────────────────────── + // + // Case 1 — SKIP. A DATA record (item metadata, embedding) that fails + // NODE-INDEPENDENT validation is structurally unapplicable on every + // replica of this group, forever: the verdict reads only the record's + // own bytes and the shared schema, so no binary upgrade, no catch-up + // and no restart can make it apply. Halting on it converts one + // malformed client request into a shard-wide write outage — on + // 2026-08-31 a 128-dim vector posted to a 1536-dim slot pinned both + // shard-1 followers and writes to the group returned 503 until a leader + // election was forced. So it is DROPPED here: never journaled into this + // node's own log (a restart must not re-poison the receiver), never + // applied, counted on `tidaldb_cluster_blobs_apply_failed_total`, and + // logged at ERROR. The rest of the round applies and the frontier + // advances past the poison. + // + // Case 2 — HALT. A CONTROL record (term marker, membership) that fails + // validation returns an error and halts the receiver: dropping it would + // silently lose term/roster state the surviving replicas still hold, + // trading an availability bug for a divergence bug. Every other failure + // (durability, storage, poisoned lock, read-only, an unknown batch kind + // rejected before this point) is node-LOCAL or version-dependent — it + // may well apply on the next attempt or after an upgrade, and skipping + // it would silently drop replicated data that is perfectly valid. + // + // Deliberately NOT skippable: this node's REGISTERED slot width. It is + // derived from local storage, so two replicas can legitimately disagree + // about it, and acting on it here would let one node apply a record + // another dropped. Only `validate_embedding_replicable` runs here. + let mut accepted: Vec> = Vec::with_capacity(records.len()); + // Item metadata deserialized during the partition, carried so Phase 1b + // never re-parses it. Index-aligned with `accepted`. + let mut parsed_metadata: Vec>> = + Vec::with_capacity(records.len()); + for record in records { + let verdict = match &*record { + BlobRecord::ItemMetadata(r) => { + let metadata = deserialize_metadata(&r.metadata_bytes); + match Self::validate_item_write(EntityId::new(r.entity_id), &metadata) { + Ok(()) => Ok(Some(metadata)), + Err(e) => Err(e), + } + } + BlobRecord::Embedding(r) => self + .validate_embedding_replicable(EntityKind::Item, &r.values) + .map(|()| None), + // Control records: validated in Phase 1b, where a failure halts. + BlobRecord::TermMarker(_) | BlobRecord::Membership(_) => Ok(None), + }; + match verdict { + Ok(metadata) => { + accepted.push(record); + parsed_metadata.push(metadata); + } + Err(e) => { + let kind = record.blob_kind(); + rejected[kind.index()] += 1; + tracing::error!( + blob_kind = kind.label(), + entity_id = record.entity_id(), + error = %e, + "replicated blob record is unapplicable on every replica; \ + SKIPPING it and continuing the stream (counted on \ + tidaldb_cluster_blobs_apply_failed_total) — halting here would \ + cost the shard group its write quorum" + ); + } + } + } + let records = accepted; + + // Phase 1b — validate the control records that survived the partition + // (a failure HALTS, case 2 above) and capture every record's parsed + // form for the Phase 3 upserts. let mut applies: Vec> = Vec::with_capacity(records.len()); for (slot, record) in records.iter().enumerate() { match &**record { BlobRecord::ItemMetadata(r) => { - let id = EntityId::new(r.entity_id); - let metadata = deserialize_metadata(&r.metadata_bytes); - Self::validate_item_write(id, &metadata)?; - applies.push(BlobApply::Item { id, metadata }); + // Moved out of the partition's parse, never re-deserialized. + // `None` is unreachable (the partition fills every accepted + // item record's slot) but is surfaced as an error rather + // than defaulted to an empty map, which would silently + // erase the item's metadata on this replica only. + let metadata = parsed_metadata[slot].take().ok_or_else(|| { + TidalError::internal( + "apply_replicated_blobs", + "accepted item-metadata record lost its parsed form", + ) + })?; + applies.push(BlobApply::Item { + id: EntityId::new(r.entity_id), + metadata, + }); } BlobRecord::Embedding(r) => { - Self::validate_embedding_for_log(&r.values)?; applies.push(BlobApply::Embedding { id: EntityId::new(r.entity_id), values: &r.values, @@ -307,8 +416,10 @@ impl TidalDb { // Re-validate the roster before it enters THIS node's log: // a structurally-invalid record decoded clean would fail // identically on every follower, so refuse it here rather - // than journal an unappliable record (the segment receiver - // halts on apply errors, never silently skips). + // than journal an unappliable record. This one HALTS rather + // than skipping (case 2): a dropped roster record diverges + // the cluster's membership view, which is worse than a + // stalled receiver an operator can see. r.validate() .map_err(|e| TidalError::invalid_input(format!("{e}")))?; applies.push(BlobApply::Membership { @@ -701,16 +812,26 @@ impl TidalDb { /// /// # Errors /// + /// - `TidalError::InvalidInput` if the embedding is empty, non-finite, + /// zero-norm, or its length does not equal the slot's declared / + /// registered dimensions. A malformed vector is a CALLER error and is + /// rejected before the WAL append, so it never becomes durable and never + /// ships (see [`Self::validate_embedding_for_log`]). /// - `TidalError::Internal` if no storage backend is wired or lock is poisoned. /// - `TidalError::Storage` on storage engine failure. - /// - `TidalError::Internal` if the embedding has zero norm. pub fn write_item_embedding( &self, id: EntityId, embedding: &[f32], ) -> crate::Result> { self.require_writeable("write_item_embedding")?; - Self::validate_embedding_for_log(embedding)?; + // BEFORE `wal_blob_first`, never after. The WAL is the durability + // boundary: once a record crosses it, rejecting it is no longer + // possible — only halting is. On 2026-08-31 a 128-dim vector posted to + // a 1536-dim slot was appended first and validated second; the client + // got a 500 while the already-durable record shipped to both followers, + // halted both receivers, and cost shard 1 its write quorum (503). + self.validate_embedding_for_log(EntityKind::Item, embedding)?; // WAL-first in cluster mode (m11p2; see `write_item_with_metadata`). // The record carries the CALLER's raw values: every replica runs the @@ -753,11 +874,58 @@ impl TidalDb { } /// Reject embeddings that could never apply BEFORE they enter the - /// replicated log: an empty / non-finite / zero-norm vector fails - /// `insert_embedding` identically on every replica, and a record that is - /// guaranteed to fail apply must not ship (a follower halts its receiver - /// on apply errors rather than silently skipping). - fn validate_embedding_for_log(embedding: &[f32]) -> crate::Result<()> { + /// replicated log: an empty / non-finite / zero-norm vector, or one whose + /// width does not match the slot, fails `insert_embedding` identically on + /// every replica, and a record that is guaranteed to fail apply must not + /// ship (a follower halts its receiver on apply errors rather than + /// silently skipping). + /// + /// # Two tiers, and why they are separate + /// + /// - **Node-independent** (shape + the schema-DECLARED width): every + /// replica reaches the identical verdict from the identical bytes, + /// because the schema is the cluster-wide contract. This tier is what the + /// replicated apply path (`apply_replicated_blobs_inner` Phase 1) may act + /// on, since a verdict every node shares can be skipped without + /// diverging anyone. + /// - **Node-local** (this node's REGISTERED slot width, which is derived + /// from its own stored vectors): checked only on the ORIGIN write path, + /// where its only effect is to refuse a write this node's own apply would + /// reject. Applying it on the replicated path would let two replicas + /// disagree about the same record and silently diverge, so it is + /// deliberately not reachable from there — see `replicable` below. + /// + /// The width comparison itself is `storage::vector::validate_dimensions` — + /// the same function the apply path's `normalize_and_store` calls. One rule, + /// one definition: a second copy would drift, and it is the apply-path copy + /// that halts replication when the two disagree. + fn validate_embedding_for_log(&self, kind: EntityKind, embedding: &[f32]) -> crate::Result<()> { + self.validate_embedding_replicable(kind, embedding)?; + + // Node-local tier: origin-only (see above). + let slot_name = self.entity_embedding_slot(kind); + if let Some(registered) = self.registered_embedding_dimensions(kind, slot_name) { + crate::storage::vector::validate_dimensions(registered, embedding.len()).map_err( + |e| TidalError::invalid_input(format!("embedding slot \"{slot_name}\": {e}")), + )?; + } + Ok(()) + } + + /// The node-INDEPENDENT half of [`Self::validate_embedding_for_log`]: the + /// checks whose verdict is identical on every replica of the group, so a + /// failure here means the record is structurally unapplicable everywhere, + /// forever, and the replication stream may drop it instead of halting. + /// + /// Every check below reads only the record's own bytes and the shared + /// schema. Nothing here may consult this node's storage, registry, or + /// binary version — that is the invariant that makes the receiver's + /// skip-vs-halt decision sound. + fn validate_embedding_replicable( + &self, + kind: EntityKind, + embedding: &[f32], + ) -> crate::Result<()> { if embedding.is_empty() { return Err(TidalError::invalid_input("embedding must not be empty")); } @@ -772,9 +940,51 @@ impl TidalDb { "embedding must have a positive finite L2 norm", )); } + if let Some(declared) = self.declared_embedding_dimensions(kind) { + let slot_name = self.entity_embedding_slot(kind); + crate::storage::vector::validate_dimensions(declared, embedding.len()).map_err( + |e| TidalError::invalid_input(format!("embedding slot \"{slot_name}\": {e}")), + )?; + } Ok(()) } + /// The width the SCHEMA declares for `kind`'s first embedding slot, or + /// `None` when the schema declares none. + /// + /// Node-independent by construction: every replica of a group is started + /// from the same schema, so this is the one width they all agree on. Paired + /// with [`Self::entity_embedding_slot`], which picks the same slot. + fn declared_embedding_dimensions(&self, kind: EntityKind) -> Option { + self.schema_def.as_ref().and_then(|schema| { + schema + .embedding_slots() + .iter() + .find(|slot| slot.entity_kind == kind) + .map(|slot| slot.dimensions) + }) + } + + /// The width THIS node's registry currently holds for `(kind, slot_name)`, + /// or `None` when the slot is not yet registered — the first write + /// auto-registers it at the vector's own length, so an unregistered slot + /// constrains nothing. + /// + /// Node-LOCAL: the registered width is derived from this node's stored + /// vectors (`rebuild_or_load_from_store` treats the stored header as + /// authoritative), so two replicas can legitimately hold different values + /// while one is still catching up. A poisoned lock yields `None` rather + /// than failing the write: the apply path re-takes the same lock and + /// surfaces the poison there, and a pre-check must never invent a rejection + /// out of a lock fault. + fn registered_embedding_dimensions(&self, kind: EntityKind, slot_name: &str) -> Option { + let registry = self + .embedding_registry + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + registry.get(kind, slot_name).map(|slot| slot.dimensions) + } + /// Read the stored content embedding for an item. /// /// Returns `None` if no embedding has been written for this item. The slot diff --git a/tidal/src/db/items_tests.rs b/tidal/src/db/items_tests.rs index 94c22e2..817587b 100644 --- a/tidal/src/db/items_tests.rs +++ b/tidal/src/db/items_tests.rs @@ -473,3 +473,263 @@ fn cluster_node_stamps_configured_partition_id() { ); db.close().unwrap(); } + +/// A schema declaring a fixed-width item embedding slot — the node-independent +/// dimension contract every replica of a group is started from. `minimal_schema` +/// deliberately declares none, so the two are not interchangeable. +fn schema_with_embedding_slot(dimensions: usize) -> crate::schema::Schema { + use crate::schema::{DecaySpec, EntityKind, SchemaBuilder, Window}; + let mut b = SchemaBuilder::new(); + let _ = b + .signal( + "view", + EntityKind::Item, + DecaySpec::Exponential { + half_life: std::time::Duration::from_secs(3600), + }, + ) + .windows(&[Window::AllTime]) + .velocity(false) + .add(); + b.embedding_slot("content_vector", EntityKind::Item, dimensions); + b.build().expect("schema must be valid") +} + +/// The core 2026-08-31 regression: a dimension-mismatched embedding is rejected +/// as a CALLER error (400, not 500) and appends NOTHING to the WAL. +/// +/// The WAL is the durability boundary — once a record crosses it, rejecting it +/// is no longer possible, only halting is. The live incident appended first and +/// validated second: the client got a 500 while the already-durable record +/// shipped to both followers, halted both receivers, and cost shard 1 its write +/// quorum. The frontier assertion is the load-bearing half of this test; a fix +/// that only changed the status code would still poison the stream. +#[test] +fn a_dimension_mismatched_embedding_is_rejected_before_the_wal_append() { + use crate::db::config::NodeConfig; + use crate::replication::ShardId; + + let dir = tempfile::tempdir().unwrap(); + let db = TidalDb::builder() + .with_data_dir(dir.path()) + .with_schema(schema_with_embedding_slot(1536)) + .with_cluster(NodeConfig { + peer_shards: vec![ShardId(1)], + ..NodeConfig::default() + }) + .open() + .unwrap(); + + let frontier_before = db.last_wal_seq(); + let err = db + .write_item_embedding(EntityId::new(1), &[0.1, 0.2]) + .expect_err("a 2-dim vector must not be accepted for a 1536-dim slot"); + + assert!( + matches!(err, crate::TidalError::InvalidInput(_)), + "a malformed vector is a caller error and must map to 400, not 500: {err:?}" + ); + let msg = err.to_string(); + assert!( + msg.contains("expected 1536") && msg.contains("got 2"), + "the rejection must name both widths so the caller can fix its model: {msg}" + ); + assert_eq!( + db.last_wal_seq(), + frontier_before, + "a rejected embedding must append NOTHING: the WAL frontier may not advance, \ + or the record ships and halts every follower that receives it" + ); + + // The same slot still accepts a correctly-sized vector, and THAT one journals. + let seq = db + .write_item_embedding(EntityId::new(2), &vec![0.5; 1536]) + .expect("a 1536-dim vector matches the declared slot"); + assert!( + seq.is_some(), + "a valid cluster-mode embedding write must still be journaled" + ); + assert!( + db.last_wal_seq() > frontier_before, + "the accepted write must advance the frontier the rejected one left alone" + ); + db.close().unwrap(); +} + +/// The pre-append guard and the apply path share ONE dimension rule +/// (`storage::vector::validate_dimensions`), so a vector the write path accepts +/// can never fail the apply that follows it. +/// +/// Proven by the observable consequence: the accepted write reaches storage. If +/// the two rules ever drift, the write journals and then the local apply fails, +/// which is precisely the durable-but-unappliable record this task removes. +#[test] +fn the_pre_append_guard_and_the_apply_path_agree_on_the_slot_width() { + let db = TidalDb::builder() + .ephemeral() + .with_schema(schema_with_embedding_slot(4)) + .open() + .unwrap(); + + db.write_item_embedding(EntityId::new(1), &[1.0, 0.0, 0.0, 0.0]) + .expect("a 4-dim vector must pass the guard AND the apply"); + assert!( + db.read_item_embedding(EntityId::new(1)) + .unwrap() + .is_some_and(|v| v.len() == 4), + "the write that passed the guard must be durable in the slot" + ); + + // A width the declared slot forbids is refused by the guard, not by a + // post-append apply failure. + let err = db + .write_item_embedding(EntityId::new(2), &[1.0, 0.0]) + .expect_err("2 dims must be refused against a 4-dim slot"); + assert!( + matches!(err, crate::TidalError::InvalidInput(_)), + "the guard rejects; the apply path never sees it: {err:?}" + ); + db.close().unwrap(); +} + +/// Replication SKIPS a record that no replica could ever apply, and keeps +/// applying the rest of the round. +/// +/// This is the follower half of the incident. Halting on a schema-mismatched +/// embedding pinned both shard-1 followers at `13540694`/`13540693` while the +/// leader ran to `13540698`, and writes to the group returned 503 until a leader +/// election was forced. The round must therefore: drop the poison, apply the +/// good records beside it, count the drop on +/// `tidaldb_cluster_blobs_apply_failed_total{kind="embedding"}`, and return Ok +/// so the receiver advances its frontier past it. +#[test] +fn a_schema_unapplicable_replicated_embedding_is_skipped_not_halted() { + use crate::wal::format::batch::{BlobRecord, EmbeddingRecord}; + + let db = TidalDb::builder() + .ephemeral() + .with_schema(schema_with_embedding_slot(4)) + .open() + .unwrap(); + + let outcome = db + .apply_replicated_blobs(vec![ + BlobRecord::Embedding(EmbeddingRecord { + entity_id: 1, + values: vec![1.0, 0.0, 0.0, 0.0], + }), + // The poison: 2 dims against a declared 4-dim slot. Node-independent + // — every replica reaches this verdict from these bytes. + BlobRecord::Embedding(EmbeddingRecord { + entity_id: 2, + values: vec![0.1, 0.2], + }), + BlobRecord::Embedding(EmbeddingRecord { + entity_id: 3, + values: vec![0.0, 1.0, 0.0, 0.0], + }), + ]) + .expect("the round must succeed: halting here is the shard-wide outage"); + assert_eq!( + outcome.rejected, 1, + "exactly the poison record is dropped, and it is reported — a silent skip \ + would be as bad as the halt" + ); + + assert!( + db.read_item_embedding(EntityId::new(1)).unwrap().is_some(), + "a good record BEFORE the poison must still apply" + ); + assert!( + db.read_item_embedding(EntityId::new(3)).unwrap().is_some(), + "a good record AFTER the poison must still apply — dropping the whole \ + round would silently lose valid replicated data" + ); + assert!( + db.read_item_embedding(EntityId::new(2)).unwrap().is_none(), + "the poison record must not be applied" + ); + + let mut out = String::new(); + db.metrics.cluster.render_into(&mut out, 0); + assert!( + out.contains( + r#"tidaldb_cluster_blobs_apply_failed_total{kind="embedding",partition_id="0"} 1"# + ), + "the skip must be COUNTED, once per dropped record:\n{out}" + ); + assert!( + out.contains(r#"tidaldb_cluster_blobs_applied_total{kind="embedding",partition_id="0"} 2"#), + "the two records that applied must be counted as applied, not the three received:\n{out}" + ); + db.close().unwrap(); +} + +/// Without a schema-declared width there is no node-independent verdict, so the +/// round must NOT skip. +/// +/// The registered slot width is derived from this node's own stored vectors, so +/// two replicas can legitimately disagree about it. Acting on it here would let +/// one node apply a record another dropped — silent divergence. The failure must +/// stay an error and halt the receiver, which an operator can see. +#[test] +fn a_node_local_width_disagreement_still_halts_rather_than_skipping() { + use crate::wal::format::batch::{BlobRecord, EmbeddingRecord}; + + // No declared embedding slot: the slot registers at the first vector's width. + let db = TidalDb::builder() + .ephemeral() + .with_schema(minimal_schema()) + .open() + .unwrap(); + + db.apply_replicated_blobs(vec![BlobRecord::Embedding(EmbeddingRecord { + entity_id: 1, + values: vec![1.0, 0.0, 0.0], + })]) + .expect("the first vector registers the slot at 3 dims"); + + let err = db + .apply_replicated_blobs(vec![BlobRecord::Embedding(EmbeddingRecord { + entity_id: 2, + values: vec![0.1, 0.2], + })]) + .expect_err("a node-LOCAL width disagreement must halt, never silently skip"); + assert!( + format!("{err}").contains("dimension mismatch"), + "the halt must name the real cause: {err}" + ); + db.close().unwrap(); +} + +/// A control record that fails validation keeps HALTING the round: dropping a +/// term marker or a roster record would diverge this replica's cluster view, +/// which is worse than a stalled receiver an operator can observe. +#[test] +fn a_malformed_control_record_still_halts_the_round() { + use crate::wal::format::batch::{BlobRecord, EmbeddingRecord, TermMarkerRecord}; + + let db = TidalDb::builder() + .ephemeral() + .with_schema(schema_with_embedding_slot(4)) + .open() + .unwrap(); + + let err = db + .apply_replicated_blobs(vec![ + BlobRecord::Embedding(EmbeddingRecord { + entity_id: 1, + values: vec![1.0, 0.0, 0.0, 0.0], + }), + BlobRecord::TermMarker(TermMarkerRecord { + term: 0, + leader_region: 0, + }), + ]) + .expect_err("term 0 must halt the round, not be skipped"); + assert!( + format!("{err}").contains("term 0"), + "the original error must reach the receiver unchanged: {err}" + ); + db.close().unwrap(); +} diff --git a/tidal/src/db/replication_ops.rs b/tidal/src/db/replication_ops.rs index 48ee63a..555bfdf 100644 --- a/tidal/src/db/replication_ops.rs +++ b/tidal/src/db/replication_ops.rs @@ -894,7 +894,10 @@ impl WeakBlobApplier { } impl crate::replication::receiver::ReplicatedBlobApplier for WeakBlobApplier { - fn apply_blobs(&self, records: Vec) -> crate::Result<()> { + fn apply_blobs( + &self, + records: Vec, + ) -> crate::Result { // The FULL write path, not the local-apply variant: the follower // re-journals every record in its OWN WAL (WAL-first, staged as one // batch so the round shares group-commit fsyncs — m11p3) so follower diff --git a/tidal/src/replication/receiver.rs b/tidal/src/replication/receiver.rs index 18dec3d..15ae85d 100644 --- a/tidal/src/replication/receiver.rs +++ b/tidal/src/replication/receiver.rs @@ -50,6 +50,22 @@ use crate::{ }, }; +/// Result of one apply round's blob batch. +/// +/// Distinguishes "everything applied" from "everything applied EXCEPT records +/// that no replica of this group could ever apply". The second case is not an +/// error — the stream must continue past such a record, or one malformed client +/// write costs the whole shard group its write quorum (2026-08-31) — but it is +/// never silent: the applier counts each dropped record on +/// `tidaldb_cluster_blobs_apply_failed_total` and logs it at ERROR, and the +/// receiver logs the seqno range it skipped over. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct BlobApplyOutcome { + /// Records dropped as structurally unapplicable on every replica, forever. + /// Zero on a clean round. + pub rejected: u64, +} + /// Applier for replicated blob (kind-1/2) records on a follower. /// /// Implemented by `TidalDb` (via a `Weak` adapter — see @@ -67,12 +83,17 @@ pub trait ReplicatedBlobApplier: Send + Sync { /// handed over by value so the engine can share them with its WAL /// writer by refcount instead of deep-cloning every payload. /// + /// A record that is unapplicable on EVERY replica is dropped rather than + /// returned as an error, and reported in [`BlobApplyOutcome::rejected`]; + /// see that type and `TidalDb::apply_replicated_blobs` for why the two + /// cases are split. + /// /// # Errors /// /// Any engine error; the receiver HALTS on it (a follower must never /// silently acknowledge an item it failed to durably record). Blobs are /// idempotent upserts, so redelivery after a mid-batch halt is safe. - fn apply_blobs(&self, records: Vec) -> crate::Result<()>; + fn apply_blobs(&self, records: Vec) -> crate::Result; } /// Handle to a running segment receiver thread. @@ -243,8 +264,15 @@ pub fn spawn_receiver( // is observable via SegmentReceiverHandle::died() while the // node is still open — not just at shutdown via join(). Log at // the failure site too -- mirrors wal::reader's per-corruption - // logging, except replication cannot skip-and-continue (a gap - // would diverge the replica), so we halt and surface. + // logging. + // + // Unconditional by design, and it is the HALT half of the + // decision made in `commit_segments`: the skippable case (a + // record no replica could ever apply) is dropped there, with + // per-record granularity, and never surfaces as an error + // here. Anything that does reach this point is node-local, + // transient, or version-dependent — skipping it would leave a + // gap and diverge the replica, so halt and surface instead. thread_died.store(true, Ordering::Release); tracing::error!( error = %e, @@ -765,6 +793,11 @@ fn commit_segments( // out of the prepared segments (ownership moves into the applier — no // deep clone); on failure the receiver halts and redelivery rebuilds // them from the segment bytes, so nothing here needs them back. + // Seqno range this round covers, captured before the blobs are drained: a + // skip must name WHICH records the stream moved past, or the ERROR line is + // unactionable. + let seq_lo = prepared.iter().map(|seg| seg.first).min().unwrap_or(0); + let seq_hi = prepared.iter().map(|seg| seg.last).max().unwrap_or(0); let group_blobs: Vec = prepared .iter_mut() .flat_map(|seg| seg.blobs.drain(..)) @@ -779,10 +812,48 @@ fn commit_segments( )))); }; let blob_count = group_blobs.len(); - if let Err(e) = applier.apply_blobs(group_blobs) { - return Err(WalError::Io(std::io::Error::other(format!( - "replicated blob batch apply failed ({blob_count} records): {e}" - )))); + // ── The halt-vs-skip decision, receiver side ── + // + // SKIP: the applier dropped one or more records that are structurally + // unapplicable on EVERY replica of this group, forever (their verdict + // reads only the record's bytes and the shared schema — see + // `TidalDb::apply_replicated_blobs_inner` Phase 1). The round still + // applied, so the frontier advances and the stream continues. Halting + // instead is what turned a single 128-dim vector into a shard-1 + // quorum-write outage on 2026-08-31: both followers pinned, `lag` + // growing, writes 503, and neither a restart nor a reseed escaped it + // (boot self-heal re-pulls the same record; the reseed snapshot is + // taken at the leader's applied frontier, which is BEHIND the poison). + // + // HALT: anything that reaches the `Err` arm is node-LOCAL (disk fault, + // poisoned lock, read-only) or version-dependent (an unknown batch kind + // a newer binary understands), or a control record whose loss would + // diverge this replica's term/roster view. All of those may apply on a + // later attempt or after an upgrade, so skipping them would silently + // drop valid replicated data — the exact hazard this halt exists to + // prevent. Keep halting, and let health degrade so an operator sees it. + match applier.apply_blobs(group_blobs) { + Ok(outcome) => { + if outcome.rejected > 0 { + tracing::error!( + rejected = outcome.rejected, + blob_count, + seq_lo, + seq_hi, + "replicated blob record(s) rejected as unapplicable on every replica; \ + SKIPPED and the stream continues (receiver stays healthy, frontier \ + advances). Each is counted on \ + tidaldb_cluster_blobs_apply_failed_total with its reason logged at \ + the applier" + ); + } + } + Err(e) => { + return Err(WalError::Io(std::io::Error::other(format!( + "replicated blob batch apply failed ({blob_count} records, seqnos \ + {seq_lo}..={seq_hi}): {e}" + )))); + } } } @@ -1775,7 +1846,7 @@ mod tests { applied: StdMutex>, } impl ReplicatedBlobApplier for RecordingApplier { - fn apply_blobs(&self, records: Vec) -> crate::Result<()> { + fn apply_blobs(&self, records: Vec) -> crate::Result { let lines: Vec = records .iter() .map(|record| match record { @@ -1792,7 +1863,7 @@ mod tests { }) .collect(); self.applied.lock().unwrap().extend(lines); - Ok(()) + Ok(BlobApplyOutcome::default()) } } diff --git a/tidal/src/storage/vector/lifecycle/ops.rs b/tidal/src/storage/vector/lifecycle/ops.rs index 26c3cd4..0dc248c 100644 --- a/tidal/src/storage/vector/lifecycle/ops.rs +++ b/tidal/src/storage/vector/lifecycle/ops.rs @@ -5,7 +5,7 @@ //! the source-of-truth vector in the entity store, and updates the HNSW index. use super::{ - super::{VectorError, VectorIndex}, + super::{VectorError, VectorIndex, validate_dimensions}, normalize::l2_normalize, serde::{embedding_store_key, serialize_embedding}, }; @@ -54,12 +54,11 @@ fn normalize_and_store( expected_dimensions: usize, storage: &dyn StorageEngine, ) -> Result, VectorError> { - if raw_vector.len() != expected_dimensions { - return Err(VectorError::DimensionMismatch { - expected: expected_dimensions, - got: raw_vector.len(), - }); - } + // The ONE dimension rule (see `validate_dimensions`): the pre-WAL guard on + // the write path calls the same function, so an embedding that passes there + // cannot fail here — which is what keeps a malformed vector out of the + // replicated log instead of halting every follower that receives it. + validate_dimensions(expected_dimensions, raw_vector.len())?; let normalized = l2_normalize(raw_vector)?; diff --git a/tidal/src/storage/vector/mod.rs b/tidal/src/storage/vector/mod.rs index 721f57b..c873d65 100644 --- a/tidal/src/storage/vector/mod.rs +++ b/tidal/src/storage/vector/mod.rs @@ -308,9 +308,15 @@ pub trait VectorIndex: Send + Sync { /// Validate that a vector's dimensionality matches an index's configuration. /// /// Returns [`VectorError::DimensionMismatch`] when `got != expected`. Used by -/// every [`VectorIndex`] implementation on the insert/search hot paths, so it -/// is `const` and allocation-free. -pub(super) const fn validate_dimensions(expected: usize, got: usize) -> Result<(), VectorError> { +/// every [`VectorIndex`] implementation on the insert/search hot paths, by the +/// embedding-lifecycle write prologue (`normalize_and_store`), AND by the +/// engine's pre-WAL embedding guard (`TidalDb::validate_embedding_for_log`), so +/// the "a vector must be exactly the slot's width" rule has exactly ONE +/// definition. Two copies would drift, and the copy on the apply path is the one +/// that halts a follower's receiver when it disagrees with the copy on the write +/// path — the 2026-08-31 shard-1 outage shape. It is `const` and +/// allocation-free. +pub(crate) const fn validate_dimensions(expected: usize, got: usize) -> Result<(), VectorError> { if got != expected { return Err(VectorError::DimensionMismatch { expected, got }); }