You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2026/10/26 15:20:39 UTC
(iggy) branch repro/slow-replica-transfer updated (90403cec9 -> a289768b7)
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a change to branch repro/slow-replica-transfer
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 90403cec9 test(partitions): expose bugs when a slow replica rejoins the cluster
new a289768b7 test(partitions): expose bugs when a slow replica rejoins the cluster
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (90403cec9)
\
N -- N -- N refs/heads/repro/slow-replica-transfer (a289768b7)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 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.
Summary of changes:
.../tests/cluster/partition_state_transfer.rs | 15 ++++++-
core/shard/src/lib.rs | 52 ++++++++++++++--------
2 files changed, 47 insertions(+), 20 deletions(-)
(iggy) 01/01: test(partitions): expose bugs when a slow replica rejoins the cluster
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch repro/slow-replica-transfer
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit a289768b7e2ed0222507d65525a5492cecb6c4b2
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Mon Oct 26 15:39:07 2026 +0100
test(partitions): expose bugs when a slow replica rejoins the cluster
A replica that lags or runs slow can disturb its partition after the
fault ends. A state transfer it installs can leave the partition in an
endless view change. Add a test for that failure and for two related
ones. Each test is ignored as a known bug, so CI stays green until the
fixes land.
After a replica adopts a `StartView` and then installs a transfer
below the view head, the install rewinds the sequencer but keeps the
view. Its next `DoViewChange` carries the transfer op in the header
but starts the suffix at the old head, so peers drop it as
unreadable. If the transfer is more than 128 ops below the head, the
vote also leaves uncommitted ops outside its window. Two shard unit
tests replay the consensus steps of the install, build the vote, and
decode it as a peer does.
Under sustained writes, a backup that fell past the evicted ring
never installs a transfer. The first request pins the offer at the
primary's commit point, and the pull runs one retry interval later.
By then the backup journaled a newer op and refuses the offer as
stale. An integration spec cuts node 2 off, writes one message per
second after the heal, and waits for an install.
A replica whose messages arrive later than the view-change status
timeout keeps the two healthy replicas leaving their working view. A
simulator test shows this with latency alone. For this, the packet
simulator gains a directed extra delay per link. The delay does not
draw from the PRNG, so a run with zero extra delay replays the same
trace.
---
.../tests/cluster/partition_state_transfer.rs | 201 +++++++++++++++++-
core/shard/src/lib.rs | 234 +++++++++++++++++++++
core/simulator/src/lib.rs | 3 +
core/simulator/src/network.rs | 6 +
core/simulator/src/packet.rs | 76 ++++++-
core/simulator/src/slow_replica_tests.rs | 148 +++++++++++++
6 files changed, 663 insertions(+), 5 deletions(-)
diff --git a/core/integration/tests/cluster/partition_state_transfer.rs b/core/integration/tests/cluster/partition_state_transfer.rs
index 36716ffa6..4decebf09 100644
--- a/core/integration/tests/cluster/partition_state_transfer.rs
+++ b/core/integration/tests/cluster/partition_state_transfer.rs
@@ -27,13 +27,18 @@
//! `RangeEvicted`, the repaired window cannot connect to recovered state,
//! and `complete_repair` returns the `FloorRefused` conversion trigger.
+use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::str::FromStr;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};
use iggy::prelude::*;
-use integration::harness::TestHarness;
+use integration::harness::{TestHarness, TestServerConfig, resolve_config_paths};
use integration::iggy_harness;
+use serial_test::parallel;
+use tokio::task::JoinHandle;
use tokio::time::sleep;
const STREAM_NAME: &str = "partition-transfer-stream";
@@ -68,6 +73,9 @@ const KILL_GATE_POLL: Duration = Duration::from_millis(10);
const CONVERSION_MARKER: &str = "partition repair floor unreachable; converting to state transfer";
const INSTALL_MARKER: &str = "partition state transfer installed";
+const INSTALL_FAILED_MARKER: &str = "partition state transfer install failed";
+/// The `StaleTransfer` reason, logged inside an [`INSTALL_FAILED_MARKER`] line.
+const STALE_INSTALL_MARKER: &str = "is below the local commit frontier";
/// Logged once per transfer, at the last byte of the last artifact. The
/// descriptor-time "serving" line only proves a request ARRIVED; this proves the
/// pull ran.
@@ -84,6 +92,17 @@ const ABANDON_MARKER: &str =
const REARM_ON_VIEW_MARKER: &str =
"adopted a live view while awaiting transfer; requesting partition state transfer";
+/// Sustained-write spec: the node cut off and then healed.
+const LAGGING_NODE: usize = 2;
+/// Sends committed while node 2 is cut off: more than the evicted ring (capacity
+/// 64) holds, so its repair after the heal cannot reach its own commit point.
+const ISOLATED_SENDS: u32 = 100;
+/// Pace of the writer that keeps running after the heal. One message a second
+/// already moves the commit point past every offer before its pull lands.
+const WRITE_INTERVAL: Duration = Duration::from_secs(1);
+/// How long node 2 must go without another conversion once it installed.
+const CAUGHT_UP_WINDOW: Duration = Duration::from_secs(10);
+
/// Transfer end-to-end: adoption, repair round-trip, conversion, chunk pull,
/// install, tail repair. CI runners are slow; bound without hanging the suite.
const TRANSFER_BUDGET: Duration = Duration::from_secs(60);
@@ -353,6 +372,103 @@ async fn given_transfer_peer_dies_when_stalled_should_leave_dead_peer_and_recove
}
}
+/// A backup that falls past the evicted ring must rejoin through one state
+/// transfer while clients keep writing, and then stay caught up.
+///
+/// Node 2 is cut off while the partition commits more ops than the ring holds,
+/// so its repair after the heal hits `RangeEvicted` and converts to a transfer.
+/// One message per second keeps arriving for the whole run. That is enough to
+/// keep the cluster's commit point moving during every pull. No bulky seed is
+/// needed: the pull itself is a few KiB.
+#[tokio::test(flavor = "multi_thread")]
+#[parallel]
+#[ignore = "known bug: under sustained writes the backup refuses every transfer as stale (transfer frontier below the local commit frontier); run with --run-ignored only"]
+async fn given_sustained_writes_when_a_node_falls_past_the_evicted_ring_should_rejoin_through_one_transfer()
+ {
+ let mut harness = relayed_cluster();
+ harness.start().await.expect("start the cluster");
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client to the node");
+ seed_topic(&client, None).await;
+
+ harness
+ .isolate_node(LAGGING_NODE)
+ .expect("cut node 2 off from its peers");
+ produce(&client, ISOLATED_SENDS).await;
+ let writer = SteadyWriter::start(
+ harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect the steady writer"),
+ );
+ harness
+ .heal_node(LAGGING_NODE)
+ .expect("restore node 2's peer links");
+ await_marker(&harness, LAGGING_NODE, CONVERSION_MARKER).await;
+
+ // Today every install is refused with "transfer frontier X is below the
+ // local commit frontier X+1". The first request only pins the offer at the
+ // primary's commit point (the build answers "flush is pending"), and the
+ // pull runs on the next request, one `repair_retry_interval` later. Node 2
+ // journals the op committed in between, and the install refuses an offer
+ // below an op it holds.
+ let deadline = Instant::now() + TRANSFER_BUDGET;
+ while harness
+ .node(LAGGING_NODE)
+ .stdout_occurrences(INSTALL_MARKER)
+ == 0
+ {
+ assert!(
+ Instant::now() < deadline,
+ "node 2 fell past the evicted ring and never installed a state transfer within \
+ {TRANSFER_BUDGET:?} while one message per second was written: the peers fully \
+ served {} pull(s) and node 2 failed {} install(s), {} of them as stale, so a \
+ backup cannot rejoin under sustained writes",
+ fully_served_count(&harness),
+ harness
+ .node(LAGGING_NODE)
+ .stdout_occurrences(INSTALL_FAILED_MARKER),
+ harness
+ .node(LAGGING_NODE)
+ .stdout_occurrences(STALE_INSTALL_MARKER),
+ );
+ sleep(MARKER_POLL).await;
+ }
+
+ // Installed: the tail above the offer is now ordinary journal repair. A
+ // backup that keeps up never converts to a transfer again.
+ let conversions = harness
+ .node(LAGGING_NODE)
+ .stdout_occurrences(CONVERSION_MARKER);
+ let acked_at_install = writer.acked();
+ let window_end = Instant::now() + CAUGHT_UP_WINDOW;
+ while Instant::now() < window_end {
+ let converted = harness
+ .node(LAGGING_NODE)
+ .stdout_occurrences(CONVERSION_MARKER)
+ - conversions;
+ assert_eq!(
+ converted, 0,
+ "node 2 installed a state transfer, then fell past the evicted ring again and \
+ converted to another transfer while writes continued"
+ );
+ sleep(MARKER_POLL).await;
+ }
+ // The markers only say the machinery ran. Every send committed while node
+ // 2 was cut off must be in its own segment bytes, in order.
+ let expected: Vec<String> = (0..ISOLATED_SENDS)
+ .map(|sequence| format!("message-{sequence}"))
+ .collect();
+ await_installed_payloads(&harness, LAGGING_NODE, &expected).await;
+ let acked = writer.stop().await;
+ assert!(
+ acked > acked_at_install,
+ "the steady writer must keep committing through the caught-up window"
+ );
+}
+
async fn restart_lagging_node_and_await_install(harness: &mut TestHarness) {
// Node 2 holds a durable prefix, then misses enough traffic that the
// survivors' ring moves past its durable end: its repaired window cannot
@@ -395,6 +511,89 @@ async fn restart_lagging_node_and_await_install(harness: &mut TestHarness) {
await_installed_payloads(harness, 2, &expected).await;
}
+/// The specs above go through `iggy_harness`, which cannot relay the replica
+/// links, so the one that cuts a node off builds its cluster here with the same
+/// server settings.
+fn relayed_cluster() -> TestHarness {
+ let overrides = HashMap::from([
+ ("sharding.cpu_allocation".to_owned(), "0..1".to_owned()),
+ (
+ "partition.evicted_ring_capacity".to_owned(),
+ "64".to_owned(),
+ ),
+ ]);
+ let extra_envs = resolve_config_paths(&overrides).expect("valid server config paths");
+ TestHarness::builder()
+ .server(TestServerConfig::builder().extra_envs(extra_envs).build())
+ .cluster_nodes(3)
+ .peer_relays()
+ .build()
+ .expect("build the harness")
+}
+
+/// Sends one small message per [`WRITE_INTERVAL`] from its own task until
+/// stopped. A failed send is not retried: the next tick sends a new message,
+/// and only acknowledged sends are counted.
+struct SteadyWriter {
+ stop: Arc<AtomicBool>,
+ acked: Arc<AtomicU64>,
+ task: JoinHandle<()>,
+}
+
+impl SteadyWriter {
+ fn start(client: IggyClient) -> Self {
+ let stop = Arc::new(AtomicBool::new(false));
+ let acked = Arc::new(AtomicU64::new(0));
+ let task = tokio::spawn({
+ let stop = Arc::clone(&stop);
+ let acked = Arc::clone(&acked);
+ async move {
+ let stream = Identifier::named(STREAM_NAME).expect("stream identifier");
+ let topic = Identifier::named(TOPIC_NAME).expect("topic identifier");
+ let mut sequence = 0_u64;
+ while !stop.load(Ordering::Relaxed) {
+ let mut messages = vec![
+ IggyMessage::from_str(&format!("steady-{sequence}")).expect("message"),
+ ];
+ sequence += 1;
+ if client
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .is_ok()
+ {
+ acked.fetch_add(1, Ordering::Relaxed);
+ }
+ sleep(WRITE_INTERVAL).await;
+ }
+ }
+ });
+ Self { stop, acked, task }
+ }
+
+ fn acked(&self) -> u64 {
+ self.acked.load(Ordering::Relaxed)
+ }
+
+ async fn stop(self) -> u64 {
+ self.stop.store(true, Ordering::Relaxed);
+ self.task.await.expect("the steady writer task ends");
+ self.acked.load(Ordering::Relaxed)
+ }
+}
+
+/// Pulls the survivors served to their last byte, whoever served them.
+fn fully_served_count(harness: &TestHarness) -> usize {
+ (0..harness.cluster_size())
+ .filter(|&node| node != LAGGING_NODE)
+ .map(|node| harness.node(node).stdout_occurrences(FULLY_SERVED_MARKER))
+ .sum()
+}
+
async fn connect_any(harness: &TestHarness, nodes: &[usize]) -> Option<IggyClient> {
for &node in nodes {
if let Ok(builder) = harness.node(node).tcp_client()
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 1c35bb54f..e6b166200 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -14498,6 +14498,240 @@ mod dvc_suffix_window_tests {
}
}
+#[cfg(test)]
+mod installed_transfer_dvc_tests {
+ //! A lagging replica adopts a `StartView` and later installs a state
+ //! transfer below that view's head. The install rewinds the sequencer to
+ //! the transfer's op and keeps the adopted view. Each `DoViewChange` the
+ //! replica sends from then on carries the transfer op in its header but
+ //! starts its suffix at the adopted head, and its peers drop every one as
+ //! unreadable. With a third replica down, or back with an untrusted log,
+ //! no quorum of readable votes forms and the view change never ends.
+
+ use super::*;
+ use consensus::{LocalPipeline, dvc_suffix_decode};
+ use iggy_common::{IggyByteSize, PartitionStats};
+ use journal::prepare_journal::PrepareJournal;
+ use message_bus::IggyMessageBus;
+ use server_common::iobuf::Owned;
+ use std::sync::Arc;
+
+ const CLUSTER: u128 = 1;
+ const REPLICA_COUNT: u8 = 3;
+ const GROUP: u64 = 42;
+ /// Adopts the view, then installs the transfer.
+ const LAGGING: u8 = 2;
+ /// Led by replica 0.
+ const ADOPTED_VIEW: u32 = 3;
+ /// Led by replica 1, which receives the lagging replica's vote.
+ const NEXT_VIEW: u32 = 4;
+ const HEAD: u64 = 300;
+ /// Two below the head, so the adopted view still carries uncommitted ops.
+ const COMMIT: u64 = HEAD - 2;
+
+ #[compio::test]
+ #[ignore = "known bug: a transfer installed below the adopted view head makes the replica's DoViewChange unreadable; run with --ignored"]
+ async fn given_a_transfer_installed_below_the_adopted_head_when_a_view_change_follows_should_send_a_readable_do_view_change()
+ {
+ // Inside the 128-entry window, so only the head disagrees.
+ let transfer = HEAD - 100;
+ // Boxed: the setup future carries a whole partition.
+ let (vote, message) = Box::pin(vote_after_install(transfer)).await;
+ let VsrAction::SendDoViewChange { suffix, .. } = &vote else {
+ unreachable!("matched in vote_after_install");
+ };
+ let header = message.header();
+ // What `decode_peer_suffix` runs on the receiving replica.
+ if let Err(error) = dvc_suffix_decode(
+ &message.as_slice()[size_of::<DoViewChangeHeader>()..],
+ header.op,
+ header.nack_bitset,
+ header.present_bitset,
+ ) {
+ panic!(
+ "replica {LAGGING} adopted a view with head {HEAD} and then installed a state \
+ transfer at op {transfer}. Its DoViewChange carries op {} in the header but starts \
+ its suffix at op {}, so the receiver drops it as unreadable ({error}) and the \
+ view change cannot complete without a vote from the third replica",
+ header.op,
+ suffix.headers().first().map_or(0, |entry| entry.op),
+ );
+ }
+ }
+
+ #[compio::test]
+ #[ignore = "known bug: a transfer installed far below the adopted view head leaves uncommitted ops outside the DoViewChange window; run with --ignored"]
+ async fn given_a_transfer_installed_far_below_the_adopted_head_when_a_view_change_follows_should_describe_every_uncommitted_op()
+ {
+ // Past the 128-entry window.
+ let transfer = HEAD - 200;
+ let (vote, _) = Box::pin(vote_after_install(transfer)).await;
+ let VsrAction::SendDoViewChange { commit, suffix, .. } = &vote else {
+ unreachable!("matched in vote_after_install");
+ };
+ let floor = suffix.headers().last().map_or(0, |entry| entry.op);
+ assert!(
+ floor <= *commit,
+ "replica {LAGGING}'s DoViewChange window starts at op {floor}, above its commit point \
+ {commit}, so ops {}..={} go out described by nobody and the view change cannot decide \
+ them. The install put the sequencer {} ops below the adopted head, past the \
+ {DVC_HEADERS_MAX}-entry window that `build_dvc_suffix` treats as unreachable",
+ commit + 1,
+ floor.saturating_sub(1),
+ HEAD - transfer,
+ );
+ }
+
+ /// The lagging replica's vote for `NEXT_VIEW`, and the frame the shard sends
+ /// for it. Before the vote, the replica adopts the `ADOPTED_VIEW` `StartView`
+ /// and then installs a peer's state transfer at `transfer`.
+ #[allow(clippy::future_not_send)]
+ async fn vote_after_install(transfer: u64) -> (VsrAction, Message<DoViewChangeHeader>) {
+ let bus = IggyMessageBus::new(0);
+ let sent = Rc::new(RefCell::new(Vec::new()));
+ let captured = Rc::clone(&sent);
+ bus.set_replica_forward_fn(Box::new(move |_, _, frame| {
+ captured.borrow_mut().push(frame);
+ Ok(())
+ }));
+ for peer in (0..REPLICA_COUNT).filter(|&peer| peer != LAGGING) {
+ assert!(bus.owner_table().try_claim(peer, 1));
+ }
+ let consensus = VsrConsensus::new(
+ CLUSTER,
+ LAGGING,
+ REPLICA_COUNT,
+ GROUP,
+ bus,
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ let mut partition: IggyPartition<IggyMessageBus> = IggyPartition::with_in_memory_storage(
+ Arc::new(PartitionStats::default()),
+ consensus,
+ IggyByteSize::from(1024 * 1024),
+ );
+
+ // `on_start_view`: adopt, then reconcile the journal against the view.
+ let announce = start_view();
+ let actions = partition
+ .consensus()
+ .try_handle_start_view(
+ PlaneKind::Partitions,
+ announce.header(),
+ &announce.as_slice()[size_of::<StartViewHeader>()..],
+ |op| partition.applied_header(op),
+ )
+ .expect("the StartView is well formed");
+ assert!(!actions.is_empty(), "the lagging replica adopts the view");
+ let pending = partition.consensus().pending_view_log();
+ reconcile_partition_view_divergence(0, &mut partition, pending.as_ref()).await;
+
+ // Repair finds the gap evicted at every peer, so a transfer follows. The
+ // install cannot run here: its offsets artifact is a type private to
+ // `partitions`. So these are its journal wipe and then the steps of
+ // `publish_installed_consensus`, in its order, with a checkpoint prepare
+ // built here in place of the origin's. They rewind the sequencer to
+ // `transfer` and keep the adopted view.
+ let checkpoint = Message::<PrepareHeader>::new(size_of::<PrepareHeader>())
+ .transmute_header(|_, header: &mut PrepareHeader| {
+ *header = prepare_header(transfer, 0);
+ });
+ partition.log.journal().inner.clear_all();
+ partition.log.journal().inner.restore_checkpoint_prepare(
+ transfer,
+ Owned::<MESSAGE_ALIGN>::copy_from_slice(checkpoint.as_slice()).into(),
+ );
+ let consensus = partition.consensus();
+ consensus.set_commit_floor(transfer);
+ consensus.sequencer().set_sequence(transfer);
+ consensus.set_last_prepare_checksum(checkpoint.header().checksum);
+ consensus.clear_pipeline();
+ consensus.advance_commit_max(transfer);
+
+ // `on_start_view_change`: snapshot the suffix, then vote.
+ refresh_partition_dvc_suffix(&partition);
+ let consensus = partition.consensus();
+ let actions =
+ consensus.handle_start_view_change(PlaneKind::Partitions, start_view_change().header());
+ let vote = actions
+ .iter()
+ .find(|action| matches!(action, VsrAction::SendDoViewChange { .. }))
+ .cloned()
+ .expect("the lagging replica votes");
+ dispatch_partition_wire_actions::<_, _, PrepareJournal, _>(consensus, &partition, actions)
+ .await;
+ let message = sent
+ .borrow()
+ .iter()
+ .find_map(|frame| {
+ Message::<DoViewChangeHeader>::try_from(Owned::copy_from_slice(frame.as_slice()))
+ .ok()
+ })
+ .expect("the vote goes out on the wire");
+ (vote, message)
+ }
+
+ /// The `ADOPTED_VIEW` `StartView`: its head and commit point, with the
+ /// hash-chained headers of `COMMIT..=HEAD`, high-to-low.
+ fn start_view() -> Message<StartViewHeader> {
+ let mut headers = Vec::new();
+ let mut parent = 0;
+ for op in COMMIT..=HEAD {
+ let header = prepare_header(op, parent);
+ parent = header.checksum;
+ headers.push(header);
+ }
+ headers.reverse();
+ let header_size = size_of::<StartViewHeader>();
+ let total = header_size + size_of_val(headers.as_slice());
+ let mut message = Message::<StartViewHeader>::new(total);
+ encode_prepare_headers(&headers, &mut message.as_mut_slice()[header_size..total]);
+ message.transmute_header(|_, header: &mut StartViewHeader| {
+ header.command = Command::StartView;
+ header.cluster = CLUSTER;
+ header.view = ADOPTED_VIEW;
+ header.op = HEAD;
+ header.commit = COMMIT;
+ header.replica = 0;
+ header.group = GROUP;
+ header.size = u32::try_from(total).unwrap();
+ })
+ }
+
+ fn start_view_change() -> Message<StartViewChangeHeader> {
+ Message::<StartViewChangeHeader>::new(size_of::<StartViewChangeHeader>()).transmute_header(
+ |_, header: &mut StartViewChangeHeader| {
+ header.command = Command::StartViewChange;
+ header.cluster = CLUSTER;
+ header.view = NEXT_VIEW;
+ header.replica = 1;
+ header.group = GROUP;
+ header.size = u32::try_from(size_of::<StartViewChangeHeader>()).unwrap();
+ },
+ )
+ }
+
+ /// A body-less prepare header at `op` in `ADOPTED_VIEW`, chained to `parent`.
+ fn prepare_header(op: u64, parent: u128) -> PrepareHeader {
+ let mut header = PrepareHeader {
+ command: Command::Prepare,
+ cluster: CLUSTER,
+ group: GROUP,
+ operation: Operation::StoreConsumerOffset,
+ size: u32::try_from(size_of::<PrepareHeader>()).unwrap(),
+ op,
+ commit: op - 1,
+ timestamp: op,
+ parent,
+ view: ADOPTED_VIEW,
+ ..PrepareHeader::default()
+ };
+ header.checksum = header.identity_checksum();
+ header
+ }
+}
+
#[cfg(test)]
mod adopted_view_retention_tests {
//! The view 12 stall of workload-fuzz seed 890 at 45a4ea297. Replica 0
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 199221480..4d5329eec 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -2221,6 +2221,9 @@ fn stamp_metadata_watermark(
}
}
+#[cfg(test)]
+mod slow_replica_tests;
+
#[cfg(test)]
mod stale_prepare_commit_tests;
diff --git a/core/simulator/src/network.rs b/core/simulator/src/network.rs
index 43984693c..890d62a33 100644
--- a/core/simulator/src/network.rs
+++ b/core/simulator/src/network.rs
@@ -153,6 +153,12 @@ impl Network {
self.simulator.link_drop_packet_fn(from, to)
}
+ /// Returns a mutable reference to the link's extra one-way delay, in ticks.
+ /// See [`PacketSimulator::link_extra_delay`].
+ pub fn link_extra_delay(&mut self, from: ProcessId, to: ProcessId) -> &mut u64 {
+ self.simulator.link_extra_delay(from, to)
+ }
+
/// Clog a specific link (bidirectionally).
///
/// Clogged links do not deliver any packets until unclogged.
diff --git a/core/simulator/src/packet.rs b/core/simulator/src/packet.rs
index ba648636a..30417e200 100644
--- a/core/simulator/src/packet.rs
+++ b/core/simulator/src/packet.rs
@@ -19,6 +19,7 @@
//!
//! This module provides a packet-level network simulator that models:
//! - Per-path latency with exponential delay distribution
+//! - Per-link extra latency, directed, set by the caller
//! - Packet loss and replay (duplication) at delivery time
//! - Automatic network partitioning with configurable lifecycle
//! - Automatic path clogging with exponential duration
@@ -237,6 +238,9 @@ struct Link {
/// Optional predicate to drop specific packets. Checked after the link filter
/// but before the random loss check.
drop_packet_fn: Option<fn(&Packet) -> bool>,
+ /// Ticks added to the drawn one-way delay of every packet submitted on this
+ /// link, replays included. Zero by default.
+ extra_delay: u64,
}
#[allow(clippy::missing_fields_in_debug)]
@@ -247,6 +251,7 @@ impl std::fmt::Debug for Link {
.field("clogged_till", &self.clogged_till)
.field("filter", &self.filter)
.field("drop_packet_fn", &self.drop_packet_fn.map(|_| "<fn>"))
+ .field("extra_delay", &self.extra_delay)
.finish()
}
}
@@ -258,6 +263,7 @@ impl Link {
clogged_till: 0,
filter: ALLOW_ALL,
drop_packet_fn: None,
+ extra_delay: 0,
}
}
}
@@ -527,7 +533,9 @@ impl PacketSimulator {
/// time in `step()`. If the link is at capacity, a random existing packet
/// is evicted to make room.
pub fn submit(&mut self, from: ProcessId, to: ProcessId, message: Message<GenericHeader>) {
- let delay = Self::calculate_delay(&mut self.prng, &self.options);
+ let idx = self.link_index(from, to);
+ let delay = Self::calculate_delay(&mut self.prng, &self.options)
+ .saturating_add(self.links[idx].extra_delay);
let ready_at = self.current_tick.saturating_add(delay);
let packet = Packet {
@@ -537,8 +545,6 @@ impl PacketSimulator {
ready_at,
};
- let idx = self.link_index(from, to);
-
// If at capacity, evict a random existing packet
if self.links[idx].packets.len() >= self.options.link_capacity as usize {
self.links[idx].packets.remove_random(&mut self.prng);
@@ -614,6 +620,21 @@ impl PacketSimulator {
&mut self.links[idx].drop_packet_fn
}
+ /// Returns a mutable reference to the link's extra one-way delay, in ticks.
+ ///
+ /// Directed: only packets from `from` to `to` wait longer. The extra delay
+ /// is added to the delay drawn from `one_way_delay_min` and
+ /// `one_way_delay_mean` without drawing from the PRNG, so a run with every
+ /// extra delay at zero replays the same trace. Setting it on every link out
+ /// of one replica models a slow egress on that node, as `tc netem delay`
+ /// does. A delayed link holds its packets longer, so `link_capacity` evicts
+ /// sooner on it. [`Self::heal`] leaves it in place, like
+ /// [`Self::link_drop_packet_fn`].
+ pub fn link_extra_delay(&mut self, from: ProcessId, to: ProcessId) -> &mut u64 {
+ let idx = self.link_index(from, to);
+ &mut self.links[idx].extra_delay
+ }
+
/// Mark a process down. Anything addressed to it is dropped at delivery.
///
/// Link filters are untouched, so a partition or command filter standing at crash
@@ -732,7 +753,8 @@ impl PacketSimulator {
// Random replay check: clone and re-enqueue with eviction at capacity
if prng.random::<f64>() < options.replay_probability {
- let delay = Self::calculate_delay(prng, options);
+ let delay =
+ Self::calculate_delay(prng, options).saturating_add(link.extra_delay);
let replay = Packet {
ready_at: (*current_tick).saturating_add(delay),
..packet.clone()
@@ -1147,6 +1169,52 @@ mod tests {
assert_eq!(delivered.len(), 1);
}
+ /// An extra delay on one link holds back only that direction, by at least
+ /// the extra ticks, and the reverse link keeps the drawn delay.
+ #[test]
+ fn given_extra_link_delay_when_submitting_should_delay_only_that_direction() {
+ const EXTRA: u64 = 30;
+ let options = PacketSimulatorOptions {
+ one_way_delay_min: 1,
+ one_way_delay_mean: 1,
+ node_count: 2,
+ client_count: 0,
+ seed: 42,
+ ..Default::default()
+ };
+
+ let mut sim = PacketSimulator::new(options);
+ *sim.link_extra_delay(ProcessId::Replica(0), ProcessId::Replica(1)) = EXTRA;
+ sim.submit(
+ ProcessId::Replica(0),
+ ProcessId::Replica(1),
+ create_test_message(),
+ );
+ sim.submit(
+ ProcessId::Replica(1),
+ ProcessId::Replica(0),
+ create_test_message(),
+ );
+
+ let mut delivered_at = HashMap::new();
+ for tick in 0..=EXTRA + 20 {
+ for packet in sim.step() {
+ delivered_at.insert(packet.from, tick);
+ }
+ sim.tick();
+ }
+ let slow = delivered_at[&ProcessId::Replica(0)];
+ let fast = delivered_at[&ProcessId::Replica(1)];
+ assert!(
+ slow > EXTRA,
+ "the delayed link delivered at tick {slow}, before its extra {EXTRA} ticks"
+ );
+ assert!(
+ fast <= EXTRA,
+ "the reverse link delivered at tick {fast}, so it was delayed too"
+ );
+ }
+
/// A partition standing when a process crashes still stands when it restarts.
///
/// `process_enable` used to write `ALLOW_ALL` over every link touching the
diff --git a/core/simulator/src/slow_replica_tests.rs b/core/simulator/src/slow_replica_tests.rs
new file mode 100644
index 000000000..0c22f42e3
--- /dev/null
+++ b/core/simulator/src/slow_replica_tests.rs
@@ -0,0 +1,148 @@
+// 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.
+
+//! One slow replica must not keep the two healthy replicas changing views.
+//!
+//! Replica 2 is cut off for a moment, long enough to start a view change of its
+//! own, and every message it sends from then on takes longer than the
+//! view-change status timeout to arrive. Its `StartViewChange` for the next
+//! view therefore lands after it already timed out into the view after that.
+//! Replicas 0 and 1 are Normal and commit together, yet each such
+//! `StartViewChange` takes them out of their working view, and the primary
+//! keeps moving between them. Nothing but latency is injected after the cut
+//! heals: no packet is dropped.
+
+use server_common::sharding::IggyNamespace;
+
+use crate::Simulator;
+use crate::packet::ProcessId;
+use crate::partition_repair_driver_tests::cluster;
+use crate::workload::{
+ self, Workload,
+ invariants::Invariants,
+ options::{ActionWeights, WorkloadOptions},
+};
+
+const SEED: u64 = 0x5052_494D;
+const SLOW: u8 = 2;
+/// Groups the rebalancer spreads across the replicas, so each replica leads
+/// one before the fault.
+const GROUPS: usize = 3;
+/// Ticks for the rebalancer to spread the primaries.
+const SPREAD_TICKS: u64 = 10_000;
+/// Past `NORMAL_HEARTBEAT_TICKS` (500): the cut lasts long enough that the
+/// group replica 2 led fails over and replica 2 starts a view change itself.
+const CUT_TICKS: u64 = 600;
+/// One-way delay on every link out of replica 2, just past
+/// `VIEW_CHANGE_STATUS_TICKS` (500). Below that bound the healthy pair adopts
+/// replica 2's view before it times out again, and the views settle.
+const EGRESS_DELAY_TICKS: u64 = 520;
+/// Measured after the heal: 300 s of virtual time at 10 ms per tick.
+const MEASURE_TICKS: u64 = 30_000;
+/// The heal itself may cost a few view changes while replica 2 rejoins: 450
+/// ticks of delay costs up to 3 here, 520 ticks costs about 60.
+const VIEW_CHANGES_MAX: u32 = 5;
+
+#[test]
+#[ignore = "known bug: a replica slower than the view-change status timeout keeps the healthy pair leaving its working view; run with --ignored"]
+fn given_a_replica_slower_than_the_view_change_timeout_when_it_rejoins_should_not_keep_the_healthy_pair_changing_views()
+ {
+ let (mut sim, client) = cluster(SEED);
+ let namespaces: Vec<_> = (0..GROUPS)
+ .map(|partition| IggyNamespace::new(1, 1, partition))
+ .collect();
+ for &namespace in &namespaces {
+ sim.init_partition(namespace);
+ }
+ sim.register_client_with_primary(&client);
+ let mut options = WorkloadOptions::new(SEED, sim.replica_count, namespaces.clone());
+ options.weights = ActionWeights::partition_only();
+ let mut workload = Workload::new(options);
+ let clients = [client];
+ let mut invariants = Invariants::new();
+ let mut tick = |sim: &mut Simulator| {
+ workload::run(sim, &mut workload, &clients, 1, u64::MAX, &mut invariants)
+ };
+
+ for _ in 0..SPREAD_TICKS {
+ tick(&mut sim);
+ }
+ set_peer_links(&mut sim, false);
+ for _ in 0..CUT_TICKS {
+ tick(&mut sim);
+ }
+ set_peer_links(&mut sim, true);
+ for peer in (0..sim.replica_count).filter(|&peer| peer != SLOW) {
+ *sim.network
+ .link_extra_delay(ProcessId::Replica(SLOW), ProcessId::Replica(peer)) =
+ EGRESS_DELAY_TICKS;
+ }
+
+ let first_views: Vec<u32> = namespaces
+ .iter()
+ .map(|&namespace| healthy_view(&sim, namespace))
+ .collect();
+ let mut views = first_views.clone();
+ let mut changes = vec![0_u32; GROUPS];
+ let mut replies = 0;
+ for _ in 0..MEASURE_TICKS {
+ replies += tick(&mut sim);
+ for (index, &namespace) in namespaces.iter().enumerate() {
+ let view = healthy_view(&sim, namespace);
+ if view != views[index] {
+ changes[index] += 1;
+ views[index] = view;
+ }
+ }
+ }
+ assert!(
+ replies > 0,
+ "the workload must keep committing after the heal"
+ );
+ for (index, &namespace) in namespaces.iter().enumerate() {
+ assert!(
+ changes[index] <= VIEW_CHANGES_MAX,
+ "replicas 0 and 1 changed view {} times for {namespace:?} in {MEASURE_TICKS} ticks \
+ (view {} -> {}), while only replica 2 was slow: one replica whose messages \
+ arrive later than the view-change status timeout keeps the healthy pair leaving \
+ its working view (changes per group {changes:?})",
+ changes[index],
+ first_views[index],
+ views[index],
+ );
+ }
+}
+
+/// Cuts or restores every link between the slow replica and its peers.
+fn set_peer_links(sim: &mut Simulator, enabled: bool) {
+ for peer in (0..sim.replica_count).filter(|&peer| peer != SLOW) {
+ sim.network
+ .set_link_filter(ProcessId::Replica(SLOW), ProcessId::Replica(peer), enabled);
+ sim.network
+ .set_link_filter(ProcessId::Replica(peer), ProcessId::Replica(SLOW), enabled);
+ }
+}
+
+/// The highest view replicas 0 and 1 hold for `namespace`.
+fn healthy_view(sim: &Simulator, namespace: IggyNamespace) -> u32 {
+ (0..sim.replica_count)
+ .filter(|&replica| replica != SLOW)
+ .filter_map(|replica| sim.partition_consensus_state(usize::from(replica), namespace))
+ .map(|state| state.view)
+ .max()
+ .unwrap_or(0)
+}