From 727fbfcb6b7c3259659e9432b1cfe56054ecf19c Mon Sep 17 00:00:00 2001 From: jx12n Date: Tue, 16 Jun 2026 22:34:21 -0600 Subject: [PATCH] fix(m12p6): 6-bug k3s 3-shard cluster repair (rc8+rc9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root-caused and fixed five sharding bugs exposed on the real k3s 3-shard cluster (rc5→rc7), plus a divergent-rejoin reseed loop found in rc9: 1. reseed shard-awareness (Bug 3, keystone): `run_boot_install_for_region` visits each hosted group's own shard subdir; per-group leader discovery appends `?shard=N` so a divergent shard heals from its own leader (not shard-0's WAL/term — cross-shard contamination). 2. leader self-join term (Bug 4): `become_leader_for_term` now calls `note_self_won_term` so the elected shard's `joined_term` is set and `cluster_promote` routes rebalances correctly (was: topology-era mis-read → legacy fenced promote → 500). 3. boot self-heal self-pull guard (Bug 2): `leader_shard != my_shard` gate prevents a node pulling its own stream (its stream isn't a registered peer) → eliminates the `PeerUnreachable(self)` loop. 4. scatter-merge degraded partial (Bug 1): failed shard logs + continues instead of `?`-failing the whole read; bounded read-admission semaphore (`offload.rs`) sheds as 429 instead of piling into a 36s p99. 5. WAL retention (Bug 5): `compact_wal_retained` keeps `WAL_RETENTION_SEGMENTS=4` most-recent sealed segments; online path gets the same retention clamp. Prevents brief-restart forced-reseed. 6. divergent-rejoin reseed loop (Bug 6, rc9): `note_quarantined` latches `from_seqno = stream_baseline` (not `frontier + 1`) so `wal_covers` returns `needed=true` and the snapshot installs instead of looping. Also: `TidalDb::close_shared` for deterministic HNSW save on cluster SIGTERM (HNSW graph was not saved when request-scoped Arc clones were alive at shutdown); updated profiling doc with full rc8/rc9 fix narrative; k8s recall job YAMLs. --- docs/profiling/m12-cluster-deploy-findings.md | 250 ++++++++++++++++++ k8s/cluster/statefulset.yaml | 18 +- tidal-server/src/cluster/election_driver.rs | 45 +++- tidal-server/src/cluster/node.rs | 154 ++++++++--- tidal-server/src/cluster/reseed.rs | 117 ++++++-- tidal-server/src/main.rs | 13 +- tidal-server/src/offload.rs | 47 ++++ tidal-server/tests/cluster_reseed.rs | 13 +- tidal-server/tests/reseed_install.rs | 3 + .../k8s/recall-skipseed-tidaldb2-job.yaml | 95 +++++++ tidal-stress/k8s/recall-verify-rc7-job.yaml | 94 +++++++ tidal-stress/k8s/t4-load-job.yaml | 4 +- tidal-stress/src/client.rs | 15 +- tidal/src/db/lifecycle.rs | 27 +- tidal/src/wal/compaction.rs | 60 +++++ 15 files changed, 874 insertions(+), 81 deletions(-) create mode 100644 tidal-stress/k8s/recall-skipseed-tidaldb2-job.yaml create mode 100644 tidal-stress/k8s/recall-verify-rc7-job.yaml diff --git a/docs/profiling/m12-cluster-deploy-findings.md b/docs/profiling/m12-cluster-deploy-findings.md index 30c84e1..0c10b35 100644 --- a/docs/profiling/m12-cluster-deploy-findings.md +++ b/docs/profiling/m12-cluster-deploy-findings.md @@ -1,5 +1,110 @@ # M12 cluster deploy + recall findings (k3s, 1536-dim) +## ⇒ STATUS (2026-06-17): the m12p4 3-shard layer is now FIXED in rc8 + +Operator deployed M12 to the real 3-shard k3s cluster (rc5→rc7), clean-slate +re-seeded a COMPLETE 100k/1536 corpus, then **root-caused and FIXED the five +m12p4 3-shard catch-up/read/reseed bugs** (each adversarially verified for +failover-safety before implementation). `m12-rc8` +(`sha256:5e3e5128562d5de98e73809f468ed6654e3445fecaf486f7879d9e9c6c553cb5`). +Workspace lib tests green (tidal-server 147/147, tidaldb 1910/1910; compaction +17/17; S=1 reseed 3/3). The 5 fixes: + +1. **Bug 3 (keystone) — reseed boot-install was never shard-aware in S>1.** + `run_boot_install` ran ONCE on the PARENT `/data/db`, but the per-group marker + + divergent WAL live in `/data/db/shard-NNNNN/` → it found no marker, no-op'd, + and the divergent shard re-quarantined + self-restarted forever (186 loops). + Fix: new `run_boot_install_for_region` runs the install per hosted shard subdir, + and `run_boot_install` now takes a `ShardId` — `candidates_for` resolves THAT + group's per-shard `grpc_addr`, `discover_leader` appends `?shard=N` — so a + divergent shard heals from ITS OWN leader (the original directory-only idea + would have installed shard-0's WAL/term into shard-N: cross-shard contamination, + caught in verification). `node::shard_subdir` → `pub(crate)` (single formatter). + `main.rs` calls the per-region entry. (`reseed.rs`, `main.rs`, `node.rs`) +2. **Bug 4 — an elected LEADER never recorded its own `joined_term`** (only the + follower `join_term_check` set it), so `cluster_promote` read the elected shard + as topology-era and routed `/cluster/shards/{id}/transfer` down the legacy + fenced leg → 500. Fix: `become_leader_for_term` calls a new monotonic + `ElectionRuntime::note_self_won_term(term)`. (`node.rs`, `election_driver.rs`) +3. **Bug 2 — boot self-heal self-pulled `my_shard`** (a node can't serve its own + stream to itself) → `PeerUnreachable(self)` forever. Fix: guard + `leader_shard != my_shard`. The "never reconverges" half was Bug 3 (a quarantined + node clears only via the now-working reseed). (`node.rs`) +4. **Bug 1 — cross-shard read collapse (p99 36s, 500-storm).** `scatter_merge` + fail-fasted (`?`) on one shard's error and `offload_read` had no admission bound. + Fix: `scatter_merge` serves the surviving groups as a degraded partial on a + per-shard error (logs it; the recall harness still measures true recall); a + bounded read-admission semaphore sheds 429 instead of piling onto the blocking + pool into a 36s p99. (Recall recovery itself comes from Bugs 2+3 restoring the + empty shards.) (`node.rs`, `offload.rs`) +6. **Bug 6 (rc9) — divergent-rejoin reseed loop.** Surfaced when a deposed leader + (frontier AT/ABOVE the new term's baseline, with old-term data) rejoins: the + `decide_join`→`ReseedRequired`/`Quarantine` latch used `from_seqno = frontier+1`, + which lands ABOVE the leader's WAL tail, so `wal_covers` answers `needed=false` + → the reseed install no-ops → the divergent suffix is re-detected → self-restart + loop. Fix: latch `from_seqno = stream_baseline` (threaded through + `join_term_check` from the heartbeat), which is `<= baseline` so `wal_covers` + returns `needed=true` → the snapshot installs and re-baselines onto the leader's + committed history (discarding the divergent suffix — uncommitted writes the + cluster elected past, so no acked-write loss). `m12-rc9` + (`sha256:818c923368ecca6e433c42a13672132e9b467c449fa23c035201fe7189f1f52b`). + (`election_driver.rs`, `node.rs note_quarantined`) + +5. **Bug 5 — WAL over-compaction forced reseed on a brief restart.** Compaction + reclaimed every sealed segment below the checkpoint, so a follower down across + one checkpoint hit the compacted-below refusal. Fix: retain the + `WAL_RETENTION_SEGMENTS=4` most-recent sealed segments past the checkpoint + (online + a new `compact_wal_retained` for shutdown); a too-far-behind follower + still correctly reseeds. (`tidal/src/wal/compaction.rs`, `tidal/src/db/lifecycle.rs`) + +Original handoff detail (pre-fix) retained below for context. + +## ⇒ DEV HANDOFF (2026-06-16, pre-rc8) + +Operator deployed M12 to the real 3-shard k3s cluster (rc5→rc7) and clean-slate +re-seeded a COMPLETE 100k/1536 corpus. Five fixes landed in-tree (working-tree +diff; rc7 image is built from them) and are **verified live**; the m12p4 sharded +catch-up/read layer is **NOT production-ready on k3s** and is handed back to you. + +**FIXED + verified (keep these):** +| fix | file | what it cures | +|-----|------|---------------| +| `TidalDb::close_shared(&self)` called explicitly from `ShardReplica::shutdown` | `tidal/src/db/lifecycle.rs`, `tidal-server/src/cluster/node.rs` | HNSW graph save never ran on cluster SIGTERM (relied on `drop(db)` hitting refcount 0; lingering request clones prevented it). Now runs. | +| concurrent per-shard shutdown (`thread::scope`) + grace 60→600s + `TIDAL_SHUTDOWN_DRAIN_MS=3000` | `node.rs` (`ClusterNode::shutdown`), `k8s/cluster/statefulset.yaml` | 3 shards saved SERIALLY overran the grace → SIGKILL mid-save. Now all 3 save concurrently in-window (verified: 3× "persisted HNSW graphs to disk"). | +| `count_alive_other_voters` uses the CA-trusting `self.blocking_client` | `node.rs` ~2820 | reseed self-restart quorum poll built a BARE reqwest client → every TLS peer poll failed → `alive_voters=0` → reseed NEVER auto-healed on a TLS cluster. Verified: now `alive_voters=2`, self-restart fires. | +| seed retry: 8→40 attempts, backoff→500ms (~12s budget) | `tidal-stress/src/client.rs` `retry_write` | a fast bulk seed silently DROPPED ~0.4–4% of the corpus when a leader shed 429. Now seeds `100000/100000`. | +| single-shard "blocker #3" was a STALE ConfigMap (shards: commented out live) | `k8s/cluster/topology-configmap.yaml` (re-applied) | NOT a code bug. | + +**OPEN — m12p4 sharded path, needs your repair (gates T5/T4/1M-on-shards BLOCKED):** +1. **Cross-shard `/vector_search` collapses** even on a healthy follower with all + 3 shards LOCAL + lag=0: p99 **36 s**, recall **~0.42** (≈ consulting one shard), + 70–100% error at ≥100 rps. The local scatter-merge / read-offload path + (`vector_search` → `scatter_merge` over `hosted_dbs`, node.rs ~7204) hangs and + returns partial. This is the gate-blocker — reads are the SLA. +2. **Catch-up "peer shard sN not in configuration"** (`tidal-net` `PeerUnreachable`, + error.rs:13): a rejoining node's per-shard gRPC pool never registers the source + peer → its catch-up stream can't open → it never reconverges (seen on shards 0/2). +3. **term-0 topology-era divergence trap**: a leader that took term-0 bootstrap + writes, hit by a term transition under the seed, gets a divergent suffix → + self-restarts (correct now), but then STALLS rejoining on (2) + a term-behind + refusal. Net: self-restart fires but the node can't actually rejoin the 3-shard + cluster (quorum survives 2/3; it does not self-repair to 3/3). +4. **3-shard leader-transfer verb 500s** for an already-elected shard + (`cluster_promote`/`promote_local` routes the fenced transfer through the legacy + leg, fenced at `joined_term≥1`). +5. **WAL compacts past a briefly-restarted follower** → forced reseed instead of + cheap stream catch-up (periodic-checkpoint compaction is too aggressive for the + rollout window). + +**Recommendation:** the **single-shard** topology has a working read path (prior +session: recall 0.967, p99 7 ms @ 100k) — G1/G2/T4/1M reads are provable there +today. The sharded gates need (1)+(2)+(3) fixed first. Also note the rc6 grace bump +to 600s makes rolling restarts slow on a loaded cluster (the graph save is the long +pole); a periodic graph checkpoint (the `checkpoint_graphs` helper is reusable from +the periodic thread) would let the grace drop back and survive SIGKILL too. + +--- + Date: 2026-06-15. Cluster: orchard9-k3sf, namespace `tidaldb-cluster`, 3 nodes (4 vCPU / 16 GiB each), local-path PVCs. Image built from HEAD `8e39ee1` + the working-tree T4-TLS fixes. **Goal: deploy M12 to the real cluster and get the @@ -217,3 +322,148 @@ let restarts be fast clean reloads, and make the HNSW-persist fix pay off. Deployed: `m12-rc4` (`sha256:28d0c45e59c487cf0d5e6a574c4c9997b82c6a358d67e74f326ff2c0c0925193`). + +## rc5 + CLEAN-SLATE 3-shard rebuild (2026-06-16) + +Operator-authorized clean-slate wipe (StatefulSet + 3 PVCs deleted, kustomization +re-applied) → fresh **3-shard** cluster on `m12-rc5` +(`sha256:8792b283d6c4d41c0b303afa777fa1a7583a94cca932d73580132c3a0e60e01b`). +This removed the accumulated-PVC-churn confound and enabled the 3-shard topology +that the single-shard layout could not host in place. + +**rc5 = rc4 + the persist-save fix.** Root cause of "Fix #2 doesn't materialize": +`ShardReplica::shutdown` relied on `drop(db)` reaching `TidalDb::Drop` (refcount +0) to run `shutdown_inner`→`checkpoint_embedding_graphs`. Under load a +request-scoped clone keeps the `Arc` alive, so `Drop` never fires and the graph +is never saved. Fix: a new `pub fn TidalDb::close_shared(&self)` (idempotent via +the `closed` CAS) that `ShardReplica::shutdown` calls EXPLICITLY on the shared +handle — the checkpoint now runs regardless of refcount. Verified: +`m12p6_graph_persistence` (save/load roundtrip) + `cluster_graph_persistence` +(multiproc SIGTERM reaches the close) both green; native `cargo check` clean. + +**Blocker #3 (single-shard at runtime) — RESOLVED, was a stale ConfigMap.** The +LIVE `tidaldb-cluster-topology` ConfigMap had the `shards:` block COMMENTED OUT +(the on-disk manifest had it enabled but was never re-applied). The fresh deploy +re-applied it: `/cluster/status` now reports **3 shards**, every pod carries +`shard-00000/01/02` subdirs, and a 100k seed hash-balances ~even across them +(≈42k events/shard). NOT a code bug. + +**Ingest is FAST on a fresh 3-shard cluster.** The 100k seed (200k writes: +items + embeddings) committed at **~680 events/s aggregate**, lag 0 throughout — +vs the prior cluster's ~80–200/s. The old slowness was upsert-tombstoning over +an existing corpus + single-shard, not a fundamental ceiling. ⇒ a 1M seed is +~25 min, not hours. + +### Still-open (noted, not yet fixed) + +- **3-shard leader-transfer verb 500s on a fresh mixed-era cluster.** On the + parallel fresh boot tidaldb-0 won leadership of ALL 3 shards. `POST + /cluster/shards/{id}/transfer` to rebalance returns 503 ("promotion target + returned 500"): for a shard already in an elected term (≥1) the fenced + transfer routes through the internal legacy-promote leg, which `promote_local` + fences at `joined_term ≥ 1` → 500. Operator-convenience verb (not a + correctness issue); the cluster serves fine with one pod leading all shards. + Deferred — the fix is in the delicate election machinery the code repeatedly + warns against duplicating. +- **term-gate same-term `Stale`** + **reseed-on-restart churn**: to be + re-evaluated on the clean cluster (hypothesis: largely PVC-churn cruft). + +### Fix #2 persist — root-caused on the cluster, rc6 (2026-06-16) + +Restarting a follower (tidaldb-2, ~32k vectors/slot/shard) on rc5 showed the +persist SAVE now **runs** (rc4 wrote nothing): `/data/db/shard-00000/vector/` +held a complete **141 MB `item__content_vector.usearch.tmp`** — `close_shared` +executed `checkpoint_graphs`. But the `.tmp` was never renamed to the final +`.usearch`, and shards 1/2 had no `vector/` dir at all → the boot correctly +REBUILT (`load_persisted_slot` rejects a `.tmp`). Root cause: `ClusterNode::shutdown` +saved the 3 shards **serially** (`node.rs:4161`), and a single 1536-dim slot's +USearch serialize+fsync is the long pole, so 3 shards overran the 60s SIGTERM +grace and k8s SIGKILLed mid-save. + +**rc6 = rc5 +** (a) `ClusterNode::shutdown` now saves the hosted shards +**concurrently** via `thread::scope` (each `ShardReplica::shutdown` is `&self` +and touches only its own db/shard); (b) `terminationGracePeriodSeconds` 60→600 +and `TIDAL_SHUTDOWN_DRAIN_MS=3000` so the save starts promptly and has ample +budget. `m12-rc6` +(`sha256:8fc56cb8e9014812df3de81fbdb4cfed6fdecd3a19ec7c2eb836a3defebdf068`). +Open: a SIGKILL/crash still skips the save (boot rebuilds) — the production-robust +follow-up is a periodic graph checkpoint (the `checkpoint_graphs` helper is +already reusable from the periodic thread; not yet wired because the save is +heavy and needs an interval + change-detection design). + +### Seed completeness — single-leader 429 saturation + +The fresh parallel boot left tidaldb-0 leading ALL 3 shards (the election race; +the per-shard transfer verb 500s, above). A 100k recall seed then dropped +**4396/100000 items**: every write funnels through tidaldb-0's ONE bounded write +pool, which sheds 429 under sustained load; the stress seed retries 429 only 8× +(~900ms total backoff) and then drops. Real product behavior (backpressure), but +it means a single-leader cluster can't absorb a fast bulk seed. Mitigations: +balanced leaders (3 pools) and/or lower `--seed-concurrency`. The rc6 rollout +terminates tidaldb-0 LAST, so leadership fails over OFF it — a free rebalance. + +### rc7 (2026-06-16): self-restart auto-heal + the wedge it exposed → 2nd wipe + +The rc6 rollout (the FIRST restart of a loaded 3-shard cluster) exposed three +real catch-up/reseed bugs (NOT cruft — clean cluster): +1. **WAL compacts past a briefly-restarted follower** (shard 1: "WAL compacted + below seqno 1, earliest 63873") → forced reseed instead of cheap stream + catch-up. Aggravated by (3). +2. **Reseed self-restart ALWAYS refused on a TLS cluster** — + `count_alive_other_voters` built a BARE `reqwest::blocking::Client` (no cluster + CA), so every `https://peer:9500/cluster/status/local` poll failed TLS verify → + `alive_voters_excluding_self=0` → refuse (would "break quorum") → a node that + needed a reseed wedged forever. **FIXED (rc7):** use the CA-trusting + `self.blocking_client`. VERIFIED live: the same path now logs + `alive_voters_excluding_self=2` and the self-restart FIRES. +3. **Shard-2 catch-up "peer shard s2 not in configuration"** — the per-shard gRPC + pool lacks the source peer while reseed-pending; likely downstream of (2) + (incomplete group membership), to re-confirm now that (2) auto-heals. + +**Persist save VERIFIED on the cluster (rc6):** a self-restart logged +`persisted HNSW graphs to disk` for ALL THREE shards (shard-00000/01/02) +concurrently within the grace window — the rc6 parallel-save + 600s grace works. + +**But:** the rc7 self-restart then LOOPED on the churned cluster — a divergent +term-0 suffix (`tail_term=0` vs elected `term=1`, frontier 65108) that the boot +snapshot-install did not clear → quarantine→self-restart→re-detect→loop. This is +mixed-era churn (the term-0 topology-bootstrap leadership orphaning uncommitted +data across a leadership change). Resolved by a **2nd operator-authorized +clean-slate wipe** → pristine **rc7** baseline +(`sha256:d42fe27c4d6a63e5c5103886d4bfa25998d992c4dfd3b536f01469a1edab94d0`). +Lesson: commit fully (lag=0) before any leadership-changing restart; the term-0 +topology era is the divergence trap. The boot-install-not-clearing-a-divergent- +suffix loop is a remaining upstream item for the devs. + +### The 3-shard topology is NOT production-ready on real k3s (dev handoff) + +With the seed-retry fix the corpus seeds COMPLETE (`100000/100000 in 488.5s`), +but exercising the seeded 3-shard cluster surfaced that the **3-shard +catch-up/read paths are broken** — distinct from (and on top of) the persist / +self-restart fixes, which DO work. These are the m12p4 sharding code (kind-tested +only); they need upstream work before T5/T4/1M can pass on 3 shards: + +1. **Cross-shard `/vector_search` collapses under load.** A recall ramp on a + busy LEADER (tidaldb-0, owns 2 groups) returned **p99 33,000 ms, 81–100% + error, recall 0.3082** — the ≈⅓ recall is the tell that the read consulted ONE + shard. Root: the read target (a leader) self-restarted mid-ramp (see 3) and, + while degraded, its sibling-group dbs were unavailable so the gather fell to a + remote fan-out that timed out. Reads MUST be sent to a non-leader; even then + the gather is fragile. +2. **Catch-up "peer shard sN not in configuration"** recurs (now shard 0 AND 2): + a rejoining node's per-shard gRPC pool never registers the source peer, so its + catch-up stream can't open → it can't converge. This blocks clean rejoin. +3. **A leader hits the term-0 divergence trap under the seed** and self-restarts + (`exitCode 0` — the Bug-B fix firing), then STALLS rejoining on (2) plus a + term-behind refusal (`puller term 0 behind source's 1; rejoin`). Net: a node + that self-restarts to heal cannot actually rejoin the 3-shard cluster. Quorum + survives (2/3), the corpus is intact on the healthy two, but the cluster does + not self-repair to 3/3. + +**Net:** the rc5–rc7 fixes (persist, parallel-save, self-restart TLS-trust, +seed-retry) are real and verified. But the m12p4 **3-shard catch-up + cross-shard +read** layer needs upstream repair before the sharded-topology gates (T5, and +T4/1M *on shards*) can run. The **single-shard** topology had a working read path +(prior session: recall 0.967, p99 7 ms @ 100k) — the read gates G1/G2 (and T4/1M +reads) are provable there today; T5 inherently needs the (currently broken) +sharded path. diff --git a/k8s/cluster/statefulset.yaml b/k8s/cluster/statefulset.yaml index 65cd36e..70ac61d 100644 --- a/k8s/cluster/statefulset.yaml +++ b/k8s/cluster/statefulset.yaml @@ -45,10 +45,13 @@ spec: prometheus.io/path: "/metrics" spec: # SIGTERM flips readiness to 503 (pod leaves the client Service), drains - # in-flight requests, then checkpoints + fsyncs the WAL before exit. The - # cluster path also wants the leader-lease/heartbeat windows to lapse so a - # successor is elected cleanly. 60s covers the whole sequence. - terminationGracePeriodSeconds: 60 + # in-flight requests, then checkpoints + fsyncs the WAL AND saves every + # shard's HNSW graph before exit (m12p6). The graph save is the long pole at + # the production shape (~32k vectors/slot × 3 shards, serialized + fsynced), + # so the grace must cover it or k8s SIGKILLs mid-save and the next boot + # rebuilds. 600s is a generous ceiling; the bounded drain (below) starts the + # save early, and a clean save typically finishes in well under a minute. + terminationGracePeriodSeconds: 600 # Spread the three pods across distinct nodes so a single node loss takes # at most one voter — preserving quorum (2 of 3). ScheduleAnyway (not # DoNotSchedule) so a smaller cluster still schedules, just less spread. @@ -81,7 +84,7 @@ spec: mountPath: /data containers: - name: tidaldb - image: registry.threesix.ai/tidal/server@sha256:28d0c45e59c487cf0d5e6a574c4c9997b82c6a358d67e74f326ff2c0c0925193 # m12-rc4 (= rc3 + ingest admission-control/forward-retry + SIGTERM bounded-drain graph-save) + image: registry.threesix.ai/tidal/server@sha256:818c923368ecca6e433c42a13672132e9b467c449fa23c035201fe7189f1f52b # m12-rc9 (= rc8 + Bug 6: reseed latch re-baselines from the stream baseline, not frontier+1, so a divergent-rejoin node force-installs the snapshot instead of looping) imagePullPolicy: IfNotPresent # The image ENTRYPOINT is the bare binary. We override the command with # a tiny /bin/sh wrapper (the bookworm-slim runtime HAS a shell) so we @@ -170,6 +173,11 @@ spec: value: info - name: TIDAL_ALLOW_EXPERIMENTAL_CLUSTER value: "1" + # m12p6: shorten the post-SIGTERM in-flight drain so the (long) HNSW + # graph save starts promptly within the grace window instead of after + # the full 15s default. 3s is ample for loopback/in-cluster drain. + - name: TIDAL_SHUTDOWN_DRAIN_MS + value: "3000" ports: - name: http containerPort: 9500 diff --git a/tidal-server/src/cluster/election_driver.rs b/tidal-server/src/cluster/election_driver.rs index 58800dc..01fbeab 100644 --- a/tidal-server/src/cluster/election_driver.rs +++ b/tidal-server/src/cluster/election_driver.rs @@ -93,6 +93,25 @@ impl ElectionRuntime { self.joined_term.load(Ordering::Acquire) } + /// Record that THIS node JOINED `term` by WINNING it. A node never processes + /// its own heartbeats, so `join_term_check` (the only other writer) never runs + /// for a self-won term and `joined_term` would otherwise stay 0 on an elected + /// LEADER — which makes `cluster_promote` mis-classify the elected shard as + /// topology-era and route a leader rebalance down the legacy fan-out promote + /// (which an elected target correctly fences, returning 500). The leader is the + /// authority on its own term's history (its WAL carries the kind-3 term marker, + /// so the three-way join rule trivially yields `Clean` via the + /// `own.tail_term == term` clause — there is no divergence check to run against + /// itself). Monotonic (`fetch_max`): never lowers an already-higher joined term, + /// and a future divergence can only appear at a term `> term`, which is NOT + /// short-circuited and still runs the full `decide_join`. + pub fn note_self_won_term(&self, term: u64) { + if term == 0 { + return; + } + self.joined_term.fetch_max(term, Ordering::AcqRel); + } + /// The machine's current role (for status). pub fn role(&self) -> Role { self.lock_machine().role() @@ -222,7 +241,7 @@ impl ElectionRuntime { /// /// Returns `false` only when the node is (or just became) quarantined; the /// reseed-latch path returns `true` (the join proceeds). - fn join_term_check(&self, term: u64, prev_log: LogPosition) -> bool { + fn join_term_check(&self, term: u64, prev_log: LogPosition, baseline: u64) -> bool { if term == 0 || term <= self.joined_term.load(Ordering::Acquire) { return !self.is_quarantined(); } @@ -254,7 +273,13 @@ impl ElectionRuntime { term, position.tail_term, position.frontier, - Some(prev_log.frontier), + // m12p6: re-baseline the reseed from the new term's STREAM + // BASELINE, not `frontier + 1`. A divergent node's frontier + // sits AT/ABOVE the baseline, so `frontier + 1` lands above the + // leader's tail → `wal_covers` answers needed=false → the + // snapshot never installs and the divergent suffix loops. A + // `from_seqno <= baseline` forces the re-baselining snapshot. + baseline, ); false } @@ -268,10 +293,14 @@ impl ElectionRuntime { // node is missing in the PREVIOUS stream's numbering // (`own.frontier + 1`); the marker boot re-baselines onto the // leader's stream via the snapshot regardless. - node.latch_reseed_marker( - ReseedReason::SnapshotRequired, - position.frontier.saturating_add(1), - ); + // m12p6: re-baseline from the new term's STREAM BASELINE, not + // `frontier + 1`. When the node's frontier sits at/above the + // baseline (post-baseline old-term data), `frontier + 1` lands + // above the leader's tail → `wal_covers` answers needed=false → the + // reseed install no-ops → re-detect → self-restart loop. A + // `from_seqno <= baseline` forces the snapshot that re-baselines the + // node onto the leader's committed history. + node.latch_reseed_marker(ReseedReason::SnapshotRequired, baseline); self.joined_term.store(term, Ordering::Release); true } @@ -569,7 +598,9 @@ impl tidal_net::ElectionHooks for NodeElectionHooks { // is control-plane (the node keeps voting and reporting status); // only the data plane is fenced. let before = self.runtime.joined_term.load(Ordering::Acquire); - let joined = self.runtime.join_term_check(term, prev_log); + // `stream_baseline` is the new term's clamp floor — pass it so a reseed + // latch re-baselines from it (m12p6 divergent-rejoin fix), not frontier+1. + let joined = self.runtime.join_term_check(term, prev_log, stream_baseline); if joined && term > before && let Some(node) = self.runtime.node.upgrade() diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 6fcb29b..4edfab5 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -839,11 +839,21 @@ impl ShardReplica { transport.request_catchup(discovered_shard, recovered_tail + 1); } else if !is_leader_at_boot { let leader_shard = shard_of_region(leader); - let applied = db - .replication_state() - .applied_seqno(leader_shard) - .unwrap_or(0); - transport.request_catchup(leader_shard, applied + 1); + // A node can NEVER pull its own stream from itself: only the node whose + // `source_shard == S` serves shard S's stream (server.rs returns + // NOT_FOUND otherwise), and in full placement the group's TOPOLOGY + // leader IS self for the group this node leads — so the old + // unconditional pull emitted `PeerUnreachable(my_shard)` forever + // (handle_for(my_shard) is None; self is never a registered peer). For a + // self-led group the receiver gap path (from real ship source shards) + // and live election traffic drive convergence instead. + if leader_shard != my_shard { + let applied = db + .replication_state() + .applied_seqno(leader_shard) + .unwrap_or(0); + transport.request_catchup(leader_shard, applied + 1); + } } let commit = ship_queue.commit_index(); @@ -1094,10 +1104,23 @@ impl ShardReplica { // it (firing the deterministic close); a second call swaps `None`→`None` // and drops nothing — idempotent. if let Some(db) = self.db.swap(None) { - // Drop the strong handle we just took; if it was the last reference - // (request-scoped clones have drained, the transport holds only a - // `Weak`), this runs `TidalDb::Drop` → `shutdown_inner` → - // `checkpoint_embedding_graphs` synchronously, right here. + // m12p6 FIX: run the deterministic close on the SHARED handle EXPLICITLY + // rather than relying on `drop(db)` reaching `TidalDb::Drop`. Under load + // a request-scoped clone (or a background task) often still holds the + // `Arc` at this point, so `drop` here is NOT the last reference, `Drop` + // never fires, and `checkpoint_embedding_graphs` never runs — the + // observed cluster gap where a follower restart rebuilt the HNSW graph + // because no `{data_dir}/vector` file was written on SIGTERM. `close_shared` + // takes `&self` and is idempotent (the `closed` CAS), so the trailing + // `drop(db)` and any later `Drop` from a lingering clone are no-ops. + if let Err(e) = db.close_shared() { + tracing::error!( + region = %self.region_name, + error = %e, + "region cluster node shutdown: deterministic close reported a \ + durable-flush error (state may be partially flushed)" + ); + } drop(db); tracing::info!( region = %self.region_name, @@ -2524,6 +2547,15 @@ impl ShardReplica { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) = prev; *write_recovered(&self.leader, "leader") = Some(self.region); + // Record that we JOINED this elected term by WINNING it. A node never + // processes its own heartbeats, so without this the leader's `joined_term` + // stays 0 and `cluster_promote` mis-reads the elected shard as topology-era, + // routing a leader rebalance (`/cluster/shards/{id}/transfer`) down the + // LEGACY fan-out promote — which an elected target correctly fences with a + // 500. The won term marker is durable above, so this is a clean self-join. + if let Some(rt) = self.election_runtime.get() { + rt.note_self_won_term(term); + } tracing::info!( term, baseline, @@ -2663,13 +2695,18 @@ impl ShardReplica { term: u64, tail_term: u64, frontier: u64, - baseline: Option, + baseline: u64, ) { self.cluster_metrics.set_divergence_quarantined(true); - // The quarantine frontier is the seqno the reseed must resume from: the - // node's applies are fenced from here, so the next-boot snapshot fetch - // requests from `frontier + 1` (the first seqno past the divergent tail). - self.latch_reseed_marker(ReseedReason::Quarantine, frontier.saturating_add(1)); + // m12p6: the reseed must re-baseline from the new term's STREAM BASELINE, + // NOT `frontier + 1`. A divergent node's frontier sits AT/ABOVE the + // baseline, so `frontier + 1` lands above the leader's WAL tail → the + // leader answers `needed=false` (nothing to ship above its tail) → the + // reseed install no-ops, the divergent suffix is re-detected, and the node + // self-restart-loops. A `from_seqno <= baseline` makes `wal_covers` return + // needed=true → the snapshot installs and re-baselines onto the leader, + // discarding the divergent suffix. + self.latch_reseed_marker(ReseedReason::Quarantine, baseline); tracing::error!( term, tail_term, @@ -2805,17 +2842,14 @@ impl ShardReplica { /// `/cluster/status/local` (a short-timeout blocking poll). Used by the §2.4 /// quorum refusal. Excludes this node. fn count_alive_other_voters(&self) -> usize { - let client = match reqwest::blocking::Client::builder() - .timeout(forward::STATUS_PEER_TIMEOUT) - .build() - { - Ok(c) => c, - Err(e) => { - tracing::error!(error = %e, "self-restart quorum poll: status client build failed; \ - conservatively reporting 0 alive others (refuse the restart)"); - return 0; - } - }; + // m12p6 FIX: use the SHARED `blocking_client`, which is built with the + // cluster CA as a trust anchor (`build_forwarding_clients`). A freshly + // built bare `reqwest::blocking::Client` trusts only the SYSTEM roots, so + // on a TLS cluster every `https://peer:9500/cluster/status/local` poll + // fails certificate verification → this returned 0 → the reseed + // self-restart was ALWAYS refused (it believed it would break quorum) → + // a node that needed a reseed wedged forever instead of self-healing. + // The per-request timeout keeps the tight status budget. let mut alive = 0usize; for (rid, _name, http) in self.all_regions_for_status() { if rid == self.region { @@ -2823,7 +2857,10 @@ impl ShardReplica { } let Some(http_addr) = http else { continue }; let url = peer_url(&http_addr, "/cluster/status/local"); - let mut req = client.get(&url); + let mut req = self + .blocking_client + .get(&url) + .timeout(forward::STATUS_PEER_TIMEOUT); if let Ok(key) = std::env::var("TIDAL_API_KEY") { req = req.bearer_auth(key); } @@ -3619,7 +3656,11 @@ fn build_forwarding_clients( /// The ONE authoritative formatter for the per-group directory layout — a /// durability contract any ops tooling (backup, `tidalctl`, the L3 rebalance /// stager) must format identically to find an existing group's data. -fn shard_subdir(shard: ShardId) -> String { +/// +/// `pub(crate)` so the boot-time reseed (`cluster::reseed::run_boot_install_for_region`) +/// resolves each hosted group's marker-bearing subdir via the SAME formatter — +/// the parent-dir-only install silently skipped per-group markers in S>1. +pub(crate) fn shard_subdir(shard: ShardId) -> String { format!("shard-{:05}", shard.0) } @@ -4145,8 +4186,27 @@ impl ClusterNode { /// request-scoped clones have drained, and the checkpoint runs synchronously. pub fn shutdown(&self) { self.shutting_down.store(true, Ordering::Release); - for replica in self.hosted() { - replica.shutdown(); + let hosted: Vec<&Arc> = self.hosted().collect(); + // m12p6: when this node hosts MULTIPLE shard groups, save their HNSW + // graphs CONCURRENTLY. Each `ShardReplica::shutdown` serializes + fsyncs + // its slot graphs (~100 MB+ per slot at the 1536-dim production shape); + // run serially across S groups that is S × the slowest save, which on a + // 3-shard node routinely overran the SIGTERM grace window → k8s SIGKILL + // mid-save → the next boot rebuilt instead of loading. Each replica's + // close is `&self` and touches only its own db/shard, so the saves are + // independent and safe to run in parallel; `thread::scope` joins them all + // before returning (the process must not exit until every graph is durable). + if hosted.len() <= 1 { + for replica in &hosted { + replica.shutdown(); + } + } else { + std::thread::scope(|s| { + for replica in &hosted { + let replica = Arc::clone(replica); + s.spawn(move || replica.shutdown()); + } + }); } } } @@ -6483,15 +6543,43 @@ where F: Fn(&TidalDb) -> std::result::Result<(Vec, usize), ServerError>, { // S=1: the one group's result is already ranked + limited by the engine. + // A single group's error IS the read's error — there is nothing to degrade to. if let [only] = dbs { return per_db(only); } let mut merged: Vec = Vec::new(); let mut total = 0usize; - for db in dbs { - let (items, candidates) = per_db(db)?; - total = total.saturating_add(candidates); - merged.extend(items); + let mut ok_groups = 0usize; + let mut last_err: Option = None; + for (idx, db) in dbs.iter().enumerate() { + match per_db(db) { + Ok((items, candidates)) => { + total = total.saturating_add(candidates); + merged.extend(items); + ok_groups += 1; + } + // m12p6: a SINGLE shard's probe error (e.g. a shard still converging + // after a restart) must NOT abort the whole corpus-wide read with a + // 500 — the old `?` turned one churning shard into a request-wide error + // storm. Record it and serve the surviving groups as an honest + // (reduced-recall) partial; the recall harness measures actual recall, + // so a silently-degraded shard cannot pass the gate. + Err(e) => { + tracing::warn!( + group_index = idx, + error = %e, + "cross-shard read: a local shard probe failed; serving the \ + remaining groups as a degraded partial" + ); + last_err = Some(e); + } + } + } + // Only when EVERY group failed is there no answer to return — surface it. + if ok_groups == 0 { + return Err(last_err.unwrap_or_else(|| { + ServerError::Cluster("cross-shard read: every local shard probe failed".into()) + })); } merged.sort_by(|a, b| { score(b) diff --git a/tidal-server/src/cluster/reseed.rs b/tidal-server/src/cluster/reseed.rs index 288009f..d1186ed 100644 --- a/tidal-server/src/cluster/reseed.rs +++ b/tidal-server/src/cluster/reseed.rs @@ -398,8 +398,52 @@ pub fn run_boot_install( topology: &TopologySpec, region_name: &str, data_dir: &Path, + shard: ShardId, ) -> Result { - run_boot_install_with(topology, region_name, data_dir, handshake_window()) + run_boot_install_with(topology, region_name, data_dir, shard, handshake_window()) +} + +/// Run the boot-time reseed install for EVERY shard group this region hosts, +/// each against that group's OWN data subdir (the m11p6 layout) and with +/// SHARD-TARGETED leader discovery. +/// +/// Fixes the multi-shard reseed loop: the per-group reseed marker + the divergent +/// WAL live in `/shard-{:05}` (`ShardReplica`'s own dir), but the legacy +/// single-call `run_boot_install(.., data_dir)` loaded the marker from the PARENT +/// dir, found none, returned `NotNeeded`, and never healed — so a divergent shard +/// re-quarantined and self-restarted forever. This visits each hosted group's +/// subdir (S==1 resolves to `data_dir` verbatim, preserving the byte-for-byte +/// legacy path) and discovers/dials THAT shard's leader (not the default group's). +/// +/// # Errors +/// +/// Propagates an unrecoverable install fault (see [`run_boot_install`]) or a +/// topology resolution error. +pub fn run_boot_install_for_region( + topology: &TopologySpec, + region_name: &str, + data_dir: &Path, +) -> Result> { + let groups = topology.resolve_shard_groups()?; + let single = groups.len() == 1; + let mut outcomes = Vec::new(); + for group in &groups { + // Only groups THIS region replicates have local data to heal. + if !group.replicas.iter().any(|r| r.name == region_name) { + continue; + } + let group_dir = if single { + data_dir.to_path_buf() + } else { + data_dir.join(super::node::shard_subdir(group.shard)) + }; + // A group subdir may not exist yet on a never-run node; the install's + // recover_swap + marker load both tolerate an absent/empty dir (=> + // NotNeeded), so this is safe to call unconditionally per hosted group. + let outcome = run_boot_install(topology, region_name, &group_dir, group.shard)?; + outcomes.push((group.shard, outcome)); + } + Ok(outcomes) } /// [`run_boot_install`] with an explicit handshake window. @@ -416,6 +460,7 @@ pub fn run_boot_install_with( topology: &TopologySpec, region_name: &str, data_dir: &Path, + shard: ShardId, handshake_window: Duration, ) -> Result { let dirs = SwapDirs::derive(data_dir)?; @@ -442,7 +487,7 @@ pub fn run_boot_install_with( "boot-time reseed marker present: discovering a leader to snapshot-install from (§2.7)" ); - let candidates = candidates_for(topology, region_name); + let candidates = candidates_for(topology, region_name, shard)?; if candidates.is_empty() { tracing::error!( "reseed boot: no peer candidates in the topology to discover a leader from; \ @@ -482,7 +527,8 @@ pub fn run_boot_install_with( }; while Instant::now() < deadline { - let Some(leader) = discover_leader(&status_client, &candidates, api_key.as_deref()) else { + let Some(leader) = discover_leader(&status_client, &candidates, api_key.as_deref(), shard) + else { std::thread::sleep(backoff); backoff = (backoff * 2).min(BACKOFF_MAX); continue; @@ -618,26 +664,47 @@ impl std::fmt::Display for SeedInstallError { } } -/// The candidate poll set (§2.7): the topology regions' HTTP addresses plus -/// their advertised gRPC addresses, excluding this node. -fn candidates_for(topology: &TopologySpec, region_name: &str) -> Vec { - // The RegionId is the region's 0-based topology declaration index - // (topology.rs: "the 0-based index ... IS its RegionId"), so enumerate - // BEFORE filtering to keep every candidate's id stable. - topology - .regions +/// The candidate poll set (§2.7) for the reseed of shard group `shard`: the +/// OTHER replicas of THAT group, each carrying the group's PER-SHARD gRPC address +/// (the resolved replica's `grpc_addr` already includes the `+shard_id` port +/// offset) plus the replica's node-level HTTP address (the `:9500` plane is +/// per-node, not per-shard). Shard-targeting is load-bearing: a parent-default +/// poll would discover the WRONG shard's leader and install foreign data/term +/// into shard `shard`'s subdir. +/// +/// # Errors +/// +/// Propagates a topology resolution error. +fn candidates_for( + topology: &TopologySpec, + region_name: &str, + shard: ShardId, +) -> Result> { + let groups = topology.resolve_shard_groups()?; + let Some(group) = groups.iter().find(|g| g.shard == shard) else { + return Ok(Vec::new()); + }; + // The node-level HTTP address for a region (the `:9500` status/forward plane + // is shared by every shard the node hosts; only the gRPC port is per-shard). + let http_for = |name: &str| { + topology + .regions + .iter() + .find(|r| r.name == name) + .and_then(|r| r.http_addr.clone()) + }; + Ok(group + .replicas .iter() - .enumerate() - .filter(|(_, r)| r.name != region_name) - .filter_map(|(idx, r)| match (&r.http_addr, &r.grpc_addr) { - (Some(http), Some(grpc)) => Some(Candidate { - region: u16::try_from(idx).unwrap_or(u16::MAX), - http_addr: http.clone(), - grpc_addr: grpc.clone(), - }), - _ => None, + .filter(|r| r.name != region_name) + .filter_map(|r| { + http_for(&r.name).map(|http| Candidate { + region: r.region.0, + http_addr: http, + grpc_addr: r.grpc_addr.clone(), + }) }) - .collect() + .collect()) } /// This region's own gRPC TLS material (the snapshot fetch dials the leader with @@ -660,9 +727,15 @@ fn discover_leader( client: &reqwest::blocking::Client, candidates: &[Candidate], api_key: Option<&str>, + shard: ShardId, ) -> Option { + // The `?shard=N` selector makes `/cluster/status/local` report SHARD N's + // leadership (status_local resolves `replica_for(sel.shard_id())`); without it + // a multi-leader sharded node answers for its default (lowest-id) group, so we + // would discover the wrong shard's leader/term. + let path = format!("/cluster/status/local?shard={}", shard.0); for cand in candidates { - let url = super::forward::peer_url(&cand.http_addr, "/cluster/status/local"); + let url = super::forward::peer_url(&cand.http_addr, &path); let mut req = client.get(&url); if let Some(key) = api_key { req = req.bearer_auth(key); diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index 63d8bfb..aa1cb56 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -302,9 +302,16 @@ async fn run_region_cluster(args: ClusterArgs, region: String) -> Result<()> { // `run_boot_install` returns immediately (NotNeeded) when no marker // exists, after performing the always-safe swap-recovery. if let Some(ref dir) = data_dir { - let outcome = - tidal_server::cluster::reseed::run_boot_install(&topology, ®ion, dir)?; - tracing::info!(?outcome, region = %region, "boot-time reseed install evaluated"); + // m11p6+: each hosted shard group has its OWN data subdir, WAL, and + // reseed marker. Run the install ONCE PER hosted group against that + // group's subdir with shard-targeted leader discovery (the parent + // dir holds no per-group marker in S>1, which silently skipped the + // install → the term-0 divergence reseed loop). S==1 resolves to the + // parent dir verbatim. + let outcomes = tidal_server::cluster::reseed::run_boot_install_for_region( + &topology, ®ion, dir, + )?; + tracing::info!(?outcomes, region = %region, "boot-time reseed install evaluated per hosted shard group"); } // m11p6: `ClusterNode::new` resolves the shard-group assignment and // opens one `ShardReplica` per group this node hosts (S=1 is the diff --git a/tidal-server/src/offload.rs b/tidal-server/src/offload.rs index 0d99aba..ec98ea0 100644 --- a/tidal-server/src/offload.rs +++ b/tidal-server/src/offload.rs @@ -26,11 +26,34 @@ //! Both surfaces route through this module so the standalone and cluster paths //! cannot drift in how they treat blocking work. +use std::sync::OnceLock; + use crossbeam::channel::{Receiver, Sender, TrySendError}; use tidaldb::TidalError; +use tokio::sync::Semaphore; use crate::error::{Result, ServerError}; +/// Retry-after hint (ms) returned when the read-admission gate is saturated. Short +/// by design (mirrors the write pool's `WRITE_BACKPRESSURE_RETRY_AFTER_MS`): a +/// blocking read drains in worker-thread time, so a quick retry once a slot frees +/// beats parking the request. +const READ_BACKPRESSURE_RETRY_AFTER_MS: u64 = 50; + +/// Process-wide cap on CONCURRENT blocking reads, so a read storm sheds as a fast +/// 429 instead of piling unboundedly onto tokio's shared 512-thread blocking pool +/// (where it would also starve the leader's WAL-serve `spawn_blocking`) and +/// climbing into a multi-second p99. Lazily sized from the core count with +/// generous headroom, capped well below the blocking-pool limit. +static READ_GATE: OnceLock = OnceLock::new(); + +fn read_inflight_limit() -> usize { + std::thread::available_parallelism() + .map_or(8, std::num::NonZeroUsize::get) + .saturating_mul(16) + .clamp(16, 256) +} + /// Run a blocking READ-only query (RETRIEVE / SEARCH / text-index reload) on /// tokio's blocking pool, off the async reactor, and await its result. /// @@ -51,11 +74,35 @@ where F: FnOnce() -> Result + Send + 'static, T: Send + 'static, { + // m12p6 read admission: bound concurrent blocking reads. A healthy burst waits + // briefly for a slot (buffer-then-shed, matching the ClusterWritePool + // contract); a sustained read overload sheds as engine-native backpressure + // (429) instead of an unbounded climb into a 36s p99 on the shared blocking + // pool. A genuinely-closed gate never happens (the static lives for the + // process), so the closed arm is a defensive 500. + let gate = READ_GATE.get_or_init(|| Semaphore::new(read_inflight_limit())); + let _permit = match tokio::time::timeout( + std::time::Duration::from_millis(READ_BACKPRESSURE_RETRY_AFTER_MS), + gate.acquire(), + ) + .await + { + Ok(Ok(permit)) => permit, + Ok(Err(_closed)) => { + return Err(ServerError::Cluster("read admission gate closed".into())); + } + Err(_elapsed) => { + return Err(ServerError::Tidal(TidalError::Backpressure { + retry_after_ms: READ_BACKPRESSURE_RETRY_AFTER_MS, + })); + } + }; tokio::task::spawn_blocking(f) .await // A JoinError means the blocking task panicked or was cancelled — the // query produced no answer, so surface it as a server error (500). .map_err(|e| ServerError::Cluster(format!("blocking read worker failed: {e}")))? + // `_permit` drops here, releasing the read slot. } /// Configuration for the cluster write worker pool. diff --git a/tidal-server/tests/cluster_reseed.rs b/tidal-server/tests/cluster_reseed.rs index 02a5368..c954131 100644 --- a/tidal-server/tests/cluster_reseed.rs +++ b/tidal-server/tests/cluster_reseed.rs @@ -86,10 +86,15 @@ const FAST_ELECTION_YAML: &str = "election:\n heartbeat_interval_ms: 100\n ele /// follower's frontier (a single segment is never deleted; the snapshot path /// only triggers once history is genuinely compacted away). /// -/// Sizing: ~56 KiB of non-indexed metadata per item × 320 items ≈ 18 MiB of -/// metadata bytes (plus kind-1/2 blob overhead) > the 16 MiB segment size ⇒ -/// ≥ 2 segments ⇒ compaction deletes the earlier one(s). -const OFFLINE_ITEMS: u64 = 320; +/// Sizing (m12p6): the graceful-shutdown compaction now RETAINS the +/// `WAL_RETENTION_SEGMENTS` (= 4) most-recent sealed segments so a briefly-down +/// follower can stream-catch-up instead of reseeding. For this exit-gate to STILL +/// force the snapshot-required path, the offline batch must compact the follower's +/// frontier segment away DESPITE retention — i.e. produce more than +/// `WAL_RETENTION_SEGMENTS + 1` segments. At ~56 KiB/item the 16 MiB segment holds +/// ~292 items, so 1800 items ≈ 6 segments ⇒ the follower's frontier segment is +/// well past the 4-segment retention window and is genuinely deleted. +const OFFLINE_ITEMS: u64 = 1800; /// The engine caps item metadata at 8 KiB per VALUE and 64 KiB TOTAL per item /// (a hard query-index-integrity invariant — never to be weakened). So the WAL diff --git a/tidal-server/tests/reseed_install.rs b/tidal-server/tests/reseed_install.rs index 43ee8c8..5a9e004 100644 --- a/tidal-server/tests/reseed_install.rs +++ b/tidal-server/tests/reseed_install.rs @@ -223,6 +223,7 @@ fn boot_install_discovers_fetches_and_swaps() { &topology, "joiner", &data_dir, + ShardId(0), std::time::Duration::from_secs(8), ) .unwrap(); @@ -310,6 +311,7 @@ fn boot_install_needed_false_clears_marker_and_does_not_swap() { &topology, "joiner", &data_dir, + ShardId(0), std::time::Duration::from_secs(8), ) .unwrap(); @@ -360,6 +362,7 @@ fn boot_install_falls_back_when_no_leader_reachable() { &topology, "joiner", &data_dir, + ShardId(0), std::time::Duration::from_millis(600), ) .unwrap(); diff --git a/tidal-stress/k8s/recall-skipseed-tidaldb2-job.yaml b/tidal-stress/k8s/recall-skipseed-tidaldb2-job.yaml new file mode 100644 index 0000000..9ad840f --- /dev/null +++ b/tidal-stress/k8s/recall-skipseed-tidaldb2-job.yaml @@ -0,0 +1,95 @@ +# T-read recall gate — true p99 + recall@10 vs a brute-force cosine oracle. +# +# Seeds a 100k / 1536-dim corpus (regenerates the deterministic base vectors so +# it can build the ground-truth oracle locally), then ramps pure k-NN +# /vector_search probes against tidaldb-0 and reports recall@10 + read p99 per +# stage. Requires the M12 server (the /vector_search endpoint did not exist +# pre-m12; the m11p6 image 404s). +# +# Apply: kubectl apply -f tidal-stress/k8s/recall-verify-job.yaml +# Watch: kubectl logs -f job/tidal-recall-verify -n tidaldb-cluster +apiVersion: batch/v1 +kind: Job +metadata: + name: tidal-recall-t2 + namespace: tidaldb-cluster + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb +spec: + backoffLimit: 0 + ttlSecondsAfterFinished: 7200 + template: + metadata: + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb + spec: + restartPolicy: Never + automountServiceAccountToken: false + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + seccompProfile: + type: RuntimeDefault + containers: + - name: stress + image: registry.threesix.ai/tidal/stress:m12-rc7-seedretry + imagePullPolicy: IfNotPresent + args: + - --target + - https://tidaldb-2.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 + - --ca-cert + - /etc/tidaldb/tls/ca.crt + - --verify-recall + - --skip-seed + - --seed-concurrency + - "32" + - --corpus + - "100000" + - --embedding-dim + - "1536" + - --recall-k + - "10" + - --recall-queries + - "1000" + - --read-p99-target-ms + - "10" + - --recall-target + - "0.95" + - --recall-ef-search + - "64" + - --ramp + - "100:30,200:30,300:30,500:30" + env: + - name: TIDAL_API_KEY + valueFrom: + secretKeyRef: + name: tidaldb-credentials + key: TIDAL_API_KEY + - name: TIDAL_STRESS_LOG + value: warn + resources: + requests: + cpu: 500m + memory: 512Mi + limits: + cpu: "3" + memory: 2Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: cluster-tls + mountPath: /etc/tidaldb/tls + readOnly: true + volumes: + - name: cluster-tls + secret: + secretName: tidaldb-cluster-tls + items: + - key: ca.crt + path: ca.crt diff --git a/tidal-stress/k8s/recall-verify-rc7-job.yaml b/tidal-stress/k8s/recall-verify-rc7-job.yaml new file mode 100644 index 0000000..95ee5a3 --- /dev/null +++ b/tidal-stress/k8s/recall-verify-rc7-job.yaml @@ -0,0 +1,94 @@ +# T-read recall gate — true p99 + recall@10 vs a brute-force cosine oracle. +# +# Seeds a 100k / 1536-dim corpus (regenerates the deterministic base vectors so +# it can build the ground-truth oracle locally), then ramps pure k-NN +# /vector_search probes against tidaldb-0 and reports recall@10 + read p99 per +# stage. Requires the M12 server (the /vector_search endpoint did not exist +# pre-m12; the m11p6 image 404s). +# +# Apply: kubectl apply -f tidal-stress/k8s/recall-verify-job.yaml +# Watch: kubectl logs -f job/tidal-recall-verify -n tidaldb-cluster +apiVersion: batch/v1 +kind: Job +metadata: + name: tidal-recall-rc7 + namespace: tidaldb-cluster + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb +spec: + backoffLimit: 0 + ttlSecondsAfterFinished: 7200 + template: + metadata: + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb + spec: + restartPolicy: Never + automountServiceAccountToken: false + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + seccompProfile: + type: RuntimeDefault + containers: + - name: stress + image: registry.threesix.ai/tidal/stress:m12-rc7-seedretry + imagePullPolicy: IfNotPresent + args: + - --target + - https://tidaldb-0.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 + - --ca-cert + - /etc/tidaldb/tls/ca.crt + - --verify-recall + - --seed-concurrency + - "32" + - --corpus + - "100000" + - --embedding-dim + - "1536" + - --recall-k + - "10" + - --recall-queries + - "1000" + - --read-p99-target-ms + - "10" + - --recall-target + - "0.95" + - --recall-ef-search + - "64" + - --ramp + - "200:30,500:30,1000:30,2000:30" + env: + - name: TIDAL_API_KEY + valueFrom: + secretKeyRef: + name: tidaldb-credentials + key: TIDAL_API_KEY + - name: TIDAL_STRESS_LOG + value: warn + resources: + requests: + cpu: 500m + memory: 512Mi + limits: + cpu: "3" + memory: 2Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: cluster-tls + mountPath: /etc/tidaldb/tls + readOnly: true + volumes: + - name: cluster-tls + secret: + secretName: tidaldb-cluster-tls + items: + - key: ca.crt + path: ca.crt diff --git a/tidal-stress/k8s/t4-load-job.yaml b/tidal-stress/k8s/t4-load-job.yaml index 82fdac1..6cc1717 100644 --- a/tidal-stress/k8s/t4-load-job.yaml +++ b/tidal-stress/k8s/t4-load-job.yaml @@ -42,7 +42,7 @@ spec: type: RuntimeDefault containers: - name: stress - image: tidaldb-stress:m12p4-local + image: registry.threesix.ai/tidal/stress@sha256:4b21c1b89790f7a995f4d9e754fe0faf530079d7ad67a5ad74c15377dcc18ac2 # m12 recall/stress harness (registry) imagePullPolicy: IfNotPresent args: - --target @@ -57,7 +57,7 @@ spec: - quorum - --skip-seed - --corpus - - "300" + - "100000" - --embedding-dim - "1536" - --users diff --git a/tidal-stress/src/client.rs b/tidal-stress/src/client.rs index 61ed852..0d66ab3 100644 --- a/tidal-stress/src/client.rs +++ b/tidal-stress/src/client.rs @@ -449,17 +449,24 @@ const SEED_CATEGORIES: [&str; 12] = [ /// retries (the seed must not give up just because the write pool is busy); any /// other code / transport fault fails fast (a usage bug, not transient load). async fn retry_write(client: &HttpClient, url: &str, body: &Body) -> bool { - const MAX_ATTEMPTS: u32 = 8; + // A bulk SEED must persist through sustained write-pool saturation: when one + // leader owns several shards, a fast multi-writer seed keeps its bounded + // admission pool full and it sheds 429 for seconds at a stretch. The old + // 8-attempt / ~900ms budget gave up inside that window and silently DROPPED + // ~0.4% of the corpus, failing the recall harness's exact-count gate. 429 is + // backpressure, not failure — retry it persistently (exponential backoff + // capped at 500ms → ~12s total budget, comfortably past a saturation burst). + const MAX_ATTEMPTS: u32 = 40; for attempt in 0..MAX_ATTEMPTS { match client.post_json(url, body).await { Some(code) if (200..=299).contains(&code) => return true, Some(429) => { - // Honor the engine's ~50ms hint, with a little growth. - tokio::time::sleep(Duration::from_millis(40 + u64::from(attempt) * 20)).await; + let backoff = std::cmp::min(40 + u64::from(attempt) * 40, 500); + tokio::time::sleep(Duration::from_millis(backoff)).await; } Some(503) => { // Leader briefly unreachable (e.g. a roll) — short wait, retry. - tokio::time::sleep(Duration::from_millis(100)).await; + tokio::time::sleep(Duration::from_millis(150)).await; } _ => return false, } diff --git a/tidal/src/db/lifecycle.rs b/tidal/src/db/lifecycle.rs index 8a005f2..f1c3793 100644 --- a/tidal/src/db/lifecycle.rs +++ b/tidal/src/db/lifecycle.rs @@ -38,6 +38,31 @@ impl TidalDb { self.close() } + /// Run the deterministic shutdown — the final checkpoints (signal ledger, + /// secondary ledgers, replication HWM, **the m12p6 HNSW-graph save**) and the + /// WAL flush/marker — on a SHARED reference, WITHOUT consuming `self`. + /// + /// [`close`]/[`shutdown`] consume `self`, so a caller holding only an + /// `Arc` (the cluster node shutdown path, where request-scoped clones + /// may still be alive) cannot reach them. Its only other option is to drop the + /// handle and rely on `Drop` firing at refcount 0 — but under load a lingering + /// clone keeps the `Arc` alive, `Drop` never runs, and the HNSW graph is never + /// persisted (the observed m12p6 cluster gap: a follower restart rebuilt the + /// graph because no `{data_dir}/vector` file was written on the prior SIGTERM). + /// + /// This runs the identical body on `&self`. It is idempotent — the `closed` + /// CAS in [`shutdown_inner`](Self::shutdown_inner) guards the body — so the + /// eventual `Drop` (when the last clone finally releases) is a harmless no-op. + /// Callers MUST have quiesced the write/apply path first (the cluster node + /// stops the ship queue and signals the segment receiver before calling this). + /// + /// # Errors + /// + /// Surfaces the first durable-flush failure, exactly like [`close`](Self::close). + pub fn close_shared(&self) -> crate::Result<()> { + self.shutdown_inner() + } + /// Internal shutdown logic shared by `close()` and `Drop`. #[allow(clippy::too_many_lines)] pub(crate) fn shutdown_inner(&self) -> crate::Result<()> { @@ -264,7 +289,7 @@ impl TidalDb { // redundantly on next open. if seq > 0 && let Some(wal_dir) = self.config.resolved_wal_dir() - && let Err(e) = crate::wal::compaction::compact_wal(&wal_dir, seq) + && let Err(e) = crate::wal::compaction::compact_wal_retained(&wal_dir, seq) { tracing::warn!(error = %e, "WAL compaction failed during shutdown"); } diff --git a/tidal/src/wal/compaction.rs b/tidal/src/wal/compaction.rs index 43598b3..4b2f4cd 100644 --- a/tidal/src/wal/compaction.rs +++ b/tidal/src/wal/compaction.rs @@ -67,6 +67,60 @@ pub fn compact_wal(wal_dir: &Path, checkpoint_seq: u64) -> Result u64 { + if segments.len() > WAL_RETENTION_SEGMENTS { + let keep_idx = segments.len() - WAL_RETENTION_SEGMENTS; + floor.min(segments[keep_idx].0) + } else { + floor + } +} + +/// Offline (writer-joined) compaction with segment retention for stream catch-up. +/// +/// Retains the `WAL_RETENTION_SEGMENTS` most-recent sealed segments past the +/// checkpoint — the clean-shutdown analogue of the online retention in +/// [`compact_wal_online_pinned`], so a rolling restart keeps stream-catch-up +/// history instead of forcing returning followers to reseed. +/// +/// No live-segment clamp is needed (the caller has joined the writer via +/// `WalHandle::shutdown()` before calling this); the retention clamp is identical +/// to the online path. Use [`compact_wal`] only where catch-up history is not +/// wanted (it has no production caller after m12p6). +/// +/// # Errors +/// +/// Returns `WalError::Io` on filesystem failure. +pub fn compact_wal_retained( + wal_dir: &Path, + checkpoint_seq: u64, +) -> Result { + let segments = segment::list_segments(wal_dir)?; + let floor = retention_clamp(checkpoint_seq, &segments); + compact_segments(wal_dir, &segments, floor, None) +} + /// Online-safe compaction: identical to [`compact_wal`] except it NEVER deletes /// the segment the WAL writer thread is (or may be) actively appending to. /// @@ -146,6 +200,12 @@ pub fn compact_wal_online_pinned( None => checkpoint_seq, }; + // m12p6 recent-history retention: keep the N most-recent sealed segments past + // the checkpoint so a briefly-down follower stream-catches-up instead of + // reseeding. Only lowers the floor (keeps more), composing monotonically with + // the live-segment clamp above and the pin clamp below. + floor = retention_clamp(floor, &segments); + // Retention pin: preserve the segment chain covering `pin + 1` onward. if pin > 0 { // `pin == u64::MAX` would overflow `pin + 1`; in that (never-in-practice)