This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 914bf23b7 feat(simulator): separate a vacuous run and assert through
the drain (#4191)
914bf23b7 is described below
commit 914bf23b74d14f5115f387eda5cf666f713429b0
Author: Krishna Vishal <[email protected]>
AuthorDate: Wed Sep 16 12:45:36 2026 +0530
feat(simulator): separate a vacuous run and assert through the drain (#4191)
---
core/consensus/src/impls.rs | 125 ++++++++++++++--
core/shard/src/lib.rs | 14 +-
core/simulator/src/bin/workload-fuzz.rs | 137 +++++++++++-------
core/simulator/src/lib.rs | 204 +++++++++++++++++++++++----
core/simulator/src/workload/mod.rs | 18 ++-
core/simulator/src/workload/oracle.rs | 27 +++-
core/simulator/src/workload/state_checker.rs | 66 +++++++--
7 files changed, 475 insertions(+), 116 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 2f2fa03e2..e31680eec 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -2947,13 +2947,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
actions
}
- /// Handle a received `DoViewChange` message (only relevant for primary
candidate).
- ///
- /// "When the new primary receives f + 1 DOVIEWCHANGE messages from
different
- /// replicas (including itself), it sets its view-number to that in the
messages
- /// and selects as the new log the one contained in the message with the
largest v'..."
- ///
- /// The `commit` this replica advertises in a `DoViewChange`.
+ /// The `commit` this replica advertises in a `DoViewChange`, and in any
+ /// `StartView` it announces.
///
/// `commit_max`, not `commit_min`: the new primary floors its pipeline
rebuild
/// at `max(commit)` across the quorum, and only `commit_max` bounds that
range
@@ -2964,7 +2959,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
/// the prepares and `DoViewChangeHeader::validate` rejects `commit > op`.
/// Lossless for the rebuild floor: quorum intersection guarantees some
sender
/// whose head covers the true commit point carries it.
- fn dvc_commit(&self) -> u64 {
+ #[must_use]
+ pub fn dvc_commit(&self) -> u64 {
let op = self.sequencer.current_sequence();
self.commit_max.get().min(op)
}
@@ -3235,12 +3231,14 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
vec![VsrAction::SendStartView {
view: self.view.get(),
op: self.sequencer.current_sequence(),
- commit: self.commit_max.get(),
+ commit: self.dvc_commit(),
incarnation: header.incarnation,
target: Some(header.replica),
- // A probe answer reports this primary's settled frontier, not a
- // freshly merged log, so there is no canonical suffix to publish.
- suffix: Vec::new(),
+ // A prober can hold a different entry at an op at or below this
+ // commit point, left from a view whose entry the change truncated.
+ // Without the canonical headers it cannot know, so it adopts the
+ // commit point and applies its own stale entry.
+ suffix: self.local_dvc_suffix().headers().to_vec(),
group: self.group,
}]
}
@@ -6002,3 +6000,106 @@ mod recovery_barrier_tests {
);
}
}
+
+#[cfg(test)]
+mod probe_answer_tests {
+ //! What a primary publishes when it answers a `RequestStartView` probe.
+ //!
+ //! The answer is not a bare frontier report. A prober can hold a different
+ //! entry at an op at or below the announced commit point, left from a view
+ //! whose entry the change truncated, and the canonical headers are its
only
+ //! way to find that out before it adopts the commit point.
+
+ use super::*;
+ use crate::LocalPipeline;
+ use crate::test_bus::NoopBus;
+ use crate::view_change_quorum::dvc_blank;
+
+ const PROBER: u8 = 1;
+
+ /// Replica 0 of 3, primary in view 0 with head `head` and commit point
+ /// `commit`, carrying a fresh suffix snapshot over `[commit + 1, head]`.
+ fn primary_with_suffix(head: u64, commit: u64) -> VsrConsensus<NoopBus,
LocalPipeline> {
+ let consensus = VsrConsensus::new(1, 0, 3, METADATA_GROUP, NoopBus,
LocalPipeline::new());
+ consensus.init();
+ consensus.sequencer().set_sequence(head);
+ consensus.advance_commit_max(commit);
+ // Tagged on `(head, commit)` as they stand now, so `local_dvc_suffix`
+ // returns it rather than falling back to empty.
+ let headers: Vec<PrepareHeader> = (commit +
1..=head).rev().map(dvc_blank).collect();
+ let present = (1u128 << headers.len()) - 1;
+ consensus.set_local_dvc_suffix(DvcSuffix::new(headers, 0, present));
+ consensus
+ }
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn probe(view: u32) -> RequestStartViewHeader {
+ RequestStartViewHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 1,
+ size: size_of::<RequestStartViewHeader>() as u32,
+ view,
+ release: 0,
+ command: Command::RequestStartView,
+ replica: PROBER,
+ reserved_frame: [0; 66],
+ group: METADATA_GROUP,
+ reserved: [0; 104],
+ incarnation: 0,
+ }
+ }
+
+ /// The suffix is the whole point: without it the prober cannot tell its
own
+ /// op from the view's op at the same number, adopts the commit point, and
+ /// commits the stale entry.
+ #[test]
+ fn
given_a_cached_suffix_when_answering_a_probe_should_publish_its_headers() {
+ let consensus = primary_with_suffix(7, 5);
+
+ let actions = consensus.handle_request_start_view(PlaneKind::Metadata,
&probe(0));
+
+ let [
+ VsrAction::SendStartView {
+ op,
+ commit,
+ suffix,
+ target,
+ ..
+ },
+ ] = &actions[..]
+ else {
+ panic!("a probe from a backup must be answered with one StartView:
{actions:?}");
+ };
+ assert_eq!(*op, 7);
+ assert_eq!(*commit, 5);
+ assert_eq!(*target, Some(PROBER));
+ assert_eq!(
+ suffix.iter().map(|header| header.op).collect::<Vec<_>>(),
+ vec![7, 6],
+ "the answer must carry the view's canonical headers, high op first"
+ );
+ }
+
+ /// `commit_max` legitimately runs ahead of the head, since a replica
learns
+ /// the commit point before it holds the prepares.
`StartViewHeader::validate`
+ /// refuses `commit > op`, and the dispatcher turns that refusal into a
panic,
+ /// so the announcement clamps the way `DoViewChange` already does.
+ #[test]
+ fn
given_a_commit_point_above_the_head_when_answering_a_probe_should_clamp_it() {
+ let consensus = primary_with_suffix(7, 5);
+ consensus.advance_commit_max(9);
+ assert!(consensus.commit_max() >
consensus.sequencer().current_sequence());
+
+ let actions = consensus.handle_request_start_view(PlaneKind::Metadata,
&probe(0));
+
+ let [VsrAction::SendStartView { op, commit, .. }] = &actions[..] else {
+ panic!("expected one StartView: {actions:?}");
+ };
+ assert!(
+ commit <= op,
+ "announced commit {commit} exceeds head {op}, which
StartViewHeader::validate \
+ rejects and the dispatcher panics on"
+ );
+ }
+}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index f85c29ecd..803547fba 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -6181,7 +6181,7 @@ where
// number and a backup can sit above it. Splitting on the view's number
// would drop already-executed ops with no rollback, and silently.
let announced_commit = pending.as_ref().map_or(0, |pending|
pending.commit_max);
- let applied_floor = announced_commit.max(consensus.commit_min());
+ let applied_floor = consensus.commit_min();
let mut repairable_from: Option<u64> = None;
for canonical in pending.as_ref().map_or(&[][..], |pending|
&pending.headers) {
@@ -10397,13 +10397,15 @@ where
let action = VsrAction::SendStartView {
view: consensus.view(),
op: consensus.sequencer().current_sequence(),
- commit: consensus.commit_max(),
+ commit: consensus.dvc_commit(),
incarnation: 0,
target: None,
group: consensus.group(),
- // Correcting a peer on a stale view, not concluding a view change:
this
- // publishes the settled frontier, which the peer reaches by repair.
- suffix: Vec::new(),
+ // The headers, not just the frontier. Repair skips an op whose header
is
+ // already resident, so a peer holding a DIFFERENT entry at an op under
+ // this commit point never learns of it from repair alone: it adopts
the
+ // commit point and applies what it already has.
+ suffix: consensus.local_dvc_suffix().headers().to_vec(),
};
dispatch_vsr_actions::<B, P, J>(consensus, None, &[action]).await;
}
@@ -11330,7 +11332,7 @@ async fn reconcile_partition_view_divergence<B, SB>(
// Truncation is safe only above what this replica has *applied*, which is
not
// the view's commit point: a backup can sit above it.
let announced_commit = pending.map_or(0, |pending| pending.commit_max);
- let applied_floor =
announced_commit.max(partition.consensus().commit_min());
+ let applied_floor = partition.consensus().commit_min();
let mut repairable_from: Option<u64> = None;
for canonical in pending.map_or(&[][..], |pending| &pending.headers) {
diff --git a/core/simulator/src/bin/workload-fuzz.rs
b/core/simulator/src/bin/workload-fuzz.rs
index fb6ab966a..5f5e247ce 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -44,6 +44,11 @@
//! CI campaign wants: `none`/`light`/`heavy` are three points in parameter
space,
//! so a thousand seeds against `heavy` is the same network a thousand times.
The
//! drawn values print on the `network:` line and `--seed` replays them
exactly.
+//!
+//! Exit codes: `0` passed, `2` the configuration is unusable, `3` the run
proved
+//! nothing because a vacuity floor went unmet, and `101` an invariant or an
oracle
+//! failed. Only `101` is a bug. A campaign that cannot tell `3` from `101`
reports
+//! its own thin seeds as failures, which is what splitting them apart is for.
use clap::{Parser, ValueEnum};
use iggy_common::IggyByteSize;
@@ -53,10 +58,27 @@ use simulator::Simulator;
use simulator::client::SimClient;
use simulator::packet::{COMMAND_LABELS, PacketSimulatorOptions, PartitionMode,
PartitionSymmetry};
use simulator::workload::actions::Action;
+use simulator::workload::invariants::Invariants;
use simulator::workload::options::{ActionWeights, WorkloadOptions};
use simulator::workload::{FaultInjector, Workload, oracle, run_with_faults};
use strum::IntoEnumIterator;
+/// Exit code for a run that proved nothing.
+///
+/// Apart from the `101` a panic exits with, because a run under a vacuity
floor is
+/// not a failure: every oracle held and none of them had anything to compare.
+const EXIT_VACUOUS: i32 = 3;
+
+/// Report that the run proved nothing, then exit with [`EXIT_VACUOUS`].
+///
+/// On stdout beside the coverage numbers rather than on stderr, because this
is an
+/// outcome of the run and not a diagnostic of one. No panic, so the reproduce
line
+/// the panic hook prints stays reserved for failures worth reproducing.
+fn vacuous(reason: std::fmt::Arguments<'_>) -> ! {
+ println!("vacuous: {reason}");
+ std::process::exit(EXIT_VACUOUS)
+}
+
#[derive(Parser)]
#[command(about = "Deterministic workload fuzzer for the Iggy simulator")]
#[allow(clippy::struct_excessive_bools)]
@@ -119,14 +141,14 @@ struct Args {
/// than Iggy is.
#[arg(long)]
restore_partition_frontier: bool,
- /// Fail the run if the entity oracle did not hold at quiesce.
+ /// End the run as vacuous if the entity oracle did not hold at quiesce.
///
/// An eviction disarms it (the forgotten request's fate is unknown) and it
/// re-arms only once the shadow is proven equal to committed state again.
Without
/// this flag a run whose oracle stayed disarmed still exits 0.
#[arg(long)]
require_entity_oracle: bool,
- /// Committed workload operations this run must produce, or it fails.
+ /// Committed workload operations this run must produce, or it ends as
vacuous.
///
/// A run that commits nothing proved nothing: every oracle downstream
compares an
/// empty shadow against empty committed state and agrees. `0` opts out.
@@ -135,7 +157,7 @@ struct Args {
/// Committed ops, on EITHER plane, that must have been witnessed by more
than
/// one live replica, i.e. that exercised cross-replica agreement. Ignored
below
/// two live replicas, where the property is untestable rather than
untested.
- /// `0` opts out.
+ /// `0` opts out. An unmet floor exits [`EXIT_VACUOUS`], like every floor
here.
#[arg(long, default_value_t = 1)]
min_ops_compared: usize,
/// As `--min-ops-compared`, but METADATA ops only.
@@ -147,7 +169,8 @@ struct Args {
/// with `0`.
#[arg(long, default_value_t = 1)]
min_metadata_ops_compared: usize,
- /// Fail the run if crash or restart injection was requested but never
happened.
+ /// End the run as vacuous if crash or restart injection was requested but
never
+ /// happened.
/// Off by default, since a short run at low probability may legitimately
draw
/// none; on for a campaign where such a seed is silently wasted.
#[arg(long)]
@@ -518,13 +541,16 @@ fn validate_network_options(options:
&PacketSimulatorOptions) -> Result<(), Stri
/// compare the replicas against each other and against the oracle.
///
/// Split out of `main` only for length. Every assert here is a hard failure by
-/// design; see the individual comments for why each one is not a warning.
+/// design, and the individual comments say why each one is not a warning. The
+/// vacuity floors are the exception: they end the run at [`EXIT_VACUOUS`],
since a
+/// run that compared nothing has disproved nothing either.
fn run_quiesce_phase(
args: &Args,
sim: &mut Simulator,
workload: &mut Workload,
seed: u64,
replicas: u8,
+ invariants: &mut Invariants,
) {
// Liveness phase, opt-in: a drain against a handicapped cluster has no
// verdict, but healing unconditionally resolves the wedges worth
reporting.
@@ -546,7 +572,7 @@ fn run_quiesce_phase(
// unactionable; with the client resending, a request unanswered inside the
// budget is either a wedge or a liveness bug.
assert!(
- oracle::drive_to_quiesce(sim, workload, 50_000),
+ oracle::drive_to_quiesce(sim, workload, 50_000, invariants),
"{}",
oracle::quiesce_failure_report(sim, workload),
);
@@ -554,7 +580,7 @@ fn run_quiesce_phase(
// the leader as whichever live replica claims to be primary, so asserting
// mid-view-change finds none or finds a deposed one, both false failures.
assert!(
- oracle::settle_to_stable_view(sim, workload, 50_000),
+ oracle::settle_to_stable_view(sim, workload, 50_000, invariants),
"metadata views never converged after the drain\n{}",
oracle::quiesce_failure_report(sim, workload),
);
@@ -580,36 +606,37 @@ fn run_quiesce_phase(
convergence.replicas_compared,
convergence.namespaces_checked,
);
- assert!(
- !args.require_entity_oracle || workload.strict_outcome_oracle(),
- "--require-entity-oracle: the entity oracle was {entity_oracle}, so
this run \
- proved nothing about entity state (seed={seed:#x})"
- );
+ if args.require_entity_oracle && !workload.strict_outcome_oracle() {
+ vacuous(format_args!(
+ "--require-entity-oracle: the entity oracle was {entity_oracle},
so this run \
+ proved nothing about entity state (seed={seed:#x})"
+ ));
+ }
let live = usize::from(replicas) - sim.crashed.len();
// Either plane satisfies it: a partition-plane run commits almost no
metadata,
// so the metadata count alone called every such run vacuous.
let compared = convergence.ops_compared +
convergence.partition_ops_compared;
- assert!(
- args.min_ops_compared == 0 || live < 2 || compared >=
args.min_ops_compared,
- "--min-ops-compared {}: {live} replicas live but only {compared} op(s)
witnessed \
- by more than one ({} metadata, {} partition), so cross-replica
agreement went \
- untested (seed={seed:#x})",
- args.min_ops_compared,
- convergence.ops_compared,
- convergence.partition_ops_compared,
- );
+ if args.min_ops_compared > 0 && live >= 2 && compared <
args.min_ops_compared {
+ vacuous(format_args!(
+ "--min-ops-compared {}: {live} replicas live but only {compared}
op(s) witnessed \
+ by more than one ({} metadata, {} partition), so cross-replica
agreement went \
+ untested (seed={seed:#x})",
+ args.min_ops_compared, convergence.ops_compared,
convergence.partition_ops_compared,
+ ));
+ }
// The metadata half on its own: summing the planes above lets a
partition-only
// run clear that floor while the metadata oracle compares nothing.
- assert!(
- args.min_metadata_ops_compared == 0
- || live < 2
- || convergence.ops_compared >= args.min_metadata_ops_compared,
- "--min-metadata-ops-compared {}: {live} replicas live but only {}
committed metadata \
- op(s) witnessed by more than one, so the metadata oracle compared an
empty chain \
- (seed={seed:#x})",
- args.min_metadata_ops_compared,
- convergence.ops_compared,
- );
+ if args.min_metadata_ops_compared > 0
+ && live >= 2
+ && convergence.ops_compared < args.min_metadata_ops_compared
+ {
+ vacuous(format_args!(
+ "--min-metadata-ops-compared {}: {live} replicas live but only {}
committed metadata \
+ op(s) witnessed by more than one, so the metadata oracle compared
an empty chain \
+ (seed={seed:#x})",
+ args.min_metadata_ops_compared, convergence.ops_compared,
+ ));
+ }
// 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);
@@ -679,6 +706,9 @@ fn main() {
let mut workload = Workload::new(options);
let mut injector = FaultInjector::new(seed, replicas);
+ // One checker for the whole run: the drain in `run_quiesce_phase`
continues with
+ // it, so a mark set during the active phase still holds the drain to
account.
+ let mut invariants = Invariants::new();
let replies = run_with_faults(
&mut sim,
&mut workload,
@@ -686,6 +716,7 @@ fn main() {
ticks,
u64::MAX,
&mut injector,
+ &mut invariants,
);
println!(
"ran {ticks} ticks; {replies} replies; crashes={} restarts={} still
down: {}",
@@ -700,7 +731,14 @@ fn main() {
print_coverage(&workload);
if quiesce {
- run_quiesce_phase(&args, &mut sim, &mut workload, seed, replicas);
+ run_quiesce_phase(
+ &args,
+ &mut sim,
+ &mut workload,
+ seed,
+ replicas,
+ &mut invariants,
+ );
}
// After the quiesce block, so the drain's own commits count. Rejections
are added
@@ -709,23 +747,26 @@ fn main() {
// full of them exercised the plane.
let stats = workload.auditor.stats();
let commits: u64 = stats.commits_per_action.iter().sum::<u64>() +
stats.committed_rejections;
- assert!(
- commits >= args.min_commits,
- "--min-commits {}: the run committed {commits} operation(s) on the
{plane:?} \
- plane, so every oracle above compared empty against empty
(seed={seed:#x})",
- args.min_commits,
- );
+ if commits < args.min_commits {
+ vacuous(format_args!(
+ "--min-commits {}: the run committed {commits} operation(s) on the
{plane:?} \
+ plane, so every oracle above compared empty against empty
(seed={seed:#x})",
+ args.min_commits,
+ ));
+ }
if args.require_faults {
- assert!(
- crash_prob <= 0.0 || injector.crashes() > 0,
- "--require-faults: --crash-prob {crash_prob} crashed nothing \
- (seed={seed:#x})"
- );
- assert!(
- args.restart_prob <= 0.0 || injector.restarts() > 0,
- "--require-faults: --restart-prob {} restarted nothing
(seed={seed:#x})",
- args.restart_prob,
- );
+ if crash_prob > 0.0 && injector.crashes() == 0 {
+ vacuous(format_args!(
+ "--require-faults: --crash-prob {crash_prob} crashed nothing \
+ (seed={seed:#x})"
+ ));
+ }
+ if args.restart_prob > 0.0 && injector.restarts() == 0 {
+ vacuous(format_args!(
+ "--require-faults: --restart-prob {} restarted nothing
(seed={seed:#x})",
+ args.restart_prob,
+ ));
+ }
}
print_command_coverage(&sim);
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 31527f6ea..8077668ed 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -2605,11 +2605,19 @@ mod tests {
let mut wl = Workload::new(options);
let clients = [client];
- let replies = workload::run(&mut sim, &mut wl, &clients, 2_000,
u64::MAX);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ let replies = workload::run(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 2_000,
+ u64::MAX,
+ &mut invariants,
+ );
assert!(replies > 0, "workload produced no replies");
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut wl, 5_000),
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 5_000, &mut
invariants),
"system did not drain within the tick budget"
);
// Cross-replica agreement + entity oracle (single client => strict).
@@ -2661,7 +2669,15 @@ mod tests {
let mut wl = Workload::new(options);
let clients = [client];
- let replies = workload::run(&mut sim, &mut wl, &clients, 3_000,
u64::MAX);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ let replies = workload::run(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 3_000,
+ u64::MAX,
+ &mut invariants,
+ );
assert!(replies > 0, "workload produced no replies");
assert!(
!sim.crashed.is_empty(),
@@ -2669,7 +2685,7 @@ mod tests {
);
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut wl, 5_000),
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 5_000, &mut
invariants),
"surviving quorum did not drain within the tick budget"
);
oracle::assert_converged(&sim, &mut wl);
@@ -2722,7 +2738,15 @@ mod tests {
let mut wl = Workload::new(options);
let clients = [client];
- let replies = workload::run(&mut sim, &mut wl, &clients, 3_000,
u64::MAX);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ let replies = workload::run(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 3_000,
+ u64::MAX,
+ &mut invariants,
+ );
assert!(replies > 0, "lossy workload produced no replies");
assert!(
wl.resends() > 0,
@@ -2731,7 +2755,7 @@ mod tests {
);
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut wl, 20_000),
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 20_000, &mut
invariants),
"{}",
oracle::quiesce_failure_report(&sim, &wl),
);
@@ -3822,7 +3846,15 @@ mod tests {
let mut wl = Workload::new(options);
let clients = [client];
// run() asserts the per-tick invariants every tick under injected
crashes.
- let replies = workload::run(&mut sim, &mut wl, &clients, 3_000,
u64::MAX);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ let replies = workload::run(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 3_000,
+ u64::MAX,
+ &mut invariants,
+ );
let crashed = sim.crashed.len();
assert!(
@@ -4050,6 +4082,7 @@ mod tests {
let mut workload = Workload::new(options);
let mut injector = FaultInjector::new(seed, replica_count);
let clients = [client];
+ let mut invariants =
crate::workload::invariants::Invariants::new();
let replies = run_with_faults(
&mut sim,
&mut workload,
@@ -4057,6 +4090,7 @@ mod tests {
3_000,
u64::MAX,
&mut injector,
+ &mut invariants,
);
(
replies,
@@ -4564,7 +4598,15 @@ mod tests {
let mut wl = Workload::new(options);
let clients = [client];
- let replies = workload::run(&mut sim, &mut wl, &clients, 4_000,
u64::MAX);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ let replies = workload::run(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 4_000,
+ u64::MAX,
+ &mut invariants,
+ );
assert!(replies > 0, "shell workload produced no replies");
let stats = wl.auditor.stats();
@@ -4580,7 +4622,7 @@ mod tests {
);
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut wl, 20_000),
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 20_000, &mut
invariants),
"{}",
oracle::quiesce_failure_report(&sim, &wl),
);
@@ -4662,7 +4704,16 @@ mod tests {
let clients = [client];
let mut injector = FaultInjector::new(seed, replica_count);
- run_with_faults(&mut sim, &mut wl, &clients, 1_500, u64::MAX, &mut
injector);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ run_with_faults(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 1_500,
+ u64::MAX,
+ &mut injector,
+ &mut invariants,
+ );
assert!(
wl.auditor.stats().transient_rejections > 0,
@@ -4671,7 +4722,7 @@ mod tests {
);
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut wl, 50_000),
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 50_000, &mut
invariants),
"{}",
oracle::quiesce_failure_report(&sim, &wl),
);
@@ -4742,6 +4793,7 @@ mod tests {
let clients = [client];
let mut injector = FaultInjector::new(seed, replica_count);
+ let mut invariants = crate::workload::invariants::Invariants::new();
let _ = workload::run_with_faults(
&mut sim,
&mut workload,
@@ -4749,6 +4801,7 @@ mod tests {
4_000,
u64::MAX,
&mut injector,
+ &mut invariants,
);
assert!(
@@ -4756,7 +4809,7 @@ mod tests {
"no replica crashed, so no view change ran and this proves nothing"
);
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut workload, 50_000),
+ oracle::drive_to_quiesce(&mut sim, &mut workload, 50_000, &mut
invariants),
"{}",
oracle::quiesce_failure_report(&sim, &workload),
);
@@ -4806,22 +4859,18 @@ mod tests {
let clients = [client];
let mut injector = FaultInjector::new(seed, replica_count);
+ // The checker is read afterwards, so it is declared here rather than
left to
+ // the driver: `chain` below is the accumulated canonical commit chain.
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);
- }
+ workload::run_with_faults(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 4_000,
+ u64::MAX,
+ &mut injector,
+ &mut invariants,
+ );
assert!(
injector.restarts() > 0,
@@ -4839,12 +4888,12 @@ mod tests {
);
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut wl, 50_000),
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 50_000, &mut
invariants),
"{}",
oracle::quiesce_failure_report(&sim, &wl),
);
assert!(
- oracle::settle_to_stable_view(&mut sim, &mut wl, 50_000),
+ oracle::settle_to_stable_view(&mut sim, &mut wl, 50_000, &mut
invariants),
"metadata views never converged after the drain"
);
oracle::assert_converged(&sim, &mut wl);
@@ -5222,11 +5271,19 @@ mod tests {
options.weights = ActionWeights::new(&[(Action::SendMessages, 100)]);
let mut wl = Workload::new(options);
let clients = [client];
- let replies = workload::run(&mut sim, &mut wl, &clients, 2_000,
u64::MAX);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ let replies = workload::run(
+ &mut sim,
+ &mut wl,
+ &clients,
+ 2_000,
+ u64::MAX,
+ &mut invariants,
+ );
assert!(replies > 0, "workload produced no replies");
assert!(
- oracle::drive_to_quiesce(&mut sim, &mut wl, 5_000),
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 5_000, &mut
invariants),
"system did not drain within the tick budget"
);
oracle::assert_converged(&sim, &mut wl);
@@ -7985,3 +8042,88 @@ mod review_4092_dst_tests {
);
}
}
+
+#[cfg(test)]
+mod probe_answer_divergence_tests {
+ //! End-to-end seeds for the probe-answer `StartView`. The mechanism
itself is
+ //! pinned by `consensus::impls::probe_answer_tests`; these replay the
runs that
+ //! found it.
+
+ use super::*;
+
+ /// Seed 144 of the uniform swarm lane: replica 0 prepared op 7 in view 0
+ /// without acks, view 1 truncated it and prepared a different op 7, and
+ /// replica 0 then adopted view 1 through a probe answer that carried no
+ /// canonical headers, so it kept its own op 7 and committed that instead.
+ ///
+ /// Network faults only, no crash needed, which is why it lands at op 7 and
+ /// replays fast. The mechanism is pinned separately by
+ /// `consensus::impls::probe_answer_tests`; this is the end-to-end seed.
+ #[test]
+ fn
given_a_probe_adopted_view_when_the_head_diverges_should_not_commit_the_stale_entry()
{
+ use crate::workload::{
+ self, FaultInjector, Workload,
+ options::{ActionWeights, WorkloadOptions},
+ oracle,
+ };
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let seed = 144;
+ // Swarm, not a fixed profile: the asymmetric partitions and clogs this
+ // seed draws are what let a primary prepare an op it cannot get acked.
+ let mut network_opts = packet::PacketSimulatorOptions::swarm(seed);
+ network_opts.node_count = replica_count;
+ network_opts.client_count = 1;
+ 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);
+
+ let mut options = WorkloadOptions::new(seed, replica_count, vec![ns]);
+ options.weights = ActionWeights::uniform();
+ options.crash_per_tick_ratio = 0.01;
+ options.restart_per_tick_ratio = 0.05;
+ let mut wl = Workload::new(options);
+
+ let mut injector = FaultInjector::new(seed, replica_count);
+ let mut invariants = crate::workload::invariants::Invariants::new();
+ // `run_with_faults` runs the per-tick invariants and the live state
+ // checker, which is where the divergence fired.
+ workload::run_with_faults(
+ &mut sim,
+ &mut wl,
+ &[client],
+ 6_000,
+ u64::MAX,
+ &mut injector,
+ &mut invariants,
+ );
+
+ assert!(
+ oracle::drive_to_quiesce(&mut sim, &mut wl, 50_000, &mut
invariants),
+ "{}",
+ oracle::quiesce_failure_report(&sim, &wl),
+ );
+ assert!(
+ oracle::settle_to_stable_view(&mut sim, &mut wl, 50_000, &mut
invariants),
+ "metadata views never converged after the drain"
+ );
+ let report = oracle::assert_converged(&sim, &mut wl);
+ assert!(
+ report.ops_compared > 0,
+ "no committed metadata op was witnessed on two replicas, so this
seed \
+ would pass on a diverged cluster"
+ );
+ }
+}
diff --git a/core/simulator/src/workload/mod.rs
b/core/simulator/src/workload/mod.rs
index 0995eb7d7..06a356b70 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -709,6 +709,7 @@ pub fn run(
clients: &[SimClient],
tick_budget: u64,
replies_target: u64,
+ invariants: &mut Invariants,
) -> u64 {
let mut injector = FaultInjector::new(workload.options.seed,
sim.replica_count);
run_with_faults(
@@ -718,11 +719,16 @@ pub fn run(
tick_budget,
replies_target,
&mut injector,
+ invariants,
)
}
/// [`run`] against a caller-owned [`FaultInjector`], so a test can assert what
/// was actually injected instead of trusting the probabilities to have fired.
+///
+/// [`Invariants`] is caller-owned for a second reason: the drain that follows
this
+/// call carries on with the same checker, and its high-water marks and
canonical
+/// commit chain are what let a regression spanning the two phases be seen at
all.
/// # Panics
/// If `injector` was built for a different replica count than `sim` has.
pub fn run_with_faults(
@@ -732,6 +738,7 @@ pub fn run_with_faults(
tick_budget: u64,
replies_target: u64,
injector: &mut FaultInjector,
+ invariants: &mut Invariants,
) -> u64 {
// The injector is caller-owned, and it sized `last_transition` from a
count
// nobody has checked against this simulator. Left unchecked the mismatch
@@ -747,7 +754,6 @@ pub fn run_with_faults(
sim.replica_count,
workload.options.seed,
);
- let mut invariants = Invariants::new();
let mut replies_seen = 0u64;
for _ in 0..tick_budget {
workload.tick();
@@ -1109,7 +1115,15 @@ mod tests {
// The recovery has to leave a usable session behind. Without a fresh
// registration the next request is refused with another eviction and
// nothing commits.
- let replies = run(&mut sim, &mut workload, &clients, 400, u64::MAX);
+ let mut invariants = Invariants::new();
+ let replies = run(
+ &mut sim,
+ &mut workload,
+ &clients,
+ 400,
+ u64::MAX,
+ &mut invariants,
+ );
assert!(replies > 0, "the recovered client got no replies");
assert!(
workload
diff --git a/core/simulator/src/workload/oracle.rs
b/core/simulator/src/workload/oracle.rs
index 18b4b714d..e3d1763ec 100644
--- a/core/simulator/src/workload/oracle.rs
+++ b/core/simulator/src/workload/oracle.rs
@@ -36,6 +36,7 @@
use crate::Simulator;
use crate::replica::Replica;
+use crate::workload::invariants::Invariants;
use crate::workload::shadow::Shadow;
use crate::workload::{Workload, apply_sim_commands, resubmit_due,
state_checker};
use consensus::{Consensus, MetadataHandle, Status};
@@ -102,8 +103,19 @@ impl CommittedMetadata {
///
/// Returns `true` once drained, `false` if `max_ticks` elapses with requests
/// still outstanding (a liveness failure the caller should surface).
+///
+/// `invariants` is the checker the active phase ran, carried in rather than
built
+/// here. A drain is up to 50,000 ticks of a cluster still repairing itself,
so a
+/// wedge that forms during it used to surface as nothing more than "did not
drain",
+/// and a fresh checker would start with an empty commit chain, which is the
memory
+/// that names a divergence.
#[must_use]
-pub fn drive_to_quiesce(sim: &mut Simulator, workload: &mut Workload,
max_ticks: u64) -> bool {
+pub fn drive_to_quiesce(
+ sim: &mut Simulator,
+ workload: &mut Workload,
+ max_ticks: u64,
+ invariants: &mut Invariants,
+) -> bool {
let mut drained = false;
for _ in 0..max_ticks {
// The drain keeps resending: a request lost on the way out is never
@@ -122,6 +134,7 @@ pub fn drive_to_quiesce(sim: &mut Simulator, workload: &mut
Workload, max_ticks:
for client_id in sim.take_evictions() {
workload.forget_evicted_client(client_id);
}
+ invariants.check(sim, workload);
if workload.total_in_flight() == 0 {
drained = true;
break;
@@ -135,6 +148,7 @@ pub fn drive_to_quiesce(sim: &mut Simulator, workload: &mut
Workload, max_ticks:
let cmds = workload.on_reply(&reply);
apply_sim_commands(sim, &cmds);
}
+ invariants.check(sim, workload);
}
true
}
@@ -251,8 +265,16 @@ pub fn quiesce_failure_report(sim: &Simulator, workload:
&Workload) -> String {
///
/// Returns `false` if the views never converge, which is a real liveness
failure
/// the caller should report rather than assert against an unsettled cluster.
+///
+/// Runs `invariants` per tick for the same reason [`drive_to_quiesce`] does:
this is
+/// another 50,000-tick window, and it is the one a view change wedges in.
#[must_use]
-pub fn settle_to_stable_view(sim: &mut Simulator, workload: &mut Workload,
max_ticks: u64) -> bool {
+pub fn settle_to_stable_view(
+ sim: &mut Simulator,
+ workload: &mut Workload,
+ max_ticks: u64,
+ invariants: &mut Invariants,
+) -> bool {
for _ in 0..max_ticks {
if views_are_settled(sim, workload) {
return true;
@@ -263,6 +285,7 @@ pub fn settle_to_stable_view(sim: &mut Simulator, workload:
&mut Workload, max_t
let cmds = workload.on_reply(&reply);
apply_sim_commands(sim, &cmds);
}
+ invariants.check(sim, workload);
}
views_are_settled(sim, workload)
}
diff --git a/core/simulator/src/workload/state_checker.rs
b/core/simulator/src/workload/state_checker.rs
index 53995a6ec..cc2e501dd 100644
--- a/core/simulator/src/workload/state_checker.rs
+++ b/core/simulator/src/workload/state_checker.rs
@@ -35,7 +35,7 @@ use crate::Simulator;
use iggy_binary_protocol::PrepareHeader;
use journal::Journal;
use server_common::sharding::IggyNamespace;
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::BTreeMap;
/// One op of the canonical committed chain.
#[derive(Debug)]
@@ -47,9 +47,16 @@ struct CanonicalCommit {
/// arriving header's `parent` must equal the canonical previous op's
`checksum`,
/// so keeping each entry's own parent would record a value nothing 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>,
+ /// Replicas observed committing this op, each against the metadata
incarnation
+ /// it held at the time, so the check can prove it compared something
rather
+ /// than passing over an empty chain.
+ ///
+ /// The incarnation is what separates two failure modes the checksum alone
+ /// reports identically: two replicas holding different history at one op,
and
+ /// ONE replica reporting a different header there after a restart. The
second
+ /// means it applied one entry and recovered another, which is narrower and
+ /// wants naming as such.
+ witnesses: BTreeMap<u8, u128>,
}
/// Canonical committed metadata chain, accumulated across ticks.
@@ -154,7 +161,14 @@ impl StateChecker {
);
return;
};
- self.record(replica_idx, op, &header, seed);
+ self.record(
+ replica_idx,
+ op,
+ &header,
+ replica.metadata_incarnation,
+ committed,
+ seed,
+ );
}
/// Number of ops in the canonical chain. Tests assert this is non-zero,
so a
@@ -170,11 +184,19 @@ impl StateChecker {
pub fn ops_compared(&self) -> usize {
self.commits
.values()
- .filter(|commit| commit.replicas.len() > 1)
+ .filter(|commit| commit.witnesses.len() > 1)
.count()
}
- fn record(&mut self, replica_idx: u8, op: u64, header: &PrepareHeader,
seed: u64) {
+ fn record(
+ &mut self,
+ replica_idx: u8,
+ op: u64,
+ header: &PrepareHeader,
+ incarnation: u128,
+ committed: u64,
+ 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.
@@ -207,21 +229,35 @@ impl StateChecker {
}
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,
+ // Named separately because the fix differs: a replica
disagreeing
+ // with its own earlier incarnation applied one entry and
recovered
+ // another, and no other replica has to be involved.
+ let restarted_since = canonical
+ .witnesses
+ .get(&replica_idx)
+ .is_some_and(|&recorded| recorded != incarnation);
+ assert!(
+ canonical.checksum == header.checksum,
+ "{} on committed op {op}: canonical checksum {:#x}
(witnesses \
+ {:?}) vs replica {replica_idx}'s {:#x} at incarnation \
+ {incarnation}, commit point {committed} (seed={seed:#x})",
+ if restarted_since {
+ "one replica disagrees with its own pre-restart
history"
+ } else {
+ "replicas committed different history at the same log
position"
+ },
+ canonical.checksum,
+ canonical.witnesses,
+ header.checksum,
);
- canonical.replicas.insert(replica_idx);
+ canonical.witnesses.insert(replica_idx, incarnation);
}
None => {
self.commits.insert(
op,
CanonicalCommit {
checksum: header.checksum,
- replicas: BTreeSet::from([replica_idx]),
+ witnesses: BTreeMap::from([(replica_idx,
incarnation)]),
},
);
}
