You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/28 09:39:09 UTC
(iggy) 01/01: fix(cluster): start a fresh partition group in the metadata view
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch fix_view_divergence_during_bootstrap
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit fcfe6e7e590f882ce1c64a1d41c0821374b8adba
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Fri Aug 28 11:38:58 2026 +0200
fix(cluster): start a fresh partition group in the metadata view
---
core/integration/tests/cluster/mod.rs | 1 +
.../tests/cluster/partition_primary_routing.rs | 201 +++++++++++++++++++++
core/server/src/bootstrap.rs | 5 +
core/server/src/partition_helpers.rs | 19 +-
core/server/src/partition_reconciler.rs | 27 ++-
5 files changed, 250 insertions(+), 3 deletions(-)
diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs
index 05f42ba21..07ddcde59 100644
--- a/core/integration/tests/cluster/mod.rs
+++ b/core/integration/tests/cluster/mod.rs
@@ -24,5 +24,6 @@ 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;
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..ac3edf71b
--- /dev/null
+++ b/core/integration/tests/cluster/partition_primary_routing.rs
@@ -0,0 +1,201 @@
+// 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 and therefore still at view 0. The two assertions are deliberately
+//! separate, because they distinguish the two candidate defects:
+//!
+//! - Node 0 accepts the write. A partition primary EXISTS; "partition groups
+//! never elect" is refuted.
+//! - The advertised leader accepts the write. If this fails while the first
+//! passes, the defect is purely that the roster advertises the metadata
+//! leader as the destination for partition traffic.
+
+use std::str::FromStr;
+use std::time::Duration;
+
+use iggy::prelude::*;
+use integration::harness::TestHarness;
+use integration::iggy_harness;
+use tokio::time::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);
+/// Well past a healthy send and past the SDK's own transient replay window, so
+/// exceeding it means the client is not going to recover on its own.
+const SEND_BUDGET: Duration = Duration::from_secs(45);
+
+fn message(payload: &str) -> IggyMessage {
+ IggyMessage::from_str(payload).expect("build message")
+}
+
+/// The roster's current leader as a node index, read through an already
+/// connected client so no new connection is attempted while the cluster may
+/// still be settling.
+async fn read_leader_index(harness: &TestHarness, client: &IggyClient) -> Option<usize> {
+ let metadata = client.get_cluster_metadata().await.ok()?;
+ let leader_port = metadata
+ .nodes
+ .iter()
+ .find(|node| node.role == ClusterNodeRole::Leader)?
+ .endpoints
+ .tcp;
+ (0..harness.cluster_size()).find(|index| {
+ harness
+ .node(*index)
+ .tcp_addr()
+ .is_some_and(|address| address.port() == leader_port)
+ })
+}
+
+#[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;
+
+ let probe = harness
+ .root_client_for_node(1)
+ .await
+ .expect("root client on node 1 after the election");
+ let leader = read_leader_index(harness, &probe)
+ .await
+ .expect("the cluster must name a leader once the election settled");
+ assert_ne!(
+ leader, 0,
+ "killing node 0 must have moved the metadata plane off view 0; \
+ with the leader back at node 0 the two planes agree and the split cannot show"
+ );
+
+ // A brand-new topic: its partition consensus group starts at view 0, so
+ // its primary is replica 0, while the metadata leader is not node 0.
+ 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 dialing node 0 actually ends up. If the SDK's leader
+ // check moves it to the metadata leader, that redirect is itself the
+ // defect: it walks the client off the only node that can accept the write.
+ let node_zero_address = harness
+ .node(0)
+ .tcp_addr()
+ .expect("node 0 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;
+ println!(
+ "client dialed node 0 ({node_zero_address}); it settled on {landed_on}; \
+ metadata leader is node {leader}"
+ );
+
+ // Assertion 1: node 0 accepts. Proves a partition primary exists, so the
+ // failure mode is NOT "partition groups never elect". Retried briefly so a
+ // post-CreateTopic materialisation window is not mistaken for the defect.
+ let accepted_by_node_zero =
+ send_once(&on_node_zero, &stream_id, &topic_id, &partitioning).await;
+ assert!(
+ accepted_by_node_zero.is_ok(),
+ "a client dialing node 0 (the view-0 primary of this fresh partition group) must land on \
+ a node that accepts the write; it settled on {landed_on} and got \
+ {accepted_by_node_zero:?}"
+ );
+
+ // Assertion 2: the advertised leader accepts. This is the contract a
+ // leader-aware SDK relies on, and the one the live cluster violated.
+ let accepted_by_leader = send_once(&setup, &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 an unbounded send never returns while the defect is
+/// present. The timeout turns that livelock into a readable failure.
+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/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 587744f5d..fd7553d62 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?
diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs
index 5a2f49fd2..670971ccd 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -516,6 +516,11 @@ 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 `view_fallback`
+/// comment below for why a group left at view 0 is unreachable. `None` where
+/// no view is published yet, which keeps the historical view-0 start.
+///
/// 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 +539,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();
@@ -613,7 +619,18 @@ pub async fn build_partition_fresh(
durable_view: recovered_state
.as_ref()
.map(|state| (state.view, state.log_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. Peers materialise the same namespace
+ // from the same committed event and read the same published view,
+ // so they agree; a seed that races an election lands one replica a
+ // view behind, which its `StartView` catch-up already covers.
+ view_fallback: view_seed,
incarnation: None,
join,
},
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index 699e9c94d..9485a4fc8 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; [`METADATA_VIEW_UNKNOWN`] 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,23 @@ 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, or `None`
+ /// while the metadata plane has published no view yet.
+ fn partition_view_seed(&self) -> Option<u32> {
+ let view = self.metadata_view.load(Ordering::Relaxed);
+ if view == METADATA_VIEW_UNKNOWN {
+ return None;
+ }
+ u32::try_from(view).ok()
+ }
+
fn is_backed_off(&self, ns: IggyNamespace, cause: FailureCause, now: Instant) -> bool {
let state = self.failure_state.borrow();
if state.is_empty() {
@@ -676,6 +695,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 +1231,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 +1684,7 @@ mod tests {
CLUSTER_ID,
0,
1,
+ Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)),
))
}
@@ -1809,6 +1830,7 @@ mod tests {
CLUSTER_ID,
0,
1,
+ None,
Rc::clone(&ctx.shard.bus),
)
.await
@@ -1838,6 +1860,7 @@ mod tests {
CLUSTER_ID,
0,
1,
+ None,
Rc::clone(&ctx.shard.bus),
)
.await