fix(cluster): recover writes after a fast partition primary rejoin (#3987)
A fast partition primary rejoin could leave the group unable to
commit new writes. A backup could adopt suffix headers without
obtaining their bodies because repair stopped at commit_max. Clients
then handled the typed non-admission response by checking metadata
leadership, even though metadata and partition groups can elect
different primaries.
Extend suffix repair to the available frontier and clear completed
sessions after the later commit catches up. Reject writes on stale
backups before local offset validation, so callers receive the typed
response required for safe rerouting.
Add one-pass roster routing for HTTP and the Rust, Node, Go, Java,
and C# clients. Coordinate concurrent Rust reconnects and prevent
ambiguous replicated writes from being replayed under a new session.
diff --git a/Cargo.lock b/Cargo.lock
index ac8d177..c6aa065 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2742,9 +2742,9 @@
[[package]]
name = "chacha20"
-version = "0.10.1"
+version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
dependencies = [
"cfg-if",
"cpufeatures 0.3.0",
@@ -6659,7 +6659,7 @@
[[package]]
name = "iggy"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
dependencies = [
"async-broadcast",
"async-dropper",
@@ -6693,7 +6693,7 @@
[[package]]
name = "iggy-bench"
-version = "0.6.0-edge.4"
+version = "0.6.0-edge.5"
dependencies = [
"async-trait",
"bench-report",
@@ -6750,7 +6750,7 @@
[[package]]
name = "iggy-cli"
-version = "0.14.0-edge.4"
+version = "0.14.0-edge.5"
dependencies = [
"anyhow",
"apple-native-keyring-store",
@@ -6784,7 +6784,7 @@
[[package]]
name = "iggy-connectors"
-version = "0.5.0-edge.4"
+version = "0.5.0-edge.5"
dependencies = [
"async-trait",
"axum",
@@ -6856,7 +6856,7 @@
[[package]]
name = "iggy-mcp"
-version = "0.5.0-edge.3"
+version = "0.5.0-edge.4"
dependencies = [
"axum",
"axum-server",
@@ -6890,7 +6890,7 @@
[[package]]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
dependencies = [
"aligned-vec",
"bytemuck",
@@ -6903,7 +6903,7 @@
[[package]]
name = "iggy_common"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
dependencies = [
"aes-gcm",
"async-broadcast",
@@ -12031,7 +12031,7 @@
[[package]]
name = "server"
-version = "0.9.0-edge.5"
+version = "0.9.0-edge.6"
dependencies = [
"ahash 0.8.12",
"argon2",
diff --git a/Cargo.toml b/Cargo.toml
index 32a274e..ea2d476 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -206,10 +206,10 @@
iceberg = "0.9.1"
iceberg-catalog-rest = "0.9.1"
iceberg-storage-opendal = "0.9.1"
-iggy = { path = "core/sdk", version = "0.11.0-edge.4" }
-iggy-cli = { path = "core/cli", version = "0.14.0-edge.4" }
-iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.4" }
-iggy_common = { path = "core/common", version = "0.11.0-edge.4" }
+iggy = { path = "core/sdk", version = "0.11.0-edge.5" }
+iggy-cli = { path = "core/cli", version = "0.14.0-edge.5" }
+iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.5" }
+iggy_common = { path = "core/common", version = "0.11.0-edge.5" }
iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.3" }
indexmap = "2.14.0"
integration = { path = "core/integration" }
diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock
index c0f0560..b401a9a 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev4"
+version = "0.9.0.dev5"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/bdd/rust/tests/steps/streams.rs b/bdd/rust/tests/steps/streams.rs
index 2fe2c93..ba1248f 100644
--- a/bdd/rust/tests/steps/streams.rs
+++ b/bdd/rust/tests/steps/streams.rs
@@ -18,6 +18,11 @@
use crate::common::global_context::GlobalContext;
use cucumber::{given, then, when};
use iggy::prelude::{Identifier, StreamClient, StreamUpdateOptions};
+use std::time::Duration;
+use tokio::time::{Instant, sleep};
+
+const METADATA_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(2);
+const METADATA_CONVERGENCE_POLL: Duration = Duration::from_millis(10);
#[given("I have no streams in the system")]
pub async fn given_no_streams(world: &mut GlobalContext) {
@@ -123,11 +128,18 @@
#[then("getting the stream by its numeric ID should return no stream")]
pub async fn then_get_stream_returns_no_stream(world: &mut GlobalContext) {
- get_stream_by_numeric_id(world).await;
- assert!(
- world.last_stream_name.is_none(),
- "Deleted stream should not be returned"
- );
+ let deadline = Instant::now() + METADATA_CONVERGENCE_TIMEOUT;
+ loop {
+ get_stream_by_numeric_id(world).await;
+ if world.last_stream_name.is_none() {
+ return;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "Deleted stream should not be returned after {METADATA_CONVERGENCE_TIMEOUT:?}"
+ );
+ sleep(METADATA_CONVERGENCE_POLL).await;
+ }
}
async fn create_stream(world: &mut GlobalContext, stream_name: &str) {
diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml
index e36c876..bab4769 100644
--- a/core/ai/mcp/Cargo.toml
+++ b/core/ai/mcp/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-mcp"
-version = "0.5.0-edge.3"
+version = "0.5.0-edge.4"
description = "MCP Server for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/bench/Cargo.toml b/core/bench/Cargo.toml
index cc58073..60e521d 100644
--- a/core/bench/Cargo.toml
+++ b/core/bench/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-bench"
-version = "0.6.0-edge.4"
+version = "0.6.0-edge.5"
edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/apache/iggy"
diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml
index 4cb4325..79e99c7 100644
--- a/core/binary_protocol/Cargo.toml
+++ b/core/binary_protocol/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
description = "Wire protocol types and codec for the Iggy binary protocol. Shared between server and SDK."
edition = "2024"
rust-version.workspace = true
diff --git a/core/cli/Cargo.toml b/core/cli/Cargo.toml
index fbdba38..bfecb9e 100644
--- a/core/cli/Cargo.toml
+++ b/core/cli/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-cli"
-version = "0.14.0-edge.4"
+version = "0.14.0-edge.5"
edition = "2024"
rust-version.workspace = true
authors = ["bartosz.ciesla@gmail.com"]
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index cdfaa64..8055104 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_common"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index 0c04c59..37b5351 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-connectors"
-version = "0.5.0-edge.4"
+version = "0.5.0-edge.5"
description = "Connectors runtime for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/integration/src/harness/handle/client_builder.rs b/core/integration/src/harness/handle/client_builder.rs
index 040056c..4dd449a 100644
--- a/core/integration/src/harness/handle/client_builder.rs
+++ b/core/integration/src/harness/handle/client_builder.rs
@@ -38,12 +38,12 @@
use crate::harness::error::TestBinaryError;
use iggy::http::http_client::HttpClient;
use iggy::prelude::{
- Client, HttpClientConfig, IggyClient, QuicClientConfig, TcpClient, TcpClientConfig, UserClient,
- WebSocketClientConfig,
+ Client, HttpClientConfig, IggyClient, IggyDuration, QuicClientConfig, TcpClient,
+ TcpClientConfig, UserClient, WebSocketClientConfig,
};
use iggy::quic::quic_client::QuicClient;
use iggy::websocket::websocket_client::WebSocketClient;
-use iggy_common::TransportProtocol;
+use iggy_common::{AutoLogin, Credentials, TransportProtocol};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
@@ -69,7 +69,9 @@
transport: TransportProtocol,
connection: ServerConnection,
auto_login: Option<AutoLoginConfig>,
+ reconnecting_login: bool,
tcp_nodelay: bool,
+ reestablish_after: Option<IggyDuration>,
encryptor: Option<Arc<iggy_common::EncryptorKind>>,
}
@@ -79,7 +81,9 @@
transport,
connection,
auto_login: None,
+ reconnecting_login: false,
tcp_nodelay: false,
+ reestablish_after: None,
encryptor: None,
}
}
@@ -90,6 +94,14 @@
self
}
+ /// Configure the binary transport itself to restore the root session
+ /// after reconnecting instead of running a one-time harness login.
+ pub fn with_reconnecting_root_login(mut self) -> Self {
+ self.auto_login = Some(AutoLoginConfig::root());
+ self.reconnecting_login = true;
+ self
+ }
+
/// Enable automatic login with custom credentials after connection.
pub fn with_login(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
self.auto_login = Some(AutoLoginConfig::new(username, password));
@@ -102,6 +114,13 @@
self
}
+ /// Override how long a binary transport prefers its previous endpoint
+ /// before rotating through the cluster roster after a disconnect.
+ pub fn with_reestablish_after(mut self, reestablish_after: IggyDuration) -> Self {
+ self.reestablish_after = Some(reestablish_after);
+ self
+ }
+
/// Set the client-side encryptor for encrypting/decrypting message payloads and headers.
pub fn with_encryptor(mut self, encryptor: Arc<iggy_common::EncryptorKind>) -> Self {
self.encryptor = Some(encryptor);
@@ -117,7 +136,9 @@
TransportProtocol::WebSocket => self.create_websocket_client().await?,
};
- if let Some(ref login) = self.auto_login {
+ if let Some(ref login) = self.auto_login
+ && (!self.reconnecting_login || self.transport == TransportProtocol::Http)
+ {
client
.login_user(&login.username, &login.password)
.await
@@ -142,7 +163,7 @@
let tls_enabled = self.connection.tls.is_some();
let tls_validate = self.connection.tls.as_ref().is_some_and(|t| !t.self_signed);
- let config = TcpClientConfig {
+ let mut config = TcpClientConfig {
server_address: addr.to_string(),
nodelay: self.tcp_nodelay,
tls_enabled,
@@ -153,8 +174,12 @@
.as_ref()
.map(|p| p.to_string_lossy().to_string()),
tls_validate_certificate: tls_validate,
+ auto_login: self.binary_auto_login(),
..TcpClientConfig::default()
};
+ if let Some(reestablish_after) = self.reestablish_after {
+ config.reconnection.reestablish_after = reestablish_after;
+ }
let client =
TcpClient::create(Arc::new(config)).map_err(|e| TestBinaryError::ClientCreation {
@@ -213,11 +238,15 @@
message: "QUIC transport not available".to_string(),
})?;
- let config = QuicClientConfig {
+ let mut config = QuicClientConfig {
server_address: addr.to_string(),
max_idle_timeout: 2_000_000,
+ auto_login: self.binary_auto_login(),
..QuicClientConfig::default()
};
+ if let Some(reestablish_after) = self.reestablish_after {
+ config.reconnection.reestablish_after = reestablish_after;
+ }
let client =
QuicClient::create(Arc::new(config)).map_err(|e| TestBinaryError::ClientCreation {
@@ -256,7 +285,7 @@
.as_ref()
.is_some_and(|t| !t.self_signed);
- let config = WebSocketClientConfig {
+ let mut config = WebSocketClientConfig {
server_address: addr.to_string(),
tls_enabled,
tls_domain: "localhost".to_string(),
@@ -266,8 +295,12 @@
.as_ref()
.map(|p| p.to_string_lossy().to_string()),
tls_validate_certificate: tls_validate,
+ auto_login: self.binary_auto_login(),
..WebSocketClientConfig::default()
};
+ if let Some(reestablish_after) = self.reestablish_after {
+ config.reconnection.reestablish_after = reestablish_after;
+ }
let client = WebSocketClient::create(Arc::new(config)).map_err(|e| {
TestBinaryError::ClientCreation {
@@ -292,6 +325,20 @@
))
}
+ fn binary_auto_login(&self) -> AutoLogin {
+ if !self.reconnecting_login {
+ return AutoLogin::Disabled;
+ }
+ self.auto_login
+ .as_ref()
+ .map_or(AutoLogin::Disabled, |login| {
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ login.username.clone(),
+ login.password.clone().into(),
+ ))
+ })
+ }
+
fn get_address_string(&self) -> String {
match self.transport {
TransportProtocol::Tcp => self
diff --git a/core/integration/tests/cluster/fast_primary_rejoin.rs b/core/integration/tests/cluster/fast_primary_rejoin.rs
new file mode 100644
index 0000000..435c4ce
--- /dev/null
+++ b/core/integration/tests/cluster/fast_primary_rejoin.rs
@@ -0,0 +1,472 @@
+// 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.
+
+//! Same-handle continuity when the primary leaves and comes back.
+//!
+//! Two sibling scenarios share one setup and differ only in the rejoin step.
+//! A producing client is pinned to the primary, a follower restarts and
+//! progress is required, then the primary goes away. In the baseline the
+//! primary stays down and the client must settle on a survivor. In the
+//! fast-rejoin case the primary is started again immediately, so the old
+//! endpoint accepts TCP and answers metadata while the partition group's
+//! primaryship may live elsewhere. The same client object must reach a
+//! confirmed write and read it back in both cases: an endpoint that can hold
+//! a session hostage without being able to admit the write is a routing
+//! failure, not a durability failure.
+
+use iggy::prelude::*;
+use iggy_common::store_consumer_offset::StoreConsumerOffset;
+use iggy_common::{IggyMessagesBatch, SendMessagesConfirmations};
+use integration::harness::{TestHarness, disk};
+use integration::iggy_harness;
+use reqwest::StatusCode;
+use std::time::Duration;
+use tokio::time::sleep;
+
+use crate::server::http_client::HttpClient;
+
+const STREAM_NAME: &str = "rejoin-stream";
+const TOPIC_NAME: &str = "rejoin-topic";
+const PARTITION_ID: u32 = 0;
+
+/// Acks the pinned producer must capture before any disruption, so the
+/// session is warm and mid-stream rather than freshly connected.
+const WARM_ACKS: usize = 20;
+
+/// Budget for reaching the warm ack count against the healthy cluster.
+const WARMUP_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Budget for one confirmed send after a disruption. Covers an election, a
+/// reconnect sweep, and a leader recheck, while staying far below the point
+/// where repeated full response timeouts would read as progress.
+const RESUME_BUDGET: Duration = Duration::from_secs(30);
+
+/// Budget for reading the resumed record back through the same handle.
+const READBACK_BUDGET: Duration = Duration::from_secs(10);
+
+const RETRY_PAUSE: Duration = Duration::from_millis(250);
+
+fn build_message(payload: &str) -> IggyMessage {
+ IggyMessage::builder()
+ .payload(payload.to_owned().into())
+ .build()
+ .expect("build message")
+}
+
+/// Create the stream and an eagerly flushed single-partition topic, so every
+/// confirmed send is durable and each scenario is purely a routing question.
+async fn create_stream_and_topic(harness: &TestHarness) {
+ let setup_client = harness.tcp_root_client().await.unwrap();
+ setup_client
+ .create_stream(STREAM_NAME)
+ .await
+ .expect("create stream");
+ let options = TopicCreateOptions {
+ partitions_count: Some(1),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ messages_required_to_save: Some(1),
+ enforce_fsync: Some(true),
+ ..TopicCreateOptions::default()
+ };
+ setup_client
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ &options,
+ )
+ .await
+ .expect("create topic");
+}
+
+/// Connect one client pinned to the current primary's own endpoint, the way a
+/// leader-aware SDK ends up connected to whichever node answers as leader.
+async fn pinned_producer(
+ harness: &TestHarness,
+ leader: usize,
+ transport: TransportProtocol,
+ reestablish_after: Option<IggyDuration>,
+) -> (IggyClient, String) {
+ let node = harness.node(leader);
+ let primary_endpoint = match transport {
+ TransportProtocol::Tcp => node.tcp_addr(),
+ TransportProtocol::Quic => node.quic_addr(),
+ TransportProtocol::WebSocket => node.websocket_addr(),
+ TransportProtocol::Http => panic!("HTTP does not expose a persistent Iggy client"),
+ }
+ .unwrap_or_else(|| panic!("leader exposes a {transport} endpoint"))
+ .to_string();
+ let builder = match transport {
+ TransportProtocol::Tcp => node.tcp_client(),
+ TransportProtocol::Quic => node.quic_client(),
+ TransportProtocol::WebSocket => node.websocket_client(),
+ TransportProtocol::Http => panic!("HTTP does not expose a persistent Iggy client"),
+ }
+ .unwrap_or_else(|error| panic!("leader exposes a {transport} client: {error}"));
+ let builder = match reestablish_after {
+ Some(duration) => builder.with_reestablish_after(duration),
+ None => builder,
+ };
+ let producer = builder
+ .with_reconnecting_root_login()
+ .connect()
+ .await
+ .expect("connect the producer to the primary");
+ assert_eq!(
+ producer.get_connection_info().await.server_address,
+ primary_endpoint,
+ "the producer must be pinned to the node this test disrupts, or it proves nothing"
+ );
+ (producer, primary_endpoint)
+}
+
+/// Drive confirmed sends until `count` acks land, within `budget`.
+async fn require_acked_sends(
+ producer: &IggyClient,
+ label: &str,
+ count: usize,
+ budget: Duration,
+) -> Vec<(u64, String)> {
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let partitioning = Partitioning::partition_id(PARTITION_ID);
+ let deadline = tokio::time::Instant::now() + budget;
+ let mut acked = 0usize;
+ let mut confirmations = Vec::with_capacity(count);
+ let mut attempt = 0usize;
+ let mut last_error: Option<IggyError> = None;
+ while acked < count {
+ let now = tokio::time::Instant::now();
+ assert!(
+ now < deadline,
+ "{label}: only {acked}/{count} confirmed sends within {budget:?} \
+ ({attempt} attempts, last error: {last_error:?})"
+ );
+ let payload = format!("{label}-{attempt:05}");
+ let mut messages = vec![build_message(&payload)];
+ attempt += 1;
+ let send = producer.send_messages(&stream, &topic, &partitioning, &mut messages);
+ match tokio::time::timeout(deadline - now, send).await {
+ Ok(Ok(response)) => {
+ let confirmation = response
+ .confirmations
+ .first()
+ .unwrap_or_else(|| panic!("{label}: the VSR server confirms every send"));
+ confirmations.push((confirmation.base_offset, payload));
+ acked += 1;
+ }
+ Ok(Err(error)) => {
+ last_error = Some(error);
+ sleep(RETRY_PAUSE).await;
+ }
+ Err(_elapsed) => {
+ panic!(
+ "{label}: an attempt outlived the whole {budget:?} budget \
+ ({acked}/{count} acked, {attempt} attempts, last error: {last_error:?})"
+ );
+ }
+ }
+ }
+ confirmations
+}
+
+/// Poll the topic through the same client and require every confirmed record
+/// at its assigned offset, including the write confirmed after failover.
+async fn require_readback(producer: &IggyClient, expected: &[(u64, String)]) {
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let deadline = tokio::time::Instant::now() + READBACK_BUDGET;
+ let last_offset = expected
+ .iter()
+ .map(|(offset, _)| *offset)
+ .max()
+ .expect("at least one confirmed record");
+ let count = u32::try_from(last_offset + 1).expect("test offsets fit in u32");
+ let mut last_missing = Vec::new();
+ loop {
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "the same handle must read every confirmed record after the failover \
+ (last missing offsets: {last_missing:?})"
+ );
+ match producer
+ .poll_messages(
+ &stream,
+ &topic,
+ Some(PARTITION_ID),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ count,
+ false,
+ )
+ .await
+ {
+ Ok(polled) => {
+ last_missing = expected
+ .iter()
+ .filter(|(offset, payload)| {
+ !polled.messages.iter().any(|message| {
+ message.header.offset == *offset
+ && message.payload.as_ref() == payload.as_bytes()
+ })
+ })
+ .map(|(offset, _)| *offset)
+ .collect();
+ if last_missing.is_empty() {
+ return;
+ }
+ sleep(RETRY_PAUSE).await;
+ }
+ Err(_) => sleep(RETRY_PAUSE).await,
+ }
+ }
+}
+
+/// Shared prologue: topic, pinned producer, warm acks, follower restart, and
+/// required progress after the follower is back. Returns the producer and the
+/// primary it is pinned to.
+async fn warmed_producer_past_follower_restart(
+ harness: &mut TestHarness,
+ transport: TransportProtocol,
+ reestablish_after: Option<IggyDuration>,
+) -> (IggyClient, usize, String, Vec<(u64, String)>) {
+ create_stream_and_topic(harness).await;
+ let leader = disk::leader_node_index(harness).await;
+ let (producer, primary_endpoint) =
+ pinned_producer(harness, leader, transport, reestablish_after).await;
+ let mut acked = require_acked_sends(&producer, "warmup", WARM_ACKS, WARMUP_TIMEOUT).await;
+
+ let follower = (0..harness.cluster_size())
+ .find(|index| *index != leader)
+ .expect("a three-node cluster has a follower");
+ harness
+ .restart_node(follower)
+ .expect("restart the follower with its data intact");
+ acked.extend(require_acked_sends(&producer, "post-follower-restart", 1, RESUME_BUDGET).await);
+
+ (producer, leader, primary_endpoint, acked)
+}
+
+async fn require_resume_after_primary_stop(
+ harness: &mut TestHarness,
+ transport: TransportProtocol,
+) {
+ let (producer, leader, primary_endpoint, mut acked) =
+ warmed_producer_past_follower_restart(harness, transport, None).await;
+
+ harness.stop_node(leader).expect("stop the primary");
+
+ acked.extend(require_acked_sends(&producer, "post-primary-stop", 1, RESUME_BUDGET).await);
+ assert_ne!(
+ producer.get_connection_info().await.server_address,
+ primary_endpoint,
+ "the send that resumed must have landed on a survivor"
+ );
+ require_readback(&producer, &acked).await;
+}
+
+async fn require_resume_after_fast_primary_rejoin(
+ harness: &mut TestHarness,
+ transport: TransportProtocol,
+) {
+ let (producer, leader, _primary_endpoint, mut acked) =
+ warmed_producer_past_follower_restart(harness, transport, None).await;
+
+ harness
+ .restart_node(leader)
+ .expect("restart the primary with its data intact");
+
+ acked.extend(require_acked_sends(&producer, "post-primary-restart", 1, RESUME_BUDGET).await);
+ require_readback(&producer, &acked).await;
+}
+
+async fn require_resume_after_fast_primary_rejoin_with_dead_roster_hop(
+ harness: &mut TestHarness,
+ transport: TransportProtocol,
+) {
+ let (producer, leader, _primary_endpoint, mut acked) =
+ warmed_producer_past_follower_restart(harness, transport, None).await;
+ let dead_hop = (leader + 1) % harness.cluster_size();
+ harness
+ .stop_node(dead_hop)
+ .expect("stop the first roster hop after the original primary");
+ harness
+ .restart_node(leader)
+ .expect("restart the primary with its data intact");
+
+ acked.extend(require_acked_sends(&producer, "post-dead-roster-hop", 1, RESUME_BUDGET).await);
+ require_readback(&producer, &acked).await;
+}
+
+/// Baseline sibling: the primary stops gracefully and stays down. The same
+/// client must settle on a survivor, write, and read back.
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_pinned_producer_when_its_primary_stops_and_stays_down_should_resume(
+ harness: &mut TestHarness,
+) {
+ require_resume_after_primary_stop(harness, TransportProtocol::Tcp).await;
+}
+
+/// Fast-rejoin sibling: the primary stops gracefully and is started again
+/// immediately, so its endpoint answers TCP and metadata as a rejoining
+/// follower while the group primaryship settles elsewhere. The same client
+/// must still reach a confirmed write and read the stream back.
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_pinned_producer_when_its_primary_restarts_quickly_should_resume(
+ harness: &mut TestHarness,
+) {
+ require_resume_after_fast_primary_rejoin(harness, TransportProtocol::Tcp).await;
+}
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_zero_cooldown_tcp_producer_when_replay_lands_on_a_partition_backup_should_resume(
+ harness: &mut TestHarness,
+) {
+ let (producer, leader, _primary_endpoint, mut acked) = warmed_producer_past_follower_restart(
+ harness,
+ TransportProtocol::Tcp,
+ Some(IggyDuration::from(0u64)),
+ )
+ .await;
+
+ harness
+ .restart_node(leader)
+ .expect("restart the primary with its data intact");
+
+ acked.extend(require_acked_sends(&producer, "zero-cooldown-replay", 1, RESUME_BUDGET).await);
+ require_readback(&producer, &acked).await;
+}
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_quic_producer_when_its_primary_restarts_quickly_should_resume(
+ harness: &mut TestHarness,
+) {
+ require_resume_after_fast_primary_rejoin(harness, TransportProtocol::Quic).await;
+}
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_quic_producer_when_its_first_roster_hop_is_down_should_reach_the_partition_primary(
+ harness: &mut TestHarness,
+) {
+ require_resume_after_fast_primary_rejoin_with_dead_roster_hop(harness, TransportProtocol::Quic)
+ .await;
+}
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_websocket_producer_when_its_primary_restarts_quickly_should_resume(
+ harness: &mut TestHarness,
+) {
+ require_resume_after_fast_primary_rejoin(harness, TransportProtocol::WebSocket).await;
+}
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_websocket_producer_when_its_first_roster_hop_is_down_should_reach_the_partition_primary(
+ harness: &mut TestHarness,
+) {
+ require_resume_after_fast_primary_rejoin_with_dead_roster_hop(
+ harness,
+ TransportProtocol::WebSocket,
+ )
+ .await;
+}
+
+/// The stateless HTTP transport has no client-side partition routing. After
+/// the old primary rejoins as a backup, its listener must walk the bounded
+/// server roster for acknowledged partition writes.
+#[iggy_harness(
+ cluster_nodes = 3,
+ server(
+ http.jwt.encoding_secret = "0123456789abcdef0123456789abcdef",
+ http.jwt.decoding_secret = "0123456789abcdef0123456789abcdef"
+ )
+)]
+async fn given_http_writes_on_a_rejoined_backup_when_the_primary_moved_should_forward_once(
+ harness: &mut TestHarness,
+) {
+ let (producer, leader, primary_endpoint, mut acked) =
+ warmed_producer_past_follower_restart(harness, TransportProtocol::Tcp, None).await;
+ harness
+ .restart_node(leader)
+ .expect("restart the primary with its data intact");
+
+ acked.extend(require_acked_sends(&producer, "settled-primary", 1, RESUME_BUDGET).await);
+ assert_ne!(
+ producer.get_connection_info().await.server_address,
+ primary_endpoint,
+ "the HTTP target must be a partition backup for this test"
+ );
+
+ let http_addr = harness
+ .node(leader)
+ .http_addr()
+ .expect("the restarted node exposes HTTP");
+ let http = HttpClient::login_root_no_redirect(format!("http://{http_addr}")).await;
+ let payload = "http-after-primary-rejoin".to_owned();
+ let message = build_message(&payload);
+ let messages = vec![message];
+ let body = SendMessages {
+ metadata_length: 0,
+ stream_id: Identifier::default(),
+ topic_id: Identifier::default(),
+ partitioning: Partitioning::partition_id(PARTITION_ID),
+ batch: IggyMessagesBatch::from(&messages),
+ };
+ let response = http
+ .client
+ .post(http.url(&format!(
+ "/streams/{STREAM_NAME}/topics/{TOPIC_NAME}/messages"
+ )))
+ .bearer_auth(&http.token)
+ .json(&body)
+ .send()
+ .await
+ .expect("forwarded HTTP produce");
+ assert_eq!(response.status(), StatusCode::CREATED);
+ let confirmation: SendMessagesConfirmations =
+ response.json().await.expect("decode confirmations");
+ let confirmation = confirmation
+ .confirmations
+ .first()
+ .expect("acknowledged HTTP produce has a confirmation");
+ acked.push((confirmation.base_offset, payload));
+
+ let offsets_path = format!("/streams/{STREAM_NAME}/topics/{TOPIC_NAME}/consumer-offsets");
+ let consumer_id = Identifier::numeric(1).expect("valid consumer id");
+ let store = StoreConsumerOffset {
+ consumer: Consumer::new(consumer_id),
+ partition_id: Some(PARTITION_ID),
+ offset: 0,
+ };
+ let response = http
+ .client
+ .put(http.url(&offsets_path))
+ .bearer_auth(&http.token)
+ .json(&store)
+ .send()
+ .await
+ .expect("forwarded HTTP offset store");
+ assert_eq!(response.status(), StatusCode::NO_CONTENT);
+
+ let response = http
+ .client
+ .delete(http.url(&format!("{offsets_path}/1?partition_id={PARTITION_ID}")))
+ .bearer_auth(&http.token)
+ .send()
+ .await
+ .expect("forwarded HTTP offset delete");
+ assert_eq!(response.status(), StatusCode::NO_CONTENT);
+ require_readback(&producer, &acked).await;
+}
diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs
index c1fafec..940e1fc 100644
--- a/core/integration/tests/cluster/mod.rs
+++ b/core/integration/tests/cluster/mod.rs
@@ -21,6 +21,7 @@
mod crash_offset_reuse;
mod crash_recovery_corruption;
mod failover_client_continuity;
+mod fast_primary_rejoin;
mod metadata_checkpoint_restart;
mod metadata_state_transfer;
mod multi_shard_partition_convergence;
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index f31ca33..c880723 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -39,7 +39,7 @@
mod purge_vsr;
// Shared HTTP transport plumbing (session + verb helpers) for the raw-HTTP
// server suites below.
-mod http_client;
+pub(crate) mod http_client;
// Raw-HTTP data-plane contract against the server's shard-0 listener.
mod http_vsr;
// Raw-HTTP wire-contract residue against the server (status codes + typed error
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 557af43..34900e5 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -1974,6 +1974,30 @@
message
};
+ // State-dependent admission belongs to the primary. A backup can
+ // lag the committed offset table or message frontier and would
+ // otherwise turn a routing artifact into a terminal 404 or 400.
+ // Reject it first with the only response that proves the request
+ // was never admitted, so the caller may safely retry elsewhere.
+ if consensus.is_follower() || !consensus.is_normal() || consensus.is_transferring() {
+ emit_partition_diag(
+ tracing::Level::WARN,
+ &PartitionDiagEvent::new(
+ ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions),
+ "rejecting client request on non-primary partition replica",
+ )
+ .with_operation(message.header().operation),
+ );
+ Self::send_partition_deny_or_log(
+ consensus,
+ message.header(),
+ IggyError::TransientNotAccepted.as_code(),
+ "non-primary transient reply send failed",
+ )
+ .await;
+ return;
+ }
+
// Parse once for both the delete-existence check and AckLevel dispatch.
let consumer_offset = match message.header().operation {
Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset => {
@@ -2064,32 +2088,6 @@
}
}
- // A client op landing on a non-primary (or mid-view-change)
- // replica is a routing artifact -- e.g. the roster still points
- // here while this group's primaryship moved after a restart.
- // Answer the typed transient instead of asserting: the SDK
- // replays and its leader recheck re-routes, whereas a panic
- // kills the shard and a silent drop wedges the client until its
- // read timeout.
- if consensus.is_follower() || !consensus.is_normal() || consensus.is_transferring() {
- emit_partition_diag(
- tracing::Level::WARN,
- &PartitionDiagEvent::new(
- ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions),
- "rejecting client request on non-primary partition replica",
- )
- .with_operation(message.header().operation),
- );
- Self::send_partition_deny_or_log(
- consensus,
- message.header(),
- IggyError::TransientNotAccepted.as_code(),
- "non-primary transient reply send failed",
- )
- .await;
- return;
- }
-
// NoAck -> fast path. Quorum -> VSR pipeline.
if let Some((kind, consumer_id, offset, AckLevel::NoAck)) = consumer_offset
&& matches!(
@@ -4465,10 +4463,29 @@
/// commit walk runs at `RepairDone`, after the floor is known.
pub async fn apply_repaired_prepare(&mut self, message: Message<PrepareHeader>) {
let header = *message.header();
- let Some(session) = &self.repair else {
+ let Some(session) = self.repair else {
return;
};
- if header.op <= self.consensus().commit_min() || header.op > session.to_op {
+ let consensus = self.consensus();
+ if !consensus.is_normal() || consensus.view() != session.view {
+ self.repair = None;
+ return;
+ }
+ if header.op <= consensus.commit_min() || header.op > session.fetch_to_op {
+ return;
+ }
+ let canonical_checksum = consensus
+ .with_pending_view_log(|pending| {
+ pending
+ .headers
+ .iter()
+ .find(|expected| expected.op == header.op)
+ .map(|expected| expected.checksum)
+ })
+ .flatten();
+ if canonical_checksum.is_some_and(|expected| expected != header.checksum)
+ || (header.op > session.commit_to_op && canonical_checksum.is_none())
+ {
return;
}
// Any in-window frame proves the stream is alive; only silence
@@ -4482,7 +4499,9 @@
let applied = if header.operation == Operation::SendMessages {
match self.append_repaired_send_messages(message).await {
Ok(base_offset) => {
- if let (Some(base_offset), Some(session)) = (base_offset, self.repair.as_mut())
+ if header.op <= session.commit_to_op
+ && let (Some(base_offset), Some(session)) =
+ (base_offset, self.repair.as_mut())
{
session.first_batch_offset = Some(
session
@@ -4537,6 +4556,10 @@
let Some(session) = self.repair else {
return RepairConclusion::Done;
};
+ if !self.consensus().is_normal() || self.consensus().view() != session.view {
+ self.repair = None;
+ return RepairConclusion::Done;
+ }
if let Some(floor) = session.floor {
// A peer may have evicted past this replica's commit frontier;
// an unclamped floor would drive commit_min above commit_max and
@@ -4557,6 +4580,11 @@
let stand_in = durable_end
.map(|durable| durable.saturating_add(1))
.max(self.installed_frontier);
+ let committed_shape = self
+ .log
+ .journal()
+ .inner
+ .repaired_window_shape(floor, session.commit_to_op);
let connected = match (session.first_batch_offset, stand_in) {
(Some(first), Some(bound)) => first <= bound,
(Some(first), None) => first == 0,
@@ -4568,7 +4596,11 @@
// a fully evicted window -- is indistinguishable from a
// message range below the floor that this replica does not
// durably own, and accepting it would serve a holed log.
- (None, _) => self.repaired_window_is_offsets_only(floor, session.to_op),
+ (None, _) => {
+ floor < session.commit_to_op
+ && committed_shape.complete
+ && !committed_shape.holds_messages
+ }
};
if !connected {
tracing::error!(
@@ -4593,11 +4625,11 @@
// Both are the state-transfer trigger; the session is
// dropped here so the caller's arming funnel starts clean,
// and a transfer-unavailable fallback re-arms repair fresh.
- if self.repaired_window_is_complete(floor, session.to_op) {
+ if committed_shape.complete {
self.repair = None;
return RepairConclusion::FloorRefused {
floor,
- to_op: session.to_op,
+ to_op: session.commit_to_op,
};
}
return RepairConclusion::InProgress;
@@ -4616,7 +4648,23 @@
// that reached the requested frontier closes the session; anything
// less keeps it armed and the stall retry re-requests the remains
// (`commit_min + 1..`), converging over rounds.
- let done = commit_min >= session.to_op;
+ // The residency check starts at the LIVE commit point, not the
+ // session's snapshotted one: delivering the suffix bodies is what
+ // lets the group commit past `commit_to_op`, and the `commit_journal`
+ // above then evicts exactly those headers. Judged from
+ // `commit_to_op`, a fully successful repair would report itself
+ // incomplete forever, pin the session, and block every later re-arm
+ // until a view change. Ops at or below `commit_min` are committed and
+ // applied, a monotone fact the flush cannot erase, so they need no
+ // resident header to count as fetched.
+ let fetch_complete = session.fetch_to_op <= session.commit_to_op
+ || self
+ .log
+ .journal()
+ .inner
+ .repaired_window_shape(session.commit_to_op.max(commit_min), session.fetch_to_op)
+ .complete;
+ let done = commit_min >= session.commit_to_op && fetch_complete;
if done {
self.repair = None;
}
@@ -4627,7 +4675,9 @@
commit_min_before = before,
commit_min_after = commit_min,
commit_max = self.consensus().commit_max(),
- to_op = session.to_op,
+ commit_to_op = session.commit_to_op,
+ fetch_to_op = session.fetch_to_op,
+ fetch_complete,
done,
"repair window commit walk finished"
);
@@ -4638,31 +4688,6 @@
}
}
- /// Whether every op in `(floor, to_op]` is journaled. An empty window
- /// (`floor >= to_op`) counts as complete: there is nothing left that
- /// could arrive and change the floor verdict.
- fn repaired_window_is_complete(&self, floor: u64, to_op: u64) -> bool {
- self.log
- .journal()
- .inner
- .repaired_window_shape(floor, to_op)
- .complete
- }
-
- /// Whether the served repair window `(floor, to_op]` arrived complete and
- /// holds no `SendMessages` op. Only then may a commit floor be accepted
- /// without a batch anchor: the window demonstrably moved no messages, so
- /// the consumer-offset table on disk stands in below the floor. An empty
- /// window (`floor >= to_op`) carries no evidence at all and never
- /// qualifies.
- fn repaired_window_is_offsets_only(&self, floor: u64, to_op: u64) -> bool {
- if floor >= to_op {
- return false;
- }
- let shape = self.log.journal().inner.repaired_window_shape(floor, to_op);
- shape.complete && !shape.holds_messages
- }
-
/// Journal a repaired `SendMessages` prepare, preserving its embedded
/// batch stamps. A stored prepare was stamped by `append_messages` on
/// the serving replica BEFORE it was journaled, so its `base_offset` /
@@ -5476,13 +5501,20 @@
type SentFrames = Rc<RefCell<Vec<(u128, Frozen<MESSAGE_ALIGN>)>>>;
fn recording_partition() -> (IggyPartition<RecordingBus>, SentFrames) {
+ recording_partition_at(0, 1)
+ }
+
+ fn recording_partition_at(
+ replica: u8,
+ replica_count: u8,
+ ) -> (IggyPartition<RecordingBus>, SentFrames) {
let namespace = IggyNamespace::new(1, 1, 0);
let bus = RecordingBus::default();
let sent_to_clients = bus.sent_to_clients.clone();
let consensus = VsrConsensus::new(
TEST_CLUSTER,
- 0,
- 1,
+ replica,
+ replica_count,
namespace.inner(),
bus,
LocalPipeline::new(),
@@ -5585,6 +5617,32 @@
);
}
+ #[compio::test]
+ async fn on_request_delete_on_stale_backup_replies_transient_before_not_found() {
+ let (mut partition, sent_to_clients) = recording_partition_at(1, 3);
+ let client_id = 42;
+
+ partition
+ .on_request(delete_offset_request(client_id, 7, 5))
+ .await;
+
+ let sent = sent_to_clients.borrow();
+ assert_eq!(sent.len(), 1, "exactly one routing denial");
+ let (reply_client, frame) = &sent[0];
+ assert_eq!(*reply_client, client_id);
+ let header = bytemuck::checked::try_from_bytes::<ReplyHeader>(
+ &frame.as_slice()[..std::mem::size_of::<ReplyHeader>()],
+ )
+ .expect("deny frame starts with a valid reply header");
+ assert_eq!(header.status, IggyError::TransientNotAccepted.as_code());
+ assert_eq!(header.op, 0, "the backup admitted nothing");
+ assert_eq!(
+ partition.consensus().pipeline_len(),
+ 0,
+ "a backup must not replicate the delete"
+ );
+ }
+
fn unique_temp_offset_dir() -> String {
let mut dir = std::env::temp_dir();
dir.push(format!(
@@ -6599,9 +6657,20 @@
}
fn armed_session(to_op: u64, floor: u64, first_batch_offset: Option<u64>) -> RepairSession {
+ armed_fetch_session(to_op, to_op, floor, first_batch_offset)
+ }
+
+ fn armed_fetch_session(
+ to_op: u64,
+ fetch_to_op: u64,
+ floor: u64,
+ first_batch_offset: Option<u64>,
+ ) -> RepairSession {
RepairSession {
nonce: 1,
- to_op,
+ view: 0,
+ commit_to_op: to_op,
+ fetch_to_op,
floor: Some(floor),
peer: 0,
first_batch_offset,
@@ -6729,6 +6798,26 @@
}
#[compio::test]
+ async fn given_prior_view_repair_when_a_new_view_started_should_discard_it() {
+ let mut partition = test_partition();
+ partition.repair = Some(armed_fetch_session(0, 1, 0, None));
+ partition.consensus.set_view(1);
+
+ partition
+ .apply_repaired_prepare(repaired_send_prepare(1, 0, 0x11))
+ .await;
+
+ assert!(
+ partition.repair.is_none(),
+ "the prior-view session is obsolete"
+ );
+ assert!(
+ partition.log.journal().inner.header_by_op(1).is_none(),
+ "a delayed prior-view body must not enter the new view's journal"
+ );
+ }
+
+ #[compio::test]
async fn given_session_remint_when_attempts_burned_should_survive_on_partition() {
let mut partition = test_partition();
for round in 0..consensus::STATE_TRANSFER_MAX_STALL_RETRIES {
@@ -6850,6 +6939,89 @@
}
#[compio::test]
+ async fn given_empty_committed_window_with_a_suffix_fetch_should_escape_to_state_transfer() {
+ let mut partition = test_partition();
+ partition.consensus().advance_commit_max(5);
+ partition.repair = Some(armed_fetch_session(5, 9, 5, None));
+
+ let conclusion = partition.complete_repair(&repair_config()).await;
+
+ assert_eq!(
+ conclusion,
+ RepairConclusion::FloorRefused { floor: 5, to_op: 5 },
+ "the uncommitted fetch ceiling must not postpone a definitive committed-floor refusal"
+ );
+ assert!(partition.repair.is_none());
+ }
+
+ /// A one-message `SendMessages` prepare for `op`, journaled through the
+ /// replicated-apply path (which stamps offsets and re-checksums), with the
+ /// sequencer advanced the way `on_replicate` does after a real append.
+ pub(super) async fn journal_send_batch(partition: &mut IggyPartition<IggyMessageBus>, op: u64) {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let record = build_segment_record(namespace, 0);
+ let header_size = std::mem::size_of::<PrepareHeader>();
+ let total = header_size + record.len();
+ let mut message = Message::<PrepareHeader>::new(total);
+ message.as_mut_slice()[header_size..].copy_from_slice(&record);
+ let message = message.transmute_header(|_, header: &mut PrepareHeader| {
+ header.command = Command::Prepare;
+ header.operation = Operation::SendMessages;
+ header.op = op;
+ header.timestamp = op;
+ header.group = namespace.inner();
+ header.size = u32::try_from(total).expect("prepare size fits u32");
+ });
+ partition
+ .apply_replicated_operation(message)
+ .await
+ .expect("journal send batch");
+ partition.consensus().sequencer().set_sequence(op);
+ }
+
+ #[compio::test]
+ async fn given_committed_suffix_evicted_when_completing_repair_should_close_the_session() {
+ // A successful suffix repair is what lets the group commit past
+ // `commit_to_op`, and the commit walk's flush then evicts exactly the
+ // suffix headers. The completion verdict must survive that eviction:
+ // judged from resident headers alone, the fully successful session
+ // would report itself incomplete forever, stay armed, and block every
+ // later re-arm for this partition until a view change.
+ let mut partition = test_partition();
+ for op in 1..=3 {
+ journal_send_batch(&mut partition, op).await;
+ }
+ partition.consensus().advance_commit_max(3);
+ partition.repair = Some(armed_fetch_session(2, 3, 0, Some(0)));
+ partition.commit_journal(&repair_config()).await;
+ let _ = partition.log.journal().inner.evict_prefix(3).await;
+
+ let conclusion = partition.complete_repair(&repair_config()).await;
+
+ assert_eq!(conclusion, RepairConclusion::Done);
+ assert!(
+ partition.repair.is_none(),
+ "a fully committed suffix fetch must not stay armed after its \
+ headers are flushed out of the resident journal"
+ );
+ }
+
+ #[compio::test]
+ async fn given_suffix_fetch_when_its_view_is_discarded_should_clear_the_session() {
+ let mut partition = test_partition();
+ partition.repair = Some(armed_fetch_session(0, 3, 0, None));
+ partition.consensus.set_view(1);
+
+ let conclusion = partition.complete_repair(&repair_config()).await;
+
+ assert_eq!(conclusion, RepairConclusion::Done);
+ assert!(
+ partition.repair.is_none(),
+ "a discarded view must not leave its suffix fetch blocking future repair"
+ );
+ }
+
+ #[compio::test]
async fn given_repaired_batch_above_durable_end_when_floor_arrives_should_refuse_commit_floor()
{
let mut partition = test_partition();
@@ -7458,7 +7630,7 @@
#[cfg(test)]
mod purge_floor_tests {
- use super::tests::{build_segment_record, repair_config, test_partition};
+ use super::tests::{build_segment_record, journal_send_batch, repair_config, test_partition};
use super::*;
use iggy_binary_protocol::{Command, WireConsumer, WireEncode};
@@ -7479,31 +7651,6 @@
(partition, dir)
}
- /// A one-message `SendMessages` prepare for `op`, journaled through the
- /// replicated-apply path (which stamps offsets and re-checksums), with the
- /// sequencer advanced the way `on_replicate` does after a real append.
- async fn journal_send_batch(partition: &mut IggyPartition<IggyMessageBus>, op: u64) {
- let namespace = IggyNamespace::new(1, 1, 0);
- let record = build_segment_record(namespace, 0);
- let header_size = std::mem::size_of::<PrepareHeader>();
- let total = header_size + record.len();
- let mut message = Message::<PrepareHeader>::new(total);
- message.as_mut_slice()[header_size..].copy_from_slice(&record);
- let message = message.transmute_header(|_, header: &mut PrepareHeader| {
- header.command = Command::Prepare;
- header.operation = Operation::SendMessages;
- header.op = op;
- header.timestamp = op;
- header.group = namespace.inner();
- header.size = u32::try_from(total).expect("prepare size fits u32");
- });
- partition
- .apply_replicated_operation(message)
- .await
- .expect("journal send batch");
- partition.consensus().sequencer().set_sequence(op);
- }
-
/// A `StoreConsumerOffset` prepare for `op`, journaled and staged through
/// the replicated-apply path.
async fn journal_store_offset(
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index ec45aef..fcb3a9a 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -1330,6 +1330,20 @@
}
#[compio::test]
+ async fn repaired_window_shape_rejects_unbounded_sparse_window_before_allocation() {
+ let journal = PartitionJournal::<PartitionJournalMemStorage>::default();
+ journal
+ .append(build_prepare(7, HEADER_SIZE + 16).into_frozen())
+ .await
+ .expect("append");
+
+ let shape = journal.repaired_window_shape(0, u64::MAX);
+
+ assert!(!shape.complete);
+ assert!(!shape.holds_messages);
+ }
+
+ #[compio::test]
async fn repair_headers_in_serves_the_commit_point_from_the_evicted_ring() {
// Blank AT the commit point is the one slot a merge can neither adopt nor
// discard, so a quorum that all flushed there deadlocks. A flushed replica has
diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs
index df15b43..bf9d936 100644
--- a/core/partitions/src/types.rs
+++ b/core/partitions/src/types.rs
@@ -213,10 +213,20 @@
/// One in-flight journal-repair stream for a partition group.
#[derive(Debug, Clone, Copy)]
pub struct RepairSession {
- /// Fences stale repair frames from an earlier attempt.
+ /// Fences range replies from an earlier attempt. Repair bodies carry the
+ /// stored prepare header instead, so [`Self::view`] and canonical suffix
+ /// checks fence their ingest.
pub nonce: u128,
- /// Last op the stream is expected to serve (the frontier at request time).
- pub to_op: u64,
+ /// Consensus view in which this session was armed. A later view discards
+ /// the session before any delayed repair body can enter its journal.
+ pub view: u32,
+ /// Committed frontier this repair must make locally walkable. Floor
+ /// completeness and session completion are bounded here.
+ pub commit_to_op: u64,
+ /// Highest op requested from the peer. This may extend above
+ /// [`Self::commit_to_op`] only for the canonical suffix carried by the
+ /// adopted `StartView`.
+ pub fetch_to_op: u64,
/// Commit floor learned from `RangeEvicted { retained_from }`:
/// `retained_from - 1`. `None` until (unless) the serving peer reports a
/// truncated prefix.
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index e6658fd..0e3e504 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/sdk/src/leader_aware.rs b/core/sdk/src/leader_aware.rs
index b5c038f..2d1c61b 100644
--- a/core/sdk/src/leader_aware.rs
+++ b/core/sdk/src/leader_aware.rs
@@ -20,8 +20,13 @@
use iggy_common::{
ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, IggyError, TransportProtocol,
};
+use std::collections::VecDeque;
+use std::future::Future;
use std::net::SocketAddr;
use std::str::FromStr;
+use std::sync::Mutex as StdMutex;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use tokio::sync::Notify;
use tracing::{debug, info, warn};
/// Maximum number of leader redirections to prevent infinite loops
@@ -339,6 +344,248 @@
.replace("[::]", "[::1]")
}
+/// One bounded pass over the cluster endpoints a request has not tried yet.
+///
+/// The metadata leader check cannot repair a persistent not-admitted refusal:
+/// metadata and partition consensus groups elect independently, so the
+/// metadata leader can hold a follower replica of the partition a request
+/// targets. `TransientNotAccepted` marks the request as never admitted and
+/// safe to re-issue anywhere. The walk owns a roster snapshot for one request
+/// and removes every endpoint as it is attempted, so a stale roster or a
+/// failed dial cannot cycle the request back through nodes it already tried.
+#[derive(Debug)]
+pub(crate) struct RosterWalk {
+ remaining: VecDeque<String>,
+ attempted: Vec<String>,
+}
+
+impl RosterWalk {
+ pub(crate) fn new(current: &str, roster: &[String]) -> Self {
+ let mut ordered: VecDeque<String> = VecDeque::with_capacity(roster.len());
+ let start = roster
+ .iter()
+ .position(|endpoint| is_same_spelling(endpoint, current))
+ .map_or(0, |position| position + 1);
+
+ for offset in 0..roster.len() {
+ let endpoint = roster[(start + offset) % roster.len()].clone();
+ if is_same_spelling(&endpoint, current)
+ || ordered
+ .iter()
+ .any(|queued| is_same_spelling(queued, &endpoint))
+ {
+ continue;
+ }
+ ordered.push_back(endpoint);
+ }
+
+ Self {
+ remaining: ordered,
+ attempted: vec![current.to_owned()],
+ }
+ }
+
+ /// Record an endpoint chosen outside the roster walk, such as a metadata
+ /// leader redirect. Returns whether this request had not tried it before.
+ pub(crate) fn record_attempt(&mut self, endpoint: &str) -> bool {
+ self.remaining
+ .retain(|queued| !is_same_spelling(queued, endpoint));
+ if self
+ .attempted
+ .iter()
+ .any(|attempted| is_same_spelling(attempted, endpoint))
+ {
+ return false;
+ }
+ self.attempted.push(endpoint.to_owned());
+ true
+ }
+
+ pub(crate) fn next(&mut self) -> Option<String> {
+ let endpoint = self.remaining.pop_front()?;
+ self.attempted.push(endpoint.clone());
+ Some(endpoint)
+ }
+}
+
+/// Coordinates callers of one client's complete connect and authentication
+/// sequence. Only the owner runs connection work. Concurrent callers receive
+/// that exact result instead of treating `Connecting` as success.
+#[derive(Debug)]
+pub(crate) struct ConnectCoordinator {
+ id: u64,
+ active: AtomicBool,
+ abandoned: AtomicBool,
+ active_token: AtomicU64,
+ next_token: AtomicU64,
+ generation: AtomicU64,
+ result: StdMutex<Option<(u64, Result<(), IggyError>)>>,
+ changed: Notify,
+}
+
+impl ConnectCoordinator {
+ pub(crate) fn new() -> Self {
+ Self {
+ id: NEXT_CONNECT_COORDINATOR_ID.fetch_add(1, Ordering::SeqCst),
+ active: AtomicBool::new(false),
+ abandoned: AtomicBool::new(false),
+ active_token: AtomicU64::new(0),
+ next_token: AtomicU64::new(1),
+ generation: AtomicU64::new(0),
+ result: StdMutex::new(None),
+ changed: Notify::new(),
+ }
+ }
+
+ pub(crate) fn is_active(&self) -> bool {
+ self.active.load(Ordering::SeqCst)
+ }
+
+ pub(crate) fn current_owner_context(&self) -> Option<ConnectOwnerContext> {
+ let context = CONNECT_OWNER_CONTEXT.try_with(|context| *context).ok()?;
+ (context.coordinator_id == self.id
+ && context.token == self.active_token.load(Ordering::SeqCst))
+ .then_some(context)
+ }
+
+ pub(crate) fn owner_context(
+ &self,
+ token: ConnectOwnerToken,
+ settle_off_leader: bool,
+ single_attempt: bool,
+ ) -> ConnectOwnerContext {
+ ConnectOwnerContext {
+ coordinator_id: self.id,
+ token: token.0,
+ settle_off_leader,
+ single_attempt,
+ }
+ }
+
+ pub(crate) async fn scope_owner<Fut, T>(&self, context: ConnectOwnerContext, future: Fut) -> T
+ where
+ Fut: Future<Output = T>,
+ {
+ CONNECT_OWNER_CONTEXT.scope(context, future).await
+ }
+
+ pub(crate) async fn run<F, Fut>(&self, operation: F) -> Result<(), IggyError>
+ where
+ F: FnOnce(bool, ConnectOwnerToken) -> Fut,
+ Fut: Future<Output = Result<(), IggyError>>,
+ {
+ let observed_generation = self.generation.load(Ordering::SeqCst);
+ if self
+ .active
+ .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+ .is_ok()
+ {
+ let abandoned = self.abandoned.swap(false, Ordering::SeqCst);
+ let token = self.next_token.fetch_add(1, Ordering::SeqCst).max(1);
+ self.active_token.store(token, Ordering::SeqCst);
+ let mut owner = ConnectOwner {
+ coordinator: self,
+ token,
+ completed: false,
+ };
+ let result = operation(abandoned, ConnectOwnerToken(token)).await;
+ owner.complete(result.clone());
+ return result;
+ }
+
+ loop {
+ let changed = self.changed.notified();
+ let generation = self.generation.load(Ordering::SeqCst);
+ if generation > observed_generation {
+ return self.result_for(generation);
+ }
+ changed.await;
+ }
+ }
+
+ fn result_for(&self, generation: u64) -> Result<(), IggyError> {
+ self.result
+ .lock()
+ .expect("connect result mutex poisoned")
+ .as_ref()
+ .filter(|(completed_generation, _)| *completed_generation >= generation)
+ .map(|(_, result)| result.clone())
+ .unwrap_or(Err(IggyError::Disconnected))
+ }
+
+ fn finish(&self, token: u64, result: Result<(), IggyError>, abandoned: bool) {
+ if abandoned {
+ self.abandoned.store(true, Ordering::SeqCst);
+ }
+ let generation = self.generation.load(Ordering::SeqCst) + 1;
+ self.result
+ .lock()
+ .expect("connect result mutex poisoned")
+ .replace((generation, result));
+ self.generation.store(generation, Ordering::SeqCst);
+ if self.active_token.load(Ordering::SeqCst) == token {
+ self.active_token.store(0, Ordering::SeqCst);
+ }
+ self.active.store(false, Ordering::SeqCst);
+ self.changed.notify_waiters();
+ }
+}
+
+impl Default for ConnectCoordinator {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+struct ConnectOwner<'a> {
+ coordinator: &'a ConnectCoordinator,
+ token: u64,
+ completed: bool,
+}
+
+impl ConnectOwner<'_> {
+ fn complete(&mut self, result: Result<(), IggyError>) {
+ self.coordinator.finish(self.token, result, false);
+ self.completed = true;
+ }
+}
+
+impl Drop for ConnectOwner<'_> {
+ fn drop(&mut self) {
+ if !self.completed {
+ self.coordinator
+ .finish(self.token, Err(IggyError::Disconnected), true);
+ }
+ }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct ConnectOwnerToken(u64);
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct ConnectOwnerContext {
+ coordinator_id: u64,
+ token: u64,
+ settle_off_leader: bool,
+ single_attempt: bool,
+}
+
+impl ConnectOwnerContext {
+ pub(crate) fn settle_off_leader(self) -> bool {
+ self.settle_off_leader
+ }
+
+ pub(crate) fn single_attempt(self) -> bool {
+ self.single_attempt
+ }
+}
+
+static NEXT_CONNECT_COORDINATOR_ID: AtomicU64 = AtomicU64::new(1);
+
+tokio::task_local! {
+ static CONNECT_OWNER_CONTEXT: ConnectOwnerContext;
+}
+
/// Struct to track leader redirection state
#[derive(Debug, Clone)]
pub struct LeaderRedirectionState {
@@ -378,6 +625,8 @@
#[cfg(test)]
mod tests {
use super::*;
+ use std::sync::Arc;
+ use std::sync::atomic::AtomicUsize;
#[test]
fn only_unauthenticated_cluster_metadata_is_a_pre_login_probe() {
@@ -518,6 +767,278 @@
}
#[test]
+ fn a_roster_walk_visits_each_other_endpoint_once() {
+ let roster = vec![
+ "10.0.0.1:8090".to_string(),
+ "10.0.0.2:8090".to_string(),
+ "10.0.0.3:8090".to_string(),
+ ];
+ let mut walk = RosterWalk::new("10.0.0.1:8090", &roster);
+ assert_eq!(walk.next().as_deref(), Some("10.0.0.2:8090"));
+ assert_eq!(walk.next().as_deref(), Some("10.0.0.3:8090"));
+ assert_eq!(walk.next(), None);
+
+ let mut from_last = RosterWalk::new("10.0.0.3:8090", &roster);
+ assert_eq!(from_last.next().as_deref(), Some("10.0.0.1:8090"));
+ assert_eq!(from_last.next().as_deref(), Some("10.0.0.2:8090"));
+ assert_eq!(from_last.next(), None);
+ }
+
+ #[test]
+ fn a_roster_walk_never_revisits_redirects_or_duplicate_spellings() {
+ let mut walk = RosterWalk::new(
+ "localhost:8090",
+ &[
+ "127.0.0.1:8090".to_string(),
+ "10.0.0.2:8090".to_string(),
+ "10.0.0.2:8090".to_string(),
+ "10.0.0.3:8090".to_string(),
+ ],
+ );
+ assert!(walk.record_attempt("10.0.0.3:8090"));
+ assert!(!walk.record_attempt("10.0.0.3:8090"));
+ assert_eq!(walk.next().as_deref(), Some("10.0.0.2:8090"));
+ assert_eq!(walk.next(), None);
+
+ assert_eq!(RosterWalk::new("10.0.0.1:8090", &[]).next(), None);
+ // Same endpoint under a different spelling still counts as nowhere.
+ assert_eq!(
+ RosterWalk::new("localhost:8090", &["127.0.0.1:8090".to_string()]).next(),
+ None
+ );
+ }
+
+ #[tokio::test]
+ async fn concurrent_connect_callers_share_the_owners_result() {
+ let coordinator = Arc::new(ConnectCoordinator::new());
+ let operations = Arc::new(AtomicUsize::new(0));
+ let started = Arc::new(Notify::new());
+ let release = Arc::new(Notify::new());
+
+ let owner = {
+ let coordinator = Arc::clone(&coordinator);
+ let operations = Arc::clone(&operations);
+ let started = Arc::clone(&started);
+ let release = Arc::clone(&release);
+ tokio::spawn(async move {
+ coordinator
+ .run(|abandoned, _token| async move {
+ assert!(!abandoned);
+ operations.fetch_add(1, Ordering::SeqCst);
+ started.notify_one();
+ release.notified().await;
+ Err(IggyError::InvalidCredentials)
+ })
+ .await
+ })
+ };
+ started.notified().await;
+ let waiter = {
+ let coordinator = Arc::clone(&coordinator);
+ let operations = Arc::clone(&operations);
+ tokio::spawn(async move {
+ coordinator
+ .run(|_, _token| async move {
+ operations.fetch_add(1, Ordering::SeqCst);
+ Ok(())
+ })
+ .await
+ })
+ };
+ tokio::task::yield_now().await;
+ release.notify_one();
+
+ assert!(matches!(
+ owner.await.unwrap(),
+ Err(IggyError::InvalidCredentials)
+ ));
+ assert!(matches!(
+ waiter.await.unwrap(),
+ Err(IggyError::InvalidCredentials)
+ ));
+ assert_eq!(operations.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test]
+ async fn a_cancelled_connect_releases_waiters_and_marks_cleanup_needed() {
+ let coordinator = Arc::new(ConnectCoordinator::new());
+ let started = Arc::new(Notify::new());
+ let owner = {
+ let coordinator = Arc::clone(&coordinator);
+ let started = Arc::clone(&started);
+ tokio::spawn(async move {
+ coordinator
+ .run(|_, _token| async move {
+ started.notify_one();
+ std::future::pending::<Result<(), IggyError>>().await
+ })
+ .await
+ })
+ };
+ started.notified().await;
+ let waiter = {
+ let coordinator = Arc::clone(&coordinator);
+ tokio::spawn(async move { coordinator.run(|_, _token| async { Ok(()) }).await })
+ };
+ tokio::task::yield_now().await;
+ owner.abort();
+
+ assert!(matches!(
+ waiter.await.unwrap(),
+ Err(IggyError::Disconnected)
+ ));
+ let observed_abandoned = Arc::new(AtomicBool::new(false));
+ let marker = Arc::clone(&observed_abandoned);
+ coordinator
+ .run(|abandoned, _token| async move {
+ marker.store(abandoned, Ordering::SeqCst);
+ Ok(())
+ })
+ .await
+ .unwrap();
+ assert!(observed_abandoned.load(Ordering::SeqCst));
+ }
+
+ #[tokio::test]
+ async fn connect_owner_context_is_visible_only_to_the_owner_task() {
+ let coordinator = Arc::new(ConnectCoordinator::new());
+ let started = Arc::new(Notify::new());
+ let release = Arc::new(Notify::new());
+ let owner = {
+ let coordinator = Arc::clone(&coordinator);
+ let owner_coordinator = Arc::clone(&coordinator);
+ let started = Arc::clone(&started);
+ let release = Arc::clone(&release);
+ tokio::spawn(async move {
+ coordinator
+ .run(move |_, token| async move {
+ let context = owner_coordinator.owner_context(token, true, true);
+ owner_coordinator
+ .scope_owner(context, async {
+ assert_eq!(
+ owner_coordinator.current_owner_context(),
+ Some(context)
+ );
+ started.notify_one();
+ release.notified().await;
+ Ok(())
+ })
+ .await
+ })
+ .await
+ })
+ };
+ started.notified().await;
+
+ assert!(coordinator.is_active());
+ assert_eq!(coordinator.current_owner_context(), None);
+ release.notify_one();
+ owner.await.unwrap().unwrap();
+ assert_eq!(coordinator.current_owner_context(), None);
+ }
+
+ #[tokio::test]
+ async fn owner_tokens_cannot_collide_across_connect_coordinators() {
+ let first = Arc::new(ConnectCoordinator::new());
+ let second = Arc::new(ConnectCoordinator::new());
+ let second_started = Arc::new(Notify::new());
+ let second_release = Arc::new(Notify::new());
+ let second_owner = {
+ let second = Arc::clone(&second);
+ let owner_coordinator = Arc::clone(&second);
+ let second_started = Arc::clone(&second_started);
+ let second_release = Arc::clone(&second_release);
+ tokio::spawn(async move {
+ second
+ .run(move |_, token| async move {
+ let context = owner_coordinator.owner_context(token, false, false);
+ owner_coordinator
+ .scope_owner(context, async {
+ second_started.notify_one();
+ second_release.notified().await;
+ Ok(())
+ })
+ .await
+ })
+ .await
+ })
+ };
+ second_started.notified().await;
+
+ let first_owner = Arc::clone(&first);
+ let second_from_first = Arc::clone(&second);
+ first
+ .run(move |_, token| async move {
+ let context = first_owner.owner_context(token, true, true);
+ first_owner
+ .scope_owner(context, async {
+ assert_eq!(first_owner.current_owner_context(), Some(context));
+ assert_eq!(second_from_first.current_owner_context(), None);
+ Ok(())
+ })
+ .await
+ })
+ .await
+ .unwrap();
+
+ second_release.notify_one();
+ second_owner.await.unwrap().unwrap();
+ }
+
+ #[tokio::test]
+ async fn a_waiters_settlement_mode_cannot_leak_into_the_owner() {
+ let coordinator = Arc::new(ConnectCoordinator::new());
+ let started = Arc::new(Notify::new());
+ let release = Arc::new(Notify::new());
+ let waiter_ran = Arc::new(AtomicBool::new(false));
+ let owner = {
+ let coordinator = Arc::clone(&coordinator);
+ let owner_coordinator = Arc::clone(&coordinator);
+ let started = Arc::clone(&started);
+ let release = Arc::clone(&release);
+ tokio::spawn(async move {
+ coordinator
+ .run(move |_, token| async move {
+ let context = owner_coordinator.owner_context(token, false, false);
+ owner_coordinator
+ .scope_owner(context, async {
+ assert!(!context.settle_off_leader());
+ assert!(!context.single_attempt());
+ started.notify_one();
+ release.notified().await;
+ Ok(())
+ })
+ .await
+ })
+ .await
+ })
+ };
+ started.notified().await;
+ let waiter = {
+ let coordinator = Arc::clone(&coordinator);
+ let waiter_coordinator = Arc::clone(&coordinator);
+ let waiter_ran = Arc::clone(&waiter_ran);
+ tokio::spawn(async move {
+ coordinator
+ .run(move |_, token| async move {
+ waiter_ran.store(true, Ordering::SeqCst);
+ let context = waiter_coordinator.owner_context(token, true, true);
+ assert!(context.settle_off_leader());
+ assert!(context.single_attempt());
+ Ok(())
+ })
+ .await
+ })
+ };
+ tokio::task::yield_now().await;
+ release.notify_one();
+
+ owner.await.unwrap().unwrap();
+ waiter.await.unwrap().unwrap();
+ assert!(!waiter_ran.load(Ordering::SeqCst));
+ }
+
+ #[test]
fn test_normalize_address() {
assert_eq!(normalize_address("localhost:8090"), "127.0.0.1:8090");
assert_eq!(normalize_address("LOCALHOST:8090"), "127.0.0.1:8090");
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index b1f423a..28fb652 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -16,10 +16,12 @@
// under the License.
use crate::leader_aware::{
- LeaderRedirectionState, check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
+ ConnectCoordinator, ConnectOwnerContext, LeaderRedirectionState, RosterWalk,
+ check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
};
use crate::prelude::AutoLogin;
use crate::session::ConsensusSession;
+use crate::vsr::replay_after_session_reset_is_safe;
use iggy_common::VsrSessionControl as _;
use iggy_common::{BinaryClient, BinaryTransport, Client, PersonalAccessTokenClient, UserClient};
@@ -30,7 +32,9 @@
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
use bytes::Bytes;
-use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE};
+use iggy_binary_protocol::codes::{
+ GET_CLUSTER_METADATA_CODE, LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE,
+};
use iggy_common::{
ClientState, ConnectionString, ConnectionStringUtils, Credentials, DiagnosticEvent,
QuicConnectionStringOptions, TransportProtocol, validate_server_address,
@@ -66,6 +70,12 @@
/// Bounded overall by `RESPONSE_READ_TIMEOUT`.
const NOT_READY_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+/// How long a request replays `TransientNotAccepted` on the SAME connection
+/// before it is handed back for a leader recheck or a roster walk. A node
+/// that is not the target group's primary refuses forever, so replaying on it
+/// for the whole request budget would burn the budget against a verdict.
+const TRANSIENT_FAILOVER_CHECK_INTERVAL: Duration = Duration::from_secs(2);
+
/// QUIC client for interacting with the Iggy API.
#[derive(Debug)]
pub struct QuicClient {
@@ -81,6 +91,14 @@
// `std::sync::Mutex` rationale (pure-CPU critical section).
consensus_session: Arc<StdMutex<ConsensusSession>>,
skip_auto_login_once: Mutex<bool>,
+ /// Every endpoint the cluster roster named on the last leader check, kept
+ /// as walk candidates for a request the current node keeps refusing to
+ /// admit (its replica of the target partition group is not the primary).
+ roster_endpoints: Mutex<Vec<String>>,
+ /// Serializes leader checks and roster walks after refused requests, so
+ /// concurrent QUIC streams cannot tear down each other's new connection.
+ routing_lock: Mutex<()>,
+ connect_coordinator: ConnectCoordinator,
consumer_group_state: Arc<iggy_common::ConsumerGroupClientState>,
}
@@ -129,7 +147,145 @@
}
async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
- let result = self.send_raw(code, payload.clone()).await;
+ let roster_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ let mut result = self.send_raw(code, payload.clone()).await;
+
+ // A persistent not-admitted refusal is a verdict about who leads the
+ // TARGET group, which the metadata leader check alone cannot repair:
+ // metadata and partition consensus groups elect independently. Recheck
+ // the leader once, then walk the roster, one visit per endpoint. Only
+ // recoverable with a session to re-establish, hence the auto-login
+ // gate, and a login/register replay stays on its own connection.
+ if matches!(result, Err(IggyError::TransientNotAccepted))
+ && !is_login_register_code(code)
+ && code != GET_CLUSTER_METADATA_CODE
+ && self.config.reconnection.enabled
+ && !matches!(self.config.auto_login, AutoLogin::Disabled)
+ {
+ let _routing_guard =
+ match tokio::time::timeout_at(roster_deadline, self.routing_lock.lock()).await {
+ Ok(guard) => guard,
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ };
+ let overall_deadline = roster_deadline;
+ // A concurrent refused request may have completed the movement
+ // while this request waited for the gate.
+ result = match tokio::time::timeout_at(
+ overall_deadline,
+ self.send_raw(code, payload.clone()),
+ )
+ .await
+ {
+ Ok(result) => result,
+ // The frame is on the wire with the reply unread, so the
+ // outcome is unknown: it may be admitted, replicated, and
+ // committed. `TransientNotAccepted` here would license the
+ // walk to re-issue the payload under a fresh session the
+ // server's dedup fence cannot match. `TransientNotCommitted`
+ // states the truth and also ends the hop chain.
+ Err(_) => Err(IggyError::TransientNotCommitted),
+ };
+ let mut roster_walk: Option<RosterWalk> = None;
+ // Once the walk starts it keeps walking: a leader recheck between
+ // hops would put the request straight back on the node whose
+ // partition replica refused it.
+ let mut checked_metadata_leader = false;
+ while matches!(result, Err(IggyError::TransientNotAccepted)) {
+ let current = self.current_server_address.lock().await.clone();
+ let redirected = if checked_metadata_leader {
+ false
+ } else {
+ checked_metadata_leader = true;
+ let redirected = match tokio::time::timeout_at(
+ overall_deadline,
+ self.handle_leader_redirection(),
+ )
+ .await
+ {
+ Ok(result) => matches!(result, Ok(true)),
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ };
+ let roster = self.roster_endpoints.lock().await.clone();
+ roster_walk = Some(RosterWalk::new(¤t, &roster));
+ redirected
+ };
+ let (mut target, mut needs_settle) = if redirected {
+ let target = self.current_server_address.lock().await.clone();
+ if let Some(walk) = roster_walk.as_mut() {
+ walk.record_attempt(&target);
+ }
+ (target, false)
+ } else if let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) {
+ (next, true)
+ } else {
+ break;
+ };
+
+ loop {
+ if tokio::time::Instant::now() >= overall_deadline {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ let settled = if needs_settle {
+ match tokio::time::timeout_at(
+ overall_deadline,
+ self.settle_on_endpoint(target.clone()),
+ )
+ .await
+ {
+ Ok(Ok(())) => true,
+ Ok(Err(IggyError::CannotEstablishConnection)) => false,
+ Ok(Err(error)) => return Err(error),
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ }
+ } else {
+ true
+ };
+ if settled {
+ let connect_result = if needs_settle {
+ tokio::time::timeout_at(overall_deadline, self.connect_off_leader())
+ .await
+ } else {
+ tokio::time::timeout_at(overall_deadline, self.connect()).await
+ };
+ match connect_result {
+ Ok(Ok(())) => {
+ let connected = self.current_server_address.lock().await.clone();
+ let first_visit = roster_walk
+ .as_mut()
+ .is_some_and(|walk| walk.record_attempt(&connected));
+ if crate::leader_aware::is_same_spelling(&connected, &target)
+ || first_visit
+ {
+ break;
+ }
+ }
+ Ok(Err(IggyError::CannotEstablishConnection)) => {}
+ Ok(Err(error)) => return Err(error),
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ }
+ }
+
+ let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) else {
+ return Err(IggyError::TransientNotAccepted);
+ };
+ target = next;
+ needs_settle = true;
+ }
+ result = match tokio::time::timeout_at(
+ overall_deadline,
+ self.send_raw(code, payload.clone()),
+ )
+ .await
+ {
+ Ok(result) => result,
+ // On the wire, reply unread: unknown outcome. See the
+ // matching arm above; a fabricated not-admitted would
+ // re-issue a possibly committed write on the next hop.
+ Err(_) => Err(IggyError::TransientNotCommitted),
+ };
+ }
+ }
+
if result.is_ok() {
return result;
}
@@ -152,6 +308,10 @@
return Err(error);
}
+ if code == GET_CLUSTER_METADATA_CODE {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -166,8 +326,25 @@
return Err(error);
}
- self.disconnect().await?;
+ let replay_after_reconnect = replay_after_session_reset_is_safe(code, &error);
let skip_auto_login = is_login_register_code(code);
+ let owner_context = skip_auto_login
+ .then(|| self.connect_coordinator.current_owner_context())
+ .flatten();
+ let nested_connect = owner_context.is_some();
+ let _routing_guard = if nested_connect {
+ None
+ } else {
+ Some(self.routing_lock.lock().await)
+ };
+ if !nested_connect && self.connect_coordinator.is_active() {
+ self.connect().await?;
+ if !replay_after_reconnect {
+ return Err(error);
+ }
+ return self.send_raw(code, payload).await;
+ }
+ self.disconnect().await?;
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
}
@@ -176,11 +353,23 @@
"Reconnecting to the server: {}, by client: {}",
server_address, self.config.client_address
);
- let reconnect = self.connect().await;
+ let reconnect = if nested_connect {
+ self.connect_inner(owner_context.expect("owner context checked above"))
+ .await
+ } else {
+ self.connect().await
+ };
if skip_auto_login && reconnect.is_err() {
*self.skip_auto_login_once.lock().await = false;
}
reconnect?;
+ if !replay_after_reconnect {
+ warn!(
+ "Reconnected, but command: {code} may have committed before its reply was lost; \
+ replaying it under the new session could apply it twice."
+ );
+ return Err(error);
+ }
self.send_raw(code, payload).await
}
@@ -299,6 +488,9 @@
current_server_address: Mutex::new(server_address),
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
skip_auto_login_once: Mutex::new(false),
+ roster_endpoints: Mutex::new(Vec::new()),
+ routing_lock: Mutex::new(()),
+ connect_coordinator: ConnectCoordinator::new(),
consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
@@ -337,6 +529,36 @@
}
async fn connect(&self) -> Result<(), IggyError> {
+ self.connect_with_settlement(false).await
+ }
+
+ pub(crate) async fn connect_off_leader(&self) -> Result<(), IggyError> {
+ self.connect_with_settlement(true).await
+ }
+
+ async fn connect_with_settlement(&self, settle_off_leader: bool) -> Result<(), IggyError> {
+ self.connect_coordinator
+ .run(|abandoned, token| async move {
+ let context = self.connect_coordinator.owner_context(
+ token,
+ settle_off_leader,
+ settle_off_leader,
+ );
+ self.connect_coordinator
+ .scope_owner(context, async move {
+ if abandoned {
+ self.clear_abandoned_connect().await?;
+ }
+ self.connect_inner(context).await
+ })
+ .await
+ })
+ .await
+ }
+
+ async fn connect_inner(&self, context: ConnectOwnerContext) -> Result<(), IggyError> {
+ let settle_off_leader = context.settle_off_leader();
+ let single_attempt = context.single_attempt();
loop {
match self.get_state().await {
ClientState::Shutdown => {
@@ -357,7 +579,7 @@
}
self.set_state(ClientState::Connecting).await;
- if let Some(connected_at) = self.connected_at.lock().await.as_ref() {
+ if !single_attempt && let Some(connected_at) = self.connected_at.lock().await.as_ref() {
let now = IggyTimestamp::now();
let elapsed = now.as_micros() - connected_at.as_micros();
let interval = self.config.reconnection.reestablish_after.as_micros();
@@ -395,14 +617,26 @@
"{NAME} client is connecting to server: {}...",
server_address
);
- let connection_result = self
+ let connection_result = match self
.endpoint
.connect(server_address, &self.config.server_name)
- .unwrap()
- .await;
+ {
+ Ok(connecting) => connecting.await,
+ Err(error) => {
+ error!("Failed to start QUIC connection: {error}");
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::CannotEstablishConnection);
+ }
+ };
if connection_result.is_err() {
error!("Failed to connect to server: {}", server_address);
+ if single_attempt {
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::CannotEstablishConnection);
+ }
if !self.config.reconnection.enabled {
warn!("Automatic reconnection is disabled.");
return Err(IggyError::CannotEstablishConnection);
@@ -494,12 +728,23 @@
}
}
- // The sole leader settlement, and it runs
- // authenticated. Any node completes a login now -- a
- // backup forwards the register to the primary -- so
- // this decides where later ops land, not whether
- // sign-in works.
- self.handle_leader_redirection().await?
+ // A roster walk stays on the endpoint it dialed: the
+ // leader settlement below would put the connection
+ // straight back on the node whose partition replica
+ // keeps refusing the request. One connect only.
+ if settle_off_leader {
+ info!(
+ "{NAME} client stays on the dialed node for a partition failover."
+ );
+ false
+ } else {
+ // The sole leader settlement, and it runs
+ // authenticated. Any node completes a login now --
+ // a backup forwards the register to the primary --
+ // so this decides where later ops land, not
+ // whether sign-in works.
+ self.handle_leader_redirection().await?
+ }
}
}
};
@@ -512,19 +757,35 @@
}
}
+ async fn clear_abandoned_connect(&self) -> Result<(), IggyError> {
+ if let Some(connection) = self.connection.lock().await.take() {
+ connection.close(0u32.into(), b"");
+ }
+ self.endpoint.wait_idle().await;
+ self.reset_vsr_session().await?;
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ Ok(())
+ }
+
/// Checks cluster metadata and handles leader redirection if needed.
/// Returns true if redirection occurred and reconnection is needed.
pub(crate) async fn handle_leader_redirection(&self) -> Result<bool, IggyError> {
let current_address = self.current_server_address.lock().await.clone();
- // The roster's other endpoints are dropped here: only the TCP client
- // dials failover candidates so far.
- let leader_address = check_and_redirect_to_leader(
+ let leader_check = check_and_redirect_to_leader(
self,
¤t_address,
iggy_common::TransportProtocol::Quic,
)
- .await?
- .redirect;
+ .await?;
+ // Replaced wholesale rather than merged: the roster is the cluster's
+ // own answer about where its nodes are. Kept for the roster walk a
+ // persistently refused request runs, not for dead-node redial (which
+ // remains TCP-only).
+ if !leader_check.endpoints.is_empty() {
+ *self.roster_endpoints.lock().await = leader_check.endpoints;
+ }
+ let leader_address = leader_check.redirect;
if let Some(new_leader_address) = leader_address {
let mut redirection_state = self.leader_redirection_state.lock().await;
@@ -552,6 +813,24 @@
}
}
+ /// Move the connection to the roster endpoint after the current one, for
+ /// a request the current node keeps refusing to admit. See the TCP twin:
+ /// metadata and partition consensus groups elect independently, so the
+ /// metadata leader can hold a follower replica of the target partition,
+ /// and only walking the roster reaches that group's primary.
+ async fn settle_on_endpoint(&self, next: String) -> Result<(), IggyError> {
+ let current = self.current_server_address.lock().await.clone();
+
+ info!(
+ "The request keeps being refused on {current} while the roster names it the \
+ metadata leader; trying the next cluster node at {next}."
+ );
+ self.connected_at.lock().await.take();
+ self.disconnect().await?;
+ *self.current_server_address.lock().await = next;
+ Ok(())
+ }
+
async fn shutdown(&self) -> Result<(), IggyError> {
if self.get_state().await == ClientState::Shutdown {
return Ok(());
@@ -649,6 +928,16 @@
// reconnect path in `send_raw_with_response`, same as TCP.
let header_bytes = bytemuck::bytes_of(&request_header);
let deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ // `TransientNotAccepted` gets a short same-connection window
+ // only: past it the refusal is a verdict about who leads, not
+ // load, and the caller runs a leader recheck or roster walk.
+ // Login/register keeps the full budget on this connection: the
+ // connect flow owns its leader settlement.
+ let not_accepted_deadline = if is_login_register_code(code) {
+ deadline
+ } else {
+ deadline.min(tokio::time::Instant::now() + TRANSIENT_FAILOVER_CHECK_INTERVAL)
+ };
loop {
let (mut send, mut recv) = connection.open_bi().await.map_err(|error| {
error!("Failed to open a bidirectional stream: {error}");
@@ -682,14 +971,22 @@
{
Ok(reply) => return Ok(reply),
// `TransientNotCommitted` = the server replied with an
- // explicit retry frame because it could not commit yet
- // (not-caught-up / in-flight / pipeline-full /
- // view-change cancel). Nothing committed, so replaying
- // the same request id on a fresh bidi is safe; the
- // session stays intact so no reconnect/relogin is
- // needed (login replays idempotently too). Anything
- // else - including a silent read timeout - is
+ // explicit retry frame with an outcome that may still
+ // be resolving (not-caught-up / in-flight /
+ // pipeline-full / view-change cancel). Replaying the
+ // same request id on the same session is safe because
+ // metadata dedup returns the committed reply if needed.
+ // Anything else, including a silent read timeout, is
// terminal here and handled by the caller.
+ Err(IggyError::TransientNotAccepted)
+ if tokio::time::Instant::now() >= not_accepted_deadline =>
+ {
+ // Never admitted, so re-issuable anywhere: hand it
+ // back for a leader recheck or a roster walk
+ // instead of replaying into the same refusal for
+ // the whole request budget.
+ return Err(IggyError::TransientNotAccepted);
+ }
Err(IggyError::TransientNotCommitted | IggyError::TransientNotAccepted)
if tokio::time::Instant::now() < deadline =>
{
@@ -793,6 +1090,25 @@
use super::*;
#[tokio::test]
+ async fn a_roster_hop_does_not_enter_the_reconnect_ladder() {
+ let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("reserve UDP address");
+ let server_address = socket.local_addr().unwrap().to_string();
+ drop(socket);
+ let client = QuicClient::create(Arc::new(QuicClientConfig {
+ server_address,
+ max_idle_timeout: 100,
+ ..QuicClientConfig::default()
+ }))
+ .expect("create QUIC client");
+
+ let result = tokio::time::timeout(Duration::from_secs(10), client.connect_off_leader())
+ .await
+ .expect("one QUIC dial must not enter unlimited reconnect");
+ assert!(matches!(result, Err(IggyError::CannotEstablishConnection)));
+ assert_eq!(client.get_state().await, ClientState::Disconnected);
+ }
+
+ #[tokio::test]
async fn should_fail_with_a_zero_heartbeat_interval() {
let value = "iggy+quic://user:secret@127.0.0.1:1234?heartbeat_interval=none";
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index a271cef..23f70f7 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -16,8 +16,9 @@
// under the License.
use crate::leader_aware::{
- LeaderRedirectionState, check_and_redirect_to_leader, is_same_spelling,
- is_unauthenticated_metadata_probe, read_transport_endpoints,
+ ConnectCoordinator, ConnectOwnerContext, LeaderRedirectionState, RosterWalk,
+ check_and_redirect_to_leader, is_same_spelling, is_unauthenticated_metadata_probe,
+ read_transport_endpoints,
};
use crate::prelude::Client;
use crate::prelude::TcpClientConfig;
@@ -25,12 +26,13 @@
use crate::tcp::tcp_connection_stream::TcpConnectionStream;
use crate::tcp::tcp_connection_stream_kind::ConnectionStreamKind;
use crate::tcp::tcp_tls_connection_stream::TcpTlsConnectionStream;
-use crate::vsr::operation_for_code;
+use crate::vsr::replay_after_session_reset_is_safe;
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
use bytes::{Bytes, BytesMut};
-use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE};
-use iggy_binary_protocol::consensus::Operation;
+use iggy_binary_protocol::codes::{
+ GET_CLUSTER_METADATA_CODE, LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE,
+};
#[cfg(test)]
use iggy_common::TcpClientReconnectionConfig;
use iggy_common::VsrSessionControl as _;
@@ -124,6 +126,11 @@
// contention with zero correctness benefit.
consensus_session: Arc<StdMutex<ConsensusSession>>,
skip_auto_login_once: Mutex<bool>,
+ /// Serializes connection movement after a refused request. The stream is
+ /// lockstep, but the refusal releases it before the leader check and
+ /// reconnect, where another request could otherwise run a competing walk.
+ routing_lock: Mutex<()>,
+ connect_coordinator: ConnectCoordinator,
consumer_group_state: Arc<iggy_common::ConsumerGroupClientState>,
}
@@ -183,7 +190,6 @@
}
#[async_trait]
-#[async_trait]
impl BinaryTransport for TcpClient {
async fn get_state(&self) -> ClientState {
*self.state.lock().await
@@ -223,6 +229,10 @@
return Err(error);
}
+ if code == GET_CLUSTER_METADATA_CODE {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -248,11 +258,28 @@
// Login and register are the exception: the server stays deliberately
// silent on a transient register failure and relies on the client
// replaying, so that replay is the protocol rather than a retry.
- let replay_after_reconnect = replay_is_safe(code, &error);
-
- self.disconnect_transport().await?;
+ let replay_after_reconnect = replay_after_session_reset_is_safe(code, &error);
let skip_auto_login = is_login_register_code(code);
+ let owner_context = skip_auto_login
+ .then(|| self.connect_coordinator.current_owner_context())
+ .flatten();
+ let nested_connect = owner_context.is_some();
+ let _routing_guard = if nested_connect {
+ None
+ } else {
+ Some(self.routing_lock.lock().await)
+ };
+ if !nested_connect && self.connect_coordinator.is_active() {
+ self.connect().await?;
+ if !replay_after_reconnect {
+ return Err(error);
+ }
+ drop(_routing_guard);
+ return self.send_raw(code, payload).await;
+ }
+ self.disconnect_transport().await?;
+
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
}
@@ -266,7 +293,12 @@
);
}
- let reconnect = self.connect().await;
+ let reconnect = if nested_connect {
+ self.connect_inner(owner_context.expect("owner context checked above"))
+ .await
+ } else {
+ self.connect().await
+ };
if skip_auto_login && reconnect.is_err() {
*self.skip_auto_login_once.lock().await = false;
}
@@ -281,6 +313,7 @@
return Err(error);
}
+ drop(_routing_guard);
self.send_raw(code, payload).await
}
@@ -293,38 +326,6 @@
}
}
-/// Whether replaying `code` over a fresh connection cannot apply it twice.
-///
-/// The reconnect registers a new client identity, so the server's dedup fence
-/// no longer covers the original request: only requests that provably never
-/// reached the log may be re-sent.
-///
-/// - the errors raised before the frame was written, and the server's own
-/// refusals, which precede execution. A `StaleClient` eviction is neither:
-/// it arrives out of band and is consumed in place of the pending reply, so
-/// the request it interrupted may already have committed;
-/// - operations that never enter the log: a non-replicated read, and a logout,
-/// which ends whatever session the connection carried -- the reconnect
-/// brought a new one, and refusing the replay would strand
-/// `logout_before_relogin`, whose failure aborts the sign-in that was about
-/// to replace the session;
-/// - login and register, where the replay is the protocol: the server stays
-/// deliberately silent on a transient register failure and relies on the
-/// client resending.
-fn replay_is_safe(code: u32, error: &IggyError) -> bool {
- is_login_register_code(code)
- || matches!(
- error,
- IggyError::NotConnected
- | IggyError::CannotEstablishConnection
- | IggyError::Unauthenticated
- )
- || matches!(
- operation_for_code(code),
- Operation::NonReplicated | Operation::Logout
- )
-}
-
impl iggy_common::VsrSessionSealed for TcpClient {}
#[async_trait::async_trait]
@@ -498,11 +499,40 @@
session_credentials: Mutex::new(None),
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
skip_auto_login_once: Mutex::new(false),
+ routing_lock: Mutex::new(()),
+ connect_coordinator: ConnectCoordinator::new(),
consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
async fn connect(&self) -> Result<(), IggyError> {
+ self.connect_with_settlement(false).await
+ }
+
+ async fn connect_off_leader(&self) -> Result<(), IggyError> {
+ self.connect_with_settlement(true).await
+ }
+
+ async fn connect_with_settlement(&self, settle_off_leader: bool) -> Result<(), IggyError> {
+ self.connect_coordinator
+ .run(|abandoned, token| async move {
+ let context =
+ self.connect_coordinator
+ .owner_context(token, settle_off_leader, false);
+ self.connect_coordinator
+ .scope_owner(context, async move {
+ if abandoned {
+ self.clear_abandoned_connect().await?;
+ }
+ self.connect_inner(context).await
+ })
+ .await
+ })
+ .await
+ }
+
+ async fn connect_inner(&self, context: ConnectOwnerContext) -> Result<(), IggyError> {
+ let settle_off_leader = context.settle_off_leader();
loop {
// Read and claimed under one lock acquisition. Apart, two callers
// both find `Disconnected` and both sweep: the loser's
@@ -548,7 +578,6 @@
let mut guard = self.skip_auto_login_once.lock().await;
std::mem::take(&mut *guard)
};
-
let mut retry_count = 0;
let mut candidate = 0;
// A fault no retry can fix, remembered rather than returned at
@@ -595,7 +624,7 @@
self.publish_event(DiagnosticEvent::Connected).await;
match self
- .establish_session(client_address, skip_auto_login)
+ .establish_session(client_address, skip_auto_login, settle_off_leader)
.await
{
Ok(should_redirect) => break should_redirect,
@@ -684,12 +713,21 @@
}
}
+ async fn clear_abandoned_connect(&self) -> Result<(), IggyError> {
+ self.stream.lock().await.take();
+ self.reset_vsr_session().await?;
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ Ok(())
+ }
+
/// Re-establish the session on a connection that just came up and settle it
/// on the leader. Reports whether the leader check asks for a redirect.
async fn establish_session(
&self,
client_address: SocketAddr,
skip_auto_login: bool,
+ settle_off_leader: bool,
) -> Result<bool, SignInFailure> {
let Some(credentials) = self.sign_in_credentials().await else {
info!("No credentials to sign in with.");
@@ -722,6 +760,18 @@
Err(error) => return Err(self.fail_sign_in(error).await),
}
+ // A failover walking the roster past the metadata leader stays where
+ // it dialed: the leader settlement below would put the connection
+ // straight back on the node whose partition replica keeps refusing
+ // the request. One connect only; the next ordinary connect settles
+ // normally.
+ if settle_off_leader {
+ info!(
+ "{NAME} client: {client_address} stays on the dialed node for a partition failover."
+ );
+ return Ok(false);
+ }
+
// The sole leader settlement, and it runs authenticated. Any node
// completes a login now -- a backup forwards the register to the
// primary -- so this decides where later ops land, not whether sign-in
@@ -828,6 +878,31 @@
}
}
+ /// Move the connection to the roster endpoint after the current one, for
+ /// a request the current node keeps refusing to admit.
+ ///
+ /// The metadata leader check cannot repair that refusal: metadata and
+ /// partition consensus groups elect independently, so the metadata leader
+ /// can hold a follower replica of the partition the request targets.
+ /// `TransientNotAccepted` marks the request as never admitted and safe to
+ /// re-issue anywhere, so walking the roster is correct, and the caller's
+ /// request budget bounds the walk. Reports whether there was another
+ /// endpoint to move to.
+ async fn settle_on_endpoint(&self, next: String) -> Result<(), IggyError> {
+ let current = self.current_server_address.lock().await.clone();
+
+ info!(
+ "The request keeps being refused on {current} while the roster names it the \
+ metadata leader; trying the next cluster node at {next}."
+ );
+ // No reestablish pacing: the node being left is healthy, the one being
+ // dialed owes no cooldown, and the request is already burning budget.
+ self.connected_at.lock().await.take();
+ self.disconnect_transport().await?;
+ *self.current_server_address.lock().await = next;
+ Ok(())
+ }
+
/// Whether an `AutoLogin` is configured on this client, which makes the
/// session after any connect the configured user's rather than whoever
/// signed in by hand.
@@ -1012,12 +1087,7 @@
let connector = TlsConnector::from(Arc::new(config));
let tls_domain = if self.config.tls_domain.is_empty() {
- // Extract hostname/IP from server_address when tls_domain is not specified
- server_address
- .split(':')
- .next()
- .unwrap_or(server_address)
- .to_string()
+ tls_server_name(server_address)
} else {
self.config.tls_domain.to_owned()
};
@@ -1172,7 +1242,12 @@
// sign-in handshake, and reconnecting from underneath it would
// recurse.
let overall_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
- let mut preencoded = None;
+ // Set once this request starts walking the roster past the metadata
+ // leader, so later rounds keep walking instead of being redirected
+ // back onto the node whose partition replica keeps refusing them.
+ let mut roster_walk: Option<RosterWalk> = None;
+ let mut checked_metadata_leader = false;
+ let mut routing_guard = None;
loop {
let transient_deadline = if is_login_register_code(code) {
overall_deadline
@@ -1180,11 +1255,11 @@
overall_deadline
.min(tokio::time::Instant::now() + TRANSIENT_FAILOVER_CHECK_INTERVAL)
};
- let (header, result) = self
+ let (_header, result) = self
.send_raw_vsr_attempt(
code,
payload.clone(),
- preencoded,
+ None,
transient_deadline,
overall_deadline,
)
@@ -1194,6 +1269,17 @@
if tokio::time::Instant::now() < overall_deadline
&& !is_login_register_code(code) =>
{
+ if code == GET_CLUSTER_METADATA_CODE {
+ return Err(IggyError::TransientNotAccepted);
+ }
+
+ if routing_guard.is_none() {
+ routing_guard = Some(self.routing_lock.lock().await);
+ // A concurrent refused request may have moved the
+ // shared client while this request waited.
+ continue;
+ }
+
// The server explicitly did NOT admit the request, so
// re-issuing it -- same id on this session, or a fresh
// id under a new session after a failover -- cannot
@@ -1204,10 +1290,64 @@
// its outcome is unknown, so the attempt loop replays
// it same-session for the whole budget and then the
// error propagates to the caller.)
- preencoded = header;
- if let Ok(true) = self.handle_leader_redirection().await {
- self.connect().await?;
- preencoded = None;
+ let current = self.current_server_address.lock().await.clone();
+ let mut redirected = false;
+ if !checked_metadata_leader {
+ checked_metadata_leader = true;
+ redirected = matches!(self.handle_leader_redirection().await, Ok(true));
+ let roster = self.roster_endpoints.lock().await.clone();
+ roster_walk = Some(RosterWalk::new(¤t, &roster));
+ }
+ let (mut target, mut needs_settle) = if redirected {
+ let target = self.current_server_address.lock().await.clone();
+ if let Some(walk) = roster_walk.as_mut() {
+ walk.record_attempt(&target);
+ }
+ (target, false)
+ } else if let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) {
+ // The roster says this node IS the metadata leader
+ // (or answered nothing usable), yet it keeps refusing
+ // to admit the request: its replica of the target
+ // partition group is not that group's primary, and no
+ // metadata redirect can fix that. Walk the roster
+ // instead of replaying into the same refusal until
+ // the whole request budget burns. Once walking, keep
+ // walking: rechecking the metadata leader between
+ // hops would bounce the request between two nodes and
+ // never reach the rest of the roster.
+ (next, true)
+ } else {
+ return Err(IggyError::TransientNotAccepted);
+ };
+
+ loop {
+ if needs_settle {
+ self.settle_on_endpoint(target.clone()).await?;
+ }
+ let connect = if needs_settle {
+ self.connect_off_leader().await
+ } else {
+ self.connect().await
+ };
+ match connect {
+ Ok(()) => {
+ let connected = self.current_server_address.lock().await.clone();
+ let first_visit = roster_walk
+ .as_mut()
+ .is_some_and(|walk| walk.record_attempt(&connected));
+ if is_same_spelling(&connected, &target) || first_visit {
+ break;
+ }
+ }
+ Err(IggyError::CannotEstablishConnection) => {}
+ Err(error) => return Err(error),
+ }
+
+ let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) else {
+ return Err(IggyError::TransientNotAccepted);
+ };
+ target = next;
+ needs_settle = true;
}
}
Err(IggyError::Disconnected) => {
@@ -1405,6 +1545,16 @@
matches!(code, LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE)
}
+fn tls_server_name(server_address: &str) -> String {
+ if let Ok(address) = SocketAddr::from_str(server_address) {
+ return address.ip().to_string();
+ }
+ server_address
+ .rsplit_once(':')
+ .map_or(server_address, |(host, _port)| host)
+ .to_owned()
+}
+
/// Unit tests for TcpClient.
/// Currently only tests for "from_connection_string()" are implemented.
/// TODO: Add complete unit tests for TcpClient.
@@ -1417,6 +1567,13 @@
const SESSION_USER_ID: u32 = 7;
+ #[test]
+ fn tls_server_names_support_dns_ipv4_and_ipv6_endpoints() {
+ assert_eq!(tls_server_name("iggy-1:8090"), "iggy-1");
+ assert_eq!(tls_server_name("127.0.0.1:8090"), "127.0.0.1");
+ assert_eq!(tls_server_name("[fd00::1]:8090"), "fd00::1");
+ }
+
fn client_with(server_address: &str) -> TcpClient {
TcpClient::create(Arc::new(TcpClientConfig {
server_address: server_address.to_string(),
@@ -1469,6 +1626,82 @@
counted_endpoint_that_hangs_up().await.0
}
+ #[tokio::test]
+ async fn concurrent_connect_waits_for_the_owners_result() {
+ let (_listener, silent) = live_endpoint().await;
+ let client = Arc::new(
+ TcpClient::create(Arc::new(TcpClientConfig {
+ server_address: silent,
+ tls_enabled: true,
+ tls_validate_certificate: false,
+ reconnection: TcpClientReconnectionConfig {
+ enabled: false,
+ ..TcpClientReconnectionConfig::default()
+ },
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client"),
+ );
+ let owner = {
+ let client = Arc::clone(&client);
+ tokio::spawn(async move { TcpClient::connect(&client).await })
+ };
+ while client.get_state().await != ClientState::Connecting {
+ tokio::task::yield_now().await;
+ }
+ let mut waiter = {
+ let client = Arc::clone(&client);
+ tokio::spawn(async move { TcpClient::connect(&client).await })
+ };
+
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(100), &mut waiter)
+ .await
+ .is_err(),
+ "a concurrent caller reported success while the owner was still connecting"
+ );
+ owner.abort();
+ assert!(matches!(
+ waiter.await.unwrap(),
+ Err(IggyError::Disconnected)
+ ));
+ }
+
+ #[tokio::test]
+ async fn an_unrelated_public_login_cannot_take_over_a_connect_owner() {
+ let (_listener, silent) = live_endpoint().await;
+ let client = Arc::new(
+ TcpClient::create(Arc::new(TcpClientConfig {
+ server_address: silent,
+ tls_enabled: true,
+ tls_validate_certificate: false,
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client"),
+ );
+ let owner = {
+ let client = Arc::clone(&client);
+ tokio::spawn(async move { TcpClient::connect(&client).await })
+ };
+ while client.get_state().await != ClientState::Connecting {
+ tokio::task::yield_now().await;
+ }
+ let mut login = {
+ let client = Arc::clone(&client);
+ tokio::spawn(async move { client.login_user("iggy", "iggy").await })
+ };
+
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(100), &mut login)
+ .await
+ .is_err(),
+ "an unrelated login bypassed the connect owner instead of waiting"
+ );
+ assert_eq!(client.get_state().await, ClientState::Connecting);
+ owner.abort();
+ assert!(login.await.unwrap().is_err());
+ }
+
// With reconnection off there are no retries, but the endpoints the roster
// named are still there to be tried and each gets its one turn.
#[tokio::test]
@@ -1742,8 +1975,11 @@
#[test]
fn only_requests_that_cannot_double_apply_are_replayed() {
// Never written, or refused before execution.
- assert!(replay_is_safe(SEND_MESSAGES_CODE, &IggyError::NotConnected));
- assert!(replay_is_safe(
+ assert!(replay_after_session_reset_is_safe(
+ SEND_MESSAGES_CODE,
+ &IggyError::NotConnected
+ ));
+ assert!(replay_after_session_reset_is_safe(
SEND_MESSAGES_CODE,
&IggyError::CannotEstablishConnection
));
@@ -1751,22 +1987,31 @@
// re-sent under a session the fence cannot match it against. An
// eviction is consumed in place of the reply, so it says nothing about
// whether the write committed.
- assert!(!replay_is_safe(SEND_MESSAGES_CODE, &IggyError::StaleClient));
- assert!(!replay_is_safe(
+ assert!(!replay_after_session_reset_is_safe(
+ SEND_MESSAGES_CODE,
+ &IggyError::StaleClient
+ ));
+ assert!(!replay_after_session_reset_is_safe(
SEND_MESSAGES_CODE,
&IggyError::Disconnected
));
- assert!(!replay_is_safe(
+ assert!(!replay_after_session_reset_is_safe(
SEND_MESSAGES_CODE,
&IggyError::EmptyResponse
));
// A read never enters the log, and a logout ends a session the
// reconnect already replaced -- `logout_before_relogin` depends on it.
- assert!(replay_is_safe(GET_ME_CODE, &IggyError::Disconnected));
- assert!(replay_is_safe(LOGOUT_USER_CODE, &IggyError::Disconnected));
+ assert!(replay_after_session_reset_is_safe(
+ GET_ME_CODE,
+ &IggyError::Disconnected
+ ));
+ assert!(replay_after_session_reset_is_safe(
+ LOGOUT_USER_CODE,
+ &IggyError::Disconnected
+ ));
// The register replay is the protocol: the server stays silent on a
// transient failure and waits for the resend.
- assert!(replay_is_safe(
+ assert!(replay_after_session_reset_is_safe(
LOGIN_REGISTER_CODE,
&IggyError::Disconnected
));
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 9625b02..2e8a210 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -28,20 +28,19 @@
const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4;
-// TODO(vsr): transparent retry-after-disconnect can weaken at-most-once semantics
-// for replicated writes. The transports currently disconnect, reconnect, and encode
-// the retry from the current ConsensusSession. If disconnect created a fresh VSR
-// client/session, the retried request gets a new (client_id, request_id) tuple, so
-// server-side deduplication cannot match a mutation that may already have committed
-// before the transport failure.
+// A reconnect creates a fresh VSR client and session. A replicated request
+// retried there gets a new (client_id, request_id) tuple, so server-side
+// deduplication cannot match a mutation that may already have committed before
+// the transport failure. `replay_after_session_reset_is_safe` therefore keeps
+// ambiguous replicated outcomes with the caller instead of replaying them.
//
-// The fix is to keep the ConsensusSession's client_id and request counter across
-// reconnects and retry replicated writes under the same (client_id, request_id)
-// instead of re-registering fresh. Resume happens through the LOGIN path: the
+// A future transparent replay path can keep the ConsensusSession's client id
+// and request counter across reconnects and retry under the same identity.
+// Resume would happen through the LOGIN path: the
// reconnecting client re-authenticates presenting its previous client_id, the
// server verifies the authenticated user owns that entry, and the rebind commits
// a Register that adopts the entry with its watermark and reply ring intact. Note
-// the epoch changes -- the rebind moves the fence to the new register's op -- so
+// the epoch changes because the rebind moves the fence to the new register's op, so
// the session field must be taken from the new login reply, not carried over.
//
// There is deliberately no credential-free rebind: presenting (client, session)
@@ -167,6 +166,22 @@
Operation::from_command_code(code).unwrap_or(Operation::NonReplicated)
}
+/// Whether replaying `code` after reconnecting with a new session cannot
+/// apply it twice.
+pub(crate) fn replay_after_session_reset_is_safe(code: u32, error: &IggyError) -> bool {
+ matches!(code, LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE)
+ || matches!(
+ error,
+ IggyError::NotConnected
+ | IggyError::CannotEstablishConnection
+ | IggyError::Unauthenticated
+ )
+ || matches!(
+ operation_for_code(code),
+ Operation::NonReplicated | Operation::Logout
+ )
+}
+
pub(crate) fn response_size(header: &[u8]) -> Result<usize, IggyError> {
let size = read_size_field(header).ok_or(IggyError::InvalidCommand)? as usize;
if size < HEADER_SIZE {
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index 56261f6..f40bc91 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -16,9 +16,11 @@
// under the License.
use crate::leader_aware::{
- LeaderRedirectionState, check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
+ ConnectCoordinator, ConnectOwnerContext, LeaderRedirectionState, RosterWalk,
+ check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
};
use crate::session::ConsensusSession;
+use crate::vsr::replay_after_session_reset_is_safe;
use crate::websocket::websocket_connection_stream::WebSocketConnectionStream;
use crate::websocket::websocket_stream_kind::WebSocketStreamKind;
use crate::websocket::websocket_tls_connection_stream::WebSocketTlsConnectionStream;
@@ -28,7 +30,9 @@
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
use bytes::Bytes;
-use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE};
+use iggy_binary_protocol::codes::{
+ GET_CLUSTER_METADATA_CODE, LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE,
+};
use iggy_common::VsrSessionControl as _;
use iggy_common::{
AutoLogin, ClientState, ConnectionString, Credentials, DiagnosticEvent, IggyDuration,
@@ -44,7 +48,7 @@
use tokio::sync::Mutex;
use tokio::time::sleep;
use tokio_tungstenite::{
- Connector, client_async_with_config, connect_async_tls_with_config,
+ Connector, client_async_tls_with_config, client_async_with_config,
tungstenite::client::IntoClientRequest,
};
use tracing::{debug, error, info, trace, warn};
@@ -63,6 +67,12 @@
/// `RESPONSE_READ_TIMEOUT`.
const NOT_READY_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
+/// How long a request replays `TransientNotAccepted` on the SAME connection
+/// before it is handed back for a leader recheck or a roster walk. A node
+/// that is not the target group's primary refuses forever, so replaying on it
+/// for the whole request budget would burn the budget against a verdict.
+const TRANSIENT_FAILOVER_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);
+
#[derive(Debug)]
pub struct WebSocketClient {
stream: Arc<Mutex<Option<WebSocketStreamKind>>>,
@@ -77,6 +87,14 @@
// `std::sync::Mutex` rationale (pure-CPU critical section).
consensus_session: Arc<StdMutex<ConsensusSession>>,
skip_auto_login_once: Mutex<bool>,
+ /// Every endpoint the cluster roster named on the last leader check, kept
+ /// as walk candidates for a request the current node keeps refusing to
+ /// admit (its replica of the target partition group is not the primary).
+ roster_endpoints: Mutex<Vec<String>>,
+ /// Serializes leader checks and roster walks after refused requests, so
+ /// concurrent callers cannot tear down each other's new connection.
+ routing_lock: Mutex<()>,
+ connect_coordinator: ConnectCoordinator,
consumer_group_state: Arc<iggy_common::ConsumerGroupClientState>,
}
@@ -122,7 +140,145 @@
}
async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
- let result = self.send_raw(code, payload.clone()).await;
+ let roster_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ let mut result = self.send_raw(code, payload.clone()).await;
+
+ // A persistent not-admitted refusal is a verdict about who leads the
+ // TARGET group, which the metadata leader check alone cannot repair:
+ // metadata and partition consensus groups elect independently. Recheck
+ // the leader once, then walk the roster, one visit per endpoint. Only
+ // recoverable with a session to re-establish, hence the auto-login
+ // gate, and a login/register replay stays on its own connection.
+ if matches!(result, Err(IggyError::TransientNotAccepted))
+ && !is_login_register_code(code)
+ && code != GET_CLUSTER_METADATA_CODE
+ && self.config.reconnection.enabled
+ && !matches!(self.config.auto_login, AutoLogin::Disabled)
+ {
+ let _routing_guard =
+ match tokio::time::timeout_at(roster_deadline, self.routing_lock.lock()).await {
+ Ok(guard) => guard,
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ };
+ let overall_deadline = roster_deadline;
+ // A concurrent refused request may have completed the movement
+ // while this request waited for the gate.
+ result = match tokio::time::timeout_at(
+ overall_deadline,
+ self.send_raw(code, payload.clone()),
+ )
+ .await
+ {
+ Ok(result) => result,
+ // The frame is on the wire with the reply unread, so the
+ // outcome is unknown: it may be admitted, replicated, and
+ // committed. `TransientNotAccepted` here would license the
+ // walk to re-issue the payload under a fresh session the
+ // server's dedup fence cannot match. `TransientNotCommitted`
+ // states the truth and also ends the hop chain.
+ Err(_) => Err(IggyError::TransientNotCommitted),
+ };
+ let mut roster_walk: Option<RosterWalk> = None;
+ // Once the walk starts it keeps walking: a leader recheck between
+ // hops would put the request straight back on the node whose
+ // partition replica refused it.
+ let mut checked_metadata_leader = false;
+ while matches!(result, Err(IggyError::TransientNotAccepted)) {
+ let current = self.current_server_address.lock().await.clone();
+ let redirected = if checked_metadata_leader {
+ false
+ } else {
+ checked_metadata_leader = true;
+ let redirected = match tokio::time::timeout_at(
+ overall_deadline,
+ self.handle_leader_redirection(),
+ )
+ .await
+ {
+ Ok(result) => matches!(result, Ok(true)),
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ };
+ let roster = self.roster_endpoints.lock().await.clone();
+ roster_walk = Some(RosterWalk::new(¤t, &roster));
+ redirected
+ };
+ let (mut target, mut needs_settle) = if redirected {
+ let target = self.current_server_address.lock().await.clone();
+ if let Some(walk) = roster_walk.as_mut() {
+ walk.record_attempt(&target);
+ }
+ (target, false)
+ } else if let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) {
+ (next, true)
+ } else {
+ break;
+ };
+
+ loop {
+ if tokio::time::Instant::now() >= overall_deadline {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ let settled = if needs_settle {
+ match tokio::time::timeout_at(
+ overall_deadline,
+ self.settle_on_endpoint(target.clone()),
+ )
+ .await
+ {
+ Ok(Ok(())) => true,
+ Ok(Err(IggyError::CannotEstablishConnection)) => false,
+ Ok(Err(error)) => return Err(error),
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ }
+ } else {
+ true
+ };
+ if settled {
+ let connect_result = if needs_settle {
+ tokio::time::timeout_at(overall_deadline, self.connect_off_leader())
+ .await
+ } else {
+ tokio::time::timeout_at(overall_deadline, self.connect()).await
+ };
+ match connect_result {
+ Ok(Ok(())) => {
+ let connected = self.current_server_address.lock().await.clone();
+ let first_visit = roster_walk
+ .as_mut()
+ .is_some_and(|walk| walk.record_attempt(&connected));
+ if crate::leader_aware::is_same_spelling(&connected, &target)
+ || first_visit
+ {
+ break;
+ }
+ }
+ Ok(Err(IggyError::CannotEstablishConnection)) => {}
+ Ok(Err(error)) => return Err(error),
+ Err(_) => return Err(IggyError::TransientNotAccepted),
+ }
+ }
+
+ let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) else {
+ return Err(IggyError::TransientNotAccepted);
+ };
+ target = next;
+ needs_settle = true;
+ }
+ result = match tokio::time::timeout_at(
+ overall_deadline,
+ self.send_raw(code, payload.clone()),
+ )
+ .await
+ {
+ Ok(result) => result,
+ // On the wire, reply unread: unknown outcome. See the
+ // matching arm above; a fabricated not-admitted would
+ // re-issue a possibly committed write on the next hop.
+ Err(_) => Err(IggyError::TransientNotCommitted),
+ };
+ }
+ }
+
if result.is_ok() {
return result;
}
@@ -148,6 +304,10 @@
return Err(error);
}
+ if code == GET_CLUSTER_METADATA_CODE {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -156,9 +316,26 @@
return Err(error);
}
+ let replay_after_reconnect = replay_after_session_reset_is_safe(code, &error);
+ let skip_auto_login = is_login_register_code(code);
+ let owner_context = skip_auto_login
+ .then(|| self.connect_coordinator.current_owner_context())
+ .flatten();
+ let nested_connect = owner_context.is_some();
+ let _routing_guard = if nested_connect {
+ None
+ } else {
+ Some(self.routing_lock.lock().await)
+ };
+ if !nested_connect && self.connect_coordinator.is_active() {
+ self.connect().await?;
+ if !replay_after_reconnect {
+ return Err(error);
+ }
+ return self.send_raw(code, payload).await;
+ }
self.disconnect().await?;
- let skip_auto_login = is_login_register_code(code);
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
}
@@ -171,11 +348,23 @@
);
}
- let reconnect = self.connect().await;
+ let reconnect = if nested_connect {
+ self.connect_inner(owner_context.expect("owner context checked above"))
+ .await
+ } else {
+ self.connect().await
+ };
if skip_auto_login && reconnect.is_err() {
*self.skip_auto_login_once.lock().await = false;
}
reconnect?;
+ if !replay_after_reconnect {
+ warn!(
+ "Reconnected, but command: {code} may have committed before its reply was lost; \
+ replaying it under the new session could apply it twice."
+ );
+ return Err(error);
+ }
self.send_raw(code, payload).await
}
@@ -246,6 +435,9 @@
current_server_address: Mutex::new(server_address),
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
skip_auto_login_once: Mutex::new(false),
+ roster_endpoints: Mutex::new(Vec::new()),
+ routing_lock: Mutex::new(()),
+ connect_coordinator: ConnectCoordinator::new(),
consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
@@ -267,6 +459,36 @@
}
async fn connect(&self) -> Result<(), IggyError> {
+ self.connect_with_settlement(false).await
+ }
+
+ pub(crate) async fn connect_off_leader(&self) -> Result<(), IggyError> {
+ self.connect_with_settlement(true).await
+ }
+
+ async fn connect_with_settlement(&self, settle_off_leader: bool) -> Result<(), IggyError> {
+ self.connect_coordinator
+ .run(|abandoned, token| async move {
+ let context = self.connect_coordinator.owner_context(
+ token,
+ settle_off_leader,
+ settle_off_leader,
+ );
+ self.connect_coordinator
+ .scope_owner(context, async move {
+ if abandoned {
+ self.clear_abandoned_connect().await?;
+ }
+ self.connect_inner(context).await
+ })
+ .await
+ })
+ .await
+ }
+
+ async fn connect_inner(&self, context: ConnectOwnerContext) -> Result<(), IggyError> {
+ let settle_off_leader = context.settle_off_leader();
+ let single_attempt = context.single_attempt();
loop {
if self.get_state().await == ClientState::Connected {
return Ok(());
@@ -318,7 +540,10 @@
})?;
let connection_stream = if self.config.tls_enabled {
- match self.connect_tls(server_addr, &mut retry_count).await {
+ match self
+ .connect_tls(server_addr, &mut retry_count, single_attempt)
+ .await
+ {
Ok(stream) => stream,
Err(IggyError::CannotEstablishConnection) => {
return Err(IggyError::CannotEstablishConnection);
@@ -326,7 +551,10 @@
Err(_) => continue, // retry
}
} else {
- match self.connect_plain(server_addr, &mut retry_count).await {
+ match self
+ .connect_plain(server_addr, &mut retry_count, single_attempt)
+ .await
+ {
Ok(stream) => stream,
Err(IggyError::CannotEstablishConnection) => {
return Err(IggyError::CannotEstablishConnection);
@@ -350,16 +578,25 @@
break;
}
- if !self.check_and_maybe_redirect().await? {
+ if !self.check_and_maybe_redirect(settle_off_leader).await? {
return Ok(());
}
}
}
+ async fn clear_abandoned_connect(&self) -> Result<(), IggyError> {
+ self.stream.lock().await.take();
+ self.reset_vsr_session().await?;
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ Ok(())
+ }
+
async fn connect_plain(
&self,
server_addr: SocketAddr,
retry_count: &mut u32,
+ single_attempt: bool,
) -> Result<WebSocketStreamKind, IggyError> {
let tcp_stream = match TcpStream::connect(&server_addr).await {
Ok(stream) => stream,
@@ -368,7 +605,9 @@
"Failed to connect to server: {}. Error: {}",
self.config.server_address, error
);
- return self.handle_connection_error(retry_count).await;
+ return self
+ .handle_connection_error(retry_count, single_attempt)
+ .await;
}
};
@@ -385,7 +624,9 @@
Ok(result) => result,
Err(error) => {
error!("WebSocket handshake failed: {}", error);
- return self.handle_connection_error(retry_count).await;
+ return self
+ .handle_connection_error(retry_count, single_attempt)
+ .await;
}
};
@@ -402,7 +643,17 @@
&self,
server_addr: SocketAddr,
retry_count: &mut u32,
+ single_attempt: bool,
) -> Result<WebSocketStreamKind, IggyError> {
+ let tcp_stream = match TcpStream::connect(server_addr).await {
+ Ok(stream) => stream,
+ Err(error) => {
+ error!("Failed to connect to server: {server_addr}. Error: {error}");
+ return self
+ .handle_connection_error(retry_count, single_attempt)
+ .await;
+ }
+ };
let tls_config = self.build_tls_config()?;
let connector = Connector::Rustls(Arc::new(tls_config));
@@ -412,14 +663,19 @@
server_addr.ip().to_string()
};
- let ws_url = format!("wss://{}:{}", domain, server_addr.port());
+ let uri_domain = if domain.contains(':') && !domain.starts_with('[') {
+ format!("[{domain}]")
+ } else {
+ domain
+ };
+ let ws_url = format!("wss://{}:{}", uri_domain, server_addr.port());
let tungstenite_config = self.config.ws_config.to_tungstenite_config();
debug!("Initiating WebSocket TLS connection to: {}", ws_url);
- let (websocket_stream, response) = match connect_async_tls_with_config(
+ let (websocket_stream, response) = match client_async_tls_with_config(
ws_url,
+ tcp_stream,
Some(tungstenite_config),
- false,
Some(connector),
)
.await
@@ -427,7 +683,9 @@
Ok(result) => result,
Err(error) => {
error!("WebSocket TLS handshake failed: {}", error);
- return self.handle_connection_error(retry_count).await;
+ return self
+ .handle_connection_error(retry_count, single_attempt)
+ .await;
}
};
@@ -486,7 +744,16 @@
Ok(config)
}
- async fn handle_connection_error<T>(&self, retry_count: &mut u32) -> Result<T, IggyError> {
+ async fn handle_connection_error<T>(
+ &self,
+ retry_count: &mut u32,
+ single_attempt: bool,
+ ) -> Result<T, IggyError> {
+ if single_attempt {
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::CannotEstablishConnection);
+ }
if !self.config.reconnection.enabled {
warn!("Automatic reconnection is disabled.");
return Err(IggyError::CannotEstablishConnection);
@@ -518,7 +785,7 @@
Err(IggyError::CannotEstablishConnection)
}
- async fn check_and_maybe_redirect(&self) -> Result<bool, IggyError> {
+ async fn check_and_maybe_redirect(&self, settle_off_leader: bool) -> Result<bool, IggyError> {
match &self.config.auto_login {
// Only `IggyClient` redirects after a manual sign-in, so a raw
// transport can stay on a backup, and nothing on the send path
@@ -527,6 +794,14 @@
AutoLogin::Disabled => Ok(false),
AutoLogin::Enabled(_) => {
self.auto_login().await?;
+ // A roster walk stays on the endpoint it dialed: the leader
+ // settlement below would put the connection straight back on
+ // the node whose partition replica keeps refusing the request.
+ // One connect only.
+ if settle_off_leader {
+ info!("{NAME} client stays on the dialed node for a partition failover.");
+ return Ok(false);
+ }
// The sole leader settlement, and it runs authenticated. Any
// node completes a login now -- a backup forwards the register
// to the primary -- so this decides where later ops land, not
@@ -540,15 +815,20 @@
/// Returns true if redirection occurred and reconnection is needed.
pub(crate) async fn handle_leader_redirection(&self) -> Result<bool, IggyError> {
let current_address = self.current_server_address.lock().await.clone();
- // The roster's other endpoints are dropped here: only the TCP client
- // dials failover candidates so far.
- let leader_address = check_and_redirect_to_leader(
+ let leader_check = check_and_redirect_to_leader(
self,
¤t_address,
iggy_common::TransportProtocol::WebSocket,
)
- .await?
- .redirect;
+ .await?;
+ // Replaced wholesale rather than merged: the roster is the cluster's
+ // own answer about where its nodes are. Kept for the roster walk a
+ // persistently refused request runs, not for dead-node redial (which
+ // remains TCP-only).
+ if !leader_check.endpoints.is_empty() {
+ *self.roster_endpoints.lock().await = leader_check.endpoints;
+ }
+ let leader_address = leader_check.redirect;
if let Some(new_leader_address) = leader_address {
let mut redirection_state = self.leader_redirection_state.lock().await;
@@ -574,6 +854,24 @@
}
}
+ /// Move the connection to the roster endpoint after the current one, for
+ /// a request the current node keeps refusing to admit. See the TCP twin:
+ /// metadata and partition consensus groups elect independently, so the
+ /// metadata leader can hold a follower replica of the target partition,
+ /// and only walking the roster reaches that group's primary.
+ async fn settle_on_endpoint(&self, next: String) -> Result<(), IggyError> {
+ let current = self.current_server_address.lock().await.clone();
+
+ info!(
+ "The request keeps being refused on {current} while the roster names it the \
+ metadata leader; trying the next cluster node at {next}."
+ );
+ self.connected_at.lock().await.take();
+ self.disconnect().await?;
+ *self.current_server_address.lock().await = next;
+ Ok(())
+ }
+
async fn auto_login(&self) -> Result<(), IggyError> {
let client_address = self.get_client_address_value().await;
let skip_auto_login = {
@@ -671,13 +969,18 @@
ClientState::Connected | ClientState::Authenticating | ClientState::Authenticated => {}
}
- let mut stream_guard = self.stream.lock().await;
- if stream_guard.is_none() {
- trace!("Cannot send data. Client is not connected.");
- return Err(IggyError::NotConnected);
- }
+ let stream = self.stream.clone();
+ let consensus_session = self.consensus_session.clone();
+ // The spawned task owns the lockstep exchange to completion. Cancelling
+ // the caller after a partial WebSocket frame or response header must
+ // not release the stream lock while leaving that connection reusable.
+ tokio::spawn(async move {
+ let mut stream_guard = stream.lock().await;
+ if stream_guard.is_none() {
+ trace!("Cannot send data. Client is not connected.");
+ return Err(IggyError::NotConnected);
+ }
- {
// Encode the request ONCE: `next_request_id` advances here, so a
// transient replay must reuse the same id for the server's dedup.
// The connection is lockstep (one request in flight per client), so a
@@ -686,8 +989,7 @@
// lets us resend the SAME request on the SAME connection with no
// reconnect and the session intact. Bounded by RESPONSE_READ_TIMEOUT.
let request = {
- let mut consensus_session = self
- .consensus_session
+ let mut consensus_session = consensus_session
.lock()
.expect("consensus session mutex poisoned");
crate::vsr::encode_contiguous_request(&mut consensus_session, code, &payload)?
@@ -698,6 +1000,16 @@
);
// One deadline bounds the whole request including transient replays.
let retry_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ // `TransientNotAccepted` gets a short same-connection window only:
+ // past it the refusal is a verdict about who leads, not load, and
+ // the caller runs a leader recheck or roster walk. Login/register
+ // keeps the full budget here: the connect flow owns its own
+ // leader settlement.
+ let not_accepted_deadline = if is_login_register_code(code) {
+ retry_deadline
+ } else {
+ retry_deadline.min(tokio::time::Instant::now() + TRANSIENT_FAILOVER_CHECK_INTERVAL)
+ };
loop {
let stream = stream_guard.as_mut().ok_or(IggyError::NotConnected)?;
stream.write(&request).await?;
@@ -705,9 +1017,8 @@
// One deadline spans both the header and body reads so a reply
// that delivers a header then stalls cannot wait up to 2x the
- // timeout. On expiry drop the stream: the read runs inline here
- // (no spawned task to cancel), so without an explicit drop a late
- // reply would desync framing for the next request.
+ // timeout. On expiry drop the stream so a late reply cannot
+ // desync framing for the next request.
let mut response_header = [0u8; iggy_binary_protocol::HEADER_SIZE];
let header_read =
tokio::time::timeout_at(retry_deadline, stream.read(&mut response_header))
@@ -741,9 +1052,19 @@
};
match crate::vsr::decode_response_split(&response_header, body) {
- // Server could not commit yet but answered with a complete
- // frame; the lockstep stream is in sync, so replay the same
- // request id on this connection after a short pause.
+ Err(IggyError::TransientNotAccepted)
+ if tokio::time::Instant::now() >= not_accepted_deadline =>
+ {
+ // Never admitted, so re-issuable anywhere: hand it
+ // back for a leader recheck or a roster walk instead
+ // of replaying into the same refusal for the whole
+ // request budget.
+ return Err(IggyError::TransientNotAccepted);
+ }
+ // The server answered with a complete transient frame. The
+ // lockstep stream is in sync, and replaying the same request
+ // id on this session preserves metadata dedup even when the
+ // original outcome is still resolving.
Err(IggyError::TransientNotCommitted | IggyError::TransientNotAccepted)
if tokio::time::Instant::now() < retry_deadline =>
{
@@ -754,7 +1075,12 @@
other => return other,
}
}
- }
+ })
+ .await
+ .map_err(|error| {
+ error!("Task execution failed during {NAME} request: {error}");
+ IggyError::WebSocketSendError
+ })?
}
}
@@ -767,6 +1093,29 @@
use super::*;
use std::str::FromStr;
+ #[tokio::test]
+ async fn a_roster_hop_does_not_enter_the_reconnect_ladder() {
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+ .await
+ .expect("reserve TCP address");
+ let server_address = listener.local_addr().unwrap().to_string();
+ drop(listener);
+ let client = WebSocketClient::create(Arc::new(WebSocketClientConfig {
+ server_address,
+ ..WebSocketClientConfig::default()
+ }))
+ .expect("create WebSocket client");
+
+ let result = tokio::time::timeout(
+ std::time::Duration::from_secs(1),
+ client.connect_off_leader(),
+ )
+ .await
+ .expect("one WebSocket dial must not enter unlimited reconnect");
+ assert!(matches!(result, Err(IggyError::CannotEstablishConnection)));
+ assert_eq!(client.get_state().await, ClientState::Disconnected);
+ }
+
#[test]
fn should_be_created_with_default_config() {
let client = WebSocketClient::default();
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index e47cc52..22d6ba2 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "server"
-version = "0.9.0-edge.5"
+version = "0.9.0-edge.6"
edition = "2024"
license = "Apache-2.0"
publish = false
diff --git a/core/server/src/http.rs b/core/server/src/http.rs
index 8f49f30..b8180f7 100644
--- a/core/server/src/http.rs
+++ b/core/server/src/http.rs
@@ -275,13 +275,7 @@
.is_some()
.then(|| state.metrics.request_counter());
let forwardable = forwardable_routes(state.clone());
- // The partition-plane routes (produce, consumer-offset writes) stay local:
- // each partition is its own consensus group whose primary can diverge from
- // the metadata primary, so forwarding them to the metadata primary would
- // livelock whenever the two disagree.
- // TODO: forward partition-plane writes to their own partition group's
- // primary (requires resolving the target partition from the request before
- // dispatch, and rewriting balanced partitioning to an explicit partition).
+ let partition_writes = partition_write_routes(state.clone());
let local = Router::new()
.route(PING_PATH, get(ping))
.route("/users/login", post(login_user))
@@ -292,15 +286,11 @@
)
.route(
"/streams/{stream_id}/topics/{topic_id}/messages",
- get(poll_messages).post(send_messages),
+ get(poll_messages),
)
.route(
"/streams/{stream_id}/topics/{topic_id}/consumer-offsets",
- get(get_consumer_offset).put(store_consumer_offset),
- )
- .route(
- "/streams/{stream_id}/topics/{topic_id}/consumer-offsets/{consumer_id}",
- delete(delete_consumer_offset),
+ get(get_consumer_offset),
)
.route("/stats", get(get_stats))
.route("/options/{scope}", get(describe_options))
@@ -314,6 +304,7 @@
};
let router = Router::new()
.merge(forwardable)
+ .merge(partition_writes)
.merge(local)
.with_state(state)
.layer(DefaultBodyLimit::max(max_request_size))
@@ -358,6 +349,26 @@
merge_web_ui(router, web_ui).with_state(())
}
+/// Acknowledged partition writes use a bounded HTTP roster fallback. This is
+/// a correctness path for the existing stateless HTTP transport. Direct
+/// partition-primary routing remains the scalable long-term design.
+fn partition_write_routes(state: HttpState) -> Router<HttpState> {
+ Router::new()
+ .route(
+ "/streams/{stream_id}/topics/{topic_id}/messages",
+ post(send_messages),
+ )
+ .route(
+ "/streams/{stream_id}/topics/{topic_id}/consumer-offsets",
+ put(store_consumer_offset),
+ )
+ .route(
+ "/streams/{stream_id}/topics/{topic_id}/consumer-offsets/{consumer_id}",
+ delete(delete_consumer_offset),
+ )
+ .route_layer(from_fn_with_state(state, forward::forward_partition_write))
+}
+
/// The control-plane route table: every write here commits through the
/// metadata consensus group, so the shared `forward_to_primary` route layer
/// (holding `state`) relays them on a follower.
diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs
index 5e4572f..8a153bf 100644
--- a/core/server/src/http/forward.rs
+++ b/core/server/src/http/forward.rs
@@ -60,7 +60,6 @@
use axum::body::{Body, to_bytes};
use axum::extract::{Request, State};
use axum::http::header::{AUTHORIZATION, CONTENT_TYPE, RETRY_AFTER};
-use axum::http::request::Parts;
use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
@@ -245,6 +244,21 @@
SendWrapper::new(forward_or_pass(state, request, next)).await
}
+/// Route-layer fallback for acknowledged partition writes over HTTP.
+///
+/// HTTP has no persistent leader-aware connection to retarget. Execute on the
+/// contacted node first, then walk every other configured HTTP node at most
+/// once when the response is the typed `TransientNotAccepted` denial. That
+/// denial proves the write never entered a partition pipeline. Every ambiguous
+/// outcome is returned without replay.
+pub(in crate::http) async fn forward_partition_write(
+ State(state): State<HttpState>,
+ request: Request,
+ next: Next,
+) -> Response {
+ SendWrapper::new(forward_partition_or_pass(state, request, next)).await
+}
+
async fn forward_or_pass(state: HttpState, request: Request, next: Next) -> Response {
if !state.forward.active || state.is_metadata_primary() {
return next.run(request).await;
@@ -283,6 +297,69 @@
forward(&state, request).await
}
+async fn forward_partition_or_pass(state: HttpState, request: Request, next: Next) -> Response {
+ if !state.forward.active || request.headers().contains_key(FORWARDED_HEADER) {
+ return next.run(request).await;
+ }
+ let bearer = match bearer_token(request.headers()) {
+ Ok(bearer) => bearer,
+ Err(error) => return CustomError::from(error).into_response(),
+ };
+ if let Err(rejection) = resolve_credential(&state, bearer).await {
+ return rejection.into_response();
+ }
+
+ let (parts, request_body) = request.into_parts();
+ let Ok(body) = to_bytes(request_body, state.forward.body_limit).await else {
+ return error_response(
+ StatusCode::PAYLOAD_TOO_LARGE,
+ "payload_too_large",
+ "request body exceeds http.max_request_size",
+ );
+ };
+ let method = parts.method.clone();
+ let request_headers = parts.headers.clone();
+ let path_and_query = parts
+ .uri
+ .path_and_query()
+ .map_or("/", |path_and_query| path_and_query.as_str())
+ .to_owned();
+ let local = next
+ .run(Request::from_parts(parts, Body::from(body.clone())))
+ .await;
+ if let AttemptOutcome::Relay(response) = classify_local_partition_reply(local).await {
+ return response;
+ }
+
+ let Some(_guard) = ForwardGuard::admit(&state.forward.in_flight) else {
+ return with_retry_after(error_response(
+ StatusCode::SERVICE_UNAVAILABLE,
+ "forward_busy",
+ "node is at its forward budget; retry with backoff",
+ ));
+ };
+ let self_id = state
+ .shard
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .map(consensus::VsrConsensus::replica);
+ let deadline = Instant::now() + FORWARD_RETRY_DEADLINE;
+ for socket in partition_http_sockets(&state.roster, self_id) {
+ if Instant::now() >= deadline {
+ break;
+ }
+ let url = format!("{}://{socket}{path_and_query}", state.forward.scheme);
+ match attempt(&state, &method, &request_headers, &body, &url, false).await {
+ AttemptOutcome::Relay(response) => return response,
+ AttemptOutcome::Retry => {}
+ }
+ }
+
+ with_retry_after(CustomError::from(IggyError::TransientNotAccepted).into_response())
+}
+
/// Buffer the request and drive forward attempts until one yields a relayable
/// outcome or the retry budget runs out.
async fn forward(state: &HttpInner, request: Request) -> Response {
@@ -309,7 +386,7 @@
None => AttemptOutcome::Retry,
Some(socket) => {
let url = format!("{}://{socket}{path_and_query}", state.forward.scheme);
- attempt(state, &parts, &body, &url).await
+ attempt(state, &parts.method, &parts.headers, &body, &url, true).await
}
};
match outcome {
@@ -342,8 +419,15 @@
/// Run one forward attempt end to end (connect, send, read the full reply)
/// under [`FORWARD_ATTEMPT_TIMEOUT`].
-async fn attempt(state: &HttpInner, parts: &Parts, body: &Bytes, url: &str) -> AttemptOutcome {
- let builder = match state.forward.client.request(parts.method.clone(), url) {
+async fn attempt(
+ state: &HttpInner,
+ method: &Method,
+ request_headers: &HeaderMap,
+ body: &Bytes,
+ url: &str,
+ retry_redirect: bool,
+) -> AttemptOutcome {
+ let builder = match state.forward.client.request(method.clone(), url) {
Ok(builder) => builder,
Err(error) => {
warn!(%error, "forward request build failed");
@@ -351,7 +435,7 @@
}
};
let request = builder
- .headers(forwarded_headers(&parts.headers))
+ .headers(forwarded_headers(request_headers))
.body(body.clone())
.build();
let attempt = async {
@@ -403,7 +487,7 @@
}
body.extend_from_slice(&chunk);
}
- classify_reply(status, relayed_headers, Bytes::from(body))
+ classify_reply(status, relayed_headers, Bytes::from(body), retry_redirect)
};
match compio::time::timeout(FORWARD_ATTEMPT_TIMEOUT, attempt).await {
// Elapsed: the request may be mid-commit on the primary. Outcome
@@ -463,8 +547,9 @@
status: StatusCode,
relayed_headers: Vec<(HeaderName, HeaderValue)>,
body: Bytes,
+ retry_redirect: bool,
) -> AttemptOutcome {
- if status == StatusCode::TEMPORARY_REDIRECT {
+ if retry_redirect && status == StatusCode::TEMPORARY_REDIRECT {
return AttemptOutcome::Retry;
}
if status == StatusCode::SERVICE_UNAVAILABLE && is_transient_not_accepted_body(&body) {
@@ -478,6 +563,23 @@
AttemptOutcome::Relay(response)
}
+/// Inspect a response produced on this node without changing any terminal
+/// response. Only the typed never-admitted denial opens the roster fallback.
+async fn classify_local_partition_reply(response: Response) -> AttemptOutcome {
+ let (parts, body) = response.into_parts();
+ let body = match to_bytes(body, RESPONSE_BODY_LIMIT).await {
+ Ok(body) => body,
+ Err(error) => {
+ warn!(%error, "local partition response body read failed; outcome unknown");
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ };
+ if parts.status == StatusCode::SERVICE_UNAVAILABLE && is_transient_not_accepted_body(&body) {
+ return AttemptOutcome::Retry;
+ }
+ AttemptOutcome::Relay(Response::from_parts(parts, Body::from(body)))
+}
+
/// True when a 503 body is the JSON `ErrorResponse` whose `id` is the
/// `TransientNotAccepted` code. Unparsable or foreign bodies are NOT
/// transient: when in doubt the reply is relayed, never retried.
@@ -499,6 +601,26 @@
primary_http_socket(&state.roster, primary_index)
}
+/// Private HTTP sockets for every other configured replica, in stable roster
+/// order. The caller tries each once. Invalid or HTTP-disabled entries are
+/// skipped because they cannot accept the forwarded request.
+fn partition_http_sockets(
+ roster: &crate::cluster_meta::ClusterRoster,
+ self_id: Option<u8>,
+) -> Vec<SocketAddr> {
+ roster
+ .nodes
+ .iter()
+ .filter(|node| Some(node.config().replica_id) != self_id)
+ .filter_map(|node| {
+ Some(SocketAddr::new(
+ node.replica_ip()?,
+ node.config().ports.http?,
+ ))
+ })
+ .collect()
+}
+
fn wants_linearizable(query: Option<&str>) -> bool {
query.is_some_and(|query| {
query
@@ -593,6 +715,38 @@
mod tests {
use super::*;
+ use configs::cluster::{ClusterNodeConfig, TransportPorts};
+
+ fn node(replica_id: u8, ip: &str, http: Option<u16>) -> ClusterNodeConfig {
+ ClusterNodeConfig {
+ name: format!("node-{replica_id}"),
+ ip: ip.to_owned(),
+ advertised_address: None,
+ advertised_addresses: Vec::new(),
+ replica_id,
+ ports: TransportPorts {
+ tcp: None,
+ quic: None,
+ http,
+ websocket: None,
+ tcp_replica: None,
+ },
+ }
+ }
+
+ fn roster(nodes: Vec<ClusterNodeConfig>) -> crate::cluster_meta::ClusterRoster {
+ crate::cluster_meta::ClusterRoster {
+ enabled: true,
+ name: "test-cluster".to_owned(),
+ nodes: nodes.into_iter().map(Into::into).collect(),
+ self_ip: "127.0.0.1".to_owned(),
+ self_ports: TransportPorts::default(),
+ metadata_view: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(
+ crate::cluster_meta::METADATA_VIEW_UNKNOWN,
+ )),
+ }
+ }
+
#[test]
fn linearizable_query_detected_only_on_exact_pair() {
assert!(wants_linearizable(Some("consistency=linearizable")));
@@ -629,4 +783,37 @@
assert_eq!(in_flight.get(), 0);
assert!(ForwardGuard::admit(&in_flight).is_some());
}
+
+ #[test]
+ fn partition_roster_walk_skips_self_and_undialable_nodes_once() {
+ let roster = roster(vec![
+ node(0, "10.0.0.1", Some(8080)),
+ node(1, "10.0.0.2", Some(8081)),
+ node(2, "not-an-ip", Some(8082)),
+ node(3, "10.0.0.4", None),
+ ]);
+
+ assert_eq!(
+ partition_http_sockets(&roster, Some(0)),
+ vec!["10.0.0.2:8081".parse().expect("valid socket")]
+ );
+ }
+
+ #[compio::test]
+ async fn partition_fallback_opens_only_for_typed_never_admitted_reply() {
+ let retry = classify_local_partition_reply(
+ CustomError::from(IggyError::TransientNotAccepted).into_response(),
+ )
+ .await;
+ assert!(matches!(retry, AttemptOutcome::Retry));
+
+ let terminal = classify_local_partition_reply(
+ CustomError::from(IggyError::TransientNotCommitted).into_response(),
+ )
+ .await;
+ let AttemptOutcome::Relay(terminal) = terminal else {
+ panic!("an ambiguous commit outcome must never be retried")
+ };
+ assert_eq!(terminal.status(), StatusCode::SERVICE_UNAVAILABLE);
+ }
}
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index 051d81c..e9114da 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -2707,7 +2707,9 @@
.expect("partition is materialised")
.repair = Some(RepairSession {
nonce: NONCE,
- to_op: 5,
+ view: 0,
+ commit_to_op: 5,
+ fetch_to_op: 5,
floor: None,
peer: 1,
first_batch_offset: None,
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 9098c7d..eadf688 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3993,6 +3993,10 @@
.handle_start_view(PlaneKind::Partitions, &header, suffix_body);
let adopted = !actions.is_empty();
if adopted {
+ // Any stream armed before this adoption belongs to the superseded
+ // view. Repair bodies carry no nonce, so drop the receiving session
+ // before reconciling or arming the new view's canonical range.
+ partition.repair = None;
// Ahead of the local dispatch, which rebuilds the pipeline out of the
// journal this rewrites. Same position as the metadata arm's twin, and
// like it, pending-less adoptions (empty StartView suffix) still sweep
@@ -4366,7 +4370,17 @@
);
return;
}
- let to_op = header.to_op.min(partition.consensus().commit_max());
+ // The frontier bounds the serve, not `commit_max` alone, mirroring the
+ // metadata twin: a rejoining backup needs the BODIES of the adopted
+ // suffix above the commit point. Its ack for those ops is withheld
+ // until the body is journaled, and the primary's retransmit is dropped
+ // by the backup gap check (adoption already advanced its sequencer to
+ // the head), so repair is the only channel that can deliver them.
+ let to_op = repair_serve_ceiling(
+ header.to_op,
+ partition.consensus().commit_max(),
+ partition.consensus().sequencer().current_sequence(),
+ );
// `None` means the journal holds NOTHING, not "nothing was evicted":
// the partition journal is memory-only and `clear_all` wipes the
// evicted ring with it, so a freshly installed or freshly restarted
@@ -4732,6 +4746,10 @@
if header.nonce != session.nonce {
return;
}
+ if !partition.consensus().is_normal() || partition.consensus().view() != session.view {
+ partition.repair = None;
+ return;
+ }
// Receiver half of the serve-side purge gate: while a committed purge
// is not yet locally applied, this replica's `recovered_durable_offset`
// still describes the PRE-purge segments, so a floor from a peer that
@@ -4841,7 +4859,7 @@
} else {
let commit_min = partition.consensus().commit_min();
let next = partition.repair.as_ref().and_then(|live| {
- (commit_min > before).then_some((live.peer, live.nonce, live.to_op))
+ (commit_min > before).then_some((live.peer, live.nonce, live.fetch_to_op))
});
let cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
@@ -6522,9 +6540,42 @@
continue;
};
let consensus_normal = partition.consensus().is_normal();
+ let consensus_view = partition.consensus().view();
let commit_min = partition.consensus().commit_min();
let cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
+ let repair_finished = partition.repair.is_some_and(|session| {
+ if !consensus_normal || consensus_view != session.view {
+ return true;
+ }
+ // Floored at the LIVE commit point, like `complete_repair`:
+ // committing past `commit_to_op` evicts exactly the suffix
+ // headers this shape would look for, and ops at or below
+ // `commit_min` are committed and applied, a monotone fact
+ // the flush cannot erase.
+ let fetch_complete = session.fetch_to_op <= session.commit_to_op
+ || partition
+ .log
+ .journal()
+ .inner
+ .repaired_window_shape(
+ session.commit_to_op.max(commit_min),
+ session.fetch_to_op,
+ )
+ .complete;
+ commit_min >= session.commit_to_op && fetch_complete
+ });
+ if repair_finished {
+ partition.repair = None;
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ commit_min,
+ consensus_view,
+ "partition journal repair completed or was superseded"
+ );
+ continue;
+ }
partition.repair.as_mut().and_then(|session| {
if !consensus_normal {
return None;
@@ -6537,8 +6588,8 @@
Some((
session.peer,
session.nonce,
- commit_min + 1,
- session.to_op,
+ commit_min.saturating_add(1),
+ session.fetch_to_op,
cluster,
self_id,
))
@@ -7450,22 +7501,48 @@
B: MessageBus,
{
let consensus = partition.consensus();
- if !consensus.is_normal()
- || consensus.is_transferring()
- || consensus.commit_min() >= consensus.commit_max()
- || partition.repair.is_some()
- {
+ if !consensus.is_normal() || consensus.is_transferring() || partition.repair.is_some() {
+ return;
+ }
+ // The window ends at the group head when suffix bodies are missing,
+ // not at the commit point. A backup that adopted a StartView holds
+ // suffix HEADERS above `commit_max` whose bodies it may never have
+ // received: its ack for them is withheld until the body is journaled,
+ // and the primary's retransmit is dropped by the backup gap check
+ // because adoption already advanced the sequencer to the head. With a
+ // commit-bounded window nothing ever delivers those bodies, the
+ // primary cannot gather quorum for the suffix, and the group wedges
+ // one op below its head with the client write never confirmed.
+ let commit_to_op = consensus.commit_max();
+ let commit_lag = consensus.commit_min() < commit_to_op;
+ let head = consensus.sequencer().current_sequence();
+ if !commit_lag && head <= commit_to_op {
+ return;
+ }
+ let canonical_suffix = consensus
+ .with_pending_view_log(|pending| pending_covers_suffix(pending, commit_to_op, head))
+ .unwrap_or(false);
+ let missing_suffix = canonical_suffix
+ && !partition
+ .log
+ .journal()
+ .inner
+ .repaired_window_shape(commit_to_op, head)
+ .complete;
+ if !commit_lag && !missing_suffix {
return;
}
let nonce = iggy_common::random_id::get_uuid();
let from_op = consensus.commit_min() + 1;
- let to_op = consensus.commit_max();
+ let fetch_to_op = if missing_suffix { head } else { commit_to_op };
let cluster = consensus.cluster();
let self_id = consensus.replica();
let namespace = consensus.group();
partition.repair = Some(partitions::RepairSession {
nonce,
- to_op,
+ view: consensus.view(),
+ commit_to_op,
+ fetch_to_op,
floor: None,
peer,
first_batch_offset: None,
@@ -7475,11 +7552,20 @@
shard = self.id,
namespace_raw = namespace,
from_op,
- to_op,
+ commit_to_op,
+ fetch_to_op,
"partition behind the group frontier; requesting repair"
);
- self.send_request_prepares(cluster, self_id, peer, nonce, from_op, to_op, namespace)
- .await;
+ self.send_request_prepares(
+ cluster,
+ self_id,
+ peer,
+ nonce,
+ from_op,
+ fetch_to_op,
+ namespace,
+ )
+ .await;
}
/// Receiver side of a partition descriptor: accept the manifest, adopt
@@ -8731,6 +8817,27 @@
requested_to_op.min(commit_max.max(head))
}
+/// Whether the parked `StartView` log names every op in the uncommitted
+/// suffix `(commit_max, head]`, in descending order. Only this canonical list
+/// makes fetching bodies above the commit point safe.
+fn pending_covers_suffix(pending: &MergedLog, commit_max: u64, head: u64) -> bool {
+ if head <= commit_max || pending.commit_max != commit_max || pending.op_head != head {
+ return false;
+ }
+ let mut expected = head;
+ for header in pending
+ .headers
+ .iter()
+ .filter(|header| header.op > commit_max)
+ {
+ if header.op != expected {
+ return false;
+ }
+ expected -= 1;
+ }
+ expected == commit_max
+}
+
/// Read this replica's uncommitted suffix out of the metadata journal, for the
/// window `commit..=op`.
///
@@ -9502,7 +9609,7 @@
use iggy_binary_protocol::{Command, PrepareHeader};
- use super::{MergedLog, repair_op_in_scope, repair_serve_ceiling};
+ use super::{MergedLog, pending_covers_suffix, repair_op_in_scope, repair_serve_ceiling};
fn header(op: u64) -> PrepareHeader {
PrepareHeader {
@@ -9573,6 +9680,20 @@
// `commit_max` above the local head still counts: heartbeats outrun prepares.
assert_eq!(repair_serve_ceiling(u64::MAX, 120, 90), 120);
}
+
+ #[test]
+ fn given_a_parked_view_when_fetching_above_commit_should_require_dense_canonical_suffix() {
+ let pending = parked();
+ assert!(pending_covers_suffix(&pending, 98, 100));
+
+ let mut missing = pending.clone();
+ missing.headers.retain(|header| header.op != 99);
+ assert!(!pending_covers_suffix(&missing, 98, 100));
+
+ let mut wrong_frontier = pending;
+ wrong_frontier.commit_max = 97;
+ assert!(!pending_covers_suffix(&wrong_frontier, 98, 100));
+ }
}
#[cfg(test)]
diff --git a/examples/node/package-lock.json b/examples/node/package-lock.json
index c798107..ab1a938 100644
--- a/examples/node/package-lock.json
+++ b/examples/node/package-lock.json
@@ -23,7 +23,7 @@
},
"../../foreign/node": {
"name": "apache-iggy",
- "version": "0.10.0-edge.4",
+ "version": "0.10.0-edge.5",
"license": "Apache-2.0",
"dependencies": {
"@node-rs/xxhash": "1.7.6",
diff --git a/examples/python/uv.lock b/examples/python/uv.lock
index b52ecc4..1f9c5b3 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev4"
+version = "0.9.0.dev5"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/foreign/cpp/Cargo.toml b/foreign/cpp/Cargo.toml
index d1f5afc..0325327 100644
--- a/foreign/cpp/Cargo.toml
+++ b/foreign/cpp/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-cpp"
-version = "0.1.1"
+version = "0.1.2"
edition = "2024"
[package.metadata.cargo-machete]
diff --git a/foreign/cpp/MODULE.bazel b/foreign/cpp/MODULE.bazel
index 86706c0..f274abe 100644
--- a/foreign/cpp/MODULE.bazel
+++ b/foreign/cpp/MODULE.bazel
@@ -17,7 +17,7 @@
module(
name = "iggy_cpp",
- version = "0.1.1",
+ version = "0.1.2",
)
bazel_dep(name = "rules_cc", version = "0.2.22")
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
index e100fcc..d0c52f6 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
@@ -522,8 +522,10 @@
var clearSensitiveReply = HasSensitiveReply(code);
var overallDeadline = Environment.TickCount64 + VsrRequestTimeoutMs;
var requestEncoded = false;
- var redirects = 0;
+ var leaderRedirects = 0;
var redirectBudgetLogged = false;
+ var walkingRoster = false;
+ HashSet<string> walkedRosterEndpoints = new(StringComparer.OrdinalIgnoreCase);
VsrConnection? lastConnection = null;
try
@@ -552,12 +554,27 @@
&& allowRedirect
&& Environment.TickCount64 < overallDeadline)
{
- if (redirects < VsrMaxLeaderRedirects && await RedirectAsync(token))
+ var moved = false;
+ if (leaderRedirects < VsrMaxLeaderRedirects && !walkingRoster)
{
- redirects++;
- await ConnectAsync(token);
+ moved = await RedirectAsync(token);
+ if (moved)
+ {
+ leaderRedirects++;
+ }
}
- else if (redirects >= VsrMaxLeaderRedirects && !redirectBudgetLogged)
+
+ if (!moved)
+ {
+ moved = await RedirectToNextRosterNodeAsync(walkedRosterEndpoints, token);
+ walkingRoster |= moved;
+ }
+
+ if (moved)
+ {
+ await ConnectAsync(true, !walkingRoster, token);
+ }
+ else if (!walkingRoster && leaderRedirects >= VsrMaxLeaderRedirects && !redirectBudgetLogged)
{
redirectBudgetLogged = true;
_logger.LogWarning("Maximum leader redirections reached, continuing on {Address}",
@@ -602,6 +619,64 @@
}
/// <summary>
+ /// Moves to the next unvisited roster endpoint after a never-admitted refusal. Metadata and partition
+ /// groups elect independently, so the metadata leader is not necessarily the primary for the request's
+ /// partition. The caller's request deadline and redirect budget bound the walk.
+ /// </summary>
+ private async Task<bool> RedirectToNextRosterNodeAsync(HashSet<string> visited, CancellationToken token)
+ {
+ var roster = _rosterAddresses;
+ if (roster.Length == 0)
+ {
+ return false;
+ }
+
+ var currentIndex = Array.FindIndex(roster, address =>
+ (_connectedAddress.Length > 0 && ServerAddress.IsSame(address, _connectedAddress))
+ || (_currentRemoteAddress.Length > 0 && ServerAddress.IsSame(address, _currentRemoteAddress)));
+ if (currentIndex >= 0)
+ {
+ visited.Add(roster[currentIndex]);
+ }
+
+ string? next = null;
+ for (var offset = 1; offset <= roster.Length; offset++)
+ {
+ var candidate = roster[(Math.Max(currentIndex, -1) + offset) % roster.Length];
+ if (visited.Contains(candidate)
+ || (_connectedAddress.Length > 0 && ServerAddress.IsSame(candidate, _connectedAddress))
+ || (_currentRemoteAddress.Length > 0 && ServerAddress.IsSame(candidate, _currentRemoteAddress)))
+ {
+ continue;
+ }
+
+ next = candidate;
+ break;
+ }
+
+ if (next is null)
+ {
+ return false;
+ }
+ visited.Add(next);
+ _logger.LogInformation("The request was refused on {Address}, walking the roster to {NextAddress}",
+ _connectedAddress, next);
+
+ await _sendingSemaphore.WaitAsync(token);
+ try
+ {
+ _currentAddress = next;
+ DropVsrConnectionLocked(_connection);
+ }
+ finally
+ {
+ _sendingSemaphore.Release();
+ }
+
+ return true;
+ }
+
+ /// <summary>
/// Whether the failure carries the server's verdict on this request. A lost connection, a reply frame
/// the client refused or discarded, and a NOT_COMMITTED that outlived its replay deadline all leave the
/// outcome of a request the server may still commit unknowable.
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index 245299d..928d5e9 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
@@ -608,7 +608,7 @@
"Reconnection is enabled without auto login: a lost session can only be restored once the client has signed in at least once");
}
- return ConnectAsync(true, token);
+ return ConnectAsync(true, true, token);
}
/// <inheritdoc />
@@ -817,6 +817,15 @@
/// </summary>
private async Task ConnectAsync(bool autoLogin, CancellationToken token)
{
+ await ConnectAsync(autoLogin, true, token);
+ }
+
+ /// <summary>
+ /// Connects and optionally leaves an authenticated roster walk on the endpoint it reached instead of
+ /// settling it back onto the metadata leader.
+ /// </summary>
+ private async Task ConnectAsync(bool autoLogin, bool settleOnLeader, CancellationToken token)
+ {
if (State is ConnectionState.Connected
or ConnectionState.Authenticating
or ConnectionState.Authenticated)
@@ -847,7 +856,7 @@
}
SetConnectionState(ConnectionState.Connecting);
- await TryEstablishConnectionAsync(autoLogin, token);
+ await TryEstablishConnectionAsync(autoLogin, settleOnLeader, token);
// Dispose is synchronous and cannot take the sending semaphore, so a Dispose that ran while this
// connect was dialing already read a connection that did not exist yet. Reading the flag after the
@@ -1029,7 +1038,7 @@
};
}
- private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken token)
+ private async Task TryEstablishConnectionAsync(bool autoLogin, bool settleOnLeader, CancellationToken token)
{
var retryCount = 0;
var redirects = 0;
@@ -1118,7 +1127,7 @@
{
await AutoLoginAsync(signInSettings, token);
- if (await RedirectAsync(token))
+ if (settleOnLeader && await RedirectAsync(token))
{
await BackoffOrThrowAsync();
continue;
diff --git a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
index 158ea6d..baa0fe5 100644
--- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
+++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
@@ -26,7 +26,7 @@
<TargetFrameworks>net8.0;net10.0</TargetFrameworks>
<AssemblyName>Apache.Iggy</AssemblyName>
<RootNamespace>Apache.Iggy</RootNamespace>
- <Version>0.9.0-edge.5</Version>
+ <Version>0.9.0-edge.6</Version>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
</PropertyGroup>
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
index e0e0203..3f6ad31 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
@@ -55,6 +55,7 @@
private const byte OperationNonReplicated = 2;
private const int GetClusterMetadataCode = 12;
private const int PingCode = 1;
+ private const uint TransientNotAccepted = 58;
[Fact]
public async Task ResumesOnASurvivorAfterTheSignedInNodeDies()
@@ -103,6 +104,101 @@
Assert.True(survivor.Pings >= 1, "the request landed on the survivor");
}
+ [Fact]
+ public async Task WalksPastTwoRefusingReplicasToThePartitionPrimary()
+ {
+ const uint commandCode = 60_040;
+ using var metadataLeader = new MockNode();
+ using var follower = new MockNode();
+ using var partitionPrimary = new MockNode();
+
+ metadataLeader.Serve(request => request.Code switch
+ {
+ GetClusterMetadataCode => Reply(OperationNonReplicated,
+ ThreeNodeClusterMetadata(metadataLeader.Port, follower.Port, partitionPrimary.Port)),
+ (int)commandCode => Reply(request.Operation, [], TransientNotAccepted),
+ _ => Answer(request)
+ });
+ follower.Serve(request => request.Code == (int)commandCode
+ ? Reply(request.Operation, [], TransientNotAccepted)
+ : Answer(request));
+ partitionPrimary.Serve(Answer);
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{metadataLeader.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 1,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration, NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.LoginUserAsync("iggy", "iggy", TestContext.Current.CancellationToken);
+
+ Assert.Empty(await client.SendBinaryRequestAsync(commandCode, [], TestContext.Current.CancellationToken));
+ Assert.True(follower.Connections >= 1, "the roster walk skipped the second replica");
+ Assert.True(partitionPrimary.Connections >= 1, "the roster walk never reached the partition primary");
+ }
+
+ [Fact]
+ public async Task WalksTheWholeRosterBeyondTheMetadataRedirectCap()
+ {
+ const uint commandCode = 60_041;
+ using var metadataLeader = new MockNode();
+ using var second = new MockNode();
+ using var third = new MockNode();
+ using var fourth = new MockNode();
+ using var partitionPrimary = new MockNode();
+ var roster = new[]
+ {
+ metadataLeader.Port,
+ second.Port,
+ third.Port,
+ fourth.Port,
+ partitionPrimary.Port
+ };
+
+ byte[] Refuse(MockRequest request)
+ {
+ return request.Code == (int)commandCode
+ ? Reply(request.Operation, [], TransientNotAccepted)
+ : Answer(request);
+ }
+
+ metadataLeader.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, RosterMetadata(metadataLeader.Port, roster))
+ : Refuse(request));
+ second.Serve(Refuse);
+ third.Serve(Refuse);
+ fourth.Serve(Refuse);
+ partitionPrimary.Serve(Answer);
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{metadataLeader.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 1,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration, NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.LoginUserAsync("iggy", "iggy", TestContext.Current.CancellationToken);
+
+ Assert.Empty(await client.SendBinaryRequestAsync(commandCode, [], TestContext.Current.CancellationToken));
+ Assert.True(partitionPrimary.Connections >= 1,
+ "the arbitrary metadata redirect cap stopped the bounded roster walk");
+ }
+
/// <summary>
/// Mirrors the integration contract (HeartbeatTests
/// EvictedClient_WithoutAutoLogin_Should_ReestablishItsSession): an eviction comes off the server's
@@ -371,11 +467,16 @@
private static byte[] Reply(byte operation, byte[] body)
{
+ return Reply(operation, body, 0);
+ }
+
+ private static byte[] Reply(byte operation, byte[] body, uint status)
+ {
var frame = new byte[HeaderSize + body.Length];
BinaryPrimitives.WriteUInt32LittleEndian(frame.AsSpan(SizeOffset, 4), (uint)frame.Length);
frame[CommandOffset] = CommandReply;
frame[ReplyOperationOffset] = operation;
- BinaryPrimitives.WriteUInt32LittleEndian(frame.AsSpan(ReplyStatusOffset, 4), 0);
+ BinaryPrimitives.WriteUInt32LittleEndian(frame.AsSpan(ReplyStatusOffset, 4), status);
body.CopyTo(frame.AsSpan(HeaderSize));
return frame;
@@ -410,6 +511,31 @@
return body.ToArray();
}
+ private static byte[] ThreeNodeClusterMetadata(ushort firstPort, ushort secondPort, ushort thirdPort)
+ {
+ var body = new List<byte>();
+ WriteString(body, "test-cluster");
+ body.AddRange(BitConverter.GetBytes(3u));
+ WriteNode(body, "metadata-leader", firstPort, true);
+ WriteNode(body, "follower", secondPort, false);
+ WriteNode(body, "partition-primary", thirdPort, false);
+
+ return body.ToArray();
+ }
+
+ private static byte[] RosterMetadata(ushort leaderPort, IReadOnlyList<ushort> ports)
+ {
+ var body = new List<byte>();
+ WriteString(body, "test-cluster");
+ body.AddRange(BitConverter.GetBytes((uint)ports.Count));
+ for (var index = 0; index < ports.Count; index++)
+ {
+ WriteNode(body, $"node-{index}", ports[index], ports[index] == leaderPort);
+ }
+
+ return body.ToArray();
+ }
+
private static void WriteNode(List<byte> body, string name, ushort port, bool leader)
{
WriteString(body, name);
diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go
index 057907d..0bd8b36 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -488,12 +488,21 @@
// already-connected gate, and then suppresses somebody else's auto-login.
type skipAutoLogin struct{}
+// skipLeaderSettlement keeps the sign-in owned by one roster-walk Connect on
+// the endpoint it dialed. Context scope prevents a failed or cancelled Connect
+// from suppressing an unrelated later sign-in.
+type skipLeaderSettlement struct{}
+
// suppressAutoLogin returns ctx marked so the Connect it drives does not sign
// in by itself.
func suppressAutoLogin(ctx context.Context) context.Context {
return context.WithValue(ctx, skipAutoLogin{}, struct{}{})
}
+func suppressLeaderSettlement(ctx context.Context) context.Context {
+ return context.WithValue(ctx, skipLeaderSettlement{}, struct{}{})
+}
+
// localPreconditionError marks a request that failed before its frame was
// written. The connection is healthy, so exchange must not tear it down and
// re-dial over what is purely local state.
@@ -632,6 +641,12 @@
deadline := time.Now().Add(responseReadTimeout)
stamped := false
+ // Once this request starts walking the roster it keeps walking: a leader
+ // recheck between hops would put it straight back on the metadata leader
+ // whose partition replica refused it, and the walk would bounce between
+ // two nodes without ever reaching the rest of the roster.
+ walkingRoster := false
+ visitedRosterEndpoints := make(map[string]struct{})
for {
// A sign-in owns the whole budget on this connection: any node
// completes it (a backup forwards the register to the primary), and
@@ -656,12 +671,38 @@
// The server never admitted the request, so re-issuing it cannot
// double-apply. A same-connection replay keeps the stamped request
// id; a redirect registers again, so the frame is stamped afresh.
- redirect, redirectErr := c.HandleLeaderRedirection(ctx)
- if redirectErr != nil {
- return nil, redirectErr
+ redirect := false
+ walked := false
+ if !walkingRoster {
+ var redirectErr error
+ redirect, redirectErr = c.HandleLeaderRedirection(ctx)
+ if redirectErr != nil {
+ return nil, redirectErr
+ }
+ }
+ if !redirect {
+ // The roster names this node as the metadata leader (or said
+ // nothing usable), yet it keeps refusing to admit the
+ // request: its replica of the target partition group is not
+ // that group's primary, because metadata and partition
+ // consensus groups elect independently. Walk the roster
+ // instead of replaying into the same refusal until the whole
+ // request budget burns.
+ var walkErr error
+ walked, walkErr = c.settleOnNextEndpoint(visitedRosterEndpoints)
+ if walkErr != nil {
+ return nil, walkErr
+ }
+ if walked {
+ walkingRoster = true
+ redirect = true
+ }
}
if redirect {
redirectCtx := ctx
+ if walked {
+ redirectCtx = suppressLeaderSettlement(redirectCtx)
+ }
if ctx.Value(connectScoped{}) != nil {
// Issued from inside the sign-in transaction, which holds
// registerMtx: the automatic sign-in on the reconnect path
diff --git a/foreign/go/client/tcp/tcp_failover_test.go b/foreign/go/client/tcp/tcp_failover_test.go
index e7946a4..1a035de 100644
--- a/foreign/go/client/tcp/tcp_failover_test.go
+++ b/foreign/go/client/tcp/tcp_failover_test.go
@@ -101,6 +101,67 @@
"the remembered credentials signed in again on the survivor")
}
+func TestFailover_WalksPastTwoRefusingReplicasToThePartitionPrimary(t *testing.T) {
+ var metadataLeader *testListener
+
+ partitionPrimary := listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch read.operation() {
+ case vsr.OperationRegister:
+ return registerReplyFrame(7, 384)
+ case vsr.OperationCreateStream:
+ return replyFrame(vsr.OperationCreateStream, resultSection())
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+ follower := listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if read.operation() == vsr.OperationRegister {
+ return registerReplyFrame(7, 256)
+ }
+ return statusReplyFrame(vsr.OperationCreateStream,
+ uint32(ierror.TransientNotAcceptedCode), nil)
+ })
+ metadataLeader = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 0, metadataLeader.address(),
+ follower.address(), partitionPrimary.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 128)
+ default:
+ return statusReplyFrame(vsr.OperationCreateStream,
+ uint32(ierror.TransientNotAcceptedCode), nil)
+ }
+ })
+
+ client := newDialingClient(t, metadataLeader.address())
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+
+ _, err = client.do(ctx, &command.CreateStream{Name: "orders"})
+ require.NoError(t, err)
+ assert.Equal(t, partitionPrimary.address(), client.currentServerAddress)
+ assert.True(t, len(follower.recorded()) >= 2,
+ "the roster walk skipped the second replica")
+}
+
+func TestFailover_RosterWalkVisitsEachEndpointOnce(t *testing.T) {
+ roster := []string{"iggy-0:8090", "iggy-1:8090", "iggy-2:8090"}
+ visited := make(map[string]struct{})
+
+ second := nextRosterEndpoint(roster[0], roster, visited)
+ third := nextRosterEndpoint(second, roster, visited)
+ exhausted := nextRosterEndpoint(third, roster, visited)
+
+ assert.Equal(t, roster[1], second)
+ assert.Equal(t, roster[2], third)
+ assert.Empty(t, exhausted)
+ assert.Len(t, visited, len(roster))
+}
+
// Without any credentials there is nothing to sign in with, so a request on a
// dead node fails instead of reconnecting into an unauthenticated session.
func TestFailover_FailsFastWhenNothingEverSignedIn(t *testing.T) {
diff --git a/foreign/go/client/tcp/tcp_session_management.go b/foreign/go/client/tcp/tcp_session_management.go
index 4da1fb2..a34d055 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -162,6 +162,15 @@
//
// Returns nil when the client stays where it is.
func (c *IggyTcpClient) settleOnLeader(ctx context.Context, code uint32, body []byte) (*iggcon.IdentityInfo, error) {
+ // A roster walk stays on the endpoint it dialed: the settlement below
+ // would put the connection straight back on the node whose partition
+ // replica keeps refusing the walked request. The marker is scoped to the
+ // Connect that owns this sign-in, so a failed walk cannot leak into another.
+ if ctx.Value(skipLeaderSettlement{}) != nil {
+ c.logger.Info("Staying on the dialed node for a partition failover.")
+ return nil, nil
+ }
+
var settled *iggcon.IdentityInfo
for {
// The roster read runs while register holds the sign-in lock, so it must
@@ -296,3 +305,64 @@
return true, nil
}
+
+// settleOnNextEndpoint moves the connection to the roster endpoint after the
+// current one, for a request the current node keeps refusing to admit.
+// Metadata and partition consensus groups elect independently, so the
+// metadata leader can hold a follower replica of the partition a request
+// targets, and only walking the roster reaches that group's primary. The
+// refusal marks the request as never admitted and safe to re-issue anywhere;
+// the caller's request budget bounds the walk.
+func (c *IggyTcpClient) settleOnNextEndpoint(visited map[string]struct{}) (bool, error) {
+ c.mtx.Lock()
+ current := c.currentServerAddress
+ roster := append([]string(nil), c.knownServerAddresses...)
+ c.mtx.Unlock()
+
+ next := nextRosterEndpoint(current, roster, visited)
+ if next == "" {
+ return false, nil
+ }
+
+ c.logger.Info(
+ "The request keeps being refused while the roster names this node the metadata leader; trying the next cluster node.",
+ slog.String("current", current),
+ slog.String("next", next),
+ )
+ if err := c.disconnect(); err != nil {
+ return false, err
+ }
+
+ c.mtx.Lock()
+ c.connectedAt = time.Time{}
+ c.currentServerAddress = next
+ c.mtx.Unlock()
+ return true, nil
+}
+
+func nextRosterEndpoint(current string, roster []string, visited map[string]struct{}) string {
+ currentIndex := -1
+ for index, endpoint := range roster {
+ if endpoint == current {
+ currentIndex = index
+ visited[endpoint] = struct{}{}
+ break
+ }
+ }
+
+ next := ""
+ for offset := 1; offset <= len(roster); offset++ {
+ index := (currentIndex + offset) % len(roster)
+ candidate := roster[index]
+ if _, seen := visited[candidate]; seen || candidate == current {
+ continue
+ }
+ next = candidate
+ break
+ }
+ if next == "" || next == current {
+ return ""
+ }
+ visited[next] = struct{}{}
+ return next
+}
diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go
index 9f51ca3..32ff7a2 100644
--- a/foreign/go/contracts/version.go
+++ b/foreign/go/contracts/version.go
@@ -17,4 +17,4 @@
package iggcon
-const Version = "0.9.0-edge.4"
+const Version = "0.9.0-edge.5"
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
index 3f6ebd0..d65622f 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
@@ -557,7 +557,8 @@
int commandCode,
ByteBuf payload,
long requestDeadlineNanos,
- IggyServerException rejection) {
+ IggyServerException rejection,
+ AsyncTcpConnection.TransientFailoverState failoverState) {
AtomicReference<ByteBuf> requestPayload = new AtomicReference<>(payload);
Optional<AsyncTcpConnection.AuthenticationSnapshot> authentication = source.authenticationSnapshot();
AtomicReference<ByteBuf> authenticationPayload = new AtomicReference<>(authentication
@@ -565,7 +566,7 @@
.orElse(null));
CompletableFuture<Void> ready = prepareTransientFailover(
- source, authentication, authenticationPayload, requestDeadlineNanos, rejection);
+ source, authentication, authenticationPayload, requestDeadlineNanos, rejection, failoverState);
CompletableFuture<ByteBuf> retried = ready.thenCompose(ignored -> {
if (requestDeadlineNanos - System.nanoTime() <= 0) {
return CompletableFuture.failedFuture(rejection);
@@ -574,7 +575,8 @@
if (currentConnection == null) {
return CompletableFuture.failedFuture(new IggyNotConnectedException());
}
- return currentConnection.send(commandCode, takePayload(requestPayload), requestDeadlineNanos);
+ return currentConnection.send(
+ commandCode, takePayload(requestPayload), requestDeadlineNanos, failoverState);
});
return retried.whenComplete((response, error) -> {
releasePayload(requestPayload);
@@ -587,10 +589,10 @@
Optional<AsyncTcpConnection.AuthenticationSnapshot> authentication,
AtomicReference<ByteBuf> authenticationPayload,
long requestDeadlineNanos,
- IggyServerException rejection) {
+ IggyServerException rejection,
+ AsyncTcpConnection.TransientFailoverState failoverState) {
CompletableFuture<Void> gate = new CompletableFuture<>();
CompletableFuture<Void> previous = loginChain.getAndSet(gate);
- LeaderRedirectionState redirectionState = new LeaderRedirectionState();
CompletableFuture<Void> transaction = previous.thenCompose(ignored -> {
if (closed) {
return CompletableFuture.failedFuture(new IggyNotConnectedException());
@@ -601,7 +603,7 @@
if (connection.get() != source) {
return CompletableFuture.completedFuture(null);
}
- return redirectToLeader(redirectionState).thenCompose(redirected -> {
+ return walkRoster(requestDeadlineNanos, failoverState).thenCompose(ignoredWalk -> {
AsyncTcpConnection currentConnection = connection.get();
if (currentConnection == null || currentConnection == source || authentication.isEmpty()) {
return CompletableFuture.completedFuture(null);
@@ -883,43 +885,44 @@
}
/**
- * One discovery hop for transient failover. When the roster names a healthy leader elsewhere,
- * reconnect to it and re-check from the new node, since mid-election
- * metadata can point at a node that is itself not the leader. Register is
- * sent only after this bounded process settles. Reading the roster needs a
- * bound session, so a connection that has none fails the fetch locally and
- * stays where it is. Login settlement uses {@link #loginAndSettleOnLeader}
- * so every redirected connection binds before its next roster read.
+ * Walks the known roster once after a never-admitted refusal. Metadata and
+ * partition groups elect independently, so redirecting to the metadata
+ * leader can bounce a partition request away from its primary. Failed
+ * dials are skipped within the same request deadline.
*/
- private CompletableFuture<Void> redirectToLeader(LeaderRedirectionState redirectionState) {
+ private CompletableFuture<Void> walkRoster(
+ long requestDeadlineNanos, AsyncTcpConnection.TransientFailoverState failoverState) {
ConnectionInfo currentTarget = connectionInfo;
- return findLeaderElsewhere(currentTarget).thenCompose(leaderTarget -> {
- if (leaderTarget.isEmpty()) {
- return CompletableFuture.completedFuture(null);
- }
- if (!redirectionState.canRedirect()) {
- log.warn(
- "Maximum leader redirections ({}) reached, connection will continue on server node {}",
- LeaderAwareness.MAX_LEADER_REDIRECTS,
- currentTarget.serverAddress());
- return CompletableFuture.completedFuture(null);
- }
- return retarget(leaderTarget.get())
- .handle((ignored, error) -> {
- if (error != null) {
- log.warn(
- "Failed to reconnect to leader at {}: {}, connection will continue"
- + " on server node {}",
- leaderTarget.get().serverAddress(),
- error.getMessage(),
- currentTarget.serverAddress());
- return CompletableFuture.<Void>completedFuture(null);
- }
- redirectionState.recordRedirect();
- return redirectToLeader(redirectionState);
- })
- .thenCompose(Function.identity());
- });
+ failoverState.visitedTargets().add(currentTarget);
+ List<ConnectionInfo> targets =
+ LeaderAwareness.rosterWalkTargets(rosterTargets, currentTarget, failoverState.visitedTargets());
+ return walkRoster(targets, 0, currentTarget, requestDeadlineNanos, failoverState);
+ }
+
+ private CompletableFuture<Void> walkRoster(
+ List<ConnectionInfo> targets,
+ int index,
+ ConnectionInfo originalTarget,
+ long requestDeadlineNanos,
+ AsyncTcpConnection.TransientFailoverState failoverState) {
+ if (index >= targets.size() || requestDeadlineNanos - System.nanoTime() <= 0) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ConnectionInfo target = targets.get(index);
+ failoverState.visitedTargets().add(target);
+ log.info(
+ "The request was refused on {}, walking the roster to {}",
+ originalTarget.serverAddress(),
+ target.serverAddress());
+ return retarget(target)
+ .handle((ignored, error) -> {
+ if (error == null) {
+ return CompletableFuture.<Void>completedFuture(null);
+ }
+ log.warn("Roster walk to {} failed: {}", target.serverAddress(), error.getMessage());
+ return walkRoster(targets, index + 1, originalTarget, requestDeadlineNanos, failoverState);
+ })
+ .thenCompose(Function.identity());
}
/**
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
index 48d70ab..bdf97f9 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
@@ -39,6 +39,7 @@
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.FutureListener;
import io.netty.util.concurrent.ScheduledFuture;
+import org.apache.iggy.client.ConnectionInfo;
import org.apache.iggy.client.async.tcp.vsr.ConsensusSession;
import org.apache.iggy.client.async.tcp.vsr.VsrFrameDecoder;
import org.apache.iggy.client.async.tcp.vsr.VsrRequestEncoder;
@@ -59,8 +60,10 @@
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
@@ -342,6 +345,11 @@
}
CompletableFuture<ByteBuf> send(int commandCode, ByteBuf payload, long requestDeadlineNanos) {
+ return send(commandCode, payload, requestDeadlineNanos, new TransientFailoverState());
+ }
+
+ CompletableFuture<ByteBuf> send(
+ int commandCode, ByteBuf payload, long requestDeadlineNanos, TransientFailoverState failoverState) {
if (isLoginCode(commandCode) && authenticated) {
return logoutThenLogin(commandCode, payload);
}
@@ -366,7 +374,8 @@
failoverPayload,
responseFuture,
callerFuture,
- requestDeadlineNanos);
+ requestDeadlineNanos,
+ failoverState);
});
return callerFuture;
@@ -380,9 +389,17 @@
ByteBuf failoverPayload,
CompletableFuture<ByteBuf> responseFuture,
CompletableFuture<ByteBuf> callerFuture,
- long requestDeadlineNanos) {
+ long requestDeadlineNanos,
+ TransientFailoverState failoverState) {
Runnable dispatch = () -> dispatchOnChannel(
- channel, commandCode, payload, failoverPayload, responseFuture, callerFuture, requestDeadlineNanos);
+ channel,
+ commandCode,
+ payload,
+ failoverPayload,
+ responseFuture,
+ callerFuture,
+ requestDeadlineNanos,
+ failoverState);
if (channel.eventLoop().inEventLoop()) {
dispatch.run();
return;
@@ -397,6 +414,7 @@
}
}
+ @SuppressWarnings("checkstyle:ParameterNumber")
private void dispatchOnChannel(
Channel channel,
int commandCode,
@@ -404,7 +422,8 @@
ByteBuf failoverPayload,
CompletableFuture<ByteBuf> responseFuture,
CompletableFuture<ByteBuf> callerFuture,
- long inheritedRequestDeadlineNanos) {
+ long inheritedRequestDeadlineNanos,
+ TransientFailoverState failoverState) {
boolean isLoginCommand = isLoginCode(commandCode);
boolean holdLeaseUntilResponse = mutatesSessionState(commandCode);
long requestDeadlineNanos = inheritedRequestDeadlineNanos == 0
@@ -419,6 +438,7 @@
requestDeadlineNanos,
holdLeaseUntilResponse,
callerFuture,
+ failoverState,
response,
error));
authenticationStep(channel, commandCode, requestDeadlineNanos)
@@ -483,6 +503,7 @@
long requestDeadlineNanos,
boolean holdLeaseUntilResponse,
CompletableFuture<ByteBuf> callerFuture,
+ TransientFailoverState failoverState,
ByteBuf response,
Throwable error) {
try {
@@ -493,6 +514,7 @@
failoverPayload,
requestDeadlineNanos,
callerFuture,
+ failoverState,
response,
error);
} finally {
@@ -510,6 +532,7 @@
ByteBuf failoverPayload,
long requestDeadlineNanos,
CompletableFuture<ByteBuf> callerFuture,
+ TransientFailoverState failoverState,
ByteBuf response,
Throwable error) {
try {
@@ -522,7 +545,7 @@
completeWithResponse(callerFuture, response);
return;
}
- completeFailedRequest(commandCode, failoverPayload, requestDeadlineNanos, callerFuture, error);
+ completeFailedRequest(commandCode, failoverPayload, requestDeadlineNanos, callerFuture, failoverState, error);
}
private void completeFailedRequest(
@@ -530,6 +553,7 @@
ByteBuf failoverPayload,
long requestDeadlineNanos,
CompletableFuture<ByteBuf> callerFuture,
+ TransientFailoverState failoverState,
Throwable error) {
IggyTimeoutException timeout = findResponseTimeout(error);
if (timeout != null) {
@@ -537,7 +561,8 @@
}
IggyServerException serverError = findServerError(error);
if (shouldRecheckLeader(serverError, failoverPayload, requestDeadlineNanos)) {
- retryAfterLeaderRecheck(commandCode, failoverPayload, requestDeadlineNanos, serverError, callerFuture);
+ retryAfterLeaderRecheck(
+ commandCode, failoverPayload, requestDeadlineNanos, serverError, callerFuture, failoverState);
return;
}
releaseIfPresent(failoverPayload);
@@ -557,10 +582,12 @@
ByteBuf payload,
long requestDeadlineNanos,
IggyServerException rejection,
- CompletableFuture<ByteBuf> callerFuture) {
+ CompletableFuture<ByteBuf> callerFuture,
+ TransientFailoverState failoverState) {
CompletableFuture<ByteBuf> retry;
try {
- retry = transientFailoverHandler.retry(this, commandCode, payload, requestDeadlineNanos, rejection);
+ retry = transientFailoverHandler.retry(
+ this, commandCode, payload, requestDeadlineNanos, rejection, failoverState);
} catch (RuntimeException retryError) {
payload.release();
callerFuture.completeExceptionally(retryError);
@@ -953,6 +980,15 @@
record AuthenticationSnapshot(int commandCode, ByteBuf payload) {}
+ static final class TransientFailoverState {
+
+ private final Set<ConnectionInfo> visitedTargets = new HashSet<>();
+
+ Set<ConnectionInfo> visitedTargets() {
+ return visitedTargets;
+ }
+ }
+
@FunctionalInterface
interface TransientFailoverHandler {
CompletableFuture<ByteBuf> retry(
@@ -960,7 +996,8 @@
int commandCode,
ByteBuf payload,
long requestDeadlineNanos,
- IggyServerException rejection);
+ IggyServerException rejection,
+ TransientFailoverState failoverState);
}
public static class TcpConnectionPoolConfig {
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java
index c316733..4500ee2 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java
@@ -30,10 +30,12 @@
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
@@ -165,6 +167,37 @@
}
/**
+ * Every roster endpoint after the current one, wrapping once and omitting
+ * aliases of the current endpoint. This is the bounded candidate order for
+ * a request a partition follower has refused as never admitted.
+ */
+ static List<ConnectionInfo> rosterWalkTargets(
+ List<ConnectionInfo> roster, ConnectionInfo currentTarget, Set<ConnectionInfo> visitedTargets) {
+ if (roster.isEmpty()) {
+ return List.of();
+ }
+ int currentIndex = -1;
+ for (int index = 0; index < roster.size(); index++) {
+ if (isSameAddress(roster.get(index), currentTarget)) {
+ currentIndex = index;
+ break;
+ }
+ }
+ int start = currentIndex < 0 ? 0 : (currentIndex + 1) % roster.size();
+ var targets = new ArrayList<ConnectionInfo>(roster.size());
+ for (int offset = 0; offset < roster.size(); offset++) {
+ var candidate = roster.get((start + offset) % roster.size());
+ if (isSameAddress(candidate, currentTarget)
+ || visitedTargets.stream().anyMatch(visited -> isSameAddress(visited, candidate))
+ || targets.stream().anyMatch(target -> isSameAddress(target, candidate))) {
+ continue;
+ }
+ targets.add(candidate);
+ }
+ return List.copyOf(targets);
+ }
+
+ /**
* One leader-check verdict from a cluster-metadata snapshot.
*/
static LeaderCheck checkLeader(ClusterMetadata metadata, ConnectionInfo currentTarget) {
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java
index 897c5e4..abc7a3e 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java
@@ -68,7 +68,7 @@
private static final int EVICTION_STALE_CLIENT = 13;
@Test
- void shouldRecheckLeaderAndReplayNotAcceptedMutation() throws Exception {
+ void shouldWalkRosterAndReplayNotAcceptedMutation() throws Exception {
InetAddress loopback = InetAddress.getLoopbackAddress();
try (ServerSocket oldLeaderSocket = new ServerSocket(0, 1, loopback);
ServerSocket newLeaderSocket = new ServerSocket(0, 1, loopback)) {
@@ -109,6 +109,77 @@
}
@Test
+ void shouldWalkPastTwoRefusingReplicasToThePartitionPrimary() throws Exception {
+ InetAddress loopback = InetAddress.getLoopbackAddress();
+ try (ServerSocket metadataLeaderSocket = new ServerSocket(0, 1, loopback);
+ ServerSocket followerSocket = new ServerSocket(0, 1, loopback);
+ ServerSocket partitionPrimarySocket = new ServerSocket(0, 1, loopback)) {
+ int metadataLeaderPort = metadataLeaderSocket.getLocalPort();
+ int followerPort = followerSocket.getLocalPort();
+ int partitionPrimaryPort = partitionPrimarySocket.getLocalPort();
+ AtomicInteger accepted = new AtomicInteger();
+ CompletableFuture<Void> metadataLeader = serve(metadataLeaderSocket, request -> {
+ if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) {
+ return Response.success(
+ OPERATION_NON_REPLICATED,
+ threeNodeMetadata(metadataLeaderPort, followerPort, partitionPrimaryPort));
+ }
+ if (request.operation() == OPERATION_REGISTER) {
+ return Response.success(OPERATION_REGISTER, registerBody(1));
+ }
+ if (request.operation() == OPERATION_CREATE_STREAM) {
+ return Response.error(OPERATION_CREATE_STREAM, TRANSIENT_NOT_ACCEPTED);
+ }
+ throw new IllegalStateException("Unexpected request to metadata leader: " + request);
+ });
+ CompletableFuture<Void> follower = serve(followerSocket, request -> {
+ if (request.operation() == OPERATION_REGISTER) {
+ return Response.success(OPERATION_REGISTER, registerBody(2));
+ }
+ if (request.operation() == OPERATION_CREATE_STREAM) {
+ return Response.error(OPERATION_CREATE_STREAM, TRANSIENT_NOT_ACCEPTED);
+ }
+ throw new IllegalStateException("Unexpected request to follower: " + request);
+ });
+ CompletableFuture<Void> partitionPrimary = serve(partitionPrimarySocket, request -> {
+ if (request.operation() == OPERATION_REGISTER) {
+ return Response.success(OPERATION_REGISTER, registerBody(3));
+ }
+ if (request.operation() == OPERATION_CREATE_STREAM) {
+ accepted.incrementAndGet();
+ ByteBuf body = Unpooled.buffer(Integer.BYTES);
+ body.writeIntLE(0);
+ return Response.success(OPERATION_CREATE_STREAM, body);
+ }
+ throw new IllegalStateException("Unexpected request to partition primary: " + request);
+ });
+
+ AsyncIggyTcpClient client = AsyncIggyTcpClient.builder()
+ .host(loopback.getHostAddress())
+ .port(metadataLeaderPort)
+ .credentials("iggy", "iggy")
+ .requestTimeout(Duration.ofSeconds(15))
+ .build();
+ try {
+ client.connect().get(5, TimeUnit.SECONDS);
+ client.login().get(5, TimeUnit.SECONDS);
+
+ byte[] response = client.sendBinaryRequest(CREATE_STREAM_CODE, new byte[0])
+ .get(15, TimeUnit.SECONDS);
+
+ assertThat(response).isEmpty();
+ assertThat(client.getConnectionInfo().port()).isEqualTo(partitionPrimaryPort);
+ assertThat(accepted).hasValue(1);
+ } finally {
+ client.close().get(5, TimeUnit.SECONDS);
+ }
+ metadataLeader.get(5, TimeUnit.SECONDS);
+ follower.get(5, TimeUnit.SECONDS);
+ partitionPrimary.get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
void shouldReplayTransientImplicitLoginAfterEviction() throws Exception {
InetAddress loopback = InetAddress.getLoopbackAddress();
try (ServerSocket serverSocket = new ServerSocket(0, 1, loopback)) {
@@ -386,23 +457,38 @@
return serve(server, 1, handler);
}
+ /**
+ * Runs blocking socket I/O on one dedicated daemon thread per mock node.
+ * The common fork-join pool has only cores minus one workers on small CI
+ * runners, so three blocking nodes can starve the client continuations the
+ * test is waiting for when the full suite runs concurrently.
+ */
private static CompletableFuture<Void> serve(ServerSocket server, int connectionCount, RequestHandler handler) {
- return CompletableFuture.runAsync(() -> {
- try {
- for (int connection = 0; connection < connectionCount; connection++) {
- try (Socket socket = server.accept()) {
- InputStream input = socket.getInputStream();
- OutputStream output = socket.getOutputStream();
- Request request;
- while ((request = readRequest(input)) != null) {
- writeResponse(output, request, handler.handle(request));
+ CompletableFuture<Void> serving = new CompletableFuture<>();
+ Thread serverThread = new Thread(
+ () -> {
+ try {
+ for (int connection = 0; connection < connectionCount; connection++) {
+ try (Socket socket = server.accept()) {
+ InputStream input = socket.getInputStream();
+ OutputStream output = socket.getOutputStream();
+ Request request;
+ while ((request = readRequest(input)) != null) {
+ writeResponse(output, request, handler.handle(request));
+ }
+ }
}
+ serving.complete(null);
+ } catch (IOException error) {
+ serving.completeExceptionally(new IllegalStateException("Mock VSR server failed", error));
+ } catch (RuntimeException error) {
+ serving.completeExceptionally(error);
}
- }
- } catch (IOException error) {
- throw new IllegalStateException("Mock VSR server failed", error);
- }
- });
+ },
+ "transient-failover-server-" + server.getLocalPort());
+ serverThread.setDaemon(true);
+ serverThread.start();
+ return serving;
}
private static Request readRequest(InputStream input) throws IOException {
@@ -481,6 +567,16 @@
return body;
}
+ private static ByteBuf threeNodeMetadata(int firstPort, int secondPort, int thirdPort) {
+ ByteBuf body = Unpooled.buffer();
+ writeString(body, "test-cluster");
+ body.writeIntLE(3);
+ writeNode(body, "metadata-leader", firstPort, true);
+ writeNode(body, "follower", secondPort, false);
+ writeNode(body, "partition-primary", thirdPort, false);
+ return body;
+ }
+
private static void writeNode(ByteBuf body, String name, int port, boolean leader) {
writeString(body, name);
writeString(body, InetAddress.getLoopbackAddress().getHostAddress());
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java
index a847acb..efae342 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java
@@ -33,6 +33,7 @@
import java.time.Duration;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -153,6 +154,47 @@
}
@Test
+ void shouldWalkEveryOtherRosterEndpointOnce() {
+ var first = new ConnectionInfo("iggy-0", 8090);
+ var second = new ConnectionInfo("iggy-1", 8091);
+ var third = new ConnectionInfo("iggy-2", 8092);
+
+ assertThat(LeaderAwareness.rosterWalkTargets(List.of(first, second, third), second, Set.of(second)))
+ .containsExactly(third, first);
+ }
+
+ @Test
+ void shouldOmitAliasesOfTheCurrentEndpointFromRosterWalk() {
+ var current = new ConnectionInfo("localhost", 8090);
+ var alias = new ConnectionInfo("127.0.0.1", 8090);
+ var other = new ConnectionInfo("127.0.0.1", 8091);
+
+ assertThat(LeaderAwareness.rosterWalkTargets(List.of(alias, other), current, Set.of(current)))
+ .containsExactly(other);
+ }
+
+ @Test
+ void shouldNotRevisitAnEndpointAlreadyTriedByTheRequest() {
+ var first = new ConnectionInfo("iggy-0", 8090);
+ var second = new ConnectionInfo("iggy-1", 8091);
+ var third = new ConnectionInfo("iggy-2", 8092);
+
+ assertThat(LeaderAwareness.rosterWalkTargets(List.of(first, second, third), second, Set.of(first, second)))
+ .containsExactly(third);
+ }
+
+ @Test
+ void shouldStopAfterOneCompleteRosterPass() {
+ var first = new ConnectionInfo("iggy-0", 8090);
+ var second = new ConnectionInfo("iggy-1", 8091);
+ var third = new ConnectionInfo("iggy-2", 8092);
+
+ assertThat(LeaderAwareness.rosterWalkTargets(
+ List.of(first, second, third), third, Set.of(first, second, third)))
+ .isEmpty();
+ }
+
+ @Test
void shouldMatchLocalhostAgainstLoopback() {
assertThat(isSameAddress("localhost", 8090, "127.0.0.1", 8090)).isTrue();
}
diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json
index 5f7238a..31539e7 100644
--- a/foreign/node/package-lock.json
+++ b/foreign/node/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "apache-iggy",
- "version": "0.10.0-edge.4",
+ "version": "0.10.0-edge.5",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "apache-iggy",
- "version": "0.10.0-edge.4",
+ "version": "0.10.0-edge.5",
"license": "Apache-2.0",
"dependencies": {
"@node-rs/xxhash": "1.7.7",
diff --git a/foreign/node/package.json b/foreign/node/package.json
index 755019a..17cbfca 100644
--- a/foreign/node/package.json
+++ b/foreign/node/package.json
@@ -1,7 +1,7 @@
{
"name": "apache-iggy",
"type": "module",
- "version": "0.10.0-edge.4",
+ "version": "0.10.0-edge.5",
"description": "Official Apache Iggy NodeJS SDK",
"keywords": [
"iggy",
diff --git a/foreign/node/src/client/client.connection.test.ts b/foreign/node/src/client/client.connection.test.ts
index 5e320b0..1d671eb 100644
--- a/foreign/node/src/client/client.connection.test.ts
+++ b/foreign/node/src/client/client.connection.test.ts
@@ -459,6 +459,33 @@
}
);
+ it('walks each roster endpoint once per request', async () => {
+ const seed = await startServer();
+ const seedPort = (seed.address() as AddressInfo).port;
+ const connection = new IggyConnection(connectionConfig(seed));
+ connection.on('error', () => undefined);
+ try {
+ await connection.connect();
+ connection.rememberRoster([
+ { host: '127.0.0.1', port: seedPort },
+ { host: '127.0.0.1', port: seedPort + 1 },
+ { host: '127.0.0.1', port: seedPort + 2 }
+ ]);
+ const visited = new Set<string>();
+
+ assert.deepEqual(connection.nextRosterEndpoint(visited), {
+ host: '127.0.0.1', port: seedPort + 1
+ });
+ assert.deepEqual(connection.nextRosterEndpoint(visited), {
+ host: '127.0.0.1', port: seedPort + 2
+ });
+ assert.equal(connection.nextRosterEndpoint(visited), undefined);
+ assert.equal(visited.size, 3);
+ } finally {
+ await closeConnection(connection, seed);
+ }
+ });
+
it('dials the endpoint it is on, then the seed, then the roster',
async () => {
const seed = await startServer();
diff --git a/foreign/node/src/client/client.connection.ts b/foreign/node/src/client/client.connection.ts
index 936c6f1..06103cf 100644
--- a/foreign/node/src/client/client.connection.ts
+++ b/foreign/node/src/client/client.connection.ts
@@ -504,6 +504,35 @@
}
/**
+ * The roster endpoint after the one this connection is on, for a request
+ * the current node keeps refusing to admit. Metadata and partition
+ * consensus groups elect independently, so the metadata leader can hold a
+ * follower replica of the partition a request targets, and only walking
+ * the roster reaches that group's primary. `undefined` when the roster
+ * names nowhere else to go.
+ */
+ nextRosterEndpoint(visited: Set<string>): Endpoint | undefined {
+ const roster = this.rosterEndpoints;
+ if (roster.length === 0)
+ return undefined;
+ const index = roster.findIndex(
+ (endpoint) => this.isConnectedTo(endpoint.host, endpoint.port)
+ );
+ if (index >= 0)
+ visited.add(endpointKey(roster[index]));
+ const start = index < 0 ? 0 : (index + 1) % roster.length;
+ for (let offset = 0; offset < roster.length; offset += 1) {
+ const candidate = roster[(start + offset) % roster.length];
+ const key = endpointKey(candidate);
+ if (visited.has(key) || this.isConnectedTo(candidate.host, candidate.port))
+ continue;
+ visited.add(key);
+ return candidate;
+ }
+ return undefined;
+ }
+
+ /**
* Endpoints a redial rotates through, likeliest first: where the client
* currently is, the endpoint it was configured with, then the roster it
* learned while connected. After a leader redirect the current endpoint may
@@ -637,3 +666,6 @@
? '127.0.0.1'
: normalized;
};
+
+export const endpointKey = (endpoint: Endpoint): string =>
+ `${normalizeHost(endpoint.host)}:${endpoint.port}`;
diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts
index 2b1aae1..7506350 100644
--- a/foreign/node/src/client/client.socket.test.ts
+++ b/foreign/node/src/client/client.socket.test.ts
@@ -183,6 +183,38 @@
]);
};
+const threeNodeMetadataBody = (
+ firstPort: number,
+ secondPort: number,
+ thirdPort: number
+): Buffer => {
+ const node = (name: string, port: number, role: number): Buffer => {
+ const nodeName = Buffer.from(name);
+ const ip = Buffer.from('127.0.0.1');
+ const encoded = Buffer.alloc(4 + nodeName.length + 4 + ip.length + 8 + 2);
+ let offset = 0;
+ encoded.writeUInt32LE(nodeName.length, offset); offset += 4;
+ nodeName.copy(encoded, offset); offset += nodeName.length;
+ encoded.writeUInt32LE(ip.length, offset); offset += 4;
+ ip.copy(encoded, offset); offset += ip.length;
+ encoded.writeUInt16LE(port, offset); offset += 8;
+ encoded.writeUInt8(role, offset); offset += 1;
+ encoded.writeUInt8(0, offset);
+ return encoded;
+ };
+ const name = Buffer.from('iggy-cluster');
+ const header = Buffer.alloc(4 + name.length + 4);
+ header.writeUInt32LE(name.length, 0);
+ name.copy(header, 4);
+ header.writeUInt32LE(3, 4 + name.length);
+ return Buffer.concat([
+ header,
+ node('iggy-node-1', firstPort, 0),
+ node('iggy-node-2', secondPort, 1),
+ node('iggy-node-3', thirdPort, 1)
+ ]);
+};
+
/** Register, metadata, and echo behavior of a healthy single VSR node. */
const singleNodeHandler = (port: number): FrameHandler =>
(frame, socket) => {
@@ -1058,6 +1090,75 @@
}
);
+ it('walks past two refusing replicas to the partition primary',
+ async () => {
+ const third = await startVsrServer((frame, socket) => {
+ singleNodeHandler(third.port)(frame, socket);
+ });
+ const second = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ if (frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_040) {
+ socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
+ return;
+ }
+ socket.write(replyFrame(operation));
+ });
+ const metadataLeader = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ if (code === COMMAND_CODE.GetClusterMetadata) {
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ threeNodeMetadataBody(
+ metadataLeader.port,
+ second.port,
+ third.port
+ )
+ ));
+ return;
+ }
+ if (code === 60_040) {
+ socket.write(replyFrame(Operation.NonReplicated, Buffer.alloc(0), 58));
+ return;
+ }
+ socket.write(replyFrame(operation));
+ });
+ const client = new CommandResponseStream(vsrConfig(metadataLeader.port));
+ try {
+ await client.authenticate(vsrConfig(metadataLeader.port).credentials);
+
+ const response = await client.sendCommand(60_040, Buffer.alloc(0), {
+ deadline: Date.now() + 15_000
+ });
+
+ assert.equal(response.status, 0);
+ assert.ok(
+ second.frames.some((frame) =>
+ frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_040),
+ 'the roster walk skipped the second replica'
+ );
+ assert.ok(
+ third.frames.some((frame) =>
+ frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_040),
+ 'the roster walk never reached the partition primary'
+ );
+ } finally {
+ client.destroy();
+ await metadataLeader.close();
+ await second.close();
+ await third.close();
+ }
+ }
+ );
+
it('holds a queued command instead of writing it to the node being left',
async () => {
// A refusal sends its caller to re-read the roster, and the drain that
@@ -1212,6 +1313,44 @@
}
);
+ it('does not suppress leader settlement after a failed roster walk',
+ async () => {
+ const server = await startVsrServer(
+ (frame, socket) => singleNodeHandler(server.port)(frame, socket)
+ );
+ const client = new CommandResponseStream(vsrConfig(server.port));
+ try {
+ await client.authenticate(vsrConfig(server.port).credentials);
+ const routing = client as unknown as {
+ walkSettleSuppressed: boolean,
+ _followLeaderMove: (walkPastLeader: boolean) => Promise<unknown>,
+ connection: {
+ nextRosterEndpoint: () => { host: string, port: number },
+ redirect: (host: string, port: number) => Promise<void>
+ }
+ };
+ routing.connection.nextRosterEndpoint = () => ({
+ host: '127.0.0.1',
+ port: server.port + 1
+ });
+ routing.connection.redirect = async () => {
+ throw new Error('redirect failed');
+ };
+
+ assert.deepEqual(await routing._followLeaderMove(true), {
+ endpoint: { host: '127.0.0.1', port: server.port + 1 },
+ moved: false
+ });
+ assert.equal(routing.walkSettleSuppressed, false,
+ 'a failed walk leaked its one-shot suppression into a later login'
+ );
+ } finally {
+ client.destroy();
+ await server.close();
+ }
+ }
+ );
+
it('surfaces the refusal when the move left too little of the budget',
async () => {
// The move itself costs budget: a roster read, an election it waited out,
diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts
index 06670e5..7e47a03 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -25,7 +25,11 @@
} from '../client/client.type.js';
import { ResponseError, responseError } from '../wire/error.utils.js';
import { debug } from './client.debug.js';
-import { type Endpoint, IggyConnection } from './client.connection.js';
+import {
+ endpointKey,
+ type Endpoint,
+ IggyConnection
+} from './client.connection.js';
import { LOGIN, LOGIN_WITH_TOKEN, LOGOUT, PING } from '../wire/index.js';
import { GET_CLUSTER_METADATA } from '../wire/cluster/get-cluster-metadata.command.js';
import { COMMAND_CODE } from '../wire/command.code.js';
@@ -73,6 +77,17 @@
}
/**
+ * How a leader move resolved: onto the metadata leader, onto the next roster
+ * node (for a request the metadata leader itself keeps refusing), or not at
+ * all.
+ */
+type RosterWalkVerdict = {
+ endpoint: Endpoint,
+ moved: boolean
+};
+type LeaderMoveVerdict = 'leader' | RosterWalkVerdict | false;
+
+/**
* Command codes that can be executed without authentication.
*/
const UNLOGGED_COMMAND_CODE = [
@@ -136,7 +151,13 @@
* The leader re-check a refused request started, shared with every other
* request refused by the same node so one demotion moves the client once.
*/
- private leaderMoveInFlight?: Promise<boolean>;
+ private leaderMoveInFlight?: Promise<LeaderMoveVerdict>;
+ /**
+ * Set when a roster walk just redirected this client, so the login that
+ * re-authenticates it stays on the dialed node instead of settling back on
+ * the metadata leader whose partition replica refused the request.
+ */
+ private walkSettleSuppressed = false;
/**
* Refusals handed out to callers that have not decided what to do with them
* yet. The queue holds while any are outstanding: the caller of a refused
@@ -287,6 +308,8 @@
// opening a second one.
const deadline = options.deadline ?? Date.now() + VSR_RESPONSE_TIMEOUT_MS;
let response: CommandResponse;
+ let walkingRoster = false;
+ const visitedRosterEndpoints = new Set<string>();
for (;;) {
try {
response = await this._queueCommand(command, payload, handleResponse,
@@ -304,11 +327,22 @@
// refusal the server actually gave: re-issued into what is left, the
// request would time out instead and the caller would see a timeout
// where the answer was "not admitted".
- let moved = false;
+ let moved: LeaderMoveVerdict = false;
try {
if (!followsLeaderMoves || !worthAnotherAttempt(deadline))
throw responseError(command, error.refusal.errorCode);
- moved = await this._followLeaderMove();
+ // Once this request starts walking the roster it keeps walking: a
+ // leader recheck between hops would put it straight back on the
+ // metadata leader whose partition replica refused it, and the
+ // walk would bounce between two nodes without reaching the rest.
+ moved = await this._followLeaderMove(
+ walkingRoster,
+ visitedRosterEndpoints
+ );
+ if (typeof moved === 'object') {
+ walkingRoster = true;
+ visitedRosterEndpoints.add(endpointKey(moved.endpoint));
+ }
} finally {
// Released as soon as the move is decided, before the pace below
// and before any re-authentication: those go through the queue
@@ -317,7 +351,9 @@
if (followsLeaderMoves)
this._releaseUndecidedMove();
}
- if (!moved) {
+ const connectionMoved = moved === 'leader' ||
+ (typeof moved === 'object' && moved.moved);
+ if (!connectionMoved) {
// Nowhere else to go yet: the roster still names this node, or it
// could not be read. Paced, because the in-connection replay
// window belongs to the request's budget and has already been
@@ -339,6 +375,13 @@
}
if (!isLoginCommand(command) || this.settlingLeader)
return response;
+ // A login that re-authenticates a roster walk stays on the dialed node:
+ // settling would put the client back on the metadata leader whose
+ // partition replica refused the walked request. One login only.
+ if (this.walkSettleSuppressed) {
+ this.walkSettleSuppressed = false;
+ return response;
+ }
this.settlingLeader = true;
try {
const settled = await this._settleOnLeader(command, payload);
@@ -393,19 +436,50 @@
*
* @returns Whether the client moved
*/
- private _followLeaderMove(): Promise<boolean> {
+ private _followLeaderMove(
+ walkPastLeader = false,
+ visitedRosterEndpoints = new Set<string>()
+ ): Promise<LeaderMoveVerdict> {
const inFlight = this.leaderMoveInFlight;
if (inFlight)
return inFlight;
- const move = (async () => {
+ const move = (async (): Promise<LeaderMoveVerdict> => {
try {
- const leader = await this._readLeaderEndpoint();
- if (!leader || this.connection.isConnectedTo(leader.host, leader.port))
+ if (!walkPastLeader) {
+ const leader = await this._readLeaderEndpoint();
+ if (leader && !this.connection.isConnectedTo(leader.host, leader.port)) {
+ debug(`the leader moved to ${leader.host}:${leader.port}, following it`);
+ await this.connection.redirect(leader.host, leader.port);
+ return 'leader';
+ }
+ }
+ // The roster names this node as the metadata leader (or said nothing
+ // usable), yet it keeps refusing to admit the request: its replica of
+ // the target partition group is not that group's primary, because
+ // metadata and partition consensus groups elect independently. Walk
+ // the roster instead of re-issuing into the same refusal.
+ const next = this.connection.nextRosterEndpoint(visitedRosterEndpoints);
+ if (!next)
return false;
- debug(`the leader moved to ${leader.host}:${leader.port}, following it`);
- await this.connection.redirect(leader.host, leader.port);
- return true;
+ debug(
+ 'the request keeps being refused here, walking the roster to ' +
+ `${next.host}:${next.port}`
+ );
+ // The re-authentication after this redirect runs a login, and a login
+ // normally settles on the metadata leader, which would put the walk
+ // right back on the node that refused. One login only.
+ this.walkSettleSuppressed = true;
+ try {
+ await this.connection.redirect(next.host, next.port);
+ } catch (error) {
+ // The suppression belongs to the redirect above. If that redirect
+ // never lands, a later unrelated login must settle normally.
+ this.walkSettleSuppressed = false;
+ debug('the roster endpoint could not be reached', error);
+ return { endpoint: next, moved: false };
+ }
+ return { endpoint: next, moved: true };
} catch (error) {
debug('the leader could not be re-checked, staying on this node', error);
return false;
diff --git a/foreign/php/Cargo.toml b/foreign/php/Cargo.toml
index 1c8245e..5850259 100644
--- a/foreign/php/Cargo.toml
+++ b/foreign/php/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-php"
-version = "0.1.1"
+version = "0.1.2"
edition = "2024"
authors = ["Iggy Committers <dev@iggy.apache.org>"]
license = "Apache-2.0"
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index 9446a1c..eefeaa1 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "apache-iggy"
-version = "0.9.0-dev4"
+version = "0.9.0-dev5"
edition = "2024"
authors = ["Iggy Committers <dev@iggy.apache.org>"]
license = "Apache-2.0"
@@ -37,7 +37,7 @@
[dependencies]
bytes = "1.12.1"
futures = "0.3.33"
-iggy = { path = "../../core/sdk", version = "0.11.0-edge.2" }
+iggy = { path = "../../core/sdk", version = "0.11.0-edge.5" }
paste = "1"
pyo3 = "0.29.0"
pyo3-async-runtimes = { version = "0.29.0", features = [
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index b7c0d2f..7a82108 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@
[project]
name = "apache-iggy"
requires-python = ">=3.10"
-version = "0.9.0.dev4"
+version = "0.9.0.dev5"
description = "Apache Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
readme = "README.md"
license = { file = "LICENSE" }
diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock
index c78536a..963b58a 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock
@@ -8,7 +8,7 @@
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev4"
+version = "0.9.0.dev5"
source = { editable = "." }
[package.optional-dependencies]
