You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/08/18 11:01:37 UTC
(iggy) branch sim-workload-faults created (now 1771001c8)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 1771001c8 feat(simulator): re-establish a session after the cluster evicts a client
This branch includes the following new commits:
new 46df538e1 fix(simulator): draw the wire consumer kind, not a bare boolean
new c0d9a2104 feat(simulator): select the fuzzer op mix and surface server diagnostics
new f6e5db6a1 feat(simulator): expose the network fault space to the fuzzer
new 72e58db67 feat(simulator): resend timed-out client requests
new 43d293c98 feat(simulator): inject replica restarts with stability windows
new b98d3a5a6 feat(simulator): carry partition logs across a replica restart
new 98936dcb7 fix(metadata): gate the backup gap check on the journal, not the sequencer
new 495db105c fix(simulator): skip re-materialising a namespace the workload deleted
new 06e11a116 feat(simulator): assert cross-replica committed log equality
new 059362667 feat(simulator): drive the workload through the dispatch shell
new deb2e974e feat(simulator): cover register and logout forwarding, and measure command reach
new 3d49a07cb feat(simulator): reach state transfer through the snapshot floor
new 007e6537d feat(simulator): checkpoint for real and serve a chunked state transfer
new 1b5bb4a2b chore(simulator): record tempfile dev-dependency in the lockfile
new a1c8918d8 fix(simulator): send the client PAT shape through the dispatch layer
new 9b0dd0586 style: satisfy the pedantic lints the new simulator code introduced
new 4e9d69cc5 fix(simulator): restore the partition consensus frontier on restart
new 1771001c8 feat(simulator): re-establish a session after the cluster evicts a client
The 18 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 12/18: feat(simulator): reach state transfer through the snapshot floor
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3d49a07cbd63f947bf94ea5860c29900c817f30b
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 09:15:12 2026 +0530
feat(simulator): reach state transfer through the snapshot floor
`SimJournal` answered a constant 0 for its snapshot watermark and dropped
`set_snapshot_op` on the floor. The repair server floors what it serves at
`snapshot_op + 1` and announces the skipped prefix as `RangeEvicted`, so a
watermark stuck at zero made no range evictable and left `RangeEvicted`
unreachable however the cluster was driven. That matters more than one
frame: `RangeEvicted` is the single authoritative "journal repair cannot
close this gap" signal, and the only route into state transfer. Every way
a replica falls behind a checkpoint converges on it.
The watermark is now real, and the harness can stamp one. Stamping the
number alone is enough for the whole escalation, which splits the problem
honestly: the simulator has no `SnapshotCoordinator` (metadata is built
without a data directory, so `checkpoint_if_needed` returns immediately),
and inventing snapshot bytes here would have modelled a checkpoint that
never happened.
A lagging replica whose gap sits entirely below its peers' floor now gets
told so and converts to a transfer: RangeEvicted, RequestStateTransfer and
StateTransferTarget all cross the wire, three of the five frames that no
run had ever delivered. It stops there. `RequestStateChunk` and
`StateChunk` need artifacts to fetch and the serving peer has no checkpoint
to build them from, so the requester re-arms each round instead of
converging. The test asserts the handshake it does reach rather than the
transfer completing, and says why.
Removing the stamp fails the test on its first assertion, so it cannot pass
without the mechanism it covers.
---
core/simulator/src/deps.rs | 28 ++++++++--
core/simulator/src/lib.rs | 134 +++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 156 insertions(+), 6 deletions(-)
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index f44ab3ce9..95a16c019 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -104,6 +104,9 @@ pub struct SimJournal<S: Storage> {
/// retained head in O(1) without scanning `headers` (see
/// [`SimJournal::last_op`]).
last_op: Cell<Option<u64>>,
+ /// Snapshot watermark; see the `Journal::snapshot_op` impl for why a real
+ /// value here is what makes `RangeEvicted` reachable at all.
+ snapshot_op: Cell<u64>,
/// Debug-only single-accessor tripwire. `entry` / `append` hold a
/// [`JournalAccessGuard`] across their whole body, including the storage
/// `.await`, so if a suspending storage tier ever let a second task touch
@@ -121,6 +124,7 @@ impl<S: Storage + Default> Default for SimJournal<S> {
offsets: UnsafeCell::new(HashMap::new()),
write_offset: Cell::new(0),
last_op: Cell::new(None),
+ snapshot_op: Cell::new(0),
#[cfg(debug_assertions)]
accessing: Cell::new(false),
}
@@ -207,15 +211,27 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal<S> for SimJournal<S> {
Ok(doomed.len())
}
- /// The simulated journal retains everything for the run, so nothing is
- /// ever superseded by a snapshot. Answered explicitly (the trait has no
- /// default) so a simulated state transfer has to opt into a watermark
- /// rather than silently inherit one that never moves.
+ /// The snapshot watermark: entries at or below it are evictable.
+ ///
+ /// Load-bearing despite the simulated journal retaining every entry it is
+ /// given. The repair server floors what it will serve at `snapshot_op + 1`
+ /// and announces the skipped prefix as `RangeEvicted`, which is the one
+ /// authoritative "repair cannot close this gap" signal and the only route
+ /// into state transfer. While this answered a constant 0, no range was ever
+ /// evictable, so `RangeEvicted` and every state-transfer frame behind it were
+ /// unreachable no matter what else the harness did.
fn snapshot_op(&self) -> u64 {
- 0
+ self.snapshot_op.get()
}
- fn set_snapshot_op(&self, _op: u64) {}
+ /// Advance the watermark, never lower it. Production advances it when a
+ /// checkpoint supersedes a prefix or a transfer installs one; both only move
+ /// forward, and a retreating floor would re-offer ops the serving side has
+ /// already told a peer are gone.
+ fn set_snapshot_op(&self, op: u64) {
+ let current = self.snapshot_op.get();
+ self.snapshot_op.set(current.max(op));
+ }
// TODO(hubcio): validate that the caller's checksum matches the stored
// header - currently this looks up by op only, ignoring the checksum.
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 9ad034d00..84829852b 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -645,6 +645,31 @@ impl Simulator {
client_replies
}
+ /// Stamp a metadata snapshot watermark on one replica, standing in for a
+ /// checkpoint that superseded everything at or below `op`.
+ ///
+ /// The simulator does not checkpoint: `IggyMetadata` is built without a data
+ /// directory, so it has no `SnapshotCoordinator` and `checkpoint_if_needed`
+ /// returns immediately. That left the watermark permanently at zero, so
+ /// nothing was ever evictable, the repair server never had a compacted prefix
+ /// to skip, and `RangeEvicted` -- the only signal that converts a repair into
+ /// a state transfer -- could not occur however the cluster was driven.
+ ///
+ /// Stamping the number alone reaches the whole escalation without producing
+ /// snapshot bytes to transfer. That splits the problem honestly: the protocol
+ /// path can be covered now, while a transfer that installs real state needs
+ /// the coordinator seam.
+ ///
+ /// # Panics
+ /// If the replica has no metadata journal (only shard 0 owns one).
+ pub fn stamp_metadata_snapshot(&self, replica_idx: usize, op: u64) {
+ use journal::Journal;
+
+ self.replicas[replica_idx]
+ .metadata_journal
+ .set_snapshot_op(op);
+ }
+
/// Take the clients evicted since the last call.
///
/// A driver must consume these: an evicted client's session is gone, so its
@@ -3289,6 +3314,115 @@ mod tests {
/// pipeline to `PIPELINE_PREPARE_QUEUE_MAX`; a request on `ns_b`
/// still commits while `ns_a` is wedged (no quorum without backup
/// acks); lifting the block drains `ns_a` completely.
+ /// A backup whose gap sits below the serving peer's snapshot floor is told
+ /// `RangeEvicted` and converts its repair into a state transfer.
+ ///
+ /// This is the only route into state transfer. Journal repair closes a gap by
+ /// replaying ops; when the peer has compacted past them there is nothing left
+ /// to replay, and `RangeEvicted` is the authoritative signal saying so. Every
+ /// way a replica falls behind a checkpoint converges on it -- a fresh node
+ /// joining an already-checkpointed cluster, a healed partition, a restart
+ /// whose gap is below the floor.
+ ///
+ /// The simulator never checkpoints (no `SnapshotCoordinator`, see
+ /// `Simulator::stamp_metadata_snapshot`), so the watermark sat at zero and
+ /// this whole branch was unreachable. Stamping the number is enough for the
+ /// escalation; the transfer cannot install real state without snapshot bytes,
+ /// which is the next task.
+ ///
+ /// Asserts on frames delivered rather than on internal state, so it fails if
+ /// the escalation stops happening even should the local bookkeeping still look
+ /// plausible.
+ #[test]
+ fn repair_below_the_snapshot_floor_escalates_to_state_transfer() {
+ use iggy_binary_protocol::Command2;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0xE71C_0001,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+
+ // Commit a prefix the lagging replica will later be missing.
+ for sequence in 0..6u32 {
+ let msg = client.create_stream(&format!("wl-before-{sequence}"));
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..60 {
+ sim.step();
+ }
+ }
+
+ // Take a backup down and keep committing, so its log falls behind.
+ let lagging = 2u8;
+ sim.replica_crash(lagging);
+ for sequence in 0..6u32 {
+ let msg = client.create_stream(&format!("wl-during-{sequence}"));
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..60 {
+ sim.step();
+ }
+ }
+
+ // Compact past everything the lagging replica is missing, on the peers
+ // that will serve its repair. Its gap is now entirely below their floor,
+ // which is exactly the state no amount of journal repair can fix.
+ let floor = metadata_commit(&sim, 0);
+ assert!(
+ floor > 0,
+ "nothing committed, so there is no prefix to compact and nothing to evict"
+ );
+ for replica_idx in 0..usize::from(replica_count) {
+ if replica_idx != usize::from(lagging) {
+ sim.stamp_metadata_snapshot(replica_idx, floor);
+ }
+ }
+
+ sim.replica_restart(lagging);
+ for _ in 0..4_000 {
+ sim.step();
+ }
+
+ assert!(
+ sim.network.delivered_any(Command2::RangeEvicted),
+ "no RangeEvicted: the serving peers never announced a compacted \
+ prefix, so the rejoining replica is still trying to repair ops that \
+ cannot be served"
+ );
+ assert!(
+ sim.network.delivered_any(Command2::RequestStateTransfer),
+ "RangeEvicted did not convert into a state transfer, so the replica \
+ has no route back to the cluster's state"
+ );
+ assert!(
+ sim.network.delivered_any(Command2::StateTransferTarget),
+ "the serving peer never answered with a transfer target, so the \
+ escalation stopped at the request and covers only half the handshake"
+ );
+
+ // The handshake reaches the target descriptor and goes no further:
+ // `RequestStateChunk` and `StateChunk` need artifacts to fetch, and the
+ // serving peer has no checkpoint to build them from. The requester
+ // therefore re-arms every round rather than converging, which is why this
+ // test asserts on the handshake instead of on the transfer completing.
+ // Wiring a snapshot coordinator is what closes that gap.
+ }
+
/// A client that dials a BACKUP still gets a working session, and the login
/// travels as a forwarded consensus proposal rather than a redirect.
///
(iggy) 16/18: style: satisfy the pedantic lints the new simulator code introduced
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 9b0dd0586dfbddc6c98c56fe24715ec1f65a188b
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 12:18:58 2026 +0530
style: satisfy the pedantic lints the new simulator code introduced
Ten diagnostics across the batch's own code: seven denied under the
crate's `pedantic` setting and three warnings that CI's `-D warnings`
would have turned into failures too. Missing `#[must_use]` and `# Panics`,
doc paragraphs and backticks, an argument taken by value and never
consumed, and a return type wrapped in an `Option` that was never `None`.
`await_setup_reply` returning `Option` was the one worth more than a
lint fix: it panics when the handshake never completes, so the `None` its
callers unwrapped through could not occur, and both of them dealt with it
by defaulting a session id to zero. Returning the reply directly removes
a branch that only ever obscured the real failure.
---
core/binary_protocol/src/primitives/consumer.rs | 9 +--
core/metadata/src/impls/metadata.rs | 6 +-
core/simulator/src/bin/workload-fuzz.rs | 88 ++++++++++++++-----------
core/simulator/src/lib.rs | 57 ++++++++--------
core/simulator/src/workload/auditor.rs | 1 +
core/simulator/src/workload/mod.rs | 9 +--
core/simulator/src/workload/options.rs | 12 +++-
core/simulator/src/workload/state_checker.rs | 7 +-
8 files changed, 106 insertions(+), 83 deletions(-)
diff --git a/core/binary_protocol/src/primitives/consumer.rs b/core/binary_protocol/src/primitives/consumer.rs
index 1e7739b37..cc44f4479 100644
--- a/core/binary_protocol/src/primitives/consumer.rs
+++ b/core/binary_protocol/src/primitives/consumer.rs
@@ -20,10 +20,11 @@ use crate::WireIdentifier;
use crate::codec::{WireDecode, WireEncode, read_u8};
use bytes::{BufMut, BytesMut};
-/// Wire discriminant for a single consumer (vs a `ConsumerGroup`). Public for
-/// the same reason as its sibling below, plus one more: `decode` accepts only
-/// these two values, so anything synthesizing a consumer needs to name them
-/// rather than guess a small integer.
+/// Wire discriminant for a single consumer (vs a `ConsumerGroup`).
+///
+/// Public for the same reason as its sibling below, plus one more: `decode`
+/// accepts only these two values, so anything synthesizing a consumer needs to
+/// name them rather than guess a small integer.
pub const KIND_CONSUMER: u8 = 1;
/// Wire discriminant for a consumer-group consumer (vs a single `Consumer`).
/// Public so the server dispatch can match on it by name instead of a raw `2`.
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index b3b682d02..700b8d87e 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -4787,9 +4787,9 @@ mod tests {
/// metadata workload under crash/restart injection until this was gated on the
/// journal instead.
///
- /// TigerBeetle sidesteps the question by keeping a single frontier: `self.op`
- /// IS the log head, and `on_prepare` routes anything at or below it to
- /// `on_repair`, which re-acks a prepare already held.
+ /// `TigerBeetle` sidesteps the question by keeping a single frontier:
+ /// `self.op` IS the log head, and `on_prepare` routes anything at or below it
+ /// to `on_repair`, which re-acks a prepare already held.
#[compio::test]
async fn backup_admits_the_prepare_its_journal_needs_despite_a_leading_sequencer() {
const CLIENT: u128 = 1;
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index 940120387..dd0720cf4 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -136,7 +136,7 @@ struct Args {
clog_duration_mean: Option<u64>,
}
-/// Named network fault profile, in the spirit of TigerBeetle's VOPR modes: one
+/// Named network fault profile, in the spirit of `TigerBeetle`'s VOPR modes: one
/// flag for "how hostile is the network", rather than eleven.
///
/// Progress falls off steeply with severity, because every lost frame costs a
@@ -388,40 +388,7 @@ fn main() {
bucket_capacity: 1,
});
- let client_ids: Vec<u128> = (1..=u128::from(clients)).collect();
- let mut sim = if args.shell {
- Simulator::with_shards_shell(
- usize::from(replicas),
- 1,
- client_ids.iter().copied(),
- network_opts,
- )
- } else {
- Simulator::new(
- usize::from(replicas),
- client_ids.iter().copied(),
- network_opts,
- )
- };
- let sim_clients: Vec<SimClient> = client_ids.iter().map(|&id| SimClient::new(id)).collect();
-
- let ns = IggyNamespace::new(1, 1, 0);
- sim.init_partition(ns);
- if args.shell {
- // The shell resolves a partition request's namespace against committed
- // metadata, so the stream and topic behind it have to exist as well as the
- // partition group.
- sim.seed_stream_topic_partition(ns);
- }
- for client in &sim_clients {
- if args.shell {
- // Log in rather than bare-register: the dispatch path admits a request
- // only from a bound session, and the login is what mints one.
- sim.shell_login(client);
- } else {
- sim.register_client_with_primary(client);
- }
- }
+ let (mut sim, sim_clients, ns) = build_cluster(&args, replicas, clients, network_opts);
let mut options = WorkloadOptions::new(seed, replicas, vec![ns]);
options.client_count = clients;
@@ -484,6 +451,51 @@ fn main() {
println!("workload-fuzz: OK (seed={seed})");
}
+/// Stand up the cluster, seed its namespace, and get every client a session.
+///
+/// Returns the simulator, its clients, and the namespace the workload drives.
+/// The shell path differs in two ways that have to agree: a partition request's
+/// namespace is resolved against committed metadata, so the stream and topic
+/// behind it must exist and not just the partition group; and dispatch admits a
+/// request only from a bound session, which only a login mints.
+fn build_cluster(
+ args: &Args,
+ replicas: u8,
+ clients: u8,
+ network_opts: PacketSimulatorOptions,
+) -> (Simulator, Vec<SimClient>, IggyNamespace) {
+ let client_ids: Vec<u128> = (1..=u128::from(clients)).collect();
+ let mut sim = if args.shell {
+ Simulator::with_shards_shell(
+ usize::from(replicas),
+ 1,
+ client_ids.iter().copied(),
+ network_opts,
+ )
+ } else {
+ Simulator::new(
+ usize::from(replicas),
+ client_ids.iter().copied(),
+ network_opts,
+ )
+ };
+ let sim_clients: Vec<SimClient> = client_ids.iter().map(|&id| SimClient::new(id)).collect();
+
+ let ns = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns);
+ if args.shell {
+ sim.seed_stream_topic_partition(ns);
+ }
+ for client in &sim_clients {
+ if args.shell {
+ sim.shell_login(client);
+ } else {
+ sim.register_client_with_primary(client);
+ }
+ }
+ (sim, sim_clients, ns)
+}
+
/// Which protocol commands the run actually delivered, and which it never
/// reached.
///
@@ -524,9 +536,9 @@ fn print_coverage(workload: &Workload) {
);
for action in Action::iter() {
let commits = stats.commits(action);
- let (denied, status) = stats.denials_per_action[action as usize];
- if commits > 0 || denied > 0 {
- println!(" {action:?}: {commits} commits, {denied} denied (last status {status})");
+ let (refused, code) = stats.denials_per_action[action as usize];
+ if commits > 0 || refused > 0 {
+ println!(" {action:?}: {commits} commits, {refused} denied (last status {code})");
}
}
}
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index a266dc65d..4a0b3a08d 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -281,7 +281,7 @@ impl Simulator {
replica_count: usize,
clients: impl Iterator<Item = u128>,
network_options: PacketSimulatorOptions,
- data_dir_root: std::path::PathBuf,
+ data_dir_root: &std::path::Path,
) -> Self {
Self::build_inner(
replica_count,
@@ -326,7 +326,7 @@ impl Simulator {
clients: impl Iterator<Item = u128>,
network_options: PacketSimulatorOptions,
shell: bool,
- data_dir_root: Option<std::path::PathBuf>,
+ data_dir_root: Option<&std::path::Path>,
) -> Self {
assert!(
shards_per_replica >= 1,
@@ -583,7 +583,8 @@ impl Simulator {
// (`build_reply_with_body` maps the session field to `op`).
let session = self
.await_setup_reply(client.client_id(), target, &msg, "shell_login")
- .map_or(0, |reply| reply.header().op);
+ .header()
+ .op;
assert!(session > 0, "shell_login: login reply carried no session");
client.bind_session(session);
}
@@ -591,22 +592,26 @@ impl Simulator {
/// Submit `message` to `target` and step until a client reply arrives,
/// resubmitting every [`SETUP_RETRY_STEPS`] steps.
///
- /// Returns the first reply, or `None` once [`SETUP_TOTAL_STEPS`] is spent.
/// The same message is resubmitted verbatim, so the request id is stable and
/// the metadata client table treats a retry as a duplicate.
+ ///
+ /// # Panics
+ /// If no reply arrives within [`SETUP_TOTAL_STEPS`]. A setup handshake that
+ /// never completes leaves the fixture unusable, so there is no useful
+ /// `None` for a caller to handle.
fn await_setup_reply(
&mut self,
client_id: u128,
target: u8,
message: &Message<GenericHeader>,
label: &str,
- ) -> Option<Message<ReplyHeader>> {
+ ) -> Message<ReplyHeader> {
for step in 0..SETUP_TOTAL_STEPS {
if step % SETUP_RETRY_STEPS == 0 {
self.submit_request(client_id, target, message.deep_copy());
}
if let Some(reply) = self.step().into_iter().next() {
- return Some(reply);
+ return reply;
}
}
panic!(
@@ -852,26 +857,22 @@ impl Simulator {
#[allow(clippy::cast_possible_truncation)]
pub fn register_client_with_primary(&mut self, client: &SimClient) {
let msg = client.register().into_generic();
- let session = self
- .await_setup_reply(client.client_id(), 0, &msg, "register_client_with_primary")
- .map_or(0, |reply| {
- let header = reply.header();
- debug_assert_eq!(
- header.operation,
- iggy_binary_protocol::Operation::Register,
- "register_client_with_primary: first reply was not Register"
- );
- assert_eq!(
- header.client,
- client.client_id(),
- "register_client_with_primary: reply client_id mismatch \
- (expected {}, got {})",
- client.client_id(),
- header.client,
- );
- header.commit
- });
- client.bind_session(session);
+ let reply =
+ self.await_setup_reply(client.client_id(), 0, &msg, "register_client_with_primary");
+ let header = reply.header();
+ debug_assert_eq!(
+ header.operation,
+ iggy_binary_protocol::Operation::Register,
+ "register_client_with_primary: first reply was not Register"
+ );
+ assert_eq!(
+ header.client,
+ client.client_id(),
+ "register_client_with_primary: reply client_id mismatch (expected {}, got {})",
+ client.client_id(),
+ header.client,
+ );
+ client.bind_session(header.commit);
// Partition has no `client_table`: at-least-once, no per-client
// dedup. Consumers dedup via message id / content / producer-id+seq.
@@ -3441,7 +3442,7 @@ mod tests {
usize::from(replica_count),
std::iter::once(client_id),
network_opts,
- root.path().to_path_buf(),
+ root.path(),
);
// Small enough that the ops below cross the margin; the coordinator forces
// a checkpoint once free slots fall to its margin (64 by default).
@@ -3687,7 +3688,7 @@ mod tests {
let network_opts = packet::PacketSimulatorOptions {
node_count: replica_count,
client_count: 1,
- seed: 0xF0_2D_0001,
+ seed: 0xF02D_0001,
..packet::PacketSimulatorOptions::default()
};
let mut sim = Simulator::with_shards_shell(
diff --git a/core/simulator/src/workload/auditor.rs b/core/simulator/src/workload/auditor.rs
index 903117935..96cd09258 100644
--- a/core/simulator/src/workload/auditor.rs
+++ b/core/simulator/src/workload/auditor.rs
@@ -217,6 +217,7 @@ impl ServerAuditor {
self.in_flight.get(&key).map(|entry| entry.action)
}
+ #[must_use]
pub fn in_flight_count(&self) -> usize {
self.in_flight.len()
}
diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs
index 564082bfd..9e25f9f82 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -141,7 +141,7 @@ impl Workload {
/// Advance the resend clock by one tick. Called once per driver iteration;
/// [`Self::due_resends`] measures against it.
- pub fn tick(&mut self) {
+ pub const fn tick(&mut self) {
self.now += 1;
}
@@ -550,9 +550,10 @@ pub fn run_with_faults(
replies_seen
}
-/// Crash and restart injection with stability windows, in the shape of
-/// TigerBeetle's VOPR: a crash must last a while before it may be repaired, and
-/// a repaired replica must run a while before it may fail again.
+/// Crash and restart injection with stability windows.
+///
+/// Shaped after `TigerBeetle`'s VOPR: a crash must last a while before it may be
+/// repaired, and a repaired replica must run a while before it may fail again.
///
/// Owns the fault PRNG so crash scheduling stays reproducible from the seed yet
/// independent of the traffic draw order. Draws nothing while both probabilities
diff --git a/core/simulator/src/workload/options.rs b/core/simulator/src/workload/options.rs
index f5fd38acb..c21b12d50 100644
--- a/core/simulator/src/workload/options.rs
+++ b/core/simulator/src/workload/options.rs
@@ -32,9 +32,11 @@ pub const DEFAULT_REQUEST_TIMEOUT_TICKS: u64 = 200;
/// something to repair.
pub const DEFAULT_CRASH_STABILITY_TICKS: u64 = 300;
-/// Default [`WorkloadOptions::restart_stability_ticks`]. Long enough for a
-/// rejoined replica to finish catching up before it becomes a crash candidate
-/// again, so a run does not consist entirely of half-repaired replicas.
+/// Default [`WorkloadOptions::restart_stability_ticks`].
+///
+/// Long enough for a rejoined replica to finish catching up before it becomes a
+/// crash candidate again, so a run does not consist entirely of half-repaired
+/// replicas.
pub const DEFAULT_RESTART_STABILITY_TICKS: u64 = 500;
/// Per-action sampling weights as percentages. Unlisted variants default
@@ -116,6 +118,10 @@ impl ActionWeights {
/// `Action::COUNT` does not divide 100, so the first `100 % COUNT` actions
/// carry one extra point. Spread that way rather than asserting the count
/// divides evenly, so appending an `Action` never breaks this preset.
+ ///
+ /// # Panics
+ /// Panics if the spread weights do not sum to 100, which would mean the
+ /// remainder arithmetic above is wrong rather than the caller.
#[must_use]
pub fn uniform() -> Self {
use strum::IntoEnumIterator;
diff --git a/core/simulator/src/workload/state_checker.rs b/core/simulator/src/workload/state_checker.rs
index a3c5135f8..f2ffa3fef 100644
--- a/core/simulator/src/workload/state_checker.rs
+++ b/core/simulator/src/workload/state_checker.rs
@@ -15,8 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-//! Cross-replica committed-log equality, after TigerBeetle's
-//! `testing/cluster/state_checker.zig`.
+//! Cross-replica committed-log equality.
+//!
+//! After `TigerBeetle`'s `testing/cluster/state_checker.zig`.
//!
//! The per-tick checks in [`super::invariants`] catch a single replica
//! contradicting itself, and [`super::oracle`] compares committed metadata
@@ -24,7 +25,7 @@
//! is the actual consensus property: two replicas that both committed op N must
//! have committed the same op N.
//!
-//! Modelled on TigerBeetle's checker rather than invented: it keeps one canonical
+//! Modelled on `TigerBeetle`'s checker rather than invented: it keeps one canonical
//! commit chain, asserts every replica agrees with it wherever they overlap
//! (`(commit_a == commit_b) == (checksum_a == checksum_b)`), and asserts the chain
//! is hash-linked (`header_b.parent == checksum_a`). Recording which replicas
(iggy) 03/18: feat(simulator): expose the network fault space to the fuzzer
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit f6e5db6a1f17cc68856c5894550f6ba139fea719
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 14:32:37 2026 +0530
feat(simulator): expose the network fault space to the fuzzer
`PacketSimulatorOptions` has carried loss, duplication, partition and
clog knobs since the packet simulator landed, all defaulting to zero,
and the fuzzer offered no way to set any of them. The fault space the
harness exists to explore was therefore reachable only from
hand-written tests.
`--faults` names a whole profile so a run does not have to spell out
eleven probabilities, and the individual flags override single fields of
the chosen profile for when the question is about one axis. The run
prints the resulting network so a reproduction carries what was injected
rather than what was asked for. Default stays a perfect network.
Every profile past `none` currently stalls, and not from a cluster
failure: `SimClient` has no request timeout, so the first dropped
request or reply strands its only in-flight slot for the rest of the
run. A 2% loss rate drains four replies where a perfect network drains
439. The profiles are wired anyway, since the fault space belongs in one
place, and the client resend path is the next commit.
---
core/simulator/src/bin/workload-fuzz.rs | 238 ++++++++++++++++++++++++++++++--
1 file changed, 229 insertions(+), 9 deletions(-)
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index e3eebfb50..52fe0eeae 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -25,12 +25,19 @@
//! ```text
//! workload-fuzz [--seed N] [--ticks N] [--clients N] [--replicas N]
//! [--plane partition|metadata|mixed|uniform]
-//! [--crash-prob F] [--no-quiesce]
+//! [--faults none|light|heavy] [--crash-prob F] [--no-quiesce]
+//! [network overrides: --packet-loss, --replay, --partition-mode,
+//! --partition-prob, --unpartition-prob, --clog-prob, ...]
//! ```
//!
//! `--plane` selects the op mix (see [`ActionWeights`]). Partition-plane runs
//! drain and converge most readily; `uniform` is the widest per-tick op
//! coverage.
+//!
+//! `--faults` picks a whole network fault profile; the individual network flags
+//! override single fields of the chosen profile, so exploring one axis does not
+//! mean spelling out the other ten. The default profile is `none`, a perfect
+//! network, so a run says what it injects rather than inheriting it.
use clap::{Parser, ValueEnum};
use iggy_common::IggyByteSize;
@@ -38,7 +45,7 @@ use server_common::sharding::IggyNamespace;
use server_common::{MemoryPool, MemoryPoolConfigOther};
use simulator::Simulator;
use simulator::client::SimClient;
-use simulator::packet::PacketSimulatorOptions;
+use simulator::packet::{PacketSimulatorOptions, PartitionMode, PartitionSymmetry};
use simulator::workload::actions::Action;
use simulator::workload::options::{ActionWeights, WorkloadOptions};
use simulator::workload::{Workload, oracle, run};
@@ -67,6 +74,147 @@ struct Args {
crash_prob: f32,
#[arg(long)]
no_quiesce: bool,
+
+ /// Network fault profile. Individual network flags below override single
+ /// fields of the profile.
+ #[arg(long, value_enum, default_value_t = Faults::None)]
+ faults: Faults,
+ /// Chance a packet is dropped at delivery time.
+ #[arg(long, value_parser = parse_unit_interval_f64)]
+ packet_loss: Option<f64>,
+ /// Chance a packet is duplicated at delivery time.
+ #[arg(long, value_parser = parse_unit_interval_f64)]
+ replay: Option<f64>,
+ /// Minimum one-way delay, in ticks.
+ #[arg(long)]
+ one_way_delay_min: Option<u64>,
+ /// Mean one-way delay, in ticks (exponentially distributed).
+ #[arg(long)]
+ one_way_delay_mean: Option<u64>,
+ /// Maximum packets queued on a single link; beyond it the link drops.
+ #[arg(long)]
+ link_capacity: Option<u8>,
+ /// How an automatic partition picks its sides.
+ #[arg(long, value_enum)]
+ partition_mode: Option<PartitionModeArg>,
+ /// Whether a partition blocks both directions or just one.
+ #[arg(long, value_enum)]
+ partition_symmetry: Option<PartitionSymmetryArg>,
+ /// Per-tick chance a partition forms while connectivity is whole.
+ #[arg(long, value_parser = parse_unit_interval_f64)]
+ partition_prob: Option<f64>,
+ /// Per-tick chance a standing partition heals.
+ #[arg(long, value_parser = parse_unit_interval_f64)]
+ unpartition_prob: Option<f64>,
+ /// Minimum ticks a partition lasts once formed.
+ #[arg(long)]
+ partition_stability: Option<u32>,
+ /// Minimum ticks of whole connectivity before another partition may form.
+ #[arg(long)]
+ unpartition_stability: Option<u32>,
+ /// Per-tick chance any one path clogs (stops delivering, keeps queueing).
+ #[arg(long, value_parser = parse_unit_interval_f64)]
+ clog_prob: Option<f64>,
+ /// Mean clog duration, in ticks (exponentially distributed).
+ #[arg(long)]
+ clog_duration_mean: Option<u64>,
+}
+
+/// Named network fault profile, in the spirit of TigerBeetle's VOPR modes: one
+/// flag for "how hostile is the network", rather than eleven.
+///
+/// Anything but [`Faults::None`] currently stalls the run, and not because the
+/// cluster fails to make progress: `SimClient` has no request timeout, so a
+/// client holds its single in-flight slot forever once the request or its reply
+/// is dropped. Every profile below is therefore write-once-read-later until the
+/// client grows a resend path; they are wired now so the fault space is
+/// described in one place rather than rediscovered later.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
+enum Faults {
+ /// Perfect network. Delays only, no loss and no partitions.
+ None,
+ /// Occasional loss, duplication and short one-sided partitions. Meant to
+ /// stay inside the range where a healthy cluster still drains.
+ Light,
+ /// Frequent loss, long partitions and clogged paths. Expected to stall
+ /// progress for stretches; use with a generous tick budget.
+ Heavy,
+}
+
+/// Clap mirror of [`PartitionMode`], so the library type stays free of a clap
+/// derive.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
+enum PartitionModeArg {
+ None,
+ UniformSize,
+ UniformPartition,
+ IsolateSingle,
+}
+
+impl From<PartitionModeArg> for PartitionMode {
+ fn from(value: PartitionModeArg) -> Self {
+ match value {
+ PartitionModeArg::None => Self::None,
+ PartitionModeArg::UniformSize => Self::UniformSize,
+ PartitionModeArg::UniformPartition => Self::UniformPartition,
+ PartitionModeArg::IsolateSingle => Self::IsolateSingle,
+ }
+ }
+}
+
+/// Clap mirror of [`PartitionSymmetry`]; see [`PartitionModeArg`].
+#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
+enum PartitionSymmetryArg {
+ Symmetric,
+ Asymmetric,
+}
+
+impl From<PartitionSymmetryArg> for PartitionSymmetry {
+ fn from(value: PartitionSymmetryArg) -> Self {
+ match value {
+ PartitionSymmetryArg::Symmetric => Self::Symmetric,
+ PartitionSymmetryArg::Asymmetric => Self::Asymmetric,
+ }
+ }
+}
+
+impl Faults {
+ /// Base network options for this profile. `node_count`, `client_count` and
+ /// `seed` are filled by the caller.
+ fn options(self) -> PacketSimulatorOptions {
+ match self {
+ // `PacketSimulatorOptions::default` is already a perfect network:
+ // delay only, every probability zero.
+ Self::None => PacketSimulatorOptions::default(),
+ Self::Light => PacketSimulatorOptions {
+ packet_loss_probability: 0.02,
+ replay_probability: 0.01,
+ partition_probability: 0.005,
+ unpartition_probability: 0.05,
+ partition_stability: 20,
+ unpartition_stability: 40,
+ partition_mode: PartitionMode::IsolateSingle,
+ partition_symmetry: PartitionSymmetry::Asymmetric,
+ path_clog_probability: 0.002,
+ path_clog_duration_mean: 10,
+ ..PacketSimulatorOptions::default()
+ },
+ Self::Heavy => PacketSimulatorOptions {
+ packet_loss_probability: 0.10,
+ replay_probability: 0.03,
+ one_way_delay_mean: 8,
+ partition_probability: 0.02,
+ unpartition_probability: 0.02,
+ partition_stability: 50,
+ unpartition_stability: 50,
+ partition_mode: PartitionMode::UniformSize,
+ partition_symmetry: PartitionSymmetry::Asymmetric,
+ path_clog_probability: 0.01,
+ path_clog_duration_mean: 25,
+ ..PacketSimulatorOptions::default()
+ },
+ }
+ }
}
/// Which plane the sampled ops target. Maps onto an [`ActionWeights`] preset.
@@ -105,6 +253,68 @@ fn parse_unit_interval(raw: &str) -> Result<f32, String> {
}
}
+/// [`parse_unit_interval`] for the network knobs, which are `f64`.
+fn parse_unit_interval_f64(raw: &str) -> Result<f64, String> {
+ let value: f64 = raw
+ .parse()
+ .map_err(|_| format!("`{raw}` is not a number"))?;
+ if (0.0..=1.0).contains(&value) {
+ Ok(value)
+ } else {
+ Err(format!("must be within [0.0, 1.0], got {value}"))
+ }
+}
+
+/// The chosen fault profile with any individually-set network flag applied over
+/// it, plus the cluster shape and seed.
+fn network_options(args: &Args, replicas: u8, clients: u8, seed: u64) -> PacketSimulatorOptions {
+ let mut options = args.faults.options();
+ options.node_count = replicas;
+ options.client_count = clients;
+ options.seed = seed;
+
+ if let Some(value) = args.packet_loss {
+ options.packet_loss_probability = value;
+ }
+ if let Some(value) = args.replay {
+ options.replay_probability = value;
+ }
+ if let Some(value) = args.one_way_delay_min {
+ options.one_way_delay_min = value;
+ }
+ if let Some(value) = args.one_way_delay_mean {
+ options.one_way_delay_mean = value;
+ }
+ if let Some(value) = args.link_capacity {
+ options.link_capacity = value;
+ }
+ if let Some(value) = args.partition_mode {
+ options.partition_mode = value.into();
+ }
+ if let Some(value) = args.partition_symmetry {
+ options.partition_symmetry = value.into();
+ }
+ if let Some(value) = args.partition_prob {
+ options.partition_probability = value;
+ }
+ if let Some(value) = args.unpartition_prob {
+ options.unpartition_probability = value;
+ }
+ if let Some(value) = args.partition_stability {
+ options.partition_stability = value;
+ }
+ if let Some(value) = args.unpartition_stability {
+ options.unpartition_stability = value;
+ }
+ if let Some(value) = args.clog_prob {
+ options.path_clog_probability = value;
+ }
+ if let Some(value) = args.clog_duration_mean {
+ options.path_clog_duration_mean = value;
+ }
+ options
+}
+
fn main() {
let args = Args::parse();
@@ -134,9 +344,25 @@ fn main() {
eprintln!("workload-fuzz FAILED — reproduce with --seed {seed}\n{info}");
}));
+ let network_opts = network_options(&args, replicas, clients, seed);
println!(
"workload-fuzz: seed={seed} ticks={ticks} clients={clients} replicas={replicas} \
- plane={plane:?} crash_prob={crash_prob} quiesce={quiesce}"
+ plane={plane:?} faults={:?} crash_prob={crash_prob} quiesce={quiesce}",
+ args.faults,
+ );
+ println!(
+ "network: loss={} replay={} delay={}..{} partition={:?}/{:?} \
+ p_partition={} p_unpartition={} clog={} link_capacity={}",
+ network_opts.packet_loss_probability,
+ network_opts.replay_probability,
+ network_opts.one_way_delay_min,
+ network_opts.one_way_delay_mean,
+ network_opts.partition_mode,
+ network_opts.partition_symmetry,
+ network_opts.partition_probability,
+ network_opts.unpartition_probability,
+ network_opts.path_clog_probability,
+ network_opts.link_capacity,
);
// poll_messages / reply paths panic without an initialized pool; disabled
@@ -148,12 +374,6 @@ fn main() {
});
let client_ids: Vec<u128> = (1..=u128::from(clients)).collect();
- let network_opts = PacketSimulatorOptions {
- node_count: replicas,
- client_count: clients,
- seed,
- ..PacketSimulatorOptions::default()
- };
let mut sim = Simulator::new(
usize::from(replicas),
client_ids.iter().copied(),
(iggy) 15/18: fix(simulator): send the client PAT shape through the dispatch layer
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit a1c8918d874a107e112ce599fc1b5b051cce7f9b
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 12:08:49 2026 +0530
fix(simulator): send the client PAT shape through the dispatch layer
Both personal-access-token operations were refused with `InvalidCommand`
on every attempt through the dispatch shell, never once committing, while
the raw path accepted them. `SimClient` sent the REPLICATED shape,
`[user_id][name][expiry][token_hash]`, where a client sends
`[name][expiry]` and the server resolves the acting user from the session
and mints the token and its hash itself in `maybe_rewrite_pat_request`.
A client cannot produce the replicated shape, since it does not know the
hash; the previous code filled it with a stub. That went unnoticed because
the raw path has no dispatch layer and therefore no rewrite, so a request
submitted there must already be replicated and the wrong shape was the
right one.
Both remain reachable, chosen by which path the client took rather than by
configuration, mirroring the register/login split that already exists for
sessions. Through the shell the tokens now commit.
---
core/simulator/src/client.rs | 54 ++++++++++++++++++++++++++++++++-----
core/simulator/src/lib.rs | 63 ++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 111 insertions(+), 6 deletions(-)
diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs
index 88e0d9d08..f70bd5a48 100644
--- a/core/simulator/src/client.rs
+++ b/core/simulator/src/client.rs
@@ -31,6 +31,10 @@ use iggy_binary_protocol::requests::messages::{
use iggy_binary_protocol::requests::partitions::{
CreatePartitionsRequest, DeletePartitionsRequest,
};
+use iggy_binary_protocol::requests::personal_access_tokens::{
+ CreatePersonalAccessTokenRequest as WireCreatePersonalAccessTokenRequest,
+ DeletePersonalAccessTokenRequest as WireDeletePersonalAccessTokenRequest,
+};
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
use iggy_binary_protocol::requests::streams::{
CreateStreamRequest, DeleteStreamRequest, PurgeStreamRequest, UpdateStreamRequest,
@@ -79,6 +83,21 @@ pub struct SimClient {
/// the body a pure function of the seed. See [`SimClient::next_message_id`].
message_counter: Cell<u64>,
session: Cell<u64>,
+ /// Whether this client talks to the server's real dispatch layer.
+ ///
+ /// It changes what a PAT request must contain. A real client sends
+ /// `[name][expiry]` and the server mints the token and its hash in
+ /// `maybe_rewrite_pat_request`, rewriting the request into the replicated
+ /// form before it reaches consensus. The simulator's raw path has no dispatch
+ /// layer and therefore no rewrite, so a request submitted there has to arrive
+ /// already in the replicated form.
+ ///
+ /// The same split already exists for sessions (`register` for the raw path,
+ /// `login` for the shell); this makes it explicit for the one op family whose
+ /// BODY differs rather than just its envelope. Set by
+ /// `Simulator::shell_login_via`, so it follows the path the client actually
+ /// took rather than being configured separately.
+ shell_wire: Cell<bool>,
}
impl SimClient {
@@ -90,9 +109,17 @@ impl SimClient {
partition_counter: Cell::new(0),
message_counter: Cell::new(0),
session: Cell::new(0),
+ shell_wire: Cell::new(false),
}
}
+ /// Mark this client as talking to the real dispatch layer, so PAT requests
+ /// carry the client wire shape rather than the replicated one. See
+ /// [`SimClient::shell_wire`].
+ pub fn set_shell_wire(&self) {
+ self.shell_wire.set(true);
+ }
+
#[must_use]
pub const fn client_id(&self) -> u128 {
self.client_id
@@ -503,14 +530,23 @@ impl SimClient {
name: &str,
expiry: u64,
) -> Message<RoutedRequestHeader> {
+ let name = WireName::new(name).expect("PAT name must be valid");
+ // Through dispatch, send what a real client sends: the server resolves
+ // the acting user from the session and mints the token and its hash in
+ // `maybe_rewrite_pat_request`, rewriting this into the replicated form
+ // before consensus sees it. A client cannot produce that form itself --
+ // it does not know the hash -- so sending it here is what made every PAT
+ // request fail to decode as `InvalidCommand`.
+ if self.shell_wire.get() {
+ let wire = WireCreatePersonalAccessTokenRequest { name, expiry };
+ return self.build_request(Operation::CreatePersonalAccessToken, &wire.to_bytes());
+ }
+ // Raw path: no dispatch layer, so no rewrite ever happens and the request
+ // has to arrive already replicated.
let wire = CreatePersonalAccessTokenRequest {
user_id: 0,
- name: WireName::new(name).expect("PAT name must be valid"),
+ name,
expiry,
- // Deterministic stub for the simulator. Production servers mint
- // this in `maybe_rewrite_pat_request` on the primary; the
- // simulator drives the wire path directly without that rewrite
- // step.
token_hash: [b'a'; 64],
};
self.build_request(Operation::CreatePersonalAccessToken, &wire.to_bytes())
@@ -519,9 +555,15 @@ impl SimClient {
/// # Panics
/// Panics if `name` is not a valid `WireName`.
pub fn delete_personal_access_token(&self, name: &str) -> Message<RoutedRequestHeader> {
+ let name = WireName::new(name).expect("PAT name must be valid");
+ // See `create_personal_access_token` for why the shape depends on the path.
+ if self.shell_wire.get() {
+ let wire = WireDeletePersonalAccessTokenRequest { name };
+ return self.build_request(Operation::DeletePersonalAccessToken, &wire.to_bytes());
+ }
let wire = DeletePersonalAccessTokenRequest {
user_id: 0,
- name: WireName::new(name).expect("PAT name must be valid"),
+ name,
only_if_expired: false,
};
self.build_request(Operation::DeletePersonalAccessToken, &wire.to_bytes())
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 27135d4b0..a266dc65d 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -573,6 +573,9 @@ impl Simulator {
));
}
+ // From here this client talks the real client protocol; see
+ // `SimClient::shell_wire`.
+ client.set_shell_wire();
let msg = client
.login(replica::SHELL_ROOT_USERNAME, replica::SHELL_ROOT_PASSWORD)
.into_generic();
@@ -3596,6 +3599,66 @@ mod tests {
// Wiring a snapshot coordinator is what closes that gap.
}
+ /// Personal-access-token requests commit through the dispatch layer.
+ ///
+ /// They used to be refused there with `InvalidCommand` on every attempt,
+ /// because `SimClient` sent the REPLICATED shape
+ /// (`[user_id][name][expiry][token_hash]`) rather than the client one
+ /// (`[name][expiry]`). A client cannot produce the replicated shape: it does
+ /// not know the token hash, which the server mints in
+ /// `maybe_rewrite_pat_request` after resolving the acting user from the
+ /// session. The raw path has no dispatch layer and so no rewrite, which is
+ /// why the wrong shape worked there and hid the bug.
+ ///
+ /// Asserts a commit rather than merely a reply: a denial is also a reply, and
+ /// it was the denials that went unnoticed for so long.
+ #[test]
+ fn personal_access_tokens_commit_through_the_dispatch_shell() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0x9A7_0001,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::with_shards_shell(
+ usize::from(replica_count),
+ 1,
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+ sim.shell_login(&client);
+
+ let create = client.create_personal_access_token("wl-pat-token", 0);
+ let request = create.header().request;
+ sim.submit_request(client_id, 0, create.into_generic());
+ let mut status = None;
+ for _ in 0..400 {
+ if let Some(reply) = sim
+ .step()
+ .into_iter()
+ .find(|reply| reply.header().request == request)
+ {
+ status = Some(reply.header().status);
+ break;
+ }
+ }
+ assert_eq!(
+ status,
+ Some(0),
+ "creating a personal access token through dispatch was refused; a \
+ nonzero status here means the request never reached consensus"
+ );
+ }
+
/// A client that dials a BACKUP still gets a working session, and the login
/// travels as a forwarded consensus proposal rather than a redirect.
///
(iggy) 01/18: fix(simulator): draw the wire consumer kind, not a bare boolean
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 46df538e1d4ae4eec4a2d903297e9dce6a4aaca2
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 14:28:55 2026 +0530
fix(simulator): draw the wire consumer kind, not a bare boolean
The four consumer-offset ops sampled `consumer_kind` as `u8::from(bool)`,
so half of every request they emitted carried kind 0. `WireConsumer`
recognises only 1 (consumer) and 2 (consumer group), so those requests
failed to decode; the partition plane answers an unparseable
consumer-offset request by logging a WARN and dropping the frame without
a reply.
A client blocked on that request never hears back, and with one
in-flight slot per client the first malformed draw wedged the workload
for the rest of the run. `ActionWeights::default` carries
`StoreConsumerOffset2` at weight 25, so every default-weight workload
hit it within a handful of requests; a 3000-tick run drained 6 replies
instead of ~270. It went unnoticed because the fuzzer hardcoded
`SendMessages: 100` and the in-tree workload tests only assert that some
reply arrived, never that the run drains.
`KIND_CONSUMER` becomes public alongside the sibling it is always
compared against, so a synthesized consumer names the discriminant
rather than guessing a small integer.
Still one bool draw per sample, so the PRNG trace shape is unchanged and
only the encoded bytes move. The determinism baseline is re-locked: the
old value was recorded over a trace that wedged a few replies in.
---
core/binary_protocol/src/lib.rs | 2 +-
core/binary_protocol/src/primitives/consumer.rs | 6 +++++-
core/simulator/src/lib.rs | 9 ++++++++-
.../src/workload/ops/delete_consumer_offset.rs | 3 ++-
.../src/workload/ops/delete_consumer_offset_2.rs | 3 ++-
core/simulator/src/workload/ops/mod.rs | 21 ++++++++++++++++++++-
.../src/workload/ops/store_consumer_offset.rs | 3 ++-
.../src/workload/ops/store_consumer_offset_2.rs | 3 ++-
8 files changed, 42 insertions(+), 8 deletions(-)
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index 0825d0f44..4ed40c33e 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -88,7 +88,7 @@ pub use message_view::{
WireMessageIterator, WireMessageIteratorMut, WireMessageView, WireMessageViewMut,
};
pub use primitives::ack_level::AckLevel;
-pub use primitives::consumer::{KIND_CONSUMER_GROUP, WireConsumer};
+pub use primitives::consumer::{KIND_CONSUMER, KIND_CONSUMER_GROUP, WireConsumer};
pub use primitives::identifier::{MAX_WIRE_NAME_LENGTH, WireIdentifier, WireName};
pub use primitives::partition_assignment::CreatedPartitionAssignment;
pub use primitives::partitioning::{MAX_MESSAGES_KEY_LENGTH, WirePartitioning};
diff --git a/core/binary_protocol/src/primitives/consumer.rs b/core/binary_protocol/src/primitives/consumer.rs
index 202de84f1..1e7739b37 100644
--- a/core/binary_protocol/src/primitives/consumer.rs
+++ b/core/binary_protocol/src/primitives/consumer.rs
@@ -20,7 +20,11 @@ use crate::WireIdentifier;
use crate::codec::{WireDecode, WireEncode, read_u8};
use bytes::{BufMut, BytesMut};
-const KIND_CONSUMER: u8 = 1;
+/// Wire discriminant for a single consumer (vs a `ConsumerGroup`). Public for
+/// the same reason as its sibling below, plus one more: `decode` accepts only
+/// these two values, so anything synthesizing a consumer needs to name them
+/// rather than guess a small integer.
+pub const KIND_CONSUMER: u8 = 1;
/// Wire discriminant for a consumer-group consumer (vs a single `Consumer`).
/// Public so the server dispatch can match on it by name instead of a raw `2`.
pub const KIND_CONSUMER_GROUP: u8 = 2;
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 70f24b7d3..2aafb34ee 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -1675,8 +1675,15 @@ mod tests {
// Re-locked again when replies stopped echoing a group id (the
// client wire lost its namespace field): the reply-hash tuple
// dropped that component.
+ // Re-locked again when the consumer-offset ops began drawing the WIRE
+ // consumer kind (1 / 2) instead of a bare boolean (0 / 1). Kind 0 is
+ // not a `WireConsumer` discriminant, so every such request used to be
+ // dropped unparsed with no reply, wedging the client's only in-flight
+ // slot; the previous baseline was recorded over a trace that stalled a
+ // few replies in. Same single bool draw, so only the encoded bytes and
+ // the replies they now earn moved. See `ops::sample_consumer_kind`.
assert_eq!(
- h1, 0xCF1F_BC79_B44A_65F7,
+ h1, 0x25C7_0F6A_5D0F_A8B2,
"workload reply hash drifted from locked baseline"
);
}
diff --git a/core/simulator/src/workload/ops/delete_consumer_offset.rs b/core/simulator/src/workload/ops/delete_consumer_offset.rs
index c05c7960a..aac37c71b 100644
--- a/core/simulator/src/workload/ops/delete_consumer_offset.rs
+++ b/core/simulator/src/workload/ops/delete_consumer_offset.rs
@@ -25,6 +25,7 @@ use server_common::sharding::IggyNamespace;
use crate::client::SimClient;
use crate::workload::effect::Effect;
+use crate::workload::ops::sample_consumer_kind;
use crate::workload::options::WorkloadOptions;
use crate::workload::shadow::Shadow;
@@ -51,7 +52,7 @@ pub fn sample(
match outcome {
Outcome::Success => {
let ns = shadow.pick_namespace(prng)?;
- let consumer_kind: u8 = u8::from(prng.random::<bool>());
+ let consumer_kind = sample_consumer_kind(prng);
let consumer_id: u32 = prng.random_range(0..options.consumer_pool_size.max(1));
Some(Input {
ns,
diff --git a/core/simulator/src/workload/ops/delete_consumer_offset_2.rs b/core/simulator/src/workload/ops/delete_consumer_offset_2.rs
index ad2a6201a..8cd936145 100644
--- a/core/simulator/src/workload/ops/delete_consumer_offset_2.rs
+++ b/core/simulator/src/workload/ops/delete_consumer_offset_2.rs
@@ -25,6 +25,7 @@ use server_common::sharding::IggyNamespace;
use crate::client::SimClient;
use crate::workload::effect::Effect;
+use crate::workload::ops::sample_consumer_kind;
use crate::workload::options::WorkloadOptions;
use crate::workload::shadow::Shadow;
@@ -52,7 +53,7 @@ pub fn sample(
match outcome {
Outcome::Success => {
let ns = shadow.pick_namespace(prng)?;
- let consumer_kind: u8 = u8::from(prng.random::<bool>());
+ let consumer_kind = sample_consumer_kind(prng);
let consumer_id: u32 = prng.random_range(0..options.consumer_pool_size.max(1));
let f: f32 = prng.random();
let ack = if f < options.ack_quorum_ratio {
diff --git a/core/simulator/src/workload/ops/mod.rs b/core/simulator/src/workload/ops/mod.rs
index df6ffed48..d74d5b0fa 100644
--- a/core/simulator/src/workload/ops/mod.rs
+++ b/core/simulator/src/workload/ops/mod.rs
@@ -55,7 +55,8 @@ pub mod update_stream;
pub mod update_topic;
pub mod update_user;
-use iggy_binary_protocol::RoutedRequestHeader;
+use iggy_binary_protocol::{KIND_CONSUMER, KIND_CONSUMER_GROUP, RoutedRequestHeader};
+use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -65,6 +66,24 @@ use crate::workload::effect::Effect;
use crate::workload::options::WorkloadOptions;
use crate::workload::shadow::Shadow;
+/// Draw a consumer kind for the four consumer-offset ops, as the WIRE
+/// discriminant rather than a bare boolean.
+///
+/// `WireConsumer::decode` accepts only [`KIND_CONSUMER`] (1) and
+/// [`KIND_CONSUMER_GROUP`] (2); anything else is an `UnknownDiscriminant`,
+/// which `parse_consumer_offset_request` maps to `IggyError::InvalidCommand`
+/// and the partition plane answers by logging a WARN and dropping the frame
+/// with NO reply. A client blocked on that request never gets one back, so a
+/// single malformed draw wedges its in-flight slot for the rest of the run.
+/// One bool draw either way, so the PRNG trace shape is unchanged.
+pub(crate) fn sample_consumer_kind(prng: &mut Xoshiro256Plus) -> u8 {
+ if prng.random::<bool>() {
+ KIND_CONSUMER_GROUP
+ } else {
+ KIND_CONSUMER
+ }
+}
+
/// Generates per-op enums (`InFlightInput`, `InFlightOutcome`) plus four
/// dispatch fns over a fixed `(Action, module)` table. Missing variants
/// are a compile error via the exhaustive `match` arms.
diff --git a/core/simulator/src/workload/ops/store_consumer_offset.rs b/core/simulator/src/workload/ops/store_consumer_offset.rs
index edefbfc72..2a9795b7d 100644
--- a/core/simulator/src/workload/ops/store_consumer_offset.rs
+++ b/core/simulator/src/workload/ops/store_consumer_offset.rs
@@ -26,6 +26,7 @@ use server_common::sharding::IggyNamespace;
use crate::client::SimClient;
use crate::workload::effect::Effect;
+use crate::workload::ops::sample_consumer_kind;
use crate::workload::options::WorkloadOptions;
use crate::workload::shadow::Shadow;
@@ -53,7 +54,7 @@ pub fn sample(
match outcome {
Outcome::Success => {
let ns = shadow.pick_namespace(prng)?;
- let consumer_kind: u8 = u8::from(prng.random::<bool>());
+ let consumer_kind = sample_consumer_kind(prng);
let consumer_id: u32 = prng.random_range(0..options.consumer_pool_size.max(1));
// Draw against the configured ceiling, then clamp to committed
// reality so the offset is reachable. Clamping post-draw keeps
diff --git a/core/simulator/src/workload/ops/store_consumer_offset_2.rs b/core/simulator/src/workload/ops/store_consumer_offset_2.rs
index baac692d0..e745d8ced 100644
--- a/core/simulator/src/workload/ops/store_consumer_offset_2.rs
+++ b/core/simulator/src/workload/ops/store_consumer_offset_2.rs
@@ -31,6 +31,7 @@ use server_common::sharding::IggyNamespace;
use crate::client::SimClient;
use crate::workload::effect::Effect;
+use crate::workload::ops::sample_consumer_kind;
use crate::workload::options::WorkloadOptions;
use crate::workload::shadow::Shadow;
@@ -59,7 +60,7 @@ pub fn sample(
match outcome {
Outcome::Success => {
let ns = shadow.pick_namespace(prng)?;
- let consumer_kind: u8 = u8::from(prng.random::<bool>());
+ let consumer_kind = sample_consumer_kind(prng);
let consumer_id: u32 = prng.random_range(0..options.consumer_pool_size.max(1));
// Draw against the configured ceiling, then clamp to committed
// reality so the offset is reachable. Clamping post-draw keeps
(iggy) 04/18: feat(simulator): resend timed-out client requests
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 72e58db678d348477516d46b621471173a0e2a42
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 14:40:41 2026 +0530
feat(simulator): resend timed-out client requests
A client held one in-flight slot and nothing ever timed out, so the first
dropped request or reply stranded that slot for the rest of the run. Any
packet loss therefore wedged the workload rather than exercising
recovery: at 2% loss a 5000-tick run drained four replies where a perfect
network drained 439. It also forced the crash injector to spare
primaries, since a request lost to a dead primary could never be
answered.
The workload now retains each outstanding request and resubmits it once
the reply is overdue, rotating to the next replica so a client whose
primary died finds the new one. The retained message is the original
byte-for-byte, because the request id is what the metadata client table
dedups on; rebuilding it would draw a fresh id and commit a second time
instead of replaying the cached reply. Outstanding requests live in a
BTreeMap, as the resulting submit order is observable and hash order
would break replay.
The two setup handshakes had the same defect and gave up after 100 and
200 steps; both now resubmit on the same reasoning, which is what let
seed 1 fail during registration before the workload even started.
A failed drain becomes a hard failure carrying which requests are
outstanding, how many attempts each has had, and what every live replica
believes about view, commit offset and primary. It was a warning only
because a stall was previously expected and unactionable.
---
core/simulator/src/bin/workload-fuzz.rs | 25 +++--
core/simulator/src/lib.rs | 155 +++++++++++++++++++++++------
core/simulator/src/workload/mod.rs | 171 +++++++++++++++++++++++++++-----
core/simulator/src/workload/options.rs | 17 ++++
core/simulator/src/workload/oracle.rs | 53 +++++++++-
5 files changed, 350 insertions(+), 71 deletions(-)
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index 52fe0eeae..4ac434628 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -401,24 +401,29 @@ fn main() {
);
if quiesce {
- if oracle::drive_to_quiesce(&mut sim, &mut workload, 50_000) {
- oracle::assert_converged(&sim, &workload);
- println!("quiesced and converged (leader-relative + entity oracle)");
- } else {
- println!(
- "WARN: did not quiesce within budget — expected when crashing to bare quorum; \
- per-tick invariants still held"
- );
- }
+ // A failed drain is a hard failure, not a warning. It used to be one
+ // because a lost request could not be retried, so a stall was expected
+ // and unactionable; with the client resending, a request that never gets
+ // answered inside the budget is either a wedge or a liveness bug, and
+ // the report says which replicas were live and what they believed.
+ assert!(
+ oracle::drive_to_quiesce(&mut sim, &mut workload, 50_000),
+ "{}",
+ oracle::quiesce_failure_report(&sim, &workload),
+ );
+ oracle::assert_converged(&sim, &workload);
+ println!("quiesced and converged (leader-relative + entity oracle)");
}
let stats = workload.auditor.stats();
println!(
- "coverage: replies_seen={} replies_unknown={} committed_rejections={} samples_none={}",
+ "coverage: replies_seen={} replies_unknown={} committed_rejections={} \
+ samples_none={} resends={}",
stats.replies_seen,
stats.replies_unknown,
stats.committed_rejections,
workload.samples_none(),
+ workload.resends(),
);
for action in Action::iter() {
let commits = stats.commits(action);
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 2aafb34ee..3b8d621ae 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -67,6 +67,18 @@ const POLL_BUDGET: u32 = 100_000;
/// keeps those draws from perturbing network or workload traces.
pub const ENTRY_SHARD_SEED_SALT: u64 = 0x5A1A_F0E5_FACE_0003;
+/// Steps a setup handshake (`register_client_with_primary`, `shell_login`) waits
+/// before resubmitting its request, and the total it will spend.
+///
+/// These helpers used to submit once and give up, which is sound only on a
+/// perfect network: under injected packet loss the request or its reply is
+/// eventually dropped and the fixture fails before the workload ever starts.
+/// Both requests are metadata-plane ops carrying a stable request id, so the
+/// client table dedups a resend and replays the cached reply rather than
+/// committing twice.
+const SETUP_RETRY_STEPS: u32 = 50;
+const SETUP_TOTAL_STEPS: u32 = 4_000;
+
/// One simulated replica: its shards plus the executor bookkeeping needed
/// to crash it. One entry per shard in `shards`/`pump_tasks` (a single
/// shard until multi-shard lands).
@@ -412,7 +424,8 @@ impl Simulator {
/// (replica 0), as [`Self::register_client_with_primary`] does.
///
/// # Panics
- /// If no login reply arrives within 200 steps or it carries no session.
+ /// If no login reply arrives within [`SETUP_TOTAL_STEPS`], or it carries no
+ /// session.
pub fn shell_login(&mut self, client: &SimClient) {
// Register the client's connection metadata on every replica, as
// `install_client_fd` does in production. `ensure_transport_connection`
@@ -427,24 +440,46 @@ impl Simulator {
));
}
- let msg = client.login(replica::SHELL_ROOT_USERNAME, replica::SHELL_ROOT_PASSWORD);
- self.submit_request(client.client_id(), 0, msg.into_generic());
- let mut session = 0u64;
- let mut got_reply = false;
- for _ in 0..200 {
- if let Some(reply) = self.step().first() {
- // The login reply carries the assigned session in `op`
- // (`build_reply_with_body` maps the session field to `op`).
- session = reply.header().op;
- got_reply = true;
- break;
- }
- }
- assert!(got_reply, "shell_login: no login reply within 200 steps");
+ let msg = client
+ .login(replica::SHELL_ROOT_USERNAME, replica::SHELL_ROOT_PASSWORD)
+ .into_generic();
+ // The login reply carries the assigned session in `op`
+ // (`build_reply_with_body` maps the session field to `op`).
+ let session = self
+ .await_setup_reply(client.client_id(), 0, &msg, "shell_login")
+ .map_or(0, |reply| reply.header().op);
assert!(session > 0, "shell_login: login reply carried no session");
client.bind_session(session);
}
+ /// Submit `message` to `target` and step until a client reply arrives,
+ /// resubmitting every [`SETUP_RETRY_STEPS`] steps.
+ ///
+ /// Returns the first reply, or `None` once [`SETUP_TOTAL_STEPS`] is spent.
+ /// The same message is resubmitted verbatim, so the request id is stable and
+ /// the metadata client table treats a retry as a duplicate.
+ fn await_setup_reply(
+ &mut self,
+ client_id: u128,
+ target: u8,
+ message: &Message<GenericHeader>,
+ label: &str,
+ ) -> Option<Message<ReplyHeader>> {
+ for step in 0..SETUP_TOTAL_STEPS {
+ if step % SETUP_RETRY_STEPS == 0 {
+ self.submit_request(client_id, target, message.deep_copy());
+ }
+ if let Some(reply) = self.step().into_iter().next() {
+ return Some(reply);
+ }
+ }
+ panic!(
+ "{label}: no reply for client {client_id} within {SETUP_TOTAL_STEPS} steps \
+ (seed {:#x})",
+ self.seed,
+ );
+ }
+
/// Advance the simulation by one tick. Returns client replies delivered.
///
/// Every shard runs its real message pump as an executor task, so a
@@ -633,17 +668,14 @@ impl Simulator {
/// `SimClient`.
///
/// # Panics
- /// If no reply arrives within 100 steps.
+ /// If no reply arrives within [`SETUP_TOTAL_STEPS`].
#[allow(clippy::cast_possible_truncation)]
pub fn register_client_with_primary(&mut self, client: &SimClient) {
- let msg = client.register();
- self.submit_request(client.client_id(), 0, msg.into_generic());
- let mut session = 0u64;
- let mut got_reply = false;
- for _ in 0..100 {
- let replies = self.step();
- if !replies.is_empty() {
- let header = replies[0].header();
+ let msg = client.register().into_generic();
+ let session = self
+ .await_setup_reply(client.client_id(), 0, &msg, "register_client_with_primary")
+ .map_or(0, |reply| {
+ let header = reply.header();
debug_assert_eq!(
header.operation,
iggy_binary_protocol::Operation::Register,
@@ -657,15 +689,8 @@ impl Simulator {
client.client_id(),
header.client,
);
- session = header.commit;
- got_reply = true;
- break;
- }
- }
- assert!(
- got_reply,
- "register_client_with_primary: no reply within 100 steps"
- );
+ header.commit
+ });
client.bind_session(session);
// Partition has no `client_table`: at-least-once, no per-client
@@ -2013,6 +2038,70 @@ mod tests {
oracle::assert_converged(&sim, &wl);
}
+ /// A lossy network drains, because the client resends.
+ ///
+ /// Without [`workload::Workload::due_resends`] this wedges immediately and
+ /// permanently: a client holds one in-flight slot, nothing times out, so the
+ /// first dropped request or reply strands that slot for the rest of the run.
+ /// At 5% loss a 3000-tick run used to drain a handful of replies and then
+ /// stop, and `drive_to_quiesce` could never finish because the reply it
+ /// waited on had already been discarded by the network.
+ ///
+ /// Asserts the resend path actually ran rather than the seed getting lucky.
+ #[test]
+ fn packet_loss_resends_and_drains() {
+ use crate::workload::{
+ self, Workload,
+ options::{ActionWeights, WorkloadOptions},
+ oracle,
+ };
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let seed = 0x105_5A1A;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed,
+ packet_loss_probability: 0.05,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = client::SimClient::new(client_id);
+ let ns_a = server_common::sharding::IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns_a);
+ sim.register_client_with_primary(&client);
+
+ let mut options = WorkloadOptions::new(seed, replica_count, vec![ns_a]);
+ options.weights = ActionWeights::partition_only();
+ let mut wl = Workload::new(options);
+
+ let clients = [client];
+ let replies = workload::run(&mut sim, &mut wl, &clients, 3_000, u64::MAX);
+ assert!(replies > 0, "lossy workload produced no replies");
+ assert!(
+ wl.resends() > 0,
+ "no request timed out at 5% packet loss, so the resend path never ran; \
+ raise the loss rate or lower request_timeout_ticks"
+ );
+
+ assert!(
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 20_000),
+ "{}",
+ oracle::quiesce_failure_report(&sim, &wl),
+ );
+ oracle::assert_converged(&sim, &wl);
+ }
+
/// Drive Create-heavy then Delete-heavy workload; assert shadow tracks
/// live streams:
///
diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs
index 2e641066c..4ac4735da 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -49,22 +49,49 @@ use rand_xoshiro::Xoshiro256Plus;
use rand_xoshiro::rand_core::SeedableRng;
use server_common::Message;
use shadow::Shadow;
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashSet};
/// Max in-flight requests per client. Must stay under the consensus
/// pipeline's queue limits.
pub const CLIENT_REQUEST_QUEUE_MAX: usize = 1;
+/// An outstanding request, retained so the client can resend it.
+///
+/// The encoded message is kept verbatim rather than rebuilt from the sampled
+/// `Input`, because rebuilding would draw a fresh request id from the client and
+/// a resend must reuse the original: that id is what the metadata plane's client
+/// table dedups on, so a renumbered retry commits a second time instead of
+/// returning the cached reply.
+struct Outstanding {
+ message: Message<RoutedRequestHeader>,
+ /// Replica the most recent attempt went to. A resend moves to the next one,
+ /// so a client whose primary died eventually finds the new one.
+ target: u8,
+ /// Tick of the most recent attempt, not of the first.
+ attempted_tick: u64,
+ attempts: u32,
+}
+
pub struct Workload {
prng: Xoshiro256Plus,
pub auditor: ServerAuditor,
pub shadow: Shadow,
pub options: WorkloadOptions,
- /// Number of in-flight requests per client.
+ /// Outstanding requests keyed exactly as the auditor keys its expectations,
+ /// so the two are removed together.
+ ///
+ /// A `BTreeMap`, not a `HashMap`: [`Self::due_resends`] walks it and the
+ /// resulting submit order is observable, so hash iteration order would make
+ /// replay diverge from the seed.
///
/// TODO: reap on client disconnect; bounded today by the fixed
/// `Simulator::new` set.
- in_flight_per_client: HashMap<u128, usize>,
+ outstanding: BTreeMap<(u128, u64), Outstanding>,
+ /// Driver tick, advanced by [`Self::tick`]. A driver that never ticks never
+ /// resends, which is what the hand-written scenario tests rely on.
+ now: u64,
+ /// Total resends issued, for the run summary.
+ resends: u64,
/// Debug counter for `sample()` returning `None` (a targeted outcome whose
/// shadow precondition is unmet). Flags PRNG-trace drift during development.
samples_none: u64,
@@ -90,7 +117,9 @@ impl Workload {
auditor: ServerAuditor::new(),
shadow,
options,
- in_flight_per_client: HashMap::new(),
+ outstanding: BTreeMap::new(),
+ now: 0,
+ resends: 0,
samples_none: 0,
strict_outcome_oracle,
}
@@ -99,18 +128,83 @@ impl Workload {
/// True if the client has a free in-flight slot.
#[must_use]
pub fn client_idle(&self, client_id: u128) -> bool {
- self.in_flight_per_client
- .get(&client_id)
- .copied()
- .unwrap_or(0)
- < CLIENT_REQUEST_QUEUE_MAX
+ self.client_in_flight(client_id) < CLIENT_REQUEST_QUEUE_MAX
}
/// Total in-flight requests across all clients. Read by the
/// [`Invariants`]; draws no PRNG.
#[must_use]
pub(crate) fn total_in_flight(&self) -> usize {
- self.in_flight_per_client.values().copied().sum()
+ self.outstanding.len()
+ }
+
+ /// Advance the resend clock by one tick. Called once per driver iteration;
+ /// [`Self::due_resends`] measures against it.
+ pub fn tick(&mut self) {
+ self.now += 1;
+ }
+
+ /// Total resends issued so far.
+ #[must_use]
+ pub const fn resends(&self) -> u64 {
+ self.resends
+ }
+
+ /// Requests whose reply has not arrived within
+ /// [`WorkloadOptions::request_timeout_ticks`], each paired with the replica
+ /// to retry it against. Callers must submit every returned message.
+ ///
+ /// This is what a real client's read timeout does, and the harness needs it
+ /// for two reasons. A dropped request or reply otherwise strands the
+ /// client's only in-flight slot for the rest of the run, so any packet loss
+ /// wedges the workload. And a request lost to a crashed primary can only be
+ /// answered by the next one, which the client reaches by rotating its
+ /// target.
+ ///
+ /// Resending is safe on both planes but not equally cheap: the metadata
+ /// plane dedups on the retained request id and replays the cached reply,
+ /// while the partition plane is at-least-once and may commit the op twice.
+ /// The shadow already models that (`Effect` application is driven by what
+ /// committed, not by what was targeted).
+ #[must_use = "returned requests must be submitted or the client stays wedged"]
+ pub fn due_resends(&mut self) -> Vec<(u8, Message<RoutedRequestHeader>)> {
+ let timeout = self.options.request_timeout_ticks;
+ if timeout == 0 {
+ return Vec::new();
+ }
+ let replica_count = self.options.replica_count.max(1);
+ let now = self.now;
+ let mut due = Vec::new();
+ for entry in self.outstanding.values_mut() {
+ if now.saturating_sub(entry.attempted_tick) < timeout {
+ continue;
+ }
+ entry.target = (entry.target + 1) % replica_count;
+ entry.attempted_tick = now;
+ entry.attempts += 1;
+ due.push((entry.target, entry.message.deep_copy()));
+ }
+ self.resends += due.len() as u64;
+ due
+ }
+
+ /// Outstanding requests as `(client, request, target, attempts)`, in key
+ /// order. Diagnostic only: names what a run was still waiting on when it
+ /// failed to drain.
+ #[must_use]
+ pub(crate) fn outstanding_summary(&self) -> Vec<(u128, u64, u8, u32)> {
+ self.outstanding
+ .iter()
+ .map(|(&(client, request), entry)| (client, request, entry.target, entry.attempts))
+ .collect()
+ }
+
+ /// In-flight count for one client. Keys are `(client, request)`, so the
+ /// client's entries are one contiguous range.
+ fn client_in_flight(&self, client_id: u128) -> usize {
+ self.outstanding
+ .range((client_id, 0)..=(client_id, u64::MAX))
+ .count()
}
/// Aggregate in-flight ceiling: one queue's worth per declared client.
@@ -173,10 +267,15 @@ impl Workload {
request_namespace: header.group,
},
);
- *self
- .in_flight_per_client
- .entry(client.client_id())
- .or_insert(0) += 1;
+ self.outstanding.insert(
+ key,
+ Outstanding {
+ message: message.deep_copy(),
+ target,
+ attempted_tick: self.now,
+ attempts: 1,
+ },
+ );
Some((target, message))
}
@@ -200,7 +299,7 @@ impl Workload {
OnReply::Match(entry) => entry,
OnReply::NsMismatch => {
// Entry consumed; release slot, skip effects (misrouted).
- self.decrement_in_flight(header.client);
+ self.release_outstanding(key);
return Vec::new();
}
OnReply::Unknown => return Vec::new(),
@@ -283,24 +382,27 @@ impl Workload {
self.auditor.note_committed(entry.action);
}
- self.decrement_in_flight(header.client);
+ self.release_outstanding(key);
result.sim_commands
}
- /// Release one in-flight slot. Panics on underflow so a future
- /// double-decrement surfaces instead of being silently clamped.
+ /// Drop a request's retry entry, freeing the client's slot. Paired with the
+ /// auditor consuming its expectation for the same key, so the two never
+ /// disagree about what is outstanding.
///
/// # Panics
- /// Panics if no entry exists for `client`, or if the counter is 0.
- fn decrement_in_flight(&mut self, client: u128) {
- let count = self
- .in_flight_per_client
- .get_mut(&client)
- .expect("decrement_in_flight: no entry for client; record_in_flight must precede");
- *count = count
- .checked_sub(1)
- .expect("in_flight underflow: per-client counter went below 0");
+ /// Panics if no entry exists for `key`; the auditor only reports a match or
+ /// a namespace mismatch for a key it was given, and `build_request` records
+ /// both sides together, so a miss here means the two drifted.
+ fn release_outstanding(&mut self, key: (u128, u64)) {
+ assert!(
+ self.outstanding.remove(&key).is_some(),
+ "no outstanding entry for (client={}, request={}); the auditor \
+ matched a key the retry buffer never recorded",
+ key.0,
+ key.1,
+ );
}
/// Debug counter for `sample()` returning `None`. Surfaces sampling
@@ -374,9 +476,13 @@ pub fn run(
let mut fault_prng = Xoshiro256Plus::seed_from_u64(workload.options.seed ^ FAULT_SEED_SALT);
let mut replies_seen = 0u64;
for _ in 0..tick_budget {
+ workload.tick();
if workload.options.crash_per_tick_ratio > 0.0 {
maybe_inject_crash(sim, workload, &mut fault_prng);
}
+ // Resend before sampling: a timed-out request still holds the client's
+ // slot, so `build_request` would decline it anyway.
+ resubmit_due(sim, workload);
for client in clients {
if let Some((target, msg)) = workload.build_request(client) {
sim.submit_request(client.client_id(), target, msg.into_generic());
@@ -439,6 +545,17 @@ fn maybe_inject_crash(sim: &mut Simulator, workload: &Workload, prng: &mut Xoshi
sim.replica_crash(victim);
}
+/// Submit every request whose reply is overdue (see [`Workload::due_resends`]).
+///
+/// The client id rides the retained message's header, so a resend re-enters the
+/// network exactly as the original did, only aimed at the next replica.
+pub fn resubmit_due(sim: &mut Simulator, workload: &mut Workload) {
+ for (target, message) in workload.due_resends() {
+ let client_id = message.header().client;
+ sim.submit_request(client_id, target, message.into_generic());
+ }
+}
+
/// Apply `SimCommand`s returned by [`Workload::on_reply`].
///
/// Callers must invoke this (or [`run`]) for every batch of returned
diff --git a/core/simulator/src/workload/options.rs b/core/simulator/src/workload/options.rs
index a4a29447a..d707b2961 100644
--- a/core/simulator/src/workload/options.rs
+++ b/core/simulator/src/workload/options.rs
@@ -19,6 +19,14 @@ use crate::workload::actions::Action;
use server_common::sharding::IggyNamespace;
use strum::EnumCount;
+/// Default [`WorkloadOptions::request_timeout_ticks`].
+///
+/// Four times the primary's commit-broadcast interval (50 ticks, see
+/// `oracle::QUIESCE_SETTLE_TICKS`), so a request whose commit is merely waiting
+/// on the next broadcast is never resent, while one lost to a dropped packet or
+/// a crashed primary is retried long before the run's budget runs out.
+pub const DEFAULT_REQUEST_TIMEOUT_TICKS: u64 = 200;
+
/// Per-action sampling weights as percentages. Unlisted variants default
/// to 0 (never picked). Listed weights must sum to 100.
#[derive(Debug, Clone, Copy)]
@@ -171,6 +179,14 @@ pub struct WorkloadOptions {
/// Floor on live replicas the driver will not crash below, preserving a
/// commit quorum. Defaults to `replica_count / 2 + 1`.
pub min_survivors: u8,
+ /// Ticks a request may stay outstanding before the client resends it.
+ ///
+ /// Must clear the primary's commit-broadcast interval with room to spare, or
+ /// a client resends work that was about to be answered and the run spends
+ /// its budget on duplicates. Must also stay well under the tick budget, or a
+ /// request lost to a dropped packet never gets retried and the client's slot
+ /// strands. `0` disables resending.
+ pub request_timeout_ticks: u64,
}
impl WorkloadOptions {
@@ -191,6 +207,7 @@ impl WorkloadOptions {
max_offset: 1_000_000,
crash_per_tick_ratio: 0.0,
min_survivors: replica_count / 2 + 1,
+ request_timeout_ticks: DEFAULT_REQUEST_TIMEOUT_TICKS,
}
}
}
diff --git a/core/simulator/src/workload/oracle.rs b/core/simulator/src/workload/oracle.rs
index b5a2b6ce0..4bdfc50f5 100644
--- a/core/simulator/src/workload/oracle.rs
+++ b/core/simulator/src/workload/oracle.rs
@@ -38,7 +38,7 @@
use crate::Simulator;
use crate::replica::Replica;
use crate::workload::shadow::Shadow;
-use crate::workload::{Workload, apply_sim_commands};
+use crate::workload::{Workload, apply_sim_commands, resubmit_due};
use consensus::{MetadataHandle, Status};
use metadata::impls::metadata::StreamsFrontend;
use std::collections::BTreeSet;
@@ -96,6 +96,11 @@ impl CommittedMetadata {
pub fn drive_to_quiesce(sim: &mut Simulator, workload: &mut Workload, max_ticks: u64) -> bool {
let mut drained = false;
for _ in 0..max_ticks {
+ // The drain keeps resending: a request lost on the way out is never
+ // answered, so without retries the drain would spend its whole budget
+ // waiting on a reply that cannot arrive.
+ workload.tick();
+ resubmit_due(sim, workload);
for reply in sim.step() {
let cmds = workload.on_reply(&reply);
apply_sim_commands(sim, &cmds);
@@ -117,6 +122,52 @@ pub fn drive_to_quiesce(sim: &mut Simulator, workload: &mut Workload, max_ticks:
true
}
+/// Why the run did not drain, as a multi-line report.
+///
+/// A failed drain is either a wedge or a cluster that is merely slow, and the
+/// bare boolean [`drive_to_quiesce`] returns cannot tell them apart. Once
+/// crashes, restarts and packet loss are all in play, that distinction is the
+/// whole diagnosis, so name what is still outstanding and what every live
+/// replica thinks the world looks like.
+#[must_use]
+pub fn quiesce_failure_report(sim: &Simulator, workload: &Workload) -> String {
+ use std::fmt::Write;
+
+ let mut report = format!(
+ "did not drain: {} request(s) still outstanding (seed={:#x})\n",
+ workload.total_in_flight(),
+ workload.options.seed,
+ );
+ for (client, request, target, attempts) in workload.outstanding_summary() {
+ let _ = writeln!(
+ report,
+ " outstanding client={client} request={request} \
+ last_target=replica {target} attempts={attempts}",
+ );
+ }
+ let _ = writeln!(report, " resends issued: {}", workload.resends());
+ for replica_idx in 0..sim.replica_count {
+ if sim.is_crashed(replica_idx) {
+ let _ = writeln!(report, " replica {replica_idx}: CRASHED");
+ continue;
+ }
+ let _ = write!(report, " replica {replica_idx}: live");
+ for &ns in &workload.options.namespaces {
+ let view = sim.consensus_view(usize::from(replica_idx), ns);
+ let commit = sim
+ .offsets(usize::from(replica_idx), ns)
+ .map(|offsets| offsets.commit_offset);
+ let primary = sim.primary_index(ns);
+ let _ = write!(
+ report,
+ " | ns {ns:?} view={view:?} commit_offset={commit:?} primary={primary:?}",
+ );
+ }
+ report.push('\n');
+ }
+ report
+}
+
/// Post-drain consensus checks that hold today.
///
/// Asserts no live replica is ahead of the leader, and (on a serial run) that
(iggy) 05/18: feat(simulator): inject replica restarts with stability windows
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 43d293c986dd308a63fdc2a1fe20d97dbfb4a5f8
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 14:56:32 2026 +0530
feat(simulator): inject replica restarts with stability windows
Crash injection was permanent: a crashed replica never came back, so
nothing in the harness ever exercised rejoin, the view probe, or log
repair. The reason was the same missing resend path that forced primaries
to be spared, and that landed in the previous commit.
Crash and restart now run through one FaultInjector owning the fault
PRNG, with TigerBeetle-style stability windows: a crash lasts long enough
for the surviving primary to commit past the victim's log, so its rejoin
has something to repair, and a rejoined replica runs long enough to catch
up before becoming a candidate again. Restart is considered before crash
so a tick never both revives and kills. `spare_primary` becomes a knob
rather than an unconditional rule, since crashing the primary under live
traffic is what puts a view change on the critical path.
Restart injection is metadata-plane only for now. The harness retains the
metadata WAL and the superblocks across a restart but not partition data,
so a restarted replica reports commit_offset 0 and the monotonicity
invariant trips on a harness limitation rather than a real regression.
The invariant doc now says so instead of claiming both quantities survive
a restart. Retaining the partition journal per namespace is the faithful
fix, since a real server's segments are on disk.
Two tests pin the rejoin paths that do work: a lost PrepareOk recovers by
retransmit, and a prepare journaled but unacked commits after its backups
restart. Both are worth stating because the backup's admission check
drops a prepare it already holds, so neither outcome follows from reading
that check alone.
The fuzzer finds a metadata-plane wedge these do not explain
(--seed 42 --replicas 5 --plane metadata --crash-prob 0.01
--restart-prob 0.02): two live backups hold op 45, the primary sits at
commit 44, and both log the gap drop for the whole drain while the
client's request goes unanswered across 276 attempts. Neither reduction
above reproduces it, so it stays an open finding rather than a test.
---
core/simulator/src/bin/workload-fuzz.rs | 66 ++++++---
core/simulator/src/lib.rs | 221 ++++++++++++++++++++++++++++++
core/simulator/src/workload/auditor.rs | 8 ++
core/simulator/src/workload/invariants.rs | 16 ++-
core/simulator/src/workload/mod.rs | 221 +++++++++++++++++++++++-------
core/simulator/src/workload/options.rs | 39 +++++-
core/simulator/src/workload/oracle.rs | 44 +++++-
7 files changed, 542 insertions(+), 73 deletions(-)
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index 4ac434628..c7abfecc5 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -17,15 +17,17 @@
//! Deterministic workload fuzzer for the Iggy simulator.
//!
-//! Drives [`simulator::workload::run`] (per-tick invariants + optional crash
-//! injection) for a number of ticks, then optionally quiesces and asserts the
-//! Phase C consensus checks. Everything is a function of `--seed`, logged at
-//! start and on panic so any failure replays with `--seed <value>`.
+//! Drives [`simulator::workload::run_with_faults`] (per-tick invariants plus
+//! crash, restart and network fault injection) for a number of ticks, then
+//! optionally quiesces and asserts the Phase C consensus checks. Everything is a
+//! function of `--seed`, logged at start and on panic so any failure replays
+//! with `--seed <value>`.
//!
//! ```text
//! workload-fuzz [--seed N] [--ticks N] [--clients N] [--replicas N]
//! [--plane partition|metadata|mixed|uniform]
-//! [--faults none|light|heavy] [--crash-prob F] [--no-quiesce]
+//! [--faults none|light|heavy] [--no-quiesce]
+//! [--crash-prob F] [--restart-prob F] [--crash-primary]
//! [network overrides: --packet-loss, --replay, --partition-mode,
//! --partition-prob, --unpartition-prob, --clog-prob, ...]
//! ```
@@ -48,7 +50,7 @@ use simulator::client::SimClient;
use simulator::packet::{PacketSimulatorOptions, PartitionMode, PartitionSymmetry};
use simulator::workload::actions::Action;
use simulator::workload::options::{ActionWeights, WorkloadOptions};
-use simulator::workload::{Workload, oracle, run};
+use simulator::workload::{FaultInjector, Workload, oracle, run_with_faults};
use strum::IntoEnumIterator;
#[derive(Parser)]
@@ -70,8 +72,16 @@ struct Args {
/// `NoAck`. `1.0` keeps every offset op on the replicated path.
#[arg(long, default_value_t = 0.5, value_parser = parse_unit_interval)]
ack_quorum_ratio: f32,
+ /// Per-tick chance one eligible replica is crashed.
#[arg(long, default_value_t = 0.0, value_parser = parse_unit_interval)]
crash_prob: f32,
+ /// Per-tick chance one crashed replica is restarted. Without this a crash
+ /// is permanent and nothing exercises rejoin or log repair.
+ #[arg(long, default_value_t = 0.0, value_parser = parse_unit_interval)]
+ restart_prob: f32,
+ /// Crash the primary too, putting a view change under live traffic.
+ #[arg(long)]
+ crash_primary: bool,
#[arg(long)]
no_quiesce: bool,
@@ -123,12 +133,11 @@ struct Args {
/// Named network fault profile, in the spirit of TigerBeetle's VOPR modes: one
/// flag for "how hostile is the network", rather than eleven.
///
-/// Anything but [`Faults::None`] currently stalls the run, and not because the
-/// cluster fails to make progress: `SimClient` has no request timeout, so a
-/// client holds its single in-flight slot forever once the request or its reply
-/// is dropped. Every profile below is therefore write-once-read-later until the
-/// client grows a resend path; they are wired now so the fault space is
-/// described in one place rather than rediscovered later.
+/// Progress falls off steeply with severity, because every lost frame costs a
+/// resend timeout: on one namespace with one client, a 3-replica cluster drains
+/// roughly 440 replies in 5000 ticks on a perfect network, 240 under `light` and
+/// 40 under `heavy`. All three still drain and converge; budget ticks
+/// accordingly rather than reading a low reply count as a stall.
#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
enum Faults {
/// Perfect network. Delays only, no loss and no partitions.
@@ -390,16 +399,33 @@ fn main() {
let mut options = WorkloadOptions::new(seed, replicas, vec![ns]);
options.client_count = clients;
options.crash_per_tick_ratio = crash_prob;
+ options.restart_per_tick_ratio = args.restart_prob;
+ options.spare_primary = !args.crash_primary;
options.ack_quorum_ratio = args.ack_quorum_ratio;
options.weights = plane.weights();
let mut workload = Workload::new(options);
- let replies = run(&mut sim, &mut workload, &sim_clients, ticks, u64::MAX);
+ let mut injector = FaultInjector::new(seed, replicas);
+ let replies = run_with_faults(
+ &mut sim,
+ &mut workload,
+ &sim_clients,
+ ticks,
+ u64::MAX,
+ &mut injector,
+ );
println!(
- "ran {ticks} ticks; {replies} replies; crashed replicas: {}",
- sim.crashed.len()
+ "ran {ticks} ticks; {replies} replies; crashes={} restarts={} still down: {}",
+ injector.crashes(),
+ injector.restarts(),
+ sim.crashed.len(),
);
+ // Printed before the quiesce assert, so a failed drain still reports what
+ // the run managed to do. Reading it after the assert meant the failure that
+ // most needs the numbers is the one that never shows them.
+ print_coverage(&workload);
+
if quiesce {
// A failed drain is a hard failure, not a warning. It used to be one
// because a lost request could not be retried, so a stall was expected
@@ -413,8 +439,16 @@ fn main() {
);
oracle::assert_converged(&sim, &workload);
println!("quiesced and converged (leader-relative + entity oracle)");
+ // Again after the drain: the drain both answers outstanding requests and
+ // issues its own resends, so the pre-drain numbers are not the final ones.
+ print_coverage(&workload);
}
+ println!("workload-fuzz: OK (seed={seed})");
+}
+
+/// Reply, rejection and resend counters plus per-action commits.
+fn print_coverage(workload: &Workload) {
let stats = workload.auditor.stats();
println!(
"coverage: replies_seen={} replies_unknown={} committed_rejections={} \
@@ -431,6 +465,4 @@ fn main() {
println!(" {action:?}: {commits} commits");
}
}
-
- println!("workload-fuzz: OK (seed={seed})");
}
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 3b8d621ae..7c38b1668 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -3135,6 +3135,227 @@ mod tests {
/// pipeline to `PIPELINE_PREPARE_QUEUE_MAX`; a request on `ns_b`
/// still commits while `ns_a` is wedged (no quorum without backup
/// acks); lifting the block drains `ns_a` completely.
+ /// A lost `PrepareOk` does not wedge the metadata plane: once the acks flow
+ /// again the primary reaches its commit quorum without client involvement.
+ ///
+ /// Worth pinning because the obvious reading of the code says otherwise. The
+ /// backup's admission check in `metadata::on_replicate` is a single
+ /// `header.op != current_op + 1`, so a prepare the backup ALREADY HOLDS is
+ /// dropped exactly like a forward gap, logging "dropping out-of-order
+ /// prepare (gap)" — and the primary's retransmit of an unacked prepare is
+ /// precisely such a duplicate. Reading only that check, a lost ack should
+ /// livelock: the primary retransmits forever and every retransmit is
+ /// dropped.
+ ///
+ /// It recovers anyway, so recovery does not depend on that check accepting
+ /// the duplicate, and those gap warnings are benign rather than evidence of
+ /// a stall. This test exists to keep that distinction honest: it fails if
+ /// recovery ever does come to rest on the retransmit being re-acked.
+ ///
+ /// Drops acks rather than crashing anyone, so the property under test is
+ /// about lost acks in general, not about restart recovery.
+ #[test]
+ fn lost_prepare_ok_is_recovered_by_retransmit() {
+ use iggy_binary_protocol::Command2;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0x5EED_0077,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+
+ let committed_before = metadata_commit(&sim, 0);
+
+ // Drop only PrepareOk on both backup links. Everything else still flows,
+ // so the backups receive and journal the prepare; only the primary's
+ // evidence of that is lost.
+ for backup in 1..replica_count {
+ sim.network
+ .link_filter_mut(ProcessId::Replica(backup), ProcessId::Replica(0))
+ .remove(Command2::PrepareOk);
+ }
+
+ let msg = client.create_stream("wl-lost-ack");
+ sim.submit_request(client_id, 0, msg.into_generic());
+
+ // Long enough for the prepare to reach and be journaled by both backups
+ // while the primary sees no acks.
+ for _ in 0..200 {
+ sim.step();
+ }
+ assert_eq!(
+ metadata_commit(&sim, 0),
+ committed_before,
+ "the primary must not commit while every backup ack is dropped"
+ );
+ for backup in 1..replica_count {
+ assert!(
+ metadata_op(&sim, usize::from(backup)) > committed_before,
+ "backup {backup} must have journaled the prepare, else this test \
+ proves nothing about a LOST ack"
+ );
+ }
+
+ // Restore the acks. From here the primary's retransmit is the only route
+ // to a commit, which is exactly the mechanism under test.
+ for backup in 1..replica_count {
+ sim.network
+ .link_filter_mut(ProcessId::Replica(backup), ProcessId::Replica(0))
+ .insert(Command2::PrepareOk);
+ }
+
+ for _ in 0..5_000 {
+ sim.step();
+ if metadata_commit(&sim, 0) > committed_before {
+ return;
+ }
+ }
+ panic!(
+ "metadata commit stuck at {} after 5000 ticks with healthy links: a lost \
+ PrepareOk is no longer recovered, so the backup's gap check is now \
+ swallowing the primary's retransmit",
+ metadata_commit(&sim, 0),
+ );
+ }
+
+ /// A prepare that every backup journaled but never acked still commits after
+ /// those backups restart.
+ ///
+ /// The rejoin path is what recovers it: a restarted replica comes back with
+ /// `current_op` at N from its own WAL, rejoins as a probing backup
+ /// (`Status::Recovering`, see `new_shard`), and its probe draws a targeted
+ /// `StartView` from the primary that returns it to `Normal` and gets the
+ /// tail acked. The primary's own retransmit cannot do it: the backup's
+ /// admission check in `metadata::on_replicate` is a single
+ /// `header.op != current_op + 1`, so a prepare it already holds is dropped
+ /// like a forward gap.
+ ///
+ /// Pinned because the fuzzer finds a run where this recovery does NOT happen
+ /// (see the module-level note on seed 42): there, two live backups hold op 45
+ /// with the primary at commit 44, and both log the gap drop for the whole
+ /// drain. This test covers the case that works, so a regression here narrows
+ /// where that one diverges rather than leaving both unexplained.
+ #[test]
+ fn unacked_prepare_commits_after_the_backup_restarts() {
+ use iggy_binary_protocol::Command2;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0x5EED_0078,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+ let committed_before = metadata_commit(&sim, 0);
+
+ // Lose every backup ack, so the prepare is journaled cluster-wide while
+ // the primary stays one short of its commit quorum.
+ for backup in 1..replica_count {
+ sim.network
+ .link_filter_mut(ProcessId::Replica(backup), ProcessId::Replica(0))
+ .remove(Command2::PrepareOk);
+ }
+
+ let msg = client.create_stream("wl-unacked");
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..200 {
+ sim.step();
+ }
+ for backup in 1..replica_count {
+ assert!(
+ metadata_op(&sim, usize::from(backup)) > committed_before,
+ "backup {backup} must hold the prepare before it is restarted"
+ );
+ }
+ assert_eq!(
+ metadata_commit(&sim, 0),
+ committed_before,
+ "the primary must not have committed while its acks were dropped"
+ );
+
+ // Restart every backup. Each recovers the unacked op from its own WAL and
+ // rejoins as a probing backup, which is the state the primary's
+ // retransmit cannot get an ack out of.
+ for backup in 1..replica_count {
+ sim.replica_crash(backup);
+ for _ in 0..50 {
+ sim.tick();
+ }
+ sim.replica_restart(backup);
+ }
+
+ // Healthy links from here: nothing but the protocol stands between the
+ // primary and its quorum.
+ for backup in 1..replica_count {
+ sim.network
+ .link_filter_mut(ProcessId::Replica(backup), ProcessId::Replica(0))
+ .insert(Command2::PrepareOk);
+ }
+
+ for _ in 0..10_000 {
+ sim.step();
+ if metadata_commit(&sim, 0) > committed_before {
+ return;
+ }
+ }
+ panic!(
+ "metadata commit stuck at {} after 10000 ticks with healthy links and \
+ every replica holding op {}: the restarted backups never re-acked the \
+ prepare they recovered from their own WALs",
+ metadata_commit(&sim, 0),
+ metadata_op(&sim, 1),
+ );
+ }
+
+ /// Committed metadata op on a replica's shard 0.
+ fn metadata_commit(sim: &Simulator, replica_idx: usize) -> u64 {
+ sim.replicas[replica_idx].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .expect("shard 0 owns metadata consensus")
+ .commit_min()
+ }
+
+ /// Highest metadata op a replica has journaled.
+ fn metadata_op(sim: &Simulator, replica_idx: usize) -> u64 {
+ sim.replicas[replica_idx]
+ .metadata_journal
+ .last_op()
+ .unwrap_or(0)
+ }
+
#[test]
fn per_partition_consensus_independence() {
use consensus::PIPELINE_PREPARE_QUEUE_MAX;
diff --git a/core/simulator/src/workload/auditor.rs b/core/simulator/src/workload/auditor.rs
index d9616ec4e..ca51a6016 100644
--- a/core/simulator/src/workload/auditor.rs
+++ b/core/simulator/src/workload/auditor.rs
@@ -184,6 +184,14 @@ impl ServerAuditor {
}
#[must_use]
+ /// The action of an outstanding request, if one is recorded for `key`.
+ /// Diagnostic only: names what a stalled run is waiting on, which the bare
+ /// `(client, request)` pair cannot.
+ #[must_use]
+ pub fn in_flight_action(&self, key: (u128, u64)) -> Option<Action> {
+ self.in_flight.get(&key).map(|entry| entry.action)
+ }
+
pub fn in_flight_count(&self) -> usize {
self.in_flight.len()
}
diff --git a/core/simulator/src/workload/invariants.rs b/core/simulator/src/workload/invariants.rs
index bf00d9286..a20b99281 100644
--- a/core/simulator/src/workload/invariants.rs
+++ b/core/simulator/src/workload/invariants.rs
@@ -51,8 +51,20 @@ impl Invariants {
/// Globally:
/// - total in-flight requests stay within the per-client queue ceiling.
///
- /// Crashed replicas are skipped: their last-seen marks are retained, which
- /// stays correct because both quantities are monotonic across a restart.
+ /// Crashed replicas are skipped and their last-seen marks retained, which is
+ /// correct for `view` (the superblock carries it across a restart) but NOT
+ /// for partition `commit_offset`, because the simulator does not retain
+ /// partition data across a restart the way it retains the metadata WAL and
+ /// the superblocks. `IggyPartitions` is dropped and rebuilt empty, so a
+ /// restarted replica reports `commit_offset` 0 and this trips, reporting a
+ /// harness limitation as a consensus regression.
+ ///
+ /// So restart injection (`WorkloadOptions::restart_per_tick_ratio`) is
+ /// metadata-plane only today. Driving it with partition traffic needs the
+ /// partition journal held by the harness, per-namespace, exactly as
+ /// `SimReplica::metadata_journal` already is; a real server's segments are on
+ /// disk and do survive, so retaining them is the faithful model, and the
+ /// alternative of relaxing this check would instead model total data loss.
///
/// # Panics
/// On any regression or in-flight overflow. The workload seed is in the
diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs
index 4ac4735da..322d66c22 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -188,14 +188,23 @@ impl Workload {
due
}
- /// Outstanding requests as `(client, request, target, attempts)`, in key
- /// order. Diagnostic only: names what a run was still waiting on when it
- /// failed to drain.
+ /// Outstanding requests as `(client, request, action, target, attempts)`, in
+ /// key order. Diagnostic only: names what a run was still waiting on when it
+ /// failed to drain, including which op, since some ops cannot be answered
+ /// twice and a resend of those is a permanent stall rather than a slow one.
#[must_use]
- pub(crate) fn outstanding_summary(&self) -> Vec<(u128, u64, u8, u32)> {
+ pub(crate) fn outstanding_summary(&self) -> Vec<(u128, u64, Option<Action>, u8, u32)> {
self.outstanding
.iter()
- .map(|(&(client, request), entry)| (client, request, entry.target, entry.attempts))
+ .map(|(&key, entry)| {
+ (
+ key.0,
+ key.1,
+ self.auditor.in_flight_action(key),
+ entry.target,
+ entry.attempts,
+ )
+ })
.collect()
}
@@ -463,23 +472,44 @@ const FAULT_SEED_SALT: u64 = 0x5A1A_F0E5_FACE_0001;
///
/// The invariants are asserted after every tick, so a consensus or
/// workload regression panics at the tick it occurs (the seed in the message
-/// replays it). When `crash_per_tick_ratio > 0` the driver also injects
-/// crash-only faults via [`maybe_inject_crash`].
+/// replays it). Crash and restart injection runs through [`FaultInjector`],
+/// idle unless one of the two probabilities is set.
+///
+/// Discards the injector; use [`run_with_faults`] to read the crash and restart
+/// counts back.
pub fn run(
sim: &mut Simulator,
workload: &mut Workload,
clients: &[SimClient],
tick_budget: u64,
replies_target: u64,
+) -> u64 {
+ let mut injector = FaultInjector::new(workload.options.seed, sim.replica_count);
+ run_with_faults(
+ sim,
+ workload,
+ clients,
+ tick_budget,
+ replies_target,
+ &mut injector,
+ )
+}
+
+/// [`run`] against a caller-owned [`FaultInjector`], so a test can assert what
+/// was actually injected instead of trusting the probabilities to have fired.
+pub fn run_with_faults(
+ sim: &mut Simulator,
+ workload: &mut Workload,
+ clients: &[SimClient],
+ tick_budget: u64,
+ replies_target: u64,
+ injector: &mut FaultInjector,
) -> u64 {
let mut invariants = Invariants::new();
- let mut fault_prng = Xoshiro256Plus::seed_from_u64(workload.options.seed ^ FAULT_SEED_SALT);
let mut replies_seen = 0u64;
for _ in 0..tick_budget {
workload.tick();
- if workload.options.crash_per_tick_ratio > 0.0 {
- maybe_inject_crash(sim, workload, &mut fault_prng);
- }
+ injector.step(sim, workload);
// Resend before sampling: a timed-out request still holds the client's
// slot, so `build_request` would decline it anyway.
resubmit_due(sim, workload);
@@ -501,48 +531,141 @@ pub fn run(
replies_seen
}
-/// With probability `crash_per_tick_ratio`, crash one live non-primary replica,
-/// provided doing so leaves at least `min_survivors` live. Crash-only: a
-/// crashed replica is never restarted (that needs consensus durability).
-///
-/// "Non-primary" is partition-plane only: the exclusion set comes from
-/// `Simulator::primary_index`, which reads `partitions()`. The metadata-plane
-/// primary is not consulted; it is spared only by co-location, since every group
-/// starts at view 0 with `primary = view % replica_count` (so replica 0 leads
-/// both planes) and `min_survivors` keeps a commit quorum, so no view change
-/// moves it. Were the two planes' primaries to diverge, the metadata primary
-/// could be crashed.
+/// Crash and restart injection with stability windows, in the shape of
+/// TigerBeetle's VOPR: a crash must last a while before it may be repaired, and
+/// a repaired replica must run a while before it may fail again.
///
-/// Primaries are spared at all because the driver has no request-timeout/resend
-/// path: a request lost to a crashed primary would wedge the client's only
-/// in-flight slot. Forcing primary crashes (and the view change they trigger)
-/// while keeping traffic flowing is future work gated on that resend path.
-fn maybe_inject_crash(sim: &mut Simulator, workload: &Workload, prng: &mut Xoshiro256Plus) {
- let live: Vec<u8> = (0..sim.replica_count)
- .filter(|replica_idx| !sim.is_crashed(*replica_idx))
- .collect();
- if live.len() <= usize::from(workload.options.min_survivors) {
- return;
+/// Owns the fault PRNG so crash scheduling stays reproducible from the seed yet
+/// independent of the traffic draw order. Draws nothing while both probabilities
+/// are zero, so a fault-free run replays bit-identically.
+pub struct FaultInjector {
+ prng: Xoshiro256Plus,
+ /// Tick of each replica's last crash or restart, indexed by replica id.
+ /// Compared against the stability windows to decide eligibility.
+ last_transition: Vec<u64>,
+ now: u64,
+ crashes: u64,
+ restarts: u64,
+}
+
+impl FaultInjector {
+ #[must_use]
+ pub fn new(seed: u64, replica_count: u8) -> Self {
+ Self {
+ prng: Xoshiro256Plus::seed_from_u64(seed ^ FAULT_SEED_SALT),
+ last_transition: vec![0; usize::from(replica_count)],
+ now: 0,
+ crashes: 0,
+ restarts: 0,
+ }
+ }
+
+ #[must_use]
+ pub const fn crashes(&self) -> u64 {
+ self.crashes
}
- let roll: f32 = prng.random();
- if roll >= workload.options.crash_per_tick_ratio {
- return;
+
+ #[must_use]
+ pub const fn restarts(&self) -> u64 {
+ self.restarts
+ }
+
+ /// Advance one tick and maybe crash or restart one replica.
+ ///
+ /// Restart is considered before crash so a single tick never both revives
+ /// and kills, which would make the stability windows meaningless.
+ pub fn step(&mut self, sim: &mut Simulator, workload: &Workload) {
+ self.now += 1;
+ self.maybe_restart(sim, workload);
+ self.maybe_crash(sim, workload);
+ }
+
+ /// With probability `restart_per_tick_ratio`, restart one replica that has
+ /// been down at least `crash_stability_ticks`.
+ ///
+ /// This is what exercises rejoin: the replica comes back with its durable
+ /// superblock and metadata WAL but no volatile consensus state, asks the
+ /// current view's primary for a `StartView`, and repairs the log it missed.
+ fn maybe_restart(&mut self, sim: &mut Simulator, workload: &Workload) {
+ if workload.options.restart_per_tick_ratio <= 0.0 {
+ return;
+ }
+ let eligible: Vec<u8> = (0..sim.replica_count)
+ .filter(|replica_idx| sim.is_crashed(*replica_idx))
+ .filter(|replica_idx| {
+ self.stable_for(*replica_idx) >= workload.options.crash_stability_ticks
+ })
+ .collect();
+ if eligible.is_empty() {
+ return;
+ }
+ let roll: f32 = self.prng.random();
+ if roll >= workload.options.restart_per_tick_ratio {
+ return;
+ }
+ let revived = eligible[self.prng.random_range(0..eligible.len())];
+ sim.replica_restart(revived);
+ self.last_transition[usize::from(revived)] = self.now;
+ self.restarts += 1;
}
- let primaries: HashSet<u8> = workload
- .options
- .namespaces
- .iter()
- .filter_map(|ns| sim.primary_index(*ns))
- .collect();
- let eligible: Vec<u8> = live
- .into_iter()
- .filter(|replica_idx| !primaries.contains(replica_idx))
- .collect();
- if eligible.is_empty() {
- return;
+
+ /// With probability `crash_per_tick_ratio`, crash one live replica that has
+ /// been up at least `restart_stability_ticks`, provided doing so leaves at
+ /// least `min_survivors` live.
+ ///
+ /// Primaries are excluded unless `spare_primary` is off. The exclusion set
+ /// comes from `Simulator::primary_index`, which reads `partitions()`, so it
+ /// names partition-plane primaries; the metadata primary is spared only by
+ /// co-location, since every group starts at view 0 with
+ /// `primary = view % replica_count`. Once views diverge across planes the
+ /// metadata primary can be crashed even with this on.
+ fn maybe_crash(&mut self, sim: &mut Simulator, workload: &Workload) {
+ if workload.options.crash_per_tick_ratio <= 0.0 {
+ return;
+ }
+ let live: Vec<u8> = (0..sim.replica_count)
+ .filter(|replica_idx| !sim.is_crashed(*replica_idx))
+ .collect();
+ if live.len() <= usize::from(workload.options.min_survivors) {
+ return;
+ }
+ let roll: f32 = self.prng.random();
+ if roll >= workload.options.crash_per_tick_ratio {
+ return;
+ }
+ let primaries: HashSet<u8> = if workload.options.spare_primary {
+ workload
+ .options
+ .namespaces
+ .iter()
+ .filter_map(|ns| sim.primary_index(*ns))
+ .collect()
+ } else {
+ HashSet::new()
+ };
+ let eligible: Vec<u8> = live
+ .into_iter()
+ .filter(|replica_idx| !primaries.contains(replica_idx))
+ .filter(|replica_idx| {
+ self.stable_for(*replica_idx) >= workload.options.restart_stability_ticks
+ })
+ .collect();
+ if eligible.is_empty() {
+ return;
+ }
+ let victim = eligible[self.prng.random_range(0..eligible.len())];
+ sim.replica_crash(victim);
+ self.last_transition[usize::from(victim)] = self.now;
+ self.crashes += 1;
+ }
+
+ /// Ticks since this replica last changed state. A replica that never
+ /// transitioned counts from tick 0, so the first crash still has to wait out
+ /// `restart_stability_ticks`.
+ fn stable_for(&self, replica_idx: u8) -> u64 {
+ self.now
+ .saturating_sub(self.last_transition[usize::from(replica_idx)])
}
- let victim = eligible[prng.random_range(0..eligible.len())];
- sim.replica_crash(victim);
}
/// Submit every request whose reply is overdue (see [`Workload::due_resends`]).
diff --git a/core/simulator/src/workload/options.rs b/core/simulator/src/workload/options.rs
index d707b2961..f5fd38acb 100644
--- a/core/simulator/src/workload/options.rs
+++ b/core/simulator/src/workload/options.rs
@@ -27,6 +27,16 @@ use strum::EnumCount;
/// a crashed primary is retried long before the run's budget runs out.
pub const DEFAULT_REQUEST_TIMEOUT_TICKS: u64 = 200;
+/// Default [`WorkloadOptions::crash_stability_ticks`]. Long enough that the
+/// surviving primary commits past the crashed replica's log, so its rejoin has
+/// something to repair.
+pub const DEFAULT_CRASH_STABILITY_TICKS: u64 = 300;
+
+/// Default [`WorkloadOptions::restart_stability_ticks`]. Long enough for a
+/// rejoined replica to finish catching up before it becomes a crash candidate
+/// again, so a run does not consist entirely of half-repaired replicas.
+pub const DEFAULT_RESTART_STABILITY_TICKS: u64 = 500;
+
/// Per-action sampling weights as percentages. Unlisted variants default
/// to 0 (never picked). Listed weights must sum to 100.
#[derive(Debug, Clone, Copy)]
@@ -172,10 +182,29 @@ pub struct WorkloadOptions {
pub consumer_pool_size: u32,
/// Upper bound on offset carried by `StoreConsumerOffset2`.
pub max_offset: u64,
- /// Probability per tick that the driver crashes one live non-primary
- /// replica (crash-only, no restart). `0.0` disables injection: the fault
- /// PRNG draws nothing, so traffic stays bit-identical.
+ /// Probability per tick that the driver crashes one eligible replica.
+ /// `0.0` disables injection entirely: the fault PRNG draws nothing, so
+ /// traffic stays bit-identical.
pub crash_per_tick_ratio: f32,
+ /// Probability per tick that the driver restarts one crashed replica.
+ /// Meaningless without `crash_per_tick_ratio`, since nothing is ever down.
+ pub restart_per_tick_ratio: f32,
+ /// Ticks a replica must stay down before it may be restarted. Keeps a crash
+ /// long enough to actually matter: a replica restarted the tick after it
+ /// crashed never falls behind, so nothing needs repairing.
+ pub crash_stability_ticks: u64,
+ /// Ticks a replica must stay up before it may be crashed again. Stops a
+ /// single unlucky replica from being crash-looped while its peers never
+ /// fail.
+ pub restart_stability_ticks: u64,
+ /// Leave the primary of every tracked namespace out of the crash pool.
+ ///
+ /// Defaults to `true`, which is what the driver did unconditionally before
+ /// clients could resend: a request lost to a crashed primary was never
+ /// retried, so it stranded the client's only in-flight slot. With resending
+ /// in place, setting this to `false` is what puts a view change under live
+ /// traffic, the scenario the harness exists for.
+ pub spare_primary: bool,
/// Floor on live replicas the driver will not crash below, preserving a
/// commit quorum. Defaults to `replica_count / 2 + 1`.
pub min_survivors: u8,
@@ -206,6 +235,10 @@ impl WorkloadOptions {
consumer_pool_size: 4,
max_offset: 1_000_000,
crash_per_tick_ratio: 0.0,
+ restart_per_tick_ratio: 0.0,
+ crash_stability_ticks: DEFAULT_CRASH_STABILITY_TICKS,
+ restart_stability_ticks: DEFAULT_RESTART_STABILITY_TICKS,
+ spare_primary: true,
min_survivors: replica_count / 2 + 1,
request_timeout_ticks: DEFAULT_REQUEST_TIMEOUT_TICKS,
}
diff --git a/core/simulator/src/workload/oracle.rs b/core/simulator/src/workload/oracle.rs
index 4bdfc50f5..567172421 100644
--- a/core/simulator/src/workload/oracle.rs
+++ b/core/simulator/src/workload/oracle.rs
@@ -138,10 +138,10 @@ pub fn quiesce_failure_report(sim: &Simulator, workload: &Workload) -> String {
workload.total_in_flight(),
workload.options.seed,
);
- for (client, request, target, attempts) in workload.outstanding_summary() {
+ for (client, request, action, target, attempts) in workload.outstanding_summary() {
let _ = writeln!(
report,
- " outstanding client={client} request={request} \
+ " outstanding client={client} request={request} action={action:?} \
last_target=replica {target} attempts={attempts}",
);
}
@@ -152,6 +152,26 @@ pub fn quiesce_failure_report(sim: &Simulator, workload: &Workload) -> String {
continue;
}
let _ = write!(report, " replica {replica_idx}: live");
+ // Metadata plane first: a rejoining replica is quorum-invisible until it
+ // completes its view probe, so its status is the difference between "the
+ // cluster is slow" and "the cluster has no quorum despite enough live
+ // replicas".
+ if let Some(consensus) = sim.replicas[usize::from(replica_idx)].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ {
+ let _ = write!(
+ report,
+ " | metadata status={:?} view={} log_view={} commit={} primary={}",
+ consensus.status(),
+ consensus.view(),
+ consensus.log_view(),
+ consensus.commit_min(),
+ consensus.is_primary(),
+ );
+ }
for &ns in &workload.options.namespaces {
let view = sim.consensus_view(usize::from(replica_idx), ns);
let commit = sim
@@ -164,6 +184,26 @@ pub fn quiesce_failure_report(sim: &Simulator, workload: &Workload) -> String {
);
}
report.push('\n');
+ // Per-client table state. `check_request` admits a metadata request only
+ // when it is exactly `watermark + 1`, so the watermark says whether an
+ // outstanding request is still expected, already answered (and thus owed
+ // a cached-reply replay), or ahead of what this replica will accept.
+ let table = sim.replicas[usize::from(replica_idx)].shards[0]
+ .plane
+ .metadata()
+ .client_table
+ .borrow();
+ for client_id in table.client_ids() {
+ let _ = writeln!(
+ report,
+ " client {client_id}: watermark={:?} epoch={:?} cached_reply_request={:?}",
+ table.get_watermark(client_id),
+ table.get_epoch(client_id),
+ table
+ .get_reply(client_id)
+ .map(|reply| reply.header().request),
+ );
+ }
}
report
}
(iggy) 06/18: feat(simulator): carry partition logs across a replica restart
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b98d3a5a66f3d88debbfe360776536e3213ada03
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 18:58:34 2026 +0530
feat(simulator): carry partition logs across a replica restart
A restart drops the whole partition and rebuilds it, which costs a real
server nothing: its messages are in segment files and boot recovers the
offset counter from them. The simulator's partitions are in-memory, so a
rebuilt one came back empty, its commit_offset regressed to zero, and the
monotonicity invariant reported a discarded log as a consensus
regression. That confined restart injection to the metadata plane, whose
WAL the harness already retained.
The harness now holds each group's log across the rebuild and hands it
back on re-materialisation, the same role it already plays for the
metadata WAL and the superblocks. `adopt_retained_log` lives in the
partitions crate beside `restore_offset_frontier` rather than in the
harness, because the two have to agree on maxing against what is already
proved, and the one place built to catch violations of that rule is the
wrong place to keep a second copy of it.
Relaxing the invariant instead would have modelled total data loss rather
than a restart, so it stays strict and its doc no longer claims the
partition plane is exempt.
The regression test asserts both the counter and a poll that returns
messages: a carried-over counter over an empty log would satisfy the
counter alone while having lost everything behind it.
Partition-plane crash/restart runs now drain and converge across seeds
where they previously tripped the invariant (commit_offset 79 -> 0). The
metadata-plane wedge reported earlier is untouched by this and is now
better characterised: it reproduces on every plane containing metadata
ops at every seed tried, so it is systematic rather than seed luck.
---
core/partitions/src/iggy_partition.rs | 45 ++++++++
core/partitions/src/lib.rs | 13 +++
core/shard/Cargo.toml | 6 +-
core/shard/src/lib.rs | 18 ++++
core/simulator/src/lib.rs | 166 +++++++++++++++++++++++++++++-
core/simulator/src/workload/auditor.rs | 1 -
core/simulator/src/workload/invariants.rs | 21 ++--
7 files changed, 250 insertions(+), 20 deletions(-)
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 8dc976169..16e07230d 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -752,6 +752,51 @@ where
/// simulator share one implementation. A copy in the harness was a copy of
/// the max rule that had lost the max, in the one place built to catch
/// violations of it.
+ /// Adopt a log carried over from a previous incarnation of this partition,
+ /// standing in for what segment recovery reads off disk at boot.
+ ///
+ /// A restart drops the whole partition and rebuilds it, which on a real
+ /// server loses nothing: the messages are in segment files and boot recovers
+ /// the offset counter from them. The simulator's partitions are in-memory, so
+ /// without this the rebuilt partition comes back empty and its
+ /// `commit_offset` regresses to zero -- reported as a consensus regression
+ /// when it is really the harness having thrown the data away.
+ ///
+ /// `durable_offset` and `write_offset` are what the caller recovered, exactly
+ /// as `segment_recovery` derives them from the segments it read. Applied as a
+ /// MAX against whatever the superblock frontier already proved, never as an
+ /// overwrite, for the same reason [`Self::restore_offset_frontier`] maxes: a
+ /// recovered value that is behind the frontier must not lower it.
+ ///
+ /// Lives here rather than in the harness so the rule is stated once, beside
+ /// the frontier restore it has to agree with.
+ #[cfg(any(test, feature = "simulator"))]
+ pub fn adopt_retained_log(
+ &mut self,
+ log: SegmentedLog<PartitionJournal<PartitionJournalMemStorage>, PartitionJournalMemStorage>,
+ durable_offset: u64,
+ write_offset: u64,
+ ) {
+ self.log = log;
+ // Empty carry-over: the previous incarnation never took a write, so there
+ // is no offset space to restore and claiming one would make the next
+ // prepare mint from a base no peer agrees on.
+ if write_offset == 0 && durable_offset == 0 && !self.should_increment_offset {
+ return;
+ }
+ let durable = durable_offset.max(self.offset.load(Ordering::Acquire));
+ let dirty = write_offset
+ .max(durable)
+ .max(self.dirty_offset.load(Ordering::Relaxed));
+ self.offset.store(durable, Ordering::Release);
+ self.dirty_offset.store(dirty, Ordering::Relaxed);
+ self.should_increment_offset = true;
+ // Everything carried over is already persisted as far as this replica is
+ // concerned, so the flush and commit paths must not re-persist or re-count
+ // it -- the same contract boot gives a partition recovered from segments.
+ self.recovered_durable_offset = Some(durable);
+ }
+
pub fn restore_offset_frontier(&mut self, recovered: Option<&consensus::VsrState>) {
let Some(frontier) = recovered
.map(|state| state.offset_frontier)
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 2b62402cf..1a773e747 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -38,6 +38,19 @@ pub use iggy_index_reader::IggyIndexReader;
pub use iggy_index_writer::IggyIndexWriter;
pub use iggy_partition::{IggyPartition, PurgeError};
pub use iggy_partitions::IggyPartitions;
+
+/// A partition's message log, named so a caller can carry one across a rebuild.
+///
+/// Exists for the simulator, which has no segment files to recover from and so
+/// must hold the log itself for a restarted replica to come back with its data
+/// (see [`IggyPartition::adopt_retained_log`]). The generic parameters are the
+/// only pair `IggyPartition::log` is ever instantiated with, so this names the
+/// concrete type rather than widening anything.
+#[cfg(any(test, feature = "simulator"))]
+pub type RetainedPartitionLog = log::SegmentedLog<
+ journal::PartitionJournal<journal::PartitionJournalMemStorage>,
+ journal::PartitionJournalMemStorage,
+>;
pub use journal::{EVICTED_RING_BYTES_MAX, EVICTED_RING_CAPACITY};
pub use messages_writer::MessagesWriter;
pub use offset_storage::delete_persisted_offset;
diff --git a/core/shard/Cargo.toml b/core/shard/Cargo.toml
index 874fb5353..826f747ce 100644
--- a/core/shard/Cargo.toml
+++ b/core/shard/Cargo.toml
@@ -28,7 +28,11 @@ publish = false
# off the pump task. A `-p iggy-server` build excludes it; `cargo build
# --workspace` unifies features and compiles it into the shared `shard`
# unit (simulator requests it). Benign: no production caller.
-simulator = []
+# Forwards to `partitions/simulator` because `init_partition` now names
+# `partitions::RetainedPartitionLog` and calls `adopt_retained_log`, both gated
+# there. Without the forward this feature only compiles when something else in
+# the build happens to enable the partitions one.
+simulator = ["partitions/simulator"]
[dependencies]
compio = { workspace = true }
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index bae836f8b..607713758 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3424,12 +3424,21 @@ where
/// `recovered_state` is that store's last record, read by the caller (the
/// store's read is async and this is not), mirroring how `new_shard` takes the
/// metadata plane's.
+ ///
+ /// `retained` is the log a previous incarnation of this group left behind,
+ /// standing in for the segments a real boot recovers from. `None` builds an
+ /// empty partition, which is right for a first materialisation and wrong for a
+ /// restart: a rebuilt partition with no data reports `commit_offset` 0 and
+ /// looks like a regression rather than a harness that discarded the log. The
+ /// recovered offsets ride along because the caller plays the storage layer
+ /// here, exactly as it does for the metadata WAL.
#[cfg(any(test, feature = "simulator"))]
pub fn init_partition(
&self,
namespace: IggyNamespace,
superblock: Option<Rc<SB>>,
recovered_state: Option<consensus::VsrState>,
+ retained: Option<(partitions::RetainedPartitionLog, u64, u64)>,
) where
B: MessageBus + Clone,
{
@@ -3466,6 +3475,15 @@ where
if let Some(superblock) = superblock {
partition.set_superblock(superblock, recovered_state.as_ref());
}
+ // Retained log before the frontier restore, so the restore maxes against
+ // the offsets the log actually proved rather than the zeroes of an empty
+ // one. Both are max rules, so the order only decides which value each sees
+ // first, never the outcome; doing it in this order keeps the frontier
+ // restore's own precondition (`should_increment_offset` already set by a
+ // recovered offset space) meaningful.
+ if let Some((log, durable_offset, write_offset)) = retained {
+ partition.adopt_retained_log(log, durable_offset, write_offset);
+ }
// The SAME call the boot paths make, not a copy of it: this restore is
// a max against what the segments already proved, and a harness running
// a divergent copy of that rule cannot catch a violation of it. Without
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 7c38b1668..7d0ce2413 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -38,7 +38,9 @@ use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
use metadata::impls::metadata::StreamsFrontend;
use network::Network;
use packet::{PacketSimulatorOptions, ProcessId};
-use partitions::{Partition, PartitionOffsets, PollFragments, PollingArgs, PollingConsumer};
+use partitions::{
+ Partition, PartitionOffsets, PollFragments, PollingArgs, PollingConsumer, RetainedPartitionLog,
+};
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use rand_xoshiro::rand_core::SeedableRng;
@@ -106,6 +108,16 @@ pub struct SimReplica {
/// leaves the gate, its write-failure fence, and view recovery all
/// unexercised.
pub partition_superblocks: RefCell<HashMap<IggyNamespace, Rc<SimSuperblock>>>,
+ /// One retained message log per partition group, with the offsets recovered
+ /// from it. Harness-owned for the same reason as `metadata_journal`: a
+ /// restart drops and rebuilds the partition, and a real server loses nothing
+ /// there because its messages are in segment files that boot recovers the
+ /// offset counter from. The simulator has no segment files, so without this
+ /// the rebuilt partition comes back empty, `commit_offset` regresses to zero,
+ /// and the monotonicity invariant reports a discarded log as a consensus
+ /// regression. Populated on the way into a restart (see
+ /// [`Simulator::replica_restart`]) and consumed by `materialise_partition`.
+ partition_logs: RefCell<HashMap<IggyNamespace, (RetainedPartitionLog, u64, u64)>>,
/// Keeps each pump's stop channel alive; dropping one would end that
/// pump gracefully, which is reserved for future shutdown/restart
/// tests (crash uses `DetExecutor::abort` instead).
@@ -347,6 +359,7 @@ impl Simulator {
metadata_journal,
metadata_incarnation,
partition_superblocks: RefCell::new(HashMap::new()),
+ partition_logs: RefCell::new(HashMap::new()),
_stop_txs: stop_txs,
pump_tasks,
});
@@ -771,6 +784,12 @@ impl Simulator {
// as a rebooted server partition reads the record in its directory.
let partition_superblocks =
std::mem::take(&mut *self.replicas[idx].partition_superblocks.borrow_mut());
+ // Take each live partition's log while its shard is still standing. This
+ // is the harness playing the storage layer: a real server's messages are
+ // in segment files and its boot recovers the offset counter from them, so
+ // a partition rebuilt with nothing would model total data loss rather than
+ // a restart. Read before the rebuild because the rebuild drops the shards.
+ let partition_logs = self.retain_partition_logs(idx, &partition_superblocks);
// Recover the durable VSR state from the retained superblock before the
// rebuild, as production reads it in restore_metadata_consensus.
@@ -834,6 +853,7 @@ impl Simulator {
metadata_journal,
metadata_incarnation,
partition_superblocks: RefCell::new(partition_superblocks),
+ partition_logs: RefCell::new(partition_logs),
_stop_txs: stop_txs,
pump_tasks,
};
@@ -863,6 +883,42 @@ impl Simulator {
self.crashed.remove(&replica_index);
}
+ /// Take the message log out of every partition this replica has
+ /// materialised, together with the offsets recovered from it.
+ ///
+ /// Called while the outgoing shards are still alive, so this is the last
+ /// point the data can be read. `std::mem::take` leaves the doomed partition
+ /// with an empty log, which nothing observes: the shard it belongs to is
+ /// dropped moments later.
+ ///
+ /// Keyed off `partition_superblocks` rather than the live partitions map
+ /// because that is already the record of which groups this replica has
+ /// materialised, and it is what the restart re-materialises from.
+ fn retain_partition_logs(
+ &self,
+ replica_idx: usize,
+ materialised: &HashMap<IggyNamespace, Rc<SimSuperblock>>,
+ ) -> HashMap<IggyNamespace, (RetainedPartitionLog, u64, u64)> {
+ let replica = &self.replicas[replica_idx];
+ let mut retained = HashMap::with_capacity(materialised.len());
+ for &namespace in materialised.keys() {
+ let partitions = replica.partition_shard(namespace).plane.partitions();
+ let Some(partition) = partitions.get_mut_by_ns(&namespace) else {
+ continue;
+ };
+ let offsets = partition.offsets();
+ retained.insert(
+ namespace,
+ (
+ std::mem::take(&mut partition.log),
+ offsets.commit_offset,
+ offsets.write_offset,
+ ),
+ );
+ }
+ retained
+ }
+
/// Advance consensus timeouts on every live replica without a full
/// step cycle: fires the pumps' virtual tick timers and runs the
/// executor to quiescence.
@@ -988,7 +1044,17 @@ fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace) {
let recovered_state = superblock
.read_latest_sync()
.and_then(|bytes| VsrState::try_from(bytes.as_slice()).ok());
- replica.shards[usize::from(owner)].init_partition(namespace, Some(superblock), recovered_state);
+ // Hand back the log this group left behind, if it has been materialised
+ // before on this replica. Removed rather than cloned: the rebuilt partition
+ // becomes its sole owner, and a second materialisation without a restart in
+ // between would otherwise resurrect a log the live partition has moved past.
+ let retained = replica.partition_logs.borrow_mut().remove(&namespace);
+ replica.shards[usize::from(owner)].init_partition(
+ namespace,
+ Some(superblock),
+ recovered_state,
+ retained,
+ );
// Commit the namespace before stamping the rows: a partition the metadata
// plane never heard of is a shape production cannot produce, and the shard
// refuses to serve client traffic whose routing-row epoch it cannot match
@@ -2984,7 +3050,7 @@ mod tests {
executor.run_until_stalled(POLL_BUDGET); // borrow acquired; task parks
let grow = Rc::clone(&sim.replicas[0].shards[0]);
executor.spawn(async move {
- grow.init_partition(ns_grow, None, None);
+ grow.init_partition(ns_grow, None, None, None);
});
executor.run_until_stalled(POLL_BUDGET); // grow while the borrow is live
}))
@@ -3019,7 +3085,7 @@ mod tests {
executor.run_until_stalled(POLL_BUDGET);
let grow = Rc::clone(&sim.replicas[0].shards[0]);
executor.spawn(async move {
- grow.init_partition(ns_grow, None, None);
+ grow.init_partition(ns_grow, None, None, None);
});
executor.run_until_stalled(POLL_BUDGET);
@@ -3135,6 +3201,98 @@ mod tests {
/// pipeline to `PIPELINE_PREPARE_QUEUE_MAX`; a request on `ns_b`
/// still commits while `ns_a` is wedged (no quorum without backup
/// acks); lifting the block drains `ns_a` completely.
+ /// A restarted replica comes back with the partition data it had, so its
+ /// `commit_offset` does not regress.
+ ///
+ /// The simulator has no segment files, so the log has to be carried across
+ /// the rebuild by hand (`retain_partition_logs` into
+ /// `IggyPartition::adopt_retained_log`). Without that the rebuilt partition
+ /// is empty and reports `commit_offset` 0, which models total data loss
+ /// rather than a restart and trips the monotonicity invariant on a harness
+ /// artefact. Asserted on a backup, which is where the driver's crash
+ /// injection puts a restart.
+ #[test]
+ fn restarted_replica_keeps_its_partition_offsets() {
+ use bytes::Bytes;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0x5EED_0079,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+ let ns = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns);
+ sim.register_client_with_primary(&client);
+
+ // Enough committed writes that an empty rebuild is unmistakable.
+ for sequence in 0..8u32 {
+ let msg = client.send_messages(ns, &[Bytes::from(format!("retained-{sequence}"))]);
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..40 {
+ sim.step();
+ }
+ }
+
+ let backup = 1usize;
+ let before = sim
+ .offsets(backup, ns)
+ .expect("backup hosts the namespace")
+ .commit_offset;
+ assert!(
+ before > 0,
+ "backup must have committed some writes before the restart, else a \
+ rebuilt-empty partition would be indistinguishable from this state"
+ );
+
+ sim.replica_crash(u8::try_from(backup).expect("replica index fits u8"));
+ for _ in 0..50 {
+ sim.tick();
+ }
+ sim.replica_restart(u8::try_from(backup).expect("replica index fits u8"));
+
+ let after = sim
+ .offsets(backup, ns)
+ .expect("namespace re-materialised on restart")
+ .commit_offset;
+ assert_eq!(
+ after, before,
+ "restarted backup lost its partition offsets: the retained log did not \
+ carry across the rebuild"
+ );
+
+ // The data itself, not just the counter: a carried-over counter with an
+ // empty log would satisfy the assert above and still have lost every
+ // message.
+ let fragments = sim
+ .poll_messages(
+ backup,
+ ns,
+ PollingConsumer::Consumer(0, 0),
+ &PollingArgs::new(iggy_common::PollingStrategy::offset(0), 16, false),
+ )
+ .expect("poll against the re-materialised partition");
+ assert!(
+ !fragments.is_empty(),
+ "restarted backup served no messages: the offset counter came across \
+ but the log behind it did not"
+ );
+ }
+
/// A lost `PrepareOk` does not wedge the metadata plane: once the acks flow
/// again the primary reaches its commit quorum without client involvement.
///
diff --git a/core/simulator/src/workload/auditor.rs b/core/simulator/src/workload/auditor.rs
index ca51a6016..411a83af7 100644
--- a/core/simulator/src/workload/auditor.rs
+++ b/core/simulator/src/workload/auditor.rs
@@ -183,7 +183,6 @@ impl ServerAuditor {
&self.stats
}
- #[must_use]
/// The action of an outstanding request, if one is recorded for `key`.
/// Diagnostic only: names what a stalled run is waiting on, which the bare
/// `(client, request)` pair cannot.
diff --git a/core/simulator/src/workload/invariants.rs b/core/simulator/src/workload/invariants.rs
index a20b99281..d7f1b8a8d 100644
--- a/core/simulator/src/workload/invariants.rs
+++ b/core/simulator/src/workload/invariants.rs
@@ -51,20 +51,13 @@ impl Invariants {
/// Globally:
/// - total in-flight requests stay within the per-client queue ceiling.
///
- /// Crashed replicas are skipped and their last-seen marks retained, which is
- /// correct for `view` (the superblock carries it across a restart) but NOT
- /// for partition `commit_offset`, because the simulator does not retain
- /// partition data across a restart the way it retains the metadata WAL and
- /// the superblocks. `IggyPartitions` is dropped and rebuilt empty, so a
- /// restarted replica reports `commit_offset` 0 and this trips, reporting a
- /// harness limitation as a consensus regression.
- ///
- /// So restart injection (`WorkloadOptions::restart_per_tick_ratio`) is
- /// metadata-plane only today. Driving it with partition traffic needs the
- /// partition journal held by the harness, per-namespace, exactly as
- /// `SimReplica::metadata_journal` already is; a real server's segments are on
- /// disk and do survive, so retaining them is the faithful model, and the
- /// alternative of relaxing this check would instead model total data loss.
+ /// Crashed replicas are skipped and their last-seen marks retained, which
+ /// holds across a restart for both quantities: the superblock carries `view`,
+ /// and the harness carries each partition's log
+ /// (`Simulator::retain_partition_logs`) so a rebuilt partition recovers the
+ /// offsets it had rather than reporting zero. The latter is the harness
+ /// standing in for the segment files a real server recovers from; without it
+ /// this check trips on a discarded log and calls it a consensus regression.
///
/// # Panics
/// On any regression or in-flight overflow. The workload seed is in the
(iggy) 13/18: feat(simulator): checkpoint for real and serve a chunked state transfer
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 007e6537d6e66bac332751da9ce68be559cff438
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 11:12:23 2026 +0530
feat(simulator): checkpoint for real and serve a chunked state transfer
Three things each independently prevented a checkpoint, and all three had
to go before one could happen. Metadata was built without a data
directory, so there was no `SnapshotCoordinator` and `checkpoint_if_needed`
returned immediately. The simulated journal reported no capacity limit, so
`should_checkpoint` read `None` and never fired even where a coordinator
existed. And `drain` was inherited from the trait default, which drains
nothing, so a checkpoint that did run persisted a snapshot and then left
the entire WAL in place.
That last one is why arming the coordinator alone changed nothing
observable: with every op still retained, a rejoining peer found
everything it asked for, ordinary journal repair always sufficed, and no
`RangeEvicted` was ever sent. The symptom looked like the transfer being
unreachable when the cause was a WAL that never shrank.
With all three fixed a lagging replica now completes a real transfer:
RangeEvicted, RequestStateTransfer, StateTransferTarget, RequestStateChunk
and StateChunk each cross the wire exactly once, followed by the tail
repair that finishes the rejoin. Stamping a watermark by hand reached the
handshake and then re-armed 335 times without converging; this converges.
Those are the last two of the five frames no run had ever delivered.
Checkpoints are opt-in through a separate constructor rather than the
default. The coordinator persists through `std::fs`, and a harness whose
defining property is that it touches nothing outside memory should not
start writing files because a scenario forgot to say otherwise. The writes
are synchronous and never reach the executor, so replay stays
deterministic, and the caller owns the directory. The data directory is
retained across a restart for the same reason the WAL and superblocks
already are: a rebuilt replica has to read back the snapshot its previous
incarnation wrote.
---
core/simulator/Cargo.toml | 3 +
core/simulator/src/deps.rs | 100 ++++++++++++++++++++++++
core/simulator/src/lib.rs | 173 ++++++++++++++++++++++++++++++++++++++++++
core/simulator/src/replica.rs | 8 +-
4 files changed, 283 insertions(+), 1 deletion(-)
diff --git a/core/simulator/Cargo.toml b/core/simulator/Cargo.toml
index adc3582a0..8c82f68da 100644
--- a/core/simulator/Cargo.toml
+++ b/core/simulator/Cargo.toml
@@ -50,6 +50,9 @@ strum = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
+[dev-dependencies]
+tempfile = { workspace = true }
+
[lints.clippy]
enum_glob_use = "deny"
pedantic = "deny"
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index 95a16c019..499bbce02 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -27,6 +27,7 @@ use metadata::stm::user::Users;
use server_common::{Message, iobuf::Owned};
use std::cell::{Cell, RefCell, UnsafeCell};
use std::collections::HashMap;
+use std::ops::RangeInclusive;
/// Fixed synthetic epoch for [`SimClock`]: 2026-01-01T00:00:00Z in micros.
///
@@ -107,6 +108,15 @@ pub struct SimJournal<S: Storage> {
/// Snapshot watermark; see the `Journal::snapshot_op` impl for why a real
/// value here is what makes `RangeEvicted` reachable at all.
snapshot_op: Cell<u64>,
+ /// Slots this journal pretends to have, or `None` for unbounded.
+ ///
+ /// A real WAL is a fixed ring, and running low on slots is what forces a
+ /// checkpoint (`SnapshotCoordinator::should_checkpoint` gates on
+ /// `remaining_capacity`). An unbounded journal answers `None` there and so
+ /// never triggers one, which is why the simulator had never checkpointed even
+ /// where a coordinator existed. Defaults to unbounded so existing scenarios
+ /// keep their behaviour; a test that wants checkpoints sets a small count.
+ slot_count: Cell<Option<usize>>,
/// Debug-only single-accessor tripwire. `entry` / `append` hold a
/// [`JournalAccessGuard`] across their whole body, including the storage
/// `.await`, so if a suspending storage tier ever let a second task touch
@@ -125,6 +135,7 @@ impl<S: Storage + Default> Default for SimJournal<S> {
write_offset: Cell::new(0),
last_op: Cell::new(None),
snapshot_op: Cell::new(0),
+ slot_count: Cell::new(None),
#[cfg(debug_assertions)]
accessing: Cell::new(false),
}
@@ -184,6 +195,26 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal<S> for SimJournal<S> {
self.last_op.get()
}
+ /// Slots left before a checkpoint is forced, mirroring
+ /// `PrepareJournal::remaining_capacity`: the ring holds `slot_count` entries
+ /// and everything at or below the snapshot watermark is reclaimable, so what
+ /// is occupied is `last_op - snapshot_op`.
+ ///
+ /// `None` while unbounded, which is what the trait default gave before and
+ /// what `should_checkpoint` reads as "never checkpoint".
+ fn remaining_capacity(&self) -> Option<usize> {
+ let slot_count = self.slot_count.get()?;
+ let Some(last) = self.last_op.get() else {
+ return Some(slot_count);
+ };
+ let snapshot = self.snapshot_op.get();
+ if last <= snapshot {
+ return Some(slot_count);
+ }
+ let used = usize::try_from(last - snapshot).unwrap_or(usize::MAX);
+ Some(slot_count.saturating_sub(used))
+ }
+
/// Drop the suffix, so a simulated backup whose entries disagree with a started
/// view reconciles the way a real one does. Mirrors
/// `PrepareJournal::truncate_from`, whose watermark stays put; here it never moves.
@@ -296,6 +327,66 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal<S> for SimJournal<S> {
let headers = unsafe { &*self.headers.get() };
headers.get(&(idx as u64))
}
+
+ /// Reclaim the prefix a checkpoint superseded, advancing the snapshot
+ /// watermark to the end of the drained range.
+ ///
+ /// Required, not inherited: the trait's default drains nothing, so a
+ /// simulated checkpoint persisted a snapshot and then left the whole WAL in
+ /// place. A peer's repair then found every op it asked for and journal repair
+ /// always sufficed, which is why arming the coordinator alone still never
+ /// produced a `RangeEvicted` or a state transfer.
+ ///
+ /// The watermark moves last, mirroring `PrepareJournal::drain`: advancing it
+ /// before the entries are gone would make live entries look evictable.
+ async fn drain(&self, ops: RangeInclusive<u64>) -> std::io::Result<Vec<Self::Entry>> {
+ #[cfg(debug_assertions)]
+ let _guard = JournalAccessGuard::new(&self.accessing);
+ let end_op = *ops.end();
+ let doomed: Vec<u64> = {
+ let headers = unsafe { &*self.headers.get() };
+ let mut doomed: Vec<u64> = headers
+ .keys()
+ .copied()
+ .filter(|op| ops.contains(op))
+ .collect();
+ // Sorted: the trait promises the removed entries in op order, and
+ // hash order would also make a replay of this drain diverge.
+ doomed.sort_unstable();
+ doomed
+ };
+
+ let mut drained = Vec::with_capacity(doomed.len());
+ for op in doomed {
+ // Read before removing, through `Storage` rather than the
+ // `MemStorage`-only sync path, so this stays generic like the rest of
+ // the impl. The borrow spans the read for the same reason `entry`'s
+ // does, and is sound on the same grounds: `MemStorage` never
+ // suspends, and the guard above trips if that ever changes.
+ let located = {
+ let headers = unsafe { &*self.headers.get() };
+ let offsets = unsafe { &*self.offsets.get() };
+ headers
+ .get(&op)
+ .and_then(|header| offsets.get(&op).map(|offset| (header.size, *offset)))
+ };
+ if let Some((size, offset)) = located
+ && let Ok(buffer) = self.storage.read_at(offset, vec![0; size as usize]).await
+ && let Ok(message) = Message::try_from(Owned::<4096>::copy_from_slice(&buffer))
+ {
+ drained.push(message);
+ }
+ let headers = unsafe { &mut *self.headers.get() };
+ let offsets = unsafe { &mut *self.offsets.get() };
+ headers.remove(&op);
+ offsets.remove(&op);
+ }
+
+ if end_op > self.snapshot_op.get() {
+ self.snapshot_op.set(end_op);
+ }
+ Ok(drained)
+ }
}
impl JournalHandle for SimJournal<MemStorage> {
@@ -315,6 +406,15 @@ impl SimJournal<MemStorage> {
self.last_op.get()
}
+ /// Bound this journal to `slots`, so running low on them forces a checkpoint.
+ ///
+ /// Unbounded by default (see the `slot_count` field): a journal that never
+ /// runs out never triggers `should_checkpoint`, so no checkpoint ever
+ /// happens and nothing produces the snapshot a state transfer serves.
+ pub fn set_slot_count(&self, slots: usize) {
+ self.slot_count.set(Some(slots));
+ }
+
/// Forget one op, leaving a hole exactly where a lost prepare would.
///
/// Tests only. The alternative is choreographing `Prepare`, `Commit` and
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 84829852b..27135d4b0 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -118,6 +118,10 @@ pub struct SimReplica {
/// regression. Populated on the way into a restart (see
/// [`Simulator::replica_restart`]) and consumed by `materialise_partition`.
partition_logs: RefCell<HashMap<IggyNamespace, (RetainedPartitionLog, u64, u64)>>,
+ /// This replica's data directory when checkpoints are enabled, `None`
+ /// otherwise. Retained for the same reason as the WAL and the superblocks: a
+ /// restart must read back the snapshot its previous incarnation wrote.
+ data_dir: Option<std::path::PathBuf>,
/// Keeps each pump's stop channel alive; dropping one would end that
/// pump gracefully, which is reserved for future shutdown/restart
/// tests (crash uses `DetExecutor::abort` instead).
@@ -252,6 +256,51 @@ impl Simulator {
)
}
+ /// [`Simulator::new`] with checkpoints enabled, each replica rooted at
+ /// `<data_dir_root>/replica-N`.
+ ///
+ /// A data directory is what arms the metadata `SnapshotCoordinator`; without
+ /// one `checkpoint_if_needed` returns immediately and no replica ever
+ /// checkpoints, so nothing produces the snapshot a state transfer serves.
+ ///
+ /// Opt-in and separate from the other constructors because the coordinator
+ /// persists through `std::fs`, and a harness whose defining property is that
+ /// it touches nothing outside memory should not start writing files because a
+ /// scenario forgot to say otherwise. The writes are synchronous and never
+ /// touch the executor, so replay stays deterministic; the caller owns the
+ /// directory's lifetime (a `tempfile::TempDir` in tests).
+ ///
+ /// Pair with [`Simulator::set_metadata_journal_slots`]: a checkpoint is forced
+ /// by the journal running low on slots, and the simulated journal is unbounded
+ /// until told otherwise.
+ ///
+ /// # Panics
+ /// Panics on duplicate `client_id`s, `shards_per_replica == 0`, or if the
+ /// per-replica directories cannot be created.
+ pub fn with_checkpoints(
+ replica_count: usize,
+ clients: impl Iterator<Item = u128>,
+ network_options: PacketSimulatorOptions,
+ data_dir_root: std::path::PathBuf,
+ ) -> Self {
+ Self::build_inner(
+ replica_count,
+ 1,
+ clients,
+ network_options,
+ false,
+ Some(data_dir_root),
+ )
+ }
+
+ /// Bound every replica's metadata journal to `slots`, so filling it forces a
+ /// checkpoint. See [`deps::SimJournal::set_slot_count`].
+ pub fn set_metadata_journal_slots(&self, slots: usize) {
+ for replica in &self.replicas {
+ replica.metadata_journal.set_slot_count(slots);
+ }
+ }
+
#[allow(clippy::cast_possible_truncation, clippy::too_many_lines)]
fn build(
replica_count: usize,
@@ -259,6 +308,25 @@ impl Simulator {
clients: impl Iterator<Item = u128>,
network_options: PacketSimulatorOptions,
shell: bool,
+ ) -> Self {
+ Self::build_inner(
+ replica_count,
+ shards_per_replica,
+ clients,
+ network_options,
+ shell,
+ None,
+ )
+ }
+
+ #[allow(clippy::cast_possible_truncation, clippy::too_many_lines)]
+ fn build_inner(
+ replica_count: usize,
+ shards_per_replica: u16,
+ clients: impl Iterator<Item = u128>,
+ network_options: PacketSimulatorOptions,
+ shell: bool,
+ data_dir_root: Option<std::path::PathBuf>,
) -> Self {
assert!(
shards_per_replica >= 1,
@@ -312,6 +380,17 @@ impl Simulator {
// restart increments in `replica_restart` never collide across replicas.
// Non-zero.
let metadata_incarnation = 1 + (u128::from(id) << 64);
+ // One directory per replica when checkpoints are enabled, created up
+ // front because the snapshot coordinator writes into
+ // `<dir>/metadata/` and does not create it. Retained on `SimReplica`
+ // so a restart reads back the snapshot its previous incarnation wrote,
+ // exactly as the retained WAL and superblock already do.
+ let replica_data_dir = data_dir_root.as_ref().map(|root| {
+ let dir = root.join(format!("replica-{i}"));
+ std::fs::create_dir_all(dir.join(metadata::impls::METADATA_DIR))
+ .expect("simulator data directory is creatable");
+ dir
+ });
// One crossfire mesh per replica; every shard gets a clone of
// the canonical senders vec and exclusively takes its inbox.
@@ -354,6 +433,7 @@ impl Simulator {
shard_journal,
None, // fresh boot: no recovered VSR state
metadata_incarnation,
+ (shard_idx == 0).then(|| replica_data_dir.clone()).flatten(),
);
if shard_idx == 0 {
metadata_bundle = Some(
@@ -381,6 +461,7 @@ impl Simulator {
metadata_incarnation,
partition_superblocks: RefCell::new(HashMap::new()),
partition_logs: RefCell::new(HashMap::new()),
+ data_dir: replica_data_dir,
_stop_txs: stop_txs,
pump_tasks,
});
@@ -880,6 +961,10 @@ impl Simulator {
.read_latest_sync()
.and_then(|bytes| VsrState::try_from(bytes.as_slice()).ok());
+ // Carried across the rebuild like the WAL and superblocks: the rebuilt
+ // replica has to find the snapshot its previous incarnation persisted.
+ let replica_data_dir = self.replicas[idx].data_dir.clone();
+
let consensus_clock = ConsensusClock::new(Rc::new(SimClock::new(self.executor.timer())));
let outbox = Rc::clone(&self.outboxes[idx]);
let (senders, mut inboxes) =
@@ -914,6 +999,7 @@ impl Simulator {
shard_journal,
recovered_state,
metadata_incarnation,
+ (shard_idx == 0).then(|| replica_data_dir.clone()).flatten(),
);
if shard_idx == 0 {
metadata_bundle =
@@ -937,6 +1023,7 @@ impl Simulator {
metadata_incarnation,
partition_superblocks: RefCell::new(partition_superblocks),
partition_logs: RefCell::new(partition_logs),
+ data_dir: replica_data_dir,
_stop_txs: stop_txs,
pump_tasks,
};
@@ -3314,6 +3401,92 @@ mod tests {
/// pipeline to `PIPELINE_PREPARE_QUEUE_MAX`; a request on `ns_b`
/// still commits while `ns_a` is wedged (no quorum without backup
/// acks); lifting the block drains `ns_a` completely.
+ /// A replica that checkpoints serves a real state transfer: the rejoining
+ /// peer fetches the snapshot in chunks rather than stalling at the handshake.
+ ///
+ /// The companion to
+ /// [`repair_below_the_snapshot_floor_escalates_to_state_transfer`], which
+ /// stamps a watermark without producing snapshot bytes and so reaches only
+ /// `StateTransferTarget`. Here the coordinator is armed with a data directory
+ /// and the journal is bounded, so the cluster checkpoints for real and the
+ /// transfer has artifacts to serve.
+ ///
+ /// Two things had to be true for a checkpoint to happen at all, and neither
+ /// was: metadata was built without a data directory, so there was no
+ /// `SnapshotCoordinator`; and the simulated journal reported no capacity
+ /// limit, so `should_checkpoint` never fired even where a coordinator existed.
+ #[test]
+ fn checkpointing_cluster_serves_a_chunked_state_transfer() {
+ use iggy_binary_protocol::Command2;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let root = tempfile::tempdir().expect("temp dir for the simulator's snapshots");
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0xC4E0_0001,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::with_checkpoints(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ root.path().to_path_buf(),
+ );
+ // Small enough that the ops below cross the margin; the coordinator forces
+ // a checkpoint once free slots fall to its margin (64 by default).
+ sim.set_metadata_journal_slots(80);
+
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+
+ let lagging = 2u8;
+ sim.replica_crash(lagging);
+
+ // Commit past the checkpoint margin while the lagging replica is down, so
+ // the survivors checkpoint and compact the prefix it is missing.
+ for sequence in 0..40u32 {
+ let msg = client.create_stream(&format!("wl-checkpoint-{sequence}"));
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..40 {
+ sim.step();
+ }
+ }
+
+ let snapshot = root
+ .path()
+ .join("replica-0")
+ .join("metadata")
+ .join("snapshot.bin");
+ assert!(
+ snapshot.exists(),
+ "the primary never checkpointed, so there is no snapshot to transfer: \
+ raise the op count or lower the journal slot count"
+ );
+
+ sim.replica_restart(lagging);
+ for _ in 0..8_000 {
+ sim.step();
+ }
+
+ assert!(
+ sim.network.delivered_any(Command2::RequestStateChunk),
+ "the rejoining replica never asked for a chunk, so the transfer \
+ stalled at the handshake exactly as it does without a checkpoint"
+ );
+ assert!(
+ sim.network.delivered_any(Command2::StateChunk),
+ "no chunk was served: the peer offered a transfer it could not fulfil"
+ );
+ }
+
/// A backup whose gap sits below the serving peer's snapshot floor is told
/// `RangeEvicted` and converts its repair into a state transfer.
///
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 2764b96da..473083d59 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -125,6 +125,7 @@ pub fn new_shard(
metadata_journal: Option<Rc<SimJournal<MemStorage>>>,
recovered_state: Option<VsrState>,
incarnation: u128,
+ data_dir: Option<std::path::PathBuf>,
) -> (Rc<Replica>, Option<SimMetadataBundle>) {
// Metadata is single-writer, mirroring the server bootstrap. Shard 0 owns
// the only writable STM; every peer shard rebuilds a reader-mode mirror from
@@ -243,13 +244,18 @@ pub fn new_shard(
});
let metadata_snapshot = (shard_idx == 0).then(SimSnapshot::default);
+ // A data directory arms the `SnapshotCoordinator`, without which
+ // `checkpoint_if_needed` returns immediately and the replica never
+ // checkpoints. `None` keeps that (the default for scenarios that do not care);
+ // `Some` is opt-in, because the coordinator persists through `std::fs` and a
+ // harness that writes files is not what most specs want.
let metadata = IggyMetadata::new(
metadata_consensus,
metadata_journal,
metadata_snapshot,
superblock,
mux,
- None,
+ data_dir,
);
// Reconstruct shard 0's committed metadata from the retained WAL, the sim analog
(iggy) 17/18: fix(simulator): restore the partition consensus frontier on restart
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 4e9d69cc5994769e63b6f74279fa79f56a5c76b9
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 12:24:46 2026 +0530
fix(simulator): restore the partition consensus frontier on restart
A restarted replica adopted its retained partition log but rebuilt the
group's consensus at op 0, so it advertised an empty frontier in its
`DoViewChange` while holding a log full of ops. A quorum of such replicas
merged to a log shorter than what peers had already committed -- one run
merged `op_head=2` against a replica whose `commit_min` was 34 -- and the
new primary then replayed from the start into a replica far ahead of it,
tripping the sequential-advance assert in `advance_commit_min`.
The metadata plane already restores this in `restore_metadata_consensus`;
the partition plane had the log carried across but nothing that read it
back into consensus. The commit watermark is derived the same way, as the
highest `commit` any journaled prepare stamped, which is a lower bound
that re-commits on rejoin.
Under `--crash-primary` this takes the sweep from 17 of 20 to 19 of 20,
and it also removes the `transmute_header` validity panic entirely, which
turned out to share the cause: a group whose frontier disagreed with its
log built a prepare its own peers could not revalidate.
Lives in `init_partition`, which is simulator-gated, so no production path
changes.
---
core/shard/src/lib.rs | 30 ++++++++++++++++++++++++++++++
1 file changed, 30 insertions(+)
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 607713758..2e477816b 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3483,6 +3483,36 @@ where
// recovered offset space) meaningful.
if let Some((log, durable_offset, write_offset)) = retained {
partition.adopt_retained_log(log, durable_offset, write_offset);
+ // Restore the consensus frontier from the log we just adopted, the
+ // partition-plane counterpart of what `restore_metadata_consensus`
+ // does for shard 0. Without it a restarted replica rebuilds its
+ // consensus at op 0 while holding a log full of ops, and then
+ // ADVERTISES that empty frontier in its `DoViewChange`. A quorum of
+ // such replicas merges to a log shorter than what peers have already
+ // committed, and the new primary replays from the start into a replica
+ // whose `commit_min` is far ahead, tripping the sequential-advance
+ // assert in `advance_commit_min`.
+ //
+ // The commit watermark is the highest `commit` any journaled prepare
+ // stamped: a lower bound, since a prepare records the primary's commit
+ // point at send time, so the true point may be one higher and
+ // re-commits on rejoin. Same rule `SimJournal::recovery_commit_watermark`
+ // applies on the metadata side.
+ let journal = &partition.log.journal().inner;
+ if let Some(head) = journal.last_op() {
+ let mut watermark = 0;
+ for op in 1..=head {
+ if let Some(header) = journal.header_by_op(op) {
+ watermark = watermark.max(header.commit);
+ }
+ }
+ let consensus = partition.consensus();
+ consensus.sequencer().set_sequence(head);
+ consensus.restore_commit_state(watermark, watermark);
+ if let Some(header) = journal.header_by_op(head) {
+ consensus.set_last_prepare_checksum(header.checksum);
+ }
+ }
}
// The SAME call the boot paths make, not a copy of it: this restore is
// a max against what the segments already proved, and a harness running
(iggy) 18/18: feat(simulator): re-establish a session after the cluster evicts a client
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 1771001c876a4d78c20a9164df61744326f75d74
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 14:45:30 2026 +0530
feat(simulator): re-establish a session after the cluster evicts a client
Session bindings live in the per-connection `SessionManager` and do not
survive a replica restart, so a client that had a session there becomes
unbound and its next replicated request is refused with
`Eviction(NoSession)`. The driver treated that as unrecoverable and
stopped the run, which made the dispatch shell and crash/restart mutually
exclusive: every combination of the two failed.
A real client reconnects and logs in, so the driver does too. The client
table is replicated metadata and survives the restart, so the re-login
rebinds the existing entry rather than creating one, and request numbering
continues instead of restarting. The login rotates targets across retries
for the same reason the workload's resends do: a register has to reach the
metadata primary, and retrying one replica is useless once it is
partitioned or crashed mid-handshake. Fixing that alone took the sweep
from 4 of 9 to 11 of 12.
Outstanding requests are forgotten rather than resent, since the message
the retry buffer holds carries the old session id and would be refused
again. That forgetting disarms the strict outcome oracle for the rest of
the run. The refusal proves only that the ATTEMPT which drew it did not
commit, and that attempt may have been a resend whose original had already
committed with its reply lost; the shadow is then missing an effect that
really happened. Keeping the oracle armed would turn a known unknown into
a spurious failure, so it stands down and says why.
Shell with faults and crash/restart now drains and converges 11 of 12
across four op mixes, where it previously could not run at all. The raw
path is unchanged at 9 of 9.
---
core/simulator/src/bin/workload-fuzz.rs | 3 +-
core/simulator/src/lib.rs | 7 +++
core/simulator/src/workload/auditor.rs | 12 +++++
core/simulator/src/workload/mod.rs | 93 +++++++++++++++++++++++++--------
core/simulator/src/workload/oracle.rs | 8 +++
5 files changed, 101 insertions(+), 22 deletions(-)
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index dd0720cf4..6638cf0d4 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -526,13 +526,14 @@ fn print_coverage(workload: &Workload) {
let stats = workload.auditor.stats();
println!(
"coverage: replies_seen={} replies_unknown={} committed_rejections={} \
- samples_none={} resends={} denials={}",
+ samples_none={} resends={} denials={} evictions={}",
stats.replies_seen,
stats.replies_unknown,
stats.committed_rejections,
workload.samples_none(),
workload.resends(),
stats.denials,
+ workload.evictions(),
);
for action in Action::iter() {
let commits = stats.commits(action);
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 4a0b3a08d..61b8cca2d 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -606,9 +606,16 @@ impl Simulator {
message: &Message<GenericHeader>,
label: &str,
) -> Message<ReplyHeader> {
+ let mut target = target;
for step in 0..SETUP_TOTAL_STEPS {
if step % SETUP_RETRY_STEPS == 0 {
self.submit_request(client_id, target, message.deep_copy());
+ // Rotate, exactly as the workload's resend path does. Retrying the
+ // same replica forever is enough on a perfect network and useless
+ // once one is partitioned or crashed mid-handshake: a register has
+ // to reach the metadata primary, and the replica that looked live
+ // when this started may be neither reachable nor able to forward.
+ target = (target + 1) % self.replica_count.max(1);
}
if let Some(reply) = self.step().into_iter().next() {
return reply;
diff --git a/core/simulator/src/workload/auditor.rs b/core/simulator/src/workload/auditor.rs
index 96cd09258..44fd724d4 100644
--- a/core/simulator/src/workload/auditor.rs
+++ b/core/simulator/src/workload/auditor.rs
@@ -209,6 +209,18 @@ impl ServerAuditor {
&self.stats
}
+ /// Drop every in-flight expectation for `client`, returning how many went.
+ ///
+ /// For a client the cluster evicted: its session is gone, so nothing it had
+ /// outstanding will ever be answered and an expectation left behind would
+ /// wait forever. The requests themselves were refused before commit, so
+ /// forgetting them loses no committed state.
+ pub fn forget_client(&mut self, client: u128) -> usize {
+ let before = self.in_flight.len();
+ self.in_flight.retain(|&(owner, _), _| owner != client);
+ before - self.in_flight.len()
+ }
+
/// The action of an outstanding request, if one is recorded for `key`.
/// Diagnostic only: names what a stalled run is waiting on, which the bare
/// `(client, request)` pair cannot.
diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs
index 9e25f9f82..95ca36c63 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -93,6 +93,9 @@ pub struct Workload {
now: u64,
/// Total resends issued, for the run summary.
resends: u64,
+ /// Client evictions survived, for the run summary. Only the dispatch shell
+ /// produces them, and in practice only once replicas restart under it.
+ evictions: u64,
/// Debug counter for `sample()` returning `None` (a targeted outcome whose
/// shadow precondition is unmet). Flags PRNG-trace drift during development.
samples_none: u64,
@@ -121,6 +124,7 @@ impl Workload {
outstanding: BTreeMap::new(),
now: 0,
resends: 0,
+ evictions: 0,
samples_none: 0,
strict_outcome_oracle,
}
@@ -151,6 +155,42 @@ impl Workload {
self.resends
}
+ /// Total client evictions survived.
+ #[must_use]
+ pub const fn evictions(&self) -> u64 {
+ self.evictions
+ }
+
+ /// Forget everything outstanding for a client the cluster evicted.
+ ///
+ /// An eviction is session-terminal: the server refused the request BEFORE
+ /// commit (`Eviction(NoSession)` from an unbound transport, which is what a
+ /// replica restart leaves behind, since session bindings live in the
+ /// per-connection `SessionManager` and do not survive it). The client logs in
+ /// again and carries on.
+ ///
+ /// Resending is not an option: the retained message carries the old session
+ /// id, so it would be refused again. A fresh sample after the re-login
+ /// carries the new one.
+ ///
+ /// The forgotten request's fate is genuinely unknown, which is why this
+ /// disarms the strict outcome oracle. The refusal proves only that the
+ /// ATTEMPT that drew it did not commit, and that attempt may have been a
+ /// resend of a request whose original had already committed with its reply
+ /// lost. The shadow is then missing an effect that did happen, and every
+ /// later outcome targeted against it can disagree with what commits. Losing
+ /// the oracle for the rest of the run is the honest price; claiming the
+ /// shadow is still authoritative would turn a known unknown into a spurious
+ /// failure.
+ ///
+ /// Returns how many requests were forgotten.
+ pub fn forget_evicted_client(&mut self, client_id: u128) -> usize {
+ self.evictions += 1;
+ self.strict_outcome_oracle = false;
+ self.outstanding.retain(|&(owner, _), _| owner != client_id);
+ self.auditor.forget_client(client_id)
+ }
+
/// Requests whose reply has not arrived within
/// [`WorkloadOptions::request_timeout_ticks`], each paired with the replica
/// to retry it against. Callers must submit every returned message.
@@ -541,7 +581,7 @@ pub fn run_with_faults(
apply_sim_commands(sim, &cmds);
replies_seen += 1;
}
- assert_no_evictions(sim);
+ recover_evicted_clients(sim, workload, clients);
invariants.check(sim, workload);
if replies_seen >= replies_target {
break;
@@ -688,30 +728,41 @@ impl FaultInjector {
}
}
-/// Fail loudly if the cluster evicted a client, which the workload cannot yet
-/// survive.
+/// Log any evicted client back in, which is what a real client does.
///
-/// An eviction ends the session: the client's outstanding requests become
-/// unanswerable and it must log in again before submitting anything. Modelling
-/// that means re-establishing the session mid-run and discarding the auditor's
-/// expectations for it, which the driver does not do. Until it does, an eviction
-/// presents as a client that has silently stopped making progress, so name it
-/// here rather than let the run time out with no explanation.
+/// A replica restart drops its `SessionManager`, since bindings are
+/// per-connection and volatile, so a client that had a session there is unbound
+/// and its next replicated request is refused with `Eviction(NoSession)`. The
+/// client table itself is replicated metadata and survives, so the re-login
+/// rebinds the existing entry (bumping its fence epoch) and request numbering
+/// continues rather than restarting.
///
-/// Only reachable through the dispatch shell, and in practice only once crashes
-/// and restarts are also in play.
+/// Outstanding requests are forgotten rather than resent: the refused request
+/// never committed, and the message the retry buffer holds carries the old
+/// session id, so resending it would only be refused again. See
+/// [`Workload::forget_evicted_client`].
///
/// # Panics
-/// If any client was evicted since the last step.
-fn assert_no_evictions(sim: &mut Simulator) {
- let evicted = sim.take_evictions();
- assert!(
- evicted.is_empty(),
- "cluster evicted client(s) {evicted:?}: their sessions are gone, so their \
- outstanding requests can never be answered and their next request is \
- refused. The workload does not re-establish a session, so the run cannot \
- continue"
- );
+/// If an evicted client id is not one the driver knows about, which would mean
+/// the simulator and the driver disagree about who is connected.
+fn recover_evicted_clients(sim: &mut Simulator, workload: &mut Workload, clients: &[SimClient]) {
+ for client_id in sim.take_evictions() {
+ let client = clients
+ .iter()
+ .find(|client| client.client_id() == client_id)
+ .unwrap_or_else(|| {
+ panic!("cluster evicted unknown client {client_id}: not one the driver drives")
+ });
+ workload.forget_evicted_client(client_id);
+ // Any live replica: a client dialing a backup is a supported path (the
+ // backup forwards the register), and the default target may itself be the
+ // replica whose restart caused the eviction, in which case the login just
+ // times out.
+ let Some(target) = (0..sim.replica_count).find(|idx| !sim.is_crashed(*idx)) else {
+ continue;
+ };
+ sim.shell_login_via(client, target);
+ }
}
/// Submit every request whose reply is overdue (see [`Workload::due_resends`]).
diff --git a/core/simulator/src/workload/oracle.rs b/core/simulator/src/workload/oracle.rs
index b1eba8e5d..63daa2d72 100644
--- a/core/simulator/src/workload/oracle.rs
+++ b/core/simulator/src/workload/oracle.rs
@@ -116,6 +116,14 @@ pub fn drive_to_quiesce(sim: &mut Simulator, workload: &mut Workload, max_ticks:
let cmds = workload.on_reply(&reply);
apply_sim_commands(sim, &cmds);
}
+ // A resend can land on a transport a restart left unbound, which the
+ // server refuses with an eviction. No re-login here, unlike the active
+ // driver: the drain submits nothing new, and the refused request was
+ // rejected before commit, so forgetting it is what "drained" means for a
+ // request that can never be answered.
+ for client_id in sim.take_evictions() {
+ workload.forget_evicted_client(client_id);
+ }
if workload.total_in_flight() == 0 {
drained = true;
break;
(iggy) 07/18: fix(metadata): gate the backup gap check on the journal, not the sequencer
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 98936dcb7b5b8b777f7e7bc30d810dfa64131136
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 23:52:51 2026 +0530
fix(metadata): gate the backup gap check on the journal, not the sequencer
A backup admitted a prepare only at `sequencer + 1`, but the sequencer and
the journal are two frontiers that legitimately disagree: `on_start_view`
sets the sequencer to the view's announced head, deliberately ahead of
what this replica holds, because the bodies arrive afterwards by
retransmit or repair.
So a rejoining backup with journal head 44 and an adopted head of 45
rejected op 45 -- exactly the prepare that would heal its log -- as an
out-of-order gap. The primary retransmits that op for the life of the
process, every backup drops it the same way, it never reaches a commit
quorum, and the client that submitted it is never answered. Nothing
recovers: the one message that would produce the missing ack is refused
by the receiver.
The journal is the only frontier that answers "what can be appended
next", which is what this check is for, and the hash-chain verification
beside it is already stated against the journal. A prepare at or below the
head that the replica already holds is re-acked and returned earlier, so
anything reaching the check is either the next op or a real forward gap
for repair to fill. The sequencer is still reported in the warn, since a
gap with the two frontiers disagreeing means something else.
TigerBeetle avoids the question by keeping a single frontier: `self.op` IS
the log head, and `on_prepare` routes anything at or below it to
`on_repair`, which re-acks a prepare already held.
Found by the deterministic simulator, which wedged on every workload
containing metadata ops under crash/restart injection, at every seed
tried. Those runs now drain and converge: 20 of 20 across four op mixes
and five seeds, where metadata and uniform mixes were 0 of 3 before.
---
core/metadata/src/impls/metadata.rs | 123 +++++++++++++++++++++++++++++++++---
1 file changed, 115 insertions(+), 8 deletions(-)
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 34c80fbc2..b3b682d02 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -1204,21 +1204,38 @@ where
// guard, not here.
self.checkpoint_if_needed(consensus, journal).await;
- // Backup: gap check (op == current_op + 1).
- // Primary: sequencer pre-advanced by push_prepare_entry (guards
- // sibling on_request races during journal.append await).
- // TODO: promote the backup gap warn below to a hard assert or a
- // repair-session trigger (message repair has landed; the drop-and-
- // wait-for-retransmit path is the last soft handling left here).
+ // Backup: gap check against the JOURNAL head, not the sequencer.
+ //
+ // These are two different frontiers and they can disagree: the sequencer
+ // is pre-advanced on the primary by `push_prepare_entry` and re-synced on a
+ // backup only after a successful append, so a replica can carry a sequencer
+ // one ahead of what its WAL holds. Gating admission on the sequencer then
+ // rejects the very prepare that would heal the log -- observed as a backup
+ // with `last_op = 44` refusing op 45 because its sequencer said to expect
+ // 46. The primary retransmits that op for the life of the process, every
+ // backup logs it as an out-of-order gap, it never reaches a commit quorum,
+ // and its client is never answered.
+ //
+ // The journal is the only frontier that answers "what can be appended
+ // next", which is what this check is for; the hash-chain verification below
+ // is likewise stated against the journal. TigerBeetle avoids the question
+ // by keeping a single frontier (`self.op` IS the log head, see
+ // `on_prepare`), so nothing can drift.
+ //
+ // A prepare at or below the head that this replica already holds was
+ // re-acked and returned above, so anything reaching here is either the next
+ // op or a real forward gap for repair to fill.
let is_backup = consensus.is_follower();
if is_backup {
- if header.op != current_op + 1 {
+ let journal_head = journal.handle().last_op().unwrap_or(0);
+ if header.op != journal_head + 1 {
warn!(
target: "iggy.metadata.diag",
plane = "metadata",
replica_id = consensus.replica(),
op = header.op,
- expected = current_op + 1,
+ expected = journal_head + 1,
+ sequencer_op = current_op,
"on_replicate: dropping out-of-order prepare (gap)"
);
return;
@@ -4755,6 +4772,96 @@ mod tests {
);
}
+ /// A backup admits the prepare its JOURNAL needs next, even when its
+ /// sequencer has run ahead of the journal.
+ ///
+ /// The two frontiers legitimately disagree: `on_start_view` sets the
+ /// sequencer to the view's announced head, deliberately ahead of what this
+ /// replica holds, because the bodies arrive afterwards by retransmit or
+ /// repair. Gating admission on the sequencer therefore rejected exactly the
+ /// prepare that would heal the log -- a backup with journal head 44 refusing
+ /// op 45 because its adopted head said to expect 46. The primary retransmits
+ /// that op forever, every backup logs it as an out-of-order gap, it never
+ /// reaches a commit quorum, and its client is never answered. Systematic for
+ /// any rejoining replica, so the deterministic simulator wedged on every
+ /// metadata workload under crash/restart injection until this was gated on the
+ /// journal instead.
+ ///
+ /// TigerBeetle sidesteps the question by keeping a single frontier: `self.op`
+ /// IS the log head, and `on_prepare` routes anything at or below it to
+ /// `on_repair`, which re-acks a prepare already held.
+ #[compio::test]
+ async fn backup_admits_the_prepare_its_journal_needs_despite_a_leading_sequencer() {
+ const CLIENT: u128 = 1;
+ const SESSION: u64 = 1;
+ const ACTING_USER: u32 = 7;
+ /// Stands in for a head adopted from a `StartView` whose bodies have not
+ /// arrived, so it sits well above the empty journal.
+ const ADOPTED_HEAD: u64 = 5;
+
+ let dir = tempfile::tempdir().unwrap();
+ std::fs::create_dir_all(dir.path().join(crate::impls::METADATA_DIR)).unwrap();
+ let journal =
+ journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"), 0)
+ .await
+ .unwrap();
+ // Replica 1 of 3 at view 0, so `primary_index(0) == 0` makes this a backup
+ // and `on_replicate` takes the gap-check branch.
+ let consensus = VsrConsensus::new(
+ 1,
+ 1,
+ 3,
+ server_common::sharding::METADATA_GROUP,
+ NoopBus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> =
+ IggyMetadata::new(
+ Some(consensus),
+ Some(journal),
+ None,
+ None,
+ TestMux::default(),
+ Some(dir.path().to_path_buf()),
+ );
+ let consensus = md.consensus.as_ref().unwrap();
+ assert!(
+ consensus.is_follower(),
+ "replica 1 of 3 at view 0 must be a backup for this to exercise the gap check"
+ );
+ md.client_table.borrow_mut().commit_register(
+ CLIENT,
+ ACTING_USER,
+ register_reply(CLIENT, SESSION),
+ );
+
+ // Minted while the sequencer is still at 0, so it carries op 1: exactly
+ // what the empty journal needs next.
+ let prepare = md
+ .prepare_request(create_stream_request(CLIENT, 1, "s1"))
+ .expect("CreateStream is client-allowed");
+ assert_eq!(prepare.header().op, 1, "the first prepare must be op 1");
+
+ // Now run the sequencer ahead, as adopting a started view does.
+ consensus.sequencer().set_sequence(ADOPTED_HEAD);
+ let journal = md.journal.as_ref().unwrap();
+ assert_eq!(
+ journal.last_op(),
+ None,
+ "the journal must still be empty, else the divergence under test is absent"
+ );
+
+ md.on_replicate(prepare).await;
+
+ assert!(
+ journal.header(1).is_some(),
+ "backup dropped the prepare its journal needed next because its sequencer \
+ was ahead; the primary's retransmit of this op can never be accepted, so \
+ the op never commits and its client never gets a reply"
+ );
+ }
+
/// A checkpoint reclaims the WAL prefix the snapshot supersedes, but must
/// stop one op short of the checkpoint op itself.
///
(iggy) 11/18: feat(simulator): cover register and logout forwarding, and measure command reach
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit deb2e974e2fb0b901653172f38f929437b2e57ff
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 07:33:30 2026 +0530
feat(simulator): cover register and logout forwarding, and measure command reach
Register and logout forwarding landed with no deterministic coverage and
could not have had any: forwarding fires only when a client dials a
BACKUP, and the harness only ever dialed the primary. The backup verifies
the credentials itself, sends the consensus proposal on as
`ForwardRegister`, parks the login until `ForwardRegisterResult` returns,
then answers on the connection it owns; logout works the same way. All
four frames are now exercised by a login and logout issued against a
backup.
The test asserts the frames were DELIVERED rather than inferring them from
a working session, because dialing a backup would also appear to work if
the server had silently redirected the client, which is the design
forwarding replaced. Pointing the same test at the primary fails on that
assertion, so it cannot pass vacuously.
Proving that needed a way to observe the wire, so the packet simulator now
counts deliveries per command. That answers a question the harness could
not previously answer about itself: which of the protocol it actually
reaches. A run with crashes and restarts delivers ten commands and
notably includes RequestStartView, RequestPrepares, RepairPrepare and
RepairDone, so restart injection does drive rejoin and log repair rather
than merely being configured to. Adding `--crash-primary` adds
StartViewChange and DoViewChange. Nineteen commands are still never
delivered, among them RangeEvicted and the four state-transfer frames,
which is the next task's target.
Logging out needed a `SimClient::logout`, which the client had never had
because nothing tore a session down.
---
core/simulator/src/bin/workload-fuzz.rs | 28 +++++++-
core/simulator/src/client.rs | 12 ++++
core/simulator/src/lib.rs | 116 +++++++++++++++++++++++++++++++-
core/simulator/src/network.rs | 18 ++++-
core/simulator/src/packet.rs | 71 +++++++++++++++++++
5 files changed, 241 insertions(+), 4 deletions(-)
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index fd918e483..940120387 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -47,7 +47,7 @@ use server_common::sharding::IggyNamespace;
use server_common::{MemoryPool, MemoryPoolConfigOther};
use simulator::Simulator;
use simulator::client::SimClient;
-use simulator::packet::{PacketSimulatorOptions, PartitionMode, PartitionSymmetry};
+use simulator::packet::{COMMAND_LABELS, PacketSimulatorOptions, PartitionMode, PartitionSymmetry};
use simulator::workload::actions::Action;
use simulator::workload::options::{ActionWeights, WorkloadOptions};
use simulator::workload::{FaultInjector, Workload, oracle, run_with_faults};
@@ -480,9 +480,35 @@ fn main() {
print_coverage(&workload);
}
+ print_command_coverage(&sim);
println!("workload-fuzz: OK (seed={seed})");
}
+/// Which protocol commands the run actually delivered, and which it never
+/// reached.
+///
+/// The harness wires far more of the command space than any one scenario drives,
+/// and "is this path covered?" was previously answered by grepping the source.
+/// Counted at delivery, so a command listed here really arrived somewhere.
+fn print_command_coverage(sim: &Simulator) {
+ let counts = sim.network.command_counts();
+ let mut seen: Vec<String> = Vec::new();
+ let mut unseen: Vec<&str> = Vec::new();
+ for (discriminant, &count) in counts.iter().enumerate() {
+ let label = COMMAND_LABELS[discriminant];
+ if label == "Reserved" {
+ continue;
+ }
+ if count > 0 {
+ seen.push(format!("{label}={count}"));
+ } else {
+ unseen.push(label);
+ }
+ }
+ println!("commands delivered: {}", seen.join(" "));
+ println!("commands never delivered: {}", unseen.join(" "));
+}
+
/// Reply, rejection and resend counters plus per-action commits.
fn print_coverage(workload: &Workload) {
let stats = workload.auditor.stats();
diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs
index 02ea37323..88e0d9d08 100644
--- a/core/simulator/src/client.rs
+++ b/core/simulator/src/client.rs
@@ -231,6 +231,18 @@ impl SimClient {
.expect("login request must be valid")
}
+ /// Tear down this client's bound session.
+ ///
+ /// Replicates through the metadata plane like any other session op, so it
+ /// carries the bound session and a metadata request id and needs no body. A
+ /// logout issued to a BACKUP is what produces `ForwardLogout`: the backup owns
+ /// the connection but not the log, so it asks the primary to commit the
+ /// teardown and answers the client itself once `ForwardLogoutResult` returns.
+ #[must_use]
+ pub fn logout(&self) -> Message<RoutedRequestHeader> {
+ self.build_request(Operation::Logout, &[])
+ }
+
/// # Panics
/// Panics if the stream name is not a valid wire name.
pub fn create_stream(&self, name: &str) -> Message<RoutedRequestHeader> {
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 6f0ecef54..9ad034d00 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -462,6 +462,23 @@ impl Simulator {
/// If no login reply arrives within [`SETUP_TOTAL_STEPS`], or it carries no
/// session.
pub fn shell_login(&mut self, client: &SimClient) {
+ self.shell_login_via(client, 0);
+ }
+
+ /// [`Self::shell_login`] against a chosen replica.
+ ///
+ /// Dialing a BACKUP takes a different path than dialing the primary, and it
+ /// is the only way to reach register forwarding: the backup verifies the
+ /// credentials itself and sends only the consensus proposal onward as
+ /// `ForwardRegister`, parking the login until the matching
+ /// `ForwardRegisterResult` comes back, then answering on the connection it
+ /// owns. A client that always dials the primary never produces one of those
+ /// four frames.
+ ///
+ /// # Panics
+ /// If no login reply arrives within [`SETUP_TOTAL_STEPS`], or it carries no
+ /// session.
+ pub fn shell_login_via(&mut self, client: &SimClient, target: u8) {
// Register the client's connection metadata on every replica, as
// `install_client_fd` does in production. `ensure_transport_connection`
// reads it to admit the connection into the SessionManager, which the
@@ -481,7 +498,7 @@ impl Simulator {
// The login reply carries the assigned session in `op`
// (`build_reply_with_body` maps the session field to `op`).
let session = self
- .await_setup_reply(client.client_id(), 0, &msg, "shell_login")
+ .await_setup_reply(client.client_id(), target, &msg, "shell_login")
.map_or(0, |reply| reply.header().op);
assert!(session > 0, "shell_login: login reply carried no session");
client.bind_session(session);
@@ -3272,6 +3289,103 @@ mod tests {
/// pipeline to `PIPELINE_PREPARE_QUEUE_MAX`; a request on `ns_b`
/// still commits while `ns_a` is wedged (no quorum without backup
/// acks); lifting the block drains `ns_a` completely.
+ /// A client that dials a BACKUP still gets a working session, and the login
+ /// travels as a forwarded consensus proposal rather than a redirect.
+ ///
+ /// Register forwarding exists so a client need not find the primary itself:
+ /// the backup verifies the credentials locally, sends only the proposal on as
+ /// `ForwardRegister`, parks the login until the matching
+ /// `ForwardRegisterResult` returns, then answers on the connection it owns.
+ /// The whole subsystem landed with no deterministic coverage, and could not
+ /// have any while the harness only ever dialed the primary.
+ ///
+ /// Asserts the four frames were delivered rather than inferring them from a
+ /// working session: dialing a backup would also "work" if the server had
+ /// silently redirected the client instead, which is the design this replaced.
+ #[test]
+ fn login_via_backup_forwards_the_register_to_the_primary() {
+ use iggy_binary_protocol::Command2;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed: 0xF0_2D_0001,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::with_shards_shell(
+ usize::from(replica_count),
+ 1,
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let ns = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns);
+ sim.seed_stream_topic_partition(ns);
+
+ // Replica 0 leads both planes at view 0, so replica 1 is a backup and the
+ // login has to be forwarded.
+ let client = SimClient::new(client_id);
+ sim.shell_login_via(&client, 1);
+
+ assert!(
+ sim.network.delivered_any(Command2::ForwardRegister),
+ "no ForwardRegister crossed the wire: the backup answered the login \
+ itself, so this covers nothing"
+ );
+ assert!(
+ sim.network.delivered_any(Command2::ForwardRegisterResult),
+ "the forwarded register was never answered, so the login below \
+ succeeded by some other route"
+ );
+
+ // Log out on the SAME backup, which both covers the other half of the
+ // forwarding subsystem and proves the session was real: only a bound
+ // session can be torn down, and the teardown replicates through the
+ // primary exactly as the register did.
+ //
+ // A logout, not a data request, because the session belongs to the
+ // connection: a real client holds one connection to one replica, and a
+ // partition write is refused by a backup for routing reasons
+ // (`TransientNotAccepted`) which says nothing about the session.
+ let logout = client.logout();
+ let request = logout.header().request;
+ sim.submit_request(client_id, 1, logout.into_generic());
+ let mut answered = false;
+ for _ in 0..400 {
+ if let Some(reply) = sim
+ .step()
+ .into_iter()
+ .find(|reply| reply.header().request == request)
+ {
+ assert_eq!(
+ reply.header().status,
+ 0,
+ "logout on the forwarded session was refused (status {})",
+ reply.header().status,
+ );
+ answered = true;
+ break;
+ }
+ }
+ assert!(answered, "no reply to the logout issued on the backup");
+ assert!(
+ sim.network.delivered_any(Command2::ForwardLogout),
+ "the backup committed the logout without asking the primary"
+ );
+ assert!(
+ sim.network.delivered_any(Command2::ForwardLogoutResult),
+ "the forwarded logout was never answered"
+ );
+ }
+
/// The workload drains and converges when every request goes through the
/// server's real dispatch handlers rather than the raw `on_message` path.
///
diff --git a/core/simulator/src/network.rs b/core/simulator/src/network.rs
index 6a4aa795e..abe7eeb66 100644
--- a/core/simulator/src/network.rs
+++ b/core/simulator/src/network.rs
@@ -22,9 +22,10 @@
//! process-to-bus routing, and node enable/disable logic.
use crate::packet::{
- ALLOW_ALL, BLOCK_ALL, LinkFilter, Packet, PacketSimulator, PacketSimulatorOptions, ProcessId,
+ ALLOW_ALL, BLOCK_ALL, COMMAND_COUNT_MAX, LinkFilter, Packet, PacketSimulator,
+ PacketSimulatorOptions, ProcessId,
};
-use iggy_binary_protocol::GenericHeader;
+use iggy_binary_protocol::{Command2, GenericHeader};
use server_common::Message;
/// Network layer for the cluster simulation.
@@ -164,4 +165,17 @@ impl Network {
pub fn packets_in_flight(&self) -> usize {
self.simulator.packets_in_flight()
}
+
+ /// Packets delivered so far, per [`Command2`](iggy_binary_protocol::Command2)
+ /// discriminant. Names come from [`COMMAND_LABELS`].
+ #[must_use]
+ pub const fn command_counts(&self) -> &[u64; COMMAND_COUNT_MAX] {
+ self.simulator.command_counts()
+ }
+
+ /// Whether any packet of this command reached its destination.
+ #[must_use]
+ pub const fn delivered_any(&self, command: Command2) -> bool {
+ self.simulator.delivered_any(command)
+ }
}
diff --git a/core/simulator/src/packet.rs b/core/simulator/src/packet.rs
index 093615aa9..c01016c68 100644
--- a/core/simulator/src/packet.rs
+++ b/core/simulator/src/packet.rs
@@ -241,8 +241,62 @@ pub struct PacketSimulator {
auto_partition_nodes: Vec<usize>,
/// Reusable buffer for delivered packets.
delivered: Vec<Packet>,
+ /// Packets actually delivered, per [`Command2`] discriminant.
+ ///
+ /// Counted at delivery, past every drop path, so a command shows up only if a
+ /// process really received one. This is how a run answers which parts of the
+ /// protocol it exercised: the harness reaches far more of the command space
+ /// than any single scenario drives, and a test asserting a command was
+ /// observed is the difference between covering a path and merely compiling it.
+ command_counts: [u64; COMMAND_COUNT_MAX],
}
+/// One past the highest [`Command2`] discriminant, sizing [`COMMAND_LABELS`] and
+/// the delivery counters. Raising it is part of adding a command.
+pub const COMMAND_COUNT_MAX: usize = 30;
+
+/// Names for each [`Command2`] discriminant, so a coverage report reads as
+/// protocol rather than as integers. Indexed by discriminant; the trailing
+/// assert keeps it aligned with the enum.
+pub const COMMAND_LABELS: [&str; COMMAND_COUNT_MAX] = [
+ "Reserved",
+ "Ping",
+ "Pong",
+ "PingClient",
+ "PongClient",
+ "Request",
+ "Prepare",
+ "PrepareOk",
+ "Reply",
+ "Commit",
+ "StartViewChange",
+ "DoViewChange",
+ "StartView",
+ "Eviction",
+ "ReplicaHello",
+ "ReplicaChallenge",
+ "ReplicaFinish",
+ "RequestStartView",
+ "RequestPrepares",
+ "RepairPrepare",
+ "RepairDone",
+ "RangeEvicted",
+ "RequestStateTransfer",
+ "StateTransferTarget",
+ "RequestStateChunk",
+ "StateChunk",
+ "ForwardRegister",
+ "ForwardRegisterResult",
+ "ForwardLogout",
+ "ForwardLogoutResult",
+];
+
+const _: () = {
+ // Adding a command without extending the table above would silently report it
+ // under the wrong name, or index past the end.
+ assert!(Command2::ForwardLogoutResult as usize == COMMAND_COUNT_MAX - 1);
+};
+
impl PacketSimulator {
/// Create a new packet simulator.
///
@@ -322,6 +376,7 @@ impl PacketSimulator {
auto_partition_stability: initial_stability,
auto_partition_nodes: (0..node_count).collect(),
delivered: Vec::new(),
+ command_counts: [0; COMMAND_COUNT_MAX],
}
}
@@ -524,6 +579,7 @@ impl PacketSimulator {
delivered,
max_processes,
next_index,
+ command_counts,
..
} = self;
@@ -580,6 +636,7 @@ impl PacketSimulator {
tracing::trace!("packet replayed");
}
+ command_counts[command as usize] += 1;
delivered.push(packet);
}
}
@@ -588,6 +645,20 @@ impl PacketSimulator {
std::mem::take(&mut self.delivered)
}
+ /// Packets delivered so far, per [`Command2`] discriminant. See
+ /// [`Self::command_counts`]'s field docs for why this counts at delivery.
+ #[must_use]
+ pub const fn command_counts(&self) -> &[u64; COMMAND_COUNT_MAX] {
+ &self.command_counts
+ }
+
+ /// Whether any packet of this command has been delivered. The question a
+ /// coverage assertion actually asks.
+ #[must_use]
+ pub const fn delivered_any(&self, command: Command2) -> bool {
+ self.command_counts[command as usize] > 0
+ }
+
/// Return a previously taken buffer for reuse.
pub fn recycle_buffer(&mut self, mut buf: Vec<Packet>) {
buf.clear();
(iggy) 02/18: feat(simulator): select the fuzzer op mix and surface server diagnostics
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c0d9a2104ac18c8eb02b7252cad5561fb5cf3ecb
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 14:29:42 2026 +0530
feat(simulator): select the fuzzer op mix and surface server diagnostics
The fuzzer hardcoded `SendMessages: 100` and its module docs claimed
metadata and mixed workloads were gated on the metadata request-gap.
That gap was a `SimClient` numbering bug, fixed when the client began
numbering requests per plane, so the gate is gone and the doc was stale.
`--plane` now picks between partition-only, metadata-only, the mixed
default and uniform-across-every-action, as named `ActionWeights`
presets rather than a table spelled out at each call site. `uniform`
spreads the remainder rather than requiring the action count to divide
100, so appending an action never breaks it. `--ack-quorum-ratio`
exposes the knob that decides whether an offset store takes the
replicated path or the leader-local one.
The binary also installs a tracing subscriber, `RUST_LOG`-gated and off
by default. Server-side diagnostics are the only record of a request
dropped after logging, which is precisely the shape that wedges a
client's in-flight slot, and without a subscriber such a run reads as an
unexplained stall.
---
Cargo.lock | 1 +
core/simulator/Cargo.toml | 1 +
core/simulator/src/bin/workload-fuzz.rs | 66 ++++++++++++++++++++++++------
core/simulator/src/workload/options.rs | 72 +++++++++++++++++++++++++++++++++
4 files changed, 127 insertions(+), 13 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 9a69b24d1..14607f416 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -12191,6 +12191,7 @@ dependencies = [
"shard",
"strum 0.28.0",
"tracing",
+ "tracing-subscriber",
]
[[package]]
diff --git a/core/simulator/Cargo.toml b/core/simulator/Cargo.toml
index 37df9a52e..adc3582a0 100644
--- a/core/simulator/Cargo.toml
+++ b/core/simulator/Cargo.toml
@@ -48,6 +48,7 @@ server_common = { path = "../server_common", features = ["simulator"] }
shard = { path = "../shard", features = ["simulator"] }
strum = { workspace = true }
tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
[lints.clippy]
enum_glob_use = "deny"
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index fc495f8af..e3eebfb50 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -24,18 +24,15 @@
//!
//! ```text
//! workload-fuzz [--seed N] [--ticks N] [--clients N] [--replicas N]
+//! [--plane partition|metadata|mixed|uniform]
//! [--crash-prob F] [--no-quiesce]
//! ```
//!
-//! The default workload is partition-plane (`SendMessages`): it drains and
-//! converges. Metadata and mixed-plane workloads are gated on the metadata
-//! request-gap: a client's replicated-metadata request ids must arrive
-//! contiguously (`committed + 1`), so a dropped or reordered metadata request
-//! opens a permanent `RequestGap` that wedges that client's metadata plane.
-//! Broader op coverage lands once the workload generator models that
-//! constraint.
-
-use clap::Parser;
+//! `--plane` selects the op mix (see [`ActionWeights`]). Partition-plane runs
+//! drain and converge most readily; `uniform` is the widest per-tick op
+//! coverage.
+
+use clap::{Parser, ValueEnum};
use iggy_common::IggyByteSize;
use server_common::sharding::IggyNamespace;
use server_common::{MemoryPool, MemoryPoolConfigOther};
@@ -59,12 +56,43 @@ struct Args {
clients: u8,
#[arg(long, default_value_t = 3, value_parser = clap::value_parser!(u8).range(1..))]
replicas: u8,
+ /// Op mix to draw from.
+ #[arg(long, value_enum, default_value_t = Plane::Partition)]
+ plane: Plane,
+ /// Probability a consumer-offset store asks for `Quorum` rather than
+ /// `NoAck`. `1.0` keeps every offset op on the replicated path.
+ #[arg(long, default_value_t = 0.5, value_parser = parse_unit_interval)]
+ ack_quorum_ratio: f32,
#[arg(long, default_value_t = 0.0, value_parser = parse_unit_interval)]
crash_prob: f32,
#[arg(long)]
no_quiesce: bool,
}
+/// Which plane the sampled ops target. Maps onto an [`ActionWeights`] preset.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
+enum Plane {
+ /// Writes and consumer offsets only.
+ Partition,
+ /// Replicated metadata mutations only.
+ Metadata,
+ /// Stream creates over a write-heavy base.
+ Mixed,
+ /// Every action equally likely.
+ Uniform,
+}
+
+impl Plane {
+ fn weights(self) -> ActionWeights {
+ match self {
+ Self::Partition => ActionWeights::partition_only(),
+ Self::Metadata => ActionWeights::metadata_only(),
+ Self::Mixed => ActionWeights::default(),
+ Self::Uniform => ActionWeights::uniform(),
+ }
+ }
+}
+
/// Clap value parser: accept a probability in `[0.0, 1.0]`.
fn parse_unit_interval(raw: &str) -> Result<f32, String> {
let value: f32 = raw
@@ -80,12 +108,23 @@ fn parse_unit_interval(raw: &str) -> Result<f32, String> {
fn main() {
let args = Args::parse();
+ // Server-side diagnostics (`emit_partition_diag` and friends) are the only
+ // record of a request the server dropped after logging, which is exactly the
+ // shape that wedges a client's in-flight slot. Without a subscriber they go
+ // nowhere and the run looks like an unexplained stall, so install one and let
+ // `RUST_LOG` select. Off by default: a WARN per dropped frame drowns the run.
+ tracing_subscriber::fmt()
+ .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+ .with_writer(std::io::stderr)
+ .init();
+
// A provided seed reproduces a prior run exactly; otherwise draw one and
// log it. Both the network and workload PRNGs derive from it.
let seed = args.seed.unwrap_or_else(rand::random);
let ticks = args.ticks;
let clients = args.clients;
let replicas = args.replicas;
+ let plane = args.plane;
let crash_prob = args.crash_prob;
let quiesce = !args.no_quiesce;
@@ -97,7 +136,7 @@ fn main() {
println!(
"workload-fuzz: seed={seed} ticks={ticks} clients={clients} replicas={replicas} \
- crash_prob={crash_prob} quiesce={quiesce}"
+ plane={plane:?} crash_prob={crash_prob} quiesce={quiesce}"
);
// poll_messages / reply paths panic without an initialized pool; disabled
@@ -131,7 +170,8 @@ fn main() {
let mut options = WorkloadOptions::new(seed, replicas, vec![ns]);
options.client_count = clients;
options.crash_per_tick_ratio = crash_prob;
- options.weights = ActionWeights::new(&[(Action::SendMessages, 100)]);
+ options.ack_quorum_ratio = args.ack_quorum_ratio;
+ options.weights = plane.weights();
let mut workload = Workload::new(options);
let replies = run(&mut sim, &mut workload, &sim_clients, ticks, u64::MAX);
@@ -146,8 +186,8 @@ fn main() {
println!("quiesced and converged (leader-relative + entity oracle)");
} else {
println!(
- "WARN: did not quiesce within budget — expected when crashing to bare quorum \
- or under the metadata request-gap limitation; per-tick invariants still held"
+ "WARN: did not quiesce within budget — expected when crashing to bare quorum; \
+ per-tick invariants still held"
);
}
}
diff --git a/core/simulator/src/workload/options.rs b/core/simulator/src/workload/options.rs
index a6c0bbece..a4a29447a 100644
--- a/core/simulator/src/workload/options.rs
+++ b/core/simulator/src/workload/options.rs
@@ -47,6 +47,78 @@ impl ActionWeights {
Self { weights }
}
+ /// Partition plane only: writes plus consumer-offset traffic, no metadata
+ /// mutation. The regime that drains and converges most readily, so it is
+ /// what a run reaches for when the question is about replication rather
+ /// than about the state machine.
+ #[must_use]
+ pub fn partition_only() -> Self {
+ Self::new(&[
+ (Action::SendMessages, 60),
+ (Action::StoreConsumerOffset2, 25),
+ (Action::DeleteConsumerOffset2, 5),
+ (Action::StoreConsumerOffset, 7),
+ (Action::DeleteConsumerOffset, 3),
+ ])
+ }
+
+ /// Metadata plane only: every replicated metadata mutation, weighted so
+ /// creates outrun deletes and the shadow keeps a live population to sample
+ /// duplicate- and missing-target outcomes against. `DeleteSegments` is
+ /// excluded: it resolves against partition state the partition-plane
+ /// presets build, so it belongs to a mixed run.
+ #[must_use]
+ pub fn metadata_only() -> Self {
+ Self::new(&[
+ (Action::CreateStream, 12),
+ (Action::UpdateStream, 6),
+ (Action::DeleteStream, 6),
+ (Action::PurgeStream, 4),
+ (Action::CreateTopic, 12),
+ (Action::UpdateTopic, 6),
+ (Action::DeleteTopic, 6),
+ (Action::PurgeTopic, 4),
+ (Action::CreatePartitions, 6),
+ (Action::DeletePartitions, 4),
+ (Action::CreateConsumerGroup, 6),
+ (Action::DeleteConsumerGroup, 4),
+ (Action::CreateUser, 6),
+ (Action::UpdateUser, 3),
+ (Action::DeleteUser, 3),
+ (Action::ChangePassword, 3),
+ (Action::UpdatePermissions, 3),
+ (Action::CreatePersonalAccessToken, 3),
+ (Action::DeletePersonalAccessToken, 3),
+ ])
+ }
+
+ /// Every action equally likely. Widest op coverage per tick, at the cost of
+ /// a shallow population per entity kind.
+ ///
+ /// `Action::COUNT` does not divide 100, so the first `100 % COUNT` actions
+ /// carry one extra point. Spread that way rather than asserting the count
+ /// divides evenly, so appending an `Action` never breaks this preset.
+ #[must_use]
+ pub fn uniform() -> Self {
+ use strum::IntoEnumIterator;
+
+ let count = u32::try_from(Action::COUNT).expect("Action::COUNT fits u32");
+ let base = 100 / count;
+ let remainder = 100 % count;
+ let entries: Vec<(Action, u8)> = Action::iter()
+ .enumerate()
+ .map(|(idx, action)| {
+ let extra = u32::try_from(idx).expect("action index fits u32") < remainder;
+ let weight = base + u32::from(extra);
+ (
+ action,
+ u8::try_from(weight).expect("per-action weight is at most 100"),
+ )
+ })
+ .collect();
+ Self::new(&entries)
+ }
+
#[must_use]
pub const fn weight(&self, action: Action) -> u8 {
self.weights[action as usize]
(iggy) 08/18: fix(simulator): skip re-materialising a namespace the workload deleted
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 495db105cc0272d2c867ef9ea89c733220e1eaaf
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Aug 14 23:53:12 2026 +0530
fix(simulator): skip re-materialising a namespace the workload deleted
`materialise_partition` seeded its namespace into committed metadata and
then unwrapped the committed revision, which held only while nothing ever
removed the stream or topic behind it. A metadata workload does remove
them, and the seed's `CreatePartitions` is then a committed REJECTION
rather than an error, so it reports nothing and the unwrap panicked on the
next restart that re-materialised the group.
Skip the group instead, and decide it before building anything: a rebooted
server does not re-open a deleted partition's directory either, so a
skipped group should leave behind neither a partition nor a routing row.
Surfaced by metadata-plane crash/restart runs, which previously wedged
before ever reaching a restart that re-materialised a deleted namespace.
---
core/simulator/src/lib.rs | 25 ++++++++++++++++---------
1 file changed, 16 insertions(+), 9 deletions(-)
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 7d0ce2413..5d8a42f19 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -1032,6 +1032,22 @@ impl Simulator {
fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace) {
let shard_count = u32::try_from(replica.shards.len()).expect("shard count fits u32");
let owner = calculate_shard_assignment(&namespace, shard_count);
+ // Commit the namespace before anything else: a partition the metadata plane
+ // never heard of is a shape production cannot produce, and the shard refuses
+ // to serve client traffic whose routing-row epoch it cannot match against a
+ // committed `created_revision`.
+ let streams = replica.shards[0].plane.metadata().mux_stm.streams();
+ streams.seed_namespace(namespace, namespace.inner());
+ // No committed revision means the seed could not re-add the namespace, which
+ // happens once a metadata workload has deleted its stream or topic: the seed's
+ // `CreatePartitions` is then a committed REJECTION, not an error, so it
+ // reports nothing. Skip the group rather than build a partition no committed
+ // metadata names -- a rebooted server does not re-open a deleted partition's
+ // directory either. Checked before the build so a skipped group leaves
+ // neither a partition nor a routing row behind.
+ let Some(epoch) = streams.created_revision_for_namespace(namespace) else {
+ return;
+ };
// One store per group, minted on first materialisation and reused on every
// later one, so the recorded view survives a replica restart.
let superblock = Rc::clone(
@@ -1055,15 +1071,6 @@ fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace) {
recovered_state,
retained,
);
- // Commit the namespace before stamping the rows: a partition the metadata
- // plane never heard of is a shape production cannot produce, and the shard
- // refuses to serve client traffic whose routing-row epoch it cannot match
- // against a committed `created_revision`.
- let streams = replica.shards[0].plane.metadata().mux_stm.streams();
- streams.seed_namespace(namespace, namespace.inner());
- let epoch = streams
- .created_revision_for_namespace(namespace)
- .expect("namespace committed by the seed above");
for shard in &replica.shards {
shard.shards_table().insert(
namespace,
(iggy) 14/18: chore(simulator): record tempfile dev-dependency in the lockfile
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 1b5bb4a2b37042b2a7aed9df4fe0873d38df5eed
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 11:12:44 2026 +0530
chore(simulator): record tempfile dev-dependency in the lockfile
---
Cargo.lock | 1 +
1 file changed, 1 insertion(+)
diff --git a/Cargo.lock b/Cargo.lock
index 14607f416..96242459a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -12190,6 +12190,7 @@ dependencies = [
"server_common",
"shard",
"strum 0.28.0",
+ "tempfile",
"tracing",
"tracing-subscriber",
]
(iggy) 10/18: feat(simulator): drive the workload through the dispatch shell
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 0593626672b0d142c41bc7aceced69bef2f22aa5
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 01:33:26 2026 +0530
feat(simulator): drive the workload through the dispatch shell
The shell wires the server's real dispatch handlers, and it was reachable
only from two hand-written tests. Authorization, session binding and the
pre-commit deny replies exist nowhere else -- the raw `on_message` path
has no deny site at all -- so none of them had ever seen a workload.
Running one immediately found that the workload could not read a denial.
`ReplyHeader::status` nonzero means a request refused before commit, and
its body is EMPTY by contract, mutually exclusive with the metadata result
section. The reply decoder read a result section unconditionally, found no
bytes, and reported the reply as corrupt. Denials are now their own
outcome: the op never entered the log, so the shadow does not move and
there is nothing to classify.
Denials are counted per action, not just totalled, because the two causes
need telling apart. An op the server refuses for this particular input
looks like a mix of commits and denials; an op whose wire shape the
dispatch layer cannot decode is denied every single time. The consumer
offset ops are the first kind (`ResourceNotFound`, offsets and consumers
the workload has not created yet). Both PAT ops are the second: zero
commits and `InvalidCommand` on every attempt, so `SimClient` builds a
shape the real path rejects, the same class of bug as the consumer kind
fixed earlier. Left for the fix pass.
The harness also assumed every client-addressed frame is a reply, so an
`Eviction` -- a command nothing had exercised before -- failed to decode
and surfaced as "wrong command type". Evictions are now classified and
recorded. The driver refuses to continue on one, since a lost session
makes every outstanding request unanswerable and modelling recovery means
re-establishing the session mid-run and discarding the auditor's
expectations for that client. Naming it beats presenting as a stall.
Shell runs drain and converge 12 of 12 across four op mixes and three
seeds. Shell with crashes and restarts hits the eviction limitation
instead, which is now a clear message rather than a decode panic. The raw
path is unchanged: 12 of 12 with crash and restart injection.
---
core/simulator/src/bin/workload-fuzz.rs | 51 +++++++++++----
core/simulator/src/lib.rs | 112 +++++++++++++++++++++++++++++++-
core/simulator/src/workload/auditor.rs | 26 ++++++++
core/simulator/src/workload/mod.rs | 44 +++++++++++++
4 files changed, 220 insertions(+), 13 deletions(-)
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index c41d5be9f..fd918e483 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -82,6 +82,12 @@ struct Args {
/// Crash the primary too, putting a view change under live traffic.
#[arg(long)]
crash_primary: bool,
+ /// Route every client request through the server's real dispatch handlers
+ /// instead of the raw `on_message` fast path. Clients then log in against the
+ /// seeded root user and carry a bound session, so the run also covers
+ /// authorization and session lifecycle, which exist only on this path.
+ #[arg(long)]
+ shell: bool,
#[arg(long)]
no_quiesce: bool,
@@ -356,8 +362,8 @@ fn main() {
let network_opts = network_options(&args, replicas, clients, seed);
println!(
"workload-fuzz: seed={seed} ticks={ticks} clients={clients} replicas={replicas} \
- plane={plane:?} faults={:?} crash_prob={crash_prob} quiesce={quiesce}",
- args.faults,
+ plane={plane:?} faults={:?} shell={} crash_prob={crash_prob} quiesce={quiesce}",
+ args.faults, args.shell,
);
println!(
"network: loss={} replay={} delay={}..{} partition={:?}/{:?} \
@@ -383,17 +389,38 @@ fn main() {
});
let client_ids: Vec<u128> = (1..=u128::from(clients)).collect();
- let mut sim = Simulator::new(
- usize::from(replicas),
- client_ids.iter().copied(),
- network_opts,
- );
+ let mut sim = if args.shell {
+ Simulator::with_shards_shell(
+ usize::from(replicas),
+ 1,
+ client_ids.iter().copied(),
+ network_opts,
+ )
+ } else {
+ Simulator::new(
+ usize::from(replicas),
+ client_ids.iter().copied(),
+ network_opts,
+ )
+ };
let sim_clients: Vec<SimClient> = client_ids.iter().map(|&id| SimClient::new(id)).collect();
let ns = IggyNamespace::new(1, 1, 0);
sim.init_partition(ns);
+ if args.shell {
+ // The shell resolves a partition request's namespace against committed
+ // metadata, so the stream and topic behind it have to exist as well as the
+ // partition group.
+ sim.seed_stream_topic_partition(ns);
+ }
for client in &sim_clients {
- sim.register_client_with_primary(client);
+ if args.shell {
+ // Log in rather than bare-register: the dispatch path admits a request
+ // only from a bound session, and the login is what mints one.
+ sim.shell_login(client);
+ } else {
+ sim.register_client_with_primary(client);
+ }
}
let mut options = WorkloadOptions::new(seed, replicas, vec![ns]);
@@ -461,17 +488,19 @@ fn print_coverage(workload: &Workload) {
let stats = workload.auditor.stats();
println!(
"coverage: replies_seen={} replies_unknown={} committed_rejections={} \
- samples_none={} resends={}",
+ samples_none={} resends={} denials={}",
stats.replies_seen,
stats.replies_unknown,
stats.committed_rejections,
workload.samples_none(),
workload.resends(),
+ stats.denials,
);
for action in Action::iter() {
let commits = stats.commits(action);
- if commits > 0 {
- println!(" {action:?}: {commits} commits");
+ let (denied, status) = stats.denials_per_action[action as usize];
+ if commits > 0 || denied > 0 {
+ println!(" {action:?}: {commits} commits, {denied} denied (last status {status})");
}
}
}
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 6223a4f42..6f0ecef54 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -32,7 +32,7 @@ use deps::SimClock;
use deps::SimSuperblock;
use deps::{MemStorage, SimJournal};
use executor::{DetExecutor, RunOutcome, TaskId};
-use iggy_binary_protocol::{GenericHeader, ReplyHeader};
+use iggy_binary_protocol::{Command2, GenericHeader, ReplyHeader};
use iggy_common::IggyError;
use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
use metadata::impls::metadata::StreamsFrontend;
@@ -176,6 +176,14 @@ pub struct Simulator {
entry_rng: Xoshiro256Plus,
/// Network seed, kept for livelock diagnostics.
seed: u64,
+ /// Clients the cluster has evicted since the last drain, in delivery order.
+ ///
+ /// An eviction ends a session: every request that client has outstanding will
+ /// never be answered, and it must log in again before it can submit anything.
+ /// Recorded rather than acted on here, because re-establishing a session needs
+ /// to step the simulator and drop the workload's expectations for that client,
+ /// neither of which belongs inside packet delivery.
+ evicted: Vec<u128>,
/// Dispatch-shell mode: when set, inbound client packets are delivered
/// through the real `on_client_request` handler (see
/// [`shard::IggyShard::deliver_client_request`]) instead of the raw
@@ -388,6 +396,7 @@ impl Simulator {
client_ids,
executor,
entry_rng: Xoshiro256Plus::seed_from_u64(seed ^ ENTRY_SHARD_SEED_SALT),
+ evicted: Vec::new(),
seed,
shell,
}
@@ -564,7 +573,17 @@ impl Simulator {
}
// Crashed or missing: packet silently dropped.
}
- ProcessId::Client(_) => {
+ ProcessId::Client(client_id) => {
+ // Not every client-addressed frame is a reply. `Eviction`
+ // tells a client its session is gone, which the server sends
+ // once the client table drops it -- reachable as soon as the
+ // dispatch shell runs with crashes and restarts. Decoding it
+ // as a reply fails on the command discriminant, so classify
+ // first and record the eviction for the driver.
+ if packet.message.header().command == Command2::Eviction {
+ self.evicted.push(client_id);
+ continue;
+ }
let reply: Message<ReplyHeader> = packet
.message
.deep_copy()
@@ -609,6 +628,15 @@ impl Simulator {
client_replies
}
+ /// Take the clients evicted since the last call.
+ ///
+ /// A driver must consume these: an evicted client's session is gone, so its
+ /// outstanding requests are unanswerable and its next request is refused
+ /// until it logs in again. Ignoring them looks exactly like a wedge.
+ pub fn take_evictions(&mut self) -> Vec<u128> {
+ std::mem::take(&mut self.evicted)
+ }
+
/// Rolling hash of the executor schedule (every poll and timer fire).
/// Two runs from the same seed and inputs must agree; determinism
/// tests assert on it alongside the reply-trace hash.
@@ -3244,6 +3272,86 @@ mod tests {
/// pipeline to `PIPELINE_PREPARE_QUEUE_MAX`; a request on `ns_b`
/// still commits while `ns_a` is wedged (no quorum without backup
/// acks); lifting the block drains `ns_a` completely.
+ /// The workload drains and converges when every request goes through the
+ /// server's real dispatch handlers rather than the raw `on_message` path.
+ ///
+ /// Worth its own test because the shell path is where authorization, session
+ /// binding and the pre-commit deny replies live; the raw path has no deny site
+ /// at all. Running the workload here is the only thing that exercises them,
+ /// and it is what surfaced that the workload had never modelled a denial:
+ /// `ReplyHeader::status` nonzero means an empty body, and the reply decoder
+ /// was reading a result section off it and calling the reply corrupt.
+ ///
+ /// Asserts denials were actually observed, so the test cannot pass by taking
+ /// a path where nothing is ever denied and thus proving nothing about the
+ /// shell.
+ #[test]
+ fn shell_workload_drains_and_converges() {
+ use crate::workload::{
+ self, Workload,
+ options::{ActionWeights, WorkloadOptions},
+ oracle,
+ };
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let seed = 0x5E11_0001;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::with_shards_shell(
+ usize::from(replica_count),
+ 1,
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let ns = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns);
+ // The dispatch path resolves a partition request's namespace against
+ // committed metadata, so the stream and topic have to exist too.
+ sim.seed_stream_topic_partition(ns);
+
+ let client = SimClient::new(client_id);
+ // Log in rather than bare-register: dispatch admits a request only from a
+ // bound session.
+ sim.shell_login(&client);
+
+ let mut options = WorkloadOptions::new(seed, replica_count, vec![ns]);
+ options.weights = ActionWeights::uniform();
+ let mut wl = Workload::new(options);
+
+ let clients = [client];
+ let replies = workload::run(&mut sim, &mut wl, &clients, 4_000, u64::MAX);
+ assert!(replies > 0, "shell workload produced no replies");
+
+ let stats = wl.auditor.stats();
+ assert!(
+ stats.commits_per_action.iter().sum::<u64>() > 0,
+ "shell workload committed nothing, so the dispatch path never got past \
+ admission"
+ );
+ assert!(
+ stats.denials > 0,
+ "no request was denied, so the pre-commit deny path this test exists to \
+ cover never ran"
+ );
+
+ assert!(
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 20_000),
+ "{}",
+ oracle::quiesce_failure_report(&sim, &wl),
+ );
+ oracle::assert_converged(&sim, &wl);
+ }
+
/// The cross-replica equality check actually compares replicas against each
/// other, and holds over a metadata workload with crashes and restarts.
///
diff --git a/core/simulator/src/workload/auditor.rs b/core/simulator/src/workload/auditor.rs
index 411a83af7..903117935 100644
--- a/core/simulator/src/workload/auditor.rs
+++ b/core/simulator/src/workload/auditor.rs
@@ -58,6 +58,22 @@ pub struct AuditorStats {
/// rejection). The shadow does not mutate on these; in a serial run the
/// `on_reply` equality oracle asserts the rejection was the targeted outcome.
pub committed_rejections: u64,
+ /// Replies denied before commit, carrying `ReplyHeader::status` and an empty
+ /// body. Distinct from a committed rejection: the op never entered the log,
+ /// so the shadow must not move and no result section exists to classify.
+ /// Only the dispatch shell produces these (authorization runs there); the raw
+ /// path never denies.
+ pub denials: u64,
+ /// Per-action denial counter and the last status seen for it, indexed by
+ /// `Action as usize`.
+ ///
+ /// Per-action rather than a single total because the two causes need telling
+ /// apart: an op the server legitimately refuses for this input (an offset the
+ /// partition cannot accept yet) versus an op the dispatch layer cannot decode
+ /// at all, which means the workload builds a wire shape the real path rejects.
+ /// The second is a workload bug and shows up as every request for that action
+ /// being denied with the same status.
+ pub denials_per_action: [(u64, u32); Action::COUNT],
}
impl Default for AuditorStats {
@@ -67,6 +83,8 @@ impl Default for AuditorStats {
replies_unknown: 0,
commits_per_action: [0u64; Action::COUNT],
committed_rejections: 0,
+ denials: 0,
+ denials_per_action: [(0, 0); Action::COUNT],
}
}
}
@@ -174,6 +192,14 @@ impl ServerAuditor {
/// Record a committed business rejection (nonzero result code). Either
/// targeted by outcome-first generation (duplicate name, fabricated missing
/// entity) or produced by a race.
+ /// Record a pre-commit denial (`ReplyHeader::status` nonzero).
+ pub const fn note_denial(&mut self, action: Action, status: u32) {
+ self.stats.denials += 1;
+ let entry = &mut self.stats.denials_per_action[action as usize];
+ entry.0 += 1;
+ entry.1 = status;
+ }
+
pub const fn note_committed_rejection(&mut self) {
self.stats.committed_rejections += 1;
}
diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs
index 164367084..564082bfd 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -315,6 +315,23 @@ impl Workload {
OnReply::Unknown => return Vec::new(),
};
+ // A pre-commit denial short-circuits everything below. `ReplyHeader`'s
+ // contract makes the two channels mutually exclusive: a reply either
+ // commits (status 0, result section present) or is denied before commit
+ // (status set, EMPTY body). Reading a result section off a denial finds
+ // no bytes, which the metadata branch below would report as a corrupt
+ // reply.
+ //
+ // The op never entered the log, so the shadow must not move and there is
+ // nothing to classify. Only the dispatch shell produces these, since
+ // authorization runs there; the raw path has no denial site at all, which
+ // is why this went unmodelled until the workload ran through the shell.
+ if header.status != 0 {
+ self.auditor.note_denial(entry.action, header.status);
+ self.release_outstanding(key);
+ return Vec::new();
+ }
+
// Decode the committed result code. Metadata replies carry a
// result section (see `ApplyReply::to_reply_body`); partition-plane
// replies do not, hence the `is_metadata` gate.
@@ -524,6 +541,7 @@ pub fn run_with_faults(
apply_sim_commands(sim, &cmds);
replies_seen += 1;
}
+ assert_no_evictions(sim);
invariants.check(sim, workload);
if replies_seen >= replies_target {
break;
@@ -669,6 +687,32 @@ impl FaultInjector {
}
}
+/// Fail loudly if the cluster evicted a client, which the workload cannot yet
+/// survive.
+///
+/// An eviction ends the session: the client's outstanding requests become
+/// unanswerable and it must log in again before submitting anything. Modelling
+/// that means re-establishing the session mid-run and discarding the auditor's
+/// expectations for it, which the driver does not do. Until it does, an eviction
+/// presents as a client that has silently stopped making progress, so name it
+/// here rather than let the run time out with no explanation.
+///
+/// Only reachable through the dispatch shell, and in practice only once crashes
+/// and restarts are also in play.
+///
+/// # Panics
+/// If any client was evicted since the last step.
+fn assert_no_evictions(sim: &mut Simulator) {
+ let evicted = sim.take_evictions();
+ assert!(
+ evicted.is_empty(),
+ "cluster evicted client(s) {evicted:?}: their sessions are gone, so their \
+ outstanding requests can never be answered and their next request is \
+ refused. The workload does not re-establish a session, so the run cannot \
+ continue"
+ );
+}
+
/// Submit every request whose reply is overdue (see [`Workload::due_resends`]).
///
/// The client id rides the retained message's header, so a resend re-enters the
(iggy) 09/18: feat(simulator): assert cross-replica committed log equality
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch sim-workload-faults
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 06e11a116147e102119d3a86cce220bc51cc73ac
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 15 00:10:03 2026 +0530
feat(simulator): assert cross-replica committed log equality
The harness compared each replica against itself (offsets and views never
regress) and against the workload's shadow, but never against the OTHER
replicas, which is the actual consensus property: two replicas that both
committed op N must have committed the same op N. A divergence was
invisible as long as every replica stayed internally consistent.
`state_checker` follows TigerBeetle's `testing/cluster/state_checker.zig`
rather than inventing a shape: one canonical commit chain, every replica
asserted to agree with it wherever they overlap, and the chain asserted to
be hash-linked so a replica cannot match op-by-op while having been built
on a different prefix. It runs per tick alongside the other invariants, so
a divergence is reported at the tick it appears rather than at the next
quiesce.
Equality is over the committed PREFIX, not equal heads: a replica that
missed the last commit broadcast or rejoined recently may legitimately
trail, and requiring equal heads would fail on ordinary lag while saying
nothing more about safety.
Two oracle checks were unsound once primaries can be crashed, and both
failed on correct clusters rather than catching anything:
The leader-relative partition check measured `commit_offset`, which is the
highest durably PERSISTED message offset, not a committed watermark. A
backup that persisted an op the view then settled below is ordinary VSR.
It now measures the group's consensus `commit_min`. The check also needs a
settled view to resolve a leader at all, so `settle_to_stable_view` waits
for one agreed view on both planes before asserting, instead of guessing
at a leader mid-view-change.
The entity oracle filtered committed topics by their STREAM's name, so the
harness's own `sim-topic-*` fillers counted as workload state once they
landed inside a workload-created stream, and the shadow was blamed for not
predicting them. Every level now filters on its own name.
Non-vacuity is asserted, not assumed: the test requires ops witnessed on
more than one replica, since an equality oracle that never finds a shared
op passes in silence. A 4000-tick metadata run with crashes and restarts
compares all 134 ops of its chain.
Crash/restart sweeps: 24 of 24 without primary crashes, 17 of 20 with
`--crash-primary`. The three remaining failures are production assertions
(`advance_commit_min` sequentiality, and a `transmute_header` validity
check), not oracle disagreements.
---
core/simulator/src/bin/workload-fuzz.rs | 9 ++
core/simulator/src/lib.rs | 131 ++++++++++++++++
core/simulator/src/workload/invariants.rs | 19 ++-
core/simulator/src/workload/mod.rs | 1 +
core/simulator/src/workload/oracle.rs | 222 ++++++++++++++++++++++-----
core/simulator/src/workload/state_checker.rs | 219 ++++++++++++++++++++++++++
6 files changed, 559 insertions(+), 42 deletions(-)
diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs
index c7abfecc5..c41d5be9f 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -437,6 +437,15 @@ fn main() {
"{}",
oracle::quiesce_failure_report(&sim, &workload),
);
+ // Then wait for one agreed view before asserting. `assert_converged`
+ // resolves the leader as whichever live replica claims to be primary, so
+ // asserting mid-view-change either finds none or finds a deposed one --
+ // false failures rather than divergences.
+ assert!(
+ oracle::settle_to_stable_view(&mut sim, &mut workload, 50_000),
+ "metadata views never converged after the drain\n{}",
+ oracle::quiesce_failure_report(&sim, &workload),
+ );
oracle::assert_converged(&sim, &workload);
println!("quiesced and converged (leader-relative + entity oracle)");
// Again after the drain: the drain both answers outstanding requests and
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 5d8a42f19..6223a4f42 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -141,6 +141,19 @@ impl SimReplica {
}
}
+/// One replica's view of a partition group's consensus. Read by the quiesce
+/// oracle; see [`Simulator::partition_consensus_state`].
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct PartitionConsensusState {
+ pub status: consensus::Status,
+ pub view: u32,
+ pub is_primary: bool,
+ /// Ops this replica has committed in the group. The committed watermark, as
+ /// opposed to `PartitionOffsets::commit_offset`, which is the highest
+ /// durably PERSISTED message offset and so counts an uncommitted suffix too.
+ pub commit_min: u64,
+}
+
pub struct Simulator {
/// All replicas, indexed by replica id. Always fully populated — crashed
/// replicas are kept alive but skipped during dispatch.
@@ -997,6 +1010,29 @@ impl Simulator {
Some(u64::from(partition.consensus().view()))
}
+ /// One replica's view of a partition group's consensus, or `None` when that
+ /// replica does not host the namespace.
+ ///
+ /// Read by the quiesce oracle to decide whether a group has settled into one
+ /// view, which its leader-relative checks depend on once partition primaries
+ /// can be crashed.
+ #[must_use]
+ pub(crate) fn partition_consensus_state(
+ &self,
+ replica_idx: usize,
+ namespace: IggyNamespace,
+ ) -> Option<PartitionConsensusState> {
+ let shard = self.replicas[replica_idx].partition_shard(namespace);
+ let partition = shard.plane.partitions().get_by_ns(&namespace)?;
+ let consensus = partition.consensus();
+ Some(PartitionConsensusState {
+ status: consensus.status(),
+ view: consensus.view(),
+ is_primary: consensus.is_primary(),
+ commit_min: consensus.commit_min(),
+ })
+ }
+
/// Index of the current primary for `namespace`, as seen by the first live
/// replica hosting it, or `None` if no live replica hosts it.
///
@@ -3208,6 +3244,101 @@ mod tests {
/// pipeline to `PIPELINE_PREPARE_QUEUE_MAX`; a request on `ns_b`
/// still commits while `ns_a` is wedged (no quorum without backup
/// acks); lifting the block drains `ns_a` completely.
+ /// The cross-replica equality check actually compares replicas against each
+ /// other, and holds over a metadata workload with crashes and restarts.
+ ///
+ /// Non-vacuity is the point of the assertions on the chain. An equality
+ /// oracle that never finds two replicas at the same op passes in silence, so
+ /// a green run would say nothing: `ops_compared` counts only ops witnessed on
+ /// more than one replica, which is the subset that exercised the property.
+ #[test]
+ fn committed_metadata_agrees_across_replicas() {
+ use crate::workload::{
+ self, FaultInjector, Workload,
+ invariants::Invariants,
+ options::{ActionWeights, WorkloadOptions},
+ oracle,
+ };
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 5;
+ let client_id: u128 = 1;
+ let seed = 0x57A7_E000;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+ let ns = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns);
+ sim.register_client_with_primary(&client);
+
+ // Metadata ops, since the committed chain this checks is the metadata WAL.
+ // Crash and restart so replicas rejoin and repair, which is when a
+ // divergence would be introduced if one could be.
+ let mut options = WorkloadOptions::new(seed, replica_count, vec![ns]);
+ options.weights = ActionWeights::metadata_only();
+ options.crash_per_tick_ratio = 0.01;
+ options.restart_per_tick_ratio = 0.02;
+ let mut wl = Workload::new(options);
+
+ let clients = [client];
+ let mut injector = FaultInjector::new(seed, replica_count);
+ let mut invariants = Invariants::new();
+ // Driven here rather than through `workload::run` so the accumulated
+ // chain is readable afterwards; `run` builds its own `Invariants`.
+ for _ in 0..4_000u32 {
+ wl.tick();
+ injector.step(&mut sim, &wl);
+ workload::resubmit_due(&mut sim, &mut wl);
+ if let Some((target, msg)) = wl.build_request(&clients[0]) {
+ sim.submit_request(clients[0].client_id(), target, msg.into_generic());
+ }
+ for reply in sim.step() {
+ let cmds = wl.on_reply(&reply);
+ workload::apply_sim_commands(&mut sim, &cmds);
+ }
+ invariants.check(&sim, &wl);
+ }
+
+ assert!(
+ injector.restarts() > 0,
+ "no replica restarted, so rejoin and repair never ran"
+ );
+ let chain = invariants.state_checker();
+ assert!(
+ chain.chain_len() > 0,
+ "the canonical commit chain is empty: nothing was ever recorded"
+ );
+ assert!(
+ chain.ops_compared() > 0,
+ "no committed op was witnessed on two replicas, so the equality check \
+ never actually compared anything and would pass on a diverged cluster"
+ );
+
+ assert!(
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 50_000),
+ "{}",
+ oracle::quiesce_failure_report(&sim, &wl),
+ );
+ assert!(
+ oracle::settle_to_stable_view(&mut sim, &mut wl, 50_000),
+ "metadata views never converged after the drain"
+ );
+ oracle::assert_converged(&sim, &wl);
+ }
+
/// A restarted replica comes back with the partition data it had, so its
/// `commit_offset` does not regress.
///
diff --git a/core/simulator/src/workload/invariants.rs b/core/simulator/src/workload/invariants.rs
index d7f1b8a8d..c4f57095d 100644
--- a/core/simulator/src/workload/invariants.rs
+++ b/core/simulator/src/workload/invariants.rs
@@ -24,6 +24,7 @@
//! unchanged.
use crate::Simulator;
+use crate::workload::state_checker::StateChecker;
use crate::workload::{CLIENT_REQUEST_QUEUE_MAX, Workload};
use server_common::sharding::IggyNamespace;
use std::collections::HashMap;
@@ -34,6 +35,10 @@ use std::collections::HashMap;
pub struct Invariants {
commit_offset: HashMap<(u8, IggyNamespace), u64>,
view: HashMap<(u8, IggyNamespace), u64>,
+ /// Cross-replica committed-log agreement. Lives here so it runs every tick
+ /// like the rest: a divergence is reported at the tick it appears rather than
+ /// at the next quiesce, by which point the run has moved on.
+ state_checker: StateChecker,
}
impl Invariants {
@@ -49,7 +54,9 @@ impl Invariants {
/// - consensus `view` never regresses (a view change only advances it).
///
/// Globally:
- /// - total in-flight requests stay within the per-client queue ceiling.
+ /// - total in-flight requests stay within the per-client queue ceiling,
+ /// - live replicas agree on every committed metadata op they share, and the
+ /// committed chain stays hash-linked (see [`StateChecker`]).
///
/// Crashed replicas are skipped and their last-seen marks retained, which
/// holds across a restart for both quantities: the superblock carries `view`,
@@ -94,6 +101,8 @@ impl Invariants {
(client_count={}, queue_max={CLIENT_REQUEST_QUEUE_MAX}) (seed={seed:#x})",
workload.options.client_count,
);
+
+ self.state_checker.check(sim, seed);
}
/// Number of `(replica, namespace)` pairs observed so far. Used by tests to
@@ -103,6 +112,14 @@ impl Invariants {
pub(crate) fn tracked_pairs(&self) -> usize {
self.commit_offset.len()
}
+
+ /// The canonical committed chain built so far. Tests read it to prove the
+ /// equality check compared replicas against each other rather than passing
+ /// over an empty chain.
+ #[must_use]
+ pub const fn state_checker(&self) -> &StateChecker {
+ &self.state_checker
+ }
}
/// Panic if `cur < prev`. Pure so the catch logic is unit-testable without a
diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs
index 322d66c22..164367084 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -33,6 +33,7 @@ pub mod ops;
pub mod options;
pub mod oracle;
pub mod shadow;
+pub mod state_checker;
use crate::Simulator;
use crate::client::SimClient;
diff --git a/core/simulator/src/workload/oracle.rs b/core/simulator/src/workload/oracle.rs
index 567172421..b1eba8e5d 100644
--- a/core/simulator/src/workload/oracle.rs
+++ b/core/simulator/src/workload/oracle.rs
@@ -23,22 +23,22 @@
//!
//! - no live replica is ahead of the leader on any namespace (a backup ahead of
//! the leader is a split-brain / divergence bug),
+//! - every live replica agrees with every other on each committed metadata op
+//! they both hold, the real consensus property (see
+//! [`super::state_checker`]),
//! - on a serial run, the workload's predicted [`Shadow`] equals the metadata
//! committed on the leader, the payoff of the name-keyed shadow.
//!
-//! Full cross-replica EQUALITY (every live replica holding the same committed
-//! log) is the real consensus property, but it is not asserted yet. Backups
-//! apply prepares in strict order and drop any gap (`op != current_op + 1` in
-//! `metadata::on_replicate` / `iggy_partition`), relying on the primary's
-//! retransmit and the repair sessions (`MetadataRepairSession` / partition
-//! `RepairSession`) to refill. Message repair has landed on both planes, so
-//! the equality assert is unblocked but not yet re-enabled: the sim must
-//! first drive quiesce long enough for repair rounds to converge.
+//! Equality is asserted over the committed PREFIX, not over equal heads: a
+//! replica that missed the last commit broadcast, or that rejoined recently, may
+//! legitimately trail. What it may not do is hold different history at an op it
+//! did commit. Requiring equal heads instead would fail on ordinary lag and say
+//! nothing extra about safety.
use crate::Simulator;
use crate::replica::Replica;
use crate::workload::shadow::Shadow;
-use crate::workload::{Workload, apply_sim_commands, resubmit_due};
+use crate::workload::{Workload, apply_sim_commands, resubmit_due, state_checker};
use consensus::{MetadataHandle, Status};
use metadata::impls::metadata::StreamsFrontend;
use std::collections::BTreeSet;
@@ -72,14 +72,25 @@ struct CommittedMetadata {
impl CommittedMetadata {
/// Restrict to workload-generated entities (see [`WORKLOAD_PREFIX`]), so the
/// entity oracle compares like with like against the shadow.
+ ///
+ /// Every level is filtered on its OWN name, not on its stream's. The harness
+ /// seeds filler topics and partitions of its own (`sim-topic-*`, see
+ /// `Streams::seed_namespace`) to keep slab ids dense, and once the workload
+ /// has created enough streams those fillers land inside a stream named
+ /// `wl-...`. Filtering topics by their stream alone then admits harness state
+ /// into the comparison and the shadow is blamed for not predicting it.
fn workload_owned(mut self) -> Self {
self.streams
.retain(|name| name.starts_with(WORKLOAD_PREFIX));
- self.topics
- .retain(|(stream, _)| stream.starts_with(WORKLOAD_PREFIX));
+ self.topics.retain(|(stream, topic)| {
+ stream.starts_with(WORKLOAD_PREFIX) && topic.starts_with(WORKLOAD_PREFIX)
+ });
self.users.retain(|name| name.starts_with(WORKLOAD_PREFIX));
- self.consumer_groups
- .retain(|(stream, _, _)| stream.starts_with(WORKLOAD_PREFIX));
+ self.consumer_groups.retain(|(stream, topic, group)| {
+ stream.starts_with(WORKLOAD_PREFIX)
+ && topic.starts_with(WORKLOAD_PREFIX)
+ && group.starts_with(WORKLOAD_PREFIX)
+ });
self
}
}
@@ -208,28 +219,141 @@ pub fn quiesce_failure_report(sim: &Simulator, workload: &Workload) -> String {
report
}
-/// Post-drain consensus checks that hold today.
+/// Step until every live replica's metadata plane is `Normal` in one shared
+/// view, or `max_ticks` elapses.
///
-/// Asserts no live replica is ahead of the leader, and (on a serial run) that
-/// the shadow equals the metadata committed on the leader. See the module docs
-/// for why full cross-replica equality is deferred.
+/// Needed before [`assert_converged`], which resolves the leader as "the live
+/// replica whose metadata consensus says it is primary". With primaries spared
+/// from crashes that was always the same replica in view 0. Once a primary can
+/// be crashed, live replicas transiently hold different views and there may be no
+/// `Normal` primary at all, so the leader lookup fails or names a deposed one --
+/// a false failure, not a divergence. Waiting for one view removes the ambiguity
+/// rather than guessing at a leader.
///
-/// Assumes one stable primary that every live replica agrees on: the leader is
-/// `Simulator::primary_index` (a single replica's view), and both checks treat
-/// it as the authoritative, most-advanced log. Sound today because the driver
-/// spares primaries from crashes, so no view change runs mid-test. Once
-/// primary-crash injection lands, live replicas can hold different views and
-/// this breaks: it may pick a stale or crashed leader (a correctly-ahead new
-/// primary then trips "exceeds leader"), or find no `Normal` primary mid-view
-/// change (`metadata_leader` returns `None`). Both are false failures. Fix
-/// then: resolve the leader by highest `(view, commit_offset)`, or quiesce
-/// until live replicas reconverge to one view before asserting. Crash injection
-/// already runs but spares primaries (`maybe_inject_crash`); this is deferred
-/// until primary-crash injection lands, itself gated on a request-resend path.
+/// Returns `false` if the views never converge, which is a real liveness failure
+/// and the caller should report it rather than assert against an unsettled
+/// cluster.
+#[must_use]
+pub fn settle_to_stable_view(sim: &mut Simulator, workload: &mut Workload, max_ticks: u64) -> bool {
+ for _ in 0..max_ticks {
+ if views_are_settled(sim, workload) {
+ return true;
+ }
+ workload.tick();
+ resubmit_due(sim, workload);
+ for reply in sim.step() {
+ let cmds = workload.on_reply(&reply);
+ apply_sim_commands(sim, &cmds);
+ }
+ }
+ views_are_settled(sim, workload)
+}
+
+/// Both planes settled: the metadata group and every tracked partition group.
+///
+/// The partition half matters for the leader-relative offset check in
+/// [`assert_converged`], which resolves its leader from `Simulator::primary_index`
+/// -- a single replica's view of that group's primary. Each partition group runs
+/// its own view change, so settling only the metadata plane leaves that check
+/// asserting against a deposed partition leader, and a correctly-ahead new one
+/// then trips "exceeds leader".
+fn views_are_settled(sim: &Simulator, workload: &Workload) -> bool {
+ if !metadata_view_is_settled(sim) {
+ return false;
+ }
+ workload
+ .options
+ .namespaces
+ .iter()
+ .all(|&ns| partition_view_is_settled(sim, ns))
+}
+
+/// True when every live replica's metadata consensus is `Normal` in the same
+/// view and exactly one of them claims to be primary.
+fn metadata_view_is_settled(sim: &Simulator) -> bool {
+ let mut view = None;
+ let mut primaries = 0usize;
+ let mut live = 0usize;
+ for replica_idx in 0..sim.replica_count {
+ if sim.is_crashed(replica_idx) {
+ continue;
+ }
+ let Some(consensus) = sim.replicas[usize::from(replica_idx)].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ else {
+ continue;
+ };
+ live += 1;
+ if consensus.status() != Status::Normal {
+ return false;
+ }
+ match view {
+ Some(agreed) if agreed != consensus.view() => return false,
+ Some(_) => {}
+ None => view = Some(consensus.view()),
+ }
+ if consensus.is_primary() {
+ primaries += 1;
+ }
+ }
+ live > 0 && primaries == 1
+}
+
+/// True when every live replica hosting `ns` has that group `Normal` in one
+/// shared view with exactly one primary.
+///
+/// A replica that does not host the namespace is skipped rather than treated as
+/// disagreement: a group only materialises on its hash-owning shard, and after a
+/// restart a group whose stream the workload deleted is not re-materialised at
+/// all.
+fn partition_view_is_settled(sim: &Simulator, ns: server_common::sharding::IggyNamespace) -> bool {
+ let mut view = None;
+ let mut primaries = 0usize;
+ let mut hosts = 0usize;
+ for replica_idx in 0..sim.replica_count {
+ if sim.is_crashed(replica_idx) {
+ continue;
+ }
+ let Some(state) = sim.partition_consensus_state(usize::from(replica_idx), ns) else {
+ continue;
+ };
+ hosts += 1;
+ if state.status != Status::Normal {
+ return false;
+ }
+ match view {
+ Some(agreed) if agreed != state.view => return false,
+ Some(_) => {}
+ None => view = Some(state.view),
+ }
+ if state.is_primary {
+ primaries += 1;
+ }
+ }
+ // No live host is settled by default: there is no leader for the offset check
+ // to resolve either, and it skips the namespace for the same reason.
+ hosts == 0 || primaries == 1
+}
+
+/// Post-drain consensus checks.
+///
+/// Asserts no live replica is ahead of the leader, that every live replica agrees
+/// with every other on each committed metadata op they share, and (on a serial
+/// run) that the shadow equals the metadata committed on the leader.
+///
+/// Assumes one stable primary that every live replica agrees on, which
+/// [`settle_to_stable_view`] establishes and which callers should run first once
+/// primaries can be crashed. Without it this may pick a deposed leader (a
+/// correctly-ahead new primary then trips "exceeds leader") or find no `Normal`
+/// primary at all mid-view-change. Both are false failures.
///
/// # Panics
-/// If a replica is ahead of the leader or the shadow mismatches the leader. The
-/// workload seed is in the message so the failing run replays deterministically.
+/// If a replica is ahead of the leader, two replicas disagree on a committed op,
+/// or the shadow mismatches the leader. The workload seed is in the message so the
+/// failing run replays deterministically.
pub fn assert_converged(sim: &Simulator, workload: &Workload) {
let seed = workload.options.seed;
let live: Vec<usize> = (0..sim.replica_count)
@@ -241,25 +365,41 @@ pub fn assert_converged(sim: &Simulator, workload: &Workload) {
"no live replicas at quiesce (seed={seed:#x})"
);
- // Safety direction: no live replica may be ahead of the leader on any
- // namespace. A backup may trail (no idle catch-up yet, see module docs),
- // but a backup whose commit_offset exceeds the leader's is a divergence.
+ // The consensus property proper: replicas compared against each other, not
+ // against the workload's expectations. Runs before the leader-relative checks
+ // because a genuine divergence explains any leader confusion below it.
+ state_checker::assert_committed_prefixes_agree(sim, seed);
+
+ // Safety direction: no live replica may have COMMITTED more of a group than
+ // its leader has. A backup may trail (no idle catch-up yet, see module docs),
+ // but a backup committed past the leader is a divergence.
+ //
+ // Measured on the group's consensus `commit_min`, not on
+ // `PartitionOffsets::commit_offset`. The latter is the highest durably
+ // persisted message offset, which counts an uncommitted suffix: a backup that
+ // persisted op N while the view that elected the new leader settled on N-1 is
+ // ordinary VSR, not divergence. That only stayed invisible while primaries
+ // were spared, since a never-crashed primary is always the furthest ahead;
+ // with primary crashes it fires on a correct cluster.
for &ns in &workload.options.namespaces {
let Some(leader) = sim.primary_index(ns) else {
continue;
};
- let Some(leader_offset) = sim
- .offsets(usize::from(leader), ns)
- .map(|o| o.commit_offset)
+ let Some(leader_committed) = sim
+ .partition_consensus_state(usize::from(leader), ns)
+ .map(|state| state.commit_min)
else {
continue;
};
for &replica_idx in &live {
- if let Some(offset) = sim.offsets(replica_idx, ns).map(|o| o.commit_offset) {
+ if let Some(committed) = sim
+ .partition_consensus_state(replica_idx, ns)
+ .map(|state| state.commit_min)
+ {
assert!(
- offset <= leader_offset,
- "replica {replica_idx} commit_offset {offset} exceeds leader {leader} \
- ({leader_offset}) on ns {ns:?} at quiesce (seed={seed:#x})",
+ committed <= leader_committed,
+ "replica {replica_idx} committed {committed} ops exceeds leader {leader} \
+ ({leader_committed}) on ns {ns:?} at quiesce (seed={seed:#x})",
);
}
}
diff --git a/core/simulator/src/workload/state_checker.rs b/core/simulator/src/workload/state_checker.rs
new file mode 100644
index 000000000..a3c5135f8
--- /dev/null
+++ b/core/simulator/src/workload/state_checker.rs
@@ -0,0 +1,219 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Cross-replica committed-log equality, after TigerBeetle's
+//! `testing/cluster/state_checker.zig`.
+//!
+//! The per-tick checks in [`super::invariants`] catch a single replica
+//! contradicting itself, and [`super::oracle`] compares committed metadata
+//! against the workload's shadow. Neither compares replicas to EACH OTHER, which
+//! is the actual consensus property: two replicas that both committed op N must
+//! have committed the same op N.
+//!
+//! Modelled on TigerBeetle's checker rather than invented: it keeps one canonical
+//! commit chain, asserts every replica agrees with it wherever they overlap
+//! (`(commit_a == commit_b) == (checksum_a == checksum_b)`), and asserts the chain
+//! is hash-linked (`header_b.parent == checksum_a`). Recording which replicas
+//! reached each op also makes the check provably non-vacuous, which matters:
+//! a chain nothing was ever compared against passes silently.
+
+use crate::Simulator;
+use consensus::MetadataHandle;
+use iggy_binary_protocol::PrepareHeader;
+use journal::Journal;
+use std::collections::{BTreeMap, BTreeSet};
+
+/// One op of the canonical committed chain.
+#[derive(Debug)]
+struct CanonicalCommit {
+ /// Identity of the prepare committed at this op. Two replicas disagreeing
+ /// here is a divergence: the same log position holds different history.
+ ///
+ /// The chain's hash link is checked against this rather than against a stored
+ /// `parent`: an arriving header's `parent` must equal the canonical previous
+ /// op's `checksum`, so keeping each entry's own parent as well would record a
+ /// value nothing ever reads.
+ checksum: u128,
+ /// Replicas observed committing this op, so the check can prove it compared
+ /// something rather than passing over an empty chain.
+ replicas: BTreeSet<u8>,
+}
+
+/// Canonical committed metadata chain, accumulated across ticks.
+#[derive(Debug, Default)]
+pub struct StateChecker {
+ commits: BTreeMap<u64, CanonicalCommit>,
+ /// Highest op already verified per replica, so a tick only walks what is
+ /// new. A high-water mark, never lowered: a restart recovers its commit
+ /// point from a lower bound (`SimJournal::recovery_commit_watermark`) and so
+ /// may report a smaller `commit_min` than it did before, and re-verifying
+ /// that prefix every tick would make this quadratic for no gain.
+ verified_upto: BTreeMap<u8, u64>,
+}
+
+impl StateChecker {
+ #[must_use]
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Fold every live replica's newly committed metadata ops into the canonical
+ /// chain, asserting agreement.
+ ///
+ /// Reads committed state only (ops at or below `commit_min`), so a prepare
+ /// still in flight is never compared: replicas are allowed to disagree about
+ /// uncommitted tails, and that is what a view change resolves.
+ ///
+ /// Crashed replicas are skipped rather than dropped: their high-water mark is
+ /// kept, so a restart re-verifies only what it commits anew.
+ ///
+ /// # Panics
+ /// On any disagreement about a committed op, or a broken hash chain. The
+ /// message names both replicas and the op, and the seed replays the run.
+ pub fn check(&mut self, sim: &Simulator, seed: u64) {
+ for replica_idx in 0..sim.replica_count {
+ if sim.is_crashed(replica_idx) {
+ continue;
+ }
+ let replica = &sim.replicas[usize::from(replica_idx)];
+ let Some(consensus) = replica.shards[0].plane.metadata().consensus.as_ref() else {
+ continue;
+ };
+ let committed = consensus.commit_min();
+ let verified = self.verified_upto.get(&replica_idx).copied().unwrap_or(0);
+ for op in (verified + 1)..=committed {
+ // Absent header at a committed op: the sim never checkpoints, so
+ // nothing drains the prefix and this would be a real hole. Left to
+ // the journal's own invariants rather than asserted here, since a
+ // recovered replica legitimately reports a commit point one above
+ // its head (the watermark is a lower bound).
+ let Some(header) = journaled_header(replica, op) else {
+ continue;
+ };
+ self.record(replica_idx, op, &header, seed);
+ }
+ self.verified_upto
+ .insert(replica_idx, verified.max(committed));
+ }
+ }
+
+ /// Number of ops in the canonical chain. Tests assert this is non-zero, so a
+ /// green run cannot mean "never compared anything".
+ #[must_use]
+ pub fn chain_len(&self) -> usize {
+ self.commits.len()
+ }
+
+ /// Ops witnessed by more than one replica. The only ops that actually
+ /// exercised the equality property: an op only ever seen on one replica was
+ /// recorded, never compared.
+ #[must_use]
+ pub fn ops_compared(&self) -> usize {
+ self.commits
+ .values()
+ .filter(|commit| commit.replicas.len() > 1)
+ .count()
+ }
+
+ fn record(&mut self, replica_idx: u8, op: u64, header: &PrepareHeader, seed: u64) {
+ // Hash-chain link, checked before the identity comparison so a diverged
+ // prefix is reported at the op where the chains part rather than at the
+ // first op whose contents happen to differ.
+ if let Some(previous) = self.commits.get(&(op - 1))
+ && header.parent != previous.checksum
+ {
+ panic!(
+ "replica {replica_idx} committed op {op} whose parent {:#x} is not the \
+ canonical op {} checksum {:#x}: its committed history forked below this \
+ op (seed={seed:#x})",
+ header.parent,
+ op - 1,
+ previous.checksum,
+ );
+ }
+ match self.commits.get_mut(&op) {
+ Some(canonical) => {
+ assert_eq!(
+ canonical.checksum, header.checksum,
+ "replicas disagree on committed op {op}: canonical checksum {:#x} \
+ (committed by {:?}) vs replica {replica_idx}'s {:#x}. Two replicas \
+ committed different history at the same log position (seed={seed:#x})",
+ canonical.checksum, canonical.replicas, header.checksum,
+ );
+ canonical.replicas.insert(replica_idx);
+ }
+ None => {
+ self.commits.insert(
+ op,
+ CanonicalCommit {
+ checksum: header.checksum,
+ replicas: BTreeSet::from([replica_idx]),
+ },
+ );
+ }
+ }
+ }
+}
+
+/// Assert every live replica's committed metadata prefix agrees, op for op.
+///
+/// The quiesce-time counterpart to [`StateChecker::check`]: that one folds ops in
+/// as they commit and so compares whatever happened to overlap, while this walks
+/// the full committed prefix of every live replica at rest and requires the
+/// shorter to be a genuine PREFIX of the longer. A replica may still trail (it
+/// may have missed the last commit broadcast), but where it has committed
+/// anything it must match.
+///
+/// # Panics
+/// If two live replicas disagree on any committed op.
+pub fn assert_committed_prefixes_agree(sim: &Simulator, seed: u64) {
+ let mut canonical: BTreeMap<u64, (u128, u8)> = BTreeMap::new();
+ for replica_idx in 0..sim.replica_count {
+ if sim.is_crashed(replica_idx) {
+ continue;
+ }
+ let replica = &sim.replicas[usize::from(replica_idx)];
+ let Some(consensus) = replica.shards[0].plane.metadata().consensus.as_ref() else {
+ continue;
+ };
+ for op in 1..=consensus.commit_min() {
+ let Some(header) = journaled_header(replica, op) else {
+ continue;
+ };
+ match canonical.get(&op) {
+ Some(&(checksum, owner)) => assert_eq!(
+ checksum, header.checksum,
+ "at quiesce replica {replica_idx} and replica {owner} disagree on \
+ committed metadata op {op}: {:#x} vs {checksum:#x} (seed={seed:#x})",
+ header.checksum,
+ ),
+ None => {
+ canonical.insert(op, (header.checksum, replica_idx));
+ }
+ }
+ }
+ }
+}
+
+/// The header a replica has journaled at `op`, if any.
+///
+/// Reads shard 0's retained metadata WAL, which is where the committed metadata
+/// log lives; the journal is harness-owned so this also works across a restart.
+fn journaled_header(replica: &crate::SimReplica, op: u64) -> Option<PrepareHeader> {
+ let slot = usize::try_from(op).ok()?;
+ replica.metadata_journal.header(slot).copied()
+}