diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 2e4741dab2..28865344f7 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -1127,7 +1127,7 @@ pub struct ConsensusTimers { } /// How a restored replica joins its group. -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum JoinMode { /// Fresh group or solo replica: plain init; the group needs its view-0 /// primary to exist. @@ -1151,11 +1151,72 @@ pub struct VsrRestore<'a> { /// View inferred from the last journaled prepare, consulted only when no /// durable record exists; `log_view` cannot be inferred and stays 0. pub view_fallback: Option, + /// Starting `(view, log_view)` for a group with NO history of its own, + /// consulted only when neither of the two above applies. + /// + /// Sets `log_view` as well as `view`, unlike `view_fallback`: the log is + /// empty, so there is no history to misattribute to the view, and a + /// primary whose `log_view` lags its `view` is treated as mid-transition + /// and answers no `RequestStartView` probe (see + /// [`VsrConsensus::handle_request_start_view`]) - it would hold its own group's + /// probes open forever. + /// + /// Deliberately NOT marked superblock-durable: nothing has been written + /// yet, and claiming otherwise would let a restart resume a view no record + /// holds. The group persists on its first tick instead. + pub seed_view: Option, /// Non-zero boot incarnation; `None` keeps the default. pub incarnation: Option, pub join: JoinMode, } +/// How a group with no consensus state in memory should come up. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FreshGroupStart { + pub join: JoinMode, + /// Starting view for a group with no durable record; see + /// [`VsrRestore::seed_view`]. Always `None` when a durable record exists, + /// so a caller can apply the two in either order. + pub seed_view: Option, +} + +/// Decide how a group being materialised should join, and what view it starts +/// in when it has no record of its own. +/// +/// One function because there are two materialisation paths and they must not +/// drift: `build_partition_fresh` on the server, and the simulator's +/// `init_partition`, which cannot call it (that builder does real filesystem +/// work, and the simulator runs on in-memory storage). Both previously decided +/// this for themselves, which is how the simulator came to exercise neither +/// the seed nor the split it closes. +/// +/// `restarted` is the caller's own evidence of a prior life, since the two +/// paths read it differently: a partition directory already on disk for the +/// server, a retained in-memory log for the simulator. +#[must_use] +pub const fn fresh_group_start( + restarted: bool, + durable_view: Option<(u32, u32)>, + metadata_view: Option, +) -> FreshGroupStart { + let join = if restarted { + JoinMode::ProbeAsBackup { + await_state_transfer: false, + } + } else { + JoinMode::Init + }; + // A durable record outranks the seed, and a probing backup takes neither: + // it must sit at or below the group's real view for the primary's + // `StartView` to move it forward, and seeded above that the reply reads as + // stale and is dropped. + let seed_view = match (durable_view, join) { + (None, JoinMode::Init) => metadata_view, + _ => None, + }; + FreshGroupStart { join, seed_view } +} + impl> VsrConsensus { /// # Panics /// - If `replica >= replica_count`. @@ -1230,6 +1291,14 @@ impl> VsrConsensus { consensus.mark_superblock_durable(view, log_view); } else if let Some(view) = restore.view_fallback { consensus.set_view(view); + } else if let Some(view) = restore.seed_view { + tracing::info!( + group, + view, + "seeded a group with no history of its own into the current view" + ); + consensus.set_view(view); + consensus.set_log_view(view); } match restore.join { JoinMode::Init => consensus.init(), @@ -3988,6 +4057,51 @@ where } } +#[cfg(test)] +mod fresh_group_start_tests { + use super::{JoinMode, fresh_group_start}; + + const METADATA_VIEW: u32 = 4; + + /// The case the seed exists for: a group created after the metadata plane + /// elected must not start at view 0, or it names a replica the roster does + /// not advertise and no client can be routed to it. + #[test] + fn given_a_fresh_group_when_the_metadata_view_moved_should_seed_that_view() { + let start = fresh_group_start(false, None, Some(METADATA_VIEW)); + assert_eq!(start.join, JoinMode::Init); + assert_eq!(start.seed_view, Some(METADATA_VIEW)); + } + + /// No published view is "no opinion", not "view 0". The caller defers + /// materialising rather than seeding; this only asserts nothing is + /// invented here. + #[test] + fn given_a_fresh_group_when_no_metadata_view_is_known_should_not_seed() { + let start = fresh_group_start(false, None, None); + assert_eq!(start.join, JoinMode::Init); + assert_eq!(start.seed_view, None); + } + + /// A durable record outranks the seed: it is what this replica actually + /// promised, and the seed is a guess about someone else's plane. + #[test] + fn given_a_durable_record_when_seeding_should_prefer_the_record() { + let start = fresh_group_start(false, Some((7, 7)), Some(METADATA_VIEW)); + assert_eq!(start.seed_view, None); + } + + /// A probing backup must sit at or below the group's real view for the + /// primary's `StartView` to move it forward. Seeded above it, the reply + /// reads as stale and the replica never rejoins. + #[test] + fn given_a_prior_life_when_seeding_should_probe_without_a_seed() { + let start = fresh_group_start(true, None, Some(METADATA_VIEW)); + assert!(matches!(start.join, JoinMode::ProbeAsBackup { .. })); + assert_eq!(start.seed_view, None); + } +} + #[cfg(test)] mod request_queue_tests { use super::*; diff --git a/core/harness_derive/src/attrs.rs b/core/harness_derive/src/attrs.rs index 9d1f07cbdd..136b379010 100644 --- a/core/harness_derive/src/attrs.rs +++ b/core/harness_derive/src/attrs.rs @@ -57,6 +57,12 @@ pub struct IggyTestAttrs { pub seed_fn: Option, pub cluster_nodes: ClusterNodesValue, pub jwks_server: Option, + /// Skip the generated `start()` call so the test brings nodes up itself. + /// For boot-ordering tests, which need a cluster that begins serving + /// before every configured node has arrived; `start` waits for all of + /// them. Pairs with `TestHarness::start_nodes` / `start_node`. A `seed` + /// cannot run under it, since no node is up when it would fire. + pub manual_start: bool, } /// JWKS server attributes. @@ -90,6 +96,7 @@ impl IggyTestAttrs { seed_fn: None, cluster_nodes: ClusterNodesValue::None, jwks_server: None, + manual_start: false, } } } @@ -265,6 +272,9 @@ impl Parse for IggyTestAttrs { AttrItem::JwksServer(jwks) => { attrs.jwks_server = Some(jwks); } + AttrItem::ManualStart => { + attrs.manual_start = true; + } } } @@ -282,6 +292,7 @@ enum AttrItem { Seed(syn::Path), ClusterNodes(ClusterNodesValue), JwksServer(JwksAttrs), + ManualStart, } impl Parse for AttrItem { @@ -317,6 +328,7 @@ impl Parse for AttrItem { let jwks = parse_jwks_attrs(&content)?; Ok(AttrItem::JwksServer(jwks)) } + "manual_start" => Ok(AttrItem::ManualStart), _ => Err(syn::Error::new( ident.span(), format!("unknown attribute: {ident_str}"), diff --git a/core/harness_derive/src/codegen.rs b/core/harness_derive/src/codegen.rs index 481e39af49..591f11fad5 100644 --- a/core/harness_derive/src/codegen.rs +++ b/core/harness_derive/src/codegen.rs @@ -557,6 +557,11 @@ fn generate_harness_setup( /// after server but before MCP and connectors runtime (which may depend on seed data). /// Fixture seeds are combined with the global seed. fn generate_start_and_seed(attrs: &IggyTestAttrs, fixture_seed: TokenStream) -> TokenStream { + // `manual_start` hands the bring-up to the test body, so there is nothing + // to emit and no client for a seed to run against. + if attrs.manual_start { + return TokenStream::new(); + } let has_fixture_seed = !fixture_seed.is_empty(); match (&attrs.seed_fn, has_fixture_seed) { (Some(seed_fn), true) => { @@ -837,6 +842,7 @@ mod tests { }, seed_fn: None, cluster_nodes: crate::attrs::ClusterNodesValue::None, + manual_start: false, jwks_server: None, }; let variants = generate_variants(&attrs); diff --git a/core/integration/src/harness/disk.rs b/core/integration/src/harness/disk.rs index 332a531efc..97d42b16c3 100644 --- a/core/integration/src/harness/disk.rs +++ b/core/integration/src/harness/disk.rs @@ -448,14 +448,29 @@ pub fn find_partition_superblock_dir(root: &Path) -> Option { /// Index of the node the metadata roster marks as leader, resolved by matching /// the roster's TCP port against each node's bound address. /// +/// Reads through node 0. Use [`leader_node_index_via`] where node 0 may be +/// down or where the roster must be read from a named node. +/// /// # Panics /// If no root client connects, the roster query fails, no node is marked /// leader, or the leader's port matches no harness node. pub async fn leader_node_index(harness: &TestHarness) -> usize { + leader_node_index_via(harness, 0).await +} + +/// [`leader_node_index`] reading the roster through node `via`. +/// +/// Named separately because which node answers matters once node 0 may be +/// down: the roster read is auth-gated, so it needs a node that can complete a +/// login, and a client built against a dead node cannot. +/// +/// # Panics +/// As [`leader_node_index`]. +pub async fn leader_node_index_via(harness: &TestHarness, via: usize) -> usize { let client = harness - .root_client_for_node(0) + .root_client_for_node(via) .await - .expect("a root client (redirecting to the leader if node 0 is not it)"); + .expect("a root client (redirecting to the leader if the dialed node is not it)"); let metadata = client .get_cluster_metadata() .await diff --git a/core/integration/src/harness/orchestrator/harness.rs b/core/integration/src/harness/orchestrator/harness.rs index f324ca39be..60324adc1c 100644 --- a/core/integration/src/harness/orchestrator/harness.rs +++ b/core/integration/src/harness/orchestrator/harness.rs @@ -216,7 +216,6 @@ impl TestHarness { const CLUSTER_READY_TIMEOUT: Duration = Duration::from_secs(15); const CLUSTER_READY_RETRY_INTERVAL: Duration = Duration::from_millis(200); - const LOGIN_ATTEMPT_TIMEOUT: Duration = Duration::from_millis(750); let deadline = Instant::now() + CLUSTER_READY_TIMEOUT; @@ -247,32 +246,96 @@ impl TestHarness { }); } - let mut last_error = None; + self.wait_for_login_ready(deadline).await + } + } - while Instant::now() < deadline { - match timeout(LOGIN_ATTEMPT_TIMEOUT, self.tcp_root_client()).await { - Ok(Ok(client)) => { - let _ = client.disconnect().await; - return Ok(()); - } - Ok(Err(error)) => { - last_error = Some(error.to_string()); - sleep(CLUSTER_READY_RETRY_INTERVAL).await; - } - Err(_) => { - last_error = Some("login attempt timed out".to_string()); - sleep(CLUSTER_READY_RETRY_INTERVAL).await; - } + /// Poll a root login until one succeeds or `deadline` passes. + /// + /// A root login is a replicated `Register`, so it cannot commit without a + /// commit quorum. One succeeding is therefore direct evidence that the + /// nodes which ARE up formed a working quorum, which is what the mesh + /// marker cannot tell you: that marker only reports peer TCP links, and + /// `mesh_expected_peers` counts every CONFIGURED peer, so a deliberately + /// partial cluster never emits it however healthy its quorum is. + async fn wait_for_login_ready(&self, deadline: Instant) -> Result<(), TestBinaryError> { + const RETRY_INTERVAL: Duration = Duration::from_millis(200); + const LOGIN_ATTEMPT_TIMEOUT: Duration = Duration::from_millis(750); + + let mut last_error = None; + while Instant::now() < deadline { + match timeout(LOGIN_ATTEMPT_TIMEOUT, self.tcp_root_client()).await { + Ok(Ok(client)) => { + let _ = client.disconnect().await; + return Ok(()); + } + Ok(Err(error)) => { + last_error = Some(error.to_string()); + sleep(RETRY_INTERVAL).await; + } + Err(_) => { + last_error = Some("login attempt timed out".to_string()); + sleep(RETRY_INTERVAL).await; } } + } - Err(TestBinaryError::InvalidState { - message: format!( - "Timed out waiting for VSR cluster readiness: {}", - last_error.unwrap_or_else(|| "unknown error".to_string()) - ), - }) + Err(TestBinaryError::InvalidState { + message: format!( + "Timed out waiting for VSR cluster readiness: {}", + last_error.unwrap_or_else(|| "unknown error".to_string()) + ), + }) + } + + /// Spawn only the nodes at `indexes` and wait for them to form a quorum. + /// + /// The counterpart of [`Self::start`] for boot-ordering tests: it models a + /// cluster that begins serving before every configured node has arrived, + /// which is what production does (nothing in the server's bootstrap waits + /// for peers) and what `start` deliberately does not. + /// + /// Readiness is a root login succeeding, not the all-nodes mesh gate + /// `start` uses: `mesh_expected_peers` counts every CONFIGURED peer, so a + /// deliberately partial cluster never reports a complete mesh however + /// healthy its quorum is. A login is a replicated `Register` and cannot + /// commit without quorum, which is the stronger signal anyway. + /// + /// Dependents (MCP, connectors runtime) and the configured clients are NOT + /// started: their addresses resolve against nodes this call deliberately + /// left down. Bring the rest up with [`Self::start_node`] and build clients + /// per node with [`Self::root_client_for_node`]. + pub async fn start_nodes(&mut self, indexes: &[usize]) -> Result<(), TestBinaryError> { + if self.started { + return Err(TestBinaryError::AlreadyStarted); + } + const READY_TIMEOUT: Duration = Duration::from_secs(15); + + for &index in indexes { + let server = self + .servers + .get_mut(index) + .ok_or(TestBinaryError::MissingServer)?; + server.start()?; } + self.started = true; + + let deadline = Instant::now() + READY_TIMEOUT; + self.wait_for_login_ready(deadline).await + } + + /// Spawn one node into an already-running cluster, without waiting for it. + /// + /// The late-joiner half of [`Self::start_nodes`]. No readiness wait: what + /// "ready" means for a node joining an established cluster is the caller's + /// question (rejoin at the live view, journal repair, state transfer), and + /// each has its own observable. + pub fn start_node(&mut self, index: usize) -> Result<(), TestBinaryError> { + let server = self + .servers + .get_mut(index) + .ok_or(TestBinaryError::MissingServer)?; + server.start() } async fn start_dependents(&mut self) -> Result<(), TestBinaryError> { @@ -298,7 +361,9 @@ impl TestHarness { Ok(()) } - /// Restart the primary server and reconnect all clients. + /// Restart the node at index 0 and reconnect all clients. Index 0 is not + /// "the primary" (see [`Self::running_server`]); use + /// [`Self::restart_node`] to name a different one. pub async fn restart_server(&mut self) -> Result<(), TestBinaryError> { if self.servers.is_empty() { return Err(TestBinaryError::MissingServer); @@ -416,12 +481,38 @@ impl TestHarness { Ok(()) } - /// Get reference to the first (primary) server handle. + /// The node client helpers dial when the caller names none: node 0 while + /// it is alive, otherwise the lowest-indexed node whose process is. + /// + /// Index 0 is a default, not a role. It carries no leadership: the + /// metadata primary is `view % replica_count` and moves on every + /// election, and the SDK redirects to whoever that is on connect. Pinning + /// these helpers to node 0 meant every client helper failed outright once + /// a test stopped it, which is why cluster tests reach for + /// [`Self::root_client_for_node`] instead. Preferring 0 keeps the common + /// case deterministic; falling through keeps the harness usable with it + /// down. + pub fn running_server(&self) -> Result<&ServerHandle, TestBinaryError> { + if self.servers.is_empty() { + return Err(TestBinaryError::MissingServer); + } + self.servers + .iter() + .find(|server| server.is_running()) + .ok_or(TestBinaryError::MissingServer) + } + + /// Get reference to the node at index 0. + /// + /// Index 0 is not "the primary": see [`Self::running_server`]. Use + /// [`Self::node`] when the index matters and `running_server` when any + /// live node will do. pub fn server(&self) -> &ServerHandle { self.servers.first().expect("No servers configured") } - /// Get mutable reference to the first (primary) server handle. + /// Get mutable reference to the node at index 0. Index 0 is not "the + /// primary": see [`Self::running_server`]. pub fn server_mut(&mut self) -> &mut ServerHandle { self.servers.first_mut().expect("No servers configured") } @@ -556,7 +647,7 @@ impl TestHarness { &self, transport: TransportProtocol, ) -> Result { - let server = self.servers.first().ok_or(TestBinaryError::MissingServer)?; + let server = self.running_server()?; match transport { TransportProtocol::Tcp => server.tcp_client(), TransportProtocol::Http => server.http_client(), @@ -583,7 +674,7 @@ impl TestHarness { &self, transport: TransportProtocol, ) -> Result { - let server = self.servers.first().ok_or(TestBinaryError::MissingServer)?; + let server = self.running_server()?; let builder = match transport { TransportProtocol::Tcp => server.tcp_client()?, TransportProtocol::Http => server.http_client()?, diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs index 05f42ba212..c1fafecdc6 100644 --- a/core/integration/tests/cluster/mod.rs +++ b/core/integration/tests/cluster/mod.rs @@ -24,5 +24,7 @@ mod failover_client_continuity; mod metadata_checkpoint_restart; mod metadata_state_transfer; mod multi_shard_partition_convergence; +mod partition_primary_routing; mod partition_state_transfer; mod register_forwarding; +mod staggered_bootstrap; diff --git a/core/integration/tests/cluster/partition_primary_routing.rs b/core/integration/tests/cluster/partition_primary_routing.rs new file mode 100644 index 0000000000..064f0134bc --- /dev/null +++ b/core/integration/tests/cluster/partition_primary_routing.rs @@ -0,0 +1,198 @@ +// 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. + +//! The node a client is told is "the leader" must be the node that accepts a +//! partition write. +//! +//! `get_cluster_metadata` marks a node `Leader` from the METADATA plane's +//! `primary_index` alone, while a partition write is only accepted by the +//! primary of that partition's OWN consensus group. Both planes pick their +//! primary as `view % replica_count`, but their views are independent +//! counters, so the two answers agree only while the views are congruent mod +//! the replica count. Every other 3-node test happens to run with both planes +//! at view 0, where node 0 is leader and partition primary at once, so none of +//! them can see the split. +//! +//! This test forces the views apart: it moves the metadata plane off view 0, +//! brings every node back, then creates a topic whose partition group is brand +//! new. Seeded from the metadata view it lands on the advertised leader; left +//! at view 0 it would name node 0, and no client could be told to go there. +//! +//! Both assertions are about the SAME node, and deliberately so. The SDK +//! follows the roster's leader on connect, so a client asking for node 0 does +//! not stay there - which is itself worth pinning down, since a test that +//! believes it is exercising node 0 while sitting on the leader proves +//! nothing. One assertion records where the client actually lands, the other +//! is the contract: the node the roster advertises accepts a partition write. + +use std::str::FromStr; +use std::time::Duration; + +use iggy::prelude::*; +use integration::harness::disk::leader_node_index_via; +use integration::iggy_harness; +use tokio::time::{Instant, sleep}; + +const STREAM_NAME: &str = "partition-routing-stream"; +const TOPIC_NAME: &str = "partition-routing-topic"; +const PARTITION_ID: u32 = 0; + +/// Long enough for the backups to miss `cluster.heartbeat_timeout` (5s by +/// default) and conclude an election. +const ELECTION_SETTLE: Duration = Duration::from_secs(15); +/// Long enough for the restarted node 0 to rejoin at the new view. +const REJOIN_SETTLE: Duration = Duration::from_secs(10); +/// Under the SDK's own `RESPONSE_READ_TIMEOUT` (30s), so this fires first and +/// names the failure. Above it the SDK's timeout always wins and the budget is +/// dead code. +const SEND_BUDGET: Duration = Duration::from_secs(20); +/// How long the metadata plane gets to settle on a leader that is not node 0, +/// the state this test needs before it can observe anything. +const PRECONDITION_BUDGET: Duration = Duration::from_secs(20); +const PRECONDITION_POLL: Duration = Duration::from_millis(500); + +fn message(payload: &str) -> IggyMessage { + IggyMessage::from_str(payload).expect("build message") +} + +#[iggy_harness(cluster_nodes = 3)] +async fn given_metadata_view_moved_when_producing_to_a_fresh_topic_should_reach_the_advertised_leader( + harness: &mut TestHarness, +) { + // Kill node 0: it is the view-0 primary of BOTH planes, so the metadata + // plane must elect someone else. Nothing has been written yet, so no + // partition group exists to move with it. Fixed waits rather than polling: + // dialing a leaderless cluster blocks for the SDK's own budget, and a poll + // loop that opens a fresh connection each round never converges. + harness.kill_node(0).expect("kill node 0"); + sleep(ELECTION_SETTLE).await; + harness.restart_node(0).expect("restart node 0"); + sleep(REJOIN_SETTLE).await; + + // Read through node 1: node 0 has only just restarted, and the roster read + // is auth-gated, so it needs a node that can complete a login now. + // + // A SETUP PRECONDITION, not an invariant of the system. `primary_index` is + // `view % replica_count` with no `Status::Normal` gate, so a cluster that + // elected three times is back to advertising node 0 while perfectly + // healthy. Polled rather than asserted once: the split this test is about + // is only observable while the planes disagree, and one kill normally + // lands view 1 immediately. + let leader = { + let deadline = Instant::now() + PRECONDITION_BUDGET; + loop { + let index = leader_node_index_via(harness, 1).await; + if index != 0 { + break index; + } + assert!( + Instant::now() < deadline, + "the metadata plane never settled on a leader other than node 0 within \ + {PRECONDITION_BUDGET:?}; with the leader at node 0 both planes agree and \ + the split this test is about cannot show" + ); + sleep(PRECONDITION_POLL).await; + } + }; + + // A brand-new topic. Its partition group is seeded from the metadata view, + // so its primary is the advertised leader; left at view 0 it would be + // replica 0, the node that was just killed and restarted. + let setup = harness + .root_client_for_node(leader) + .await + .expect("root client on the metadata leader"); + setup + .create_stream(STREAM_NAME) + .await + .expect("create stream"); + let stream_id = Identifier::named(STREAM_NAME).expect("stream identifier"); + setup + .create_topic( + &stream_id, + TOPIC_NAME, + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + ..TopicCreateOptions::default() + }, + ) + .await + .expect("create topic"); + let topic_id = Identifier::named(TOPIC_NAME).expect("topic identifier"); + let partitioning = Partitioning::partition_id(PARTITION_ID); + + // Where a client asking for node 0 actually ends up. + let leader_address = harness + .node(leader) + .tcp_addr() + .expect("the leader exposes a TCP endpoint") + .to_string(); + let on_node_zero = harness + .root_client_for_node(0) + .await + .expect("root client on node 0"); + let landed_on = on_node_zero.get_connection_info().await.server_address; + + // Asserted, not printed. `root_client_for_node` signs in, and sign-in ends + // in the SDK's leader check, so this client is on the LEADER whatever node + // it dialed. Pinning that down is what stops the send below from being + // read as "node 0 accepted it": nothing here ever reaches node 0, and a + // reader who assumes otherwise draws the opposite conclusion from a pass. + assert_eq!( + landed_on, leader_address, + "a signed-in client follows the roster's leader, so one dialing node 0 must settle on \ + node {leader}; landing anywhere else means the redirect did not run and the send below \ + is testing a different node than this test claims" + ); + + // The contract: the node the roster advertises accepts a partition write. + // Seeded from the metadata view the group's primary IS that node; left at + // view 0 it would be replica 0, and every client would be steered away + // from the only node that could accept. + let accepted_by_leader = send_once(&on_node_zero, &stream_id, &topic_id, &partitioning).await; + assert!( + accepted_by_leader.is_ok(), + "node {leader} is advertised as the cluster leader, so a partition write sent there must \ + be accepted (or forwarded), got {accepted_by_leader:?}" + ); +} + +/// One send, bounded. The SDK replays `TransientNotAccepted` and then hands the +/// request to its failover path, which re-reads the same roster and returns to +/// the same wrong node, so with the defect present the send burns its whole +/// budget. The timeout fires before the SDK's own and names which it was. +async fn send_once( + client: &IggyClient, + stream_id: &Identifier, + topic_id: &Identifier, + partitioning: &Partitioning, +) -> Result<(), String> { + let mut messages = vec![message("probe")]; + match tokio::time::timeout( + SEND_BUDGET, + client.send_messages(stream_id, topic_id, partitioning, &mut messages), + ) + .await + { + Ok(Ok(_)) => Ok(()), + Ok(Err(error)) => Err(format!("{error:?}")), + Err(_) => Err(format!( + "no answer within {SEND_BUDGET:?} (client livelocked)" + )), + } +} diff --git a/core/integration/tests/cluster/staggered_bootstrap.rs b/core/integration/tests/cluster/staggered_bootstrap.rs new file mode 100644 index 0000000000..a9979906f9 --- /dev/null +++ b/core/integration/tests/cluster/staggered_bootstrap.rs @@ -0,0 +1,180 @@ +// 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. + +//! A cluster whose replica 0 arrives after the others have already elected. +//! +//! Nothing in the server's bootstrap waits for peers: `await_bootstrap_complete` +//! is intra-node (shard 0 waiting on its sibling shards), and each node binds +//! its listeners as soon as its own shards load. So a slow replica 0 does not +//! hold the cluster back. Replicas 1 and 2 miss its heartbeats, conclude an +//! election, and start serving at a view whose primary is not replica 0. +//! +//! Every other cluster test starts through `TestHarness::start`, which waits +//! for all nodes to mesh before the first client op, so none of them can reach +//! this state. `partition_primary_routing` reaches the same view split by +//! killing node 0 mid-test; this reaches it the way production does, and the +//! two entry points exercise different code (a group materialised on a node +//! that was never in the founding quorum, versus one materialised on a node +//! that was). +//! +//! Two independent things are asserted, because they can fail separately: +//! +//! - A topic created while replica 0 is absent is writable at the advertised +//! leader. This is the routing contract, and it holds only because a fresh +//! partition group seeds its view from the metadata plane rather than +//! starting at view 0 (which would name replica 0, the node that was late). +//! - The late replica converges. It missed the ops committed before it +//! arrived, and the harness's usual all-nodes mesh gate is what normally +//! guarantees no node is ever in that position. + +use std::str::FromStr; +use std::time::Duration; + +use iggy::prelude::*; +use integration::harness::disk::leader_node_index_via; +use integration::iggy_harness; +use tokio::time::{Instant, sleep}; + +const STREAM_NAME: &str = "staggered-bootstrap-stream"; +const TOPIC_NAME: &str = "staggered-bootstrap-topic"; +const PARTITION_ID: u32 = 0; + +/// Long enough for replicas 1 and 2 to miss `cluster.heartbeat_timeout` (5s by +/// default) and conclude an election without replica 0. +const ELECTION_SETTLE: Duration = Duration::from_secs(15); +/// Long enough for the late replica 0 to probe for the live view and rejoin. +const REJOIN_SETTLE: Duration = Duration::from_secs(10); +/// Under the SDK's own `RESPONSE_READ_TIMEOUT` (30s), so this fires first and +/// names the failure rather than being shadowed by it. +const SEND_BUDGET: Duration = Duration::from_secs(20); +/// How long the late replica gets to show it has caught up. +const CONVERGE_BUDGET: Duration = Duration::from_secs(30); +const MARKER_POLL: Duration = Duration::from_millis(250); + +/// The late replica joining the view the others elected without it. Its own +/// recorded view is 0, which names ITSELF primary, so this line is where it +/// gives that up. +const VIEW_ADOPTED_MARKER: &str = "adopting view from StartView"; + +/// Markers that each independently prove the late replica pulled committed +/// state it did not have. Which one fires depends on whether the gap sits +/// above or below the serving peers' retained journal floor: repair refills +/// from the peers' journals, state transfer is what a gap below the retained +/// floor converts into. A cluster this small usually stays above the floor and +/// repairs, but the size of the founding quorum's log is not something this +/// test fixes, so either counts. +const CAUGHT_UP_MARKERS: [&str; 2] = [ + "metadata journal repair walked", + "metadata state transfer installed", +]; + +fn message(payload: &str) -> IggyMessage { + IggyMessage::from_str(payload).expect("build message") +} + +#[iggy_harness(cluster_nodes = 3, manual_start)] +async fn given_replica_zero_arrives_late_when_producing_to_a_fresh_topic_should_reach_the_advertised_leader( + harness: &mut TestHarness, +) { + // Replicas 1 and 2 only. A successful root login inside `start_nodes` is a + // committed Register, so returning at all proves the two of them formed a + // quorum with replica 0 still absent. + harness + .start_nodes(&[1, 2]) + .await + .expect("replicas 1 and 2 must form a quorum without replica 0"); + sleep(ELECTION_SETTLE).await; + + // Read through node 1: node 0 is not running, so it can answer no login, + // and the roster read is auth-gated. + // + // Unlike the restart-driven twin of this test, this one is an invariant + // rather than a precondition: replica 0 has never been started, so no view + // it could be elected in exists yet. + let leader = leader_node_index_via(harness, 1).await; + assert_ne!( + leader, 0, + "replica 0 was never started, so it cannot be the metadata leader" + ); + + // Replica 0 arrives into a cluster that has already elected past it. + harness.start_node(0).expect("start the late replica 0"); + sleep(REJOIN_SETTLE).await; + + let setup = harness + .root_client_for_node(leader) + .await + .expect("root client on the metadata leader"); + setup + .create_stream(STREAM_NAME) + .await + .expect("create stream"); + let stream_id = Identifier::named(STREAM_NAME).expect("stream identifier"); + setup + .create_topic( + &stream_id, + TOPIC_NAME, + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + ..TopicCreateOptions::default() + }, + ) + .await + .expect("create topic"); + let topic_id = Identifier::named(TOPIC_NAME).expect("topic identifier"); + let partitioning = Partitioning::partition_id(PARTITION_ID); + + // The routing contract. The partition group is brand new, so its view is + // whatever it was seeded with: the metadata view, which is not 0 here. + // Seeded at 0 instead it would name replica 0, the node that arrived late, + // and no client could be told to go there. + let mut messages = vec![message("probe")]; + let accepted = tokio::time::timeout( + SEND_BUDGET, + setup.send_messages(&stream_id, &topic_id, &partitioning, &mut messages), + ) + .await; + assert!( + matches!(accepted, Ok(Ok(_))), + "node {leader} is advertised as the cluster leader, so a partition write sent there must \ + be accepted, got {accepted:?}" + ); + + // The late replica missed every op committed before it arrived. Read off + // its own log rather than through a client: the SDK redirects to the + // leader on connect, so a client dialing node 0 reports the leader's state, + // not node 0's. + let deadline = Instant::now() + CONVERGE_BUDGET; + loop { + let late = harness.node(0); + let adopted_view = late.stdout_contains(VIEW_ADOPTED_MARKER); + let caught_up = CAUGHT_UP_MARKERS + .iter() + .any(|marker| late.stdout_contains(marker)); + if adopted_view && caught_up { + break; + } + assert!( + Instant::now() < deadline, + "the late replica 0 never converged within {CONVERGE_BUDGET:?} \ + (adopted the live view: {adopted_view}, caught up on committed ops: {caught_up}); \ + the harness's all-nodes mesh gate is what normally keeps a node out of this position" + ); + sleep(MARKER_POLL).await; + } +} diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 587744f5dc..7d8a277162 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -1282,6 +1282,7 @@ async fn shard_main( topology.cluster_id, topology.self_replica_id, topology.replica_count, + Arc::clone(&metadata_view), )); let reconcile_periodic = config .system @@ -2034,6 +2035,10 @@ async fn build_shard_for_thread( topology.cluster_id, topology.self_replica_id, topology.replica_count, + // Quarantine-and-rebuild always finds a partition + // directory already there, so this joins as a probing + // backup and learns the live view; nothing to seed. + None, Rc::clone(&bus), ) .await? @@ -2448,6 +2453,9 @@ fn restore_metadata_consensus( // re-derives, and it re-probes as a backup. durable_view: recovered_state.map(|state| (state.view, state.log_view)), view_fallback: last_header.map(|header| header.view), + // Metadata, not a partition group: it has a journal to infer from + // and no second plane to line up with. + seed_view: None, // Fresh random incarnation each boot, so a StartView addressed to // a previous incarnation still in flight is ignored // (`handle_start_view` guard). `| 1` guarantees the non-zero the @@ -2639,6 +2647,7 @@ async fn load_partition( .as_ref() .map(|state| (state.view, state.log_view)), view_fallback: None, + seed_view: None, incarnation: None, join, }, diff --git a/core/server/src/cluster_meta.rs b/core/server/src/cluster_meta.rs index 7372d3e597..c251ef6028 100644 --- a/core/server/src/cluster_meta.rs +++ b/core/server/src/cluster_meta.rs @@ -83,9 +83,13 @@ impl ClusterRoster { } } - /// The current metadata primary's roster index, from the shard-0-published + /// The current metadata primary's REPLICA ID, from the shard-0-published /// view; `None` until the first publish or with no roster. - pub fn current_primary_index(&self) -> Option { + /// + /// A replica id, not a position in [`Self::nodes`]: `role_for` compares it + /// against each node's configured `replica_id`, and the two coincide only + /// while the roster is listed in replica-id order. + pub fn current_primary_replica_id(&self) -> Option { if self.nodes.is_empty() { return None; } diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 5a2f49fd21..f548f5de71 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -28,7 +28,9 @@ use crate::offset_recovery::{load_consumer_group_offsets, load_consumer_offsets} use crate::server_error::ServerError; use compio::fs::create_dir_all; use configs::server::ServerConfig; -use consensus::{JoinMode, LocalPipeline, VsrConsensus, VsrRestore, VsrState}; +use consensus::{ + FreshGroupStart, LocalPipeline, VsrConsensus, VsrRestore, VsrState, fresh_group_start, +}; use iggy_common::{ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, IggyError, IggyTimestamp, PartitionStats, TopicRuntimeOptions, @@ -516,6 +518,12 @@ pub(crate) async fn open_partition_superblock( /// The namespace arrives packed, so its components are in range by /// construction. Metadata admission is what bounds them. /// +/// `view_seed` is the view a group with no durable record of its own starts +/// in, and it is the metadata plane's current view: see the `seed_view` +/// comment below for why a group left at view 0 is unreachable. `None` keeps +/// the historical view-0 start, and is also what a restart materialization +/// gets, since it probes for the live view instead. +/// /// The returned partition's `offset` / `dirty_offset` are `0` and /// `should_increment_offset` is `false`, mirroring a clean append starting /// at the empty segment. @@ -534,6 +542,7 @@ pub async fn build_partition_fresh( cluster_id: u128, self_replica_id: u8, replica_count: u8, + view_seed: Option, bus: Rc, ) -> Result>, ServerError> { let stream_id = namespace.stream_id(); @@ -590,13 +599,12 @@ pub async fn build_partition_fresh( // peer, byte-identical by the deterministic-roll/replicated-ciphertext // design. A truly fresh create keeps the plain init: every group needs // its view-0 primary to exist. - let join = if restarted { - JoinMode::ProbeAsBackup { - await_state_transfer: false, - } - } else { - JoinMode::Init - }; + let durable_view = recovered_state + .as_ref() + .map(|state| (state.view, state.log_view)); + // Shared with the simulator's `init_partition`, which cannot call this + // builder; see `fresh_group_start`. + let FreshGroupStart { join, seed_view } = fresh_group_start(restarted, durable_view, view_seed); // Request queue holds 2x the prepare depth (buffered requests drain as // prepares commit); depth is the per-partition `[partition]` knob. let prepare_queue_depth = config.partition.prepare_queue_depth; @@ -610,10 +618,25 @@ pub async fn build_partition_fresh( LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), VsrRestore { timers: &timers, - durable_view: recovered_state - .as_ref() - .map(|state| (state.view, state.log_view)), + durable_view, view_fallback: None, + // Both planes pick their primary as `view % replica_count` from + // their OWN view counter. A group left at view 0 while the + // metadata plane sits elsewhere therefore names a different node + // than the roster advertises as leader, and nothing routes a + // partition write across that gap: the client is sent to the + // metadata leader and refused there for the whole budget. Seeding + // from the metadata view keeps the two congruent for a group born + // after a metadata election. + // + // Replicas can still disagree on the seed: each publishes its own + // metadata view on a 100ms poll, so one may read V while another + // has yet to see the election. That resolves the way any view + // disagreement does, the higher view winning through `StartView` - + // but only because the seed sets `log_view` too. The caller is + // what keeps a replica from seeding a view it has no opinion on; + // see `ReconcilerCtx::partition_view_seed`. + seed_view, incarnation: None, join, }, diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 699e9c94da..051d81c694 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -170,6 +170,7 @@ //! -- `PrepareHeader.reserved` has room, but it is a `#[repr(C)]` wire change. use crate::bootstrap::ServerShard; +use crate::cluster_meta::METADATA_VIEW_UNKNOWN; use crate::partition_helpers::{build_partition_fresh, delete_partitions_from_disk}; use ahash::{AHashMap, AHashSet}; use configs::server::ServerConfig; @@ -187,6 +188,7 @@ use shard::{Receiver, Sender}; use std::cell::{Cell, RefCell}; use std::rc::Rc; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; use tracing::{debug, error, trace, warn}; @@ -225,6 +227,11 @@ pub struct ReconcilerCtx { pub cluster_id: u128, pub self_replica_id: u8, pub replica_count: u8, + /// The metadata plane's view, published by shard 0 and shared with every + /// shard's roster. Read when materialising a partition so its consensus + /// group starts in the same view the roster's advertised leader comes + /// from; the unknown-view sentinel until the first publish. + pub metadata_view: Arc, failure_state: RefCell>, /// `Streams::revision` observed at the end of the last pass that fully /// converged. Paired with `last_pass_noop` for the fast-skip in @@ -244,6 +251,7 @@ impl ReconcilerCtx { cluster_id: u128, self_replica_id: u8, replica_count: u8, + metadata_view: Arc, ) -> Self { Self { shard, @@ -252,12 +260,48 @@ impl ReconcilerCtx { cluster_id, self_replica_id, replica_count, + metadata_view, failure_state: RefCell::new(AHashMap::new()), last_revision: Cell::new(None), last_pass_noop: Cell::new(false), } } + /// The view a partition group materialised now should start in. + /// + /// `None` means this replica has no opinion yet, NOT "start at view 0": + /// shard 0 publishes the unknown-view sentinel before its first tick and + /// for as long as it has ceded a recovered view's primaryship. Treating + /// that as 0 is how a single replica of a group ends up view-0 while its + /// peers are at V, which is the split this seed exists to close. Callers + /// on a replicated group defer materialising instead; see + /// [`Self::partition_view_seed_ready`]. + /// + /// # Panics + /// If the published view does not fit a `u32`. The publisher writes + /// `u64::from(consensus.view())` or the sentinel handled above, so a value + /// past `u32::MAX` is memory corruption, not a state a retry can clear. + /// Silently falling back to `None` here would restore the view-0 start + /// this change exists to remove, on the one path nobody would look at. + fn partition_view_seed(&self) -> Option { + let view = self.metadata_view.load(Ordering::Relaxed); + if view == METADATA_VIEW_UNKNOWN { + return None; + } + Some(u32::try_from(view).expect("published metadata view must fit a u32")) + } + + /// Whether a group may be materialised now. + /// + /// A solo replica is always ready: with one replica every view names it + /// primary, so there is no peer to disagree with and no split to cause. + /// A replicated group waits until this replica knows the metadata view, + /// so it cannot seed a view-0 group underneath peers that already moved. + /// The wait is one reconcile pass; shard 0 republishes every 100ms. + fn partition_view_seed_ready(&self) -> bool { + self.replica_count <= 1 || self.partition_view_seed().is_some() + } + fn is_backed_off(&self, ns: IggyNamespace, cause: FailureCause, now: Instant) -> bool { let state = self.failure_state.borrow(); if state.is_empty() { @@ -417,6 +461,12 @@ struct PassCounters { /// has not applied yet. Counted so the pass does not arm the fast-skip /// while work is in flight; applying it bumps no revision. already_staged: usize, + /// Materialisations held back because this replica has no metadata view to + /// seed from yet. Counted so the pass does not arm the fast-skip: shard 0 + /// publishing a view bumps no `Streams::revision` and does not wake the + /// reconciler, so an armed skip would strand every deferred group until + /// some unrelated commit happened to bump the revision. + view_unpublished: usize, } impl PassCounters { @@ -433,6 +483,7 @@ impl PassCounters { + self.deferred + self.parked_reclaimed + self.already_staged + + self.view_unpublished } } @@ -659,6 +710,22 @@ async fn reconcile_additions( continue; } + // Materialising before this replica knows the metadata view would seed + // the group at view 0 under peers that already elected past it. Defer + // the whole pass rather than the namespace: every group on this shard + // reads the same view, so none of them can be seeded correctly yet. + if !ctx.partition_view_seed_ready() { + debug!( + shard = shard_id, + "metadata view not published yet; deferring partition materialisation" + ); + // Every remaining namespace reads the same view, so none of them + // can be seeded either. Counted before breaking so the pass does + // not read as converged, which would fast-skip the retry. + counters.view_unpublished += 1; + break; + } + // Resolve the shared stats `Arc` only for namespaces actually // built, not once per committed partition every pass. A topic that // vanished between the target snapshot and this read defers to the @@ -676,6 +743,7 @@ async fn reconcile_additions( ctx.cluster_id, ctx.self_replica_id, ctx.replica_count, + ctx.partition_view_seed(), Rc::clone(&ctx.shard.bus), ) .await @@ -1211,8 +1279,8 @@ pub fn install_tick_handler(shard: &Rc, wake_tx: WakeTx) { #[cfg(test)] mod tests { use super::{ - FailureCause, FailureRecord, ReconcilerCtx, build_partition_fresh, - delete_partitions_from_disk, fetch_partition_stats, reconcile_once, + AtomicU64, FailureCause, FailureRecord, METADATA_VIEW_UNKNOWN, ReconcilerCtx, + build_partition_fresh, delete_partitions_from_disk, fetch_partition_stats, reconcile_once, }; use configs::server::{ServerConfig, ServerSystemConfig}; use consensus::{MetadataHandle, PartitionsHandle}; @@ -1664,6 +1732,31 @@ mod tests { CLUSTER_ID, 0, 1, + Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)), + )) + } + + /// [`make_ctx`] for a REPLICATED group, sharing `metadata_view` with the + /// caller so a test can publish a view mid-run. + /// + /// Replica count is what decides whether materialisation waits on a + /// published view: a solo replica is primary in every view, so there is no + /// peer to disagree with and nothing to wait for. Every other test here + /// runs solo and takes that short circuit. + fn make_cluster_ctx( + shard: Rc, + total_shards: u16, + config: Rc, + metadata_view: Arc, + ) -> Rc { + Rc::new(ReconcilerCtx::new( + shard, + total_shards, + config, + CLUSTER_ID, + 0, + 3, + metadata_view, )) } @@ -1809,6 +1902,7 @@ mod tests { CLUSTER_ID, 0, 1, + None, Rc::clone(&ctx.shard.bus), ) .await @@ -1838,6 +1932,7 @@ mod tests { CLUSTER_ID, 0, 1, + None, Rc::clone(&ctx.shard.bus), ) .await @@ -1961,6 +2056,73 @@ mod tests { /// `CreatePartitions` on an existing topic adds new namespaces; the /// reconciler picks them up on the next pass without touching the /// partitions it already materialised. + /// A replicated group is NOT materialised while this replica has no + /// metadata view to seed from, and IS once one is published. + /// + /// Seeding is a local read: shard 0 publishes the unknown sentinel before + /// its first tick and for as long as it has ceded a recovered view. Taking + /// that for view 0 would start the group naming replica 0 underneath peers + /// that already elected past it, which is the split the seed exists to + /// close, reintroduced one replica at a time. Waiting costs one reconcile + /// pass; the publisher reposts every 100ms. + #[compio::test] + async fn given_no_published_metadata_view_when_reconciling_should_defer_materialisation() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-deferred"); + seed_topic(&mux, 2, 0, "topic-deferred", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let metadata_view = Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)); + let ctx = make_cluster_ctx( + Rc::clone(&shard), + 1, + Rc::new(config), + Arc::clone(&metadata_view), + ); + + reconcile_pass(&ctx).await; + assert_eq!( + shard.plane.partitions().len(), + 0, + "a replicated group must not materialise before this replica knows the metadata \ + view: seeded at 0 it names replica 0 whatever the metadata plane elected" + ); + + // The publisher posts a real view; the deferred pass now converges. + metadata_view.store(1, Ordering::Relaxed); + reconcile_pass(&ctx).await; + assert_eq!( + shard.plane.partitions().len(), + 1, + "once a view is published the deferred group must materialise on the next pass" + ); + } + + /// A SOLO replica never waits. It is primary in every view, so there is no + /// peer for it to disagree with, and blocking on a publisher that a + /// single-node deployment may never run would wedge materialisation + /// outright. + #[compio::test] + async fn given_a_solo_replica_when_no_view_is_published_should_still_materialise() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-solo"); + seed_topic(&mux, 2, 0, "topic-solo", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + reconcile_pass(&ctx).await; + assert_eq!( + shard.plane.partitions().len(), + 1, + "a solo replica must materialise without waiting on a published metadata view" + ); + } + #[compio::test] async fn reconcile_picks_up_create_partitions_increments() { let tmp = TempDir::new().expect("tempdir for system path"); diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs index f4e6bc4e56..21b8e6497c 100644 --- a/core/server/src/responses.rs +++ b/core/server/src/responses.rs @@ -682,7 +682,7 @@ where (!(consensus.has_ceded_primaryship() && primary_index == consensus.replica())) .then_some(primary_index) }) - .or_else(|| roster.current_primary_index()); + .or_else(|| roster.current_primary_replica_id()); let metadata = roster.cluster_metadata(primary_index, client_ip); ClusterMetadataResponse { name: metadata.name, diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 358b8e62bf..9098c7d639 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -3567,6 +3567,7 @@ where recovered_state: Option, retained: Option, restore_frontier: bool, + metadata_view: Option, ) where B: MessageBus + Clone, { @@ -3584,26 +3585,42 @@ where LocalPipeline::new(), self.partition_consensus.clock.clone(), ); + // The SAME decision `build_partition_fresh` makes, not a copy of it. + // This path cannot call that builder (it does real filesystem work and + // this runs on in-memory storage), and while the two decided + // separately the simulator exercised neither the metadata-view seed nor + // the plane split it closes. `retained` is populated only by the + // restart path, which is this path's evidence of a prior life. + let durable_view = recovered_state + .as_ref() + .map(|state| (state.view, state.log_view)); + let restarted = retained.is_some() && self.partition_consensus.replica_count > 1; + let consensus::FreshGroupStart { join, seed_view } = + consensus::fresh_group_start(restarted, durable_view, metadata_view); + // Recorded view first, exactly as the two boot paths order it: restoring // after `init` would advertise a view older than the recorded one. - if let Some(state) = recovered_state.as_ref() { - consensus.set_view(state.view); - consensus.set_log_view(state.log_view); - consensus.mark_superblock_durable(state.view, state.log_view); - } - // Boot as `load_partition` does. A rebuilt replica cannot know the group's - // `(op, commit)`: the partition journal is in-memory and segments carry no - // op numbers. So in a cluster it joins quorum-invisible and asks the view's - // primary rather than resuming as a primary its peers may have replaced. - // Plain `init` would set `Status::Normal` and arm the commit broadcast on - // whichever replica is primary-by-index, the split-brain `init_as_backup` - // exists to prevent. A first materialisation has no view to rejoin and - // keeps the plain init; `retained` is populated only by the restart path. - if retained.is_some() && self.partition_consensus.replica_count > 1 { - consensus.init_as_backup(); - consensus.begin_view_probe(); - } else { - consensus.init(); + if let Some((view, log_view)) = durable_view { + consensus.set_view(view); + consensus.set_log_view(log_view); + consensus.mark_superblock_durable(view, log_view); + } else if let Some(view) = seed_view { + consensus.set_view(view); + consensus.set_log_view(view); + } + // A rebuilt replica cannot know the group's `(op, commit)`: the + // partition journal is in-memory and segments carry no op numbers. So + // in a cluster it joins quorum-invisible and asks the view's primary + // rather than resuming as a primary its peers may have replaced. Plain + // `init` would set `Status::Normal` and arm the commit broadcast on + // whichever replica is primary-by-index, the split-brain + // `init_as_backup` exists to prevent. + match join { + consensus::JoinMode::ProbeAsBackup { .. } => { + consensus.init_as_backup(); + consensus.begin_view_probe(); + } + consensus::JoinMode::Init => consensus.init(), } let stats = Arc::new(PartitionStats::default()); diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 2a2f396bae..19d89dc879 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1297,6 +1297,33 @@ impl Simulator { /// `--crash-primary`, so a crash-triggered view change can run mid-run and this /// may name a stale or crashed primary. Callers needing a real answer run /// `workload::oracle::settle_to_stable_view` first. + /// The replica the METADATA plane currently names primary, read from the + /// first live replica that owns a metadata consensus. + /// + /// The twin of [`Self::primary_index`], which answers for a partition + /// group. The two planes count views independently, so they agree only + /// while their views are congruent mod the replica count, and a group + /// materialised after a metadata election is the case where they part. + /// + /// Test-only. The workload oracle deliberately does NOT assert the two + /// planes agree: that holds for a group at the view it was seeded in, and + /// a later election on either plane parts them again with nothing to pull + /// them back, so a live invariant would fire on correct runs. + #[cfg(test)] + #[must_use] + pub(crate) fn metadata_primary_index(&self) -> Option { + (0..self.replica_count) + .filter(|replica_idx| !self.crashed.contains(replica_idx)) + .find_map(|replica_idx| { + let consensus = self.replicas[usize::from(replica_idx)].shards[0] + .plane + .metadata() + .consensus + .as_ref()?; + Some(consensus.primary_index(consensus.view())) + }) + } + #[must_use] pub(crate) fn primary_index(&self, namespace: IggyNamespace) -> Option { (0..self.replica_count) @@ -1355,12 +1382,24 @@ fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace, restore // a second materialisation with no restart between would otherwise resurrect a // log the live partition has moved past. let retained = replica.partition_logs.borrow_mut().remove(&namespace); + // The view this replica's metadata plane is in, which is what a fresh + // group seeds from. Production reads it off the roster value shard 0 + // publishes; here shard 0's consensus is right there. Without it the + // simulator materialises every group at view 0 and can never produce the + // plane split that costs production its writes. + let metadata_view = replica.shards[0] + .plane + .metadata() + .consensus + .as_ref() + .map(consensus::VsrConsensus::view); replica.shards[usize::from(owner)].init_partition( namespace, Some(superblock), recovered_state, retained, restore_frontier, + metadata_view, ); for shard in &replica.shards { shard.shards_table().insert( @@ -1475,6 +1514,99 @@ mod tests { ); } + /// A partition group materialised AFTER a metadata election starts in the + /// metadata plane's view, so both planes name the same primary. + /// + /// Left at view 0 the group names replica 0 whatever the metadata plane has + /// got to. Nothing on the wire can express a partition primary + /// (`ClusterNode` carries one cluster-wide `role`) and partition ops route + /// within a node rather than to a peer, so the node clients are sent to + /// refuses every write to that group and the SDK burns its budget + /// rediscovering the same wrong answer. + /// + /// The simulator reaches this where the integration tests cannot: no + /// client, no transport, just the two planes' views read directly. It is + /// also the only place the SEED itself is asserted rather than inferred + /// from a send succeeding. + #[test] + fn given_a_metadata_election_when_a_group_materialises_should_seed_the_metadata_view() { + 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 network_opts = packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + ..packet::PacketSimulatorOptions::default() + }; + let mut sim = Simulator::new( + replica_count as usize, + std::iter::once(client_id), + network_opts, + ); + + // Move the metadata plane off view 0 by crashing its view-0 primary. + // Nothing has been written, so no partition group exists to move with + // it: the group created below is genuinely fresh. + sim.replica_crash(0); + for _ in 0..800 { + sim.step(); + } + + let metadata_primary = sim + .metadata_primary_index() + .expect("a live replica must own metadata consensus"); + assert_ne!( + metadata_primary, 0, + "crashing replica 0 must have moved the metadata plane off view 0; with the \ + primary back at replica 0 both planes agree and the split cannot show" + ); + + // Materialise a brand-new group. Every live replica seeds from its own + // metadata view, which is the value production reads off the roster. + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + + let partition_primary = sim + .primary_index(namespace) + .expect("the group must exist on a live replica after materialisation"); + assert_eq!( + partition_primary, metadata_primary, + "a group materialised after a metadata election must name the same primary as the \ + metadata plane; seeded at view 0 instead it names replica 0, which no client can \ + be routed to" + ); + + // Per replica, not just the aggregate: the seed is read locally on each + // one, so a single replica left at view 0 would still elect itself + // primary of that group while its peers disagree, and the aggregate + // read above would not see it. + for replica_idx in 0..replica_count { + if sim.is_crashed(replica_idx) { + continue; + } + let state = sim + .partition_consensus_state(usize::from(replica_idx), namespace) + .expect("a live replica must host the freshly materialised group"); + // Compared as primaries, not as view numbers: the two coincide only + // while the view is below `replica_count`, and pinning the view + // itself would make this fail on a second election for no reason. + let seeded_primary = u8::try_from(state.view % u32::from(replica_count)) + .expect("a value modulo replica_count fits the u8 replica_count"); + assert_eq!( + seeded_primary, metadata_primary, + "replica {replica_idx} seeded its group at view {}, naming replica \ + {seeded_primary} primary while the metadata plane names {metadata_primary}; \ + replicas that seed different views disagree on their own group's primary", + state.view + ); + } + } + /// A replica that advanced its view, persisted it through the superblock gate, /// then crashed recovers that view from its own disk, not a fresh 0. The /// split-brain guarantee: a replica never forgets a view it acted in. Impossible @@ -2389,7 +2521,7 @@ mod tests { executor.run_until_stalled(POLL_BUDGET); // borrow acquired; task parks let grow = Rc::clone(&sim.replicas[0].shards[0]); executor.spawn(async move { - grow.init_partition(ns_grow, None, None, None, false); + grow.init_partition(ns_grow, None, None, None, false, None); }); executor.run_until_stalled(POLL_BUDGET); // grow while the borrow is live })) @@ -2424,7 +2556,7 @@ mod tests { executor.run_until_stalled(POLL_BUDGET); let grow = Rc::clone(&sim.replicas[0].shards[0]); executor.spawn(async move { - grow.init_partition(ns_grow, None, None, None, false); + grow.init_partition(ns_grow, None, None, None, false, None); }); executor.run_until_stalled(POLL_BUDGET);