You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/08/28 12:03:28 UTC
(iggy) 01/01: fix(cluster): recover writes after a fast partition primary rejoin
This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch recover_writes
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 21ab39636b1d82d9eae487754a2fb3b697ea7510
Author: spetz <pi...@gmail.com>
AuthorDate: Fri Aug 28 14:03:14 2026 +0200
fix(cluster): recover writes after a fast partition primary rejoin
---
Cargo.lock | 18 +-
Cargo.toml | 8 +-
bdd/python/uv.lock | 2 +-
core/bench/Cargo.toml | 2 +-
core/binary_protocol/Cargo.toml | 2 +-
core/cli/Cargo.toml | 2 +-
core/common/Cargo.toml | 2 +-
core/connectors/runtime/Cargo.toml | 2 +-
.../tests/cluster/fast_primary_rejoin.rs | 368 +++++++++++++++++++++
core/integration/tests/cluster/mod.rs | 1 +
core/integration/tests/server/mod.rs | 2 +-
core/partitions/src/iggy_partition.rs | 87 +++--
core/sdk/Cargo.toml | 2 +-
core/sdk/src/leader_aware.rs | 313 ++++++++++++++++++
core/sdk/src/quic/quic_client.rs | 233 +++++++++++--
core/sdk/src/tcp/tcp_client.rs | 321 ++++++++++++++----
core/sdk/src/vsr.rs | 35 +-
core/sdk/src/websocket/websocket_client.rs | 265 +++++++++++++--
core/server/Cargo.toml | 2 +-
core/server/src/http.rs | 37 ++-
core/server/src/http/forward.rs | 201 ++++++++++-
core/shard/src/lib.rs | 54 ++-
examples/node/package-lock.json | 2 +-
examples/python/uv.lock | 2 +-
foreign/cpp/Cargo.toml | 2 +-
foreign/cpp/MODULE.bazel | 2 +-
.../Implementations/TcpMessageStream.Vsr.cs | 85 ++++-
.../IggyClient/Implementations/TcpMessageStream.cs | 17 +-
foreign/csharp/Iggy_SDK/Iggy_SDK.csproj | 2 +-
.../VsrTests/EndpointFailoverTests.cs | 128 ++++++-
foreign/go/client/tcp/tcp_core.go | 47 ++-
foreign/go/client/tcp/tcp_failover_test.go | 63 ++++
foreign/go/client/tcp/tcp_session_management.go | 70 ++++
foreign/go/contracts/version.go | 2 +-
.../iggy/client/async/tcp/AsyncIggyTcpClient.java | 85 ++---
.../iggy/client/async/tcp/AsyncTcpConnection.java | 54 ++-
.../iggy/client/async/tcp/LeaderAwareness.java | 33 ++
.../AsyncIggyTcpClientTransientFailoverTest.java | 83 ++++-
.../iggy/client/async/tcp/LeaderAwarenessTest.java | 42 +++
foreign/node/package-lock.json | 4 +-
foreign/node/package.json | 2 +-
foreign/node/src/client/client.connection.test.ts | 27 ++
foreign/node/src/client/client.connection.ts | 32 ++
foreign/node/src/client/client.socket.test.ts | 139 ++++++++
foreign/node/src/client/client.socket.ts | 98 +++++-
foreign/php/Cargo.toml | 2 +-
foreign/python/Cargo.toml | 4 +-
foreign/python/pyproject.toml | 2 +-
foreign/python/uv.lock | 2 +-
49 files changed, 2695 insertions(+), 295 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index ac8d17755..94abb0990 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2742,9 +2742,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"
[[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 @@ checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb"
[[package]]
name = "iggy"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
dependencies = [
"async-broadcast",
"async-dropper",
@@ -6693,7 +6693,7 @@ dependencies = [
[[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 @@ dependencies = [
[[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 @@ dependencies = [
[[package]]
name = "iggy-connectors"
-version = "0.5.0-edge.4"
+version = "0.5.0-edge.5"
dependencies = [
"async-trait",
"axum",
@@ -6890,7 +6890,7 @@ dependencies = [
[[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 @@ dependencies = [
[[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 @@ dependencies = [
[[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 32a274e08..ea2d476ac 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -206,10 +206,10 @@ hyper-util = { version = "0.1.20", features = ["server-auto", "service"] }
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 c0f0560a0..b401a9ae2 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev4"
+version = "0.9.0.dev5"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/core/bench/Cargo.toml b/core/bench/Cargo.toml
index cc5807361..60e521d91 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 4cb4325d9..79e99c768 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 fbdba38ea..bfecb9ef3 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 cdfaa64b3..8055104fe 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 0c04c59b9..37b5351d5 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/tests/cluster/fast_primary_rejoin.rs b/core/integration/tests/cluster/fast_primary_rejoin.rs
new file mode 100644
index 000000000..343476740
--- /dev/null
+++ b/core/integration/tests/cluster/fast_primary_rejoin.rs
@@ -0,0 +1,368 @@
+// 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) -> (IggyClient, String) {
+ let primary_endpoint = harness
+ .node(leader)
+ .tcp_addr()
+ .expect("leader exposes a TCP endpoint")
+ .to_string();
+ let producer = harness
+ .node(leader)
+ .tcp_client()
+ .expect("leader exposes a TCP endpoint")
+ .with_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,
+) -> (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).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)
+}
+
+/// 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,
+) {
+ let (producer, leader, primary_endpoint, mut acked) =
+ warmed_producer_past_follower_restart(harness).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;
+}
+
+/// 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,
+) {
+ let (producer, leader, _primary_endpoint, mut acked) =
+ warmed_producer_past_follower_restart(harness).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;
+}
+
+/// 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).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 05f42ba21..19c1e5b2c 100644
--- a/core/integration/tests/cluster/mod.rs
+++ b/core/integration/tests/cluster/mod.rs
@@ -21,6 +21,7 @@ mod crash_durability;
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 f31ca33a0..c880723c5 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -39,7 +39,7 @@ mod stats_vsr;
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 557af4387..85c4a0906 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -1974,6 +1974,30 @@ where
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 @@ where
}
}
- // 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!(
@@ -5476,13 +5474,20 @@ mod tests {
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 +5590,32 @@ mod tests {
);
}
+ #[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!(
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index e6658fd47..0e3e504b5 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 b5c038f34..6aa67f202 100644
--- a/core/sdk/src/leader_aware.rs
+++ b/core/sdk/src/leader_aware.rs
@@ -20,8 +20,13 @@ use iggy_common::ClusterClient;
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,179 @@ fn normalize_address(addr: &str) -> String {
.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 {
+ active: AtomicBool,
+ abandoned: AtomicBool,
+ generation: AtomicU64,
+ result: StdMutex<Option<(u64, Result<(), IggyError>)>>,
+ changed: Notify,
+}
+
+impl ConnectCoordinator {
+ pub(crate) fn new() -> Self {
+ Self {
+ active: AtomicBool::new(false),
+ abandoned: AtomicBool::new(false),
+ 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) async fn run<F, Fut>(&self, operation: F) -> Result<(), IggyError>
+ where
+ F: FnOnce(bool) -> 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 mut owner = ConnectOwner {
+ coordinator: self,
+ completed: false,
+ };
+ let result = operation(abandoned).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, 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);
+ 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,
+ completed: bool,
+}
+
+impl ConnectOwner<'_> {
+ fn complete(&mut self, result: Result<(), IggyError>) {
+ self.coordinator.finish(result, false);
+ self.completed = true;
+ }
+}
+
+impl Drop for ConnectOwner<'_> {
+ fn drop(&mut self) {
+ if !self.completed {
+ self.coordinator.finish(Err(IggyError::Disconnected), true);
+ }
+ }
+}
+
/// Struct to track leader redirection state
#[derive(Debug, Clone)]
pub struct LeaderRedirectionState {
@@ -378,6 +556,8 @@ impl Default for LeaderRedirectionState {
#[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() {
@@ -517,6 +697,139 @@ mod tests {
assert!(!is_same_address("no-such-host.invalid:8090", "127.0.0.1:8090").await);
}
+ #[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| 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(|_| 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(|_| 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(|_| 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| async move {
+ marker.store(abandoned, Ordering::SeqCst);
+ Ok(())
+ })
+ .await
+ .unwrap();
+ assert!(observed_abandoned.load(Ordering::SeqCst));
+ }
+
#[test]
fn test_normalize_address() {
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 b1f423a37..88b4bb18f 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, 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 crate::quic::skip_server_verification::SkipServerVerification;
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,
@@ -43,6 +47,7 @@ use std::net::{SocketAddr, ToSocketAddrs};
use std::str::FromStr;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
+use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::sleep;
@@ -66,6 +71,12 @@ const RESPONSE_READ_TIMEOUT: Duration = Duration::from_secs(30);
/// 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 +92,18 @@ pub struct QuicClient {
// `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>>,
+ /// Set when the next connect must stay on the endpoint it dialed instead
+ /// of settling on the metadata leader: the roster walk has to survive its
+ /// own sign-in's leader check. See the TCP client's twin field.
+ settle_off_leader_once: AtomicBool,
+ /// 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 +152,60 @@ impl BinaryTransport for QuicClient {
}
async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
- let result = self.send_raw(code, payload.clone()).await;
+ 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 = self.routing_lock.lock().await;
+ // A concurrent refused request may have completed the movement
+ // while this request waited for the gate.
+ result = self.send_raw(code, payload.clone()).await;
+ 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 = matches!(self.handle_leader_redirection().await, Ok(true));
+ let roster = self.roster_endpoints.lock().await.clone();
+ roster_walk = Some(RosterWalk::new(¤t, &roster));
+ redirected
+ };
+ if !redirected {
+ let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) else {
+ break;
+ };
+ self.settle_on_endpoint(next).await?;
+ } else {
+ let target = self.current_server_address.lock().await.clone();
+ if let Some(walk) = roster_walk.as_mut() {
+ walk.record_attempt(&target);
+ }
+ }
+ self.connect().await?;
+ let connected = self.current_server_address.lock().await.clone();
+ if let Some(walk) = roster_walk.as_mut() {
+ walk.record_attempt(&connected);
+ }
+ result = self.send_raw(code, payload.clone()).await;
+ }
+ }
+
if result.is_ok() {
return result;
}
@@ -152,6 +228,10 @@ impl BinaryTransport for QuicClient {
return Err(error);
}
+ if code == GET_CLUSTER_METADATA_CODE {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -166,8 +246,22 @@ impl BinaryTransport for QuicClient {
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 nested_connect = skip_auto_login && self.connect_coordinator.is_active();
+ 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 +270,22 @@ impl BinaryTransport for QuicClient {
"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().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 +404,10 @@ impl QuicClient {
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()),
+ settle_off_leader_once: AtomicBool::new(false),
+ routing_lock: Mutex::new(()),
+ connect_coordinator: ConnectCoordinator::new(),
consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
@@ -337,6 +446,20 @@ impl QuicClient {
}
async fn connect(&self) -> Result<(), IggyError> {
+ self.connect_coordinator
+ .run(|abandoned| async move {
+ if abandoned {
+ self.clear_abandoned_connect().await?;
+ }
+ self.connect_inner().await
+ })
+ .await
+ }
+
+ async fn connect_inner(&self) -> Result<(), IggyError> {
+ // Consume before fallible connection work. A walk target that cannot
+ // connect must not make an unrelated later reconnect skip settlement.
+ let settle_off_leader = self.settle_off_leader_once.swap(false, Ordering::SeqCst);
loop {
match self.get_state().await {
ClientState::Shutdown => {
@@ -494,12 +617,23 @@ impl QuicClient {
}
}
- // 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 +646,35 @@ impl QuicClient {
}
}
+ 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 +702,25 @@ impl QuicClient {
}
}
+ /// 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;
+ self.settle_off_leader_once.store(true, Ordering::SeqCst);
+ Ok(())
+ }
+
async fn shutdown(&self) -> Result<(), IggyError> {
if self.get_state().await == ClientState::Shutdown {
return Ok(());
@@ -649,6 +818,16 @@ impl QuicClient {
// 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 +861,22 @@ impl QuicClient {
{
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 =>
{
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index a271cef8c..634c4ddda 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -16,8 +16,8 @@
// under the License.
use crate::leader_aware::{
- LeaderRedirectionState, check_and_redirect_to_leader, is_same_spelling,
- is_unauthenticated_metadata_probe, read_transport_endpoints,
+ ConnectCoordinator, 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 +25,13 @@ use crate::session::ConsensusSession;
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 +125,17 @@ pub struct TcpClient {
// contention with zero correctness benefit.
consensus_session: Arc<StdMutex<ConsensusSession>>,
skip_auto_login_once: Mutex<bool>,
+ /// Set when the next connect must stay on the endpoint it dialed instead
+ /// of settling on the metadata leader. Metadata and partition consensus
+ /// groups elect independently, so the metadata leader can hold a follower
+ /// replica of the partition a request targets; the failover that walks the
+ /// roster past it has to survive its own sign-in's leader check.
+ settle_off_leader_once: AtomicBool,
+ /// 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>,
}
@@ -182,7 +194,6 @@ impl Client for TcpClient {
}
}
-#[async_trait]
#[async_trait]
impl BinaryTransport for TcpClient {
async fn get_state(&self) -> ClientState {
@@ -223,6 +234,10 @@ impl BinaryTransport for TcpClient {
return Err(error);
}
+ if code == GET_CLUSTER_METADATA_CODE {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -248,11 +263,24 @@ impl BinaryTransport for TcpClient {
// 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);
+ let replay_after_reconnect = replay_after_session_reset_is_safe(code, &error);
+ let skip_auto_login = is_login_register_code(code);
+ let nested_connect = skip_auto_login && self.connect_coordinator.is_active();
+ 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_transport().await?;
- let skip_auto_login = is_login_register_code(code);
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
}
@@ -266,7 +294,11 @@ impl BinaryTransport for TcpClient {
);
}
- let reconnect = self.connect().await;
+ let reconnect = if nested_connect {
+ self.connect_inner().await
+ } else {
+ self.connect().await
+ };
if skip_auto_login && reconnect.is_err() {
*self.skip_auto_login_once.lock().await = false;
}
@@ -293,38 +325,6 @@ impl BinaryTransport for TcpClient {
}
}
-/// 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 +498,28 @@ impl TcpClient {
session_credentials: Mutex::new(None),
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
skip_auto_login_once: Mutex::new(false),
+ settle_off_leader_once: AtomicBool::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_coordinator
+ .run(|abandoned| async move {
+ if abandoned {
+ self.clear_abandoned_connect().await?;
+ }
+ self.connect_inner().await
+ })
+ .await
+ }
+
+ async fn connect_inner(&self) -> Result<(), IggyError> {
+ // Consume before fallible connection work. A walk target that cannot
+ // connect must not make an unrelated later reconnect skip settlement.
+ let settle_off_leader = self.settle_off_leader_once.swap(false, Ordering::SeqCst);
loop {
// Read and claimed under one lock acquisition. Apart, two callers
// both find `Disconnected` and both sweep: the loser's
@@ -548,7 +565,6 @@ impl TcpClient {
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 +611,7 @@ impl TcpClient {
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 +700,21 @@ impl TcpClient {
}
}
+ 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 +747,18 @@ impl TcpClient {
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 +865,32 @@ impl TcpClient {
}
}
+ /// 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;
+ self.settle_off_leader_once.store(true, Ordering::SeqCst);
+ 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 +1075,7 @@ impl TcpClient {
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 +1230,12 @@ impl TcpClient {
// 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 +1243,11 @@ impl TcpClient {
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 +1257,17 @@ impl TcpClient {
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 +1278,59 @@ impl TcpClient {
// 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?;
+ }
+ match self.connect().await {
+ 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 +1528,16 @@ const fn is_login_register_code(code: u32) -> bool {
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 +1550,13 @@ mod tests {
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 +1609,47 @@ mod tests {
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)
+ ));
+ }
+
// 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 +1923,11 @@ mod tests {
#[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 +1935,31 @@ mod tests {
// 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 9625b020c..2e8a21097 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -28,20 +28,19 @@ use iggy_common::{IggyError, calculate_checksum, eviction_reason_to_error};
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 @@ pub(crate) fn operation_for_code(code: u32) -> Operation {
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 56261f6d3..c808d5fca 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, 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 crate::prelude::Client;
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,
@@ -40,11 +44,12 @@ use secrecy::ExposeSecret;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
+use std::sync::atomic::{AtomicBool, Ordering};
use tokio::net::TcpStream;
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 +68,12 @@ const RESPONSE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_sec
/// `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 +88,18 @@ pub struct WebSocketClient {
// `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>>,
+ /// Set when the next connect must stay on the endpoint it dialed instead
+ /// of settling on the metadata leader: the roster walk has to survive its
+ /// own sign-in's leader check. See the TCP client's twin field.
+ settle_off_leader_once: AtomicBool,
+ /// 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 +145,60 @@ impl BinaryTransport for WebSocketClient {
}
async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
- let result = self.send_raw(code, payload.clone()).await;
+ 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 = self.routing_lock.lock().await;
+ // A concurrent refused request may have completed the movement
+ // while this request waited for the gate.
+ result = self.send_raw(code, payload.clone()).await;
+ 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 = matches!(self.handle_leader_redirection().await, Ok(true));
+ let roster = self.roster_endpoints.lock().await.clone();
+ roster_walk = Some(RosterWalk::new(¤t, &roster));
+ redirected
+ };
+ if !redirected {
+ let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) else {
+ break;
+ };
+ self.settle_on_endpoint(next).await?;
+ } else {
+ let target = self.current_server_address.lock().await.clone();
+ if let Some(walk) = roster_walk.as_mut() {
+ walk.record_attempt(&target);
+ }
+ }
+ self.connect().await?;
+ let connected = self.current_server_address.lock().await.clone();
+ if let Some(walk) = roster_walk.as_mut() {
+ walk.record_attempt(&connected);
+ }
+ result = self.send_raw(code, payload.clone()).await;
+ }
+ }
+
if result.is_ok() {
return result;
}
@@ -148,6 +224,10 @@ impl BinaryTransport for WebSocketClient {
return Err(error);
}
+ if code == GET_CLUSTER_METADATA_CODE {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -156,9 +236,23 @@ impl BinaryTransport for WebSocketClient {
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 nested_connect = skip_auto_login && self.connect_coordinator.is_active();
+ 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 +265,22 @@ impl BinaryTransport for WebSocketClient {
);
}
- let reconnect = self.connect().await;
+ let reconnect = if nested_connect {
+ self.connect_inner().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 +351,10 @@ impl WebSocketClient {
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()),
+ settle_off_leader_once: AtomicBool::new(false),
+ routing_lock: Mutex::new(()),
+ connect_coordinator: ConnectCoordinator::new(),
consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
@@ -267,6 +376,20 @@ impl WebSocketClient {
}
async fn connect(&self) -> Result<(), IggyError> {
+ self.connect_coordinator
+ .run(|abandoned| async move {
+ if abandoned {
+ self.clear_abandoned_connect().await?;
+ }
+ self.connect_inner().await
+ })
+ .await
+ }
+
+ async fn connect_inner(&self) -> Result<(), IggyError> {
+ // Consume before fallible connection work. A walk target that cannot
+ // connect must not make an unrelated later reconnect skip settlement.
+ let settle_off_leader = self.settle_off_leader_once.swap(false, Ordering::SeqCst);
loop {
if self.get_state().await == ClientState::Connected {
return Ok(());
@@ -350,12 +473,20 @@ impl WebSocketClient {
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,
@@ -403,6 +534,13 @@ impl WebSocketClient {
server_addr: SocketAddr,
retry_count: &mut u32,
) -> 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).await;
+ }
+ };
let tls_config = self.build_tls_config()?;
let connector = Connector::Rustls(Arc::new(tls_config));
@@ -412,14 +550,19 @@ impl WebSocketClient {
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
@@ -518,7 +661,7 @@ impl WebSocketClient {
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 +670,14 @@ impl WebSocketClient {
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 +691,20 @@ impl WebSocketClient {
/// 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 +730,25 @@ impl WebSocketClient {
}
}
+ /// 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;
+ self.settle_off_leader_once.store(true, Ordering::SeqCst);
+ Ok(())
+ }
+
async fn auto_login(&self) -> Result<(), IggyError> {
let client_address = self.get_client_address_value().await;
let skip_auto_login = {
@@ -671,13 +846,18 @@ impl WebSocketClient {
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 +866,7 @@ impl WebSocketClient {
// 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 +877,16 @@ impl WebSocketClient {
);
// 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 +894,8 @@ impl WebSocketClient {
// 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 +929,19 @@ impl WebSocketClient {
};
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 +952,12 @@ impl WebSocketClient {
other => return other,
}
}
- }
+ })
+ .await
+ .map_err(|error| {
+ error!("Task execution failed during {NAME} request: {error}");
+ IggyError::WebSocketSendError
+ })?
}
}
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index e47cc5258..22d6ba2f0 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 8f49f3023..b8180f726 100644
--- a/core/server/src/http.rs
+++ b/core/server/src/http.rs
@@ -275,13 +275,7 @@ fn router(
.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 @@ fn router(
)
.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 @@ fn router(
};
let router = Router::new()
.merge(forwardable)
+ .merge(partition_writes)
.merge(local)
.with_state(state)
.layer(DefaultBodyLimit::max(max_request_size))
@@ -358,6 +349,26 @@ fn router(
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 5e4572f70..8a153bfbd 100644
--- a/core/server/src/http/forward.rs
+++ b/core/server/src/http/forward.rs
@@ -60,7 +60,6 @@ use std::time::{Duration, Instant};
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 @@ pub(in crate::http) async fn forward_to_primary(
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 @@ async fn forward_or_pass(state: HttpState, request: Request, next: Next) -> Resp
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 @@ async fn forward(state: &HttpInner, request: Request) -> Response {
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 @@ enum AttemptOutcome {
/// 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 @@ async fn attempt(state: &HttpInner, parts: &Parts, body: &Bytes, url: &str) -> A
}
};
let request = builder
- .headers(forwarded_headers(&parts.headers))
+ .headers(forwarded_headers(request_headers))
.body(body.clone())
.build();
let attempt = async {
@@ -403,7 +487,7 @@ async fn attempt(state: &HttpInner, parts: &Parts, body: &Bytes, url: &str) -> A
}
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 @@ fn classify_reply(
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 @@ fn classify_reply(
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 @@ fn primary_socket(state: &HttpInner) -> Option<SocketAddr> {
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 @@ impl ServerCertVerifier for PinnedCertVerifier {
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 @@ mod tests {
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/shard/src/lib.rs b/core/shard/src/lib.rs
index 358b8e62b..22fa69015 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -4349,7 +4349,17 @@ where
);
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
@@ -6508,6 +6518,19 @@ where
let commit_min = partition.consensus().commit_min();
let cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
+ if partition
+ .repair
+ .is_some_and(|session| commit_min >= session.to_op)
+ {
+ partition.repair = None;
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ commit_min,
+ "partition journal repair completed after the commit frontier advanced"
+ );
+ continue;
+ }
partition.repair.as_mut().and_then(|session| {
if !consensus_normal {
return None;
@@ -7433,16 +7456,33 @@ where
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 head = consensus.sequencer().current_sequence();
+ let commit_lag = consensus.commit_min() < consensus.commit_max();
+ let missing_suffix = consensus.commit_max().checked_add(1).is_some_and(|first| {
+ (first..=head).any(|op| partition.log.journal().inner.header_by_op(op).is_none())
+ });
+ 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 to_op = if missing_suffix {
+ head
+ } else {
+ consensus.commit_max()
+ };
let cluster = consensus.cluster();
let self_id = consensus.replica();
let namespace = consensus.group();
diff --git a/examples/node/package-lock.json b/examples/node/package-lock.json
index c798107e0..ab1a93856 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 b52ecc40e..1f9c5b313 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[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 d1f5afc2b..032532730 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 86706c0da..f274abe1d 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 e100fcc54..d0c52f654 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 @@ public sealed partial class TcpMessageStream : ISessionGenerationProvider
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 @@ public sealed partial class TcpMessageStream : ISessionGenerationProvider
&& 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}",
@@ -601,6 +618,64 @@ public sealed partial class TcpMessageStream : ISessionGenerationProvider
}
}
+ /// <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
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index 245299d09..928d5e9ff 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
@@ -608,7 +608,7 @@ public sealed partial class TcpMessageStream : IIggyClient
"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 />
@@ -816,6 +816,15 @@ public sealed partial class TcpMessageStream : IIggyClient
/// caller is about to replace.
/// </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
@@ -847,7 +856,7 @@ public sealed partial class TcpMessageStream : IIggyClient
}
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 @@ public sealed partial class TcpMessageStream : IIggyClient
};
}
- 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 @@ public sealed partial class TcpMessageStream : IIggyClient
{
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 158ea6d8e..baa0fe580 100644
--- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
+++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
@@ -26,7 +26,7 @@ under the License.
<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 e0e020397..3f6ad316c 100644
--- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
@@ -55,6 +55,7 @@ public sealed class EndpointFailoverTests
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 @@ public sealed class EndpointFailoverTests
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
@@ -370,12 +466,17 @@ public sealed class EndpointFailoverTests
}
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 @@ public sealed class EndpointFailoverTests
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 057907d61..0bd8b3600 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -488,12 +488,21 @@ type connectScoped struct{}
// 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 @@ func (c *IggyTcpClient) sendFrame(ctx context.Context, code uint32, frame []byte
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 @@ func (c *IggyTcpClient) sendFrame(ctx context.Context, code uint32, frame []byte
// 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 e7946a462..99482ad33 100644
--- a/foreign/go/client/tcp/tcp_failover_test.go
+++ b/foreign/go/client/tcp/tcp_failover_test.go
@@ -101,6 +101,69 @@ func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) {
"the remembered credentials signed in again on the survivor")
}
+func TestFailover_WalksPastTwoRefusingReplicasToThePartitionPrimary(t *testing.T) {
+ var metadataLeader *testListener
+ var follower *testListener
+ var partitionPrimary *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 4da1fb23b..a34d05565 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -162,6 +162,15 @@ func (c *IggyTcpClient) signIn(ctx context.Context, code uint32, body []byte) (*
//
// 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 @@ func (c *IggyTcpClient) HandleLeaderRedirection(ctx context.Context) (bool, erro
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 9f51ca38c..32ff7a282 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 3f6ebd0d4..d65622f53 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 @@ public class AsyncIggyTcpClient {
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 @@ public class AsyncIggyTcpClient {
.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 @@ public class AsyncIggyTcpClient {
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 @@ public class AsyncIggyTcpClient {
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 @@ public class AsyncIggyTcpClient {
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 @@ public class AsyncIggyTcpClient {
}
/**
- * 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 48d70abb4..99062261c 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.SslContextBuilder;
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 javax.net.ssl.SSLException;
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 @@ public class AsyncTcpConnection {
}
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 @@ public class AsyncTcpConnection {
failoverPayload,
responseFuture,
callerFuture,
- requestDeadlineNanos);
+ requestDeadlineNanos,
+ failoverState);
});
return callerFuture;
@@ -380,9 +389,17 @@ public class AsyncTcpConnection {
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;
@@ -404,7 +421,8 @@ public class AsyncTcpConnection {
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 +437,7 @@ public class AsyncTcpConnection {
requestDeadlineNanos,
holdLeaseUntilResponse,
callerFuture,
+ failoverState,
response,
error));
authenticationStep(channel, commandCode, requestDeadlineNanos)
@@ -483,6 +502,7 @@ public class AsyncTcpConnection {
long requestDeadlineNanos,
boolean holdLeaseUntilResponse,
CompletableFuture<ByteBuf> callerFuture,
+ TransientFailoverState failoverState,
ByteBuf response,
Throwable error) {
try {
@@ -493,6 +513,7 @@ public class AsyncTcpConnection {
failoverPayload,
requestDeadlineNanos,
callerFuture,
+ failoverState,
response,
error);
} finally {
@@ -510,6 +531,7 @@ public class AsyncTcpConnection {
ByteBuf failoverPayload,
long requestDeadlineNanos,
CompletableFuture<ByteBuf> callerFuture,
+ TransientFailoverState failoverState,
ByteBuf response,
Throwable error) {
try {
@@ -522,7 +544,7 @@ public class AsyncTcpConnection {
completeWithResponse(callerFuture, response);
return;
}
- completeFailedRequest(commandCode, failoverPayload, requestDeadlineNanos, callerFuture, error);
+ completeFailedRequest(commandCode, failoverPayload, requestDeadlineNanos, callerFuture, failoverState, error);
}
private void completeFailedRequest(
@@ -530,6 +552,7 @@ public class AsyncTcpConnection {
ByteBuf failoverPayload,
long requestDeadlineNanos,
CompletableFuture<ByteBuf> callerFuture,
+ TransientFailoverState failoverState,
Throwable error) {
IggyTimeoutException timeout = findResponseTimeout(error);
if (timeout != null) {
@@ -537,7 +560,8 @@ public class AsyncTcpConnection {
}
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 +581,12 @@ public class AsyncTcpConnection {
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 +979,15 @@ public class AsyncTcpConnection {
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 +995,8 @@ public class AsyncTcpConnection {
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 c316733a2..4500ee218 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 org.slf4j.LoggerFactory;
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;
@@ -164,6 +166,37 @@ final class LeaderAwareness {
.toList();
}
+ /**
+ * 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.
*/
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 897c5e476..b4ea3b57b 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 @@ class AsyncIggyTcpClientTransientFailoverTest {
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)) {
@@ -108,6 +108,77 @@ class AsyncIggyTcpClientTransientFailoverTest {
}
}
+ @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();
@@ -481,6 +552,16 @@ class AsyncIggyTcpClientTransientFailoverTest {
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 a847acb1c..efae342bf 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 org.junit.jupiter.api.Test;
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;
@@ -152,6 +153,47 @@ class LeaderAwarenessTest {
assertThat(isSameAddress("127.0.0.1", 8090, "127.0.0.1", 8090)).isTrue();
}
+ @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 5f7238ab1..31539e715 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 755019ae0..17cbfca34 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 b1269d19d..bfa2ebe2d 100644
--- a/foreign/node/src/client/client.connection.test.ts
+++ b/foreign/node/src/client/client.connection.test.ts
@@ -418,6 +418,33 @@ describe('IggyConnection', () => {
}
);
+ 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 49f7752f8..81e85192d 100644
--- a/foreign/node/src/client/client.connection.ts
+++ b/foreign/node/src/client/client.connection.ts
@@ -482,6 +482,35 @@ export class IggyConnection extends EventEmitter {
this.rosterEndpoints = endpoints;
}
+ /**
+ * 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
@@ -616,3 +645,6 @@ const normalizeHost = (host?: string): string => {
? '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 2b1aae15f..750635092 100644
--- a/foreign/node/src/client/client.socket.test.ts
+++ b/foreign/node/src/client/client.socket.test.ts
@@ -183,6 +183,38 @@ const twoNodeMetadataBody = (
]);
};
+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 @@ describe('VSR client socket', () => {
}
);
+ 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 @@ describe('VSR client socket', () => {
}
);
+ 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 3924ba14a..4883839b6 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -24,7 +24,11 @@ import type {
} 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';
@@ -71,6 +75,17 @@ class LeaderMovedError extends Error {
}
}
+/**
+ * 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.
*/
@@ -135,7 +150,13 @@ export class CommandResponseStream extends EventEmitter {
* 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
@@ -286,6 +307,8 @@ export class CommandResponseStream extends EventEmitter {
// 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,
@@ -303,11 +326,22 @@ export class CommandResponseStream extends EventEmitter {
// 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
@@ -316,7 +350,9 @@ export class CommandResponseStream extends EventEmitter {
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
@@ -338,6 +374,13 @@ export class CommandResponseStream extends EventEmitter {
}
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);
@@ -392,19 +435,50 @@ export class CommandResponseStream extends EventEmitter {
*
* @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 1c8245e3e..58502591d 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 <de...@iggy.apache.org>"]
license = "Apache-2.0"
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index 9446a1ce3..eefeaa1c6 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 <de...@iggy.apache.org>"]
license = "Apache-2.0"
@@ -37,7 +37,7 @@ doc = false
[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 b7c0d2f9f..7a82108b6 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@ build-backend = "maturin"
[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 c78536a36..963b58acf 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev4"
+version = "0.9.0.dev5"
source = { editable = "." }
[package.optional-dependencies]