fix(m12): break the post-reseed false-ReseedRequired loop (durable term marker + readiness gating + restart coordinator)
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.
This commit is contained in:
jx12n
parent
580142df49
commit
c8ea05b032
@ -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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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<Self>, 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<Self>) {
|
||||
// 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.
|
||||
|
||||
@ -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<u16>,
|
||||
/// 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<u64>,
|
||||
}
|
||||
|
||||
/// 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<Option<InstallSentinel>>
|
||||
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();
|
||||
|
||||
251
tidal-server/src/cluster/reseed_restart.rs
Normal file
251
tidal-server/src/cluster/reseed_restart.rs
Normal file
@ -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<ShardId>,
|
||||
/// When the FIRST request arrived — the grace deadline is measured from it.
|
||||
first_request: Option<Instant>,
|
||||
/// 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<Option<_>>` (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<Option<GateState>> = Mutex::new(None);
|
||||
|
||||
fn with_gate<R>(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));
|
||||
}
|
||||
}
|
||||
@ -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).
|
||||
|
||||
@ -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"]
|
||||
|
||||
207
tidal/tests/m12_reseed_term_marker.rs
Normal file
207
tidal/tests/m12_reseed_term_marker.rs
Normal file
@ -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}");
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user