From c8ea05b0320b2e95f286a3826f2d90a333ba58a7 Mon Sep 17 00:00:00 2001 From: jx12n Date: Thu, 18 Jun 2026 21:06:18 -0600 Subject: [PATCH] fix(m12): break the post-reseed false-ReseedRequired loop (durable term marker + readiness gating + restart coordinator) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: after a checkpoint-based snapshot install the engine WAL is empty, so wal_term_mark() reports tail_term=0. decide_join compares (tail_term, frontier) lexicographically — tail_term FIRST — so 0 < leader_term classifies the reseeded shard ReseedRequired on EVERY boot regardless of the correctly-seeded frontier, re-latching the marker and self-restarting forever. Observed live on tidaldb-2: 30 CrashLoopBackOff restarts, leader tidaldb-1 term 5, baseline=536647, the frontier seeded correctly (from_seqno=536647) yet the loop persists because the (tail_term, frontier) compare never reaches the frontier. Fix 1 (already in tree): seed the post-open frontier from sentinel.snapshot_seq, not last_wal_seq() (which a checkpoint restore leaves at 0). Fix 2 (loop-breaker): durably synthesize the artifact's kind-3 TERM_MARKER WAL record in the post-open reseed seed, at the artifact's captured term + the reseed-leader region (threaded through an extended 18-byte install sentinel, back-compat with 10/8-byte). Makes wal_term_mark() truthful on this boot AND every reboot (blob records are NOT checkpoint-filtered on recovery), so decide_join returns Clean. Truthful, not a bypass: the artifact IS the leader's authoritative state at (term, seq); a genuinely-divergent node (no install sentinel) still surfaces tail_term > term -> Quarantine. Crash-idempotent via a monotonic-by-term guard. Fix 3: node-level reseed-restart coordinator — the single process-wide exit fires once, only after every hosted shard requests a restart or a bounded grace elapses, so one shard's self-restart never aborts a co-hosted sibling's in-flight install (S>1). No-op on the S=1 production topology. Fix 4: is_ready() returns 503 while any reseed marker (SnapshotRequired or Quarantine) is latched, closing the plain-restart serve-while-behind gap; readiness is bounded staleness, not "ready the instant the process is up". Tests: decide_join loop/fix/bounded-reseed unit; install-sentinel 18-byte round-trip + back-compat; engine durability (term marker survives a checkpoint advanced past it + crash-reopen); reseed-restart gate (5 cases); reseed_install carries the term. Verified: cluster_reseed 4/4 (zero-loss rolling restart x2, quarantine reseed, failover oracle), reseed_install 3/3, m12_reseed_term_marker 4/4, cluster_membership mp_idle/mp_dns/mp_remove x2, tidal-server lib 154/154. mp_scale_3_5_3 and mp_seed_join_snapshot_catchup OOM on this host (22GB colima VM); their /health/startup failure is process-down, not the is_ready path Fix 4 touches. --- tidal-server/src/cluster/election_driver.rs | 63 +++++ tidal-server/src/cluster/mod.rs | 5 + tidal-server/src/cluster/node.rs | 202 ++++++++++++++-- tidal-server/src/cluster/reseed.rs | 108 +++++++-- tidal-server/src/cluster/reseed_restart.rs | 251 ++++++++++++++++++++ tidal-server/tests/reseed_install.rs | 11 +- tidal/Cargo.toml | 4 + tidal/tests/m12_reseed_term_marker.rs | 207 ++++++++++++++++ 8 files changed, 799 insertions(+), 52 deletions(-) create mode 100644 tidal-server/src/cluster/reseed_restart.rs create mode 100644 tidal/tests/m12_reseed_term_marker.rs diff --git a/tidal-server/src/cluster/election_driver.rs b/tidal-server/src/cluster/election_driver.rs index a06ab1b..42d37f4 100644 --- a/tidal-server/src/cluster/election_driver.rs +++ b/tidal-server/src/cluster/election_driver.rs @@ -1018,4 +1018,67 @@ mod tests { "within-term rejoin wins over the < comparison too" ); } + + /// m12 reseed-loop-fix: the durable term marker is what stops the + /// false-reseed loop a checkpoint-restored (empty-WAL) node hits. + /// + /// `election_log_position` builds `own` from `wal_term_mark()`. Before the + /// fix the restored WAL is empty so `tail_term = 0`, and because + /// `decide_join` compares `(tail_term, frontier)` with `tail_term` FIRST, + /// `0 < leader_term` classifies the node `ReseedRequired` on EVERY boot — the + /// correctly-seeded frontier is never even consulted. The seed now durably + /// synthesizes the artifact's term marker, so `own.tail_term` is truthful and + /// the within-term clause classifies the reseeded shard `Clean`. + #[test] + fn reseed_term_marker_breaks_the_false_reseed_loop() { + let leader_term = 7; + let snapshot_seq = 1_040_000; // the seeded frontier (Fix 1) + let baseline = 1_040_000; // the leader's election-time frontier + + // THE BUG: a checkpoint-restored node reports tail_term 0. Even with the + // frontier correctly seeded to `snapshot_seq`, the term-0 tail loses the + // lexicographic compare → ReseedRequired → re-latch → loop. (A follower + // never ack=leader-writes, so leader_acked = 0.) + assert_eq!( + decide_join( + leader_term, + pos(0, snapshot_seq), + pos(leader_term, baseline), + 0 + ), + JoinDecision::ReseedRequired, + "tail_term 0 after a checkpoint restore loses the (term, frontier) compare → \ + the false reseed loop" + ); + + // THE FIX: the durable term marker makes `own.tail_term == leader_term`, + // so the within-term clause returns Clean regardless of the frontier — the + // loop is broken. The catch-up pull streams any residual tail. + assert_eq!( + decide_join( + leader_term, + pos(leader_term, snapshot_seq), + pos(leader_term, baseline), + 0 + ), + JoinDecision::Clean, + "a truthful artifact term marker classifies the reseeded shard Clean" + ); + + // Even if the leadership advanced a term between fetch and the join check, + // the node reseeds ONCE MORE against the newer term (own.tail_term < + // newer_term, nothing leader-acked) — bounded, self-healing, NOT a loop: + // the next install stamps the newer term and converges. + let newer_term = leader_term + 1; + assert_eq!( + decide_join( + newer_term, + pos(leader_term, snapshot_seq), + pos(newer_term, baseline), + 0 + ), + JoinDecision::ReseedRequired, + "a leadership that advanced a term reseeds once more, then converges — not a loop" + ); + } } diff --git a/tidal-server/src/cluster/mod.rs b/tidal-server/src/cluster/mod.rs index b4f48e1..6df7dfd 100644 --- a/tidal-server/src/cluster/mod.rs +++ b/tidal-server/src/cluster/mod.rs @@ -63,6 +63,11 @@ pub(crate) mod node; /// Public so the binary's `run_region_cluster` can invoke `run_boot_install` /// before `ShardReplica::new`. pub mod reseed; +/// Node-level reseed self-restart coordinator (m12 reseed-loop-fix, Fix 3): the +/// single process-wide exit fires once, only after every hosted shard has +/// requested a restart or a bounded grace elapsed — so one shard's self-restart +/// never aborts a co-hosted sibling's in-flight install (S>1). +pub(crate) mod reseed_restart; pub(crate) mod routes; /// m11p7 cluster security: reloadable bearer + cluster keys, per-node signed /// internal tokens, and the request principal. diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 18b617b..8f79042 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -405,6 +405,18 @@ 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, /// 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. @@ -762,10 +774,10 @@ impl ShardReplica { // leader, which for a reseeded ex-leader is NOT the boot topology // leader — that is the node itself). BEFORE the receiver, transport // serving, or any pull, seed the DISCOVERED-leader-shard frontier to the - // installed artifact's recovered WAL tail (the engine's `last_wal_seq` - // after open IS the artifact's recovered tail at this instant — own-WAL - // numbering diverges from stream numbering as soon as applies begin), - // persist the replication-state checkpoint SYNCHRONOUSLY, then delete + // installed artifact's captured stream seq (the sentinel's `snapshot_seq`; + // NOT `last_wal_seq()`, which a checkpoint-based restore leaves at 0 — see + // the seed below), persist the replication-state checkpoint SYNCHRONOUSLY, + // then delete // the sentinel. A crash anywhere before the sentinel delete finds the // WAL unchanged (no receiver ran) and re-derives the same seed // idempotently. The discovered shard + tail are carried forward to issue @@ -779,8 +791,66 @@ impl ShardReplica { let leader_shard = sentinel .leader_region .map_or_else(|| shard_of_region(leader), |r| shard_of_region(RegionId(r))); - let recovered_tail = db.last_wal_seq(); + // The artifact's captured STREAM seq, recorded authoritatively in the + // install sentinel, IS the recovered frontier. Do NOT derive it from + // `db.last_wal_seq()`: a checkpoint-based restore leaves the engine's + // own WAL EMPTY (the data lives in the restored checkpoint/keyspaces, + // not a replayed WAL), so `last_wal_seq()` reads 0 — which seeds frontier + // 0, makes the post-install pull request from seqno 1, which the leader's + // long-compacted WAL cannot serve → `snapshot_required` re-latch → reseed + // loop. (Observed in prod on an 805 MB / seq-1.04M artifact whose WAL + // restored empty; small-corpus e2e artifacts restore a non-empty WAL so + // they never tripped it.) `snapshot_seq` is the leader's stream position + // the artifact represents — exactly what the catch-up pull below + // (`recovered_tail + 1`) must resume from. + let recovered_tail = sentinel.snapshot_seq; db.replication_state().advance(leader_shard, recovered_tail); + // m12 reseed-loop-fix: durably synthesize the term marker the + // installed artifact represents. A checkpoint-based restore leaves + // the engine's own WAL EMPTY (the data lives in the restored + // keyspaces, not a replayed WAL), so `wal_term_mark()` reports + // tail_term 0 — and `decide_join` compares `(tail_term, frontier)` + // lexically, tail_term FIRST, so `0 < leader_term` classifies this + // shard `ReseedRequired` on EVERY boot regardless of the correctly- + // seeded frontier → an infinite reseed loop (observed on tidaldb-2, + // the 805 MB / seq-1.04M artifact whose WAL restored empty). Writing + // a real kind-3 TERM_MARKER record at the artifact's captured term + + // the reseed-leader's region makes tail_term truthful on this boot + // AND every reboot (blob records are deliberately NOT checkpoint- + // filtered on recovery — wal/reader.rs), so `election_log_position` + // reads the frontier in the DISCOVERED-leader stream (the shard just + // seeded) and `decide_join` returns Clean. Truthful, not a bypass: + // the artifact IS the leader's authoritative state at (term, seq) and + // the install discarded any prior suffix, so a genuinely-divergent + // node (which has NO install sentinel) still surfaces tail_term > + // term → Quarantine. Crash-idempotent: once a prior attempt's marker + // is folded by recovery the guard (`< artifact_term`, monotonic-by- + // term) skips the re-append. Skipped for a legacy sentinel that + // carries no term (cannot fabricate one) and for the term-0 topology + // era (which never journals a marker). + if let (Some(artifact_term), Some(marker_region)) = + (sentinel.artifact_term, sentinel.leader_region) + && artifact_term > 0 + && db.wal_term_mark().0 < artifact_term + { + db.append_term_marker(artifact_term, marker_region) + .map_err(|e| { + ServerError::Cluster(format!( + "reseed post-open seed: synthesize term marker (term \ + {artifact_term}, region {marker_region}) failed: {e}" + )) + })?; + tracing::info!( + region = region_name, + artifact_term, + marker_region, + recovered_tail, + "reseed install boot: durably synthesized the artifact's term marker so \ + wal_term_mark reports its term (not 0) on this and every reboot — \ + decide_join now classifies the reseeded shard on a truthful (term, \ + frontier) instead of looping ReseedRequired (§2.6 m12 reseed-loop-fix)" + ); + } db.persist_replication_checkpoint().map_err(|e| { ServerError::Cluster(format!( "reseed post-open seed: persist replication checkpoint failed: {e}" @@ -793,8 +863,9 @@ impl ShardReplica { recovered_tail, discovered_leader = sentinel.leader_region, "reseed install boot: seeded the DISCOVERED-leader-shard frontier to the \ - artifact's recovered WAL tail and persisted the checkpoint (§2.6); a \ - catch-up pull toward that shard resumes the stream after the receiver starts" + artifact's captured stream seq (sentinel snapshot_seq) and persisted the \ + checkpoint (§2.6); a catch-up pull toward that shard resumes the stream \ + after the receiver starts" ); Some((leader_shard, recovered_tail)) } @@ -1047,6 +1118,14 @@ impl ShardReplica { } } + // m12 reseed-loop-fix (readiness gating): pre-load the durable reseed + // 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. + let reseed_marker_store = ReseedMarkerStore::new(&data_dir_for_state); + let reseed_marker_latched_init = matches!(reseed_marker_store.load(), Ok(Some(_))); + Ok(Self { region, region_name: region_name.to_string(), @@ -1084,7 +1163,8 @@ impl ShardReplica { ack_default, quorum_timeout, shutting_down: AtomicBool::new(false), - reseed_marker_store: ReseedMarkerStore::new(&data_dir_for_state), + reseed_marker_store, + reseed_marker_latched: AtomicBool::new(reseed_marker_latched_init), reseed_self_restart: topology.replication.reseed_self_restart.unwrap_or(false), learner_promote_lag: topology.replication.learner_promote_lag.unwrap_or(1024), seed_joiner, @@ -2922,6 +3002,12 @@ impl ShardReplica { /// a transient fsync failure does not strand the node (the next refusal /// re-latches). 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(()) => { @@ -2985,6 +3071,9 @@ impl ShardReplica { 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, "stale reseed marker cleared — caught up to the leader via the stream \ @@ -3027,7 +3116,7 @@ impl ShardReplica { /// set in status) — exiting during a 2-voter window is total write /// unavailability. Otherwise it flips readiness to 503 and exits via the /// existing graceful-shutdown path (NOT `process::abort`). - fn maybe_self_restart(&self) { + fn maybe_self_restart(self: &Arc) { // Poll the OTHER regions' local status synchronously on a throwaway // blocking client (this runs on the write pool, never the reactor). // The voter count from the effective roster (§3): era 0 = the full @@ -3050,19 +3139,65 @@ impl ShardReplica { return; } self.self_restart_refused.store(false, Ordering::Release); - tracing::warn!( - region = %self.region_name, - alive_voters_excluding_self = alive_others, - total_voters, - "reseed_self_restart: the remaining voters sustain quorum without this node — \ - draining and exiting cleanly so the next boot reseeds via snapshot (§2.4). \ - Readiness flips to 503 before drain; this is the graceful-shutdown path, never abort." - ); - // Flip readiness to 503 so the load balancer drains us, then exit(0) on - // a detached thread so the graceful HTTP drain + checkpoint run. We use - // the existing shutdown signal path (SIGTERM-equivalent) rather than a - // hard abort: send ourselves SIGTERM so `serve_state`'s graceful - // shutdown fires (readiness 503 → drain → checkpoint → exit). + // m12 reseed-loop-fix (Fix 3): the process-wide exit is owned by the + // node-level coordinator, not this single shard. It fires once, only + // after every hosted shard has also requested a restart or a bounded + // grace has elapsed — so this shard's exit never aborts a co-hosted + // sibling's in-flight install. A single-shard host (production) fires + // immediately (FireNow), so this is a behavioral no-op there. + match super::reseed_restart::request_restart(self.group_shard) { + super::reseed_restart::Decision::FireNow => { + tracing::warn!( + region = %self.region_name, + alive_voters_excluding_self = alive_others, + total_voters, + "reseed_self_restart: quorum is safe and every hosted shard has settled — \ + draining and exiting cleanly so the next boot reseeds via snapshot (§2.4). \ + Readiness flips to 503 before drain; the graceful-shutdown path, never abort." + ); + self.fire_graceful_self_restart(); + } + super::reseed_restart::Decision::Defer { deadline } => { + tracing::warn!( + region = %self.region_name, + grace_secs = super::reseed_restart::grace().as_secs(), + "reseed_self_restart: quorum is safe, but deferring the process-wide exit so a \ + co-hosted sibling shard's in-flight install can finish first (Fix 3); the exit \ + fires once all hosted shards settle or the grace elapses." + ); + let node = Arc::clone(self); + if let Err(e) = std::thread::Builder::new() + .name("tidal-reseed-restart-defer".into()) + .spawn(move || { + let now = Instant::now(); + if deadline > now { + std::thread::sleep(deadline - now); + } + if super::reseed_restart::fire_due() { + node.fire_graceful_self_restart(); + } + }) + { + tracing::error!(error = %e, "failed to spawn reseed self-restart defer thread"); + } + } + super::reseed_restart::Decision::AlreadyFired => { + tracing::info!( + region = %self.region_name, + "reseed_self_restart: a co-hosted shard already triggered the process-wide \ + exit; this shard's marker stays latched and rides the same graceful restart." + ); + } + } + } + + /// Flip readiness to 503 and trigger the graceful self-exit (drain → + /// checkpoint → clean exit(0)), after a brief grace so an in-flight admin + /// reseed response flushes. The SINGLE firing path for both the immediate + /// (`FireNow`) and the deferred (Fix 3 grace-deadline) restart decisions. We + /// use the existing shutdown-signal path (SIGTERM-equivalent) so + /// `serve_state`'s graceful shutdown fires — never `process::abort`. + fn fire_graceful_self_restart(&self) { self.set_shutting_down(); if let Err(e) = std::thread::Builder::new() .name("tidal-reseed-restart".into()) @@ -3225,6 +3360,16 @@ impl ShardReplica { if (self.install_boot || self.seed_joiner) && !self.converged.load(Ordering::Acquire) { return false; } + // m12 reseed-loop-fix (readiness gating): an unhealed reseed marker + // (snapshot-required or quarantine) means this node holds stale data it is + // about to discard — drain it from the client VIP until it heals. This + // closes the plain-restart (install_boot == false) gap: a PVC-retained + // voter that re-latched snapshot-required while merely behind used to keep + // serving stale reads. The latch clears when the node catches up via the + // stream (clear_stale_reseed_marker_if_caught_up) or reseeds next boot. + if self.reseed_marker_latched.load(Ordering::Acquire) { + return false; + } true } @@ -3965,6 +4110,12 @@ fn build_forwarding_clients( /// `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)` is the deliberately-narrow, intent-expressing visibility for a +// crate-internal durability contract; the nursery `redundant_pub_crate` lint +// only flags it because the enclosing `node` module is private (so pub(crate) +// and pub reach identically) — keep the narrower, documented visibility. +#[allow(clippy::redundant_pub_crate)] pub(crate) fn shard_subdir(shard: ShardId) -> String { format!("shard-{:05}", shard.0) } @@ -4122,6 +4273,13 @@ impl ClusterNode { "region '{region_name}' is not a replica of any shard group" ))); } + // m12 reseed-loop-fix (Fix 3): tell the node-level reseed-restart + // coordinator how many shard groups this process hosts, so a self-restart + // from one group fires the single process-wide exit only after every + // hosted group has also requested it or a bounded grace elapsed — never + // aborting a co-hosted sibling's in-flight install. A no-op on the S=1 + // topology (one group → the gate fires immediately, today's behavior). + super::reseed_restart::register_hosted(groups.len()); // m11p8: when multiple shard groups co-locate on this node, the metrics // owner exposes the siblings' `tidaldb_cluster_*` series under their own // `shard="N"` label so one `/metrics` scrape covers every hosted group. diff --git a/tidal-server/src/cluster/reseed.rs b/tidal-server/src/cluster/reseed.rs index d1186ed..725bb82 100644 --- a/tidal-server/src/cluster/reseed.rs +++ b/tidal-server/src/cluster/reseed.rs @@ -854,10 +854,11 @@ fn attempt_install( Ok(FetchResult::Needed { snapshot_seq }) => { // Copy the node's own identity files INTO staging (§2.3 step 2), // then the install sentinel (carrying the DISCOVERED leader's region - // so the post-open seed + catch-up pull target its shard, not the - // boot topology leader's), then COMPLETE, fsync, swap. + // AND term so the post-open seed + catch-up pull target its shard — + // not the boot topology leader's — and durably synthesize the term + // marker), then COMPLETE, fsync, swap. copy_identity_into_staging(&dirs.data_dir, &dirs.staging)?; - write_install_sentinel(&dirs.staging, snapshot_seq, leader.region)?; + write_install_sentinel(&dirs.staging, snapshot_seq, leader.region, leader.term)?; write_complete_sentinel(&dirs.staging)?; complete_swap(dirs)?; tracing::info!( @@ -1080,13 +1081,15 @@ fn copy_identity_into_staging(data_dir: &Path, staging: &Path) -> Result<()> { Ok(()) } -/// The install sentinel's body (m11p5 §2.6). +/// The install sentinel's body (m11p5 §2.6, extended m12 reseed-loop-fix). /// -/// It carries the artifact's recovered seq plus the DISCOVERED leader's region. -/// Both drive the post-open path: `ShardReplica::new` seeds the frontier -/// and issues the post-install catch-up pull against the discovered leader's -/// shard (NOT the boot topology leader, which for a reseeded ex-leader is the -/// node itself). +/// It carries the artifact's recovered seq, the DISCOVERED leader's region, and +/// the artifact's captured TERM. All three drive the post-open path: +/// `ShardReplica::new` seeds the frontier, durably synthesizes the term marker +/// the artifact represents (so `wal_term_mark()` reports the artifact's term — +/// not 0 — on this boot and every reboot), and issues the post-install catch-up +/// pull against the discovered leader's shard (NOT the boot topology leader, +/// which for a reseeded ex-leader is the node itself). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct InstallSentinel { /// The installed artifact's recovered WAL tail seq. @@ -1095,21 +1098,42 @@ pub struct InstallSentinel { /// when the sentinel predates this field (an interrupted older install): the /// caller then falls back to the boot topology leader's shard. pub leader_region: Option, + /// The artifact's captured leadership TERM (the discovered leader's term at + /// fetch time). `None` when the sentinel predates this field (a legacy 8- or + /// 10-byte install): the post-open seed then SKIPS the durable term-marker + /// synthesis (it cannot fabricate a term it does not know) and falls back to + /// the historical behavior. A truthful term ≥ 1 lets the seed make + /// `wal_term_mark()` report it, so `decide_join` classifies the reseeded + /// shard on a real `(term, frontier)` instead of the false `(0, …)` that + /// loops `ReseedRequired` after a checkpoint-restore empties the WAL. + pub artifact_term: Option, } -/// The install sentinel's on-disk size: 8 bytes seq (LE) + 2 bytes region (LE). -const INSTALL_SENTINEL_SIZE: usize = 8 + 2; +/// The install sentinel's on-disk size, 18 bytes: 8 bytes seq (LE), then 2 +/// bytes region (LE), then 8 bytes term (LE). A 10-byte sentinel (seq + region, +/// no term) and an 8-byte one (seq only) are still read for back-compat with an +/// install that was staged by an older binary and interrupted across the upgrade. +const INSTALL_SENTINEL_SIZE: usize = 8 + 2 + 8; +/// Legacy size: 8 bytes seq + 2 bytes region, no term. +const INSTALL_SENTINEL_SIZE_NO_TERM: usize = 8 + 2; /// Write the `reseed-install-pending` install sentinel INTO staging BEFORE the -/// swap (§2.6): it carries the artifact's recovered seq AND the discovered -/// leader's region, so the post-open seed advances the right shard's frontier +/// swap (§2.6): it carries the artifact's recovered seq, the discovered +/// leader's region, AND the artifact's captured term, so the post-open seed +/// advances the right shard's frontier, durably synthesizes the term marker, /// and the catch-up pull dials the discovered leader. Crash-idempotent: a crash /// before the seed's sentinel-delete re-runs the seed against the unchanged WAL. -fn write_install_sentinel(staging: &Path, snapshot_seq: u64, leader_region: u16) -> Result<()> { +fn write_install_sentinel( + staging: &Path, + snapshot_seq: u64, + leader_region: u16, + artifact_term: u64, +) -> Result<()> { let path = staging.join(INSTALL_PENDING_SENTINEL); let mut bytes = [0u8; INSTALL_SENTINEL_SIZE]; bytes[0..8].copy_from_slice(&snapshot_seq.to_le_bytes()); bytes[8..10].copy_from_slice(&leader_region.to_le_bytes()); + bytes[10..18].copy_from_slice(&artifact_term.to_le_bytes()); std::fs::write(&path, bytes).map_err(|e| { ServerError::Cluster(format!( "reseed boot: write install sentinel {} failed: {e}", @@ -1144,9 +1168,11 @@ fn write_complete_sentinel(staging: &Path) -> Result<()> { /// /// Called by `ShardReplica::new` after `open_region_db` to drive the §2.6 /// post-open seed AND the post-install catch-up pull. `None` ⇒ this was not an -/// install boot. A 10-byte sentinel carries the discovered leader's region; an -/// 8-byte one (an interrupted older install) is read with `leader_region: None` -/// so the caller falls back to the boot topology leader's shard. +/// install boot. An 18-byte sentinel carries the discovered leader's region AND +/// the artifact's captured term; a 10-byte one carries the region but no term +/// (`artifact_term: None`); an 8-byte one (an interrupted older install) carries +/// neither (`leader_region: None`, `artifact_term: None`) so the caller falls +/// back to the boot topology leader's shard and skips the term-marker synthesis. /// /// # Errors /// @@ -1157,30 +1183,40 @@ pub fn read_install_sentinel(data_dir: &Path) -> Result> let path = data_dir.join(INSTALL_PENDING_SENTINEL); match std::fs::read(&path) { Ok(bytes) => { - // Both accepted lengths begin with the 8-byte seq; only the 10-byte - // form carries the trailing 2-byte region. No slice `expect` — every - // index here is bounds-checked by the length match (panic-free). + // All accepted lengths begin with the 8-byte seq; the 10-byte form + // adds the 2-byte region; the 18-byte form adds the 8-byte term. No + // slice `expect` — every index here is bounds-checked by the length + // match (panic-free). let len = bytes.len(); - if len != INSTALL_SENTINEL_SIZE && len != 8 { + if len != INSTALL_SENTINEL_SIZE && len != INSTALL_SENTINEL_SIZE_NO_TERM && len != 8 { return Err(ServerError::Cluster(format!( "reseed install sentinel {} is malformed ({len} bytes, expected \ - {INSTALL_SENTINEL_SIZE} or 8); the post-open seed cannot proceed", + {INSTALL_SENTINEL_SIZE}, {INSTALL_SENTINEL_SIZE_NO_TERM}, or 8); the \ + post-open seed cannot proceed", path.display() ))); } let mut seq_bytes = [0u8; 8]; seq_bytes.copy_from_slice(&bytes[0..8]); let snapshot_seq = u64::from_le_bytes(seq_bytes); - let leader_region = if len == INSTALL_SENTINEL_SIZE { + let leader_region = if len >= INSTALL_SENTINEL_SIZE_NO_TERM { let mut region_bytes = [0u8; 2]; region_bytes.copy_from_slice(&bytes[8..10]); Some(u16::from_le_bytes(region_bytes)) } else { None }; + let artifact_term = if len == INSTALL_SENTINEL_SIZE { + let mut term_bytes = [0u8; 8]; + term_bytes.copy_from_slice(&bytes[10..18]); + Some(u64::from_le_bytes(term_bytes)) + } else { + None + }; Ok(Some(InstallSentinel { snapshot_seq, leader_region, + artifact_term, })) } Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), @@ -1390,7 +1426,8 @@ mod tests { } /// The install + clear sentinel round-trips (carrying the discovered leader - /// region), reads an 8-byte legacy sentinel as `leader_region: None`, and + /// region AND the artifact term), reads a legacy 10-byte sentinel as + /// `artifact_term: None`, an 8-byte one as `leader_region: None` too, and /// refuses a malformed sentinel. #[test] fn install_sentinel_roundtrip_and_malformed() { @@ -1398,13 +1435,14 @@ mod tests { let data = tmp.path(); assert_eq!(read_install_sentinel(data).unwrap(), None); - // The 10-byte sentinel carries the discovered leader's region. - write_install_sentinel(data, 4096, 2).unwrap(); + // The 18-byte sentinel carries the discovered leader's region AND term. + write_install_sentinel(data, 4096, 2, 5).unwrap(); assert_eq!( read_install_sentinel(data).unwrap(), Some(InstallSentinel { snapshot_seq: 4096, leader_region: Some(2), + artifact_term: Some(5), }) ); @@ -1413,6 +1451,23 @@ mod tests { // clear is idempotent. clear_install_sentinel(data).unwrap(); + // A legacy 10-byte sentinel (region but no term — staged by an older + // binary and interrupted across the upgrade) reads with artifact_term + // None, so the seed skips the term-marker synthesis. + let mut ten = [0u8; INSTALL_SENTINEL_SIZE_NO_TERM]; + ten[0..8].copy_from_slice(&9u64.to_le_bytes()); + ten[8..10].copy_from_slice(&1u16.to_le_bytes()); + std::fs::write(data.join(INSTALL_PENDING_SENTINEL), ten).unwrap(); + assert_eq!( + read_install_sentinel(data).unwrap(), + Some(InstallSentinel { + snapshot_seq: 9, + leader_region: Some(1), + artifact_term: None, + }) + ); + clear_install_sentinel(data).unwrap(); + // A legacy 8-byte sentinel (an interrupted older install) reads with // leader_region None — the caller falls back to the boot topology leader. std::fs::write(data.join(INSTALL_PENDING_SENTINEL), 7u64.to_le_bytes()).unwrap(); @@ -1421,6 +1476,7 @@ mod tests { Some(InstallSentinel { snapshot_seq: 7, leader_region: None, + artifact_term: None, }) ); clear_install_sentinel(data).unwrap(); diff --git a/tidal-server/src/cluster/reseed_restart.rs b/tidal-server/src/cluster/reseed_restart.rs new file mode 100644 index 0000000..3ad30b1 --- /dev/null +++ b/tidal-server/src/cluster/reseed_restart.rs @@ -0,0 +1,251 @@ +//! Node-level reseed self-restart coordinator (m12 reseed-loop-fix, Fix 3). +//! +//! The per-shard `maybe_self_restart` (node.rs §2.4) used to call +//! `trigger_graceful_self_exit()` directly the instant ITS shard latched a +//! reseed marker and passed the §2.4 quorum-refusal check. In a process hosting +//! MULTIPLE shard groups (S>1) that is a hazard: one shard's restart SIGTERMs +//! the whole process while a SIBLING shard is still mid-install (streaming its +//! snapshot to convergence), aborting that install. With the durable term marker +//! (Fix 2) every shard re-converges on its own next boot regardless — so this is +//! never a correctness bug — but the process can thrash through several restarts +//! before all groups settle. +//! +//! This coordinator makes the single process-wide exit fire ONCE, only after +//! every hosted shard is settled: each hosted shard has either also requested a +//! restart, or a bounded grace deadline (measured from the FIRST request) has +//! elapsed — giving any in-flight sibling install time to finish first. The +//! grace is the only thing that bounds the wait, so a healthy sibling that never +//! requests a restart cannot wedge the exit forever. +//! +//! A node hosting a SINGLE shard group (the production topology — one region, +//! one shard per process) requests `1 == hosted`, so the gate fires immediately: +//! its one shard IS every hosted shard. This coordinator is therefore a +//! behavioral no-op there, and only changes behavior for an S>1 deployment. + +// This module is deliberately crate-internal (`pub(crate) mod`) and its small +// API is `pub(crate)` by intent; `with_gate` holds the gate mutex across the +// closure ON PURPOSE (the request/fire decision must be atomic over the state). +// Silence the nursery lints that flag those deliberate choices. +#![allow(clippy::redundant_pub_crate, clippy::significant_drop_tightening)] + +use std::collections::HashSet; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use tidaldb::replication::shard::ShardId; + +/// Grace window: once the FIRST hosted shard requests a self-restart, the +/// process waits up to this long for the others to settle (also request, or +/// finish an in-flight install) before exiting anyway. Short enough that a +/// genuinely-stuck cluster still self-heals promptly; long enough that a +/// co-hosted sibling's small catch-up finishes first. +const RESTART_GRACE: Duration = Duration::from_secs(5); + +/// The decision a restart request yields. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Decision { + /// Fire the graceful self-exit now: every hosted shard has requested a + /// restart (or this is the only hosted shard). + FireNow, + /// Defer: other hosted shards are not yet settled. The caller arms a + /// one-shot timer and calls [`GateState::fire_due`] at `deadline`; if the + /// gate has not already fired by then, the exit fires. + Defer { deadline: Instant }, + /// The exit was already triggered by an earlier request — nothing to do. + AlreadyFired, +} + +/// The pure coordinator state (no I/O, no timers — directly unit-testable). +#[derive(Debug)] +struct GateState { + /// Number of shard groups this process hosts (≥ 1). + hosted: usize, + /// The shards that have requested a self-restart. + wanting: HashSet, + /// When the FIRST request arrived — the grace deadline is measured from it. + first_request: Option, + /// Set once the exit has been triggered (guards the one-shot firing). + fired: bool, +} + +impl GateState { + fn new(hosted: usize) -> Self { + Self { + hosted: hosted.max(1), + wanting: HashSet::new(), + first_request: None, + fired: false, + } + } + + /// Record `shard`'s restart request and decide what the caller should do. + fn request(&mut self, shard: ShardId, now: Instant, grace: Duration) -> Decision { + if self.fired { + return Decision::AlreadyFired; + } + self.wanting.insert(shard); + let first = *self.first_request.get_or_insert(now); + // Every hosted shard has now asked to restart (or there is only one) — + // nothing is mid-install to protect, so fire immediately. + if self.wanting.len() >= self.hosted { + self.fired = true; + return Decision::FireNow; + } + Decision::Defer { + deadline: first + grace, + } + } + + /// Called when a deferred grace timer elapses. Fires the exit (returns + /// `true`) iff the grace has truly elapsed and nothing fired yet; idempotent + /// across the multiple timers concurrent deferrals may have armed. + fn fire_due(&mut self, now: Instant, grace: Duration) -> bool { + if self.fired { + return false; + } + let due = self + .first_request + .is_some_and(|first| now.duration_since(first) >= grace); + if due { + self.fired = true; + } + due + } +} + +/// The process-global gate. `Mutex>` (not `OnceLock`) so +/// [`register_hosted`] can (re)initialize it: a process has exactly one +/// `ClusterNode`, but in-process construction in tests may build several. +static GATE: Mutex> = Mutex::new(None); + +fn with_gate(f: impl FnOnce(&mut GateState) -> R) -> R { + let mut guard = GATE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + // Default to a single-shard gate if a request somehow precedes + // registration (groups are built before the election driver starts, so + // this is defensive): a single-shard gate fires immediately — the safe, + // today's-behavior default. + let state = guard.get_or_insert_with(|| GateState::new(1)); + f(state) +} + +/// Register how many shard groups this process hosts (called once by +/// `ClusterNode::new` after every group is built). Resets the gate's request +/// state — registration happens at boot, before any restart can be requested. +pub(crate) fn register_hosted(hosted: usize) { + let mut guard = GATE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *guard = Some(GateState::new(hosted)); +} + +/// Record `shard`'s graceful-self-restart request and decide what to do (uses +/// the live clock + the standing grace window). See [`Decision`]. +pub(crate) fn request_restart(shard: ShardId) -> Decision { + with_gate(|g| g.request(shard, Instant::now(), RESTART_GRACE)) +} + +/// Resolve a deferred request at its grace deadline: returns `true` iff THIS +/// call should fire the graceful self-exit (idempotent one-shot). +pub(crate) fn fire_due() -> bool { + with_gate(|g| g.fire_due(Instant::now(), RESTART_GRACE)) +} + +/// The grace window, for the caller's deferred timer to sleep against. +pub(crate) const fn grace() -> Duration { + RESTART_GRACE +} + +#[cfg(test)] +mod tests { + use super::*; + + fn t0() -> Instant { + Instant::now() + } + + /// A single-shard host (the production topology) fires immediately: its one + /// shard is every hosted shard, so there is nothing mid-install to protect. + #[test] + fn single_shard_fires_immediately() { + let mut g = GateState::new(1); + assert_eq!( + g.request(ShardId(0), t0(), RESTART_GRACE), + Decision::FireNow + ); + // A second (impossible for S=1, but defensive) request after firing is a + // no-op. + assert_eq!( + g.request(ShardId(0), t0(), RESTART_GRACE), + Decision::AlreadyFired + ); + } + + /// With multiple hosted shards, the first requester DEFERS (a sibling may be + /// mid-install); once the LAST hosted shard also requests, the gate fires. + #[test] + fn multi_shard_defers_until_all_request_then_fires() { + let now = t0(); + let mut g = GateState::new(3); + match g.request(ShardId(0), now, RESTART_GRACE) { + Decision::Defer { deadline } => assert_eq!(deadline, now + RESTART_GRACE), + other => panic!("first of 3 must defer, got {other:?}"), + } + // Second still defers — one shard could still be mid-install. + assert!(matches!( + g.request(ShardId(1), now, RESTART_GRACE), + Decision::Defer { .. } + )); + // Third completes the set → fire now (nothing left mid-install). + assert_eq!(g.request(ShardId(2), now, RESTART_GRACE), Decision::FireNow); + // Anything after the fire is a no-op. + assert_eq!( + g.request(ShardId(0), now, RESTART_GRACE), + Decision::AlreadyFired + ); + } + + /// The grace deadline bounds the wait: a healthy sibling that never requests + /// a restart must not wedge the exit forever. After the grace elapses, the + /// deferred timer fires exactly once. + #[test] + fn deadline_fires_once_when_siblings_never_request() { + let now = t0(); + let mut g = GateState::new(3); + assert!(matches!( + g.request(ShardId(0), now, RESTART_GRACE), + Decision::Defer { .. } + )); + // Before the grace: not due. + assert!(!g.fire_due(now + Duration::from_secs(1), RESTART_GRACE)); + // At/after the grace: fires once. + assert!(g.fire_due(now + RESTART_GRACE, RESTART_GRACE)); + // Idempotent: a second timer (from another deferral) does NOT re-fire. + assert!(!g.fire_due(now + RESTART_GRACE + Duration::from_secs(1), RESTART_GRACE)); + } + + /// A `FireNow` (all shards requested) pre-empts a pending deferred timer: the + /// timer's later `fire_due` is a no-op (the exit already fired). + #[test] + fn firenow_preempts_a_pending_deferred_timer() { + let now = t0(); + let mut g = GateState::new(2); + // Shard 0 defers, arming a timer for `now + grace`. + assert!(matches!( + g.request(ShardId(0), now, RESTART_GRACE), + Decision::Defer { .. } + )); + // Shard 1 completes the set → fires immediately. + assert_eq!(g.request(ShardId(1), now, RESTART_GRACE), Decision::FireNow); + // Shard 0's timer elapses later → no second fire. + assert!(!g.fire_due(now + RESTART_GRACE, RESTART_GRACE)); + } + + /// `fire_due` without any prior request never fires (no `first_request`). + #[test] + fn fire_due_without_request_is_inert() { + let mut g = GateState::new(2); + assert!(!g.fire_due(t0() + RESTART_GRACE, RESTART_GRACE)); + } +} diff --git a/tidal-server/tests/reseed_install.rs b/tidal-server/tests/reseed_install.rs index 5a9e004..8e89c20 100644 --- a/tidal-server/tests/reseed_install.rs +++ b/tidal-server/tests/reseed_install.rs @@ -239,17 +239,20 @@ fn boot_install_discovers_fetches_and_swaps() { file_b ); - // The install sentinel is present and records the artifact seq AND the - // discovered leader's region (drives §2.6 + the post-install catch-up pull). - // `leader` is region index 1 in the topology (`joiner` is 0). + // The install sentinel is present and records the artifact seq, the + // discovered leader's region, AND the discovered leader's term (drives + // §2.6 + the durable term-marker synthesis + the post-install catch-up + // pull). `leader` is region index 1 in the topology (`joiner` is 0); the + // leader's status server reports term 7. let sentinel = reseed::read_install_sentinel(&data_dir).unwrap(); assert_eq!( sentinel, Some(reseed::InstallSentinel { snapshot_seq: 4096, leader_region: Some(1), + artifact_term: Some(7), }), - "install sentinel carries the artifact seq + discovered leader region" + "install sentinel carries the artifact seq + discovered leader region + term" ); // The reseed marker is GONE (it lived in the old dir, discarded by the swap). diff --git a/tidal/Cargo.toml b/tidal/Cargo.toml index bf376d2..48fc41f 100644 --- a/tidal/Cargo.toml +++ b/tidal/Cargo.toml @@ -205,6 +205,10 @@ required-features = ["test-utils"] name = "m11p5_membership_record" required-features = ["test-utils"] +[[test]] +name = "m12_reseed_term_marker" +required-features = ["test-utils"] + [[test]] name = "m12p6_graph_persistence" required-features = ["test-utils"] diff --git a/tidal/tests/m12_reseed_term_marker.rs b/tidal/tests/m12_reseed_term_marker.rs new file mode 100644 index 0000000..7ce8d8b --- /dev/null +++ b/tidal/tests/m12_reseed_term_marker.rs @@ -0,0 +1,207 @@ +//! m12 reseed-loop-fix — the durable term marker the reseed seed synthesizes. +//! +//! The boot-time snapshot install restores a checkpoint-based artifact whose +//! engine WAL is EMPTY, so `wal_term_mark()` would report `tail_term = 0`. The +//! post-open seed (`ShardReplica::new`) then durably synthesizes the artifact's +//! term marker via [`TidalDb::append_term_marker`] so the WAL-tail term is +//! truthful — on this boot AND every reboot — which is what stops `decide_join` +//! from looping a checkpoint-restored node through `ReseedRequired` forever. +//! +//! This test proves the DURABILITY MECHANISM that fix relies on at the engine +//! layer (the `decide_join` classification itself is unit-tested in +//! `cluster::election_driver`): +//! +//! - `append_term_marker` folds the WAL-tail term cell immediately (this boot); +//! - the marker survives a replication checkpoint whose WAL marker advances +//! PAST the marker's seqno (the exact reseed-seed ordering: marker, then +//! `persist_replication_checkpoint`) AND a hard crash-reopen — because blob +//! records are deliberately NOT checkpoint-filtered on recovery +//! (`wal::reader`), so recovery re-derives the truthful term on every restart; +//! - the append path refuses term 0 (the topology era) and non-cluster mode. + +#![allow(clippy::unwrap_used)] + +use tidaldb::{ + TempTidalHome, TidalDb, + db::config::{NodeConfig, NodeRole}, + replication::shard::ShardId, + schema::{DecaySpec, EntityKind, SchemaBuilder, Window}, + testing::{CrashInjector, CrashPoint, crash_injector::run_with_crash}, + wal::format::{MemberEntry, MemberRole, MembershipRecord}, +}; + +use std::time::Duration; + +fn make_schema() -> tidaldb::schema::Schema { + let mut builder = SchemaBuilder::new(); + let _ = builder + .signal( + "view", + EntityKind::Item, + DecaySpec::Exponential { + half_life: Duration::from_secs(7 * 24 * 3600), + }, + ) + .windows(&[Window::AllTime]) + .velocity(false) + .add(); + builder.build().unwrap() +} + +/// Open a PERSISTENT cluster node (non-empty `peer_shards` → `replicate_blobs`, +/// the WAL becomes the one replicated log so kind-3 markers can ride it). +fn open_cluster_node(home: &TempTidalHome, schema: tidaldb::schema::Schema) -> TidalDb { + TidalDb::builder() + .with_data_dir(home.path()) + .with_schema(schema) + .with_cluster(NodeConfig { + role: NodeRole::Single, + shard_id: ShardId(0), + peer_shards: vec![ShardId(1)], + ..NodeConfig::default() + }) + .open() + .expect("persistent cluster node opens") +} + +/// A kind-4 record used only to push the WAL frontier (and thus the checkpoint +/// seq) PAST the term marker, so the durability assertion is "the marker +/// survives a checkpoint advanced beyond it", not merely "at it". +fn roster() -> MembershipRecord { + MembershipRecord { + version: 1, + term: 7, + members: vec![ + MemberEntry { + id: 0, + name: "us-east".to_string(), + grpc_addr: "tidaldb-0:9500".to_string(), + http_addr: "http://tidaldb-0:9501".to_string(), + role: MemberRole::Voter, + }, + MemberEntry { + id: 1, + name: "eu-west".to_string(), + grpc_addr: "tidaldb-1:9500".to_string(), + http_addr: "http://tidaldb-1:9501".to_string(), + role: MemberRole::Voter, + }, + ], + } +} + +/// The reseed seed's `append_term_marker` folds the WAL-tail term cell +/// immediately: `wal_term_mark()` reports the artifact's (term, region) the +/// moment the seed runs, on the very boot that installed the snapshot. +#[test] +fn append_term_marker_folds_the_cell_this_boot() { + let home = TempTidalHome::new().unwrap(); + let db = open_cluster_node(&home, make_schema()); + + // A checkpoint-restored artifact's WAL is empty → topology era, term 0. + assert_eq!(db.wal_term_mark(), (0, 0, 0)); + + // The seed synthesizes the discovered leader's (term 7, region 1). + let seq = db.append_term_marker(7, 1).unwrap(); + assert!(seq >= 1, "the marker consumes a real seqno, got {seq}"); + + let (term, marker_seq, region) = db.wal_term_mark(); + assert_eq!(term, 7, "the WAL-tail term is the artifact's term, not 0"); + assert_eq!(marker_seq, seq); + assert_eq!(region, 1, "the marker names the reseed-leader stream"); +} + +/// THE durability property the reseed-loop-fix hinges on: a synthesized term +/// marker survives a replication checkpoint whose WAL marker lands PAST the +/// marker's seqno AND a hard crash-reopen. If it did not, the next boot would +/// re-read `tail_term = 0` and re-latch `ReseedRequired` — the loop. +/// +/// A crash (not a clean `close()`) is the faithful test: it exercises the +/// re-derive-from-the-surviving-log path and sidesteps the shutdown-time WAL +/// compaction that could reclaim a control-only segment. +#[test] +fn term_marker_survives_checkpoint_past_it_and_crash_reopen() { + let home = TempTidalHome::new().unwrap(); + let schema = make_schema(); + + // The first CheckpointPreFlush passes (our explicit force_replication_checkpoint + // lands the WAL marker durably PAST the term marker's seq); the second (the + // shutdown ledger checkpoint) fires the crash, so no compaction runs. + let injector = CrashInjector::new(CrashPoint::CheckpointPreFlush, 1); + let outcome = run_with_crash(&injector, || { + let db = open_cluster_node(&home, schema.clone()); + + // The reseed seed: synthesize the artifact's term marker (lands at seq 1). + let marker_seq = db.append_term_marker(7, 1).unwrap(); + // Push the frontier PAST the marker so the checkpoint seq exceeds it — + // this is what proves the marker survives a checkpoint advanced BEYOND + // it (blobs are not checkpoint-filtered), not merely a checkpoint at it. + let later_seq = db.append_membership_record(roster()).unwrap(); + assert!( + later_seq > marker_seq, + "the membership record must sit above the term marker (got {later_seq} > {marker_seq})" + ); + // The live cell already reports the truthful term before the crash. + assert_eq!(db.wal_term_mark().0, 7); + + // Land a crash-consistent checkpoint: the WAL marker advances to + // `later_seq` — strictly past the term marker at `marker_seq`. This is + // the exact reseed-seed ordering (append marker, then persist checkpoint), + // via the SAME public API the seed calls. + db.persist_replication_checkpoint().unwrap(); + + // CRASH during the shutdown ledger checkpoint — nothing past here, and + // no compaction, reaches disk. + let _ = db.close(); + }); + assert!( + matches!(outcome, Err(CrashPoint::CheckpointPreFlush)), + "the crash injector must have fired during shutdown, got {outcome:?}" + ); + + // Reopen the crashed node: recovery scans the surviving log. The term marker + // sits BELOW the checkpoint boundary, but blob records are not checkpoint- + // filtered, so recovery re-folds it → the WAL-tail term is truthful again. + let db = open_cluster_node(&home, schema); + let (term, _seq, region) = db.wal_term_mark(); + assert_eq!( + term, 7, + "recovery re-derives the artifact's term across the checkpoint boundary — \ + not 0, so decide_join will NOT re-latch ReseedRequired" + ); + assert_eq!( + region, 1, + "the recovered marker still names the reseed-leader stream" + ); +} + +/// The append path refuses term 0 (the topology era never journals a marker, so +/// the seed correctly SKIPS the synthesis for a term-0 artifact). +#[test] +fn append_term_marker_refuses_term_zero() { + let home = TempTidalHome::new().unwrap(); + let db = open_cluster_node(&home, make_schema()); + + let err = db.append_term_marker(0, 1).unwrap_err(); + assert!( + err.to_string().contains("elected terms"), + "term 0 is refused, got: {err}" + ); + assert_eq!(db.wal_term_mark(), (0, 0, 0), "no marker entered the log"); +} + +/// A non-cluster (standalone) node refuses the append: the WAL is not a +/// replicated log there, so the reseed seed (which only runs in cluster mode) +/// could never reach this. +#[test] +fn standalone_node_refuses_term_marker_append() { + let home = TempTidalHome::new().unwrap(); + let db = TidalDb::builder() + .with_data_dir(home.path()) + .with_schema(make_schema()) + .open() + .expect("standalone node opens"); + + let err = db.append_term_marker(7, 1).unwrap_err(); + assert!(err.to_string().contains("cluster mode"), "got: {err}"); +}