diff --git a/tidal-net/src/sources.rs b/tidal-net/src/sources.rs index 4d6810c..bc451eb 100644 --- a/tidal-net/src/sources.rs +++ b/tidal-net/src/sources.rs @@ -227,6 +227,32 @@ pub trait SnapshotRequiredSink: Send + Sync + 'static { fn snapshot_required(&self, shard: ShardId, from_seqno: u64); } +/// Sink for a catch-up stream that COMPLETED, i.e. positive proof that the +/// source actually served the requested range. +/// +/// This is the counterpart of [`SnapshotRequiredSink`] and exists because a +/// reseed marker must never be discharged by comparing frontier numbers. A +/// follower's applied frontier is a HIGH-WATER-MARK: a term join re-bases it +/// onto the new leader's stream, so it can leap across history the node never +/// received. Both historical predicates (`applied >= leader_last_seq`, and its +/// replacement `applied >= marker.from_seqno`) therefore discharged markers on +/// nodes that still had a hole — reported `lag_events: 0`, and silently served +/// incomplete reads. +/// +/// A completed pull is the only cheap, sound evidence the stream closed the gap: +/// the source streamed the range and the receiver applied every chunk. The other +/// sound discharge is a snapshot install, which replaces the data dir outright +/// (and takes the marker with it). +/// +/// Wired like [`SnapshotRequiredSink`] (an `OnceLock` on [`ServingSources`]). +pub trait CatchupServedSink: Send + Sync + 'static { + /// A `StreamSegments` pull from `shard` that began at `from_seqno` ran to + /// completion. Everything from `from_seqno` up to the source's stream end at + /// open time is now applied locally, so a reseed marker whose own + /// `from_seqno` is at or above this one has been genuinely served. + fn catchup_served(&self, shard: ShardId, from_seqno: u64); +} + /// The heartbeat exchange verdict an [`ElectionHooks`] implementation /// returns: the responder's (possibly just-raised) term and whether the /// sender's leadership assertion was accepted. @@ -442,6 +468,12 @@ pub struct ServingSources { /// Late-bound consumer of typed `snapshot-required` catch-up refusals /// (m11p5 §2.4). Unset = log + retry-timer only (pre-m11p5 behavior). pub snapshot_required: Arc>>, + /// Late-bound consumer of COMPLETED catch-up pulls: the positive-evidence + /// counterpart of `snapshot_required`, and the only sound way to discharge a + /// reseed marker short of a snapshot install. Unset = a completed pull is + /// logged only (a marker then waits for the reseed, which is the safe + /// direction). + pub catchup_served: Arc>>, /// Consumer of follower durable-frontier reports (m11p3). A `OnceLock` /// because the embedding application can only build it AFTER the /// transport exists (the ship queue takes the transport): fill it via @@ -481,6 +513,12 @@ impl ServingSources { let _ = self.snapshot_required.set(sink); } + /// Late-bind the completed-catch-up sink (idempotent; first set wins). Until + /// set, a completed pull discharges nothing. + pub fn set_catchup_served_sink(&self, sink: Arc) { + let _ = self.catchup_served.set(sink); + } + /// Late-bind the `JoinCluster` hooks (idempotent; first set wins). Until /// set, `JoinCluster` answers `Unimplemented`. pub fn set_join_hooks(&self, hooks: Arc) { diff --git a/tidal-net/src/transport.rs b/tidal-net/src/transport.rs index 986cac8..2707a8c 100644 --- a/tidal-net/src/transport.rs +++ b/tidal-net/src/transport.rs @@ -169,6 +169,10 @@ struct CatchupRunner { /// invokes it so the node latches a reseed marker. Absent → today's /// behavior (log + the standing retry timer). snapshot_required: Arc>>, + /// Late-bound COMPLETED-pull sink: the positive evidence that discharges a + /// reseed marker. Absent → a completed pull discharges nothing (safe: the + /// marker then waits for a snapshot reseed). + catchup_served: Arc>>, } impl CatchupRunner { @@ -306,6 +310,15 @@ impl CatchupRunner { chunks, "catch-up stream complete" ); + // POSITIVE EVIDENCE (the only sound reseed-marker discharge + // short of a snapshot install): the source streamed the whole + // requested range and every chunk was applied. A frontier + // comparison cannot substitute for this — a term join re-bases + // the applied frontier onto the new leader's stream, so it can + // leap over history the node never received. + if let Some(sink) = self.catchup_served.get() { + sink.catchup_served(from_shard, from_seqno); + } return PullOutcome::Complete; } Err(status) => { @@ -639,6 +652,9 @@ impl GrpcTransport { // surfaces a `snapshot-required` trailer to it so the node latches a // reseed marker. Late-bound, shared with the gRPC service. let snapshot_required = Arc::clone(&sources.snapshot_required); + // The positive-evidence counterpart: a completed pull is what discharges a + // reseed marker. Same late-bound sharing as `snapshot_required`. + let catchup_served = Arc::clone(&sources.catchup_served); // The shutdown latch is created up front (m11p7): the gRPC server's mTLS // accept loop and its serve future both exit on it, so it must exist @@ -700,6 +716,7 @@ impl GrpcTransport { applied: applied_for_catchup, election: Arc::clone(&election), snapshot_required, + catchup_served, }); Ok(Self { diff --git a/tidal-net/tests/catchup_retry.rs b/tidal-net/tests/catchup_retry.rs index 16c2c38..47a53ca 100644 --- a/tidal-net/tests/catchup_retry.rs +++ b/tidal-net/tests/catchup_retry.rs @@ -109,6 +109,7 @@ fn failed_pull_retries_on_timer_with_no_push() { election: Arc::default(), snapshots: Arc::default(), snapshot_required: Arc::default(), + catchup_served: Arc::default(), join: Arc::default(), }; let _leader = GrpcTransport::new_with_sources( @@ -165,6 +166,7 @@ fn clean_completion_does_not_keep_retrying() { election: Arc::default(), snapshots: Arc::default(), snapshot_required: Arc::default(), + catchup_served: Arc::default(), join: Arc::default(), }; let _leader = GrpcTransport::new_with_sources( diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 20295f6..8454564 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -426,6 +426,10 @@ pub struct ShardReplica { /// this node's reseed marker. snapshot_required_cell: Arc>>, + /// Late-bound sink for COMPLETED catch-up pulls — the positive evidence that + /// discharges this node's reseed marker (a frontier comparison cannot; see + /// [`ShardReplica::discharge_reseed_marker_if_served`]). + catchup_served_cell: Arc>>, /// The effective roster (m11p5 §3): the single source of roster truth. /// Derived at boot from the WAL-recovered `ClusterMembership` cell when /// non-`None` (the membership era — RECORD ids), else from the topology @@ -753,6 +757,8 @@ impl ShardReplica { // so it can only be built once the node is in its final Arc). Until set, // a snapshot-required catch-up trailer falls back to log + retry. let snapshot_required_cell = Arc::clone(&sources.snapshot_required); + // The positive-evidence counterpart, same late-bound discipline. + let catchup_served_cell = Arc::clone(&sources.catchup_served); // m11p5: the snapshot-source cell, late-bound below once `cluster_metrics` // exists (the source reports the staged/fetch/force-drop series). Until // set, `FetchSnapshot` answers Unimplemented. @@ -1214,6 +1220,7 @@ impl ShardReplica { converged: AtomicBool::new(false), self_restart_refused: AtomicBool::new(false), snapshot_required_cell, + catchup_served_cell, membership, join_hooks_cell, promote_inflight: AtomicBool::new(false), @@ -2872,25 +2879,19 @@ impl ShardReplica { ); } - // Winning an election proves this node's log is at least as up-to-date as a - // quorum (the vote restriction), so a reseed marker it latched while - // transiently behind at an earlier term change is a false alarm — clear it - // (else a caught-up LEADER serves with `reseed_required:true` and reseeds on - // its next restart). A quarantined node is campaign-suppressed and can never - // reach here, so the helper's exclusion is doubly safe. + // NO reseed-marker discharge here, deliberately. Winning an election proves + // this node's log is at least as up-to-date as a QUORUM (the vote + // restriction) — it does NOT prove the node's log is CONTIGUOUS. A node + // whose applied frontier was re-based across a compacted gap satisfies the + // vote restriction while still missing history, so discharging here would + // promote a leader with a hole in its log. Earlier revisions cleared the + // marker at this point (first unconditionally, then on a frontier + // comparison); both could strand a divergent node as leader. // - // Pass this shard's REAL applied frontier, not a hardcoded `true`: winning - // proves the log beats a quorum's, which is not the same as proving a - // specific compacted gap was closed, and a `DivergentPostBaseline` marker - // is not campaign-suppressed so it CAN reach here. The marker decides. - // - // For a LEADER that frontier is the durable flushed WAL frontier, not - // `applied_seqno` — a leader writes its WAL directly instead of applying - // its own stream through the receiver, so `applied_seqno(own_shard)` stays - // at its stale follower-era value (0 on a node elected without ever - // following) and would refuse every discharge. Same derivation - // `local_status` uses for the leader row. - self.clear_stale_reseed_marker_if_discharged(self.ship_feed.flushed_seq()); + // A genuine false-alarm marker is discharged by + // `discharge_reseed_marker_if_served` when a catch-up pull actually serves + // the latching range, which is the same evidence a leader would need. A + // marker that survives to leadership is real: let the reseed run. // §3.2 — the activation membership record (the linchpin): once the // kind-3 marker is durable AND the membership era has begun (a kind-4 @@ -3130,30 +3131,35 @@ impl ShardReplica { } } - /// Clear a reseed marker that turned out to be a FALSE ALARM (rolling-restart - /// churn fix, second source). The `own < prev_log → ReseedRequired` join-check - /// arm latches a durable marker for a node that is merely BEHIND at a - /// leadership change — but a node behind by a *shippable* (non-compacted) tail - /// then catches up via the normal stream / catch-up pull and never needs a - /// snapshot reseed. The durable marker used to persist anyway, so the node - /// reseeded on its next restart (and could even lead while still flagged — the - /// observed `is_leader:true, lag:0, reseed_required:true` state). + /// Discharge a reseed marker on POSITIVE EVIDENCE that the stream served the + /// range which latched it: a `StreamSegments` pull that began at or below the + /// marker's `from_seqno` and ran to completion. /// - /// The discharge decision belongs to the MARKER, not to this call site: - /// [`ReseedMarker::discharged_by`] requires both a stream-dischargeable reason - /// and an applied frontier that has reached the marker's own `from_seqno`. + /// This exists for the genuine false-alarm case (the `own < prev_log → + /// ReseedRequired` join arm latches for a node merely BEHIND by a shippable + /// tail, which the stream then serves), without the unsoundness of the two + /// frontier comparisons that preceded it. /// - /// The predicate used to be `applied >= leader_last_seq` — the leader's TAIL — - /// with the stated invariant "a node genuinely behind a COMPACTED gap never - /// reaches caught_up". Production falsified it on 2026-08-20: tidaldb-0 was - /// short four entries the leader had already compacted away, met the tail - /// comparison anyway, cleared its own marker ~200ms after latching it, and so - /// never ran the boot reseed that was the only way to close the gap. With - /// `reseed_self_restart` on, it exit-looped 196 times in 21h — and because - /// the clear also reset `tidaldb_cluster_reseed_required`, the gauge flapped - /// 1→0 every 30s and the `TidalDBClusterReseedPending` alert (`== 1 for 10m`) - /// could never fire. Keep the comparison anchored to `from_seqno`. - fn clear_stale_reseed_marker_if_discharged(&self, applied: u64) { + /// # Why no frontier comparison can work here + /// + /// The first version asked `applied >= leader_last_seq`, asserting "a node + /// genuinely behind a COMPACTED gap never reaches caught_up". The second asked + /// `applied >= marker.from_seqno`. BOTH are unsound for the same reason: the + /// applied frontier is a HIGH-WATER-MARK, and a term join re-bases it onto the + /// new leader's stream (`replication_state().advance(.., baseline + 1)`), so it + /// leaps over history the node never received. + /// + /// Measured, not theorised — `mp_follower_reseeds_via_snapshot_after_compaction` + /// stops a follower at frontier 9, compacts the leader so it retains only from + /// 15722, and the follower's frontier is re-based to 16810. Both predicates + /// discharge the marker there. The node then skips its reseed and reports + /// `lag_events: 0` while missing 10..15721 — a silent hole, served to readers. + /// Production showed the same shape: `applied 13540660` against a marker + /// resuming at 13540653 that no live WAL could serve. + /// + /// So the only sound discharges are: a completed pull covering the gap (here), + /// or a snapshot install (which replaces the data dir and the marker with it). + fn discharge_reseed_marker_if_served(&self, served_from: u64) { if self .election_runtime .get() @@ -3164,12 +3170,11 @@ impl ShardReplica { let Ok(Some(marker)) = self.reseed_marker_store.load() else { return; }; - if !marker.discharged_by(applied) { - // Refuse silently: this runs on every heartbeat (~300ms). The - // undischarged state is already observable — the marker's latch logged - // WARN once, `tidaldb_cluster_reseed_required` stays 1 (which is what - // makes the 10m alert fire now), and `/cluster/status/local` reports - // `reseed_required: true`. + if !marker.discharged_by_served_range(served_from) { + // Refuse quietly. The undischarged state is already observable: the + // latch logged WARN, `tidaldb_cluster_reseed_required` stays 1 (which + // is what makes the 10m alert reachable), and `/cluster/status/local` + // reports `reseed_required: true`. return; } match self.reseed_marker_store.clear() { @@ -3182,16 +3187,16 @@ impl ShardReplica { region = %self.region_name, reason = marker.reason.as_str(), from_seqno = marker.from_seqno, - applied, - "stale reseed marker cleared — the stream served the entry that latched it \ - (applied >= from_seqno), so no reseed is needed (false-alarm latch from a \ - transient leadership-change classification)" + served_from, + "reseed marker discharged — a catch-up pull completed from at/below the \ + seqno that latched it, so the stream genuinely closed the gap and no \ + reseed is needed" ); } Err(e) => tracing::warn!( region = %self.region_name, error = %e, - "failed to clear a stale reseed marker; retried on the next convergence" + "failed to clear a discharged reseed marker; retried on the next served pull" ), } } @@ -3504,12 +3509,12 @@ impl ShardReplica { if (self.install_boot || self.seed_joiner) && !self.converged.load(Ordering::Acquire) { self.note_lag_for_readiness(leader_last_seq.saturating_sub(applied)); } - // Self-heal a false-alarm reseed marker: if the stream served the very entry - // whose absence latched it, no reseed is needed. Anchored to the marker's - // `from_seqno`, NOT to `leader_last_seq` — see - // `clear_stale_reseed_marker_if_discharged` for the livelock that comparison - // caused. - self.clear_stale_reseed_marker_if_discharged(applied); + // NO reseed-marker discharge on the heartbeat path. This is where both + // unsound predicates lived (`applied >= leader_last_seq`, then `applied >= + // marker.from_seqno`). A heartbeat carries frontier numbers only, and a + // frontier is a high-water-mark that a term join re-bases across + // un-received history — see `discharge_reseed_marker_if_served`, which is + // driven by a COMPLETED catch-up pull instead. } /// Whether this node is READY to serve (m11p5 §4 readiness predicate). @@ -3660,6 +3665,16 @@ impl ShardReplica { }) as Arc); + // The positive-evidence counterpart: a COMPLETED catch-up pull is what + // discharges the marker the sink above latches. Bound here for the same + // Weak-back-reference reason. + let _ = self + .catchup_served_cell + .set(Arc::new(NodeCatchupServedSink { + node: Arc::downgrade(self), + }) + as Arc); + // m11p5 §3.3: late-bind the `JoinCluster` adapter (same Weak-back-ref // discipline). Until set, `JoinCluster` answers Unimplemented. let _ = self.join_hooks_cell.set(Arc::new(NodeJoinHooks { @@ -6977,6 +6992,24 @@ impl tidal_net::sources::SnapshotRequiredSink for NodeSnapshotRequiredSink { } } +struct NodeCatchupServedSink { + node: Weak, +} + +impl tidal_net::sources::CatchupServedSink for NodeCatchupServedSink { + fn catchup_served(&self, _shard: ShardId, from_seqno: u64) { + let Some(node) = self.node.upgrade() else { + return; // node shutting down; nothing to discharge + }; + // A pull that STARTED at `from_seqno` ran to completion, so the source + // streamed that range and the receiver applied it. This is the positive + // evidence a reseed marker needs — the only alternative being a snapshot + // install. Never discharge from a frontier comparison; see + // `ShardReplica::discharge_reseed_marker_if_served`. + node.discharge_reseed_marker_if_served(from_seqno); + } +} + /// Build a write-success response carrying the `x-tidal-seq` header (the /// write's replicated-log seqno, m11p3). The dedup sentinel (`Some(0)` — an /// identical record is already durable, no new log entry exists) carries diff --git a/tidal-server/tests/cluster_reseed.rs b/tidal-server/tests/cluster_reseed.rs index 4d8b0e3..085fc3f 100644 --- a/tidal-server/tests/cluster_reseed.rs +++ b/tidal-server/tests/cluster_reseed.rs @@ -200,6 +200,54 @@ fn item_searchable(cluster: &MultiProcCluster, idx: usize, entity_id: u64) -> bo .is_some_and(|r| r.iter().any(|x| x["entity_id"].as_u64() == Some(entity_id))) } +/// The `first_seq` of every WAL segment retained on node `idx`, ascending. +/// +/// Segment filenames are `wal-{first_seq:020}.seg` for the single-shard layout +/// (`tidal::wal::segment::segment_filename`), so the retained history floor is +/// the FIRST entry: the source can serve a catch-up pull from that seq onward and +/// no earlier. +fn retained_segment_first_seqs(cluster: &MultiProcCluster, idx: usize) -> Vec { + let wal_dir = cluster.data_dir(idx).join("wal"); + let mut seqs: Vec = std::fs::read_dir(&wal_dir) + .into_iter() + .flatten() + .flatten() + .filter_map(|e| { + let name = e.file_name().to_string_lossy().into_owned(); + let rest = name.strip_prefix("wal-")?.strip_suffix(".seg")?; + rest.rsplit('-').next()?.parse::().ok() + }) + .collect(); + seqs.sort_unstable(); + seqs +} + +/// Assert the PREMISE of the compaction gate: after the leader's graceful +/// restart, its retained WAL must no longer cover `resume_seq`. +/// +/// This exists because the gate previously asserted only its CONSEQUENCE +/// (`reseed_required == true`). When the premise silently stopped holding, the +/// test failed 40s later pointing at the follower — which looked like a reseed +/// bug and was actually "the leader never compacted anything". Assert the setup +/// so an un-sized fixture names itself. +fn assert_history_compacted_past(cluster: &MultiProcCluster, idx: usize, resume_seq: u64) { + let seqs = retained_segment_first_seqs(cluster, idx); + let floor = seqs.first().copied().unwrap_or(0); + assert!( + !seqs.is_empty(), + "leader has no WAL segments at all; the fixture is broken, not the reseed path" + ); + assert!( + floor > resume_seq, + "PREMISE FAILED: the leader still serves the follower's resume seq, so no \ + snapshot-required refusal can ever happen and this gate proves nothing. \ + retained segment first_seqs = {seqs:?} (floor {floor}) must all exceed \ + resume_seq {resume_seq}. Either the offline batch no longer rotates past \ + WAL_RETENTION_SEGMENTS (re-size OFFLINE_ITEMS against that constant and the \ + 16 MiB segment size) or the graceful-shutdown compaction stopped deleting." + ); +} + /// Read a status field as a bool (defaulting `false` when the node is /// unreachable or the field is absent). fn status_bool(cluster: &MultiProcCluster, idx: usize, field: &str) -> bool { @@ -430,6 +478,17 @@ fn mp_follower_reseeds_via_snapshot_after_compaction() { promote_and_agree(&cluster, LEADER, cluster.region_name(LEADER)); println!("[reseed] leader gracefully restarted (WAL compacted) and re-promoted"); + // PREMISE CHECK (see `assert_history_compacted_past`): the follower resumes at + // `frontier_at_stop + 1`, so the leader's retained WAL must no longer cover it. + // Without this, an un-sized fixture silently turns the gate into a no-op that + // fails 40s later looking like a follower bug. + assert_history_compacted_past(&cluster, LEADER, frontier_at_stop + 1); + println!( + "[reseed] premise holds: leader retains {:?}, follower resume seq {} is gone", + retained_segment_first_seqs(&cluster, LEADER), + frontier_at_stop + 1 + ); + // ── Restart the stopped follower (first boot: NO marker yet). Its runtime // boot-time catch-up pull requests from its frontier+1, which the leader's // compacted WAL cannot serve → typed `snapshot-required` refusal → the diff --git a/tidal/src/replication/reseed.rs b/tidal/src/replication/reseed.rs index 6ca213a..06d1b71 100644 --- a/tidal/src/replication/reseed.rs +++ b/tidal/src/replication/reseed.rs @@ -163,28 +163,37 @@ pub struct ReseedMarker { } impl ReseedMarker { - /// Whether an applied frontier of `applied` PROVES this marker no longer - /// needs a snapshot reseed. + /// Whether a catch-up pull that COMPLETED from `served_from` discharges this + /// marker. /// /// Two conditions, both required: /// /// 1. the reason must be [`ReseedReason::stream_dischargeable`], and - /// 2. the node must have applied THROUGH `from_seqno` — the very entry whose - /// unavailability latched the marker. + /// 2. the completed pull must have STARTED at or below this marker's + /// `from_seqno` — i.e. the range whose unavailability latched the marker is + /// inside the range the source actually streamed. /// - /// Condition 2 is the one that makes this safe against a genuinely compacted - /// gap: the leader refused the range *starting at* `from_seqno`, so those - /// entries exist in no live WAL and `applied` can never reach it without a - /// snapshot install. Comparing against the leader's TAIL instead (the - /// pre-fix behavior) discharged the marker on any node whose frontier - /// happened to meet the leader's last seqno, which on a quiet shard is every - /// node — including one missing committed history it could never refetch. - /// That inverted the guarantee: the clear suppressed the only action that - /// closes the gap, so the node re-latched, re-cleared, and (with - /// `reseed_self_restart`) exit-looped forever without ever reseeding. + /// # Why this takes a served range and not an applied frontier + /// + /// Two earlier predicates compared frontier numbers and BOTH were unsound: + /// `applied >= leader_last_seq` (the leader's tail) and `applied >= + /// from_seqno` (this marker's own resume point). The applied frontier is a + /// HIGH-WATER-MARK, not a contiguity proof: a term join re-bases it onto the + /// new leader's stream (`replication_state().advance(.., baseline + 1)`), so + /// it can leap across history the node never received. A node 15,000 entries + /// behind a compacted leader would satisfy either comparison the instant its + /// frontier was re-based, discharge its marker, skip the reseed, and then + /// report `lag_events: 0` while serving reads from a log with a hole in it. + /// That is exactly the 2026-08-20 production failure and the local + /// `mp_follower_reseeds_via_snapshot_after_compaction` reproduction. + /// + /// A completed pull is different in kind: it is evidence that the source + /// streamed the range and the receiver applied every chunk. The only other + /// sound discharge is a snapshot install, which replaces the data dir and + /// takes the marker with it. #[must_use] - pub const fn discharged_by(self, applied: u64) -> bool { - self.reason.stream_dischargeable() && applied >= self.from_seqno + pub const fn discharged_by_served_range(self, served_from: u64) -> bool { + self.reason.stream_dischargeable() && served_from <= self.from_seqno } } @@ -380,54 +389,75 @@ mod tests { } } - /// The compacted-gap livelock regression (2026-08-20 incident): a - /// `SnapshotRequired` marker latched because the leader's WAL was compacted - /// below `from_seqno` must NOT discharge while the node is still short of - /// that seqno. The pre-fix predicate compared the applied frontier against - /// the LEADER'S TAIL, so a node missing four unrefetchable entries cleared - /// its own marker, skipped the reseed on the next boot, re-latched, and - /// exit-looped 196 times over 21h without ever reseeding. + /// The 2026-08-20 livelock AND the silent-hole bug behind it. A + /// `SnapshotRequired` marker must not discharge just because the node's + /// applied frontier moved past `from_seqno` — the frontier is a high-water + /// mark that a term join re-bases across un-received history. Only a + /// completed pull that actually covered `from_seqno` discharges it. + /// + /// Reproduced locally by `mp_follower_reseeds_via_snapshot_after_compaction`: + /// a follower stopped at frontier 9, the leader compacted past it (retaining + /// only from 15722), and the follower's frontier was re-based to 16810. Both + /// historical predicates — `applied >= leader_last_seq` and `applied >= + /// from_seqno` — discharged the marker there, skipping the reseed and leaving + /// the node reporting `lag_events: 0` while missing 10..15721. #[test] - fn compacted_gap_marker_never_discharges_below_from_seqno() { + fn compacted_gap_marker_needs_a_pull_that_actually_covered_the_gap() { + let marker = ReseedMarker { + reason: ReseedReason::SnapshotRequired, + from_seqno: 10, + }; + // A pull that started ABOVE the gap proves nothing about the gap, however + // far the frontier has since travelled. This is the exact shape that let a + // node with a 15,000-entry hole call itself converged. + assert!(!marker.discharged_by_served_range(15_722)); + assert!(!marker.discharged_by_served_range(16_810)); + // A pull that started AT or BELOW the gap did stream it. + assert!(marker.discharged_by_served_range(10)); + assert!(marker.discharged_by_served_range(1)); + } + + /// The production incident's own numbers, same rule: the marker resumed at + /// 13_540_653 and the leader's earliest available was 13_540_657, so no pull + /// can ever start at or below 13_540_653 — the marker is undischargeable by + /// the stream and the reseed MUST run. A frontier of 13_540_660 (which the + /// live node reported, and which the previous predicate accepted) is not + /// evidence of anything. + #[test] + fn production_compacted_gap_is_undischargeable_by_any_servable_pull() { let marker = ReseedMarker { reason: ReseedReason::SnapshotRequired, from_seqno: 13_540_653, }; - // The live incident's exact numbers: the follower held 13_540_652 and the - // leader's earliest available was 13_540_657, so the requested entry is - // gone for good. Every frontier below `from_seqno` must refuse. - assert!(!marker.discharged_by(13_540_652)); - assert!(!marker.discharged_by(0)); - // Even a frontier at/above the LEADER'S TAIL (13_540_660 at the time) - // must refuse while it is below `from_seqno` — impossible here by - // construction, but it is the comparison the bug actually made, so pin - // that the predicate no longer consults anything but `from_seqno`. - assert!(!marker.discharged_by(marker.from_seqno - 1)); - // Applying THROUGH the missing entry is the only proof, and it discharges. - assert!(marker.discharged_by(13_540_653)); - assert!(marker.discharged_by(13_540_999)); + for earliest_available in [13_540_657_u64, 13_540_660, 13_541_000] { + assert!( + !marker.discharged_by_served_range(earliest_available), + "a pull starting at {earliest_available} cannot have served 13540653" + ); + } + assert!(marker.discharged_by_served_range(13_540_653)); } - /// The false-alarm case the m12 reseed-loop-fix targeted still self-heals: - /// a node merely behind a SHIPPABLE tail at a leadership change latches - /// `SnapshotRequired` with `from_seqno = frontier + 1`, then the stream - /// serves exactly that entry and the marker clears with no reseed. + /// The false-alarm case the m12 reseed-loop-fix targeted still self-heals: a + /// node merely behind a SHIPPABLE tail latches at `frontier + 1`, the stream + /// serves from exactly there, and the completed pull discharges the marker + /// with no reseed. #[test] - fn merely_behind_marker_discharges_once_the_tail_is_applied() { + fn merely_behind_marker_discharges_when_the_stream_serves_that_range() { let marker = ReseedMarker { reason: ReseedReason::SnapshotRequired, from_seqno: 501, // latched at frontier 500 }; - assert!(!marker.discharged_by(500)); - assert!(marker.discharged_by(501)); + assert!(marker.discharged_by_served_range(501)); + assert!(!marker.discharged_by_served_range(502)); } - /// Structural markers are never dischargeable by forward progress, whatever - /// the frontier: a divergent post-baseline suffix must be DISCARDED by a - /// snapshot, a quarantine is a fence, and an operator request is an - /// instruction. `u64::MAX` stands in for "arbitrarily caught up". + /// Structural markers are never dischargeable by the stream at all: a + /// divergent post-baseline suffix must be DISCARDED by a snapshot, a + /// quarantine is a fence, and an operator request is an instruction. Even a + /// pull that served from seqno 1 must not clear them. #[test] - fn structural_reasons_never_discharge_by_lag() { + fn structural_reasons_never_discharge_by_the_stream() { for reason in [ ReseedReason::DivergentPostBaseline, ReseedReason::Quarantine, @@ -439,8 +469,8 @@ mod tests { from_seqno: 10, }; assert!( - !marker.discharged_by(u64::MAX), - "{} discharged by lag", + !marker.discharged_by_served_range(1), + "{} discharged by a stream pull", reason.as_str() ); }