diff --git a/core/integration/tests/cluster/parked_frame_redispatch.rs b/core/integration/tests/cluster/parked_frame_redispatch.rs index 3b1a525ed4..9ac3194a3f 100644 --- a/core/integration/tests/cluster/parked_frame_redispatch.rs +++ b/core/integration/tests/cluster/parked_frame_redispatch.rs @@ -23,8 +23,9 @@ //! that parked. This test pins the park path positively, and on the replica //! where getting it wrong creates a replica gap: a client request that never //! reaches the plane is answered with a retriable status and the SDK replays it, -//! while a replicated PREPARE has no client behind it and must wait for a later -//! commit heartbeat to arm repair if this path drops it. +//! while a replicated PREPARE has no client behind it: nothing re-sends it once +//! its op has quorum, so the backup gap-stops and waits out `tick_partitions`' +//! repair debounce before anything refetches it. //! //! What makes the window wide on a backup is the commit broadcast. A backup //! learns a metadata commit from the `commit` field of the next prepare on that @@ -50,7 +51,7 @@ //! - Every acked message is readable in dense offset order, each producer's own //! sends stay in the order it made them, and all three replicas hold //! byte-identical segments. A prepare lost to the gap check leaves a backup -//! permanently short, since the gap never closes on its own. +//! short until the repair driver's next pass closes the gap. //! //! The harness removes an ambient `RUST_LOG` when this test supplies its explicit //! logging level, and the log oracle falls back from captured stdout to the @@ -98,9 +99,14 @@ const DEGRADED_MARKERS: [&str; 3] = [ ]; /// `IggyPartition::on_replicate`'s backup gap check. A re-dispatch that appends -/// behind an op already queued on the inbox surfaces here, and the dropped op -/// forces repair that correct redispatch ordering should never need. -const GAP_MARKER: &str = "dropping out-of-order prepare (gap)"; +/// behind an op already queued on the inbox surfaces here. The dropped op is +/// refetched by `tick_partitions`' level-triggered repair driver, but only after +/// its debounce interval, so a re-dispatch that trips this has already stalled +/// the replica for ~1s and the marker still means the ordering broke. +/// +/// Names its plane: the metadata plane logs its own gap drop at `warn`, which +/// passes this test's `info` filter, and the counting below matches substrings. +const GAP_MARKER: &str = "dropping out-of-order partition prepare (gap)"; const PARTITION_PLANE_FIELD: &str = "plane=\"partitions\""; fn topic_name(index: u32) -> String { @@ -347,7 +353,8 @@ fn assert_no_degraded_park_paths(harness: &TestHarness) { assert_eq!( partition_gaps, 0, "node {node} logged {GAP_MARKER:?}: a re-dispatched prepare lost its arrival \ - position and forced avoidable partition repair" + position, and the op it displaced is recoverable only by waiting out the \ + repair driver's debounce" ); } } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 3cbdf3e177..3190b6364e 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -140,6 +140,16 @@ where /// set when the recovery handshake finds this replica behind the group's /// commit frontier, cleared when `RepairDone` completes the walk. pub repair: Option, + /// Consecutive shard-sweep ticks this partition has been seen gap-stopped + /// (committed ops it cannot walk to, because the op at its commit frontier + /// plus one is missing). Debounces the sweep's level-triggered repair arm. + /// Owned by the sweep, like [`Self::transfer_rearm`]. + pub gap_ticks: u32, + /// Prepares the backup gap check destroyed since the shard last drained + /// the count. Replicated traffic has no client to answer and retransmit + /// skips ops that already reached quorum, so nothing else records that the + /// frame existed. + prepare_gap_drops: u64, /// Highest message offset recovered from segments at boot (`None` when /// the partition booted empty). Repaired batches at or below this line /// are already persisted and counted; the flush and commit paths skip @@ -489,6 +499,8 @@ where consumer_offset_enforce_fsync: false, runtime_options: TopicRuntimeOptions::default(), repair: None, + gap_ticks: 0, + prepare_gap_drops: 0, recovered_durable_offset: None, installed_frontier: None, fatal: None, @@ -1030,6 +1042,22 @@ where self.write_superblock(superblock.as_ref(), frontier).await } + /// Record one prepare destroyed by the backup gap check. Drained by the + /// shard sweep into `partition_prepare_gap_drops_total`; the counter is the + /// only production signal that the level-triggered repair driver has work + /// to do, since the drop itself answers nobody. + const fn note_prepare_gap_drop(&mut self) { + self.prepare_gap_drops = self.prepare_gap_drops.saturating_add(1); + } + + /// Take and clear the gap-drop count. + #[must_use = "dropping the count loses the only record those prepares existed"] + pub const fn take_prepare_gap_drops(&mut self) -> u64 { + let drops = self.prepare_gap_drops; + self.prepare_gap_drops = 0; + drops + } + /// Burn one transfer stall round; `true` once the budget is exhausted. /// Lives on the partition, not the session, so a re-minted session /// cannot reset it (see [`Self::transfer_attempts`]). @@ -2425,7 +2453,7 @@ where return; } - let journal_holds_op = self.log.journal().inner.header_by_op(header.op).is_some(); + let journal_holds_op = self.log.journal().inner.holds_op(header.op); if journal_holds_op { // Retransmit after downstream flap: durable here but commit // hasn't caught up. Re-forward + re-ACK so primary's view of @@ -2521,11 +2549,12 @@ where tracing::Level::WARN, &PartitionDiagEvent::new( self.diag_ctx(), - "dropping out-of-order prepare (gap)", + "dropping out-of-order partition prepare (gap)", ) .with_operation(header.operation) .with_op(header.op), ); + self.note_prepare_gap_drop(); return; } } else { diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index 45f3b6d1d8..aa6181810b 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -762,6 +762,20 @@ where headers.iter().find(|header| header.op == op).copied() } + /// Whether `op` is resident, without reading its header. + /// + /// Equivalent to `header_by_op(op).is_some()` and answers in O(log n) + /// instead of scanning: `append` writes `headers` and + /// `op_to_storage_offset` together and every clear site + /// ([`Self::commit`], [`Self::evict_prefix`], the restore path) clears + /// both, so the two populations cannot diverge. Callers that only need + /// presence must use this - a miss is the common case on the residency + /// checks, and a miss is exactly when the scan walks the whole vec. + pub fn holds_op(&self, op: u64) -> bool { + let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() }; + op_to_storage_offset.contains_key(&op) + } + /// Presence and message-carrying shape of the repair window `(floor, to_op]` /// in ONE pass over the header vec. /// diff --git a/core/server/config.toml b/core/server/config.toml index 5584b03788..6e64cf6cbf 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -656,6 +656,12 @@ view_probe_attempts_max = 5 # fire-and-forget over the lossy bus, so a session with no retry wedges forever # on a single dropped frame. Paces both the metadata and partition repair loops; # must be nonzero. +# +# Also how long a gap-stopped partition backup waits before OPENING a repair +# session, so shortening this makes the shard react to a replication hole +# sooner as well. That second use carries its own floor +# (PARTITION_GAP_DEBOUNCE_TICKS_MIN), because a value near the tick interval +# would otherwise arm repair against ordinary pipelining. repair_retry_interval = "1s" # Prepares a peer serves per repair round before the requester walks to the next diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 4db62a2b7a..a614bca3dc 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -109,9 +109,21 @@ //! `park_dropped` when it parked and then lost its namespace. A prepare has //! nobody to answer, so the counter is the only record it existed. //! +//! A shed or discarded *prepare* is not recovered by retransmit once its op has +//! reached quorum (`consensus::retransmit_targets` skips entries with +//! `ok_quorum_received`), so the backup gap-stops. `tick_partitions` opens a +//! repair session for it: its level-triggered detector arms once a partition has +//! been gap-stopped for the repair retry interval, independently of the +//! edge-triggered arming sites (`StartView` adoption, the commit heartbeat, the +//! post-transfer tail), whose edges a produce stream can starve. A repair range +//! the primary has already evicted escalates to partition state transfer. The park policy +//! above still shrinks the exposure to a genuinely exhausted byte budget and a +//! namespace this shard cannot serve; the driver bounds how long either costs, +//! and `partition_prepare_gap_drops_total` counts what reached the gap check. +//! //! # Known gaps //! -//! Recorded here because both were previously carried as a TODO on the +//! Recorded here because they were previously carried as a TODO on the //! materialization barrier this module used to promise, and the barrier is gone //! (see above) while these are not: //! @@ -3532,9 +3544,11 @@ mod tests { drop(inbox); } - /// The replicated-prepare shape, which no other test covers and where both - /// park critical are worst: a prepare has no client, so discarding it forces - /// the backup to wait for a later commit heartbeat and journal repair. + /// The replicated-prepare shape, which no other test covers and where the + /// park path's stakes are highest: a prepare has no client, so + /// `deny_parked_client_request` no-ops on it and anything that discards it + /// loses committed data silently, recoverable only once `tick_partitions`' + /// repair driver notices the gap it left. /// /// A backup receives the prepare before its own metadata commits (so the frame /// parks unstamped), then applies the commit and materialises. The prepare must @@ -3574,7 +3588,7 @@ mod tests { shard.redispatched_frame_count(), 1, "the parked prepare must be staged for re-dispatch; discarding it is \ - an avoidable gap, since a prepare has no client to retry it" + silent committed-data loss, since a prepare has no client to answer" ); let (served, answered) = drain_inbox(&inbox); assert_eq!( diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index fe4deaea5a..deeb4f5212 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -3211,8 +3211,9 @@ where /// buffer exists to absorb -- the partition primary materialises and /// replicates as soon as its own metadata commits, well before a lagging /// backup applies the same commit. Treating that as "prior incarnation" - /// destroys live traffic and forces the backup to recover a gap that the - /// park path could have delivered directly. + /// destroys live traffic: a replicated prepare has no client to answer, so + /// it would be dropped and the backup left gap-stopped until + /// `tick_partitions`' level-triggered driver notices and repairs it. /// The residual is unchanged from before the stamp existed -- a frame parked /// while the namespace was absent, then recreated under a new incarnation, /// is served against the replacement -- and closing it needs a wire-level @@ -3475,12 +3476,13 @@ where let existing = pending.get_mut(&namespace); let parked_len = existing.as_ref().map_or(0, |entry| entry.frames.len()); let namespace_bytes = existing.as_ref().map_or(0, |entry| entry.bytes); - // A prepare is never shed on a byte budget before the budget is spent. - // It has no client to retry it, and recovery requires a later commit - // heartbeat to expose the gap and arm same-view repair. A request costs - // only a retry, so it is refused the moment admitting it would cross a - // budget. This caps prepare residency at one frame of overshoot per - // budget (worst case + // A prepare is never shed on a byte budget. No client to answer, and + // recovery is slow: `consensus::retransmit_targets` skips an op that + // already reached quorum, so shedding one gap-stops the backup until + // `tick_partitions`' driver repairs it, where shedding a request costs + // one retry. A request is refused the moment admitting it + // would cross a budget; a prepare only once one is already spent. Caps + // prepare residency at one frame of overshoot per budget (worst case // `MAX_PARKED_BYTES` + `max_message_size`, 80 MiB per shard) instead of // at the budget, and is what makes an oversize frame parkable at all. let namespace_budget_spent = parked_len > 0 @@ -6866,9 +6868,27 @@ where // mid-sweep is seen on the next tick, the same latency a capped arm // already accepts. let mut transfers_inflight: Option = None; + // Repair sessions this sweep has opened, against + // `PARTITION_REPAIR_ARMS_PER_TICK_MAX`. + let mut repair_arms = 0usize; + // Commit walks this sweep has run, against + // `PARTITION_WALKS_PER_TICK_MAX`. + let mut walks = 0usize; let mut fatal: Option = None; for namespace in namespace_scratch.drain(..) { + { + // Ahead of the fence check and every `continue` below: the + // count is the only record those prepares existed, and a + // partition that fences here never ticks again. + let Some(partition) = partitions.get_mut_by_ns(&namespace) else { + continue; + }; + let gap_drops = partition.take_prepare_gap_drops(); + if gap_drops > 0 { + self.metrics.record_partition_prepare_gap_drops(gap_drops); + } + } let Some(partition) = partitions.get_by_ns(&namespace) else { continue; }; @@ -6996,6 +7016,92 @@ where .await; } + // Level-triggered gap detector. Every other partition arming site + // is edge-triggered and the edges are starvable: the commit-heartbeat + // backstop needs `CommitOutcome::Advanced`, and a follower has + // already advanced `commit_max` from each prepare header in + // `replicate_preflight` before the gap check dropped the prepare, so + // under produce load the heartbeat lands as `Accepted` and the gap + // wedges until an unrelated view change. Those edges stay the fast + // path; this is the ~1s floor under them. + let walk_stalled = { + let Some(partition) = partitions.get_mut_by_ns(&namespace) else { + continue; + }; + let probe = partition_gap_probe(partition); + let walk_stalled = partition_is_walk_stalled(&probe); + if drive_partition_gap_debounce( + &probe, + &mut partition.gap_ticks, + repair_retry_ticks, + repair_arms, + ) { + let consensus = partition.consensus(); + let self_id = consensus.replica(); + let peer = consensus.primary_index(consensus.view()); + let commit_min = consensus.commit_min(); + let commit_max = consensus.commit_max(); + if peer == self_id { + // The primary is its own repair source, so there is + // nobody to ask: the send would fail (no self entry in + // the replica registry) AFTER the session was recorded, + // and a session nothing can answer blocks this detector + // and every edge-triggered arming site until a view + // change. Restart the debounce so the warning repeats at + // its interval rather than every tick. + partition.gap_ticks = 0; + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + commit_min, + commit_max, + "partition primary is gap-stopped below its own commit frontier; \ + repair has no peer to request from" + ); + } else { + tracing::debug!( + shard = self.id, + namespace_raw = namespace.inner(), + commit_min, + commit_max, + peer, + recovery = ?probe.recovery, + "partition gap-stopped past the debounce; arming repair from the primary" + ); + self.maybe_request_partition_repair(partition, peer).await; + repair_arms += 1; + } + } + walk_stalled + }; + + // Capped like the repair arm, and for the same reason: a node-wide + // rejoin leaves every group on the shard walk-stalled in the same + // tick, and each walk reaches a segment flush. Undebounced, though + // -- the predicate guarantees the walk finds at least the next op + // (it and `collect_committable_from_journal` read the same journal + // index), so it cannot spin, and a skipped group is re-evaluated + // unchanged next tick. + if walk_stalled && walks < PARTITION_WALKS_PER_TICK_MAX { + let config = partitions.config(); + let Some(partition) = partitions.get_mut_by_ns(&namespace) else { + continue; + }; + let consensus = partition.consensus(); + // Debug, not info: an in-flight repair journals bodies without + // walking them, so this is the steady state for the whole + // duration of a rejoin and would be one line per group per tick. + tracing::debug!( + shard = self.id, + namespace_raw = namespace.inner(), + commit_min = consensus.commit_min(), + commit_max = consensus.commit_max(), + "partition commit walk parked over resident committed ops; resuming" + ); + partition.commit_journal(config).await; + walks += 1; + } + // Transfer stall retry: descriptor and chunk frames are // fire-and-forget, so a lost one must not wedge the session (and // the rejoin behind it) forever. Budget-bounded: a peer that died @@ -7889,6 +7995,15 @@ where if !consensus.is_normal() || consensus.is_transferring() || partition.repair.is_some() { return; } + // Never against self. The session is recorded below BEFORE the send, + // and a self-addressed `RequestPrepares` cannot be delivered (the + // replica registry holds no entry for this node), so the session would + // stand forever: `repair_finished` needs a `commit_min` only the reply + // can advance, the stall retry re-sends to the same peer, and + // `repair.is_some()` fences every other arming site meanwhile. + if peer == consensus.replica() { + return; + } // The window ends at the group head when suffix bodies are missing, // not at the commit point. A backup that adopted a StartView holds // suffix HEADERS above `commit_max` whose bodies it may never have @@ -7939,6 +8054,7 @@ where from_op, commit_to_op, fetch_to_op, + peer, "partition behind the group frontier; requesting repair" ); self.send_request_prepares( @@ -9202,6 +9318,174 @@ fn repair_serve_ceiling(requested_to_op: u64, commit_max: u64, head: u64) -> u64 requested_to_op.min(commit_max.max(head)) } +/// Repair sessions the partition tick sweep will OPEN per pass. +/// +/// A RATE cap, not a concurrency one: unlike +/// `IggyShard::PARTITION_TRANSFERS_INFLIGHT_MAX`, which counts live sessions +/// and refuses over the count, nothing bounds how many partition repair +/// sessions are live at once. This only spreads the opening cost, because one +/// arm is a `RequestPrepares` plus a repair stream the serving peer walks +/// synchronously and a node-wide gap (a rejoin, a lossy link) makes every group +/// on this shard due in the same tick. Over-cap groups stay due and arm on a +/// later pass. +const PARTITION_REPAIR_ARMS_PER_TICK_MAX: usize = 3; + +/// Commit walks the partition tick sweep will RUN per pass. +/// +/// Same correlated-fan-out argument as the repair arm, and the walk is the +/// costlier half: `commit_journal` reaches `commit_messages`, which flushes a +/// segment and fsyncs under `enforce_fsync`. Run serially over every group on +/// the shard in one tick body, a node-wide rejoin exceeds the view-change +/// escalation window the superblock pre-pass already chunks to stay inside. +/// +/// Capping cannot wedge a partition: the walk carries no debounce counter and +/// clears its own predicate (a walk either advances `commit_min` or fences the +/// partition), so a skipped group is re-evaluated unchanged on the next tick +/// and the eligible set drains in `ceil(groups / cap)` ticks. +const PARTITION_WALKS_PER_TICK_MAX: usize = 16; + +/// Floor under the gap detector's debounce, in ticks. +/// +/// The debounce reads `repair_retry_interval`, whose primary meaning is how +/// long a STALLED repair stream waits before re-requesting. `duration_to_ticks` +/// floors that at one tick, and one tick of lag is ordinary pipelining, so +/// without a floor of its own an operator shortening the retry interval would +/// also arm repair against a single reordered prepare. +const PARTITION_GAP_DEBOUNCE_TICKS_MIN: u32 = 50; + +/// What already owns a partition's recovery, if anything. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RecoveryOwner { + /// Nothing owns it: the sweep may arm repair. + Nobody, + Repair, + Transfer, + /// A scheduled transfer re-arm counting down. Arming repair over it would + /// defeat its backoff, as `arm_partition_transfer` documents. + TransferRearm, +} + +/// What the tick sweep reads off one partition to decide whether it is +/// gap-stopped. Split out so the guards, the debounce and the per-tick cap are +/// testable without a shard, a bus, or a journal. +#[derive(Debug, Clone, Copy)] +struct GapProbe { + normal: bool, + transferring: bool, + recovery: RecoveryOwner, + commit_min: u64, + commit_max: u64, + /// Whether `commit_min + 1` is resident in the local journal. + /// + /// Read only when the guards above already hold, and `false` otherwise: + /// both predicates test the lag first, so a probe that fails it is + /// answered without touching the journal at all. See + /// [`partition_gap_probe`]. + next_op_resident: bool, +} + +/// Whether this replica holds committed ops it cannot walk to, because the op +/// one past its commit frontier is missing from its journal. +/// +/// The journal-hole half is not redundant: a follower advances `commit_max` +/// from every prepare header in `replicate_preflight`, so `commit_min < +/// commit_max` is transiently true on every healthy pipelined tick and a bare +/// lag test would arm repair against ordinary produce. +const fn partition_is_gap_stopped(probe: &GapProbe) -> bool { + probe.normal + && !probe.transferring + && matches!(probe.recovery, RecoveryOwner::Nobody) + && probe.commit_min < probe.commit_max + && !probe.next_op_resident +} + +/// The gap predicate's disjoint sibling, not its complement: everything the +/// walk needs is resident, it just never ran (a heartbeat carrying a known +/// commit is `Accepted`, and an idle group offers no other edge). The two split +/// on `next_op_resident`, so they cannot both hold, but both are false whenever +/// a shared guard fails. Not gated on `RecoveryOwner`: repair fetches bodies +/// without walking them, so gating parks the walk all session. +const fn partition_is_walk_stalled(probe: &GapProbe) -> bool { + probe.normal + && !probe.transferring + && probe.commit_min < probe.commit_max + && probe.next_op_resident +} + +/// Count one sweep tick against `gap_ticks` and answer whether this partition +/// may arm repair now. +/// +/// Level-triggered, because every edge-triggered arming site is starvable: the +/// commit-heartbeat backstop fires only on `CommitOutcome::Advanced`, and under +/// sustained produce the prepares consume the advance in preflight before the +/// gap check drops them, so the heartbeat lands as `Accepted` and the gap wedges +/// until an unrelated view change. +/// +/// A capped-out arm keeps its debounce satisfied rather than starting over, so +/// the group arms on the next pass with a slot free. +const fn drive_partition_gap_debounce( + probe: &GapProbe, + gap_ticks: &mut u32, + debounce_ticks: u32, + arms_this_tick: usize, +) -> bool { + if !partition_is_gap_stopped(probe) { + *gap_ticks = 0; + return false; + } + let debounce_ticks = if debounce_ticks < PARTITION_GAP_DEBOUNCE_TICKS_MIN { + PARTITION_GAP_DEBOUNCE_TICKS_MIN + } else { + debounce_ticks + }; + *gap_ticks = gap_ticks.saturating_add(1); + *gap_ticks >= debounce_ticks && arms_this_tick < PARTITION_REPAIR_ARMS_PER_TICK_MAX +} + +/// Read the gap probe off a live partition. +fn partition_gap_probe(partition: &IggyPartition) -> GapProbe +where + B: MessageBus, + SB: SuperblockStore, +{ + let consensus = partition.consensus(); + let commit_min = consensus.commit_min(); + let commit_max = consensus.commit_max(); + // Transfer first: it supersedes repair, so naming it is the truthful + // diagnostic when both happen to be set. + let recovery = if partition.transfer.is_some() { + RecoveryOwner::Transfer + } else if partition.transfer_rearm.is_some() { + RecoveryOwner::TransferRearm + } else if partition.repair.is_some() { + RecoveryOwner::Repair + } else { + RecoveryOwner::Nobody + }; + let normal = consensus.is_normal(); + let transferring = consensus.is_transferring(); + // Residency last, and only once the guards both predicates share already + // hold. This runs for every group on the shard on every tick, and the + // caught-up steady state (`commit_min == commit_max`) would otherwise pay + // a journal lookup whose answer both predicates discard. + let next_op_resident = normal + && !transferring + && commit_min < commit_max + && partition + .log + .journal() + .inner + .holds_op(commit_min.saturating_add(1)); + GapProbe { + normal, + transferring, + recovery, + commit_min, + commit_max, + next_op_resident, + } +} + /// Whether the parked `StartView` log names every op in the uncommitted /// suffix `(commit_max, head]`, in descending order. Only this canonical list /// makes fetching bodies above the commit point safe. @@ -10435,3 +10719,283 @@ mod superblock_fail_stop_tests { assert!(superblock_wedged(121, 120)); } } + +#[cfg(test)] +mod gap_detector_tests { + //! The level-triggered repair arm the partition tick sweep runs. + //! + //! Its whole reason to exist is that the edge-triggered arming sites are + //! starvable, so the guards it shares with them and the debounce that keeps + //! it off healthy traffic are the parts worth pinning. + + use super::{ + GapProbe, PARTITION_GAP_DEBOUNCE_TICKS_MIN, PARTITION_REPAIR_ARMS_PER_TICK_MAX, + RecoveryOwner, drive_partition_gap_debounce, partition_is_gap_stopped, + partition_is_walk_stalled, + }; + + const DEBOUNCE: u32 = 100; + + /// A gap-stopped follower: committed through op 10, walkable only to 5, + /// because op 6 is not in its journal. + const fn gap_stopped() -> GapProbe { + GapProbe { + normal: true, + transferring: false, + recovery: RecoveryOwner::Nobody, + commit_min: 5, + commit_max: 10, + next_op_resident: false, + } + } + + /// A walk-stalled follower: the same lag, but op 6 IS in its journal, so + /// nothing needs fetching and the walk just has to run. + fn walk_stalled() -> GapProbe { + GapProbe { + next_op_resident: true, + ..gap_stopped() + } + } + + #[test] + fn given_a_lagging_follower_with_the_next_op_resident_when_probed_should_not_be_gap_stopped() { + // The half that keeps the predicate honest. A follower advances + // commit_max from every prepare header in preflight, so commit_min < + // commit_max is transiently true on any pipelined tick; without the + // journal-hole test the driver would request repair against ordinary + // produce, on every partition, forever. + let healthy = GapProbe { + next_op_resident: true, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&healthy)); + assert!(partition_is_gap_stopped(&gap_stopped())); + } + + #[test] + fn given_a_caught_up_follower_when_probed_should_not_be_gap_stopped() { + let caught_up = GapProbe { + commit_min: 10, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&caught_up)); + } + + #[test] + fn given_a_replica_outside_normal_status_when_probed_should_not_be_gap_stopped() { + // A view change owns the log while it runs, and `maybe_request_partition_repair` + // refuses outside Normal anyway; arming here would only burn a nonce. + let electing = GapProbe { + normal: false, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&electing)); + + let installing = GapProbe { + transferring: true, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&installing)); + } + + #[test] + fn given_recovery_already_owned_when_probed_should_not_be_gap_stopped() { + for owner in [ + RecoveryOwner::Repair, + RecoveryOwner::Transfer, + RecoveryOwner::TransferRearm, + ] { + let owned = GapProbe { + recovery: owner, + ..gap_stopped() + }; + assert!( + !partition_is_gap_stopped(&owned), + "{owner:?} owns the recovery; a second session would race it" + ); + } + } + + #[test] + fn given_a_gap_stopped_follower_when_debouncing_should_arm_only_at_the_threshold() { + let probe = gap_stopped(); + let mut gap_ticks = 0; + for tick in 1..DEBOUNCE { + assert!( + !drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, 0), + "armed at tick {tick}, before the debounce elapsed" + ); + } + assert!(drive_partition_gap_debounce( + &probe, + &mut gap_ticks, + DEBOUNCE, + 0 + )); + } + + #[test] + fn given_a_debounce_in_progress_when_the_gap_closes_should_reset_the_counter() { + let stopped = gap_stopped(); + let walkable = GapProbe { + next_op_resident: true, + ..stopped + }; + let mut gap_ticks = 0; + for _ in 0..DEBOUNCE - 1 { + drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0); + } + assert_eq!(gap_ticks, DEBOUNCE - 1); + + assert!(!drive_partition_gap_debounce( + &walkable, + &mut gap_ticks, + DEBOUNCE, + 0 + )); + assert_eq!(gap_ticks, 0, "progress must restart the debounce"); + assert!( + !drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0), + "a fresh gap must serve its own debounce, not inherit the old count" + ); + } + + #[test] + fn given_a_follower_with_resident_committed_ops_when_probed_should_be_walk_stalled() { + assert!(partition_is_walk_stalled(&walk_stalled())); + assert!( + !partition_is_walk_stalled(&gap_stopped()), + "a missing next op is repair's job; a walk over it would stop dead" + ); + } + + #[test] + fn given_a_caught_up_follower_when_probed_should_not_be_walk_stalled() { + let caught_up = GapProbe { + commit_min: 10, + ..walk_stalled() + }; + assert!(!partition_is_walk_stalled(&caught_up)); + } + + #[test] + fn given_a_replica_outside_normal_status_when_probed_should_not_be_walk_stalled() { + let electing = GapProbe { + normal: false, + ..walk_stalled() + }; + assert!(!partition_is_walk_stalled(&electing)); + + // Same gate as the on-commit arm: a walk during a transfer can advance + // commit_min past the incoming frontier. + let installing = GapProbe { + transferring: true, + ..walk_stalled() + }; + assert!(!partition_is_walk_stalled(&installing)); + } + + #[test] + fn given_recovery_already_owned_when_the_next_op_is_resident_should_still_be_walk_stalled() { + // Deliberate: `apply_repaired_prepare` journals without walking, so a + // gated walk would sit parked for the whole session while the resident + // prefix is already applicable. + for owner in [ + RecoveryOwner::Repair, + RecoveryOwner::Transfer, + RecoveryOwner::TransferRearm, + ] { + let owned = GapProbe { + recovery: owner, + ..walk_stalled() + }; + assert!( + partition_is_walk_stalled(&owned), + "{owner:?} owns the fetch, not the resident prefix" + ); + } + } + + #[test] + fn given_any_probe_when_evaluated_should_never_be_both_gap_stopped_and_walk_stalled() { + // The two halves split on `next_op_resident`; if they ever overlap, one + // tick both arms repair and walks the window it is fetching. + let owners = [ + RecoveryOwner::Nobody, + RecoveryOwner::Repair, + RecoveryOwner::Transfer, + RecoveryOwner::TransferRearm, + ]; + for normal in [false, true] { + for transferring in [false, true] { + for recovery in owners { + for (commit_min, commit_max) in [(5, 10), (10, 10)] { + for next_op_resident in [false, true] { + let probe = GapProbe { + normal, + transferring, + recovery, + commit_min, + commit_max, + next_op_resident, + }; + assert!( + !(partition_is_gap_stopped(&probe) + && partition_is_walk_stalled(&probe)), + "both predicates claim {probe:?}" + ); + } + } + } + } + } + } + + #[test] + fn given_the_per_tick_cap_reached_when_debouncing_should_defer_without_losing_the_debounce() { + let probe = gap_stopped(); + let mut gap_ticks = DEBOUNCE; + assert!( + !drive_partition_gap_debounce( + &probe, + &mut gap_ticks, + DEBOUNCE, + PARTITION_REPAIR_ARMS_PER_TICK_MAX + ), + "the cap must refuse the arm" + ); + assert!( + gap_ticks > DEBOUNCE, + "a capped-out group stays due; restarting its debounce would push the \ + arm a whole interval out per contended tick" + ); + assert!( + drive_partition_gap_debounce( + &probe, + &mut gap_ticks, + DEBOUNCE, + PARTITION_REPAIR_ARMS_PER_TICK_MAX - 1 + ), + "the same group arms on the next pass with a slot free" + ); + } + + #[test] + fn given_a_debounce_shorter_than_the_floor_when_driven_should_hold_until_the_floor() { + // `repair_retry_interval` is an operator knob whose primary meaning is + // the stalled-stream retry, and `duration_to_ticks` floors it at one + // tick. One tick of lag is ordinary pipelining, so without a floor of + // its own a shortened retry interval would arm repair against a single + // reordered prepare. + let probe = gap_stopped(); + let mut gap_ticks = 0; + for tick in 1..PARTITION_GAP_DEBOUNCE_TICKS_MIN { + assert!( + !drive_partition_gap_debounce(&probe, &mut gap_ticks, 1, 0), + "a 1-tick debounce armed at tick {tick}, under the floor" + ); + } + assert!(drive_partition_gap_debounce(&probe, &mut gap_ticks, 1, 0)); + } +} diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 4fbd8c7fcf..005caa7820 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -80,11 +80,14 @@ pub struct FrameDropLabel { /// (`reason=park_overflow`), a parked frame retired with no client to answer /// (`reason=park_dropped`), an incarnation rejection, or a routing send the /// target inbox refused. A shed client request is answered with a retriable -/// status. A shed prepare may no longer be covered by retransmit once its op -/// reached quorum, but a later `CommitMessage` that advances the backup's -/// frontier arms same-view journal repair. If the primary evicted the range, -/// repair escalates to partition state transfer. The counter therefore signals -/// a data-plane gap or recovery burden, not a requirement for a view change. +/// status, so the client recovers. A shed *prepare* has nobody to answer and is +/// not covered by retransmit once its op has reached quorum +/// (`consensus::retransmit_targets` skips `ok_quorum_received`), so the backup +/// gap-stops; `tick_partitions`' sweep is what repairs it, escalating to +/// partition state transfer when the primary has evicted the range, and +/// `partition_prepare_gap_drops_total` is what counts the prepares that reached +/// the gap check. The counter therefore signals a data-plane gap or recovery +/// burden, not a requirement for a view change. pub mod frame_drop_variant { pub const CONSENSUS: &str = "consensus"; pub const FD_TRANSFER: &str = "fd_transfer"; @@ -208,6 +211,7 @@ pub struct ShardMetrics { partition_frames_rejected_ahead_total: Counter, partition_requests_denied_transient_total: Counter, partition_repair_serves_deferred_purge_total: Counter, + partition_prepare_gap_drops_total: Counter, } impl ShardMetrics { @@ -232,6 +236,7 @@ impl ShardMetrics { partition_frames_rejected_ahead_total: Counter::default(), partition_requests_denied_transient_total: Counter::default(), partition_repair_serves_deferred_purge_total: Counter::default(), + partition_prepare_gap_drops_total: Counter::default(), } } @@ -405,6 +410,31 @@ impl ShardMetrics { self.partition_repair_serves_deferred_purge_total.get() } + /// Add the prepares a partition's backup gap check destroyed since the last + /// sweep, drained per tick from `IggyPartition::take_prepare_gap_drops`. + /// + /// Deliberately NOT a `frame_drops_total{variant=partition}` reason: that + /// family means the bus or the router shed a frame, and the simulator + /// asserts it stays at zero on runs with no injected loss. A gap drop is a + /// protocol-ordering drop that any real loss produces, and the sweep repairs + /// it, so folding the two would turn a routing-fault alert into noise. + /// + /// Shard-scoped, with no namespace label: a server runs hundreds of groups + /// per shard, so labelling by namespace is unbounded cardinality, and every + /// other partition counter in this file is shard-scoped for the same + /// reason. The per-group detail is in the arm's log line. The metadata + /// plane's own gap drop is NOT counted here - it has its own recovery path. + pub fn record_partition_prepare_gap_drops(&self, drops: u64) { + self.partition_prepare_gap_drops_total.inc_by(drops); + } + + /// Snapshot of `partition_prepare_gap_drops_total`. Test/simulator accessor. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn partition_prepare_gap_drops_value(&self) -> u64 { + self.partition_prepare_gap_drops_total.get() + } + /// Snapshot of `partition_frames_rejected_stale_total`. Test/simulator /// accessor, readable from any crate under those cfgs so the crates that /// drive the reconciler can assert a reject did not happen. @@ -496,6 +526,11 @@ impl ShardMetrics { "partition repair serves or completions deferred until a committed purge applies", self.partition_repair_serves_deferred_purge_total.clone(), ); + registry.register( + "partition_prepare_gap_drops", + "replicated prepares dropped out of order by a backup's gap check", + self.partition_prepare_gap_drops_total.clone(), + ); } } diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 1f4fa01170..4a3a0d2149 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -379,6 +379,8 @@ where self.process_frame(frame).await; self.process_loopback(&mut loopback_buf, &mut namespace_scratch).await; // Tail drain catches reconcile ops whose marker was dropped. + // Anything it stages is served by the arm above + // on the next pass, before this arm can run again. self.apply_reconcile_ops(); } // Guaranteed reply-lane service: `select_biased!` diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 3d2fd3feaa..b6eef2050e 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -5097,3 +5097,678 @@ mod repair_frontier_tests { ); } } + +#[cfg(test)] +mod partition_repair_driver_tests { + //! A backup that missed a committed partition prepare recovers in Normal + //! status, without waiting for a view change. + //! + //! Every edge-triggered arming site is starvable. A follower advances + //! `commit_max` from each prepare header in `replicate_preflight`, before + //! the gap check drops the prepare, so under produce the primary's commit + //! heartbeat lands as `CommitOutcome::Accepted` and the backstop that would + //! arm repair on `Advanced` never runs. These tests starve that edge + //! outright (no commit heartbeat for the group reaches the lagging replica) + //! so nothing but the level-triggered detector in `tick_partitions` can + //! close the gap. + + use super::*; + use bytes::Bytes; + use consensus::Status; + use iggy_binary_protocol::{ + CommitHeader, PrepareHeader, RepairRangeReplyHeader, RequestPreparesHeader, + }; + use packet::Packet; + use std::sync::atomic::{AtomicU64, Ordering}; + + /// Chain replication runs 0 -> 1 -> 2 and stops before the primary, so + /// replica 2 forwards to nobody: it is the only replica whose losses do not + /// also starve its successor, and therefore the group, of quorum. + const LAGGING: u8 = 2; + + const CLIENT_ID: u128 = 1; + + /// Ops that replicate cleanly before the fault, so the gap opens above a + /// committed prefix rather than at the group's first op. + const WARMUP_SENDS: usize = 3; + + /// Ticks stepped after each produce. Keeps one send's round trip inside its + /// own window so the workload is legible tick by tick. + const STEPS_PER_SEND: usize = 12; + + /// Produces issued with the fault standing. `partitions::REPAIR_RETRY_TICKS` + /// is 100, so this must carry the run past the debounce while prepares keep + /// consuming the `commit_max` advance the heartbeat backstop needs. + const GAP_SENDS: usize = 12; + + /// Quiet ticks after the produce stops, for the repair stream to land. + /// + /// Budgeted against `NORMAL_HEARTBEAT_TICKS` (500): with this group's commit + /// heartbeats withheld, the lagging replica elects once that timer fires, and + /// an election would heal the gap through `on_start_view` instead. Every + /// phase after the fault is installed has to fit inside it. + const QUIET_STEPS: usize = 160; + + /// Budget for the group to settle once the fault is lifted. Generous rather + /// than tuned: the drain loop breaks on convergence, and the tail above the + /// repaired window waits on whichever interval-driven site picks it up. + const DRAIN_STEPS: usize = 600; + + /// Ticks of healthy load in the no-false-positive test, several debounce + /// intervals' worth so the sweep gets many chances to arm. + const LOAD_TICKS: usize = 4 * partitions::REPAIR_RETRY_TICKS as usize; + + /// Paced at a round trip rather than one submit per tick: the pipeline caps + /// at `PIPELINE_PREPARE_QUEUE_MAX`, so submitting faster than the group + /// commits just collects transient rejections and the group goes quiet. + const TICKS_PER_SEND: usize = 4; + + /// What the healthy run saw of the backup's commit lag, sampled per tick. + /// + /// Lag alone is what a naive detector reads as a gap, so the run has to + /// record how much of it there was AND whether the op past the frontier was + /// resident each time, rather than assert the lag away. + #[derive(Default)] + struct LagObservations { + samples: u32, + resident_samples: u32, + longest_run: u32, + current_run: u32, + } + + impl LagObservations { + /// Fold one tick's view of `replica`'s commit frontier. + fn observe(&mut self, sim: &Simulator, replica: u8, namespace: IggyNamespace) { + let (_, _, commit_min, commit_max) = group_state(sim, replica, namespace); + if commit_min >= commit_max { + self.current_run = 0; + return; + } + self.current_run += 1; + self.longest_run = self.longest_run.max(self.current_run); + self.samples += 1; + if journal_holds(sim, replica, namespace, commit_min + 1) { + self.resident_samples += 1; + } + } + } + + /// Ops the healthy run must have committed for its verdict to mean anything. + /// A produce's round trip is four one-way hops plus tick granularity, so this + /// network commits on the order of one op per fifteen ticks however hard the + /// client pushes; the floor only has to prove the group was live across + /// several debounce intervals, not that it was saturated. + const COMMITTED_MIN: u64 = 20; + + /// Produces issued with the fault standing in the walk-starvation test: few + /// enough that the gap debounce fires only after produce stops, so the + /// repair window closes at the last carried commit and lands fully resident. + const STRAND_SENDS: usize = 3; + + /// Quiet budget for the walk-starvation test: debounce, repair stream, then + /// the drain, kept under `NORMAL_HEARTBEAT_TICKS` so an election cannot be + /// the healer. + const STRAND_QUIET_STEPS: usize = 300; + + /// The partition-plane prepare a packet carries, if it carries one for + /// `group`. + fn prepare_for(packet: &Packet, group: u64) -> Option { + if packet.message.header().command != Command::Prepare { + return None; + } + let header: &PrepareHeader = + bytemuck::checked::from_bytes(&packet.message.as_slice()[..size_of::()]); + (header.group == group).then_some(*header) + } + + /// Whether a packet is a commit heartbeat for `group`. + fn is_commit_for(packet: &Packet, group: u64) -> bool { + if packet.message.header().command != Command::Commit { + return false; + } + let header: &CommitHeader = + bytemuck::checked::from_bytes(&packet.message.as_slice()[..size_of::()]); + header.group == group + } + + /// Whether a packet is a repair request for `group`. + fn is_request_prepares_for(packet: &Packet, group: u64) -> bool { + if packet.message.header().command != Command::RequestPrepares { + return false; + } + let header: &RequestPreparesHeader = bytemuck::checked::from_bytes( + &packet.message.as_slice()[..size_of::()], + ); + header.group == group + } + + /// Whether a packet is a repair stream terminator for `group`. + fn is_repair_done_for(packet: &Packet, group: u64) -> bool { + if packet.message.header().command != Command::RepairDone { + return false; + } + let header: &RepairRangeReplyHeader = bytemuck::checked::from_bytes( + &packet.message.as_slice()[..size_of::()], + ); + header.group == group + } + + fn cluster(seed: u64) -> (Simulator, SimClient) { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let replica_count: u8 = 3; + let network_opts = packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed, + ..packet::PacketSimulatorOptions::default() + }; + let sim = Simulator::new( + replica_count as usize, + std::iter::once(CLIENT_ID), + network_opts, + ); + (sim, SimClient::new(CLIENT_ID)) + } + + /// Submit `sends` produces against `namespace`, stepping between each. + fn produce( + sim: &mut Simulator, + client: &SimClient, + namespace: IggyNamespace, + sends: usize, + tag: &str, + ) { + for index in 0..sends { + let msg = client.send_messages(namespace, &[Bytes::from(format!("{tag}-{index}"))]); + sim.submit_request(client.client_id(), 0, msg.into_generic()); + for _ in 0..STEPS_PER_SEND { + sim.step(); + } + } + } + + /// `(status, view, commit_min, commit_max)` of one replica's partition group. + fn group_state( + sim: &Simulator, + replica: u8, + namespace: IggyNamespace, + ) -> (Status, u32, u64, u64) { + let shard = sim.replicas[replica as usize].partition_shard(namespace); + let partition = shard + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the replica hosts the group"); + let consensus = partition.consensus(); + ( + consensus.status(), + consensus.view(), + consensus.commit_min(), + consensus.commit_max(), + ) + } + + fn journal_holds(sim: &Simulator, replica: u8, namespace: IggyNamespace, op: u64) -> bool { + let shard = sim.replicas[replica as usize].partition_shard(namespace); + shard + .plane + .partitions() + .get_by_ns(&namespace) + .is_some_and(|partition| partition.log.journal().inner.header_by_op(op).is_some()) + } + + fn gap_drops(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> u64 { + sim.replicas[replica as usize] + .partition_shard(namespace) + .metrics() + .partition_prepare_gap_drops_value() + } + + fn transfer_armed(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> bool { + let shard = sim.replicas[replica as usize].partition_shard(namespace); + shard + .plane + .partitions() + .get_by_ns(&namespace) + .is_some_and(|partition| { + partition.transfer.is_some() + || partition.consensus().state_transfer_stage() + != consensus::StateTransferStage::Idle + }) + } + + #[test] + fn given_a_backup_that_dropped_a_committed_prepare_when_heartbeat_advances_are_starved_should_repair_in_normal_status() + { + // Statics, not captures: the link hooks are bare `fn` pointers. Declared + // inside the test because the sibling tests in this binary run in + // parallel and would otherwise share them. + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + + /// Chain link 1 -> 2: swallow the first partition prepare, once. + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, GAP_NS.load(Ordering::Relaxed)) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + /// Primary -> 2: withhold this group's commit heartbeats, so the + /// `Advanced` backstop can never run, and withhold retransmits of the + /// dropped op. The retransmit half stands in for production behaviour + /// rather than adding a fault: `consensus::retransmit_targets` skips an + /// op that already reached quorum, and this op reaches quorum on 0 and 1 + /// alone. Without it the retry timer could heal the gap and the test + /// would pass with no repair driver at all. + fn starve_commit_edge(packet: &Packet) -> bool { + let group = GAP_NS.load(Ordering::Relaxed); + if let Some(header) = prepare_for(packet, group) { + return header.op == WITHHELD_OP.load(Ordering::Relaxed); + } + is_commit_for(packet, group) + } + + let (mut sim, client) = cluster(0x5EED_0232); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + WITHHELD_OP.store(0, Ordering::Relaxed); + + produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup"); + let (_, _, warm_commit_min, _) = group_state(&sim, LAGGING, namespace); + assert!( + warm_commit_min > 0, + "the lagging replica committed nothing before the fault, so the gap \ + below would open at the group's first op" + ); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_commit_edge); + + produce(&mut sim, &client, namespace, GAP_SENDS, "gap"); + + let withheld = WITHHELD_OP.load(Ordering::Relaxed); + assert_ne!( + withheld, 0, + "no partition prepare crossed the chain link, so the fault never armed" + ); + assert!( + gap_drops(&sim, LAGGING, namespace) > 0, + "the lagging replica never reached its backup gap check, so the \ + prepares after the withheld op were not dropped as a gap" + ); + + for _ in 0..QUIET_STEPS { + sim.step(); + } + + // Judged with the blockade still standing, so the tick driver is the only + // thing that can have armed the repair: no commit heartbeat for this + // group has reached the replica since the gap opened, and `on_commit` is + // where the `Advanced` backstop lives. + let (status, view, commit_min, _) = group_state(&sim, LAGGING, namespace); + assert_eq!( + view, 0, + "a view change healed the gap instead of the repair driver; the test \ + proves nothing about normal status" + ); + assert_eq!(status, Status::Normal, "the replica left Normal status"); + assert!( + journal_holds(&sim, LAGGING, namespace, withheld), + "op {withheld} was never repaired back into the lagging replica's journal" + ); + assert!( + commit_min >= withheld, + "the commit walk never crossed the repaired hole: stopped at \ + {commit_min}, the withheld op is {withheld}" + ); + + // Lift the blockade and let the group settle. The commit walk is driven + // by arriving frames, so with this group's heartbeats withheld the tail + // above the repaired window has nothing to advance it; that is the + // injected fault, not the gap under test. + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = None; + for _ in 0..DRAIN_STEPS { + sim.step(); + let (_, _, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + if commit_min == commit_max { + break; + } + } + let (status, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + assert_eq!((status, view), (Status::Normal, 0)); + assert_eq!( + commit_min, commit_max, + "the lagging replica is still gap-stopped: committed through \ + {commit_max} but walkable only to {commit_min}" + ); + } + + #[test] + fn given_a_repair_armed_by_the_tick_driver_when_the_range_is_evicted_should_convert_to_state_transfer() + { + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, GAP_NS.load(Ordering::Relaxed)) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + fn starve_commit_edge(packet: &Packet) -> bool { + let group = GAP_NS.load(Ordering::Relaxed); + if let Some(header) = prepare_for(packet, group) { + return header.op == WITHHELD_OP.load(Ordering::Relaxed); + } + is_commit_for(packet, group) + } + + let (mut sim, client) = cluster(0x5EED_0233); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + WITHHELD_OP.store(0, Ordering::Relaxed); + + produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup"); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_commit_edge); + + produce(&mut sim, &client, namespace, GAP_SENDS, "gap"); + assert_ne!( + WITHHELD_OP.load(Ordering::Relaxed), + 0, + "no partition prepare crossed the chain link, so the fault never armed" + ); + + // Compact the serving side past the gap. This plane's journal is + // memory-only, so wiping it IS its retention floor moving: the serve + // path reads `repair_retained_from` as `None` and reports eviction from + // its own commit frontier, which is exactly what a peer that + // checkpointed past the requested window answers. + { + let primary = sim.replicas[0].partition_shard(namespace); + let partition = primary + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the primary hosts the group"); + partition.log.journal().inner.clear_all(); + } + + for _ in 0..QUIET_STEPS { + sim.step(); + if transfer_armed(&sim, LAGGING, namespace) { + break; + } + } + + let (status, view, ..) = group_state(&sim, LAGGING, namespace); + assert_eq!( + view, 0, + "a view change armed the recovery instead of the tick-armed repair session" + ); + assert!( + transfer_armed(&sim, LAGGING, namespace), + "the tick-armed repair session hit an evicted range but never converted \ + to a state transfer (status {status:?})" + ); + } + + #[test] + fn given_healthy_pipelined_traffic_when_no_gap_exists_should_not_arm_repair() { + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static REPAIR_REQUESTS: AtomicU64 = AtomicU64::new(0); + + /// Observer, not a fault: counts this group's repair requests and passes + /// every packet through. + fn count_repair_requests(packet: &Packet) -> bool { + if is_request_prepares_for(packet, GAP_NS.load(Ordering::Relaxed)) { + REPAIR_REQUESTS.fetch_add(1, Ordering::Relaxed); + } + false + } + + // TWO replicas, so quorum spans both: no op can commit without the + // backup's ack, every reordering-induced gap therefore blocks quorum, and + // `consensus::retransmit_targets` refills it. At three, the network's + // per-tick delivery shuffle lets an op commit on the primary and its + // first chain hop while the last replica loses it for good, which is the + // very fault the sibling tests inject -- it would then be repaired here, + // correctly, and this assertion would fire on a healthy driver. + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let replica_count: u8 = 2; + let mut sim = Simulator::new( + replica_count as usize, + std::iter::once(CLIENT_ID), + packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed: 0x5EED_0234, + ..packet::PacketSimulatorOptions::default() + }, + ); + let client = SimClient::new(CLIENT_ID); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + REPAIR_REQUESTS.store(0, Ordering::Relaxed); + + for (from, to) in [(0u8, 1u8), (1, 0)] { + *sim.network + .link_drop_packet_fn(ProcessId::Replica(from), ProcessId::Replica(to)) = + Some(count_repair_requests); + } + + // Sustained, not bursty: a produce every tick, for several debounce + // intervals, so the sweep gets many chances to arm against ordinary + // pipelining. The backup's lag is sampled per tick, and the run asserts + // it went somewhere, so a green result cannot come from a workload that + // never loaded the group. + let mut observed = LagObservations::default(); + for tick in 0..LOAD_TICKS { + if tick % TICKS_PER_SEND == 0 { + let msg = + client.send_messages(namespace, &[Bytes::from(format!("healthy-{tick}"))]); + sim.submit_request(client.client_id(), 0, msg.into_generic()); + } + sim.step(); + observed.observe(&sim, 1, namespace); + } + for _ in 0..QUIET_STEPS { + sim.step(); + observed.observe(&sim, 1, namespace); + } + + let committed = group_state(&sim, 1, namespace).2; + let sends = LOAD_TICKS / TICKS_PER_SEND; + assert!( + committed >= COMMITTED_MIN, + "the backup committed only {committed} ops across {sends} sends, so the \ + sweep was never driven over a loaded group" + ); + // Lag is RECORDED, not forbidden. Asserting it away would make the + // repair-request assertion below unfalsifiable, since + // `partition_is_gap_stopped` needs `commit_min < commit_max` to fire at + // all. What must hold is that every lag this run saw was backed by a + // RESIDENT next op -- a walk waiting to run, not a hole -- which is the + // distinction a naive detector would miss. + // + // On a two-replica group the backup keeps pace and the tick's own + // walk-stalled backstop drains what little lag a step leaves, so this + // usually samples none and the check costs nothing. It is the guard for + // when that stops being true, not the proof: the predicate itself is + // pinned exhaustively by `gap_detector_tests`, and what this run adds is + // that the DRIVER sends no repair over a live, loaded group. + assert_eq!( + observed.samples, + observed.resident_samples, + "the backup lagged with a MISSING next op on {} of {} sampled ticks, so a \ + no-loss two-replica run produced a real hole and the repair assertion \ + below would be testing loss recovery instead of false positives", + observed.samples - observed.resident_samples, + observed.samples + ); + for replica in 0..replica_count { + let (status, view, commit_min, commit_max) = group_state(&sim, replica, namespace); + assert_eq!( + (status, view), + (Status::Normal, 0), + "replica {replica} left view 0 / Normal, so a view change could \ + account for repair traffic" + ); + // Residual lag is transient at worst: a heartbeat carrying a commit + // the replica already knows is `CommitOutcome::Accepted`, so a backup + // can sit with committed-but-unwalked ops until the tick sweep's + // walk-stalled backstop resumes the walk. Every one of them is + // RESIDENT, which is what separates it from a gap. + if commit_min < commit_max { + assert!( + journal_holds(&sim, replica, namespace, commit_min + 1), + "replica {replica} lags at {commit_min} of {commit_max} with op \ + {} missing, so a no-loss run produced a real hole", + commit_min + 1 + ); + } + } + assert_eq!( + REPAIR_REQUESTS.load(Ordering::Relaxed), + 0, + "the tick driver requested repair on a healthy group; its gap predicate \ + is reading ordinary commit lag as a journal hole (backup lagged on {} of \ + the sampled ticks, longest run {})", + observed.samples, + observed.longest_run + ); + } + + #[test] + fn given_a_backup_holding_resident_committed_ops_when_every_walk_edge_is_starved_should_drain_in_normal_status() + { + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + static WITHHELD_DONES: AtomicU64 = AtomicU64::new(0); + + /// Chain link 1 -> 2: swallow the first partition prepare, once. + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, GAP_NS.load(Ordering::Relaxed)) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + /// Primary -> 2: withhold every direct prepare (live ones ride the + /// chain, so this starves only retransmit heals), the group's commit + /// heartbeats, and its repair terminators. The repaired ops themselves + /// pass, so the window lands resident while `complete_repair`, the walk + /// the terminator would run, never fires. + fn starve_walk_edges(packet: &Packet) -> bool { + let group = GAP_NS.load(Ordering::Relaxed); + if prepare_for(packet, group).is_some() { + return true; + } + if is_repair_done_for(packet, group) { + WITHHELD_DONES.fetch_add(1, Ordering::Relaxed); + return true; + } + is_commit_for(packet, group) + } + + let (mut sim, client) = cluster(0x5EED_0235); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + WITHHELD_OP.store(0, Ordering::Relaxed); + WITHHELD_DONES.store(0, Ordering::Relaxed); + + produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup"); + let (_, _, warm_commit_min, _) = group_state(&sim, LAGGING, namespace); + assert!( + warm_commit_min > 0, + "the lagging replica committed nothing before the fault, so the gap \ + below would open at the group's first op" + ); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_walk_edges); + + produce(&mut sim, &client, namespace, STRAND_SENDS, "strand"); + + let withheld = WITHHELD_OP.load(Ordering::Relaxed); + assert_ne!( + withheld, 0, + "no partition prepare crossed the chain link, so the fault never armed" + ); + + for _ in 0..STRAND_QUIET_STEPS { + sim.step(); + let (_, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + if view != 0 || (commit_min >= withheld && commit_min == commit_max) { + break; + } + } + + assert!( + WITHHELD_DONES.load(Ordering::Relaxed) > 0, + "no repair terminator was withheld, so the walk was never starved and \ + a green run would not prove the tick backstop" + ); + assert!( + journal_holds(&sim, LAGGING, namespace, withheld), + "op {withheld} was never repaired back into the lagging replica's journal" + ); + let (status, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + assert_eq!( + view, 0, + "a view change drained the walk instead of the tick backstop; the test \ + proves nothing about normal status" + ); + assert_eq!(status, Status::Normal, "the replica left Normal status"); + for op in commit_min + 1..=commit_max { + assert!( + journal_holds(&sim, LAGGING, namespace, op), + "op {op} is not resident, so this run stranded on a repair gap, \ + not a parked walk" + ); + } + assert_eq!( + commit_min, commit_max, + "the walk never resumed over resident committed ops: walkable to \ + {commit_min}, committed through {commit_max}, every op between resident" + ); + } +}