This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 227786432 fix(shard): drive metadata repair and the commit walk from
the tick (#4008)
227786432 is described below
commit 227786432dc2f105c8d1139ae30a6aa8df8e2e56
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Mon Sep 7 10:45:09 2026 +0200
fix(shard): drive metadata repair and the commit walk from the tick (#4008)
A metadata backup that fell behind had no way back on its own, the same
starvation the partition plane just had. A follower advances commit_max
from every prepare header in replicate_preflight, before the gap check
drops the prepare, so the commit heartbeat lands as Accepted rather than
Advanced and the arming site inside that branch never fires. Under
sustained metadata traffic the gap wedged until an unrelated view
change. The same starvation left a backup holding resident committed ops
that nothing re-drove, since the follower walk runs at the tail of an
accepted prepare and the gap check returns before it.
The metadata tick now evaluates the same two level-triggered checks the
partition driver uses, off one probe. A hole below commit_max arms the
existing repair against the primary, debounced on the repair retry
interval. Resident committed ops re-drive the walk, follower only,
because a backup's walk ships no wire replies while a stranded primary
is resume_stranded_commits' job. A gap below the serving peer's
retention floor needs no special handling: the serve path answers
RangeEvicted and the existing conversion arms a state transfer.
---
core/configs/src/server_config/cluster.rs | 8 +-
core/metadata/src/impls/metadata.rs | 153 +++++-
core/partitions/src/iggy_partition.rs | 4 +-
core/server/config.toml | 19 +-
core/server/src/boot/recovery.rs | 6 +-
core/server/src/shell.rs | 6 +-
core/shard/src/lib.rs | 864 +++++++++++++++++++++++++-----
core/shard/src/metrics.rs | 39 +-
core/simulator/src/deps.rs | 14 +
core/simulator/src/lib.rs | 726 ++++++++++++++++++++++++-
10 files changed, 1664 insertions(+), 175 deletions(-)
diff --git a/core/configs/src/server_config/cluster.rs
b/core/configs/src/server_config/cluster.rs
index 929200f08..0667c131c 100644
--- a/core/configs/src/server_config/cluster.rs
+++ b/core/configs/src/server_config/cluster.rs
@@ -279,13 +279,15 @@ pub struct ClusterConfig {
#[serde_as(as = "DisplayFromStr")]
#[config_env(leaf)]
pub repair_retry_interval: IggyDuration,
- /// How long a partition backup must hold committed ops it cannot walk to
- /// before the shard sweep OPENS a repair session for it.
+ /// How long a backup must hold committed ops it cannot walk to before the
+ /// shard tick OPENS a repair session for it. Paces both planes' detectors:
+ /// the partition sweep in `tick_partitions` and the metadata one in
+ /// `tick_metadata`.
///
/// Separate from `repair_retry_interval`, which paces an already-open
/// stream: this one decides how long a replication hole stays open, so
/// raising the retry interval to quiet repair chatter must not widen it.
- /// Floored at `PARTITION_GAP_DEBOUNCE_TICKS_MIN` consensus ticks, since
one
+ /// Floored at `REPAIR_GAP_DEBOUNCE_TICKS_MIN` consensus ticks, since one
/// tick of lag is ordinary pipelining and repair against it would fire on
/// healthy traffic; the shard crate owns the floor and `config.toml`
states
/// its value. Zero (and the `0` / `disabled` / `unlimited` sentinels,
which
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index 726a013c1..9fbd167cf 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -218,6 +218,20 @@ impl IggySnapshot {
/// accepted (unverified, loudly) while a PRESENT but mismatching one refuses
boot. A
/// bare checksum could not tell those apart, and guessing wrong in either
direction is
/// unacceptable: silently accepting corruption, or bricking a healthy node.
+/// Committed ops one [`IggyMetadata::commit_journal`] call applies before
+/// returning to the pump.
+///
+/// The twin of `partitions::COMMIT_WALK_OPS_MAX`, and needed for the same
+/// reason: the walk reads a WAL body and applies it per op with no await the
+/// pump can interleave, and the resident `(commit_min, commit_max]` run is the
+/// whole backlog after a repair or a rejoin, not the pipeline depth.
+///
+/// Every caller is re-driven, so a truncated walk resumes rather than losing
+/// anything: `tick_metadata`'s walk backstop covers a follower and
+/// `resume_stranded_commits` covers the primary, both level-triggered on
+/// `commit_min < commit_max` every tick.
+const COMMIT_WALK_OPS_MAX: usize = 64;
+
const SNAPSHOT_TRAILER_MAGIC: u32 = 0x4953_4E50;
/// `magic` + the payload's [`checkpoint_checksum`].
@@ -776,6 +790,16 @@ pub struct IggyMetadata<C, J, S, M, SB =
PingPongSuperblock> {
/// whole snapshot on shard 0's pump, and hands each requester its own
/// multi-MB copy.
transfer_offer_cache: RefCell<Option<Rc<StateTransferOffer>>>,
+ /// Prepares the backup gap check destroyed since `tick_metadata` last
+ /// drained the count into `metadata_prepare_gap_drops_total`. What it does
+ /// and does not prove is `IggyPartition::prepare_gap_drops`, verbatim;
what
+ /// differs is the frontier the check runs against, the journal head rather
+ /// than the sequencer, so this also counts the ops that fall outside what
+ /// metadata repair can refill (an interior hole below the head, a forward
+ /// gap above `commit_max`).
+ ///
+ /// `Cell` because every method on this type takes `&self`.
+ prepare_gap_drops: Cell<u64>,
/// Highest metadata op whose apply has been PUBLISHED on this node, plus
/// the reads parked on it. Shared by every shard; see
/// [`AppliedFrontier`] for the ordering and the wake contract.
@@ -838,12 +862,19 @@ where
commit_notifier: RefCell::new(None),
client_table_frontier: Cell::new(0),
transfer_offer_cache: RefCell::new(None),
+ prepare_gap_drops: Cell::new(0),
applied_frontier: Arc::default(),
}
}
}
impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> {
+ /// Take and clear the gap-drop count (`prepare_gap_drops`).
+ #[must_use = "dropping the count loses the only record those prepares
existed"]
+ pub const fn take_prepare_gap_drops(&self) -> u64 {
+ self.prepare_gap_drops.replace(0)
+ }
+
/// Share one process-wide applied frontier with every other shard.
///
/// Consumed at construction rather than swapped in later: a shard that
@@ -1280,6 +1311,8 @@ where
sequencer_op = current_op,
"on_replicate: dropping out-of-order prepare (gap)"
);
+ self.prepare_gap_drops
+ .set(self.prepare_gap_drops.get().saturating_add(1));
return;
}
} else {
@@ -3608,15 +3641,26 @@ where
let consensus = self.consensus.as_ref().unwrap();
let journal = self.journal.as_ref().unwrap();
+ let mut applied = 0usize;
while consensus.commit_min() < consensus.commit_max() {
+ if applied == COMMIT_WALK_OPS_MAX {
+ debug!(
+ "commit_journal: stopping at op={} after {applied} ops;
resuming next tick",
+ consensus.commit_min()
+ );
+ break;
+ }
+ applied += 1;
let op = consensus.commit_min() + 1;
let Some(header) = journal.handle().header(op as usize) else {
// Gap-stop: the walk halts at the first missing prepare and
- // resumes once it is refilled. Live drops refill via the
- // primary's prepare retransmit; a replica behind at recovery
- // or after StartView adoption arms a `MetadataRepairSession`
- // (shard) that re-requests the missing window.
+ // resumes once it is refilled -- by the primary's retransmit
+ // while the op still lacks quorum, otherwise by a
+ // `MetadataRepairSession` (shard), armed at recovery, at
+ // StartView adoption, or by `tick_metadata`'s gap detector,
+ // which is the only one of those a live drop under sustained
+ // traffic reaches.
break;
};
let header = *header;
@@ -4253,6 +4297,20 @@ mod tests {
IggyMetadata::new(None, None, None, None, TestMux::default(), None)
}
+ #[test]
+ fn take_prepare_gap_drops_drains_the_count() {
+ let md = peer_metadata();
+ assert_eq!(md.take_prepare_gap_drops(), 0);
+
+ md.prepare_gap_drops.set(2);
+ assert_eq!(md.take_prepare_gap_drops(), 2);
+ assert_eq!(
+ md.take_prepare_gap_drops(),
+ 0,
+ "a second drain must not re-report drops the metrics already
counted"
+ );
+ }
+
#[test]
fn commit_notifier_fires_with_received_operation() {
let md = peer_metadata();
@@ -5035,6 +5093,93 @@ mod tests {
);
}
+ /// The walk cap bounds ONE `commit_journal` call, not the backlog.
+ ///
+ /// The resident `(commit_min, commit_max]` run after a repair window or a
+ /// rejoin is the whole backlog, and the walk applies each op with no await
+ /// the pump can interleave, so an uncapped call holds the shard for all of
+ /// it. Every caller is re-driven every tick, so stopping short loses
+ /// nothing; a cap that did NOT resume would pin `commit_min` until the
next
+ /// op to commit trips `advance_commit_min`'s sequential assert.
+ #[compio::test]
+ async fn
commit_journal_stops_at_the_walk_cap_and_resumes_on_the_next_call() {
+ const CLIENT: u128 = 1;
+ const SESSION: u64 = 1;
+ const ACTING_USER: u32 = 7;
+ /// The cap, as an op count.
+ const CAP: u64 = COMMIT_WALK_OPS_MAX as u64;
+ /// One op past the cap, so the first call must stop short and the
+ /// second must have something left to finish.
+ const OPS: u64 = CAP + 1;
+
+ let dir = tempfile::tempdir().unwrap();
+
std::fs::create_dir_all(dir.path().join(crate::impls::METADATA_DIR)).unwrap();
+ let journal =
+
journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"),
0)
+ .await
+ .unwrap();
+ // Replica 1 of 3 at view 0: a backup, so `on_replicate` journals each
+ // prepare without the primary's pipeline commit path running under it.
+ let consensus = VsrConsensus::new(
+ 1,
+ 1,
+ 3,
+ server_common::sharding::METADATA_GROUP,
+ NoopBus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (),
TestMux> =
+ IggyMetadata::new(
+ Some(consensus),
+ Some(journal),
+ None,
+ None,
+ TestMux::default(),
+ Some(dir.path().to_path_buf()),
+ );
+ let consensus = md.consensus.as_ref().unwrap();
+ md.client_table.borrow_mut().commit_register(
+ CLIENT,
+ ACTING_USER,
+ register_reply(CLIENT, SESSION),
+ );
+
+ for request in 1..=OPS {
+ let prepare = md
+ .prepare_request(create_stream_request(
+ CLIENT,
+ request,
+ &format!("s{request}"),
+ ))
+ .expect("CreateStream is client-allowed");
+ md.on_replicate(prepare).await;
+ }
+ let journal = md.journal.as_ref().unwrap();
+ assert_eq!(journal.last_op(), Some(OPS), "every op must be resident");
+ assert_eq!(consensus.commit_min(), 0, "no commit heartbeat has
landed");
+
+ // What a repair window or a rejoin leaves behind: the whole run
+ // committed by the group and resident here, none of it walked.
+ consensus.advance_commit_max(OPS);
+
+ md.commit_journal().await;
+ assert_eq!(
+ consensus.commit_min(),
+ CAP,
+ "one call walked the whole backlog; the pump is blocked for as
long \
+ as the resident run is, however long that is"
+ );
+
+ md.commit_journal().await;
+ assert_eq!(
+ consensus.commit_min(),
+ OPS,
+ "the walk did not resume where it stopped, so `commit_min` is
pinned \
+ below `commit_max` with no other re-driver"
+ );
+ }
+
/// A state-transfer receiver admits the first live prepare above the
floor it
/// installed, instead of waiting for an op the snapshot already contains.
///
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 2128ae517..356a10a7e 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -159,7 +159,7 @@ where
/// plus one is missing). Debounces the sweep's level-triggered repair arm,
/// and is spent by whichever site opens the repair session.
///
- /// `Cell` for the same reason as [`Self::prepare_gap_drops`]: the sweep
+ /// `Cell` for the same reason as `prepare_gap_drops`: the sweep
/// drives it from the shared borrow it probes the partition through, so
the
/// in-flight scan the arm budget needs can run without a `&mut`
outstanding.
pub gap_ticks: Cell<u32>,
@@ -1749,7 +1749,7 @@ where
self.write_superblock_advancing(superblock, 0, claim).await
}
- /// Take and clear the gap-drop count ([`Self::prepare_gap_drops`]).
+ /// Take and clear the gap-drop count (`prepare_gap_drops`).
#[must_use = "dropping the count loses the only record those prepares
existed"]
pub const fn take_prepare_gap_drops(&self) -> u64 {
self.prepare_gap_drops.replace(0)
diff --git a/core/server/config.toml b/core/server/config.toml
index 3bae72a0c..e36912513 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -659,21 +659,22 @@ view_probe_attempts_max = 5
# waits before one is opened for it is repair_gap_debounce_interval below.
repair_retry_interval = "1s"
-# How long a partition backup must hold committed ops it cannot walk to before
-# the shard's tick sweep opens a repair session for it (duration). The
-# level-triggered floor under every edge-triggered arming site, which a produce
-# stream can starve; must be nonzero.
+# How long a backup must hold committed ops it cannot walk to before the
shard's
+# tick opens a repair session for it (duration). The level-triggered floor
under
+# every edge-triggered arming site, which a produce stream can starve; must be
+# nonzero. Paces the partition sweep and the metadata detector alike.
#
# Floored at 50 consensus ticks (500ms at the 10ms tick), so values under that
# arm no sooner: one tick of lag is ordinary pipelining, and repair against it
# would fire on healthy traffic.
#
# Recovery latency for one hole is max(this, 500ms) plus up to one tick of
sweep
-# granularity. Under a correlated fault the sweep opens at most 3 sessions per
-# tick and holds at most 8 at once, so the Nth group waiting on this shard adds
-# ceil(N / 3) ticks on top, and more while sessions are already in flight. That
-# cap is what keeps a node-wide rejoin from putting every group's repair stream
-# on one serving peer at once.
+# granularity. Under a correlated fault the partition sweep opens at most 3
+# sessions per tick and holds at most 8 at once, so the Nth group waiting on
+# this shard adds ceil(N / 3) ticks on top, and more while sessions are already
+# in flight. That cap is what keeps a node-wide rejoin from putting every
+# group's repair stream on one serving peer at once. There is one metadata
group
+# per node, so its detector is not rate-capped.
repair_gap_debounce_interval = "1s"
# Prepares a peer serves per repair round before the requester walks to the
next
diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs
index 27d7a45b7..46558ffc7 100644
--- a/core/server/src/boot/recovery.rs
+++ b/core/server/src/boot/recovery.rs
@@ -285,7 +285,7 @@ pub(in crate::boot) async fn build_shard_for_thread(
// Repair pacing is shared by both planes' repair loops, so it is a
// per-shard tunable set once here rather than per consensus group.
shard.set_repair_retry_ticks(repair_retry_ticks(config));
- shard.set_partition_gap_debounce_ticks(repair_gap_debounce_ticks(config));
+ shard.set_repair_gap_debounce_ticks(repair_gap_debounce_ticks(config));
shard.set_superblock_wedged_fatal_failures(superblock_wedged_fatal_failures(config));
shard.set_served_segment_cache_bytes_max(
config
@@ -947,13 +947,13 @@ mod tests {
#[test]
fn documented_gap_debounce_floor_matches_the_shard_constant() {
assert_eq!(
- shard::PARTITION_GAP_DEBOUNCE_TICKS_MIN,
+ shard::REPAIR_GAP_DEBOUNCE_TICKS_MIN,
50,
"the gap debounce floor moved; core/server/config.toml states it
in \
ticks and milliseconds under [cluster]
repair_gap_debounce_interval"
);
assert_eq!(
- u128::from(shard::PARTITION_GAP_DEBOUNCE_TICKS_MIN)
+ u128::from(shard::REPAIR_GAP_DEBOUNCE_TICKS_MIN)
* shard::CONSENSUS_TICK_INTERVAL.as_millis(),
500,
"the floor is no longer 500ms; core/server/config.toml states that
\
diff --git a/core/server/src/shell.rs b/core/server/src/shell.rs
index 18145cb68..10da7aecf 100644
--- a/core/server/src/shell.rs
+++ b/core/server/src/shell.rs
@@ -202,11 +202,11 @@ pub(crate) fn repair_retry_ticks(config: &ServerConfig)
-> u32 {
}
/// `[cluster] repair_gap_debounce_interval` in consensus ticks: how long a
-/// partition backup holds a hole before the sweep opens a repair session for
-/// it. Deliberately NOT the retry interval above: that one paces an open
+/// backup holds a hole before the tick opens a repair session for it, on
either
+/// plane. Deliberately NOT the retry interval above: that one paces an open
/// stream, and pairing them means quieting retry chatter also widens how long
a
/// replication hole stays open. The shard applies
-/// [`shard::PARTITION_GAP_DEBOUNCE_TICKS_MIN`] as a floor on top.
+/// [`shard::REPAIR_GAP_DEBOUNCE_TICKS_MIN`] as a floor on top.
pub(crate) fn repair_gap_debounce_ticks(config: &ServerConfig) -> u32 {
u32::try_from(duration_to_ticks(
config.cluster.repair_gap_debounce_interval.get_duration(),
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 73cf6a9f0..008db6fbc 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -884,6 +884,10 @@ pub const REPAIR_CHUNK_MAX: u64 = 128;
struct MetadataRepairSession {
nonce: u128,
to_op: u64,
+ /// Consensus view this session was armed in. A later view decides the log
+ /// again, so the window this names may no longer be the one to fetch;
+ /// `partitions::RepairSession::view` fences the partition twin the same
way.
+ view: u32,
/// Re-request target on stall.
peer: u8,
/// Ticks since the stream last made progress; at
@@ -1319,6 +1323,21 @@ where
/// repair takes over at install. See [`MetadataTransferSession`].
metadata_transfer: RefCell<Option<MetadataTransferSession>>,
+ /// Consecutive ticks the metadata group has been seen gap-stopped
+ /// (committed ops it cannot walk to, because the op one past its commit
+ /// frontier is missing from the WAL). Debounces `tick_metadata`'s
+ /// level-triggered repair arm; the partition twin is
+ /// `IggyPartition::gap_ticks`. `Cell` because the tick drives it through
+ /// `&self`, and shard-level rather than plane-level because there is one
+ /// metadata group per node (precedent:
[`Self::metadata_transfer_attempts`]).
+ metadata_gap_ticks: Cell<u32>,
+
+ /// Op the tick's commit walk last stopped on without moving, or `0`. The
+ /// journal names it but cannot produce its body, so the gap probe counts
it
+ /// as absent and lets repair fetch it. Cleared implicitly: any advance of
+ /// `commit_min` makes it stop matching `commit_min + 1`.
+ metadata_walk_stuck_op: Cell<u64>,
+
/// Serving-side cache of state-transfer offers, both planes, keyed by
/// `(namespace, requester replica id)`. Bounded by the replica count times
/// the groups this shard serves; replaced per fresh nonce.
@@ -1493,11 +1512,11 @@ where
/// plane would alias; one sweep stale at worst.
partition_repairs_inflight: Cell<usize>,
- /// Live gap debounce in consensus ticks: how long a partition holds a hole
- /// before the sweep opens a repair session for it. Defaults to
- /// [`partitions::REPAIR_RETRY_TICKS`]; the server overrides it from
- /// `[cluster] repair_gap_debounce_interval` at bootstrap.
- partition_gap_debounce_ticks: Cell<u32>,
+ /// Live gap debounce in consensus ticks: how long a group holds a hole
+ /// before the tick opens a repair session for it. Shared by both planes.
+ /// Defaults to [`partitions::REPAIR_RETRY_TICKS`]; the server overrides it
+ /// from `[cluster] repair_gap_debounce_interval` at bootstrap.
+ repair_gap_debounce_ticks: Cell<u32>,
/// Namespace the next partition sweep starts from: the first group the
/// per-tick WALK budget turned away last pass, `None` to start at the
front.
@@ -1543,6 +1562,12 @@ where
/// [`Self::metadata_transfer_decode_failures`].
metadata_transfer_attempts: Cell<u32>,
+ /// Consecutive stalled re-requests on the live metadata repair session,
+ /// against [`partitions::REPAIR_MAX_STALL_RETRIES`]. Survives the session,
+ /// so rotating the peer cannot reset it; cleared by an accepted repaired
+ /// prepare.
+ metadata_repair_attempts: Cell<u32>,
+
/// Decode failures charged against one snapshot generation, as
/// `(snapshot_seq, failures)`. `None` until a pulled artifact set first
/// fails to decode; cleared by a successful install. Past
@@ -1678,6 +1703,8 @@ where
partition_submit_stalled: Cell::new(false),
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
+ metadata_gap_ticks: Cell::new(0),
+ metadata_walk_stuck_op: Cell::new(0),
state_transfer_offers: RefCell::new(HashMap::new()),
partition_offer_builds: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
@@ -1688,12 +1715,13 @@ where
partition_artifact_len_max:
Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
- partition_gap_debounce_ticks:
Cell::new(partitions::REPAIR_RETRY_TICKS),
+ repair_gap_debounce_ticks:
Cell::new(partitions::REPAIR_RETRY_TICKS),
partition_repairs_inflight: Cell::new(0),
partition_walk_cursor: Cell::new(None),
superblock_wedged_fatal_failures: Cell::new(0),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
metadata_transfer_attempts: Cell::new(0),
+ metadata_repair_attempts: Cell::new(0),
metadata_transfer_decode_failures: Cell::new(None),
})
}
@@ -1705,12 +1733,13 @@ where
self.repair_retry_ticks.set(ticks);
}
- /// Override the partition sweep's gap debounce (consensus ticks) from
- /// configuration. Called once per shard at bootstrap; the simulator and
- /// tests keep the compile-time [`partitions::REPAIR_RETRY_TICKS`] default.
- /// [`PARTITION_GAP_DEBOUNCE_TICKS_MIN`] still floors whatever is set.
- pub fn set_partition_gap_debounce_ticks(&self, ticks: u32) {
- self.partition_gap_debounce_ticks.set(ticks);
+ /// Override the tick gap debounce (consensus ticks) from configuration,
+ /// for both planes' detectors. Called once per shard at bootstrap; the
+ /// simulator and tests keep the compile-time
+ /// [`partitions::REPAIR_RETRY_TICKS`] default.
+ /// [`REPAIR_GAP_DEBOUNCE_TICKS_MIN`] still floors whatever is set.
+ pub fn set_repair_gap_debounce_ticks(&self, ticks: u32) {
+ self.repair_gap_debounce_ticks.set(ticks);
}
/// Arm the superblock fail-stop bound (consecutive write failures).
@@ -2135,6 +2164,8 @@ where
partition_submit_stalled: Cell::new(false),
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
+ metadata_gap_ticks: Cell::new(0),
+ metadata_walk_stuck_op: Cell::new(0),
state_transfer_offers: RefCell::new(HashMap::new()),
partition_offer_builds: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
@@ -2145,12 +2176,13 @@ where
partition_artifact_len_max:
Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
- partition_gap_debounce_ticks:
Cell::new(partitions::REPAIR_RETRY_TICKS),
+ repair_gap_debounce_ticks:
Cell::new(partitions::REPAIR_RETRY_TICKS),
partition_repairs_inflight: Cell::new(0),
partition_walk_cursor: Cell::new(None),
superblock_wedged_fatal_failures: Cell::new(0),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
metadata_transfer_attempts: Cell::new(0),
+ metadata_repair_attempts: Cell::new(0),
metadata_transfer_decode_failures: Cell::new(None),
}
}
@@ -4511,17 +4543,16 @@ where
// its own WAL (a late joiner missed the ops below the
// primary's active window; the primary only
retransmits
// uncommitted ops, never the committed prefix).
Without
- // this, such a replica learns it is behind and does
- // nothing about it -- metadata repair is otherwise
only
- // rooted at StartView adoption, which a same-view
- // late joiner never sees. Request repair from the
- // primary; if it has checkpointed past the gap the
- // repair floor evicts and the handler above converts
to
- // state transfer. Idempotent:
`maybe_request_metadata_repair`
- // no-ops when caught up, already transferring, or a
- // session is live, so a caught-up replica and a
- // cold-start node (commit_max == commit_min == 0) both
- // skip it.
+ // this, such a replica waits out `tick_metadata`'s
+ // debounced gap detector; this edge is the fast path,
+ // for the runs where a heartbeat does land as
+ // `Advanced`. Request repair from the primary; if it
+ // has checkpointed past the gap the repair floor
evicts
+ // and the handler above converts to state transfer.
+ // Idempotent: `maybe_request_metadata_repair` no-ops
+ // when caught up, already transferring, or a session
is
+ // live, so a caught-up replica and a cold-start node
+ // (commit_max == commit_min == 0) both skip it.
self.maybe_request_metadata_repair(consensus,
header.replica)
.await;
}
@@ -4976,6 +5007,15 @@ where
let Some(journal) = planes.0.journal.as_ref() else {
return;
};
+ // Above the two returns below, not after them: only SILENCE should
+ // age the stream, and an in-scope frame proves the peer is serving
+ // whether or not this replica still needs the op it carries. The
+ // ops a re-request re-serves are exactly the ones already held, so
+ // counting accepted frames alone rotates away from a live peer.
+ if let Some(session) = self.metadata_repair.borrow_mut().as_mut() {
+ session.idle_ticks = 0;
+ }
+ self.note_metadata_repair_progress();
let journal = journal.handle();
#[allow(clippy::cast_possible_truncation)]
if journal.header(header.op as usize).is_some() {
@@ -5075,10 +5115,10 @@ where
// peer's served-through claim: repair frames ride a
// lossy best-effort bus, so a fully-served stream can
// still arrive with holes. Anything short keeps the
- // session armed; while the walk is making progress the
- // next chunk is pulled immediately (the window is served
- // in `REPAIR_CHUNK_MAX` slices), and a stalled one is
- // left to the retry timer.
+ // session armed; the next chunk is pulled as soon as this
+ // one is walked (the window is served in
+ // `REPAIR_CHUNK_MAX` slices), and a window still holed
+ // below `served_through` is left to the retry timer.
let commit_min = consensus.commit_min();
let done = commit_min >= session.to_op;
tracing::info!(
@@ -5090,7 +5130,7 @@ where
);
if done {
*self.metadata_repair.borrow_mut() = None;
- } else if commit_min > before {
+ } else if repair_chunk_walked(before, commit_min,
header.op) {
self.send_request_prepares(
consensus.cluster(),
consensus.replica(),
@@ -5510,19 +5550,43 @@ where
B: MessageBus,
P: Pipeline<Entry = consensus::PipelineEntry>,
{
+ // `ViewChange` too: a parked view change repairs toward its merged log
+ // and cannot start until the window fills. Gating on `Normal` alone
+ // defers a dropped frame to the 500-tick escalation, and closes the
one
+ // session that legitimately runs outside `Normal`.
+ let repairing_view =
+ consensus.view_log_is_pending() &&
consensus.is_primary_for_view(consensus.view());
+
+ // Closed at the TOP of the tick, not after an idle window: a standing
+ // session fences every arming site and holds the gap debounce at zero
+ // (`recovery_owned`), so waiting a full retry interval to notice costs
+ // that interval on every arm behind it.
+ let superseded = self.metadata_repair.borrow().is_some_and(|session| {
+ metadata_repair_superseded(
+ &session,
+ consensus.commit_min(),
+ consensus.view(),
+ consensus.is_normal(),
+ repairing_view,
+ )
+ });
+ if superseded {
+ tracing::debug!(
+ shard = self.id,
+ commit_min = consensus.commit_min(),
+ view = consensus.view(),
+ "metadata repair session walked or superseded; closing it"
+ );
+ *self.metadata_repair.borrow_mut() = None;
+ self.note_metadata_repair_progress();
+ return;
+ }
+
// Stall retry (mirrors `tick_partitions`): a lost frame must not
wedge it.
let repair_retry_ticks = self.repair_retry_ticks.get();
let stalled = {
- // `ViewChange` too: a parked view change repairs toward its
merged log
- // and cannot start until the window fills. Gating on `Normal`
alone
- // defers a dropped frame to the 500-tick escalation.
- let repairing_view =
- consensus.view_log_is_pending() &&
consensus.is_primary_for_view(consensus.view());
let mut session = self.metadata_repair.borrow_mut();
session.as_mut().and_then(|session| {
- if !consensus.is_normal() && !repairing_view {
- return None;
- }
session.idle_ticks += 1;
if session.idle_ticks < repair_retry_ticks {
return None;
@@ -5532,6 +5596,39 @@ where
})
};
if let Some((peer, nonce, to_op)) = stalled {
+ // A session pins its peer and fences every arming site while it
+ // stands, so a peer that cannot answer wedges the plane harder
than
+ // having no session at all -- and the gap-stopped-primary rotation
+ // can pick a peer that is simply down. Past the budget the session
+ // is dropped and re-armed one step around the ring; an ordinary
lost
+ // frame is re-requested long before that.
+ if self.burn_metadata_repair_attempt() {
+ let next_peer = next_transfer_peer(
+ consensus.replica(),
+ peer,
+ consensus.replica_count(),
+ consensus.primary_index(consensus.view()),
+ );
+ tracing::warn!(
+ shard = self.id,
+ peer,
+ next_peer,
+ to_op,
+ "metadata repair stalled past its retry budget; re-arming
from another \
+ replica"
+ );
+ *self.metadata_repair.borrow_mut() = None;
+ self.note_metadata_repair_progress();
+ if next_peer != peer {
+ self.maybe_request_metadata_repair(consensus, next_peer)
+ .await;
+ }
+ // Nobody else to name (a solo group, or a two-replica group
+ // whose only peer went quiet): dropping the session is still
+ // right, since it unfences the detector, which re-arms after
its
+ // debounce and logs the state each interval.
+ return;
+ }
// Primary-elect only. Its window starts at the merged log's commit
// point, which can sit below local `commit_min` (the headers
inherited
// from senders behind the canonical log_view live there), so
@@ -5561,6 +5658,20 @@ where
consensus.group(),
)
.await;
+ } else {
+ // `from_op` past `to_op` without `commit_min` reaching it: the
+ // primary-elect window above starts at the merged log's commit
+ // point, which can sit above what this replica has walked. The
+ // top-of-tick check closes the ordinary case; this closes the
+ // one it cannot see.
+ tracing::info!(
+ shard = self.id,
+ to_op,
+ peer,
+ "metadata repair window fully requested; closing the
stalled session"
+ );
+ *self.metadata_repair.borrow_mut() = None;
+ self.note_metadata_repair_progress();
}
}
}
@@ -5811,6 +5922,7 @@ where
*self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
nonce,
to_op: pending.op_head,
+ view: consensus.view(),
peer,
idle_ticks: 0,
});
@@ -5834,14 +5946,31 @@ where
}
/// Start metadata tail journal-repair from `peer` when the commit walk
- /// gap-stopped below the known frontier. Shared by `StartView` adoption
- /// and the post-install step of a state transfer.
+ /// gap-stopped below the known frontier.
+ ///
+ /// Every TAIL arming site funnels through here -- `StartView` adoption,
the
+ /// commit-heartbeat backstop, the state-transfer fallbacks, and
+ /// `tick_metadata`'s gap detector -- so the guards below are what make the
+ /// level-triggered one idempotent. The one session this does not mint is
+ /// the view-change repair `advance_pending_metadata_view` builds inline:
it
+ /// repairs toward a merged log rather than the commit frontier, from a
peer
+ /// that offered the body rather than from the primary, so none of the
+ /// guards below describe it.
#[allow(clippy::future_not_send)]
async fn maybe_request_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>, peer: u8)
where
B: MessageBus,
P: Pipeline<Entry = consensus::PipelineEntry>,
{
+ // Never against self. A self-addressed `RequestPrepares` cannot be
+ // delivered (the replica registry holds no entry for this node), and
the
+ // send fails AFTER the session is recorded, so the session would stand
+ // forever: nothing advances `commit_min` to close it, the stall retry
+ // re-sends to the same place, and `metadata_repair.is_some()` fences
+ // every other arming site meanwhile.
+ if peer == consensus.replica() {
+ return;
+ }
if consensus.is_normal()
&& !consensus.is_transferring()
&& consensus.commit_min() < consensus.commit_max()
@@ -5850,9 +5979,15 @@ where
let nonce = iggy_common::random_id::get_uuid();
let to_op = consensus.commit_max();
let from_op = consensus.commit_min() + 1;
+ // Spent here rather than at the detector, so the edge-triggered
+ // sites spend it too: an edge-armed repair that completes before
+ // the next tick would otherwise leave the count saturated and hand
+ // the next real gap an arm on its first tick.
+ self.metadata_gap_ticks.set(0);
*self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
nonce,
to_op,
+ view: consensus.view(),
peer,
idle_ticks: 0,
});
@@ -5860,6 +5995,7 @@ where
shard = self.id,
from_op,
to_op,
+ peer,
"metadata behind the group frontier; requesting repair"
);
self.send_request_prepares(
@@ -6395,6 +6531,30 @@ where
budget.clamp(1, STATE_CHUNK_LEN as usize)
}
+ /// Burn one stalled repair round; `true` once the budget is exhausted and
+ /// the session should be re-armed against a different peer.
+ ///
+ /// The partition twin is `IggyPartition::burn_repair_attempt`, and it
lives
+ /// on the shard here for the same reason `metadata_transfer_attempts`
does:
+ /// one metadata group per node. It has to outlive the SESSION either way,
+ /// or the rotation that mints a new one would reset the count and
re-target
+ /// forever without ever giving up on a peer.
+ fn burn_metadata_repair_attempt(&self) -> bool {
+ let attempts = self.metadata_repair_attempts.get() + 1;
+ self.metadata_repair_attempts.set(attempts);
+ attempts > partitions::REPAIR_MAX_STALL_RETRIES
+ }
+
+ /// The serving peer answered: reset the budget, so it bounds CONSECUTIVE
+ /// silence rather than the stalls a long healthy stream accumulates.
+ ///
+ /// Any in-scope repair frame, not only an accepted one. A re-request
+ /// re-serves ops this replica already holds, so charging those as silence
+ /// rotates away from a peer that is answering.
+ fn note_metadata_repair_progress(&self) {
+ self.metadata_repair_attempts.set(0);
+ }
+
/// Burn one retry round; `true` once the budget is exhausted.
fn burn_metadata_transfer_attempt(&self) -> bool {
let attempts = self.metadata_transfer_attempts.get() + 1;
@@ -6870,7 +7030,7 @@ where
);
let partitions = self.plane.partitions();
let repair_retry_ticks = self.repair_retry_ticks.get();
- let gap_debounce_ticks = self.partition_gap_debounce_ticks.get();
+ let gap_debounce_ticks = self.repair_gap_debounce_ticks.get();
// Fan out over every group (each partition's heartbeat/retransmit
timer
// must advance), so the keyed single-namespace lookup the
control-frame
// handlers use does not apply here. The namespaces are snapshotted
into
@@ -7190,50 +7350,27 @@ where
repairs_live += 1;
}
let probe = partition_gap_probe(partition);
- let walk_stalled = partition_is_walk_stalled(&probe);
+ let walk_stalled = group_is_walk_stalled(&probe);
// The RATE cap only. The concurrency cap lives in the arm fn,
// which is the funnel every arming site goes through; resolved
// before the debounce either way, so a refusal keeps the group
// due rather than spending its arm.
- let may_arm = partition_is_gap_stopped(&probe)
+ let may_arm = group_is_gap_stopped(&probe)
&& repair_arms < PARTITION_REPAIR_ARMS_PER_TICK_MAX;
let mut gap_ticks = partition.gap_ticks.get();
- let verdict = drive_partition_gap_debounce(
- &probe,
- &mut gap_ticks,
- gap_debounce_ticks,
- may_arm,
- );
+ let verdict =
+ drive_group_gap_debounce(&probe, &mut gap_ticks,
gap_debounce_ticks, may_arm);
partition.gap_ticks.set(gap_ticks);
let arm_peer = match verdict {
GapArm::NotDue | GapArm::Deferred => None,
GapArm::Arm => {
let consensus = partition.consensus();
- let self_id = consensus.replica();
- let primary =
consensus.primary_index(consensus.view());
- // A gap-stopped PRIMARY cannot ask itself, and leaving
- // it to warn wedged the group: no edge-triggered site
- // re-drives a primary's own hole, and the next op to
- // commit walks `advance_commit_min` into its
sequential
- // assert. Any replica in `Normal` or `ViewChange`
serves
- // `RequestPrepares`, and a primary's window is its
- // COMMITTED prefix (the suffix widening needs a
pending
- // view log, which a settled primary has none of), so a
- // peer holding those ops holds them identically.
- //
- // The pick is positional, not liveness-aware: a dead
- // choice leaves the session re-requesting on the stall
- // timer, which is where a repair abandon budget (what
- // `burn_transfer_attempt` gives transfers) would
rotate
- // it. Still strictly better than the warn this
replaced,
- // which recovered nothing at all.
- let peer = if primary == self_id {
- next_transfer_peer(self_id, self_id,
consensus.replica_count(), primary)
- } else {
- primary
- };
- if peer == self_id {
- // Solo group: the rotation had nobody to return.
+ let peer = gap_repair_peer(
+ consensus.replica(),
+ consensus.replica_count(),
+ consensus.primary_index(consensus.view()),
+ );
+ if peer.is_none() {
// Restart the debounce so this repeats at its
// interval rather than every tick.
partition.gap_ticks.set(0);
@@ -7245,10 +7382,8 @@ where
"partition is gap-stopped below its own commit
frontier with no \
peer to repair from"
);
- None
- } else {
- Some(peer)
}
+ peer
}
};
(walk_stalled, arm_peer)
@@ -7271,7 +7406,7 @@ where
// 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,
- // so it cannot spin: `partition_is_walk_stalled` reads residency
off
+ // so it cannot spin: `group_is_walk_stalled` reads residency off
// `op_to_storage_offset` while the walk reads `headers`, and those
// two are written and cleared together (see `Journal::holds_op`),
so
// a group the predicate admits has an op for the walk to take.
@@ -9218,7 +9353,114 @@ where
}
}
+ /// Drop the WAL entry at `stuck_op` and the suffix above it, so repair can
+ /// refill a header whose body the commit walk cannot read.
+ ///
+ /// Nothing else clears it: `on_repair_prepare` returns early for an op
+ /// whose header is resident, and the append under it is refused anyway.
+ /// `stuck_op` is at `commit_min + 1` under `commit_max`, so a quorum holds
+ /// it and repair can serve it back.
+ ///
+ /// SERIALIZATION: same argument as `reconcile_metadata_view_divergence`,
+ /// which is the other shard-side `truncate_from` caller. This runs on the
+ /// pump between frames, so no append is in flight for these ops.
#[allow(clippy::future_not_send)]
+ async fn drop_unwalkable_metadata_entry<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ journal: &MJ,
+ stuck_op: u64,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ {
+ match journal.handle().truncate_from(stuck_op).await {
+ Ok(removed) => {
+ // The snapshot's `(op, commit)` tag does not move when entries
+ // are removed under it, so the next `DoViewChange` would
+ // otherwise advertise headers this replica can no longer
serve.
+ consensus.invalidate_local_dvc_suffix();
+ tracing::warn!(
+ shard = self.id,
+ stuck_op,
+ removed,
+ "metadata commit walk found a resident header with no body
at op {stuck_op}; \
+ dropped {removed} entries from it so repair can refill
the range"
+ );
+ }
+ Err(error) => {
+ tracing::error!(
+ shard = self.id,
+ stuck_op,
+ %error,
+ "could not drop the unwalkable entry at op {stuck_op};
journal repair skips \
+ ops it already holds a header for, so this replica will
not walk past \
+ it until it is restarted"
+ );
+ }
+ }
+ }
+
+ /// Read the gap probe off the metadata plane; [`partition_gap_probe`]'s
+ /// twin. A shard method because the recovery slots live here, on the
shard,
+ /// not on the plane.
+ fn metadata_gap_probe<P>(&self, consensus: &VsrConsensus<B, P>, journal:
&MJ) -> GapProbe
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ {
+ let commit_min = consensus.commit_min();
+ let commit_max = consensus.commit_max();
+ let normal = consensus.is_normal();
+ let transferring = consensus.is_transferring();
+ let recovery_owned =
+ self.metadata_transfer.borrow().is_some() ||
self.metadata_repair.borrow().is_some();
+ // Residency last, and only once the guards both predicates share hold,
+ // as in `partition_gap_probe`: a caught-up plane would otherwise pay a
+ // journal lookup whose answer both predicates discard.
+ //
+ // Safe against the snapshot floor: a checkpoint drains only to
+ // `commit_min`, so `commit_min + 1` never sits below it and a `None`
is
+ // a real hole.
+ //
+ // The header ring is only half of what the walk needs.
`commit_journal`
+ // reads the BODY through `entry()`, which answers `None` for an op the
+ // ring names but the WAL cannot produce, and then breaks without
moving
+ // `commit_min`. Reading the body here instead is not an option (it is
an
+ // async WAL read, per tick, on the walk's fast path), so the walk
+ // reports the op it stopped on and this treats that op as absent --
+ // which it is, for every purpose this probe serves. Without it the two
+ // disagree forever: the walk cannot move, the probe keeps calling the
+ // group walk-stalled, the debounce keeps resetting, and repair never
+ // arms.
+ //
+ // Self-clearing: any path that advances `commit_min` past the stuck op
+ // leaves `stuck_op != commit_min + 1`, so nothing has to retract it.
+ let next_op = commit_min.saturating_add(1);
+ #[allow(clippy::cast_possible_truncation)]
+ let next_op_resident = normal
+ && !transferring
+ && commit_min < commit_max
+ && self.metadata_walk_stuck_op.get() != next_op
+ && journal.handle().header(next_op as usize).is_some();
+ GapProbe {
+ normal,
+ transferring,
+ recovery_owned,
+ commit_min,
+ commit_max,
+ next_op_resident,
+ missing_suffix: false,
+ }
+ }
+
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
pub async fn tick_metadata(&self)
where
B: MessageBus,
@@ -9283,6 +9525,106 @@ where
self.advance_pending_metadata_view().await;
self.expire_idle_state_transfer_offers();
+ // Level-triggered gap detector, the metadata twin of the one in
+ // `tick_partitions`, and starvable in exactly the same way:
+ // `replicate_preflight` advances `commit_max` before the gap check
+ // drops the prepare, so under sustained traffic the heartbeat lands as
+ // `Accepted` and the `Advanced`-gated arm in `on_commit` never fires.
+ //
+ // Placed before the transfer-stall block below: that block's exhausted
+ // branch returns early, so a detector after it would be skipped on the
+ // tick that abandons a transfer.
+ if let Some(journal) = metadata.journal.as_ref() {
+ let gap_drops = metadata.take_prepare_gap_drops();
+ if gap_drops > 0 {
+ self.metrics.record_metadata_prepare_gap_drops(gap_drops);
+ }
+ let probe = self.metadata_gap_probe(consensus, journal);
+ let mut gap_ticks = self.metadata_gap_ticks.get();
+ // Always budgeted: one metadata group per node, so there is no
+ // correlated fan-out for a per-tick rate cap to spread.
+ let verdict = drive_group_gap_debounce(
+ &probe,
+ &mut gap_ticks,
+ self.repair_gap_debounce_ticks.get(),
+ true,
+ );
+ self.metadata_gap_ticks.set(gap_ticks);
+ if verdict == GapArm::Arm {
+ match gap_repair_peer(
+ consensus.replica(),
+ consensus.replica_count(),
+ consensus.primary_index(consensus.view()),
+ ) {
+ None => {
+ // Restart the debounce so this repeats at its
interval,
+ // not every tick.
+ self.metadata_gap_ticks.set(0);
+ tracing::warn!(
+ shard = self.id,
+ commit_min = probe.commit_min,
+ commit_max = probe.commit_max,
+ "metadata is gap-stopped below its own commit
frontier with no peer \
+ to repair from"
+ );
+ }
+ // Always repair, never classify the gap up front: a window
+ // below the peer's retention floor is answered
+ // `RangeEvicted`, and `on_repair_range_reply` converts
that
+ // to a state transfer. The floor is only ever learned
+ // through that refusal. The arm logs the window it settled
+ // on, so nothing is logged here.
+ Some(peer) =>
self.maybe_request_metadata_repair(consensus, peer).await,
+ }
+ }
+ // Undebounced, like the partition walk arm, and unrated: there is
+ // one group to walk here rather than a shard-wide fan-out, so
+ // nothing needs spreading across ticks. How FAR one walk goes is
+ // still capped, inside `commit_journal` itself.
+ //
+ // Both roles, like the partition arm. `resume_stranded_commits`
+ // above re-drives a primary's PIPELINE, and `(commit_min,
+ // commit_max]` is journal-only once it has run, so an inherited
+ // prefix or the tail of a capped walk has no other re-driver here
+ // and pins `commit_min` until the next op to commit trips
+ // `advance_commit_min`'s sequential assert.
+ //
+ // Not gated on `recovery_owned` (repaired prepares are journaled
+ // without being walked, so gating parks the walk for the whole
+ // session); `group_is_walk_stalled` itself refuses mid-transfer,
+ // where a walk past the incoming `snapshot_seq` would break the
+ // install.
+ if group_is_walk_stalled(&probe) {
+ // Debug, not info: a repair stream journals its prepares
without
+ // walking them, so this is the steady state for the whole
+ // duration of a rejoin and would be one line per tick.
+ tracing::debug!(
+ shard = self.id,
+ commit_min = probe.commit_min,
+ commit_max = probe.commit_max,
+ "metadata commit walk parked over resident committed ops;
resuming"
+ );
+ metadata.commit_journal().await;
+ // A walk that moved nothing found the header and not the body.
+ // Recording the op stops the detector calling this a parked
+ // walk, but arming repair alone cannot refill it: the ingest
+ // skips an op whose header is resident and `append` refuses
the
+ // slot under it, so the header has to go first.
+ let walked = consensus.commit_min();
+ if walked == probe.commit_min {
+ let stuck_op = walked.saturating_add(1);
+ // Once per op: a failed truncation leaves the header where
+ // it is, and retrying every tick only repeats the error.
+ if self.metadata_walk_stuck_op.replace(stuck_op) !=
stuck_op {
+ self.drop_unwalkable_metadata_entry(consensus,
journal, stuck_op)
+ .await;
+ }
+ } else {
+ self.metadata_walk_stuck_op.set(0);
+ }
+ }
+ }
+
// Stall retry for an in-flight state transfer: descriptor or chunk
// frames are fire-and-forget, so a lost one must not wedge the
// session (and the boot flow behind it) forever.
@@ -9318,9 +9660,17 @@ where
consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
}
metadata.commit_journal().await;
- let current_primary =
consensus.primary_index(consensus.view());
- self.maybe_request_metadata_repair(consensus, current_primary)
- .await;
+ // Rotated, not `primary_index` raw: this replica can BE the
+ // primary here (a leading replica that transferred to catch up
+ // on a checkpoint it lacked), and the arm refuses self.
+ if let Some(next_peer) = gap_repair_peer(
+ consensus.replica(),
+ consensus.replica_count(),
+ consensus.primary_index(consensus.view()),
+ ) {
+ self.maybe_request_metadata_repair(consensus, next_peer)
+ .await;
+ }
return;
}
tracing::info!(
@@ -9681,12 +10031,15 @@ const PARTITION_WALKS_PER_TICK_MAX: usize = 16;
/// Public because it bounds what that operator knob can do: gap recovery
starts
/// after `max(repair_gap_debounce_interval, this)`, which the `[cluster]`
/// documentation states.
-pub const PARTITION_GAP_DEBOUNCE_TICKS_MIN: u32 = 50;
+pub const REPAIR_GAP_DEBOUNCE_TICKS_MIN: u32 = 50;
-/// What the tick sweep reads off one partition to decide whether it is
+/// What a tick driver reads off one consensus group 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.
///
+/// Both planes fill it: `partition_gap_probe` off a live partition, and
+/// `IggyShard::metadata_gap_probe` off the metadata plane's consensus and WAL.
+///
/// The flags are independent readings of one instant, not states of one
/// machine, and the exhaustive predicate test below enumerates them as such,
so
/// the lint's two-variant enums would only rename `true` and `false`.
@@ -9696,8 +10049,10 @@ struct GapProbe {
normal: bool,
transferring: bool,
/// Whether a repair session, a transfer, or a scheduled transfer re-arm
- /// already owns this partition's recovery. Arming a second one would race
- /// it, or defeat the re-arm's backoff as `arm_partition_transfer`
documents.
+ /// already owns this group's recovery. Arming a second one would race it,
+ /// or defeat the re-arm's backoff as `arm_partition_transfer` documents.
+ /// The re-arm shape is the partition plane's alone; metadata has no
+ /// re-arm state, so its probe reads the other two.
recovery_owned: bool,
commit_min: u64,
commit_max: u64,
@@ -9713,6 +10068,12 @@ struct GapProbe {
/// below the frontier: the group cannot gather quorum for that suffix
until
/// the bodies land, and the only other site that notices is the single
/// `on_start_view` edge that adopted them. See
[`partition_missing_suffix`].
+ ///
+ /// Always `false` on a metadata probe: the shape it names is read off the
+ /// partition's own journal window, and the metadata plane's equivalent is
+ /// still only noticed at the `advance_pending_metadata_view` edge. So the
+ /// metadata detector covers the hole BELOW the frontier and nothing above
+ /// it.
missing_suffix: bool,
}
@@ -9722,8 +10083,8 @@ struct GapProbe {
/// 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 {
+/// lag test would arm repair against ordinary traffic.
+const fn group_is_gap_stopped(probe: &GapProbe) -> bool {
if !probe.normal || probe.transferring || probe.recovery_owned {
return false;
}
@@ -9745,18 +10106,18 @@ const fn partition_is_gap_stopped(probe: &GapProbe) ->
bool {
/// commit is `Accepted`, and an idle group offers no other edge).
///
/// The two split on `next_op_resident` while a lag stands, and
-/// [`partition_is_gap_stopped`] defers to that split even for a missing
suffix,
+/// [`group_is_gap_stopped`] defers to that split even for a missing suffix,
/// so they cannot both hold. Both are false whenever a shared guard fails. Not
/// gated on `recovery_owned`: repair fetches bodies without walking them, so
/// gating parks the walk all session.
-const fn partition_is_walk_stalled(probe: &GapProbe) -> bool {
+const fn group_is_walk_stalled(probe: &GapProbe) -> bool {
probe.normal
&& !probe.transferring
&& probe.commit_min < probe.commit_max
&& probe.next_op_resident
}
-/// What the debounce says about one partition on one sweep.
+/// What the debounce says about one group on one tick.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum GapArm {
/// Not gap-stopped, or gap-stopped for less than the debounce.
@@ -9765,39 +10126,46 @@ enum GapArm {
/// so the group is due again on the next pass rather than serving a fresh
/// interval. It moves no cursor: the sweep resumes where the WALK budget
/// ran out, and arms drain their own queue as sessions open.
+ ///
+ /// Partition-plane only. The metadata driver holds one group per node, so
+ /// it always passes a budget and never sees this.
Deferred,
/// Open a repair session now.
Arm,
}
-/// Count one sweep tick against `gap_ticks` and answer whether this partition
-/// may arm repair now.
+/// Count one tick against `gap_ticks` and answer whether this group 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.
+/// Level-triggered, because every edge-triggered arming site is starvable, on
+/// both planes: the commit-heartbeat backstop fires only on
+/// `CommitOutcome::Advanced`, and under sustained traffic 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.
///
-/// `budget_available` is the sweep's per-tick arm rate; the live-session
-/// ceiling is applied by `maybe_request_partition_repair`, which every arming
-/// site funnels through. A refused arm keeps its debounce satisfied rather
than
-/// starting over, so the group arms on the next pass with a slot free.
Spending it is `maybe_request_partition_repair`'s job, which resets
-/// `gap_ticks` for EVERY arming site, not just this one: an edge-armed repair
-/// that completes before the next sweep would otherwise leave the count
-/// saturated and hand the next gap an arm on its first tick.
-const fn drive_partition_gap_debounce(
+/// `budget_available` is the partition sweep's per-tick arm rate; the
+/// live-session ceiling is applied by `maybe_request_partition_repair`, which
+/// every partition arming site funnels through. A refused arm keeps its
+/// debounce satisfied rather than starting over, so the group arms on the next
+/// pass with a slot free.
+///
+/// Spending the count is the arm function's job, not this one's, and it resets
+/// `gap_ticks` for EVERY arming site rather than only the tick: an edge-armed
+/// repair that completes before the next tick would otherwise leave the count
+/// saturated and hand the next gap an arm on its first tick. Metadata's twin
of
+/// that reset lives in `maybe_request_metadata_repair`.
+const fn drive_group_gap_debounce(
probe: &GapProbe,
gap_ticks: &mut u32,
debounce_ticks: u32,
budget_available: bool,
) -> GapArm {
- if !partition_is_gap_stopped(probe) {
+ if !group_is_gap_stopped(probe) {
*gap_ticks = 0;
return GapArm::NotDue;
}
- let debounce_ticks = if debounce_ticks < PARTITION_GAP_DEBOUNCE_TICKS_MIN {
- PARTITION_GAP_DEBOUNCE_TICKS_MIN
+ let debounce_ticks = if debounce_ticks < REPAIR_GAP_DEBOUNCE_TICKS_MIN {
+ REPAIR_GAP_DEBOUNCE_TICKS_MIN
} else {
debounce_ticks
};
@@ -9812,6 +10180,61 @@ const fn drive_partition_gap_debounce(
}
}
+/// The peer a gap-stopped replica asks for repair, or `None` when there is
+/// nobody to ask.
+///
+/// The primary, except when this replica IS the primary: no site re-drives a
+/// settled primary's own hole, so leaving it to warn wedges the group, and the
+/// next op to commit walks `advance_commit_min` into its sequential assert.
Any
+/// replica in `Normal` or `ViewChange` serves `RequestPrepares`, and a
+/// gap-stopped replica's window is its COMMITTED prefix, which every peer that
+/// holds those ops holds identically.
+///
+/// Positional, not liveness-aware. A dead pick is corrected by the stall
+/// budget on either plane, which drops the session and rotates one step
further
+/// around the ring rather than re-requesting from it forever.
+///
+/// Shared by both planes so the rule cannot drift: the partition sweep and
+/// `tick_metadata` arm off the same predicate and owe the same answer.
+const fn gap_repair_peer(self_id: u8, replica_count: u8, primary: u8) ->
Option<u8> {
+ let peer = if primary == self_id {
+ next_transfer_peer(self_id, self_id, replica_count, primary)
+ } else {
+ primary
+ };
+ // A solo group (or a ring with nobody else live to name) rotates back to
+ // self, which no session can be opened against.
+ if peer == self_id { None } else { Some(peer) }
+}
+
+/// Whether a standing metadata repair session should be closed at the top of
+/// the tick: its window is walked, the view that decided that window has
+/// moved, or this replica has left the status the session belongs to.
+///
+/// `repairing_view` is the primary-elect repairing toward its merged log, the
+/// one session that runs outside `Normal`. Pinned by
+/// `metadata_repair_session_tests`.
+const fn metadata_repair_superseded(
+ session: &MetadataRepairSession,
+ commit_min: u64,
+ view: u32,
+ normal: bool,
+ repairing_view: bool,
+) -> bool {
+ commit_min >= session.to_op || session.view != view || !(normal ||
repairing_view)
+}
+
+/// Whether a walked `RepairDone` should pull the next chunk of the window.
+///
+/// `served_through` is the terminator's own op. Chunk progress, not this
+/// walk's: `tick_metadata` walks the same journal, so it can consume a chunk
+/// between the chunk's last prepare and its terminator, and requiring
+/// `commit_min` to move HERE idles the session a full retry interval on every
+/// such landing.
+const fn repair_chunk_walked(before: u64, commit_min: u64, served_through:
u64) -> bool {
+ commit_min > before || commit_min >= served_through
+}
+
/// Rotate a sweep's namespace snapshot so it resumes at `cursor`.
///
/// The per-tick caps are what make this necessary: the snapshot is in
ascending
@@ -11421,15 +11844,19 @@ mod sweep_scheduler_tests {
#[cfg(test)]
mod gap_detector_tests {
- //! The level-triggered repair arm the partition tick sweep runs.
+ //! The level-triggered repair arm the partition and metadata tick drivers
+ //! share.
//!
//! 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.
+ //! it off healthy traffic are the parts worth pinning. Probes are built
+ //! here by hand: what the two planes read off their own state is
+ //! `partition_gap_probe`'s and `metadata_gap_probe`'s business, and the
+ //! simulator's driver suites cover those end to end.
use super::{
- GapArm, GapProbe, PARTITION_GAP_DEBOUNCE_TICKS_MIN,
drive_partition_gap_debounce,
- partition_is_gap_stopped, partition_is_walk_stalled,
+ GapArm, GapProbe, REPAIR_GAP_DEBOUNCE_TICKS_MIN,
drive_group_gap_debounce,
+ group_is_gap_stopped, group_is_walk_stalled,
};
const DEBOUNCE: u32 = 100;
@@ -11474,8 +11901,8 @@ mod gap_detector_tests {
// 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.
- assert!(!partition_is_gap_stopped(&walk_stalled()));
- assert!(partition_is_gap_stopped(&gap_stopped()));
+ assert!(!group_is_gap_stopped(&walk_stalled()));
+ assert!(group_is_gap_stopped(&gap_stopped()));
}
#[test]
@@ -11484,7 +11911,7 @@ mod gap_detector_tests {
commit_min: 10,
..gap_stopped()
};
- assert!(!partition_is_gap_stopped(&caught_up));
+ assert!(!group_is_gap_stopped(&caught_up));
}
#[test]
@@ -11493,9 +11920,9 @@ mod gap_detector_tests {
// the head, so there is no lag to see, and the only other site that
// notices is the single `on_start_view` edge that adopted the headers.
// Left out, that class hangs until an unrelated view change.
- assert!(partition_is_gap_stopped(&missing_suffix()));
+ assert!(group_is_gap_stopped(&missing_suffix()));
assert!(
- !partition_is_gap_stopped(&GapProbe {
+ !group_is_gap_stopped(&GapProbe {
missing_suffix: false,
..missing_suffix()
}),
@@ -11508,11 +11935,11 @@ mod gap_detector_tests {
// 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.
for probe in [gap_stopped(), missing_suffix()] {
- assert!(!partition_is_gap_stopped(&GapProbe {
+ assert!(!group_is_gap_stopped(&GapProbe {
normal: false,
..probe
}));
- assert!(!partition_is_gap_stopped(&GapProbe {
+ assert!(!group_is_gap_stopped(&GapProbe {
transferring: true,
..probe
}));
@@ -11524,7 +11951,7 @@ mod gap_detector_tests {
// A session, a transfer, or a scheduled transfer re-arm all own the
// recovery; a second one would race it or defeat the re-arm's backoff.
for probe in [gap_stopped(), missing_suffix()] {
- assert!(!partition_is_gap_stopped(&GapProbe {
+ assert!(!group_is_gap_stopped(&GapProbe {
recovery_owned: true,
..probe
}));
@@ -11537,13 +11964,13 @@ mod gap_detector_tests {
let mut gap_ticks = 0;
for tick in 1..DEBOUNCE {
assert_eq!(
- drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE,
true),
+ drive_group_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE,
true),
GapArm::NotDue,
"armed at tick {tick}, before the debounce elapsed"
);
}
assert_eq!(
- drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE,
true),
+ drive_group_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, true),
GapArm::Arm
);
}
@@ -11557,17 +11984,17 @@ mod gap_detector_tests {
};
let mut gap_ticks = 0;
for _ in 0..DEBOUNCE - 1 {
- drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE,
true);
+ drive_group_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, true);
}
assert_eq!(gap_ticks, DEBOUNCE - 1);
assert_eq!(
- drive_partition_gap_debounce(&walkable, &mut gap_ticks, DEBOUNCE,
true),
+ drive_group_gap_debounce(&walkable, &mut gap_ticks, DEBOUNCE,
true),
GapArm::NotDue
);
assert_eq!(gap_ticks, 0, "progress must restart the debounce");
assert_eq!(
- drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE,
true),
+ drive_group_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, true),
GapArm::NotDue,
"a fresh gap must serve its own debounce, not inherit the old
count"
);
@@ -11575,9 +12002,9 @@ mod gap_detector_tests {
#[test]
fn
given_a_follower_with_resident_committed_ops_when_probed_should_be_walk_stalled()
{
- assert!(partition_is_walk_stalled(&walk_stalled()));
+ assert!(group_is_walk_stalled(&walk_stalled()));
assert!(
- !partition_is_walk_stalled(&gap_stopped()),
+ !group_is_walk_stalled(&gap_stopped()),
"a missing next op is repair's job; a walk over it would stop dead"
);
}
@@ -11588,7 +12015,7 @@ mod gap_detector_tests {
commit_min: 10,
..walk_stalled()
};
- assert!(!partition_is_walk_stalled(&caught_up));
+ assert!(!group_is_walk_stalled(&caught_up));
}
#[test]
@@ -11597,7 +12024,7 @@ mod gap_detector_tests {
normal: false,
..walk_stalled()
};
- assert!(!partition_is_walk_stalled(&electing));
+ assert!(!group_is_walk_stalled(&electing));
// Same gate as the on-commit arm: a walk during a transfer can advance
// commit_min past the incoming frontier.
@@ -11605,7 +12032,7 @@ mod gap_detector_tests {
transferring: true,
..walk_stalled()
};
- assert!(!partition_is_walk_stalled(&installing));
+ assert!(!group_is_walk_stalled(&installing));
}
#[test]
@@ -11613,7 +12040,7 @@ mod gap_detector_tests {
// 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.
- assert!(partition_is_walk_stalled(&GapProbe {
+ assert!(group_is_walk_stalled(&GapProbe {
recovery_owned: true,
..walk_stalled()
}));
@@ -11639,8 +12066,8 @@ mod gap_detector_tests {
missing_suffix,
};
assert!(
- !(partition_is_gap_stopped(&probe)
- && partition_is_walk_stalled(&probe)),
+ !(group_is_gap_stopped(&probe)
+ && group_is_walk_stalled(&probe)),
"both predicates claim {probe:?}"
);
}
@@ -11661,13 +12088,13 @@ mod gap_detector_tests {
missing_suffix: true,
..walk_stalled()
};
- assert!(partition_is_walk_stalled(&probe));
+ assert!(group_is_walk_stalled(&probe));
assert!(
- !partition_is_gap_stopped(&probe),
+ !group_is_gap_stopped(&probe),
"a walkable lag must win the tick; the suffix arm waits for it to
close"
);
assert!(
- partition_is_gap_stopped(&GapProbe {
+ group_is_gap_stopped(&GapProbe {
commit_min: probe.commit_max,
..probe
}),
@@ -11680,7 +12107,7 @@ mod gap_detector_tests {
let probe = gap_stopped();
let mut gap_ticks = DEBOUNCE;
assert_eq!(
- drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE,
false),
+ drive_group_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, false),
GapArm::Deferred,
"a spent budget must refuse the arm"
);
@@ -11690,7 +12117,7 @@ mod gap_detector_tests {
arm a whole interval out per contended tick"
);
assert_eq!(
- drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE,
true),
+ drive_group_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, true),
GapArm::Arm,
"the same group arms on the next pass with a slot free"
);
@@ -11705,16 +12132,173 @@ mod gap_detector_tests {
// reordered prepare.
let probe = gap_stopped();
let mut gap_ticks = 0;
- for tick in 1..PARTITION_GAP_DEBOUNCE_TICKS_MIN {
+ for tick in 1..REPAIR_GAP_DEBOUNCE_TICKS_MIN {
assert_eq!(
- drive_partition_gap_debounce(&probe, &mut gap_ticks, 1, true),
+ drive_group_gap_debounce(&probe, &mut gap_ticks, 1, true),
GapArm::NotDue,
"a 1-tick debounce armed at tick {tick}, under the floor"
);
}
assert_eq!(
- drive_partition_gap_debounce(&probe, &mut gap_ticks, 1, true),
+ drive_group_gap_debounce(&probe, &mut gap_ticks, 1, true),
GapArm::Arm
);
}
}
+
+#[cfg(test)]
+mod metadata_repair_session_tests {
+ //! The three rules a standing metadata repair session lives by: who it is
+ //! opened against, when it is closed, and when a walked terminator pulls
+ //! the next chunk of its window.
+ //!
+ //! All three wedge the plane rather than failing loudly. A session fences
+ //! every other arming site and holds the gap debounce at zero while it
+ //! stands, so one opened against nobody, or kept past the view that
decided
+ //! its window, or that stops pulling chunks, pins the commit frontier with
+ //! nothing else able to arm.
+
+ use super::{
+ MetadataRepairSession, gap_repair_peer, metadata_repair_superseded,
next_transfer_peer,
+ repair_chunk_walked,
+ };
+
+ /// Armed at view 3, against a window ending at op 20.
+ const fn session() -> MetadataRepairSession {
+ MetadataRepairSession {
+ nonce: 7,
+ to_op: 20,
+ view: 3,
+ peer: 0,
+ idle_ticks: 0,
+ }
+ }
+
+ #[test]
+ fn given_a_gap_stopped_backup_when_picking_a_peer_should_ask_the_primary()
{
+ assert_eq!(gap_repair_peer(2, 3, 0), Some(0));
+ assert_eq!(gap_repair_peer(1, 5, 3), Some(3));
+ }
+
+ #[test]
+ fn
given_a_gap_stopped_primary_when_picking_a_peer_should_never_ask_itself() {
+ // The case `maybe_request_metadata_repair`'s self-guard exists for: no
+ // other site re-drives a settled primary's own hole, and a
+ // self-addressed request fails to send AFTER the session is recorded.
+ for replica_count in 2..=7u8 {
+ for primary in 0..replica_count {
+ let peer = gap_repair_peer(primary, replica_count, primary);
+ assert_ne!(peer, Some(primary), "count {replica_count}");
+ assert!(peer.is_some(), "count {replica_count}");
+ }
+ }
+ }
+
+ #[test]
+ fn given_a_solo_group_when_picking_a_peer_should_answer_nobody() {
+ assert_eq!(gap_repair_peer(0, 1, 0), None);
+ }
+
+ #[test]
+ fn
given_a_silent_peer_when_the_stall_budget_is_spent_should_rotate_off_it() {
+ // Re-arming against the peer that just went quiet spends another whole
+ // budget on it, and the ring is the only thing that names anyone else.
+ for replica_count in 3..=7u8 {
+ for primary in 0..replica_count {
+ let self_id = (primary + 1) % replica_count;
+ let failed = gap_repair_peer(self_id, replica_count,
primary).expect("a peer");
+ let next = next_transfer_peer(self_id, failed, replica_count,
primary);
+ assert_ne!(next, failed, "count {replica_count}, primary
{primary}");
+ assert_ne!(next, self_id, "count {replica_count}, primary
{primary}");
+ }
+ }
+ }
+
+ #[test]
+ fn
given_two_replicas_when_the_stall_budget_is_spent_should_name_the_same_peer_back()
{
+ // Which is how the caller reads "nobody else to ask" and drops the
+ // session instead of re-arming it.
+ assert_eq!(next_transfer_peer(1, 0, 2, 0), 0);
+ }
+
+ #[test]
+ fn
given_a_session_whose_window_is_walked_when_checked_should_be_superseded() {
+ let session = session();
+ assert!(metadata_repair_superseded(
+ &session,
+ session.to_op,
+ session.view,
+ true,
+ false
+ ));
+ assert!(!metadata_repair_superseded(
+ &session,
+ session.to_op - 1,
+ session.view,
+ true,
+ false
+ ));
+ }
+
+ #[test]
+ fn
given_a_session_armed_in_an_earlier_view_when_checked_should_be_superseded() {
+ let session = session();
+ assert!(metadata_repair_superseded(
+ &session,
+ 0,
+ session.view + 1,
+ true,
+ false
+ ));
+ }
+
+ #[test]
+ fn given_a_replica_that_left_normal_when_checked_should_be_superseded() {
+ let session = session();
+ assert!(metadata_repair_superseded(
+ &session,
+ 0,
+ session.view,
+ false,
+ false
+ ));
+ }
+
+ #[test]
+ fn
given_a_primary_elect_repairing_its_merged_log_when_checked_should_stand() {
+ // The one session that runs outside `Normal`;
+ // `advance_pending_metadata_view` cannot start the view until its
+ // window fills.
+ let session = session();
+ assert!(!metadata_repair_superseded(
+ &session,
+ 0,
+ session.view,
+ false,
+ true
+ ));
+ assert!(
+ metadata_repair_superseded(&session, 0, session.view + 1, false,
true),
+ "not even the primary-elect's session survives the next view"
+ );
+ }
+
+ #[test]
+ fn given_a_chunk_this_walk_moved_when_checked_should_pull_the_next_chunk()
{
+ assert!(repair_chunk_walked(5, 8, 12));
+ }
+
+ #[test]
+ fn
given_a_chunk_the_tick_already_walked_when_checked_should_pull_the_next_chunk()
{
+ // `tick_metadata` walks the same journal, so the terminator can arrive
+ // with nothing left for its own walk to move.
+ assert!(repair_chunk_walked(12, 12, 12));
+ }
+
+ #[test]
+ fn
given_a_window_still_holed_below_the_terminator_when_checked_should_wait_for_the_retry()
{
+ // A frame was lost inside the served chunk: re-requesting now would
+ // race the retry timer for the same window.
+ assert!(!repair_chunk_walked(5, 5, 12));
+ }
+}
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index 3e4b1c14f..2d9535c25 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -212,6 +212,7 @@ pub struct ShardMetrics {
partition_requests_denied_transient_total: Counter,
partition_repair_serves_deferred_purge_total: Counter,
partition_prepare_gap_drops_total: Counter,
+ metadata_prepare_gap_drops_total: Counter,
metadata_read_frontier_refusals_total: Counter,
client_requests_denied_queue_full_total: Counter,
}
@@ -239,6 +240,7 @@ impl ShardMetrics {
partition_requests_denied_transient_total: Counter::default(),
partition_repair_serves_deferred_purge_total: Counter::default(),
partition_prepare_gap_drops_total: Counter::default(),
+ metadata_prepare_gap_drops_total: Counter::default(),
metadata_read_frontier_refusals_total: Counter::default(),
client_requests_denied_queue_full_total: Counter::default(),
}
@@ -473,10 +475,9 @@ impl ShardMetrics {
/// 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, and has no
- /// counter of its own: its repair is armed by the same edge-triggered
sites
- /// this plane's sweep exists to backstop, so that plane is starvable in
the
- /// same way and is simply not instrumented for it yet.
+ /// The metadata plane's own gap drop is NOT counted here: it has its own
+ /// counter, and its own level-triggered driver in `tick_metadata`. See
+ /// [`Self::record_metadata_prepare_gap_drops`].
pub fn record_partition_prepare_gap_drops(&self, drops: u64) {
self.partition_prepare_gap_drops_total.inc_by(drops);
}
@@ -488,6 +489,31 @@ impl ShardMetrics {
self.partition_prepare_gap_drops_total.get()
}
+ /// Add the prepares the metadata backup gap check destroyed since the last
+ /// tick, drained from `IggyMetadata::take_prepare_gap_drops`.
+ ///
+ /// The sibling of [`Self::record_partition_prepare_gap_drops`], and it
+ /// carries every caveat that one does: it counts the prepares that ARRIVED
+ /// after a hole rather than the holes, so a nonzero value proves the
+ /// metadata repair driver has work and a zero one proves nothing. There is
+ /// one metadata group per node, so unlike the partition counter it needs
no
+ /// argument about namespace cardinality.
+ ///
+ /// Deliberately NOT a `frame_drops_total` reason, for the same 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, while a gap
drop
+ /// is a protocol-ordering drop the tick repairs.
+ pub fn record_metadata_prepare_gap_drops(&self, drops: u64) {
+ self.metadata_prepare_gap_drops_total.inc_by(drops);
+ }
+
+ /// Snapshot of `metadata_prepare_gap_drops_total`. Test/simulator
accessor.
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn metadata_prepare_gap_drops_value(&self) -> u64 {
+ self.metadata_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.
@@ -584,6 +610,11 @@ impl ShardMetrics {
"replicated prepares dropped out of order by a backup's gap check",
self.partition_prepare_gap_drops_total.clone(),
);
+ registry.register(
+ "metadata_prepare_gap_drops",
+ "replicated metadata prepares dropped out of order by a backup's
gap check",
+ self.metadata_prepare_gap_drops_total.clone(),
+ );
registry.register(
"metadata_read_frontier_refusals",
"metadata reads refused because this node never applied the
caller's committed op",
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index 39b12c845..ace532859 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -435,6 +435,20 @@ impl SimJournal<MemStorage> {
headers.remove(&op).is_some()
}
+ /// Forget one op's BODY, leaving its header resident: what a commit walk
+ /// reading `entry()` cannot get past on its own, since the header the
+ /// gap check and the repair ingest both read is still there.
+ ///
+ /// Tests only, and the sibling of [`Self::forget_op`]: no drop pattern on
+ /// a link produces this, because the header and the body land in the same
+ /// append.
+ pub fn forget_body(&self, op: u64) -> bool {
+ #[cfg(debug_assertions)]
+ let _guard = JournalAccessGuard::new(&self.accessing);
+ let offsets = unsafe { &mut *self.offsets.get() };
+ offsets.remove(&op).is_some()
+ }
+
/// The committed watermark to restore after a restart, mirroring
/// `metadata::recover`. On a solo cluster every appended op commits the
instant
/// it is durable, so the head IS the commit point; otherwise the highest
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 8a9d7c631..373e9fd23 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -5209,9 +5209,14 @@ mod partition_repair_driver_tests {
};
}
- /// The partition-plane prepare a packet carries, if it carries one for
- /// `group`.
- fn prepare_for(packet: &Packet, group: u64) -> Option<PrepareHeader> {
+ /// The prepare a packet carries, if it carries one for `group`. Keyed on
+ /// the group, so `metadata_repair_driver_tests` reads the metadata plane's
+ /// prepares with the same helper.
+ ///
+ /// `pub`, not `pub(super)`, in this and the helpers below: the module is
+ /// private and `#[cfg(test)]`, so both spell the same reach, and
+ /// `clippy::redundant_pub_crate` refuses the narrower one.
+ pub fn prepare_for(packet: &Packet, group: u64) -> Option<PrepareHeader> {
if packet.message.header().command != Command::Prepare {
return None;
}
@@ -5221,7 +5226,7 @@ mod partition_repair_driver_tests {
}
/// Whether a packet is a commit heartbeat for `group`.
- fn is_commit_for(packet: &Packet, group: u64) -> bool {
+ pub fn is_commit_for(packet: &Packet, group: u64) -> bool {
if packet.message.header().command != Command::Commit {
return false;
}
@@ -5231,7 +5236,7 @@ mod partition_repair_driver_tests {
}
/// Whether a packet is a repair request for `group`.
- fn is_request_prepares_for(packet: &Packet, group: u64) -> bool {
+ pub fn is_request_prepares_for(packet: &Packet, group: u64) -> bool {
if packet.message.header().command != Command::RequestPrepares {
return false;
}
@@ -5242,7 +5247,7 @@ mod partition_repair_driver_tests {
}
/// Whether a packet is a repair stream terminator for `group`.
- fn is_repair_done_for(packet: &Packet, group: u64) -> bool {
+ pub fn is_repair_done_for(packet: &Packet, group: u64) -> bool {
if packet.message.header().command != Command::RepairDone {
return false;
}
@@ -5252,7 +5257,7 @@ mod partition_repair_driver_tests {
header.group == group
}
- fn cluster(seed: u64) -> (Simulator, SimClient) {
+ pub fn cluster(seed: u64) -> (Simulator, SimClient) {
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
@@ -5964,6 +5969,713 @@ mod partition_repair_driver_tests {
}
}
+#[cfg(test)]
+mod metadata_repair_driver_tests {
+ //! A backup that missed a committed metadata prepare recovers in Normal
+ //! status, without waiting for a view change.
+ //!
+ //! The metadata twin of `partition_repair_driver_tests`, driven by the
+ //! detector in `tick_metadata`: the same preflight `commit_max` advance
+ //! starves the `Advanced`-gated arm in `on_commit`, and
+ //! `retry_stalled_metadata_repair` re-drives only a session that already
+ //! exists. Every fault here is keyed on `METADATA_GROUP`, since both
+ //! planes share `Prepare` and `Commit` on the same links.
+
+ use super::partition_repair_driver_tests::{
+ cluster, is_commit_for, is_repair_done_for, is_request_prepares_for,
prepare_for,
+ };
+ use super::*;
+ use consensus::Status;
+ use journal::Journal;
+ use packet::Packet;
+ use server_common::sharding::METADATA_GROUP;
+ use std::sync::atomic::{AtomicU64, Ordering};
+
+ /// Chain replication runs 0 -> 1 -> 2 and stops before the primary, so
+ /// replica 2 is the only one whose losses cannot starve the group of
+ /// quorum (see `partition_repair_driver_tests::LAGGING`).
+ const LAGGING: u8 = 2;
+
+ const CLIENT_ID: u128 = 1;
+
+ /// Ops committed 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 stream creation, one round trip's worth.
+ const STEPS_PER_SEND: usize = 12;
+
+ /// Creations issued with the fault standing in the gap test. Long enough
+ /// that prepares keep consuming the `commit_max` advance the heartbeat
+ /// backstop needs; the debounce may elapse mid-produce, which the verdict
+ /// tolerates (the withheld heartbeats mean only the tick driver can arm).
+ const GAP_SENDS: usize = 12;
+
+ /// Creations issued with the fault standing in the eviction and
+ /// walk-starvation tests: few enough (under the debounce) that it fires
+ /// only after the traffic stops, so the floor stamp or the starved walk
+ /// edge is in place before the arm runs.
+ const SHORT_GAP_SENDS: usize = 3;
+
+ /// Quiet ticks for the repair stream to land, kept under
+ /// `NORMAL_HEARTBEAT_TICKS` (500) so no election can be the healer.
+ const QUIET_STEPS: usize = 160;
+
+ /// Budget for the group to settle once the fault is lifted; the drain
+ /// loop breaks on convergence.
+ const DRAIN_STEPS: usize = 600;
+
+ /// Quiet budget for the walk-starvation test: debounce, repair stream,
+ /// then the drain, still under `NORMAL_HEARTBEAT_TICKS`.
+ const STRAND_QUIET_STEPS: usize = 300;
+
+ /// Stall interval for the rotation run, shortened from
+ /// `partitions::REPAIR_RETRY_TICKS` so a whole spent budget plus the gap
+ /// debounce (floored at `shard::REPAIR_GAP_DEBOUNCE_TICKS_MIN`) fits under
+ /// `NORMAL_HEARTBEAT_TICKS`. At the production interval the run would only
+ /// prove that an election heals the gap.
+ const ROTATE_RETRY_TICKS: u32 = 10;
+
+ /// Quiet budget for the rotation run: debounce, the spent budget, then the
+ /// repair stream from the replica it rotates onto.
+ const ROTATE_QUIET_STEPS: usize = 250;
+
+ /// Ticks of healthy load in the no-false-positive test, several debounce
+ /// intervals' worth so the driver gets many chances to arm.
+ const LOAD_TICKS: usize = 4 * partitions::REPAIR_RETRY_TICKS as usize;
+
+ /// Paced at a fraction of a round trip; faster submission only collects
+ /// transient rejections once the prepare pipeline fills.
+ const TICKS_PER_SEND: usize = 4;
+
+ /// Ops the healthy run must have committed for its verdict to mean
+ /// anything: enough to prove the group was live across several debounce
+ /// intervals, not that it was saturated.
+ const COMMITTED_MIN: u64 = 20;
+
+ /// Defines this test's `withhold_one_prepare` chain hook over the static
it
+ /// names: swallow the FIRST metadata prepare, once, and record its op in
+ /// `$withheld_op`.
+ ///
+ /// A macro for the same reason as the partition twin: link hooks are bare
+ /// `fn` pointers, so the body cannot capture, and the statics must stay
+ /// per-test or the siblings in this binary would share one fault.
+ macro_rules! withhold_one_metadata_prepare {
+ ($withheld_op:ident) => {
+ fn withhold_one_prepare(packet: &Packet) -> bool {
+ let Some(header) = prepare_for(packet, METADATA_GROUP) else {
+ return false;
+ };
+ $withheld_op
+ .compare_exchange(0, header.op, Ordering::Relaxed,
Ordering::Relaxed)
+ .is_ok()
+ }
+ };
+ }
+
+ /// Defines this test's primary -> backup hook: withhold this group's
commit
+ /// heartbeats, so the `Advanced` backstop can never run, and withhold
+ /// retransmits of the op `$withheld_op` names.
+ ///
+ /// 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.
+ macro_rules! starve_commit_edge {
+ ($withheld_op:ident) => {
+ fn starve_commit_edge(packet: &Packet) -> bool {
+ if let Some(header) = prepare_for(packet, METADATA_GROUP) {
+ return header.op == $withheld_op.load(Ordering::Relaxed);
+ }
+ is_commit_for(packet, METADATA_GROUP)
+ }
+ };
+ }
+
+ /// `(status, view, commit_min, commit_max)` of one replica's metadata
group.
+ fn metadata_state(sim: &Simulator, replica: u8) -> (Status, u32, u64, u64)
{
+ let metadata = sim.replicas[replica as
usize].shards[0].plane.metadata();
+ let consensus = metadata
+ .consensus
+ .as_ref()
+ .expect("shard 0 owns metadata consensus");
+ (
+ consensus.status(),
+ consensus.view(),
+ consensus.commit_min(),
+ consensus.commit_max(),
+ )
+ }
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn journal_holds(sim: &Simulator, replica: u8, op: u64) -> bool {
+ sim.replicas[replica as usize]
+ .metadata_journal
+ .header(op as usize)
+ .is_some()
+ }
+
+ fn gap_drops(sim: &Simulator, replica: u8) -> u64 {
+ sim.replicas[replica as usize].shards[0]
+ .metrics()
+ .metadata_prepare_gap_drops_value()
+ }
+
+ fn transfer_armed(sim: &Simulator, replica: u8) -> bool {
+ let metadata = sim.replicas[replica as
usize].shards[0].plane.metadata();
+ metadata.consensus.as_ref().is_some_and(|consensus| {
+ consensus.state_transfer_stage() !=
consensus::StateTransferStage::Idle
+ })
+ }
+
+ /// Submit `sends` stream creations to the primary, stepping between each.
+ fn create_streams(sim: &mut Simulator, client: &SimClient, sends: usize,
tag: &str) {
+ for index in 0..sends {
+ let msg = client.create_stream(&format!("{tag}-{index}"));
+ sim.submit_request(client.client_id(), 0, msg.into_generic());
+ for _ in 0..STEPS_PER_SEND {
+ sim.step();
+ }
+ }
+ }
+
+ #[test]
+ fn
given_a_backup_that_dropped_a_committed_metadata_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 so parallel siblings cannot share them.
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+ withhold_one_metadata_prepare!(WITHHELD_OP);
+ starve_commit_edge!(WITHHELD_OP);
+
+ let (mut sim, client) = cluster(0x5EED_0240);
+ sim.register_client_with_primary(&client);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+
+ create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm");
+ let (_, _, warm_commit_min, _) = metadata_state(&sim, LAGGING);
+ 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);
+
+ create_streams(&mut sim, &client, GAP_SENDS, "md-gap");
+
+ let withheld = WITHHELD_OP.load(Ordering::Relaxed);
+ assert_ne!(
+ withheld, 0,
+ "no metadata prepare crossed the chain link, so the fault never
armed"
+ );
+ assert!(
+ gap_drops(&sim, LAGGING) > 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: no commit heartbeat for
+ // this group has reached the replica since the gap opened, so only
+ // the tick driver can have armed the repair.
+ let (status, view, commit_min, _) = metadata_state(&sim, LAGGING);
+ 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, withheld),
+ "op {withheld} was never repaired back into the lagging replica's
WAL"
+ );
+ 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 tail above the
+ // repaired window waits on the heartbeats the fault withheld.
+ *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) = metadata_state(&sim, LAGGING);
+ if commit_min == commit_max {
+ break;
+ }
+ }
+ let (status, view, commit_min, commit_max) = metadata_state(&sim,
LAGGING);
+ 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_metadata_repair_armed_by_the_tick_driver_when_the_floor_is_evicted_should_convert_to_state_transfer()
+ {
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+ withhold_one_metadata_prepare!(WITHHELD_OP);
+ starve_commit_edge!(WITHHELD_OP);
+
+ let (mut sim, client) = cluster(0x5EED_0241);
+ sim.register_client_with_primary(&client);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+
+ create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm");
+
+ *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);
+
+ create_streams(&mut sim, &client, SHORT_GAP_SENDS, "md-gap");
+ assert_ne!(
+ WITHHELD_OP.load(Ordering::Relaxed),
+ 0,
+ "no metadata prepare crossed the chain link, so the fault never
armed"
+ );
+
+ // Move the primary's retention floor past the whole gap window before
+ // the debounce can arm (`SHORT_GAP_SENDS`): the serve path reads only
+ // the snapshot watermark, so the request is answered `RangeEvicted`
+ // (see `stamp_metadata_snapshot`).
+ let primary_commit_min = metadata_state(&sim, 0).2;
+ sim.stamp_metadata_snapshot(0, primary_commit_min);
+
+ for _ in 0..QUIET_STEPS {
+ sim.step();
+ if transfer_armed(&sim, LAGGING) {
+ break;
+ }
+ }
+
+ let (status, view, ..) = metadata_state(&sim, LAGGING);
+ assert_eq!(
+ view, 0,
+ "a view change armed the recovery instead of the tick-armed repair
session"
+ );
+ assert!(
+ transfer_armed(&sim, LAGGING),
+ "the tick-armed repair session hit an evicted floor but never
converted \
+ to a state transfer (status {status:?})"
+ );
+ }
+
+ #[test]
+ fn
given_a_backup_holding_resident_committed_metadata_ops_when_every_walk_edge_is_starved_should_drain_in_normal_status()
+ {
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+ static WITHHELD_DONES: AtomicU64 = AtomicU64::new(0);
+ withhold_one_metadata_prepare!(WITHHELD_OP);
+
+ /// 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 the `RepairDone` that
would
+ /// run the walk never fires.
+ fn starve_walk_edges(packet: &Packet) -> bool {
+ if prepare_for(packet, METADATA_GROUP).is_some() {
+ return true;
+ }
+ if is_repair_done_for(packet, METADATA_GROUP) {
+ WITHHELD_DONES.fetch_add(1, Ordering::Relaxed);
+ return true;
+ }
+ is_commit_for(packet, METADATA_GROUP)
+ }
+
+ let (mut sim, client) = cluster(0x5EED_0242);
+ sim.register_client_with_primary(&client);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+ WITHHELD_DONES.store(0, Ordering::Relaxed);
+
+ create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm");
+ let (_, _, warm_commit_min, _) = metadata_state(&sim, LAGGING);
+ 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);
+
+ create_streams(&mut sim, &client, SHORT_GAP_SENDS, "md-strand");
+
+ let withheld = WITHHELD_OP.load(Ordering::Relaxed);
+ assert_ne!(
+ withheld, 0,
+ "no metadata prepare crossed the chain link, so the fault never
armed"
+ );
+
+ for _ in 0..STRAND_QUIET_STEPS {
+ sim.step();
+ let (_, view, commit_min, commit_max) = metadata_state(&sim,
LAGGING);
+ 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, withheld),
+ "op {withheld} was never repaired back into the lagging replica's
WAL"
+ );
+ let (status, view, commit_min, commit_max) = metadata_state(&sim,
LAGGING);
+ 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, 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"
+ );
+ }
+
+ #[test]
+ fn
given_a_resident_header_with_no_body_when_the_walk_stops_on_it_should_drop_it_for_repair()
{
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+ withhold_one_metadata_prepare!(WITHHELD_OP);
+ starve_commit_edge!(WITHHELD_OP);
+
+ let (mut sim, client) = cluster(0x5EED_0245);
+ sim.register_client_with_primary(&client);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+ sim.replicas[LAGGING as
usize].shards[0].set_repair_retry_ticks(ROTATE_RETRY_TICKS);
+
+ create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm");
+
+ *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);
+
+ create_streams(&mut sim, &client, SHORT_GAP_SENDS, "md-gap");
+ assert_ne!(
+ WITHHELD_OP.load(Ordering::Relaxed),
+ 0,
+ "no metadata prepare crossed the chain link, so the fault never
armed"
+ );
+
+ // Corrupt the op the walk is about to read, once repair has put a
+ // window back but before the walk has consumed it. Injected here
+ // rather than by dropping packets: the header and the body land in the
+ // same append, so no link fault produces this state.
+ let mut doomed = 0;
+ for _ in 0..ROTATE_QUIET_STEPS {
+ sim.step();
+ let (_, view, commit_min, commit_max) = metadata_state(&sim,
LAGGING);
+ if view != 0 {
+ break;
+ }
+ let next_op = commit_min + 1;
+ if commit_min < commit_max && journal_holds(&sim, LAGGING,
next_op) {
+ assert!(
+ sim.replicas[LAGGING as usize]
+ .metadata_journal
+ .forget_body(next_op),
+ "op {next_op} had no body to forget"
+ );
+ doomed = next_op;
+ break;
+ }
+ }
+ assert_ne!(
+ doomed, 0,
+ "the repair window never landed resident above the commit point,
so \
+ there was no walkable op to corrupt"
+ );
+ assert!(
+ journal_holds(&sim, LAGGING, doomed),
+ "op {doomed}'s header must stay resident, or the walk would read
this \
+ as an ordinary hole and repair would refill it unaided"
+ );
+
+ for _ in 0..ROTATE_QUIET_STEPS {
+ sim.step();
+ let (_, view, commit_min, _) = metadata_state(&sim, LAGGING);
+ if view != 0 || commit_min >= doomed {
+ break;
+ }
+ }
+
+ let (status, view, commit_min, commit_max) = metadata_state(&sim,
LAGGING);
+ assert_eq!(
+ view, 0,
+ "a view change healed the log instead of the walk; the test proves
\
+ nothing about the unwalkable entry"
+ );
+ assert_eq!(status, Status::Normal, "the replica left Normal status");
+ assert!(
+ commit_min >= doomed,
+ "the walk never crossed op {doomed}: the repair ingest skips an op
\
+ whose header is resident and `append` refuses the slot under it,
so \
+ the header has to be dropped first. Stopped at {commit_min} of \
+ {commit_max}"
+ );
+ }
+
+ #[test]
+ fn
given_a_repair_peer_that_never_answers_when_the_budget_is_spent_should_re_arm_from_another_replica()
+ {
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+ static ASKED_PRIMARY: AtomicU64 = AtomicU64::new(0);
+ static ASKED_SUCCESSOR: AtomicU64 = AtomicU64::new(0);
+ withhold_one_metadata_prepare!(WITHHELD_OP);
+
+ /// The peer `gap_repair_peer` names for a gap-stopped backup is the
+ /// primary, so the session can only heal by rotating off it.
+ fn blackhole(_packet: &Packet) -> bool {
+ true
+ }
+
+ /// Observers, not faults: every packet passes.
+ fn count_requests_to_primary(packet: &Packet) -> bool {
+ if is_request_prepares_for(packet, METADATA_GROUP) {
+ ASKED_PRIMARY.fetch_add(1, Ordering::Relaxed);
+ }
+ false
+ }
+ fn count_requests_to_successor(packet: &Packet) -> bool {
+ if is_request_prepares_for(packet, METADATA_GROUP) {
+ ASKED_SUCCESSOR.fetch_add(1, Ordering::Relaxed);
+ }
+ false
+ }
+
+ let (mut sim, client) = cluster(0x5EED_0244);
+ sim.register_client_with_primary(&client);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+ ASKED_PRIMARY.store(0, Ordering::Relaxed);
+ ASKED_SUCCESSOR.store(0, Ordering::Relaxed);
+ sim.replicas[LAGGING as
usize].shards[0].set_repair_retry_ticks(ROTATE_RETRY_TICKS);
+
+ create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm");
+ let (_, _, warm_commit_min, _) = metadata_state(&sim, LAGGING);
+ 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);
+ // Everything, not just this group's commit edge: the primary must be
+ // unable to answer the repair it is about to be asked for, while the
+ // chain (0 -> 1 -> 2) keeps carrying the prepares that advance
+ // `commit_max` past the hole.
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(0),
ProcessId::Replica(LAGGING)) =
+ Some(blackhole);
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(LAGGING),
ProcessId::Replica(0)) =
+ Some(count_requests_to_primary);
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(LAGGING),
ProcessId::Replica(1)) =
+ Some(count_requests_to_successor);
+
+ create_streams(&mut sim, &client, SHORT_GAP_SENDS, "md-gap");
+ let withheld = WITHHELD_OP.load(Ordering::Relaxed);
+ assert_ne!(
+ withheld, 0,
+ "no metadata prepare crossed the chain link, so the fault never
armed"
+ );
+
+ for _ in 0..ROTATE_QUIET_STEPS {
+ sim.step();
+ let (_, view, commit_min, _) = metadata_state(&sim, LAGGING);
+ if view != 0 || commit_min >= withheld {
+ break;
+ }
+ }
+
+ let (status, view, commit_min, commit_max) = metadata_state(&sim,
LAGGING);
+ assert_eq!(
+ view, 0,
+ "a view change healed the gap instead of the rotation; the test
proves \
+ nothing about the stall budget"
+ );
+ assert_eq!(status, Status::Normal, "the replica left Normal status");
+ assert!(
+ ASKED_PRIMARY.load(Ordering::Relaxed) > 1,
+ "the session was never re-requested from the silent primary, so
the \
+ stall budget (`partitions::REPAIR_MAX_STALL_RETRIES`) was not
spent \
+ and any rotation below came from somewhere else"
+ );
+ assert!(
+ ASKED_SUCCESSOR.load(Ordering::Relaxed) > 0,
+ "the budget ran out against a peer that cannot answer and the
session \
+ was never re-armed anywhere else; nothing re-drives it, so the
plane \
+ stays gap-stopped at {commit_min} of {commit_max}"
+ );
+ assert!(
+ journal_holds(&sim, LAGGING, withheld),
+ "op {withheld} was never repaired back into the lagging replica's
WAL"
+ );
+ assert!(
+ commit_min >= withheld,
+ "the commit walk never crossed the repaired hole: stopped at \
+ {commit_min}, the withheld op is {withheld}"
+ );
+ }
+
+ /// Regression canary, expected green even without the tick driver: a
+ /// healthy metadata backup walks at every accepted prepare's tail
+ /// (`on_replicate`), so per-tick lag never survives to quiescence and the
+ /// journal-hole half of the predicate is pinned by `gap_detector_tests`
+ /// instead. What this run pins is that the driver stays silent under
+ /// sustained pipelined load.
+ #[test]
+ fn
given_healthy_metadata_traffic_when_no_gap_exists_should_not_arm_repair() {
+ 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, METADATA_GROUP) {
+ REPAIR_REQUESTS.fetch_add(1, Ordering::Relaxed);
+ }
+ false
+ }
+
+ // TWO replicas, as in the partition twin: quorum spans both, so no op
+ // can commit while the backup misses it, and every reordering-induced
+ // gap blocks quorum until retransmit refills it.
+
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_0243,
+ ..packet::PacketSimulatorOptions::default()
+ },
+ );
+ let client = SimClient::new(CLIENT_ID);
+ sim.register_client_with_primary(&client);
+ 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, and the run asserts the load went somewhere,
+ // so a green result cannot come from a workload that never loaded the
+ // group.
+ let mut lag_run = 0u32;
+ let mut longest_lag_run = 0u32;
+ let sample = |sim: &Simulator, lag_run: &mut u32, longest: &mut u32| {
+ let (_, _, commit_min, commit_max) = metadata_state(sim, 1);
+ if commit_min < commit_max {
+ *lag_run += 1;
+ *longest = (*longest).max(*lag_run);
+ } else {
+ *lag_run = 0;
+ }
+ };
+ for tick in 0..LOAD_TICKS {
+ if tick % TICKS_PER_SEND == 0 {
+ let msg = client.create_stream(&format!("healthy-{tick}"));
+ sim.submit_request(client.client_id(), 0, msg.into_generic());
+ }
+ sim.step();
+ sample(&sim, &mut lag_run, &mut longest_lag_run);
+ }
+ for _ in 0..QUIET_STEPS {
+ sim.step();
+ sample(&sim, &mut lag_run, &mut longest_lag_run);
+ }
+
+ let committed = metadata_state(&sim, 1).2;
+ let sends = LOAD_TICKS / TICKS_PER_SEND;
+ assert!(
+ committed >= COMMITTED_MIN,
+ "the backup committed only {committed} ops across {sends} sends,
so the \
+ driver was never ticked over a loaded group"
+ );
+ // Asserted, not merely recorded: on this plane the tail walk in
+ // `on_replicate` leaves a healthy backup caught up at every quiescence
+ // point, so any lag at all is a change in that behaviour rather than
+ // ordinary pipelining. The predicate itself is pinned by
+ // `gap_detector_tests`; what this holds is the premise the repair
check
+ // below rests on.
+ assert_eq!(
+ longest_lag_run, 0,
+ "healthy two-replica metadata traffic left the backup lagging for \
+ {longest_lag_run} consecutive ticks; the run below then proves
nothing \
+ about false positives, and this test should sample residency the
way \
+ the partition twin's walk-starvation run does"
+ );
+ for replica in 0..replica_count {
+ let (status, view, commit_min, commit_max) = metadata_state(&sim,
replica);
+ assert_eq!(
+ (status, view),
+ (Status::Normal, 0),
+ "replica {replica} left view 0 / Normal, so a view change
could \
+ account for repair traffic"
+ );
+ if commit_min < commit_max {
+ assert!(
+ journal_holds(&sim, replica, 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"
+ );
+ }
+}
+
#[cfg(test)]
mod metadata_read_frontier_tests {
//! A client that committed a metadata write and then re-homed onto a
