diff --git a/core/integration/tests/cluster/partition_state_transfer.rs b/core/integration/tests/cluster/partition_state_transfer.rs index 36716ffa65..4decebf09e 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 = (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, + acked: Arc, + 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 { 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 1c35bb54f0..e6b1662001 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::()..], + 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) { + 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 = 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::()..], + |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::::new(size_of::()) + .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::::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::::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 { + 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::(); + let total = header_size + size_of_val(headers.as_slice()); + let mut message = Message::::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 { + Message::::new(size_of::()).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::()).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::()).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 1992214806..4d5329eecb 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 43984693cb..890d62a334 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 ba648636ab..30417e2004 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 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(|_| "")) + .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) { - 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::() < 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 0000000000..0c22f42e33 --- /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 = 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) +}