This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new db14a4b17 fix(cluster): make the advertised leader accept partition
writes (#3985)
db14a4b17 is described below
commit db14a4b1718f16a0cc0f74088e0d028e80321306
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Fri Aug 28 16:53:09 2026 +0200
fix(cluster): make the advertised leader accept partition writes (#3985)
A client is told which node leads the cluster, sends a partition
write there, and is refused for its whole budget. The node it was
told about is genuinely the leader; the write is genuinely unroutable.
Each node runs a metadata consensus group plus one per partition,
each with its own view counter, and both pick a primary as
`view % replica_count`. `get_cluster_metadata` marks a node Leader
from the metadata plane alone and `ClusterNode` carries one `role`,
so a partition group's primary cannot be expressed on the wire.
Partition ops route within a node, never to a peer, so the
advertised leader cannot forward one either.
A group materialised with no durable state started at view 0,
naming replica 0 wherever the metadata plane had got to, so any
metadata election opened the gap for every topic created after.
Seed a fresh group from the metadata view shard 0 already publishes
for the roster's leader marking. Both planes then name the same
node by construction, with no wire or client change. The seed sets
`log_view` too: a primary whose `log_view` lags counts as
mid-transition and answers no `RequestStartView`. A replica that
has not seen a view defers materialising rather than reading the
unknown sentinel as 0, and that deferral counts toward the pass so
it cannot arm the reconciler's fast-skip.
This aligns the planes once, at creation. A later election on
either side parts them again and nothing forwards a write across
the gap, so the symptom returns for that group.
---
core/consensus/src/impls.rs | 116 +++++++++++-
core/harness_derive/src/attrs.rs | 12 ++
core/harness_derive/src/codegen.rs | 6 +
core/integration/src/harness/disk.rs | 19 +-
.../src/harness/orchestrator/harness.rs | 145 ++++++++++++---
core/integration/tests/cluster/mod.rs | 2 +
.../tests/cluster/partition_primary_routing.rs | 198 +++++++++++++++++++++
.../tests/cluster/staggered_bootstrap.rs | 180 +++++++++++++++++++
core/server/src/bootstrap.rs | 9 +
core/server/src/cluster_meta.rs | 8 +-
core/server/src/partition_helpers.rs | 45 +++--
core/server/src/partition_reconciler.rs | 166 ++++++++++++++++-
core/server/src/responses.rs | 2 +-
core/shard/src/lib.rs | 53 ++++--
core/simulator/src/lib.rs | 136 +++++++++++++-
15 files changed, 1031 insertions(+), 66 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 2e4741dab..28865344f 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<u32>,
+ /// 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<u32>,
/// Non-zero boot incarnation; `None` keeps the default.
pub incarnation: Option<u128>,
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<u32>,
+}
+
+/// 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<u32>,
+) -> 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<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
/// # Panics
/// - If `replica >= replica_count`.
@@ -1230,6 +1291,14 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
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 9d1f07cbd..136b37901 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<syn::Path>,
pub cluster_nodes: ClusterNodesValue,
pub jwks_server: Option<JwksAttrs>,
+ /// 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 481e39af4..591f11fad 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 332a531ef..97d42b16c 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<PathBuf> {
/// 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 f324ca39b..60324adc1 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<ClientBuilder, TestBinaryError> {
- 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<IggyClient, TestBinaryError> {
- 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 05f42ba21..c1fafecdc 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 000000000..064f0134b
--- /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 000000000..a9979906f
--- /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 587744f5d..7d8a27716 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 7372d3e59..c251ef602 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<u8> {
+ ///
+ /// 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<u8> {
if self.nodes.is_empty() {
return None;
}
diff --git a/core/server/src/partition_helpers.rs
b/core/server/src/partition_helpers.rs
index 5a2f49fd2..f548f5de7 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<u32>,
bus: Rc<IggyMessageBus>,
) -> Result<IggyPartition<Rc<IggyMessageBus>>, 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 699e9c94d..051d81c69 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<AtomicU64>,
failure_state: RefCell<AHashMap<(IggyNamespace, FailureCause),
FailureRecord>>,
/// `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<AtomicU64>,
) -> 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<u32> {
+ 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<ServerShard>,
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<TestShard>,
+ total_shards: u16,
+ config: Rc<ServerConfig>,
+ metadata_view: Arc<AtomicU64>,
+ ) -> Rc<ReconcilerCtx> {
+ 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 f4e6bc4e5..21b8e6497 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 358b8e62b..9098c7d63 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3567,6 +3567,7 @@ where
recovered_state: Option<consensus::VsrState>,
retained: Option<partitions::RetainedPartitionState>,
restore_frontier: bool,
+ metadata_view: Option<u32>,
) 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 2a2f396ba..19d89dc87 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<u8> {
+ (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<u8>
{
(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);
