You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/04 20:54:20 UTC
(iggy) branch master updated: fix(shard): drive partition repair and the commit walk from the tick (#4006)
This is an automated email from the ASF dual-hosted git repository.
hubcio 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 9914f5a81 fix(shard): drive partition repair and the commit walk from the tick (#4006)
9914f5a81 is described below
commit 9914f5a812f22a56dabd1dc9b725d956a40e2185
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Fri Sep 4 22:54:15 2026 +0200
fix(shard): drive partition repair and the commit walk from the tick (#4006)
---
core/configs/src/server_config/cluster.rs | 45 +
core/configs/src/server_config/defaults.rs | 5 +
core/consensus/src/observability.rs | 22 +
.../tests/cluster/parked_frame_redispatch.rs | 19 +-
.../tests/data_integrity/storage_compat.rs | 146 ++-
core/partitions/Cargo.toml | 8 +
core/partitions/src/iggy_partition.rs | 130 +-
core/partitions/src/journal.rs | 105 +-
core/partitions/src/lib.rs | 7 +-
core/partitions/src/types.rs | 28 +
core/server/config.toml | 20 +-
core/server/src/boot/recovery.rs | 42 +-
core/server/src/partition_reconciler.rs | 38 +-
core/server/src/shell.rs | 13 +
core/shard/src/lib.rs | 1247 +++++++++++++++++++-
core/shard/src/metrics.rs | 56 +-
core/shard/src/router.rs | 2 +
core/simulator/Cargo.toml | 4 +
core/simulator/src/lib.rs | 858 ++++++++++++++
scripts/ci/storage-compat.sh | 17 +
20 files changed, 2702 insertions(+), 110 deletions(-)
diff --git a/core/configs/src/server_config/cluster.rs b/core/configs/src/server_config/cluster.rs
index 09b9be3d4..929200f08 100644
--- a/core/configs/src/server_config/cluster.rs
+++ b/core/configs/src/server_config/cluster.rs
@@ -170,6 +170,14 @@ fn default_repair_retry_interval() -> IggyDuration {
SERVER_CONFIG.cluster.repair_retry_interval.parse().unwrap()
}
+fn default_repair_gap_debounce_interval() -> IggyDuration {
+ SERVER_CONFIG
+ .cluster
+ .repair_gap_debounce_interval
+ .parse()
+ .unwrap()
+}
+
/// serde fallback for configs written before the field existed; the value
/// itself lives in `core/server/config.toml` like every other default.
fn default_repair_chunk_max() -> usize {
@@ -264,10 +272,28 @@ pub struct ClusterConfig {
/// partition repair loops. Sizes the retry threshold in consensus ticks.
/// Zero (and the `0` / `disabled` / `unlimited` sentinels, which all parse
/// to zero) is rejected at boot.
+ ///
+ /// Paces streams that are already open only. How long a hole waits before a
+ /// stream is opened for it is `repair_gap_debounce_interval`.
#[serde(default = "default_repair_retry_interval")]
#[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.
+ ///
+ /// 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
+ /// 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
+ /// all parse to zero) is rejected at boot.
+ #[serde(default = "default_repair_gap_debounce_interval")]
+ #[serde_as(as = "DisplayFromStr")]
+ #[config_env(leaf)]
+ pub repair_gap_debounce_interval: IggyDuration,
/// Prepares a peer serves per repair round before the requester walks to
/// the next chunk. Each frame rides the per-peer message-bus queue, so this
/// must stay below `message_bus.peer_queue_capacity` or a full round
@@ -1002,6 +1028,13 @@ impl Validatable<ConfigurationError> for ClusterConfig {
// The repair retry interval sizes a tick threshold that has to advance;
// `0` / `disabled` / `unlimited` all collapse to zero and would wedge
// every stalled repair stream - reject them.
+ if self.repair_gap_debounce_interval.get_duration().is_zero() {
+ eprintln!(
+ "Invalid cluster configuration: cluster.repair_gap_debounce_interval must be \
+ nonzero (it debounces the partition sweep's repair arm)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
if self.repair_retry_interval.get_duration().is_zero() {
eprintln!(
"Invalid cluster configuration: cluster.repair_retry_interval must be nonzero \
@@ -1471,6 +1504,7 @@ mod tests {
),
view_probe_attempts_max: default_view_probe_attempts_max(),
repair_retry_interval: default_repair_retry_interval(),
+ repair_gap_debounce_interval: default_repair_gap_debounce_interval(),
repair_chunk_max: default_repair_chunk_max(),
superblock_wedged_fatal_timeout: default_superblock_wedged_fatal_timeout(),
nodes: Vec::new(),
@@ -1878,6 +1912,7 @@ mod cluster_validate_tests {
),
view_probe_attempts_max: default_view_probe_attempts_max(),
repair_retry_interval: default_repair_retry_interval(),
+ repair_gap_debounce_interval: default_repair_gap_debounce_interval(),
repair_chunk_max: default_repair_chunk_max(),
superblock_wedged_fatal_timeout: default_superblock_wedged_fatal_timeout(),
nodes,
@@ -2005,6 +2040,16 @@ mod cluster_validate_tests {
assert!(c.validate().is_err());
}
+ #[test]
+ fn validate_rejects_zero_repair_gap_debounce_interval() {
+ // Same sentinel family as the retry interval above: `0`, `disabled` and
+ // `unlimited` all parse to zero, and a zero debounce would arm repair
+ // against a single reordered prepare on every partition.
+ let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
+ c.repair_gap_debounce_interval = IggyDuration::new(Duration::ZERO);
+ assert!(c.validate().is_err());
+ }
+
#[test]
fn validate_rejects_zero_repair_chunk_max() {
let mut c = cfg(vec![node("n1", 0), node("n2", 1)]);
diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs
index 303ccd0a3..330f39877 100644
--- a/core/configs/src/server_config/defaults.rs
+++ b/core/configs/src/server_config/defaults.rs
@@ -112,6 +112,11 @@ impl Default for ClusterConfig {
.repair_retry_interval
.parse()
.unwrap(),
+ repair_gap_debounce_interval: SERVER_CONFIG
+ .cluster
+ .repair_gap_debounce_interval
+ .parse()
+ .unwrap(),
repair_chunk_max: SERVER_CONFIG.cluster.repair_chunk_max as usize,
nodes: SERVER_CONFIG
.cluster
diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs
index 4519693d6..87473c3ad 100644
--- a/core/consensus/src/observability.rs
+++ b/core/consensus/src/observability.rs
@@ -402,6 +402,11 @@ pub struct PartitionDiagEvent<'a> {
pub message: &'static str,
pub operation: Option<Operation>,
pub op: Option<u64>,
+ /// This replica's sequencer position when the event fired. Set where `op`
+ /// alone cannot say which side of the frontier the frame landed on: a
+ /// prepare above it is a forward gap, one at or below it is a replay of an
+ /// op already sequenced.
+ pub sequence: Option<u64>,
pub prepare_checksum: Option<u128>,
pub reason: Option<&'a str>,
pub error: Option<Cow<'a, str>>,
@@ -415,6 +420,7 @@ impl<'a> PartitionDiagEvent<'a> {
message,
operation: None,
op: None,
+ sequence: None,
prepare_checksum: None,
reason: None,
error: None,
@@ -433,6 +439,12 @@ impl<'a> PartitionDiagEvent<'a> {
self
}
+ #[must_use]
+ pub const fn with_sequence(mut self, sequence: u64) -> Self {
+ self.sequence = Some(sequence);
+ self
+ }
+
#[must_use]
pub const fn with_prepare_checksum(mut self, prepare_checksum: u128) -> Self {
self.prepare_checksum = Some(prepare_checksum);
@@ -469,6 +481,7 @@ fn emit_partition_diag_error(event: &PartitionDiagEvent<'_>) {
let ctx = event.replica;
let operation = event.operation.map_or("", operation_as_str);
let op = event.op.unwrap_or_default();
+ let sequence = event.sequence.unwrap_or_default();
let prepare_checksum = event.prepare_checksum.unwrap_or_default();
let reason = event.reason.unwrap_or("");
let error = event.error.as_deref().unwrap_or("");
@@ -490,6 +503,7 @@ fn emit_partition_diag_error(event: &PartitionDiagEvent<'_>) {
role = ctx.role.as_str(),
operation,
op,
+ sequence,
prepare_checksum,
reason,
error,
@@ -501,6 +515,7 @@ fn emit_partition_diag_warn(event: &PartitionDiagEvent<'_>) {
let ctx = event.replica;
let operation = event.operation.map_or("", operation_as_str);
let op = event.op.unwrap_or_default();
+ let sequence = event.sequence.unwrap_or_default();
let prepare_checksum = event.prepare_checksum.unwrap_or_default();
let reason = event.reason.unwrap_or("");
let error = event.error.as_deref().unwrap_or("");
@@ -522,6 +537,7 @@ fn emit_partition_diag_warn(event: &PartitionDiagEvent<'_>) {
role = ctx.role.as_str(),
operation,
op,
+ sequence,
prepare_checksum,
reason,
error,
@@ -533,6 +549,7 @@ fn emit_partition_diag_info(event: &PartitionDiagEvent<'_>) {
let ctx = event.replica;
let operation = event.operation.map_or("", operation_as_str);
let op = event.op.unwrap_or_default();
+ let sequence = event.sequence.unwrap_or_default();
let prepare_checksum = event.prepare_checksum.unwrap_or_default();
let reason = event.reason.unwrap_or("");
let error = event.error.as_deref().unwrap_or("");
@@ -554,6 +571,7 @@ fn emit_partition_diag_info(event: &PartitionDiagEvent<'_>) {
role = ctx.role.as_str(),
operation,
op,
+ sequence,
prepare_checksum,
reason,
error,
@@ -565,6 +583,7 @@ fn emit_partition_diag_debug(event: &PartitionDiagEvent<'_>) {
let ctx = event.replica;
let operation = event.operation.map_or("", operation_as_str);
let op = event.op.unwrap_or_default();
+ let sequence = event.sequence.unwrap_or_default();
let prepare_checksum = event.prepare_checksum.unwrap_or_default();
let reason = event.reason.unwrap_or("");
let error = event.error.as_deref().unwrap_or("");
@@ -586,6 +605,7 @@ fn emit_partition_diag_debug(event: &PartitionDiagEvent<'_>) {
role = ctx.role.as_str(),
operation,
op,
+ sequence,
prepare_checksum,
reason,
error,
@@ -597,6 +617,7 @@ fn emit_partition_diag_trace(event: &PartitionDiagEvent<'_>) {
let ctx = event.replica;
let operation = event.operation.map_or("", operation_as_str);
let op = event.op.unwrap_or_default();
+ let sequence = event.sequence.unwrap_or_default();
let prepare_checksum = event.prepare_checksum.unwrap_or_default();
let reason = event.reason.unwrap_or("");
let error = event.error.as_deref().unwrap_or("");
@@ -618,6 +639,7 @@ fn emit_partition_diag_trace(event: &PartitionDiagEvent<'_>) {
role = ctx.role.as_str(),
operation,
op,
+ sequence,
prepare_checksum,
reason,
error,
diff --git a/core/integration/tests/cluster/parked_frame_redispatch.rs b/core/integration/tests/cluster/parked_frame_redispatch.rs
index 3b1a525ed..23bc22be7 100644
--- a/core/integration/tests/cluster/parked_frame_redispatch.rs
+++ b/core/integration/tests/cluster/parked_frame_redispatch.rs
@@ -23,8 +23,9 @@
//! that parked. This test pins the park path positively, and on the replica
//! where getting it wrong creates a replica gap: a client request that never
//! reaches the plane is answered with a retriable status and the SDK replays it,
-//! while a replicated PREPARE has no client behind it and must wait for a later
-//! commit heartbeat to arm repair if this path drops it.
+//! while a replicated PREPARE has no client behind it: nothing re-sends it once
+//! its op has quorum, so the backup gap-stops and waits out `tick_partitions`'
+//! repair debounce before anything refetches it.
//!
//! What makes the window wide on a backup is the commit broadcast. A backup
//! learns a metadata commit from the `commit` field of the next prepare on that
@@ -50,7 +51,7 @@
//! - Every acked message is readable in dense offset order, each producer's own
//! sends stay in the order it made them, and all three replicas hold
//! byte-identical segments. A prepare lost to the gap check leaves a backup
-//! permanently short, since the gap never closes on its own.
+//! short until the repair driver's next pass closes the gap.
//!
//! The harness removes an ambient `RUST_LOG` when this test supplies its explicit
//! logging level, and the log oracle falls back from captured stdout to the
@@ -98,8 +99,13 @@ const DEGRADED_MARKERS: [&str; 3] = [
];
/// `IggyPartition::on_replicate`'s backup gap check. A re-dispatch that appends
-/// behind an op already queued on the inbox surfaces here, and the dropped op
-/// forces repair that correct redispatch ordering should never need.
+/// behind an op already queued on the inbox surfaces here. The dropped op is
+/// refetched by `tick_partitions`' level-triggered repair driver, but only after
+/// its debounce interval, so a re-dispatch that trips this has already stalled
+/// the replica for ~1s and the marker still means the ordering broke.
+///
+/// The metadata plane logs the same string, so the counting below pairs it with
+/// `PARTITION_PLANE_FIELD`.
const GAP_MARKER: &str = "dropping out-of-order prepare (gap)";
const PARTITION_PLANE_FIELD: &str = "plane=\"partitions\"";
@@ -347,7 +353,8 @@ fn assert_no_degraded_park_paths(harness: &TestHarness) {
assert_eq!(
partition_gaps, 0,
"node {node} logged {GAP_MARKER:?}: a re-dispatched prepare lost its arrival \
- position and forced avoidable partition repair"
+ position, and the op it displaced is recoverable only by waiting out the \
+ repair driver's debounce"
);
}
}
diff --git a/core/integration/tests/data_integrity/storage_compat.rs b/core/integration/tests/data_integrity/storage_compat.rs
index 5c7ce6647..ea09f47e0 100644
--- a/core/integration/tests/data_integrity/storage_compat.rs
+++ b/core/integration/tests/data_integrity/storage_compat.rs
@@ -57,17 +57,30 @@
//! unseals the last segment, so a chain that ends on a rotation boundary
//! would only ever hand that path an empty file.
//!
-//! # Structural false positive to avoid
+//! # The configuration is held constant across the swap
//!
-//! The harness forwards every parent `IGGY_*` variable to the server child,
-//! and the server treats an unknown `IGGY_*` name as a `debug_assert`. A pull
-//! request that adds a config leaf AND a test that sets it therefore makes
-//! the BASELINE binary die on startup, which looks like a compatibility
-//! break. Nothing here detects that: `resolve_config_paths` validates the
-//! name against the catalog of the build under test, which is the wrong side
-//! of the swap, so it only proves the name exists on HEAD. Keeping every
-//! override to a name that already exists on the merge base stays a rule the
-//! author has to follow by hand.
+//! Both boots read the BASELINE's `core/server/config.toml`, extracted next
+//! to the baseline binary by `scripts/ci/storage-compat.sh`. Neither side may
+//! fall back to the server's relative default path: figment resolves that by
+//! walking up from the test process's directory, so the BASELINE parses THIS
+//! branch's file, and a key added under a `deny_unknown_fields` table (every
+//! `[cluster]` and `[node]` one) fails its extraction with
+//! `Config(CannotLoadConfiguration)`. That reads as a compatibility break and
+//! is not one.
+//!
+//! One file across the swap leaves the binary as the only variable, and it
+//! pins the two rules the config plane already carries: a field this branch
+//! adds needs its `#[serde(default)]`, because the build under test boots on
+//! a file written before the field existed, and a key this branch takes away
+//! needs its `RelocatedKey` entry. Deleting a key from a
+//! `deny_unknown_fields` table, or relocating one, makes the SECOND boot
+//! refuse the file; strip that key from the extracted copy when it happens.
+//!
+//! The overrides below reach both binaries as `IGGY_*` variables, and
+//! `resolve_config_paths` checks them against the catalog of the build under
+//! test only. [`assert_overrides_known_to_the_baseline`] covers the other
+//! side, where an unknown `IGGY_*` name trips a `debug_assert` and takes the
+//! debug binary down at boot.
//!
//! `IGGY_TEST_VERBOSE` makes the harness inherit the server's stdout instead
//! of capturing it, which would make the tombstone check vacuous. The
@@ -88,13 +101,21 @@ use std::time::{Duration, Instant};
use tokio::time::sleep;
/// Absolute path to the baseline `iggy-server` binary. See
-/// [`baseline_server_binary`] for why a relative value is refused.
+/// [`baseline_path_from_env`] for why a relative value is refused.
///
/// NOT `IGGY_`-prefixed on purpose: the harness forwards every parent
/// `IGGY_*` variable to the server child, where an unknown name trips a
/// `debug_assert` and kills the debug build this test runs against.
const BASELINE_SERVER_ENV: &str = "COMPAT_BASELINE_SERVER";
+/// Absolute path to the baseline's `core/server/config.toml`, which both
+/// boots read. Same naming rule as [`BASELINE_SERVER_ENV`].
+const BASELINE_CONFIG_ENV: &str = "COMPAT_BASELINE_CONFIG";
+
+/// Selects the server's config file, ahead of the relative default path the
+/// module documentation warns about.
+const CONFIG_PATH_ENV: &str = "IGGY_CONFIG_PATH";
+
/// `metadata.journal_slots` floor for `prepare_queue_depth = 32`
/// (`4 * max(64, 32)`). With the 64-op checkpoint margin a checkpoint fires
/// once the journal holds 192 committed ops.
@@ -263,11 +284,17 @@ const GRACEFUL_SHUTDOWN_MARKER: &str = "server shutdown complete";
#[ignore = "needs a baseline iggy-server built from master; run via scripts/ci/storage-compat.sh"]
async fn should_read_back_a_data_directory_written_by_the_baseline_server() {
let baseline = baseline_server_binary();
- let envs = resolve_config_paths(&HashMap::from([(
+ let baseline_config = baseline_server_config();
+ let overrides = HashMap::from([(
"metadata.journal_slots".to_string(),
JOURNAL_SLOTS.to_string(),
- )]))
- .expect("metadata.journal_slots resolves against the live config catalog");
+ )]);
+ assert_overrides_known_to_the_baseline(&overrides, &baseline_config);
+ let mut envs = resolve_config_paths(&overrides)
+ .expect("metadata.journal_slots resolves against the live config catalog");
+ // `extra_envs` is re-applied on every `start()`, so the swapped-in binary
+ // reads this same file.
+ envs.insert(CONFIG_PATH_ENV.to_string(), baseline_config.clone());
let mut harness = TestHarness::builder()
.cluster_nodes(1)
@@ -279,7 +306,13 @@ async fn should_read_back_a_data_directory_written_by_the_baseline_server() {
)
.build()
.unwrap();
- harness.start().await.unwrap();
+ harness.start().await.unwrap_or_else(|error| {
+ panic!(
+ "the baseline server did not start. It boots on the merge base's own config file, \
+ so a Config(...) error below means the harness fed it something the merge base \
+ does not carry: {error}"
+ )
+ });
let data_path = harness.server().data_path();
let client = harness.tcp_root_client().await.unwrap();
@@ -677,7 +710,14 @@ async fn should_read_back_a_data_directory_written_by_the_baseline_server() {
// The swap: `None` selects the cargo-built binary of the crate under test.
harness.server_mut().set_executable_path(None);
- harness.restart_server().await.unwrap();
+ harness.restart_server().await.unwrap_or_else(|error| {
+ panic!(
+ "the server under test did not restart on the data directory the baseline wrote. \
+ A Config(...) error below is a configuration break rather than a storage one: \
+ this branch dropped or renamed a key the merge base's config file still sets, or \
+ added a field without a `#[serde(default)]`: {error}"
+ )
+ });
// `tcp_root_client` hands out a client the harness does not own, so the
// reconnect loop inside `restart_server` never touched it. Take a fresh one.
@@ -877,46 +917,88 @@ async fn should_read_back_a_data_directory_written_by_the_baseline_server() {
);
}
-/// Resolve the baseline binary to an absolute, existing path.
+/// Resolve one of the baseline artefacts to an absolute, existing path.
///
-/// The absolute requirement is not cosmetic. `ServerHandle::start` only spawns
-/// the configured path directly when it has more than one component or already
-/// exists; a bare name that does not exist falls through to
+/// The absolute requirement is not cosmetic for the binary. `ServerHandle::start`
+/// only spawns the configured path directly when it has more than one component
+/// or already exists; a bare name that does not exist falls through to
/// `Command::cargo_bin`, which resolves the binary of the crate under test. So
/// a single-component value would boot the HEAD build while the test reported
/// it as the baseline, comparing master against master and passing forever.
/// The same silent-green family as running a stale binary.
-fn baseline_server_binary() -> String {
- let raw = std::env::var(BASELINE_SERVER_ENV).unwrap_or_else(|_| {
+fn baseline_path_from_env(variable: &str, artefact: &str) -> String {
+ let raw = std::env::var(variable).unwrap_or_else(|_| {
panic!(
- "{BASELINE_SERVER_ENV} is unset. It must be an ABSOLUTE path to the iggy-server \
- binary built from master. Without it this test would boot the build under test \
- twice and prove nothing, so it refuses to run. Use scripts/ci/storage-compat.sh, \
- which builds the baseline and exports the variable."
+ "{variable} is unset. It must be an ABSOLUTE path to the baseline {artefact}. \
+ Without it this test would run entirely on the build under test and prove nothing, \
+ so it refuses to run. Use scripts/ci/storage-compat.sh, which builds the baseline \
+ and exports both variables."
)
});
let path = PathBuf::from(&raw);
assert!(
path.is_absolute(),
- "{BASELINE_SERVER_ENV}={raw:?} is relative. It must be an ABSOLUTE path: a bare binary \
- name that does not exist on disk is silently resolved as the build under test, which \
- would compare master against master and report green forever."
+ "{variable}={raw:?} is relative. It must be an ABSOLUTE path: a bare name that does not \
+ exist on disk is silently resolved as the build under test, which would compare master \
+ against master and report green forever."
);
let resolved = fs::canonicalize(&path).unwrap_or_else(|error| {
panic!(
- "{BASELINE_SERVER_ENV}={raw:?} does not resolve: {error}. It must point at an \
- iggy-server binary built from master."
+ "{variable}={raw:?} does not resolve: {error}. It must point at the baseline \
+ {artefact}."
)
});
assert!(
resolved.is_file(),
- "{BASELINE_SERVER_ENV}={raw:?} resolves to {}, which is not a file.",
+ "{variable}={raw:?} resolves to {}, which is not a file.",
resolved.display()
);
resolved.display().to_string()
}
+fn baseline_server_binary() -> String {
+ baseline_path_from_env(BASELINE_SERVER_ENV, "iggy-server binary")
+}
+
+fn baseline_server_config() -> String {
+ baseline_path_from_env(BASELINE_CONFIG_ENV, "core/server/config.toml")
+}
+
+/// Refuse an override the baseline binary cannot read.
+///
+/// Every override travels to both binaries as an `IGGY_*` variable, and an
+/// unknown name trips a `debug_assert` in the server's env provider, which
+/// takes the debug baseline down at boot with no mention of the variable.
+/// `resolve_config_paths` cannot catch that: it validates against the catalog
+/// of the build under test, which is the wrong side of the swap. The baseline
+/// config file spells out every knob that build has, so it stands in for the
+/// catalog here.
+fn assert_overrides_known_to_the_baseline(overrides: &HashMap<String, String>, config: &str) {
+ let raw = fs::read_to_string(config)
+ .unwrap_or_else(|error| panic!("reading the baseline config {config}: {error}"));
+ let doc: toml::Value = toml::from_str(&raw)
+ .unwrap_or_else(|error| panic!("parsing the baseline config {config}: {error}"));
+
+ for path in overrides.keys() {
+ // The `system.` fallback mirrors `resolve_config_paths`.
+ let prefixed = format!("system.{path}");
+ let known = [path.as_str(), prefixed.as_str()]
+ .into_iter()
+ .any(|candidate| config_key(&doc, candidate).is_some());
+ assert!(
+ known,
+ "the override '{path}' names no key in the baseline config {config}, so the \
+ baseline server has no such config leaf and dies on the variable at boot. Drive \
+ the behaviour through a key that already exists on the merge base."
+ );
+ }
+}
+
+fn config_key<'a>(doc: &'a toml::Value, path: &str) -> Option<&'a toml::Value> {
+ path.split('.').try_fold(doc, |node, key| node.get(key))
+}
+
/// Options of the topic that carries the segment chain. Every key but
/// `compression_algorithm` is sent, so that one must come back derived while
/// the rest come back explicit.
diff --git a/core/partitions/Cargo.toml b/core/partitions/Cargo.toml
index 1f509cd2e..3a1b48cec 100644
--- a/core/partitions/Cargo.toml
+++ b/core/partitions/Cargo.toml
@@ -37,6 +37,14 @@ publish = false
# requests it. No production caller.
simulator = []
+# Fault injection for the harness (`IggyPartition::inject_commit_failure`):
+# makes a local commit fail the way a full disk does, so the shard sweep's
+# fence path can be driven over an in-memory partition. Requested ONLY by
+# `core/simulator`'s dev-dependencies, which the v3 resolver keeps out of any
+# non-test build, so the branch it gates cannot reach a shipped binary the way
+# a `[dependencies]` feature would through unification.
+fault-injection = []
+
[dependencies]
ahash = { workspace = true }
bytemuck = { workspace = true }
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 19db69e4c..2128ae517 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -30,7 +30,7 @@ use crate::poll_plan::{
};
use crate::segment::Segment;
use crate::state_transfer::{PartitionTransferSession, PendingTransferRearm};
-use crate::types::{FatalCommit, RepairConclusion, RepairSession};
+use crate::types::{COMMIT_WALK_OPS_MAX, FatalCommit, RepairConclusion, RepairSession};
use crate::{
AppendResult, Partition, PartitionOffsets, PartitionsConfig, PollQueryResult, PollingArgs,
PollingConsumer,
@@ -154,6 +154,36 @@ where
/// set when the recovery handshake finds this replica behind the group's
/// commit frontier, cleared when `RepairDone` completes the walk.
pub repair: Option<RepairSession>,
+ /// Consecutive shard-sweep ticks this partition has been seen gap-stopped
+ /// (committed ops it cannot walk to, because the op at its commit frontier
+ /// plus one is missing). Debounces the sweep's level-triggered repair arm,
+ /// and is spent by whichever site opens the repair session.
+ ///
+ /// `Cell` for the same reason as [`Self::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>,
+ /// Prepares the backup gap check destroyed since the shard last drained
+ /// the count, folded into `partition_prepare_gap_drops_total` there.
+ /// Replicated traffic has no client to answer and retransmit skips ops that
+ /// already reached quorum, so nothing else records that the frame existed.
+ /// It counts what the ordering check destroyed, which is neither the holes
+ /// nor only them: a gap opened by the last prepare of a burst leaves it at
+ /// zero, and a duplicate delivery of an op this replica already sequenced
+ /// bumps it without any hole existing. Zero proves nothing, and nonzero is
+ /// a reason to look at the `sequence` field on the drop log, which is what
+ /// separates the two shapes.
+ ///
+ /// `Cell`: the shard drains it once per sweep, off the same shared borrow
+ /// the rest of the tick reads the partition through, so a `&mut` here would
+ /// buy a second lookup of the same group per tick.
+ prepare_gap_drops: Cell<u64>,
+ /// Fault injection for the harness, armed by
+ /// [`Self::inject_commit_failure`]. Behind a feature only the simulator's
+ /// DEV-dependencies turn on, so no build that ships this crate compiles the
+ /// branch it gates.
+ #[cfg(any(test, feature = "fault-injection"))]
+ injected_commit_failure: bool,
/// Highest message offset recovered from segments at boot (`None` when
/// the partition booted empty). Repaired batches at or below this line
/// are already persisted and counted; the flush and commit paths skip
@@ -276,6 +306,10 @@ where
/// [`Self::note_transfer_rearm_scheduled`]; livelock across attempts is
/// bounded by [`Self::transfer_failures`] and its exponential backoff.
transfer_attempts: u32,
+ /// Consecutive stalled re-requests on the live repair session, against
+ /// [`crate::types::REPAIR_MAX_STALL_RETRIES`]. Survives the session, so rotating
+ /// the peer cannot reset it; cleared by real progress.
+ repair_attempts: u32,
/// CONSECUTIVE transfer failures of any class (decode, spill, install,
/// peer-unavailable, stall exhaustion). Deliberately NOT keyed on the
/// offered generation: a committing primary advances its generation
@@ -514,6 +548,10 @@ where
consumer_offset_enforce_fsync: false,
runtime_options: TopicRuntimeOptions::default(),
repair: None,
+ gap_ticks: Cell::new(0),
+ prepare_gap_drops: Cell::new(0),
+ #[cfg(any(test, feature = "fault-injection"))]
+ injected_commit_failure: false,
recovered_durable_offset: None,
installed_frontier: None,
fatal: None,
@@ -533,6 +571,7 @@ where
offset_reservation_lease: u64::from(crate::DEFAULT_OFFSET_RESERVATION_LEASE),
transfer: None,
transfer_attempts: 0,
+ repair_attempts: 0,
transfer_failures: 0,
transfer_refusals: 0,
transfer_rearm: None,
@@ -1710,6 +1749,51 @@ where
self.write_superblock_advancing(superblock, 0, claim).await
}
+ /// Take and clear the gap-drop count ([`Self::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)
+ }
+
+ /// Read the gap-drop count without clearing it, so a test can prove the
+ /// count was still buffered on the partition at the moment it was removed.
+ #[cfg(any(test, feature = "fault-injection"))]
+ #[must_use]
+ pub const fn prepare_gap_drops(&self) -> u64 {
+ self.prepare_gap_drops.get()
+ }
+
+ /// Fail the next local commit of a committed `SendMessages` op the way a
+ /// full disk fails it, fencing the partition.
+ ///
+ /// The commit path's own fence is covered by a `/dev/full` writer in this
+ /// crate's tests, but the shard sweep that must OBSERVE the fence runs over
+ /// in-memory partitions in the simulator, where no device can be made to
+ /// fail.
+ #[cfg(any(test, feature = "fault-injection"))]
+ pub const fn inject_commit_failure(&mut self) {
+ self.injected_commit_failure = true;
+ }
+
+ /// Burn one repair stall round; `true` once the budget is exhausted and the
+ /// session should be re-armed against a different peer.
+ ///
+ /// On the PARTITION, not the session, for the same reason the transfer
+ /// budget is: the rotation mints a new session, which would otherwise reset
+ /// the count and re-target forever without ever giving up on the ring.
+ #[must_use = "the bool is the rotate verdict; dropping it disables the stall budget"]
+ pub const fn burn_repair_attempt(&mut self) -> bool {
+ self.repair_attempts += 1;
+ self.repair_attempts > crate::types::REPAIR_MAX_STALL_RETRIES
+ }
+
+ /// Real repair progress (any in-window frame from the serving peer): reset
+ /// the stall budget, so it bounds CONSECUTIVE stalls rather than the ones a
+ /// long healthy stream accumulates.
+ pub const fn note_repair_progress(&mut self) {
+ self.repair_attempts = 0;
+ }
+
/// Burn one transfer stall round; `true` once the budget is exhausted.
/// Lives on the partition, not the session, so a re-minted session
/// cannot reset it (see [`Self::transfer_attempts`]).
@@ -3159,7 +3243,7 @@ where
return;
}
- let journal_holds_op = self.log.journal().inner.header_by_op(header.op).is_some();
+ let journal_holds_op = self.log.journal().inner.holds_op(header.op);
if journal_holds_op {
// Retransmit after downstream flap: durable here but commit
// hasn't caught up. Re-forward + re-ACK so primary's view of
@@ -3251,6 +3335,10 @@ where
let is_backup = self.consensus().is_follower();
if is_backup {
if header.op != current_op + 1 {
+ // `sequence` is what separates the two shapes this line covers:
+ // a forward gap (op above the sequencer, the hole the repair
+ // driver closes) and a retransmit of an op this replica already
+ // sequenced. Without it they read identically.
emit_partition_diag(
tracing::Level::WARN,
&PartitionDiagEvent::new(
@@ -3258,8 +3346,11 @@ where
"dropping out-of-order prepare (gap)",
)
.with_operation(header.operation)
- .with_op(header.op),
+ .with_op(header.op)
+ .with_sequence(current_op),
);
+ self.prepare_gap_drops
+ .set(self.prepare_gap_drops.get().saturating_add(1));
return;
}
} else {
@@ -3486,6 +3577,24 @@ where
}
}
+ /// Apply the committed prefix this replica can reach, from the pipeline if
+ /// the primary populated one and from the journal otherwise.
+ ///
+ /// The journal half is bounded by [`COMMIT_WALK_OPS_MAX`], for EVERY caller
+ /// and not just the tick sweep: `on_commit`, `StartView` adoption, the
+ /// post-transfer tail and the post-repair walk all reach this with the same
+ /// resident backlog behind them, and all four run on the shard pump. The
+ /// pipeline half is left alone, being the primary's own in-flight window,
+ /// which the pipeline depth already caps.
+ ///
+ /// Nothing is lost by stopping early: the run is re-derived from
+ /// `commit_min` on the next call, and the sweep's walk predicate stays true
+ /// until the group drains, so the next tick resumes exactly here. The one
+ /// caller with no next tick is the shutdown drain, and it is covered twice
+ /// over: [`Self::flush_committed_messages`] persists the committed prefix
+ /// by bytes rather than by walk, and what stays un-applied sits above the
+ /// `commit_min` the superblock records, which is what makes the restarted
+ /// replica ask its peers for it.
#[allow(clippy::future_not_send)]
pub async fn commit_journal(&mut self, config: &PartitionsConfig) {
if self.fatal.is_some() {
@@ -3504,7 +3613,7 @@ where
// double-count against `advance_commit_min`.
let mut drained = drain_committable_prefix(self.consensus());
if drained.is_empty() {
- drained = self.collect_committable_from_journal();
+ drained = self.collect_committable_from_journal(COMMIT_WALK_OPS_MAX);
}
if drained.is_empty() {
return;
@@ -3529,13 +3638,13 @@ where
/// the journal keeps its committed entries until they are flushed
/// (`commit_messages` drains only the committed prefix), so this read finds
/// every committed op while the uncommitted tail stays resident.
- fn collect_committable_from_journal(&self) -> Vec<PipelineEntry> {
+ fn collect_committable_from_journal(&self, max_ops: usize) -> Vec<PipelineEntry> {
let from_op = self.consensus.commit_min() + 1;
let commit_max = self.consensus.commit_max();
self.log
.journal()
.inner
- .committed_headers_from(from_op, commit_max)
+ .committed_headers_from(from_op, commit_max, max_ops)
.into_iter()
.map(PipelineEntry::new)
.collect()
@@ -3843,6 +3952,10 @@ where
}
async fn commit_messages(&mut self, config: &PartitionsConfig) -> Result<(), IggyError> {
+ #[cfg(any(test, feature = "fault-injection"))]
+ if std::mem::take(&mut self.injected_commit_failure) {
+ return Err(IggyError::CannotSaveMessagesToSegment);
+ }
self.commit_messages_inner(config, false).await
}
@@ -5773,11 +5886,12 @@ where
return;
}
// Any in-window frame proves the stream is alive; only silence
- // should age the stall counter.
+ // should age the stall counter or spend the rotation budget.
if let Some(session) = self.repair.as_mut() {
session.idle_ticks = 0;
+ self.repair_attempts = 0;
}
- if self.log.journal().inner.header_by_op(header.op).is_some() {
+ if self.log.journal().inner.holds_op(header.op) {
return;
}
let applied = if header.operation == Operation::SendMessages {
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index 45f3b6d1d..1bffcd16f 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -762,6 +762,30 @@ where
headers.iter().find(|header| header.op == op).copied()
}
+ /// Whether `op` is resident, in O(log n) instead of `header_by_op`'s scan.
+ /// Callers that only need presence must use this: a miss is the common case
+ /// on the residency checks, and a miss is exactly when the scan walks the
+ /// whole vec.
+ ///
+ /// It answers off `op_to_storage_offset`, so it is `header_by_op(op).is_some()`
+ /// everywhere except INSIDE [`Self::append_with_meta`], which pushes the
+ /// header before the storage write and inserts the offset after it: a task
+ /// that interleaves at that await sees the header without the offset and is
+ /// answered `false`. Both are cleared together at every clear site
+ /// ([`Self::commit`], [`Self::evict_prefix`], the restore path), so that
+ /// window is the only divergence.
+ ///
+ /// The window is unreachable for this journal: `PartitionJournalMemStorage`
+ /// writes to memory and its `write_at` never yields, so no task can observe
+ /// the half-inserted state. That is what lets `apply_repaired_prepare` lean
+ /// on this for idempotence, where a false negative would re-journal an op
+ /// the log already holds. A future yielding `Storage` has to insert the
+ /// offset before the write, or move that check back to the header vec.
+ pub fn holds_op(&self, op: u64) -> bool {
+ let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() };
+ op_to_storage_offset.contains_key(&op)
+ }
+
/// Presence and message-carrying shape of the repair window `(floor, to_op]`
/// in ONE pass over the header vec.
///
@@ -842,27 +866,64 @@ where
}
/// Headers for the contiguous op run `from_op ..= commit_max`, in op order,
- /// stopping at the first missing op. A replication gap must not be skipped:
- /// the caller advances `commit_min` strictly by one, so a hole would break
- /// that contract. Headers are append-ordered, which is op-ascending on a
- /// backup, so this is a single linear scan: drop ops below `from_op`, take
- /// while contiguous, stop at the first gap or past `commit_max`.
- pub fn committed_headers_from(&self, from_op: u64, commit_max: u64) -> Vec<PrepareHeader> {
+ /// stopping at the first missing op and at `limit` headers. A replication
+ /// gap must not be skipped: the caller advances `commit_min` strictly by
+ /// one, so a hole would break that contract.
+ ///
+ /// `limit` bounds what one caller commits in a single pass. The run is
+ /// re-derived from the caller's own `commit_min` every time, so a truncated
+ /// answer is resumed, not lost.
+ pub fn committed_headers_from(
+ &self,
+ from_op: u64,
+ commit_max: u64,
+ limit: usize,
+ ) -> Vec<PrepareHeader> {
// Walk by OP, not by append position: after a rejoin the journal
// interleaves live tail ops (which arrive while repair is still
// streaming) with repaired window ops, so append order is no longer
// op-ascending and a positional sequential scan would break at the
// first interleave boundary forever.
- let mut result = Vec::new();
- let mut op = from_op;
- while op <= commit_max {
- let Some(header) = self.header_by_op(op) else {
- break;
- };
- result.push(header);
- op += 1;
+ //
+ // ONE pass over the headers, like `repaired_window_shape`, not a
+ // `header_by_op` probe per op: that probe is itself a linear scan, so
+ // probing walked the window against the whole vec, and the walk this
+ // feeds runs per group per tick over a rejoin's entire backlog.
+ if from_op > commit_max {
+ return Vec::new();
}
- result
+ let headers = unsafe { &*self.headers.get() };
+ // Sized off the RUN the caller asked for, capped by the headers that
+ // could possibly cover it. Sizing off `headers.len()` alone would make
+ // a one-op run allocate a slot per resident header, and `commit_max` is
+ // a cluster frontier this replica may be arbitrarily far below, so
+ // neither bound can be dropped. `saturating_add` because `commit_max`
+ // is `u64::MAX` on a saturated frontier, where `+ 1` would panic in
+ // debug and wrap to an empty span (a walk that never resumes) in
+ // release.
+ let span = (commit_max - from_op)
+ .saturating_add(1)
+ .min(limit as u64)
+ .min(headers.len() as u64);
+ // `span` is a `min` of two `usize`-derived values on a 64-bit target,
+ // so the narrowing is lossless; `slot` is then compared against it as
+ // a `u64` BEFORE any cast, so nothing depends on the cast to bound it.
+ #[allow(clippy::cast_possible_truncation)]
+ let span = span as usize;
+ let mut slots: Vec<Option<PrepareHeader>> = vec![None; span];
+ for header in headers {
+ if header.op < from_op || header.op - from_op >= span as u64 {
+ continue;
+ }
+ #[allow(clippy::cast_possible_truncation)]
+ let slot = (header.op - from_op) as usize;
+ // First writer wins, matching the `header_by_op` probe this
+ // replaces (`find` returns the earliest match).
+ slots[slot].get_or_insert(*header);
+ }
+ // Stops at the first hole: a replication gap must not be skipped, or
+ // `advance_commit_min`'s sequential contract breaks.
+ slots.into_iter().map_while(|header| header).collect()
}
/// Oldest message offset still resident in the in-memory journal, if
@@ -1555,7 +1616,7 @@ mod tests {
// Contiguous run from op 1 stops before the missing op 3 even though
// op 4 is resident and within commit_max.
- let run = journal.committed_headers_from(1, 4);
+ let run = journal.committed_headers_from(1, 4, usize::MAX);
let ops: Vec<u64> = run.iter().map(|header| header.op).collect();
assert_eq!(
ops,
@@ -1564,9 +1625,19 @@ mod tests {
);
assert!(
- journal.committed_headers_from(5, 4).is_empty(),
+ journal.committed_headers_from(5, 4, usize::MAX).is_empty(),
"from_op past commit_max yields nothing"
);
+
+ // The limit truncates the run rather than skipping ahead in it, so the
+ // caller resumes at the op it stopped on.
+ let bounded = journal.committed_headers_from(1, 4, 1);
+ let ops: Vec<u64> = bounded.iter().map(|header| header.op).collect();
+ assert_eq!(ops, vec![1], "the limit must cut the run at its front");
+ assert!(
+ journal.committed_headers_from(1, 4, 0).is_empty(),
+ "a zero limit commits nothing"
+ );
}
/// Three-message batch with the broker append time (`base_timestamp`)
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 013bc6de6..059c795a0 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -61,9 +61,10 @@ pub use segment::Segment;
use server_common::Message;
pub use server_common::send_messages::{IggyMessage, IggyMessageHeader, IggyMessages};
pub use types::{
- AppendResult, FatalCommit, Fragment, PartitionOffsets, PartitionPathLayout, PartitionsConfig,
- PollFragments, PollQueryResult, PollingArgs, PollingConsumer, REPAIR_RETRY_TICKS,
- RepairConclusion, RepairSession, SendMessagesResult,
+ AppendResult, COMMIT_WALK_OPS_MAX, FatalCommit, Fragment, PartitionOffsets,
+ PartitionPathLayout, PartitionsConfig, PollFragments, PollQueryResult, PollingArgs,
+ PollingConsumer, REPAIR_MAX_STALL_RETRIES, REPAIR_RETRY_TICKS, RepairConclusion, RepairSession,
+ SendMessagesResult,
};
/// A partition's message log, named so a caller can carry one across a rebuild.
diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs
index 4762b772a..f2a3bd791 100644
--- a/core/partitions/src/types.rs
+++ b/core/partitions/src/types.rs
@@ -251,6 +251,34 @@ impl Default for PartitionOffsets {
/// from the serving peer.
pub const REPAIR_RETRY_TICKS: u32 = 100;
+/// Consecutive stalled re-requests tolerated before a repair session is
+/// abandoned and re-armed against a different peer.
+///
+/// A session pins its peer, and `repair.is_some()` fences the sweep's detector
+/// and every edge-triggered arming site while it stands, so a peer that went
+/// away (or never was reachable, which the gap-stopped-primary rotation can
+/// pick) would wedge the group harder than having no session at all. Three
+/// rounds is ~3s at the default retry interval: long enough that an ordinary
+/// dropped frame is re-requested rather than re-targeted, short enough that a
+/// dead peer costs one debounce interval, not a view change.
+pub const REPAIR_MAX_STALL_RETRIES: u32 = 3;
+
+/// Committed ops one journal-driven commit walk applies before returning to the
+/// pump.
+///
+/// `commit_journal`'s journal half returns the whole resident
+/// `(commit_min, commit_max]` run, and applying it reaches a segment flush per
+/// batch with no await the pump can interleave: after a rejoin that is the
+/// entire backlog in one call, with the consensus tick stopped behind it. Every
+/// caller is re-driven (the shard sweep's walk backstop is level-triggered and
+/// stays true until the group drains), so a truncated walk resumes on the next
+/// tick instead of losing anything.
+///
+/// Sized as a batch big enough that an ordinary group drains in one pass and
+/// small enough that a full one cannot hold the pump for the view-change
+/// escalation window.
+pub const COMMIT_WALK_OPS_MAX: usize = 64;
+
/// One in-flight journal-repair stream for a partition group.
#[derive(Debug, Clone, Copy)]
pub struct RepairSession {
diff --git a/core/server/config.toml b/core/server/config.toml
index 336265186..3bae72a0c 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -655,9 +655,27 @@ view_probe_attempts_max = 5
# remaining window from the serving peer (duration). Repair frames are
# fire-and-forget over the lossy bus, so a session with no retry wedges forever
# on a single dropped frame. Paces both the metadata and partition repair loops;
-# must be nonzero.
+# must be nonzero. Paces streams that are already OPEN only: how long a hole
+# 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.
+#
+# 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.
+repair_gap_debounce_interval = "1s"
+
# Prepares a peer serves per repair round before the requester walks to the next
# chunk (integer). Each frame rides the per-peer message-bus queue, so this must
# stay strictly below message_bus.peer_queue_capacity or a full round overruns
diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs
index 5d21e1f7c..27d7a45b7 100644
--- a/core/server/src/boot/recovery.rs
+++ b/core/server/src/boot/recovery.rs
@@ -24,7 +24,7 @@ use crate::server_error::ServerError;
use crate::session_manager::SessionManager;
use crate::shell::{
ServerMetadata, ServerShard, ShellHandlers, ShellShardHandle, consensus_timers,
- repair_retry_ticks,
+ repair_gap_debounce_ticks, repair_retry_ticks,
};
use configs::server::ServerConfig;
use consensus::{
@@ -285,6 +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_superblock_wedged_fatal_failures(superblock_wedged_fatal_failures(config));
shard.set_served_segment_cache_bytes_max(
config
@@ -939,6 +940,45 @@ mod tests {
);
}
+ /// The floor is prose in `config.toml` ("50 consensus ticks (500ms at the
+ /// 10ms tick)"), which is the number an operator sizes
+ /// `repair_gap_debounce_interval` against. Nothing else would notice it
+ /// drifting.
+ #[test]
+ fn documented_gap_debounce_floor_matches_the_shard_constant() {
+ assert_eq!(
+ shard::PARTITION_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)
+ * shard::CONSENSUS_TICK_INTERVAL.as_millis(),
+ 500,
+ "the floor is no longer 500ms; core/server/config.toml states that \
+ figure under [cluster] repair_gap_debounce_interval"
+ );
+ }
+
+ #[test]
+ fn default_repair_gap_debounce_interval_matches_partitions_constant() {
+ // Same lockstep the retry interval keeps: the shipped config.toml value
+ // is what an un-configured replica and the simulator run on, and the
+ // shard's own default is the compile-time constant.
+ let config_default = configs::cluster::ClusterConfig::default()
+ .repair_gap_debounce_interval
+ .get_duration()
+ .as_millis();
+ let built_in =
+ u128::from(partitions::REPAIR_RETRY_TICKS) * shard::CONSENSUS_TICK_INTERVAL.as_millis();
+ assert_eq!(
+ config_default, built_in,
+ "[cluster] repair_gap_debounce_interval default drifted from the shard's \
+ compile-time debounce"
+ );
+ }
+
#[test]
fn default_repair_chunk_max_matches_shard_constant() {
// Belt and suspenders with the static assert above: that pins the
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index 4db62a2b7..e35f62378 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -109,17 +109,33 @@
//! `park_dropped` when it parked and then lost its namespace. A prepare has
//! nobody to answer, so the counter is the only record it existed.
//!
+//! A shed or discarded *prepare* is not recovered by retransmit once its op has
+//! reached quorum (`consensus::retransmit_targets` skips entries with
+//! `ok_quorum_received`), so the backup gap-stops. `tick_partitions` opens a
+//! repair session for it: its level-triggered detector arms once a partition has
+//! been gap-stopped for `[cluster] repair_gap_debounce_interval`, independently of the
+//! edge-triggered arming sites (`StartView` adoption, the commit heartbeat, the
+//! post-transfer tail), whose edges a produce stream can starve. A repair range
+//! the primary has already evicted escalates to partition state transfer. The park policy
+//! above still shrinks the exposure to a genuinely exhausted byte budget and a
+//! namespace this shard cannot serve; the driver bounds how long either costs,
+//! and `partition_prepare_gap_drops_total` counts what reached the gap check.
+//!
//! # Known gaps
//!
-//! Recorded here because both were previously carried as a TODO on the
+//! Recorded here because they were previously carried as a TODO on the
//! materialization barrier this module used to promise, and the barrier is gone
//! (see above) while these are not:
//!
-//! A shed prepare is not retransmitted once its op reached quorum, but it is not
-//! stranded until a view change. A later `CommitMessage` that advances the
-//! backup's frontier runs `maybe_request_partition_repair`; an evicted repair
-//! range escalates to partition state transfer. The park policy still avoids
-//! manufacturing that recovery work unless a byte budget is already spent.
+//! A shed prepare is not retransmitted once its op reached quorum, and the
+//! commit heartbeat is no answer to it: a follower advances `commit_max` from
+//! every prepare header before the gap check drops the frame, so under produce
+//! the heartbeat lands as `Accepted` and the backstop inside that branch never
+//! runs. What closes it is the level-triggered sweep above, which means the
+//! residual exposure is its debounce (`[cluster] repair_gap_debounce_interval`,
+//! floored at 500ms) plus the arm cap under a correlated fault, during which the
+//! backup serves a short prefix. The park policy still avoids manufacturing that
+//! recovery work unless a byte budget is already spent.
//!
//! TODO(krishna): `serves_committed_incarnation` and the park stamp both call
//! `Streams::created_revision_for_namespace`, now on the per-request fence path.
@@ -3532,9 +3548,11 @@ mod tests {
drop(inbox);
}
- /// The replicated-prepare shape, which no other test covers and where both
- /// park critical are worst: a prepare has no client, so discarding it forces
- /// the backup to wait for a later commit heartbeat and journal repair.
+ /// The replicated-prepare shape, which no other test covers and where the
+ /// park path's stakes are highest: a prepare has no client, so
+ /// `deny_parked_client_request` no-ops on it and anything that discards it
+ /// loses committed data silently, recoverable only once `tick_partitions`'
+ /// repair driver notices the gap it left.
///
/// A backup receives the prepare before its own metadata commits (so the frame
/// parks unstamped), then applies the commit and materialises. The prepare must
@@ -3574,7 +3592,7 @@ mod tests {
shard.redispatched_frame_count(),
1,
"the parked prepare must be staged for re-dispatch; discarding it is \
- an avoidable gap, since a prepare has no client to retry it"
+ silent committed-data loss, since a prepare has no client to answer"
);
let (served, answered) = drain_inbox(&inbox);
assert_eq!(
diff --git a/core/server/src/shell.rs b/core/server/src/shell.rs
index 661c34b9c..18145cb68 100644
--- a/core/server/src/shell.rs
+++ b/core/server/src/shell.rs
@@ -200,3 +200,16 @@ pub(crate) fn repair_retry_ticks(config: &ServerConfig) -> u32 {
))
.unwrap_or(u32::MAX)
}
+
+/// `[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
+/// 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.
+pub(crate) fn repair_gap_debounce_ticks(config: &ServerConfig) -> u32 {
+ u32::try_from(duration_to_ticks(
+ config.cluster.repair_gap_debounce_interval.get_duration(),
+ ))
+ .unwrap_or(u32::MAX)
+}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index f44c40f3e..8181831fd 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1486,6 +1486,27 @@ where
/// `[cluster] repair_retry_interval` at bootstrap.
repair_retry_ticks: Cell<u32>,
+ /// Live repair sessions on this shard, republished by every partition sweep
+ /// and incremented as sessions open, for
+ /// [`PARTITION_REPAIRS_INFLIGHT_MAX`]. A tally rather than a scan because
+ /// the arming funnel holds a `&mut` to one partition, which a scan over the
+ /// 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>,
+
+ /// 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.
+ ///
+ /// The sweep visits namespaces in `BTreeMap` order, so without a carried
+ /// cursor the leading groups would spend the whole budget on every pass and
+ /// the tail would never be reached. See [`rotate_sweep_to_cursor`].
+ partition_walk_cursor: Cell<Option<IggyNamespace>>,
+
/// Consecutive metadata superblock write failures tolerated before the
/// process fail-stops. Defaults to 0 (disabled) so the simulator and tests
/// keep a wedged-but-fenced replica alive; the server arms it from
@@ -1667,6 +1688,9 @@ 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),
+ 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),
@@ -1681,6 +1705,14 @@ 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);
+ }
+
/// Arm the superblock fail-stop bound (consecutive write failures).
/// Called once per shard at bootstrap; the simulator and tests keep the
/// disabled default (0) so a wedged-but-fenced replica stays observable
@@ -2113,6 +2145,9 @@ 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),
+ 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),
@@ -2413,7 +2448,15 @@ where
self.discard_parked_partition_frames(namespace);
self.metrics.record_partition_removed();
confirmed_remove = true;
- if removed.is_none() {
+ if let Some(partition) = removed {
+ // Tail of the gap-drop count. The tick sweep drains it
+ // per pass, but `get_by_ns` stops answering the moment
+ // the reconciler tombstones the namespace, so whatever
+ // the last pass before the tombstone left would go to
+ // the floor with the partition value.
+ self.metrics
+ .record_partition_prepare_gap_drops(partition.take_prepare_gap_drops());
+ } else {
tracing::trace!(
shard = self_shard_id,
namespace_raw = namespace.inner(),
@@ -3221,8 +3264,9 @@ where
/// buffer exists to absorb -- the partition primary materialises and
/// replicates as soon as its own metadata commits, well before a lagging
/// backup applies the same commit. Treating that as "prior incarnation"
- /// destroys live traffic and forces the backup to recover a gap that the
- /// park path could have delivered directly.
+ /// destroys live traffic: a replicated prepare has no client to answer, so
+ /// it would be dropped and the backup left gap-stopped until
+ /// `tick_partitions`' level-triggered driver notices and repairs it.
/// The residual is unchanged from before the stamp existed -- a frame parked
/// while the namespace was absent, then recreated under a new incarnation,
/// is served against the replacement -- and closing it needs a wire-level
@@ -3485,12 +3529,13 @@ where
let existing = pending.get_mut(&namespace);
let parked_len = existing.as_ref().map_or(0, |entry| entry.frames.len());
let namespace_bytes = existing.as_ref().map_or(0, |entry| entry.bytes);
- // A prepare is never shed on a byte budget before the budget is spent.
- // It has no client to retry it, and recovery requires a later commit
- // heartbeat to expose the gap and arm same-view repair. A request costs
- // only a retry, so it is refused the moment admitting it would cross a
- // budget. This caps prepare residency at one frame of overshoot per
- // budget (worst case
+ // A prepare is never shed on a byte budget. No client to answer, and
+ // recovery is slow: `consensus::retransmit_targets` skips an op that
+ // already reached quorum, so shedding one gap-stops the backup until
+ // `tick_partitions`' driver repairs it, where shedding a request costs
+ // one retry. A request is refused the moment admitting it
+ // would cross a budget; a prepare only once one is already spent. Caps
+ // prepare residency at one frame of overshoot per budget (worst case
// `MAX_PARKED_BYTES` + `max_message_size`, 80 MiB per shard) instead of
// at the budget, and is what makes an oversize frame parkable at all.
let namespace_budget_spent = parked_len > 0
@@ -6825,12 +6870,16 @@ 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();
// 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
// the pump's owned scratch (as `process_loopback` does) so no
// partitions-plane borrow is held across the tick `.await`.
namespace_scratch.extend(partitions.namespaces().copied());
+ // Resume where the last sweep ran out of walk budget, so the cap below
+ // spreads over every group instead of replaying the same prefix.
+ rotate_sweep_to_cursor(namespace_scratch, self.partition_walk_cursor.get());
// Pre-pass: issue every group's pending superblock write CONCURRENTLY.
// A cluster-wide view change makes every group on this shard need one in
@@ -6890,12 +6939,45 @@ where
// mid-sweep is seen on the next tick, the same latency a capped arm
// already accepts.
let mut transfers_inflight: Option<usize> = None;
+ // Live repair sessions seen this pass, published at the end for the arm
+ // fn's concurrency cap.
+ let mut repairs_live = 0usize;
+ // Repair sessions this sweep has opened, against
+ // `PARTITION_REPAIR_ARMS_PER_TICK_MAX`.
+ let mut repair_arms = 0usize;
+ // Commit walks this sweep has run, against
+ // `PARTITION_WALKS_PER_TICK_MAX`.
+ let mut walks = 0usize;
+ // First group the WALK budget turned away, which becomes the next
+ // sweep's starting point. Recorded per SWEEP, not per group: the cursor
+ // only has to name where the budget ran out, and every group after it
+ // is reached on the next pass by the rotation above.
+ //
+ // The walk cap alone, because only its eligible set regenerates: a
+ // walked group is walk-stalled again on the next produce, so a fixed
+ // start would re-spend the budget on the same prefix forever. An ARMED
+ // group leaves the gap-stopped set for the life of its session, so the
+ // arm cap drains its own queue in namespace order with no cursor, and
+ // letting an arm deferral move this one would pull the walk's resume
+ // point backwards and break the `ceil(groups / cap)` bound below.
+ //
+ // It always advances: the walk budget is fresh at the group the sweep
+ // starts on, so the first group can never be the deferred one, and a
+ // cursor that stood still would re-skip the same tail forever.
+ let mut walk_cursor: Option<IggyNamespace> = None;
let mut fatal: Option<FatalCommit> = None;
for namespace in namespace_scratch.drain(..) {
let Some(partition) = partitions.get_by_ns(&namespace) else {
continue;
};
+ // Ahead of the fence check and every `continue` below: the count is
+ // the only record those prepares existed, and a partition that
+ // fences here never ticks again.
+ let gap_drops = partition.take_prepare_gap_drops();
+ if gap_drops > 0 {
+ self.metrics.record_partition_prepare_gap_drops(gap_drops);
+ }
// A fenced partition must not tick: its consensus would emit
// view-scoped sends for a log the cluster has already passed.
if let Some(fault) = partition.fatal() {
@@ -6999,7 +7081,7 @@ where
);
continue;
}
- partition.repair.as_mut().and_then(|session| {
+ let due = partition.repair.as_mut().and_then(|session| {
if !consensus_normal {
return None;
}
@@ -7016,11 +7098,51 @@ where
cluster,
self_id,
))
- })
+ });
+ // A session pins its peer and fences every arming site while it
+ // stands, so a peer that cannot answer wedges the group harder
+ // than having no session at all -- and the gap-stopped-primary
+ // rotation above 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.
+ due.map(|stalled| (stalled, partition.burn_repair_attempt()))
};
- if let Some((peer, nonce, from_op, to_op, cluster, self_id)) = stalled
+ if let Some(((peer, nonce, from_op, to_op, cluster, self_id), rotate)) = stalled
&& from_op <= to_op
{
+ if rotate {
+ let Some(partition) = partitions.get_mut_by_ns(&namespace) else {
+ continue;
+ };
+ let consensus = partition.consensus();
+ let primary = consensus.primary_index(consensus.view());
+ let next_peer =
+ next_transfer_peer(self_id, peer, consensus.replica_count(), primary);
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ peer,
+ next_peer,
+ from_op,
+ to_op,
+ "partition repair stalled past its retry budget; re-arming from \
+ another replica"
+ );
+ partition.repair = None;
+ partition.note_repair_progress();
+ if next_peer == peer {
+ // The ring had nobody else to offer (a solo group, or a
+ // two-replica group whose only peer is the one that
+ // went quiet). Dropping the session is still the right
+ // move: it unfences the detector, which re-arms after
+ // its debounce and logs the state each interval.
+ continue;
+ }
+ self.maybe_request_partition_repair(partition, next_peer)
+ .await;
+ continue;
+ }
tracing::info!(
shard = self.id,
namespace_raw = namespace.inner(),
@@ -7041,6 +7163,160 @@ where
.await;
}
+ // Level-triggered gap detector. Every other partition arming site
+ // is edge-triggered and the edges are starvable: the commit-heartbeat
+ // backstop needs `CommitOutcome::Advanced`, and a follower has
+ // already advanced `commit_max` from each prepare header in
+ // `replicate_preflight` before the gap check dropped the prepare, so
+ // under produce load the heartbeat lands as `Accepted` and the gap
+ // wedges until an unrelated view change. Those edges stay the fast
+ // path; this is the ~1s floor under them.
+ //
+ // Runs entirely on the shared borrow: the in-flight scan below reads
+ // every partition on the shard, so it must not run under a `&mut`,
+ // and `gap_ticks` is a `Cell` for exactly that reason.
+ let (walk_stalled, arm_peer) = {
+ let Some(partition) = partitions.get_by_ns(&namespace) else {
+ continue;
+ };
+ // Live sessions, tallied on the borrow this sweep already takes
+ // rather than by a scan: `maybe_request_partition_repair` reads
+ // the tally to refuse over the concurrency cap, and it is called
+ // from four edge sites that hold a `&mut` and so could not scan
+ // at all. Counted here, after the stall block above has cleared
+ // whatever finished, so the tally the NEXT sweep and every edge
+ // site in between read is one full pass old at worst.
+ if partition.repair.is_some() {
+ repairs_live += 1;
+ }
+ let probe = partition_gap_probe(partition);
+ let walk_stalled = partition_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)
+ && 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,
+ );
+ 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.
+ // Restart the debounce so this repeats at its
+ // interval rather than every tick.
+ partition.gap_ticks.set(0);
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ commit_min = probe.commit_min,
+ commit_max = probe.commit_max,
+ "partition is gap-stopped below its own commit frontier with no \
+ peer to repair from"
+ );
+ None
+ } else {
+ Some(peer)
+ }
+ }
+ };
+ (walk_stalled, arm_peer)
+ };
+ if let Some(peer) = arm_peer {
+ let Some(partition) = partitions.get_mut_by_ns(&namespace) else {
+ continue;
+ };
+ // Logged by `maybe_request_partition_repair` at info, with the
+ // same fields plus the window it settled on. A refusal there
+ // (the concurrency cap, or a guard the probe cannot see) spends
+ // no rate budget and leaves the debounce satisfied, so the group
+ // is due again next pass.
+ if self.maybe_request_partition_repair(partition, peer).await {
+ repair_arms += 1;
+ }
+ }
+
+ // Capped like the repair arm, and for the same reason: a node-wide
+ // rejoin leaves every group on the shard walk-stalled in the same
+ // tick, and each walk reaches a segment flush. Undebounced, though
+ // -- the predicate guarantees the walk finds at least the next op,
+ // so it cannot spin: `partition_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.
+ if walk_stalled {
+ if walks >= PARTITION_WALKS_PER_TICK_MAX {
+ // Deferred, not dropped: this group becomes the next
+ // sweep's starting point, so a shard with more owed walks
+ // than budget drains them round-robin. Without the cursor
+ // the leading groups would take the whole budget every
+ // pass and the tail would keep its committed ops resident
+ // indefinitely.
+ walk_cursor.get_or_insert(namespace);
+ } else {
+ let config = partitions.config();
+ let Some(partition) = partitions.get_mut_by_ns(&namespace) else {
+ continue;
+ };
+ let consensus = partition.consensus();
+ // Debug, not info: an in-flight repair journals bodies
+ // without walking them, so this is the steady state for the
+ // whole duration of a rejoin and would be one line per group
+ // per tick.
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ commit_min = consensus.commit_min(),
+ commit_max = consensus.commit_max(),
+ "partition commit walk parked over resident committed ops; resuming"
+ );
+ partition.commit_journal(config).await;
+ walks += 1;
+ // Re-read, because the aggregate above was sampled BEFORE
+ // this walk: a local commit failure fences the partition
+ // here, and reporting the stale verdict would let the pump
+ // keep serving a divergent replica until the next tick
+ // noticed.
+ if let Some(fault) = partition.fatal() {
+ if fatal.is_none() {
+ fatal = Some(fault.clone());
+ }
+ continue;
+ }
+ }
+ }
+
// Transfer stall retry: descriptor and chunk frames are
// fire-and-forget, so a lost one must not wedge the session (and
// the rejoin behind it) forever. Budget-bounded: a peer that died
@@ -7129,6 +7405,17 @@ where
}
}
+ // The first group the walk cap turned away is where the next sweep
+ // enters the group set; a sweep that turned nobody away resets to the
+ // front, since a stale cursor would keep re-entering at a point no cap
+ // chose.
+ self.partition_walk_cursor.set(walk_cursor);
+ // Republished from this pass, arms included: the arm fn has been
+ // incrementing it as sessions opened, and this is the recount that
+ // retires whatever completed.
+ self.partition_repairs_inflight
+ .set(repairs_live + repair_arms);
+
fatal
}
@@ -7965,13 +8252,48 @@ where
/// `maybe_request_metadata_repair`: no-op unless Normal, not
/// transferring, behind the frontier, and no session live.
#[allow(clippy::future_not_send)]
- async fn maybe_request_partition_repair(&self, partition: &mut IggyPartition<B, SB>, peer: u8)
+ /// Open a journal-repair session against `peer`, if this partition needs
+ /// one and the shard has room for it. `true` when a session was recorded.
+ ///
+ /// THE funnel: the tick sweep and the four edge-triggered sites
+ /// (`StartView` adoption, the commit heartbeat, the post-transfer tail, the
+ /// post-repair walk) all arrive here, so the concurrency ceiling and the
+ /// debounce reset live here rather than in any one caller. A node-wide view
+ /// change drives `on_start_view` for every group at once, which is exactly
+ /// the burst the sweep's own rate cap would not see.
+ async fn maybe_request_partition_repair(
+ &self,
+ partition: &mut IggyPartition<B, SB>,
+ peer: u8,
+ ) -> bool
where
B: MessageBus,
{
let consensus = partition.consensus();
if !consensus.is_normal() || consensus.is_transferring() || partition.repair.is_some() {
- return;
+ return false;
+ }
+ // Read, never scanned: the tally is republished by each sweep (see
+ // `tick_partitions`), so it is at worst one tick stale, which is all a
+ // concurrency ceiling needs. Callers here hold a `&mut` to one
+ // partition, so a scan over the plane would alias it.
+ if self.partition_repairs_inflight.get() >= PARTITION_REPAIRS_INFLIGHT_MAX {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = consensus.group(),
+ peer,
+ "partition repair not armed: shard is at its live-session ceiling"
+ );
+ return false;
+ }
+ // Never against self. The session is recorded below BEFORE the send,
+ // and a self-addressed `RequestPrepares` cannot be delivered (the
+ // replica registry holds no entry for this node), so the session would
+ // stand forever: `repair_finished` needs a `commit_min` only the reply
+ // can advance, the stall retry re-sends to the same peer, and
+ // `repair.is_some()` fences every other arming site meanwhile.
+ if peer == consensus.replica() {
+ return false;
}
// The window ends at the group head when suffix bodies are missing,
// not at the commit point. A backup that adopted a StartView holds
@@ -7986,27 +8308,29 @@ where
let commit_lag = consensus.commit_min() < commit_to_op;
let head = consensus.sequencer().current_sequence();
if !commit_lag && head <= commit_to_op {
- return;
+ return false;
}
- let canonical_suffix = consensus
- .with_pending_view_log(|pending| pending_covers_suffix(pending, commit_to_op, head))
- .unwrap_or(false);
- let missing_suffix = canonical_suffix
- && !partition
- .log
- .journal()
- .inner
- .repaired_window_shape(commit_to_op, head)
- .complete;
+ let missing_suffix = partition_missing_suffix(partition);
if !commit_lag && !missing_suffix {
- return;
+ return false;
}
let nonce = iggy_common::random_id::get_uuid();
let from_op = consensus.commit_min() + 1;
- let fetch_to_op = if missing_suffix { head } else { commit_to_op };
+ // The widening is for the replica that is LEVEL with the commit
+ // frontier and short of bodies above it. Widening while a commit lag
+ // stands would ask for `(commit_min, head]` -- the whole committed
+ // prefix this replica already holds, refetched -- and the suffix is
+ // reached anyway once the lag closes, on the arm after it.
+ let fetch_to_op = if commit_lag { commit_to_op } else { head };
let cluster = consensus.cluster();
let self_id = consensus.replica();
let namespace = consensus.group();
+ // Spent here for the same reason the ceiling is: the four edge-triggered
+ // sites never touch it, so a short edge-armed repair would leave the
+ // count saturated and hand the next real gap an arm on its first tick.
+ partition.gap_ticks.set(0);
+ self.partition_repairs_inflight
+ .set(self.partition_repairs_inflight.get() + 1);
partition.repair = Some(partitions::RepairSession {
nonce,
view: consensus.view(),
@@ -8023,6 +8347,7 @@ where
from_op,
commit_to_op,
fetch_to_op,
+ peer,
"partition behind the group frontier; requesting repair"
);
self.send_request_prepares(
@@ -8035,6 +8360,7 @@ where
namespace,
)
.await;
+ true
}
/// Receiver side of a partition descriptor: accept the manifest, adopt
@@ -9286,6 +9612,302 @@ fn repair_serve_ceiling(requested_to_op: u64, commit_max: u64, head: u64) -> u64
requested_to_op.min(commit_max.max(head))
}
+/// Repair sessions the partition tick sweep will OPEN per pass.
+///
+/// The RATE half of the pair: it spreads the cost of OPENING sessions, while
+/// [`PARTITION_REPAIRS_INFLIGHT_MAX`] bounds how many stand at once. One arm is
+/// a `RequestPrepares` plus a repair stream the serving peer walks
+/// synchronously, and a node-wide gap (a rejoin, a lossy link) makes every group
+/// on this shard due in the same tick.
+///
+/// Over-cap groups stay due with their debounce satisfied and arm on a later
+/// pass. No cursor: an armed group leaves the gap-stopped set for the life of
+/// its session, so the queue drains in namespace order on its own, and letting
+/// a deferred arm move the walk cursor would pull the walk's resume point
+/// backwards.
+const PARTITION_REPAIR_ARMS_PER_TICK_MAX: usize = 3;
+
+/// Live repair sessions this shard will hold at once.
+///
+/// The concurrency ceiling the rate cap above is not: without it a node-wide
+/// rejoin puts every group's stream in flight within `groups / arms` ticks, and
+/// each one is a window the SERVING peer walks on its own pump, so the cost
+/// lands on a node that has nothing wrong with it. Sized at twice
+/// [`IggyShard::PARTITION_TRANSFERS_INFLIGHT_MAX`]: a repair streams journal
+/// entries the peer already holds resident, where a transfer reads and hashes
+/// whole segments, so more of them fit in the same serving budget.
+///
+/// Applied inside `maybe_request_partition_repair`, not at any one caller: the
+/// four edge-triggered sites arm from frame handlers, and a node-wide view
+/// change drives `on_start_view` for every group on the shard at once, which no
+/// per-sweep budget can see. Over-cap groups stay gap-stopped with their
+/// debounce satisfied, so they arm as sessions complete.
+const PARTITION_REPAIRS_INFLIGHT_MAX: usize = 8;
+
+/// Commit walks the partition tick sweep will RUN per pass.
+///
+/// Same correlated-fan-out argument as the repair arm, and the walk is the
+/// costlier half: `commit_journal` reaches `commit_messages`, which flushes a
+/// segment and fsyncs under `enforce_fsync`.
+///
+/// The two caps together are what bound the tick: this one bounds how many
+/// groups a sweep walks, [`partitions::COMMIT_WALK_OPS_MAX`] bounds how far
+/// each walk goes (for every caller of `commit_journal`, not just this one),
+/// and the product is the sweep's worst case. Deliberately NOT the
+/// superblock pre-pass's number: that one runs its fan-out CONCURRENTLY under
+/// `join_all` and drains every group in the same body, while these walks are
+/// serial and what is over budget waits for the next tick.
+///
+/// Capping cannot starve a partition: the walk carries no debounce counter and
+/// clears its own predicate (a walk either advances `commit_min` or fences the
+/// partition), and [`rotate_sweep_to_cursor`] resumes the next sweep at the
+/// first group this one turned away, so the eligible set drains in
+/// `ceil(groups / cap)` ticks however many groups are owed at once.
+const PARTITION_WALKS_PER_TICK_MAX: usize = 16;
+
+/// Floor under the gap detector's debounce, in ticks.
+///
+/// The debounce reads `[cluster] repair_gap_debounce_interval`, and
+/// `duration_to_ticks` floors that at one tick. One tick of lag is ordinary
+/// pipelining, so without a floor of its own a shortened interval would arm
+/// repair against a single reordered prepare.
+///
+/// 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;
+
+/// What the tick sweep reads off one partition to decide whether it is
+/// gap-stopped. Split out so the guards, the debounce and the per-tick cap are
+/// testable without a shard, a bus, or a journal.
+///
+/// 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`.
+#[allow(clippy::struct_excessive_bools)]
+#[derive(Debug, Clone, Copy)]
+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.
+ recovery_owned: bool,
+ commit_min: u64,
+ commit_max: u64,
+ /// Whether `commit_min + 1` is resident in the local journal.
+ ///
+ /// Read only when the guards above already hold, and `false` otherwise:
+ /// both predicates test the lag first, so a probe that fails it is
+ /// answered without touching the journal at all. See
+ /// [`partition_gap_probe`].
+ next_op_resident: bool,
+ /// Whether this replica adopted suffix headers above `commit_max` whose
+ /// bodies never arrived. Its own recovery shape, disjoint from the lag
+ /// 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`].
+ missing_suffix: bool,
+}
+
+/// Whether this replica holds committed ops it cannot walk to, because the op
+/// one past its commit frontier is missing from its journal.
+///
+/// The journal-hole half is not redundant: a follower advances `commit_max`
+/// from every prepare header in `replicate_preflight`, so `commit_min <
+/// commit_max` is transiently true on every healthy pipelined tick and a bare
+/// lag test would arm repair against ordinary produce.
+const fn partition_is_gap_stopped(probe: &GapProbe) -> bool {
+ if !probe.normal || probe.transferring || probe.recovery_owned {
+ return false;
+ }
+ // The lag decides first, and a walkable lag wins outright. A replica that
+ // is BOTH short of a suffix and behind its own frontier would otherwise arm
+ // over `(commit_min, head]` -- refetching a committed prefix it already
+ // holds resident -- and would claim this predicate and the walk at once.
+ // The walk closes the lag within a tick or two (the suffix cannot commit
+ // meanwhile, so `commit_max` stands still), and the suffix arms on the pass
+ // after that.
+ if probe.commit_min < probe.commit_max {
+ return !probe.next_op_resident;
+ }
+ probe.missing_suffix
+}
+
+/// The gap predicate's disjoint sibling, not its complement: everything the
+/// walk needs is resident, it just never ran (a heartbeat carrying a known
+/// commit is `Accepted`, and an idle group offers no other edge).
+///
+/// The two split on `next_op_resident` while a lag stands, and
+/// [`partition_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 {
+ probe.normal
+ && !probe.transferring
+ && probe.commit_min < probe.commit_max
+ && probe.next_op_resident
+}
+
+/// What the debounce says about one partition on one sweep.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum GapArm {
+ /// Not gap-stopped, or gap-stopped for less than the debounce.
+ NotDue,
+ /// Due, but this sweep's arm budget is spent. The debounce stays satisfied,
+ /// 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.
+ Deferred,
+ /// Open a repair session now.
+ Arm,
+}
+
+/// Count one sweep tick against `gap_ticks` and answer whether this partition
+/// may arm repair now.
+///
+/// Level-triggered, because every edge-triggered arming site is starvable: the
+/// commit-heartbeat backstop fires only on `CommitOutcome::Advanced`, and under
+/// sustained produce the prepares consume the advance in preflight before the
+/// gap check drops them, so the heartbeat lands as `Accepted` and the gap wedges
+/// until an unrelated view change.
+///
+/// `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(
+ probe: &GapProbe,
+ gap_ticks: &mut u32,
+ debounce_ticks: u32,
+ budget_available: bool,
+) -> GapArm {
+ if !partition_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
+ } else {
+ debounce_ticks
+ };
+ *gap_ticks = gap_ticks.saturating_add(1);
+ if *gap_ticks < debounce_ticks {
+ return GapArm::NotDue;
+ }
+ if budget_available {
+ GapArm::Arm
+ } else {
+ GapArm::Deferred
+ }
+}
+
+/// 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
+/// namespace order, so a shard whose leading groups stay eligible would spend
+/// the whole budget on them every pass and never reach the tail. `cursor` names
+/// the first group a cap turned away last pass, so every eligible group is
+/// served within `ceil(groups / cap)` sweeps.
+///
+/// A cursor whose namespace was removed meanwhile resumes at its successor, and
+/// one past the last namespace wraps to the front. `None` means the previous
+/// sweep turned nobody away.
+fn rotate_sweep_to_cursor(namespaces: &mut [IggyNamespace], cursor: Option<IggyNamespace>) {
+ let Some(cursor) = cursor else {
+ return;
+ };
+ debug_assert!(
+ namespaces.is_sorted(),
+ "the sweep snapshot must be in namespace order for the cursor to resume in it",
+ );
+ // `partition_point` answers in `0..=len`, and `rotate_left(len)` is the
+ // no-op that wraps a cursor past the last namespace back to the front.
+ namespaces.rotate_left(namespaces.partition_point(|namespace| *namespace < cursor));
+}
+
+/// Whether this replica holds adopted suffix HEADERS above `commit_max` whose
+/// bodies never arrived.
+///
+/// The shape `maybe_request_partition_repair` widens its window for, read here
+/// so the sweep's detector and the arm agree by construction. A backup that
+/// adopted a `StartView` withholds its ack for those ops until the body is
+/// journaled, and the primary's retransmit is dropped by the backup gap check
+/// because adoption already advanced the sequencer to the head: nothing else
+/// delivers them, and the group wedges one op below its head.
+///
+/// Ordered cheapest-first, because it runs per group per tick: no suffix at all
+/// is one comparison, and a suffix nobody adopted is one `Option` check. Only a
+/// group that has both pays the header-vec walk.
+fn partition_missing_suffix<B, SB>(partition: &IggyPartition<B, SB>) -> bool
+where
+ B: MessageBus,
+ SB: SuperblockStore,
+{
+ let consensus = partition.consensus();
+ let commit_max = consensus.commit_max();
+ let head = consensus.sequencer().current_sequence();
+ if head <= commit_max {
+ return false;
+ }
+ let canonical_suffix = consensus
+ .with_pending_view_log(|pending| pending_covers_suffix(pending, commit_max, head))
+ .unwrap_or(false);
+ canonical_suffix
+ && !partition
+ .log
+ .journal()
+ .inner
+ .repaired_window_shape(commit_max, head)
+ .complete
+}
+
+/// Read the gap probe off a live partition.
+fn partition_gap_probe<B, SB>(partition: &IggyPartition<B, SB>) -> GapProbe
+where
+ B: MessageBus,
+ SB: SuperblockStore,
+{
+ let consensus = partition.consensus();
+ let commit_min = consensus.commit_min();
+ let commit_max = consensus.commit_max();
+ let recovery_owned = partition.transfer.is_some()
+ || partition.transfer_rearm.is_some()
+ || partition.repair.is_some();
+ let normal = consensus.is_normal();
+ let transferring = consensus.is_transferring();
+ // Residency last, and only once the guards both predicates share already
+ // hold. This runs for every group on the shard on every tick, and the
+ // caught-up steady state (`commit_min == commit_max`) would otherwise pay
+ // a journal lookup whose answer both predicates discard.
+ let next_op_resident = normal
+ && !transferring
+ && commit_min < commit_max
+ && partition
+ .log
+ .journal()
+ .inner
+ .holds_op(commit_min.saturating_add(1));
+ // Same discipline, one guard deeper: the suffix test walks the header vec,
+ // so it runs only for a group that HAS an unfinished suffix and already
+ // owes nothing else.
+ let missing_suffix =
+ normal && !transferring && !recovery_owned && partition_missing_suffix(partition);
+ GapProbe {
+ normal,
+ transferring,
+ recovery_owned,
+ commit_min,
+ commit_max,
+ next_op_resident,
+ missing_suffix,
+ }
+}
+
/// Whether the parked `StartView` log names every op in the uncommitted
/// suffix `(commit_max, head]`, in descending order. Only this canonical list
/// makes fetching bodies above the commit point safe.
@@ -10519,3 +11141,574 @@ mod superblock_fail_stop_tests {
assert!(superblock_wedged(121, 120));
}
}
+
+#[cfg(test)]
+mod sweep_scheduler_tests {
+ //! Fairness of the partition sweep's per-tick caps.
+ //!
+ //! The caps exist so a node-wide rejoin cannot put every group's walk (each
+ //! reaching a segment flush) into one tick body. They are only ACCEPTABLE
+ //! because the sweep resumes where the WALK budget ran out: the snapshot is
+ //! in ascending namespace order, so a fixed start would spend every pass on
+ //! the same leading groups and leave the tail holding committed ops it can
+ //! never walk to.
+ //!
+ //! Both budgets are modelled, because they share the sweep: the arm cap
+ //! runs first and can turn groups away ahead of the walks, and the walk
+ //! fairness bound has to survive that. What keeps them independent is that
+ //! the arm cap moves no cursor, which is the property these runs pin.
+
+ use super::{
+ IggyNamespace, PARTITION_REPAIR_ARMS_PER_TICK_MAX, PARTITION_REPAIRS_INFLIGHT_MAX,
+ PARTITION_WALKS_PER_TICK_MAX, rotate_sweep_to_cursor,
+ };
+
+ /// Comfortably past `PARTITION_WALKS_PER_TICK_MAX`, and deliberately not a
+ /// multiple of it, so the wrap lands mid-snapshot on most passes.
+ const GROUPS: usize = 40;
+
+ /// Every third group is gap-stopped rather than walk-stalled. The two
+ /// predicates are disjoint below the commit frontier, so a group is in one
+ /// set or the other, and this spreads the arm-capped ones through the
+ /// snapshot instead of parking them in one block.
+ const fn is_gap_stopped(partition: usize) -> bool {
+ partition.is_multiple_of(3)
+ }
+
+ fn namespaces() -> Vec<IggyNamespace> {
+ (0..GROUPS)
+ .map(|partition| IggyNamespace::new(1, 1, partition))
+ .collect()
+ }
+
+ /// Per-group tallies, indexed by partition id. No `Default`: empty vecs
+ /// next to a `new` that sizes them by `GROUPS` would panic on first index.
+ struct Served {
+ walks: Vec<u32>,
+ arms: Vec<u32>,
+ }
+
+ impl Served {
+ fn new() -> Self {
+ Self {
+ walks: vec![0; GROUPS],
+ arms: vec![0; GROUPS],
+ }
+ }
+ }
+
+ /// Sweeps a session stays open for before it completes. Long enough that
+ /// the live-session ceiling actually binds (it is reached on the third
+ /// sweep at three arms a pass), so the model spends passes waiting on
+ /// capacity the way a real rejoin does.
+ const SESSION_SWEEPS: u32 = 6;
+
+ /// Repair state per group, standing in for `partition.repair` (which fences
+ /// a group out of the gap-stopped set while it stands) and for the hole
+ /// itself (which a completed session closes, so the group stops being
+ /// eligible rather than arming again).
+ struct Repairs {
+ live: Vec<Option<u32>>,
+ done: Vec<bool>,
+ }
+
+ impl Repairs {
+ fn new() -> Self {
+ Self {
+ live: vec![None; GROUPS],
+ done: vec![false; GROUPS],
+ }
+ }
+
+ /// Age every open session by one sweep, closing the gaps that finish.
+ fn retire(&mut self) {
+ for (partition, session) in self.live.iter_mut().enumerate() {
+ let Some(remaining) = session else {
+ continue;
+ };
+ *remaining -= 1;
+ if *remaining == 0 {
+ *session = None;
+ self.done[partition] = true;
+ }
+ }
+ }
+
+ fn live_count(&self) -> usize {
+ self.live.iter().filter(|session| session.is_some()).count()
+ }
+
+ fn is_due(&self, partition: usize) -> bool {
+ is_gap_stopped(partition) && !self.done[partition] && self.live[partition].is_none()
+ }
+ }
+
+ /// One sweep of `tick_partitions`' scheduling: rotate to the carried walk
+ /// cursor, retire whatever finished, then spend the rate cap, the live
+ /// ceiling and the walk budget in snapshot order. Answers with the cursor
+ /// this sweep leaves behind.
+ fn sweep(
+ cursor: Option<IggyNamespace>,
+ served: &mut Served,
+ repairs: &mut Repairs,
+ ) -> Option<IggyNamespace> {
+ repairs.retire();
+ let mut snapshot = namespaces();
+ rotate_sweep_to_cursor(&mut snapshot, cursor);
+ let mut walks = 0;
+ let mut arms = 0;
+ let mut walk_deferred = None;
+ for namespace in snapshot {
+ let partition = namespace.partition_id();
+ if is_gap_stopped(partition) {
+ // Both ceilings, in the order the sweep applies them: the rate
+ // cap it counts itself, then the live-session count the arm fn
+ // refuses on.
+ if !repairs.is_due(partition)
+ || arms >= PARTITION_REPAIR_ARMS_PER_TICK_MAX
+ || repairs.live_count() >= PARTITION_REPAIRS_INFLIGHT_MAX
+ {
+ continue;
+ }
+ served.arms[partition] += 1;
+ repairs.live[partition] = Some(SESSION_SWEEPS);
+ arms += 1;
+ continue;
+ }
+ if walks < PARTITION_WALKS_PER_TICK_MAX {
+ served.walks[partition] += 1;
+ walks += 1;
+ } else {
+ walk_deferred.get_or_insert(namespace);
+ }
+ }
+ walk_deferred
+ }
+
+ #[test]
+ fn given_more_eligible_groups_than_the_walk_budget_when_swept_should_reach_every_one() {
+ let mut served = Served::new();
+ let mut repairs = Repairs::new();
+ let mut cursor = None;
+ let walk_eligible = (0..GROUPS).filter(|p| !is_gap_stopped(*p)).count();
+ let passes = walk_eligible.div_ceil(PARTITION_WALKS_PER_TICK_MAX);
+ for _ in 0..passes {
+ cursor = sweep(cursor, &mut served, &mut repairs);
+ }
+ let unserved: Vec<_> = (0..GROUPS)
+ .filter(|partition| !is_gap_stopped(*partition) && served.walks[*partition] == 0)
+ .collect();
+ assert!(
+ unserved.is_empty(),
+ "{} of {walk_eligible} walk-eligible groups never had their walk run in \
+ {passes} sweeps (partitions {unserved:?}); the budget is being spent on \
+ the same leading groups every pass",
+ unserved.len()
+ );
+ }
+
+ #[test]
+ fn given_a_sustained_backlog_when_swept_should_keep_every_group_within_one_walk() {
+ // Sustained, because the starvation this guards against only shows over
+ // many passes: one sweep serves the head no matter how the cursor moves.
+ // Walk-eligible groups stay eligible throughout (a walked group is
+ // walk-stalled again on the next produce), so a fair scheduler owes them
+ // walks in round-robin and none may drift a full round behind.
+ let mut served = Served::new();
+ let mut repairs = Repairs::new();
+ let mut cursor = None;
+ for _ in 0..10 * GROUPS {
+ cursor = sweep(cursor, &mut served, &mut repairs);
+ }
+ let walks: Vec<u32> = (0..GROUPS)
+ .filter(|partition| !is_gap_stopped(*partition))
+ .map(|partition| served.walks[partition])
+ .collect();
+ let most = walks.iter().max().copied().unwrap_or_default();
+ let fewest = walks.iter().min().copied().unwrap_or_default();
+ assert!(
+ most - fewest <= 1,
+ "walks are not spread evenly: one group got {most}, another {fewest}"
+ );
+ }
+
+ #[test]
+ fn given_a_capped_arm_backlog_when_swept_should_arm_every_gap_stopped_group() {
+ // The arm caps carry no cursor because arming REMOVES a group from the
+ // eligible set: the front of the queue drains, so the tail is reached
+ // without one. If that ever stops holding, this run wedges.
+ //
+ // Both ceilings are modelled, so the run also pins that the live-session
+ // cap only DELAYS: a group refused for capacity keeps its debounce and
+ // arms once a session retires. Bounded by the slower of the two, plus a
+ // session's life for the last batch to have somewhere to go.
+ let mut served = Served::new();
+ let mut repairs = Repairs::new();
+ let mut cursor = None;
+ let gap_stopped = (0..GROUPS).filter(|p| is_gap_stopped(*p)).count();
+ let by_rate = gap_stopped.div_ceil(PARTITION_REPAIR_ARMS_PER_TICK_MAX);
+ let by_capacity =
+ gap_stopped.div_ceil(PARTITION_REPAIRS_INFLIGHT_MAX) * SESSION_SWEEPS as usize;
+ for _ in 0..by_rate.max(by_capacity) + SESSION_SWEEPS as usize {
+ cursor = sweep(cursor, &mut served, &mut repairs);
+ }
+ let unarmed: Vec<_> = (0..GROUPS)
+ .filter(|partition| is_gap_stopped(*partition) && served.arms[*partition] == 0)
+ .collect();
+ assert!(
+ unarmed.is_empty(),
+ "gap-stopped groups {unarmed:?} never armed; the arm caps are queueing \
+ behind the same prefix and need a cursor after all"
+ );
+ assert!(
+ served.arms.iter().all(|arms| *arms <= 1),
+ "a group armed twice while its first session was still open"
+ );
+ }
+
+ #[test]
+ fn given_the_live_session_ceiling_when_swept_should_never_exceed_it() {
+ // The ceiling exists because each session is a window the SERVING peer
+ // walks on its own pump; the rate cap alone would let a rejoin put every
+ // group's stream in flight within `groups / 3` passes.
+ let mut served = Served::new();
+ let mut repairs = Repairs::new();
+ let mut cursor = None;
+ for _ in 0..10 * GROUPS {
+ cursor = sweep(cursor, &mut served, &mut repairs);
+ let live = repairs.live_count();
+ assert!(
+ live <= PARTITION_REPAIRS_INFLIGHT_MAX,
+ "{live} sessions live at once, past the ceiling of \
+ {PARTITION_REPAIRS_INFLIGHT_MAX}"
+ );
+ }
+ }
+
+ #[test]
+ fn given_a_cursor_naming_a_removed_namespace_when_rotating_should_resume_at_its_successor() {
+ // The group the cap turned away can be deleted before the next sweep;
+ // the resume point is then the next namespace above it, not the front.
+ let mut snapshot = namespaces();
+ let removed = snapshot.remove(20);
+ rotate_sweep_to_cursor(&mut snapshot, Some(removed));
+ assert_eq!(
+ snapshot.first().copied(),
+ Some(IggyNamespace::new(1, 1, 21))
+ );
+ }
+
+ #[test]
+ fn given_a_cursor_past_every_namespace_when_rotating_should_wrap_to_the_front() {
+ let mut snapshot = namespaces();
+ rotate_sweep_to_cursor(&mut snapshot, Some(IggyNamespace::new(1, 1, GROUPS)));
+ assert_eq!(snapshot, namespaces());
+ }
+
+ #[test]
+ fn given_no_deferral_last_pass_when_rotating_should_start_at_the_front() {
+ let mut snapshot = namespaces();
+ rotate_sweep_to_cursor(&mut snapshot, None);
+ assert_eq!(snapshot, namespaces());
+ }
+}
+
+#[cfg(test)]
+mod gap_detector_tests {
+ //! The level-triggered repair arm the partition tick sweep runs.
+ //!
+ //! Its whole reason to exist is that the edge-triggered arming sites are
+ //! starvable, so the guards it shares with them and the debounce that keeps
+ //! it off healthy traffic are the parts worth pinning.
+
+ use super::{
+ GapArm, GapProbe, PARTITION_GAP_DEBOUNCE_TICKS_MIN, drive_partition_gap_debounce,
+ partition_is_gap_stopped, partition_is_walk_stalled,
+ };
+
+ const DEBOUNCE: u32 = 100;
+
+ /// A gap-stopped follower: committed through op 10, walkable only to 5,
+ /// because op 6 is not in its journal.
+ const fn gap_stopped() -> GapProbe {
+ GapProbe {
+ normal: true,
+ transferring: false,
+ recovery_owned: false,
+ commit_min: 5,
+ commit_max: 10,
+ next_op_resident: false,
+ missing_suffix: false,
+ }
+ }
+
+ /// A walk-stalled follower: the same lag, but op 6 IS in its journal, so
+ /// nothing needs fetching and the walk just has to run.
+ fn walk_stalled() -> GapProbe {
+ GapProbe {
+ next_op_resident: true,
+ ..gap_stopped()
+ }
+ }
+
+ /// A follower level with its commit frontier that holds adopted suffix
+ /// headers whose bodies never arrived.
+ fn missing_suffix() -> GapProbe {
+ GapProbe {
+ commit_min: 10,
+ missing_suffix: true,
+ ..gap_stopped()
+ }
+ }
+
+ #[test]
+ fn given_a_lagging_follower_with_the_next_op_resident_when_probed_should_not_be_gap_stopped() {
+ // The half that keeps the predicate honest. A follower advances
+ // commit_max from every prepare header in preflight, so commit_min <
+ // commit_max is transiently true on any pipelined tick; without the
+ // journal-hole test the driver would request repair against ordinary
+ // produce, on every partition, forever.
+ assert!(!partition_is_gap_stopped(&walk_stalled()));
+ assert!(partition_is_gap_stopped(&gap_stopped()));
+ }
+
+ #[test]
+ fn given_a_caught_up_follower_when_probed_should_not_be_gap_stopped() {
+ let caught_up = GapProbe {
+ commit_min: 10,
+ ..gap_stopped()
+ };
+ assert!(!partition_is_gap_stopped(&caught_up));
+ }
+
+ #[test]
+ fn given_adopted_suffix_headers_without_bodies_when_probed_should_be_gap_stopped() {
+ // The shape a commit-frontier test alone misses: both marks sit below
+ // 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!(
+ !partition_is_gap_stopped(&GapProbe {
+ missing_suffix: false,
+ ..missing_suffix()
+ }),
+ "a caught-up replica with a complete suffix has nothing to repair"
+ );
+ }
+
+ #[test]
+ fn given_a_replica_outside_normal_status_when_probed_should_not_be_gap_stopped() {
+ // A view change owns the log while it runs, and `maybe_request_partition_repair`
+ // refuses outside Normal anyway; arming here would only burn a nonce.
+ for probe in [gap_stopped(), missing_suffix()] {
+ assert!(!partition_is_gap_stopped(&GapProbe {
+ normal: false,
+ ..probe
+ }));
+ assert!(!partition_is_gap_stopped(&GapProbe {
+ transferring: true,
+ ..probe
+ }));
+ }
+ }
+
+ #[test]
+ fn given_recovery_already_owned_when_probed_should_not_be_gap_stopped() {
+ // 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 {
+ recovery_owned: true,
+ ..probe
+ }));
+ }
+ }
+
+ #[test]
+ fn given_a_gap_stopped_follower_when_debouncing_should_arm_only_at_the_threshold() {
+ let probe = gap_stopped();
+ let mut gap_ticks = 0;
+ for tick in 1..DEBOUNCE {
+ assert_eq!(
+ drive_partition_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),
+ GapArm::Arm
+ );
+ }
+
+ #[test]
+ fn given_a_debounce_in_progress_when_the_gap_closes_should_reset_the_counter() {
+ let stopped = gap_stopped();
+ let walkable = GapProbe {
+ next_op_resident: true,
+ ..stopped
+ };
+ let mut gap_ticks = 0;
+ for _ in 0..DEBOUNCE - 1 {
+ drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, true);
+ }
+ assert_eq!(gap_ticks, DEBOUNCE - 1);
+
+ assert_eq!(
+ drive_partition_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),
+ GapArm::NotDue,
+ "a fresh gap must serve its own debounce, not inherit the old count"
+ );
+ }
+
+ #[test]
+ fn given_a_follower_with_resident_committed_ops_when_probed_should_be_walk_stalled() {
+ assert!(partition_is_walk_stalled(&walk_stalled()));
+ assert!(
+ !partition_is_walk_stalled(&gap_stopped()),
+ "a missing next op is repair's job; a walk over it would stop dead"
+ );
+ }
+
+ #[test]
+ fn given_a_caught_up_follower_when_probed_should_not_be_walk_stalled() {
+ let caught_up = GapProbe {
+ commit_min: 10,
+ ..walk_stalled()
+ };
+ assert!(!partition_is_walk_stalled(&caught_up));
+ }
+
+ #[test]
+ fn given_a_replica_outside_normal_status_when_probed_should_not_be_walk_stalled() {
+ let electing = GapProbe {
+ normal: false,
+ ..walk_stalled()
+ };
+ assert!(!partition_is_walk_stalled(&electing));
+
+ // Same gate as the on-commit arm: a walk during a transfer can advance
+ // commit_min past the incoming frontier.
+ let installing = GapProbe {
+ transferring: true,
+ ..walk_stalled()
+ };
+ assert!(!partition_is_walk_stalled(&installing));
+ }
+
+ #[test]
+ fn given_recovery_already_owned_when_the_next_op_is_resident_should_still_be_walk_stalled() {
+ // Deliberate: `apply_repaired_prepare` journals without walking, so a
+ // gated walk would sit parked for the whole session while the resident
+ // prefix is already applicable.
+ assert!(partition_is_walk_stalled(&GapProbe {
+ recovery_owned: true,
+ ..walk_stalled()
+ }));
+ }
+
+ #[test]
+ fn given_any_probe_when_evaluated_should_never_be_both_gap_stopped_and_walk_stalled() {
+ // If they ever overlap, one tick both arms repair and walks the window
+ // it is fetching, and the arm refetches a resident committed prefix.
+ for normal in [false, true] {
+ for transferring in [false, true] {
+ for recovery_owned in [false, true] {
+ for (commit_min, commit_max) in [(5, 10), (10, 10)] {
+ for next_op_resident in [false, true] {
+ for missing_suffix in [false, true] {
+ let probe = GapProbe {
+ normal,
+ transferring,
+ recovery_owned,
+ commit_min,
+ commit_max,
+ next_op_resident,
+ missing_suffix,
+ };
+ assert!(
+ !(partition_is_gap_stopped(&probe)
+ && partition_is_walk_stalled(&probe)),
+ "both predicates claim {probe:?}"
+ );
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn given_a_missing_suffix_over_a_walkable_lag_when_evaluated_should_prefer_the_walk() {
+ // Arming here would request `(commit_min, head]`: the committed prefix
+ // this replica already holds resident, refetched, plus the suffix. The
+ // walk closes the lag first -- `commit_max` cannot move while the suffix
+ // is short of quorum -- and the suffix arms on the pass after.
+ let probe = GapProbe {
+ missing_suffix: true,
+ ..walk_stalled()
+ };
+ assert!(partition_is_walk_stalled(&probe));
+ assert!(
+ !partition_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 {
+ commit_min: probe.commit_max,
+ ..probe
+ }),
+ "and the same replica arms once the lag is gone"
+ );
+ }
+
+ #[test]
+ fn given_the_arm_budget_spent_when_debouncing_should_defer_without_losing_the_debounce() {
+ let probe = gap_stopped();
+ let mut gap_ticks = DEBOUNCE;
+ assert_eq!(
+ drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, false),
+ GapArm::Deferred,
+ "a spent budget must refuse the arm"
+ );
+ assert!(
+ gap_ticks > DEBOUNCE,
+ "a refused group stays due; restarting its debounce would push the \
+ arm a whole interval out per contended tick"
+ );
+ assert_eq!(
+ drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, true),
+ GapArm::Arm,
+ "the same group arms on the next pass with a slot free"
+ );
+ }
+
+ #[test]
+ fn given_a_debounce_shorter_than_the_floor_when_driven_should_hold_until_the_floor() {
+ // `repair_retry_interval` is an operator knob whose primary meaning is
+ // the stalled-stream retry, and `duration_to_ticks` floors it at one
+ // tick. One tick of lag is ordinary pipelining, so without a floor of
+ // its own a shortened retry interval would arm repair against a single
+ // reordered prepare.
+ let probe = gap_stopped();
+ let mut gap_ticks = 0;
+ for tick in 1..PARTITION_GAP_DEBOUNCE_TICKS_MIN {
+ assert_eq!(
+ drive_partition_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),
+ GapArm::Arm
+ );
+ }
+}
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index fcf885a09..3e4b1c14f 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -80,11 +80,14 @@ pub struct FrameDropLabel {
/// (`reason=park_overflow`), a parked frame retired with no client to answer
/// (`reason=park_dropped`), an incarnation rejection, or a routing send the
/// target inbox refused. A shed client request is answered with a retriable
-/// status. A shed prepare may no longer be covered by retransmit once its op
-/// reached quorum, but a later `CommitMessage` that advances the backup's
-/// frontier arms same-view journal repair. If the primary evicted the range,
-/// repair escalates to partition state transfer. The counter therefore signals
-/// a data-plane gap or recovery burden, not a requirement for a view change.
+/// status, so the client recovers. A shed *prepare* has nobody to answer and is
+/// not covered by retransmit once its op has reached quorum
+/// (`consensus::retransmit_targets` skips `ok_quorum_received`), so the backup
+/// gap-stops; `tick_partitions`' sweep is what repairs it, escalating to
+/// partition state transfer when the primary has evicted the range, and
+/// `partition_prepare_gap_drops_total` is what counts the prepares that reached
+/// the gap check. The counter therefore signals a data-plane gap or recovery
+/// burden, not a requirement for a view change.
pub mod frame_drop_variant {
pub const CONSENSUS: &str = "consensus";
pub const FD_TRANSFER: &str = "fd_transfer";
@@ -208,6 +211,7 @@ pub struct ShardMetrics {
partition_frames_rejected_ahead_total: Counter,
partition_requests_denied_transient_total: Counter,
partition_repair_serves_deferred_purge_total: Counter,
+ partition_prepare_gap_drops_total: Counter,
metadata_read_frontier_refusals_total: Counter,
client_requests_denied_queue_full_total: Counter,
}
@@ -234,6 +238,7 @@ impl ShardMetrics {
partition_frames_rejected_ahead_total: Counter::default(),
partition_requests_denied_transient_total: Counter::default(),
partition_repair_serves_deferred_purge_total: Counter::default(),
+ partition_prepare_gap_drops_total: Counter::default(),
metadata_read_frontier_refusals_total: Counter::default(),
client_requests_denied_queue_full_total: Counter::default(),
}
@@ -447,6 +452,42 @@ impl ShardMetrics {
self.partition_repair_serves_deferred_purge_total.get()
}
+ /// Add the prepares a partition's backup gap check destroyed since the last
+ /// sweep. Drained per tick from `IggyPartition::take_prepare_gap_drops`,
+ /// and once more when `ConfirmRemove` drops the partition: a tombstoned
+ /// namespace is invisible to the sweep, so the tail it left would otherwise
+ /// go to the floor with the value.
+ ///
+ /// Counts the prepares that ARRIVED after a hole, not the holes: a gap
+ /// opened by the last prepare of a burst leaves this at zero. A nonzero
+ /// value proves the repair driver has work; a zero one proves nothing.
+ ///
+ /// Deliberately NOT a `frame_drops_total{variant=partition}` reason: that
+ /// family means the bus or the router shed a frame, and the simulator
+ /// asserts it stays at zero on runs with no injected loss. A gap drop is a
+ /// protocol-ordering drop that any real loss produces, and the sweep repairs
+ /// it, so folding the two would turn a routing-fault alert into noise.
+ ///
+ /// Shard-scoped, with no namespace label: a server runs hundreds of groups
+ /// per shard, so labelling by namespace is unbounded cardinality, and every
+ /// other partition counter in this file is shard-scoped for the same
+ /// reason. The per-group detail is in the arm's log line.
+ ///
+ /// The metadata plane's own gap drop is NOT counted here, 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.
+ pub fn record_partition_prepare_gap_drops(&self, drops: u64) {
+ self.partition_prepare_gap_drops_total.inc_by(drops);
+ }
+
+ /// Snapshot of `partition_prepare_gap_drops_total`. Test/simulator accessor.
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn partition_prepare_gap_drops_value(&self) -> u64 {
+ self.partition_prepare_gap_drops_total.get()
+ }
+
/// Snapshot of `partition_frames_rejected_stale_total`. Test/simulator
/// accessor, readable from any crate under those cfgs so the crates that
/// drive the reconciler can assert a reject did not happen.
@@ -538,6 +579,11 @@ impl ShardMetrics {
"partition repair serves or completions deferred until a committed purge applies",
self.partition_repair_serves_deferred_purge_total.clone(),
);
+ registry.register(
+ "partition_prepare_gap_drops",
+ "replicated prepares dropped out of order by a backup's gap check",
+ self.partition_prepare_gap_drops_total.clone(),
+ );
registry.register(
"metadata_read_frontier_refusals",
"metadata reads refused because this node never applied the caller's committed op",
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 1f4fa0117..4a3a0d214 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -379,6 +379,8 @@ where
self.process_frame(frame).await;
self.process_loopback(&mut loopback_buf, &mut namespace_scratch).await;
// Tail drain catches reconcile ops whose marker was dropped.
+ // Anything it stages is served by the arm above
+ // on the next pass, before this arm can run again.
self.apply_reconcile_ops();
}
// Guaranteed reply-lane service: `select_biased!`
diff --git a/core/simulator/Cargo.toml b/core/simulator/Cargo.toml
index 0e46f73fc..0ac51b6af 100644
--- a/core/simulator/Cargo.toml
+++ b/core/simulator/Cargo.toml
@@ -51,6 +51,10 @@ tracing = { workspace = true }
tracing-subscriber = { workspace = true }
[dev-dependencies]
+# Test-only feature on a crate this package already depends on: the fault
+# hooks the partition-driver tests arm must not compile into anything that
+# ships. See `partitions/Cargo.toml`.
+partitions = { workspace = true, features = ["fault-injection"] }
tempfile = { workspace = true }
[lints.clippy]
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 767f72196..8a9d7c631 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -5106,6 +5106,864 @@ mod repair_frontier_tests {
}
}
+#[cfg(test)]
+mod partition_repair_driver_tests {
+ //! A backup that missed a committed partition prepare recovers in Normal
+ //! status, without waiting for a view change.
+ //!
+ //! Every edge-triggered arming site is starvable. A follower advances
+ //! `commit_max` from each prepare header in `replicate_preflight`, before
+ //! the gap check drops the prepare, so under produce the primary's commit
+ //! heartbeat lands as `CommitOutcome::Accepted` and the backstop that would
+ //! arm repair on `Advanced` never runs. These tests starve that edge
+ //! outright (no commit heartbeat for the group reaches the lagging replica)
+ //! so nothing but the level-triggered detector in `tick_partitions` can
+ //! close the gap.
+
+ use super::*;
+ use bytes::Bytes;
+ use consensus::Status;
+ use iggy_binary_protocol::{
+ CommitHeader, PrepareHeader, RepairRangeReplyHeader, RequestPreparesHeader,
+ };
+ use packet::Packet;
+ use std::sync::atomic::{AtomicU64, Ordering};
+
+ /// Chain replication runs 0 -> 1 -> 2 and stops before the primary, so
+ /// replica 2 forwards to nobody: it is the only replica whose losses do not
+ /// also starve its successor, and therefore the group, of quorum.
+ const LAGGING: u8 = 2;
+
+ const CLIENT_ID: u128 = 1;
+
+ /// Ops that replicate cleanly before the fault, so the gap opens above a
+ /// committed prefix rather than at the group's first op.
+ const WARMUP_SENDS: usize = 3;
+
+ /// Ticks stepped after each produce. Keeps one send's round trip inside its
+ /// own window so the workload is legible tick by tick.
+ const STEPS_PER_SEND: usize = 12;
+
+ /// Produces issued with the fault standing. `partitions::REPAIR_RETRY_TICKS`
+ /// is 100, so this must carry the run past the debounce while prepares keep
+ /// consuming the `commit_max` advance the heartbeat backstop needs.
+ const GAP_SENDS: usize = 12;
+
+ /// Quiet ticks after the produce stops, for the repair stream to land.
+ ///
+ /// Budgeted against `NORMAL_HEARTBEAT_TICKS` (500): with this group's commit
+ /// heartbeats withheld, the lagging replica elects once that timer fires, and
+ /// an election would heal the gap through `on_start_view` instead. Every
+ /// phase after the fault is installed has to fit inside it.
+ const QUIET_STEPS: usize = 160;
+
+ /// Budget for the group to settle once the fault is lifted. Generous rather
+ /// than tuned: the drain loop breaks on convergence, and the tail above the
+ /// repaired window waits on whichever interval-driven site picks it up.
+ const DRAIN_STEPS: usize = 600;
+
+ /// Ticks of healthy load in the no-false-positive test, several debounce
+ /// intervals' worth so the sweep gets many chances to arm.
+ const LOAD_TICKS: usize = 4 * partitions::REPAIR_RETRY_TICKS as usize;
+
+ /// Paced at a round trip rather than one submit per tick: the pipeline caps
+ /// at `PIPELINE_PREPARE_QUEUE_MAX`, so submitting faster than the group
+ /// commits just collects transient rejections and the group goes quiet.
+ const TICKS_PER_SEND: usize = 4;
+
+ /// Ops the healthy run must have committed for its verdict to mean anything.
+ /// A produce's round trip is four one-way hops plus tick granularity, so this
+ /// network commits on the order of one op per fifteen ticks however hard the
+ /// client pushes; the floor only has to prove the group was live across
+ /// several debounce intervals, not that it was saturated.
+ const COMMITTED_MIN: u64 = 20;
+
+ /// Produces issued with the fault standing in the walk-starvation test: few
+ /// enough that the gap debounce fires only after produce stops, so the
+ /// repair window closes at the last carried commit and lands fully resident.
+ const STRAND_SENDS: usize = 3;
+
+ /// Quiet budget for the walk-starvation test: debounce, repair stream, then
+ /// the drain, kept under `NORMAL_HEARTBEAT_TICKS` so an election cannot be
+ /// the healer.
+ const STRAND_QUIET_STEPS: usize = 300;
+
+ /// Defines this test's `withhold_one_prepare` chain hook over the statics it
+ /// names: swallow the FIRST partition prepare for `$namespace`, once, and
+ /// record its op in `$withheld_op`.
+ ///
+ /// A macro because link hooks are bare `fn` pointers, so the body cannot
+ /// capture, and the statics must stay per-test: the sibling tests in this
+ /// binary run in parallel and would otherwise share one fault. The statics
+ /// are still declared in each test, where its fault setup is read.
+ macro_rules! withhold_one_prepare {
+ ($namespace:ident, $withheld_op:ident) => {
+ fn withhold_one_prepare(packet: &Packet) -> bool {
+ let Some(header) = prepare_for(packet, $namespace.load(Ordering::Relaxed)) else {
+ return false;
+ };
+ $withheld_op
+ .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed)
+ .is_ok()
+ }
+ };
+ }
+
+ /// The partition-plane prepare a packet carries, if it carries one for
+ /// `group`.
+ fn prepare_for(packet: &Packet, group: u64) -> Option<PrepareHeader> {
+ if packet.message.header().command != Command::Prepare {
+ return None;
+ }
+ let header: &PrepareHeader =
+ bytemuck::checked::from_bytes(&packet.message.as_slice()[..size_of::<PrepareHeader>()]);
+ (header.group == group).then_some(*header)
+ }
+
+ /// Whether a packet is a commit heartbeat for `group`.
+ fn is_commit_for(packet: &Packet, group: u64) -> bool {
+ if packet.message.header().command != Command::Commit {
+ return false;
+ }
+ let header: &CommitHeader =
+ bytemuck::checked::from_bytes(&packet.message.as_slice()[..size_of::<CommitHeader>()]);
+ header.group == group
+ }
+
+ /// Whether a packet is a repair request for `group`.
+ fn is_request_prepares_for(packet: &Packet, group: u64) -> bool {
+ if packet.message.header().command != Command::RequestPrepares {
+ return false;
+ }
+ let header: &RequestPreparesHeader = bytemuck::checked::from_bytes(
+ &packet.message.as_slice()[..size_of::<RequestPreparesHeader>()],
+ );
+ header.group == group
+ }
+
+ /// Whether a packet is a repair stream terminator for `group`.
+ fn is_repair_done_for(packet: &Packet, group: u64) -> bool {
+ if packet.message.header().command != Command::RepairDone {
+ return false;
+ }
+ let header: &RepairRangeReplyHeader = bytemuck::checked::from_bytes(
+ &packet.message.as_slice()[..size_of::<RepairRangeReplyHeader>()],
+ );
+ header.group == group
+ }
+
+ fn cluster(seed: u64) -> (Simulator, SimClient) {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let replica_count: u8 = 3;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let sim = Simulator::new(
+ replica_count as usize,
+ std::iter::once(CLIENT_ID),
+ network_opts,
+ );
+ (sim, SimClient::new(CLIENT_ID))
+ }
+
+ /// Submit `sends` produces against `namespace`, stepping between each.
+ fn produce(
+ sim: &mut Simulator,
+ client: &SimClient,
+ namespace: IggyNamespace,
+ sends: usize,
+ tag: &str,
+ ) {
+ for index in 0..sends {
+ let msg = client.send_messages(namespace, &[Bytes::from(format!("{tag}-{index}"))]);
+ sim.submit_request(client.client_id(), 0, msg.into_generic());
+ for _ in 0..STEPS_PER_SEND {
+ sim.step();
+ }
+ }
+ }
+
+ /// `(status, view, commit_min, commit_max)` of one replica's partition group.
+ fn group_state(
+ sim: &Simulator,
+ replica: u8,
+ namespace: IggyNamespace,
+ ) -> (Status, u32, u64, u64) {
+ let shard = sim.replicas[replica as usize].partition_shard(namespace);
+ let partition = shard
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .expect("the replica hosts the group");
+ let consensus = partition.consensus();
+ (
+ consensus.status(),
+ consensus.view(),
+ consensus.commit_min(),
+ consensus.commit_max(),
+ )
+ }
+
+ fn journal_holds(sim: &Simulator, replica: u8, namespace: IggyNamespace, op: u64) -> bool {
+ let shard = sim.replicas[replica as usize].partition_shard(namespace);
+ shard
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .is_some_and(|partition| partition.log.journal().inner.holds_op(op))
+ }
+
+ fn gap_drops(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> u64 {
+ sim.replicas[replica as usize]
+ .partition_shard(namespace)
+ .metrics()
+ .partition_prepare_gap_drops_value()
+ }
+
+ /// Gap drops still buffered on the partition, i.e. recorded since the sweep
+ /// last drained them into the shard counter.
+ fn buffered_gap_drops(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> u64 {
+ sim.replicas[replica as usize]
+ .partition_shard(namespace)
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .map_or(0, partitions::IggyPartition::prepare_gap_drops)
+ }
+
+ fn transfer_armed(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> bool {
+ let shard = sim.replicas[replica as usize].partition_shard(namespace);
+ shard
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .is_some_and(|partition| {
+ partition.transfer.is_some()
+ || partition.consensus().state_transfer_stage()
+ != consensus::StateTransferStage::Idle
+ })
+ }
+
+ #[test]
+ fn given_a_backup_that_dropped_a_committed_prepare_when_heartbeat_advances_are_starved_should_repair_in_normal_status()
+ {
+ // Statics, not captures: the link hooks are bare `fn` pointers. Declared
+ // inside the test because the sibling tests in this binary run in
+ // parallel and would otherwise share them.
+ static GAP_NS: AtomicU64 = AtomicU64::new(0);
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+
+ withhold_one_prepare!(GAP_NS, WITHHELD_OP);
+
+ /// Primary -> 2: withhold this group's commit heartbeats, so the
+ /// `Advanced` backstop can never run, and withhold retransmits of the
+ /// dropped op. The retransmit half stands in for production behaviour
+ /// rather than adding a fault: `consensus::retransmit_targets` skips an
+ /// op that already reached quorum, and this op reaches quorum on 0 and 1
+ /// alone. Without it the retry timer could heal the gap and the test
+ /// would pass with no repair driver at all.
+ fn starve_commit_edge(packet: &Packet) -> bool {
+ let group = GAP_NS.load(Ordering::Relaxed);
+ if let Some(header) = prepare_for(packet, group) {
+ return header.op == WITHHELD_OP.load(Ordering::Relaxed);
+ }
+ is_commit_for(packet, group)
+ }
+
+ let (mut sim, client) = cluster(0x5EED_0232);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ GAP_NS.store(namespace.inner(), Ordering::Relaxed);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+
+ produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup");
+ let (_, _, warm_commit_min, _) = group_state(&sim, LAGGING, namespace);
+ assert!(
+ warm_commit_min > 0,
+ "the lagging replica committed nothing before the fault, so the gap \
+ below would open at the group's first op"
+ );
+
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) =
+ Some(withhold_one_prepare);
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) =
+ Some(starve_commit_edge);
+
+ produce(&mut sim, &client, namespace, GAP_SENDS, "gap");
+
+ let withheld = WITHHELD_OP.load(Ordering::Relaxed);
+ assert_ne!(
+ withheld, 0,
+ "no partition prepare crossed the chain link, so the fault never armed"
+ );
+ assert!(
+ gap_drops(&sim, LAGGING, namespace) > 0,
+ "the lagging replica never reached its backup gap check, so the \
+ prepares after the withheld op were not dropped as a gap"
+ );
+
+ for _ in 0..QUIET_STEPS {
+ sim.step();
+ }
+
+ // Judged with the blockade still standing, so the tick driver is the only
+ // thing that can have armed the repair: no commit heartbeat for this
+ // group has reached the replica since the gap opened, and `on_commit` is
+ // where the `Advanced` backstop lives.
+ let (status, view, commit_min, _) = group_state(&sim, LAGGING, namespace);
+ assert_eq!(
+ view, 0,
+ "a view change healed the gap instead of the repair driver; the test \
+ proves nothing about normal status"
+ );
+ assert_eq!(status, Status::Normal, "the replica left Normal status");
+ assert!(
+ journal_holds(&sim, LAGGING, namespace, withheld),
+ "op {withheld} was never repaired back into the lagging replica's journal"
+ );
+ assert!(
+ commit_min >= withheld,
+ "the commit walk never crossed the repaired hole: stopped at \
+ {commit_min}, the withheld op is {withheld}"
+ );
+
+ // Lift the blockade and let the group settle. The commit walk is driven
+ // by arriving frames, so with this group's heartbeats withheld the tail
+ // above the repaired window has nothing to advance it; that is the
+ // injected fault, not the gap under test.
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = None;
+ for _ in 0..DRAIN_STEPS {
+ sim.step();
+ let (_, _, commit_min, commit_max) = group_state(&sim, LAGGING, namespace);
+ if commit_min == commit_max {
+ break;
+ }
+ }
+ let (status, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace);
+ assert_eq!((status, view), (Status::Normal, 0));
+ assert_eq!(
+ commit_min, commit_max,
+ "the lagging replica is still gap-stopped: committed through \
+ {commit_max} but walkable only to {commit_min}"
+ );
+ }
+
+ #[test]
+ fn given_a_repair_armed_by_the_tick_driver_when_the_range_is_evicted_should_convert_to_state_transfer()
+ {
+ static GAP_NS: AtomicU64 = AtomicU64::new(0);
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+
+ withhold_one_prepare!(GAP_NS, WITHHELD_OP);
+
+ fn starve_commit_edge(packet: &Packet) -> bool {
+ let group = GAP_NS.load(Ordering::Relaxed);
+ if let Some(header) = prepare_for(packet, group) {
+ return header.op == WITHHELD_OP.load(Ordering::Relaxed);
+ }
+ is_commit_for(packet, group)
+ }
+
+ let (mut sim, client) = cluster(0x5EED_0233);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ GAP_NS.store(namespace.inner(), Ordering::Relaxed);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+
+ produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup");
+
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) =
+ Some(withhold_one_prepare);
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) =
+ Some(starve_commit_edge);
+
+ produce(&mut sim, &client, namespace, GAP_SENDS, "gap");
+ assert_ne!(
+ WITHHELD_OP.load(Ordering::Relaxed),
+ 0,
+ "no partition prepare crossed the chain link, so the fault never armed"
+ );
+
+ // Compact the serving side past the gap. This plane's journal is
+ // memory-only, so wiping it IS its retention floor moving: the serve
+ // path reads `repair_retained_from` as `None` and reports eviction from
+ // its own commit frontier, which is exactly what a peer that
+ // checkpointed past the requested window answers.
+ {
+ let primary = sim.replicas[0].partition_shard(namespace);
+ let partition = primary
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .expect("the primary hosts the group");
+ partition.log.journal().inner.clear_all();
+ }
+
+ for _ in 0..QUIET_STEPS {
+ sim.step();
+ if transfer_armed(&sim, LAGGING, namespace) {
+ break;
+ }
+ }
+
+ let (status, view, ..) = group_state(&sim, LAGGING, namespace);
+ assert_eq!(
+ view, 0,
+ "a view change armed the recovery instead of the tick-armed repair session"
+ );
+ assert!(
+ transfer_armed(&sim, LAGGING, namespace),
+ "the tick-armed repair session hit an evicted range but never converted \
+ to a state transfer (status {status:?})"
+ );
+ }
+
+ #[test]
+ fn given_healthy_pipelined_traffic_when_no_gap_exists_should_not_arm_repair() {
+ static GAP_NS: AtomicU64 = AtomicU64::new(0);
+ static REPAIR_REQUESTS: AtomicU64 = AtomicU64::new(0);
+
+ /// Observer, not a fault: counts this group's repair requests and passes
+ /// every packet through.
+ fn count_repair_requests(packet: &Packet) -> bool {
+ if is_request_prepares_for(packet, GAP_NS.load(Ordering::Relaxed)) {
+ REPAIR_REQUESTS.fetch_add(1, Ordering::Relaxed);
+ }
+ false
+ }
+
+ // TWO replicas, so quorum spans both: no op can commit without the
+ // backup's ack, every reordering-induced gap therefore blocks quorum, and
+ // `consensus::retransmit_targets` refills it. At three, the network's
+ // per-tick delivery shuffle lets an op commit on the primary and its
+ // first chain hop while the last replica loses it for good, which is the
+ // very fault the sibling tests inject -- it would then be repaired here,
+ // correctly, and this assertion would fire on a healthy driver.
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let replica_count: u8 = 2;
+ let mut sim = Simulator::new(
+ replica_count as usize,
+ std::iter::once(CLIENT_ID),
+ packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0x5EED_0234,
+ ..packet::PacketSimulatorOptions::default()
+ },
+ );
+ let client = SimClient::new(CLIENT_ID);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ GAP_NS.store(namespace.inner(), Ordering::Relaxed);
+ REPAIR_REQUESTS.store(0, Ordering::Relaxed);
+
+ for (from, to) in [(0u8, 1u8), (1, 0)] {
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(from), ProcessId::Replica(to)) =
+ Some(count_repair_requests);
+ }
+
+ // Sustained, not bursty: a produce every tick, for several debounce
+ // intervals, so the sweep gets many chances to arm against ordinary
+ // pipelining.
+ for tick in 0..LOAD_TICKS {
+ if tick % TICKS_PER_SEND == 0 {
+ let msg =
+ client.send_messages(namespace, &[Bytes::from(format!("healthy-{tick}"))]);
+ sim.submit_request(client.client_id(), 0, msg.into_generic());
+ }
+ sim.step();
+ }
+ for _ in 0..QUIET_STEPS {
+ sim.step();
+ }
+
+ let committed = group_state(&sim, 1, namespace).2;
+ let sends = LOAD_TICKS / TICKS_PER_SEND;
+ assert!(
+ committed >= COMMITTED_MIN,
+ "the backup committed only {committed} ops across {sends} sends, so the \
+ sweep was never driven over a loaded group"
+ );
+ // No per-tick lag sampling: `sim.step()` runs the pumps to quiescence, so
+ // every sample lands AFTER the tick whose walk backstop drained whatever
+ // the step's produce left, and a run that asserted something about those
+ // samples would be asserting over an empty set. The commit lag a naive
+ // detector misreads lives inside a step, and it is pinned where it can
+ // be held still: exhaustively in `gap_detector_tests`, and end to end by
+ // the walk-starvation run in this module, which strands a backup in
+ // exactly that state and proves nothing arms repair over it.
+ //
+ // What this run adds is the part only a live group can show: a loaded,
+ // lossless cluster produces no repair traffic at all.
+ for replica in 0..replica_count {
+ let (status, view, commit_min, commit_max) = group_state(&sim, replica, namespace);
+ assert_eq!(
+ (status, view),
+ (Status::Normal, 0),
+ "replica {replica} left view 0 / Normal, so a view change could \
+ account for repair traffic"
+ );
+ // Residual lag is transient at worst: a heartbeat carrying a commit
+ // the replica already knows is `CommitOutcome::Accepted`, so a backup
+ // can sit with committed-but-unwalked ops until the tick sweep's
+ // walk-stalled backstop resumes the walk. Every one of them is
+ // RESIDENT, which is what separates it from a gap.
+ if commit_min < commit_max {
+ assert!(
+ journal_holds(&sim, replica, namespace, commit_min + 1),
+ "replica {replica} lags at {commit_min} of {commit_max} with op \
+ {} missing, so a no-loss run produced a real hole",
+ commit_min + 1
+ );
+ }
+ }
+ assert_eq!(
+ REPAIR_REQUESTS.load(Ordering::Relaxed),
+ 0,
+ "the tick driver requested repair on a healthy group across {sends} sends; \
+ its gap predicate is reading ordinary commit lag as a journal hole"
+ );
+ }
+
+ #[test]
+ fn given_a_backup_holding_resident_committed_ops_when_every_walk_edge_is_starved_should_drain_in_normal_status()
+ {
+ static GAP_NS: AtomicU64 = AtomicU64::new(0);
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+ static WITHHELD_DONES: AtomicU64 = AtomicU64::new(0);
+
+ withhold_one_prepare!(GAP_NS, 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 `complete_repair`, the walk
+ /// the terminator would run, never fires.
+ fn starve_walk_edges(packet: &Packet) -> bool {
+ let group = GAP_NS.load(Ordering::Relaxed);
+ if prepare_for(packet, group).is_some() {
+ return true;
+ }
+ if is_repair_done_for(packet, group) {
+ WITHHELD_DONES.fetch_add(1, Ordering::Relaxed);
+ return true;
+ }
+ is_commit_for(packet, group)
+ }
+
+ let (mut sim, client) = cluster(0x5EED_0235);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ GAP_NS.store(namespace.inner(), Ordering::Relaxed);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+ WITHHELD_DONES.store(0, Ordering::Relaxed);
+
+ produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup");
+ let (_, _, warm_commit_min, _) = group_state(&sim, LAGGING, namespace);
+ assert!(
+ warm_commit_min > 0,
+ "the lagging replica committed nothing before the fault, so the gap \
+ below would open at the group's first op"
+ );
+
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) =
+ Some(withhold_one_prepare);
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) =
+ Some(starve_walk_edges);
+
+ produce(&mut sim, &client, namespace, STRAND_SENDS, "strand");
+
+ let withheld = WITHHELD_OP.load(Ordering::Relaxed);
+ assert_ne!(
+ withheld, 0,
+ "no partition prepare crossed the chain link, so the fault never armed"
+ );
+
+ for _ in 0..STRAND_QUIET_STEPS {
+ sim.step();
+ let (_, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace);
+ if view != 0 || (commit_min >= withheld && commit_min == commit_max) {
+ break;
+ }
+ }
+
+ assert!(
+ WITHHELD_DONES.load(Ordering::Relaxed) > 0,
+ "no repair terminator was withheld, so the walk was never starved and \
+ a green run would not prove the tick backstop"
+ );
+ assert!(
+ journal_holds(&sim, LAGGING, namespace, withheld),
+ "op {withheld} was never repaired back into the lagging replica's journal"
+ );
+ let (status, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace);
+ assert_eq!(
+ view, 0,
+ "a view change drained the walk instead of the tick backstop; the test \
+ proves nothing about normal status"
+ );
+ assert_eq!(status, Status::Normal, "the replica left Normal status");
+ for op in commit_min + 1..=commit_max {
+ assert!(
+ journal_holds(&sim, LAGGING, namespace, op),
+ "op {op} is not resident, so this run stranded on a repair gap, \
+ not a parked walk"
+ );
+ }
+ assert_eq!(
+ commit_min, commit_max,
+ "the walk never resumed over resident committed ops: walkable to \
+ {commit_min}, committed through {commit_max}, every op between resident"
+ );
+ }
+
+ #[test]
+ fn given_a_gap_stopped_backup_when_repair_is_armed_should_spend_its_debounce() {
+ // The reset lives in `maybe_request_partition_repair`, the funnel every
+ // arming site goes through, precisely because the four edge-triggered
+ // sites never touch `gap_ticks` themselves. Driven here rather than
+ // modelled: a saturated count hands the NEXT gap an arm on its first
+ // tick, and the shape that reaches it is a repair short enough that no
+ // sweep ever observes the partition with its recovery owned.
+ static GAP_NS: AtomicU64 = AtomicU64::new(0);
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+
+ withhold_one_prepare!(GAP_NS, WITHHELD_OP);
+
+ /// Primary -> 2: withhold this group's commit heartbeats, so the
+ /// edge-triggered backstop cannot be what arms the repair below.
+ fn starve_commit_edge(packet: &Packet) -> bool {
+ let group = GAP_NS.load(Ordering::Relaxed);
+ if let Some(header) = prepare_for(packet, group) {
+ return header.op == WITHHELD_OP.load(Ordering::Relaxed);
+ }
+ is_commit_for(packet, group)
+ }
+
+ let (mut sim, client) = cluster(0x5EED_0238);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ GAP_NS.store(namespace.inner(), Ordering::Relaxed);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+
+ produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup");
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) =
+ Some(withhold_one_prepare);
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) =
+ Some(starve_commit_edge);
+ // Short, deliberately: long enough to open the gap, well under the
+ // debounce, so the driver has not armed on its own and the sweep below
+ // is the one that does it.
+ produce(&mut sim, &client, namespace, STRAND_SENDS, "gap");
+ assert_ne!(
+ WITHHELD_OP.load(Ordering::Relaxed),
+ 0,
+ "no partition prepare crossed the chain link, so the fault never armed"
+ );
+ let (_, _, commit_min, commit_max) = group_state(&sim, LAGGING, namespace);
+ assert!(
+ commit_min < commit_max && !journal_holds(&sim, LAGGING, namespace, commit_min + 1),
+ "the lagging replica is not gap-stopped (walkable to {commit_min} of \
+ {commit_max}), so the sweep has nothing to arm"
+ );
+
+ // Saturate the debounce by hand and take one sweep: the arm has to be
+ // what spends it, whatever the tick counted on the way in.
+ let shard = sim.replicas[LAGGING as usize].partition_shard(namespace);
+ {
+ let partitions = shard.plane.partitions();
+ let partition = partitions
+ .get_by_ns(&namespace)
+ .expect("the replica hosts the group");
+ assert!(
+ partition.repair.is_none(),
+ "a session was already open, so this sweep would arm nothing"
+ );
+ partition.gap_ticks.set(u32::MAX);
+ }
+ let mut namespace_scratch = Vec::new();
+ futures::executor::block_on(shard.tick_partitions(&mut namespace_scratch));
+
+ let partitions = shard.plane.partitions();
+ let partition = partitions
+ .get_by_ns(&namespace)
+ .expect("the replica hosts the group");
+ assert!(
+ partition.repair.is_some(),
+ "the sweep never opened a session, so the reset below proves nothing"
+ );
+ assert_eq!(
+ partition.gap_ticks.get(),
+ 0,
+ "the arm left the debounce saturated; a repair that completes before the \
+ next sweep would then hand the following gap an arm on its first tick"
+ );
+ }
+
+ #[test]
+ fn given_a_local_commit_failure_in_the_tick_walk_when_the_partition_fences_should_return_the_fault()
+ {
+ static GAP_NS: AtomicU64 = AtomicU64::new(0);
+
+ /// Primary -> 2: withhold this group's commit heartbeats, so the last
+ /// produced op stays journaled-but-uncommitted here and the walk the
+ /// test drives below is the first one that can reach it.
+ fn starve_commit_edge(packet: &Packet) -> bool {
+ is_commit_for(packet, GAP_NS.load(Ordering::Relaxed))
+ }
+
+ let (mut sim, client) = cluster(0x5EED_0236);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ GAP_NS.store(namespace.inner(), Ordering::Relaxed);
+
+ produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup");
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) =
+ Some(starve_commit_edge);
+ produce(&mut sim, &client, namespace, 1, "fence");
+
+ let (_, _, commit_min, _) = group_state(&sim, LAGGING, namespace);
+ let stranded = commit_min + 1;
+ assert!(
+ journal_holds(&sim, LAGGING, namespace, stranded),
+ "op {stranded} is not resident, so the walk below would find nothing \
+ to commit and nothing to fail on"
+ );
+
+ // The sweep is invoked DIRECTLY rather than through `sim.step()`: what
+ // is under test is the verdict of the tick whose own walk raised the
+ // fault, and a pump-driven tick would fence on one pass and be asked
+ // for its verdict on the next, where the pre-walk read already answers.
+ let shard = sim.replicas[LAGGING as usize].partition_shard(namespace);
+ {
+ let partitions = shard.plane.partitions();
+ let partition = partitions
+ .get_mut_by_ns(&namespace)
+ .expect("the replica hosts the group");
+ // Committed as far as this replica knows, with the body resident:
+ // walk-stalled, which is the state the tick's backstop walks.
+ partition.consensus().advance_commit_max(stranded);
+ partition.inject_commit_failure();
+ }
+
+ let mut namespace_scratch = Vec::new();
+ let fault = futures::executor::block_on(shard.tick_partitions(&mut namespace_scratch))
+ .expect(
+ "the tick returned clean over a partition its own walk fenced; the pump \
+ would keep serving a divergent replica on that verdict",
+ );
+ assert_eq!(fault.namespace_raw, namespace.inner());
+ assert_eq!(
+ fault.op, stranded,
+ "the fault must name the op whose local commit failed"
+ );
+ assert!(
+ sim.replicas[LAGGING as usize]
+ .partition_shard(namespace)
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .is_some_and(|partition| partition.fatal().is_some()),
+ "the partition must stay fenced after the tick reported the fault"
+ );
+ }
+
+ #[test]
+ fn given_buffered_gap_drops_when_the_namespace_is_removed_should_still_count_them() {
+ static GAP_NS: AtomicU64 = AtomicU64::new(0);
+ static WITHHELD_OP: AtomicU64 = AtomicU64::new(0);
+
+ // Every prepare after the withheld one reaches the backup's gap check
+ // and is destroyed there, which is what the counter records.
+ withhold_one_prepare!(GAP_NS, WITHHELD_OP);
+
+ let (mut sim, client) = cluster(0x5EED_0237);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ GAP_NS.store(namespace.inner(), Ordering::Relaxed);
+ WITHHELD_OP.store(0, Ordering::Relaxed);
+
+ produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup");
+ *sim.network
+ .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) =
+ Some(withhold_one_prepare);
+
+ // Stepped one at a time and stopped on the step that buffered a drop:
+ // the sweep drains the count on the following tick, and the whole point
+ // of this run is to remove the namespace while the count is still on
+ // the partition.
+ let mut buffered = 0;
+ 'produce: for index in 0..GAP_SENDS {
+ let message = client.send_messages(namespace, &[Bytes::from(format!("gap-{index}"))]);
+ sim.submit_request(client.client_id(), 0, message.into_generic());
+ for _ in 0..STEPS_PER_SEND {
+ sim.step();
+ buffered = buffered_gap_drops(&sim, LAGGING, namespace);
+ if buffered > 0 {
+ break 'produce;
+ }
+ }
+ }
+ assert!(
+ buffered > 0,
+ "no prepare reached the lagging replica's gap check, so the removal \
+ below would have nothing to lose"
+ );
+ let counted = gap_drops(&sim, LAGGING, namespace);
+
+ // The reconciler's teardown order: the tombstone lands first and the
+ // disk delete runs before `ConfirmRemove`, so from here `get_mut_by_ns`
+ // answers `None` and the sweep can never drain the count again.
+ {
+ let shard = sim.replicas[LAGGING as usize].partition_shard(namespace);
+ shard.plane.partitions().tombstone(namespace);
+ shard.enqueue_reconcile_op(shard::ReconcileOp::ConfirmRemove { namespace });
+ }
+ sim.step();
+
+ let shard = sim.replicas[LAGGING as usize].partition_shard(namespace);
+ assert!(
+ shard.plane.partitions().get_by_ns(&namespace).is_none(),
+ "ConfirmRemove must have dropped the partition, or this run proves nothing"
+ );
+ assert_eq!(
+ shard.metrics().partition_prepare_gap_drops_value(),
+ counted + buffered,
+ "the {buffered} prepare(s) buffered on the partition went to the floor \
+ with it; the drops are the only record those frames existed"
+ );
+ }
+}
+
#[cfg(test)]
mod metadata_read_frontier_tests {
//! A client that committed a metadata write and then re-homed onto a
diff --git a/scripts/ci/storage-compat.sh b/scripts/ci/storage-compat.sh
index 952e69fb5..edfabe02c 100755
--- a/scripts/ci/storage-compat.sh
+++ b/scripts/ci/storage-compat.sh
@@ -31,6 +31,9 @@ source "$(dirname "${BASH_SOURCE[0]}")/lib/init.sh"
# The test boots the baseline, seeds a data directory, swaps the binary to HEAD
# and restarts against that same directory.
#
+# The baseline's own core/server/config.toml is extracted next to its binary and
+# both boots read it, so the swap changes the binary and nothing else.
+#
# Both binaries MUST be built here. `core/integration` has no dependency on the
# `server` package; the harness only LOCATES a binary, through
# `assert_cmd::Command::cargo_bin`, which falls back to whatever file happens to
@@ -186,6 +189,19 @@ fi
# every run.
BASELINE_SERVER="${TARGET_DIR}/storage-compat/${BASELINE_SHA}/iggy-server"
+# The baseline's own config file, handed to BOTH boots. Neither binary may fall
+# back to the server's relative default path: figment resolves that by walking
+# up from the test process's directory, which hands the BASELINE this branch's
+# file, and a key added under a `deny_unknown_fields` table (every [cluster] and
+# [node] one) then fails its extraction. Written on every run, cached binary or
+# not, so the pair can never drift apart.
+BASELINE_CONFIG="${TARGET_DIR}/storage-compat/${BASELINE_SHA}/config.toml"
+mkdir -p "$(dirname "${BASELINE_CONFIG}")"
+if ! git show "${BASELINE_SHA}:core/server/config.toml" >"${BASELINE_CONFIG}"; then
+ echo "Baseline ${BASELINE_SHA} carries no core/server/config.toml"
+ exit 1
+fi
+
if [ "${REBUILD_BASELINE}" -eq 1 ]; then
rm -f "${BASELINE_SERVER}"
fi
@@ -260,6 +276,7 @@ fi
# Absolute path: ServerHandle only treats this value as a literal path when it
# has more than one component, otherwise it falls back to a cargo_bin lookup.
export COMPAT_BASELINE_SERVER="${BASELINE_SERVER}"
+export COMPAT_BASELINE_CONFIG="${BASELINE_CONFIG}"
echo "Running storage compatibility test..."
# --run-ignored only: the test is #[ignore]d, so the normal lanes skip it.