You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/07/13 15:10:43 UTC
(iggy) branch message_repair_protocol updated: add tests that restarts all of the nodes
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch message_repair_protocol
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/message_repair_protocol by this push:
new 568c85916 add tests that restarts all of the nodes
568c85916 is described below
commit 568c85916a0e4e90e98a01096c4390de69c1861a
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Mon Jul 13 17:10:38 2026 +0200
add tests that restarts all of the nodes
---
core/consensus/src/impls.rs | 72 ++++++++++++----
core/consensus/src/observability.rs | 5 ++
.../src/harness/orchestrator/harness.rs | 30 +++++++
.../scenarios/reconnect_after_restart_scenario.rs | 99 ++++++++++++++++++++++
core/integration/tests/server/specific.rs | 12 +++
5 files changed, 200 insertions(+), 18 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index c8360ce32..a203767c8 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -101,6 +101,11 @@ pub const PIPELINE_REQUEST_QUEUE_MAX: usize = 64;
/// Maximum number of replicas in a cluster.
pub const REPLICAS_MAX: usize = 32;
+/// Unanswered `RequestStartView` probes tolerated before a recovering
+/// replica gives up waiting for a settled primary and falls back to an
+/// election (a full-cluster restart leaves nobody able to answer).
+pub const PROBE_ATTEMPTS_MAX: u32 = 5;
+
/// Maximum number of clients tracked in the clients table.
/// When exceeded, the client with the oldest committed request is evicted.
pub const CLIENTS_TABLE_MAX: usize = 8192;
@@ -702,6 +707,9 @@ where
loopback_queue: RefCell<VecDeque<Message<GenericHeader>>>,
/// Tracks start view change messages received from all replicas (including self)
start_view_change_from_all_replicas: RefCell<BitSet<u32>>,
+ /// Consecutive unanswered `RequestStartView` probes while Recovering;
+ /// at [`PROBE_ATTEMPTS_MAX`] the replica falls back to an election.
+ probe_attempts: Cell<u32>,
/// Tracks DVC messages received (only used by primary candidate)
/// Stores metadata; actual log comes from message
@@ -770,6 +778,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
message_bus,
loopback_queue: RefCell::new(VecDeque::with_capacity(PIPELINE_PREPARE_QUEUE_MAX)),
start_view_change_from_all_replicas: RefCell::new(BitSet::with_capacity(REPLICAS_MAX)),
+ probe_attempts: Cell::new(0),
do_view_change_from_all_replicas: RefCell::new(dvc_quorum_array_empty()),
do_view_change_quorum: Cell::new(false),
sent_own_start_view_change: Cell::new(false),
@@ -1204,24 +1213,49 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
// backstop burns a fresh cluster-wide election. The would-be
// primary of the view skips the probe (it concludes the view
// itself or escalates).
- let probing = match self.status.get() {
- Status::Recovering => true,
- Status::ViewChange => self.primary_index(self.view.get()) != self.replica,
- Status::Normal => false,
- };
- if probing {
- self.timeouts
- .borrow_mut()
- .reset(TimeoutKind::RequestStartViewMessage);
- actions.push(VsrAction::SendRequestStartView {
- namespace: self.namespace,
- });
- } else {
- // Stale arm (e.g. went Normal without passing an exit that
- // stops it): silence it instead of refiring every tick.
- self.timeouts
- .borrow_mut()
- .stop(TimeoutKind::RequestStartViewMessage);
+ match self.status.get() {
+ Status::Recovering => {
+ // A probe answered by nobody, repeatedly, means nobody is
+ // settled -- the whole cluster restarted together and
+ // every group sits quorum-invisible waiting for a primary
+ // that cannot exist. Fall back to an election: recovered
+ // WALs compete on (log_view, op) in the DVC exchange, so
+ // the best surviving log leads; a group whose members all
+ // rejoined journal-less elects on equal terms and stands
+ // on its recovered durable state. Any still-live settled
+ // primary answers well before the fallback fires.
+ let attempts = self.probe_attempts.get() + 1;
+ self.probe_attempts.set(attempts);
+ if attempts >= PROBE_ATTEMPTS_MAX {
+ self.finish_view_probe();
+ actions.extend(
+ self.start_election(plane, ViewChangeReason::ViewProbeUnanswered),
+ );
+ } else {
+ self.timeouts
+ .borrow_mut()
+ .reset(TimeoutKind::RequestStartViewMessage);
+ actions.push(VsrAction::SendRequestStartView {
+ namespace: self.namespace,
+ });
+ }
+ }
+ Status::ViewChange if self.primary_index(self.view.get()) != self.replica => {
+ self.timeouts
+ .borrow_mut()
+ .reset(TimeoutKind::RequestStartViewMessage);
+ actions.push(VsrAction::SendRequestStartView {
+ namespace: self.namespace,
+ });
+ }
+ _ => {
+ // Stale arm (e.g. went Normal without passing an exit
+ // that stops it): silence it instead of refiring every
+ // tick.
+ self.timeouts
+ .borrow_mut()
+ .stop(TimeoutKind::RequestStartViewMessage);
+ }
}
timeouts = self.timeouts.borrow_mut();
}
@@ -1818,6 +1852,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
"beginning view probe"
);
self.status.set(Status::Recovering);
+ self.probe_attempts.set(0);
{
let mut timeouts = self.timeouts.borrow_mut();
timeouts.stop(TimeoutKind::Prepare);
@@ -1895,6 +1930,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
}
fn finish_view_probe(&self) {
+ self.probe_attempts.set(0);
self.timeouts
.borrow_mut()
.stop(TimeoutKind::RequestStartViewMessage);
diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs
index df1dd8c69..03a9b1b08 100644
--- a/core/consensus/src/observability.rs
+++ b/core/consensus/src/observability.rs
@@ -61,6 +61,10 @@ pub enum ViewChangeReason {
/// peers elect without waiting for its heartbeats to stop arriving
/// (they already have).
PrimaryProbedView,
+ /// A recovering replica's `RequestStartView` probes all went unanswered:
+ /// nobody in the cluster is settled (full-cluster restart), so waiting
+ /// for a primary is futile -- elect instead.
+ ViewProbeUnanswered,
ViewChangeStatusTimeout,
ReceivedStartViewChange,
ReceivedDoViewChange,
@@ -72,6 +76,7 @@ impl ViewChangeReason {
match self {
Self::NormalHeartbeatTimeout => "normal_heartbeat_timeout",
Self::PrimaryProbedView => "primary_probed_view",
+ Self::ViewProbeUnanswered => "view_probe_unanswered",
Self::ViewChangeStatusTimeout => "view_change_status_timeout",
Self::ReceivedStartViewChange => "received_start_view_change",
Self::ReceivedDoViewChange => "received_do_view_change",
diff --git a/core/integration/src/harness/orchestrator/harness.rs b/core/integration/src/harness/orchestrator/harness.rs
index 31320235d..df1d545e5 100644
--- a/core/integration/src/harness/orchestrator/harness.rs
+++ b/core/integration/src/harness/orchestrator/harness.rs
@@ -330,6 +330,36 @@ impl TestHarness {
Ok(())
}
+ /// Restart EVERY node and reconnect all clients: the full-cluster
+ /// restart path, where no settled primary survives to answer the rejoin
+ /// probes and the replicas must fall back to an election among their
+ /// recovered logs. All nodes stop BEFORE any starts, so no rejoiner can
+ /// lean on a still-live peer (that would be a rolling restart).
+ pub async fn restart_cluster(&mut self) -> Result<(), TestBinaryError> {
+ if self.servers.is_empty() {
+ return Err(TestBinaryError::MissingServer);
+ }
+
+ for client in &mut self.clients {
+ client.disconnect().await;
+ }
+
+ for server in self.servers.iter_mut().rev() {
+ server.stop_dependents()?;
+ server.stop()?;
+ }
+ for server in &mut self.servers {
+ server.restart()?;
+ }
+
+ self.update_client_addresses();
+ for client in &mut self.clients {
+ client.connect().await?;
+ }
+
+ Ok(())
+ }
+
/// Get reference to the first (primary) server handle.
pub fn server(&self) -> &ServerHandle {
self.servers.first().expect("No servers configured")
diff --git a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs
index 48c533e02..199384d87 100644
--- a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs
+++ b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs
@@ -507,3 +507,102 @@ pub async fn run_consumer_offset_ahead_after_crash(harness: &mut TestHarness) {
"BUG: consumer skipped to offset {first_offset}, expected messages in range 10-14"
);
}
+
+/// Full-cluster restart: every node stops before any comes back, so no
+/// settled primary survives to answer the rejoin probes. The replicas must
+/// give up probing (`ViewChangeReason::ViewProbeUnanswered`), elect among
+/// their recovered logs, and serve the durable data across the outage. The
+/// caller's server config must flush eagerly (`messages_required_to_save=1`
+/// + fsync): with every journal dying at once, only flushed bytes survive.
+pub async fn run_full_cluster_restart(harness: &mut TestHarness) {
+ let setup_client = harness
+ .root_client()
+ .await
+ .expect("Failed to create setup client");
+
+ setup_client
+ .create_stream(STREAM_NAME)
+ .await
+ .expect("Failed to create stream");
+ setup_client
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ 1,
+ Default::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("Failed to create topic");
+ send_messages(&setup_client, "pre-outage", 10).await;
+ drop(setup_client);
+ // Let the backups' commit walk flush the tail: the send ack proves quorum
+ // COMMIT, but each replica flushes on its own walk (driven by the commit
+ // heartbeat), and the partition journal does not survive the process. A
+ // kill racing the last heartbeat leaves the newest message durable only
+ // on the old primary -- the known journal-durability caveat, not what
+ // this scenario tests.
+ sleep(Duration::from_secs(2)).await;
+
+ harness
+ .restart_cluster()
+ .await
+ .expect("Failed to restart the whole cluster");
+
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to reconnect after full-cluster restart");
+ let survivors = poll_from_zero_until(&client, 10, Duration::from_secs(30)).await;
+ assert_eq!(
+ survivors,
+ (0..10).map(|i| format!("pre-outage-{i}")).collect::<Vec<_>>(),
+ "flushed messages must survive a full-cluster restart"
+ );
+
+ send_messages(&client, "post-outage", 10).await;
+ let all = poll_from_zero_until(&client, 20, Duration::from_secs(30)).await;
+ assert_eq!(all.len(), 20, "the reformed cluster must accept new writes");
+ assert_eq!(all[10], "post-outage-0");
+ assert_eq!(all[19], "post-outage-9");
+}
+
+/// Polls from offset 0 until `expected` messages arrive or the deadline
+/// passes, returning their payloads. The cluster may still be mid-election
+/// right after a restart, so a single poll is not a fair snapshot.
+async fn poll_from_zero_until(
+ client: &IggyClient,
+ expected: u32,
+ deadline: Duration,
+) -> Vec<String> {
+ let started = std::time::Instant::now();
+ loop {
+ let polled = client
+ .poll_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ expected * 2,
+ false,
+ )
+ .await;
+ if let Ok(polled) = &polled
+ && polled.messages.len() >= expected as usize
+ {
+ return polled
+ .messages
+ .iter()
+ .map(|message| String::from_utf8_lossy(&message.payload).to_string())
+ .collect();
+ }
+ if started.elapsed() > deadline {
+ let got = polled.map(|p| p.messages.len()).unwrap_or(0);
+ panic!("expected {expected} messages after full-cluster restart, got {got}");
+ }
+ sleep(Duration::from_millis(250)).await;
+ }
+}
diff --git a/core/integration/tests/server/specific.rs b/core/integration/tests/server/specific.rs
index c4eef2961..f5a724723 100644
--- a/core/integration/tests/server/specific.rs
+++ b/core/integration/tests/server/specific.rs
@@ -107,6 +107,18 @@ async fn single_message_restart_offset_zero(harness: &mut TestHarness) {
reconnect_after_restart_scenario::run_single_message_offset_zero_restart(harness).await;
}
+// Full-cluster restart is vsr-only by construction: it exercises the rejoin
+// probe's election fallback across all replicas, which a single-process
+// legacy server has no equivalent of (plain restart covers it there).
+#[cfg(feature = "vsr")]
+#[iggy_harness(server(
+ partition.messages_required_to_save = "1",
+ partition.enforce_fsync = true
+))]
+async fn full_cluster_restart_recovers_and_serves(harness: &mut TestHarness) {
+ reconnect_after_restart_scenario::run_full_cluster_restart(harness).await;
+}
+
#[iggy_harness(server(
partition.messages_required_to_save = "1",
partition.enforce_fsync = true