You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/21 09:53:54 UTC
(iggy) 01/01: feat(sdk): fail over to a surviving node when the current one dies
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch multi_endpoint_failover
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 99f5adf7e2f3d7f3daf18749b32118a9ceda6211
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Fri Aug 21 11:53:42 2026 +0200
feat(sdk): fail over to a surviving node when the current one dies
---
.../traits/binary_impls/personal_access_tokens.rs | 9 +-
core/common/src/traits/binary_impls/users.rs | 10 +-
core/common/src/traits/binary_transport.rs | 12 +-
.../configuration/tcp_config/tcp_client_config.rs | 11 +
.../tcp_config/tcp_client_config_builder.rs | 38 +++
.../tests/cluster/failover_client_continuity.rs | 47 ++-
core/sdk/src/client_provider.rs | 3 +
core/sdk/src/leader_aware.rs | 113 ++++++-
core/sdk/src/quic/quic_client.rs | 5 +-
core/sdk/src/tcp/tcp_client.rs | 253 +++++++++++++--
core/sdk/src/websocket/websocket_client.rs | 5 +-
.../Implementations/TcpMessageStream.Vsr.cs | 21 ++
.../IggyClient/Implementations/TcpMessageStream.cs | 129 ++++++--
.../Iggy_SDK_Tests/VsrTests/DialCandidatesTests.cs | 64 ++++
.../VsrTests/EndpointFailoverTests.cs | 346 +++++++++++++++++++++
foreign/go/client/tcp/tcp_core.go | 149 ++++++---
foreign/go/client/tcp/tcp_failover_test.go | 155 +++++++++
.../go/client/tcp/tcp_session_credentials_test.go | 66 ++++
foreign/go/client/tcp/tcp_session_management.go | 15 +-
.../iggy/client/async/tcp/AsyncIggyTcpClient.java | 74 ++++-
.../iggy/client/async/tcp/LeaderAwareness.java | 53 +++-
.../iggy/client/async/tcp/LoginRoutingHook.java | 6 +
.../iggy/client/async/tcp/ReconnectPlan.java | 19 +-
.../iggy/client/async/tcp/UsersTcpClient.java | 1 +
.../AsyncIggyTcpClientEndpointFailoverTest.java | 307 ++++++++++++++++++
.../iggy/client/async/tcp/LeaderAwarenessTest.java | 26 +-
.../iggy/client/async/tcp/ReconnectPlanTest.java | 24 +-
foreign/node/src/client/client.connection.test.ts | 25 ++
foreign/node/src/client/client.connection.ts | 94 ++++--
foreign/node/src/client/client.socket.test.ts | 105 +++++++
foreign/node/src/client/client.socket.ts | 7 +
31 files changed, 2001 insertions(+), 191 deletions(-)
diff --git a/core/common/src/traits/binary_impls/personal_access_tokens.rs b/core/common/src/traits/binary_impls/personal_access_tokens.rs
index 7e1299f28..d46fef8ba 100644
--- a/core/common/src/traits/binary_impls/personal_access_tokens.rs
+++ b/core/common/src/traits/binary_impls/personal_access_tokens.rs
@@ -18,8 +18,9 @@
use crate::traits::binary_auth::fail_if_not_authenticated;
use crate::wire_conversions::personal_access_tokens_from_wire;
use crate::{
- BinaryClient, ClientState, DiagnosticEvent, IdentityInfo, IggyError, PersonalAccessTokenClient,
- PersonalAccessTokenExpiry, PersonalAccessTokenInfo, RawPersonalAccessToken,
+ BinaryClient, ClientState, Credentials, DiagnosticEvent, IdentityInfo, IggyError,
+ PersonalAccessTokenClient, PersonalAccessTokenExpiry, PersonalAccessTokenInfo,
+ RawPersonalAccessToken,
};
use iggy_binary_protocol::MAX_WIRE_NAME_LENGTH;
use iggy_binary_protocol::WireName;
@@ -134,6 +135,10 @@ impl<B: BinaryClient> PersonalAccessTokenClient for B {
"authenticated against iggy server"
);
self.set_state(ClientState::Authenticated).await;
+ self.remember_session_credentials(Credentials::PersonalAccessToken(SecretString::from(
+ token.to_string(),
+ )))
+ .await;
self.publish_event(DiagnosticEvent::SignedIn).await;
Ok(IdentityInfo {
user_id: wire_resp.user_id,
diff --git a/core/common/src/traits/binary_impls/users.rs b/core/common/src/traits/binary_impls/users.rs
index eb785109a..c199f76de 100644
--- a/core/common/src/traits/binary_impls/users.rs
+++ b/core/common/src/traits/binary_impls/users.rs
@@ -18,8 +18,8 @@
use crate::traits::binary_auth::fail_if_not_authenticated;
use crate::wire_conversions::{identifier_to_wire, permissions_to_wire, users_from_wire};
use crate::{
- BinaryClient, ClientState, DiagnosticEvent, Identifier, IdentityInfo, IggyError, Permissions,
- UserClient, UserInfo, UserInfoDetails, UserStatus, UserUpdateOptions,
+ BinaryClient, ClientState, Credentials, DiagnosticEvent, Identifier, IdentityInfo, IggyError,
+ Permissions, UserClient, UserInfo, UserInfoDetails, UserStatus, UserUpdateOptions,
};
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::codes::LOGIN_REGISTER_CODE;
@@ -218,6 +218,11 @@ impl<B: BinaryClient> UserClient for B {
"authenticated against iggy server"
);
self.set_state(ClientState::Authenticated).await;
+ self.remember_session_credentials(Credentials::UsernamePassword(
+ username.to_owned(),
+ SecretString::from(password.to_string()),
+ ))
+ .await;
self.publish_event(DiagnosticEvent::SignedIn).await;
Ok(IdentityInfo {
user_id: wire_resp.user_id,
@@ -229,6 +234,7 @@ impl<B: BinaryClient> UserClient for B {
fail_if_not_authenticated(self).await?;
self.send_raw_with_response(LOGOUT_USER_CODE, LogoutUserRequest.to_bytes())
.await?;
+ self.forget_session_credentials().await;
self.reset_vsr_session().await?;
self.set_state(ClientState::Connected).await;
self.publish_event(DiagnosticEvent::SignedOut).await;
diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs
index d3c2bd2e3..9c6a923b6 100644
--- a/core/common/src/traits/binary_transport.rs
+++ b/core/common/src/traits/binary_transport.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{ClientState, DiagnosticEvent, IggyDuration, IggyError};
+use crate::{ClientState, Credentials, DiagnosticEvent, IggyDuration, IggyError};
use async_trait::async_trait;
use bytes::Bytes;
use std::sync::Arc;
@@ -51,6 +51,16 @@ mod vsr_session_sealed {
pub trait VsrSessionControl: vsr_session_sealed::Sealed + BinaryTransport {
async fn bind_vsr_session(&self, session: u64) -> Result<(), IggyError>;
async fn reset_vsr_session(&self) -> Result<(), IggyError>;
+ /// Keep the credentials a sign-in succeeded with, so a transport that
+ /// loses its connection can re-establish the session -- on this node or,
+ /// after failing over, on another one. A caller that signs in by hand is
+ /// otherwise less reconnectable than one that configures `AutoLogin`,
+ /// which is a surprising difference between two ways of doing the same
+ /// thing. Transports that cannot reconnect leave this a no-op.
+ async fn remember_session_credentials(&self, _credentials: Credentials) {}
+ /// Drop them: after an explicit logout there is no session to restore,
+ /// and a reconnect must not resurrect one.
+ async fn forget_session_credentials(&self) {}
/// SDK crate version sent in the login-register version prefix.
/// Implemented by the transports so the value is the SDK crate's own
/// `CARGO_PKG_VERSION` (`iggy` for Rust), not `iggy_common`'s.
diff --git a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
index 86997c6d7..e3584130f 100644
--- a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
+++ b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs
@@ -26,6 +26,12 @@ use std::str::FromStr;
pub struct TcpClientConfig {
/// The address of the Iggy server.
pub server_address: String,
+ /// Addresses of other nodes of the same cluster, dialed in order when
+ /// `server_address` cannot be reached. The roster the server reports is
+ /// remembered while the client is connected and dialed first, so these
+ /// seeds only have to be enough to reach the cluster once -- at the very
+ /// first connect, when nothing has been learned yet.
+ pub failover_addresses: Vec<String>,
/// Whether to use TLS when connecting to the server.
pub tls_enabled: bool,
/// The domain to use for TLS when connecting to the server.
@@ -49,6 +55,7 @@ impl Default for TcpClientConfig {
fn default() -> TcpClientConfig {
TcpClientConfig {
server_address: "127.0.0.1:8090".to_string(),
+ failover_addresses: Vec::new(),
tls_enabled: false,
tls_domain: "".to_string(),
tls_ca_file: None,
@@ -65,6 +72,10 @@ impl From<ConnectionString<TcpConnectionStringOptions>> for TcpClientConfig {
fn from(connection_string: ConnectionString<TcpConnectionStringOptions>) -> Self {
TcpClientConfig {
server_address: connection_string.server_address().into(),
+ // The connection-string grammar names a single host, so a client
+ // built from one starts with no seeds and learns the roster once
+ // it is connected.
+ failover_addresses: Vec::new(),
auto_login: connection_string.auto_login().to_owned(),
tls_enabled: connection_string.options().tls_enabled(),
tls_domain: connection_string.options().tls_domain().into(),
diff --git a/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs b/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs
index 6f665a577..607573279 100644
--- a/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs
+++ b/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs
@@ -20,6 +20,7 @@ use crate::{AutoLogin, IggyDuration, IggyError, TcpClientConfig, validate_server
/// Builder for the TCP client configuration.
/// Allows configuring the TCP client with custom settings or using defaults:
/// - `server_address`: Default is "127.0.0.1:8090"
+/// - `failover_addresses`: Default is empty.
/// - `auto_login`: Default is AutoLogin::Disabled.
/// - `reconnection`: Default is enabled unlimited retries and 1 second interval.
/// - `tls_enabled`: Default is false.
@@ -41,6 +42,13 @@ impl TcpClientConfigBuilder {
self
}
+ /// Sets the addresses of other nodes of the same cluster, dialed in order
+ /// when `server_address` cannot be reached.
+ pub fn with_failover_addresses(mut self, failover_addresses: Vec<String>) -> Self {
+ self.config.failover_addresses = failover_addresses;
+ self
+ }
+
/// Sets the auto sign in during connection.
pub fn with_auto_sign_in(mut self, auto_sign_in: AutoLogin) -> Self {
self.config.auto_login = auto_sign_in;
@@ -105,6 +113,10 @@ impl TcpClientConfigBuilder {
pub fn build(mut self) -> Result<TcpClientConfig, IggyError> {
self.config.server_address = self.config.server_address.trim().to_owned();
validate_server_address(&self.config.server_address)?;
+ for failover_address in &mut self.config.failover_addresses {
+ *failover_address = failover_address.trim().to_owned();
+ validate_server_address(failover_address)?;
+ }
Ok(self.config)
}
@@ -182,6 +194,32 @@ mod tests {
));
}
+ #[test]
+ fn valid_failover_addresses_should_succeed() {
+ let config = builder_with_address("127.0.0.1:8090")
+ .with_failover_addresses(vec![
+ " 127.0.0.1:8091 ".to_string(),
+ "iggy-server-3:8090".to_string(),
+ ])
+ .build()
+ .expect("build the configuration");
+
+ assert_eq!(
+ config.failover_addresses,
+ vec!["127.0.0.1:8091", "iggy-server-3:8090"]
+ );
+ }
+
+ #[test]
+ fn malformed_failover_address_should_fail() {
+ let builder = builder_with_address("127.0.0.1:8090")
+ .with_failover_addresses(vec!["127.0.0.1".to_string()]);
+ assert!(matches!(
+ builder.build(),
+ Err(IggyError::InvalidIpAddress(_, _))
+ ));
+ }
+
#[test]
fn docker_compose_service_name_should_succeed() {
let builder = builder_with_address("iggy-server:8090");
diff --git a/core/integration/tests/cluster/failover_client_continuity.rs b/core/integration/tests/cluster/failover_client_continuity.rs
index ab4c93f82..9eeed6c0d 100644
--- a/core/integration/tests/cluster/failover_client_continuity.rs
+++ b/core/integration/tests/cluster/failover_client_continuity.rs
@@ -15,19 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-//! RED SPEC, expected to FAIL: client continuity across a primary SIGKILL.
+//! Client continuity across a primary SIGKILL.
//!
//! A producing SDK client pinned to the primary must, after the primary dies,
//! complete its next operation against the surviving quorum within a small
-//! budget and without an authentication error. The SDK cannot: it has no
-//! multi-endpoint failover. A client is built around a single
-//! `server_address`, so it knows no other endpoint to dial; the transport's
-//! fail-fast gate (auto-login disabled, the shape this harness client runs
-//! with) returns errors without attempting a reconnect; and the
-//! leader-redirect machinery that could reroute it needs a live connection to
-//! read the cluster roster. Every retry therefore redials the dead endpoint
-//! and fails with a connection error until the caller gives up. Surviving a
-//! primary crash needs a seed roster of endpoints, not just a redirect.
+//! budget and without an authentication error. Three separate pieces of
+//! client state make that possible, and the test fails if any one of them is
+//! lost: the endpoints the cluster roster named while the connection was
+//! healthy (the roster is unreachable exactly when it is needed), the
+//! credentials the sign-in succeeded with (this harness client signs in by
+//! hand rather than configuring `AutoLogin`, and a reconnect has to
+//! re-establish the session on whichever node answers), and a reconnect that
+//! dials those endpoints in turn instead of redialing the address the client
+//! was configured with.
use std::time::Duration;
@@ -64,8 +64,6 @@ fn build_message(payload: &str) -> IggyMessage {
/// A producing client pinned to the primary; SIGKILL the primary mid-stream;
/// the same client's next send must succeed against the surviving quorum
/// within `RESUME_BUDGET` and must never surface Unauthenticated.
-// TODO(hubcio): fix this test
-#[ignore = "SDK has no multi-endpoint failover; client redials the dead primary forever"]
#[iggy_harness(cluster_nodes = 3)]
async fn given_a_client_producing_when_its_primary_is_killed_should_resume_without_hang_or_unauthenticated(
harness: &mut TestHarness,
@@ -97,6 +95,11 @@ async fn given_a_client_producing_when_its_primary_is_killed_should_resume_witho
// Pin the producing client to the primary's own endpoint, the way a
// leader-aware SDK ends up connected to whichever node answers as leader.
let leader = disk::leader_node_index(harness).await;
+ let primary_endpoint = harness
+ .node(leader)
+ .tcp_addr()
+ .expect("leader exposes a TCP endpoint")
+ .to_string();
let producer = harness
.node(leader)
.tcp_client()
@@ -106,6 +109,12 @@ async fn given_a_client_producing_when_its_primary_is_killed_should_resume_witho
.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 kills, or it proves nothing"
+ );
+
let stream = Identifier::named(STREAM_NAME).unwrap();
let topic = Identifier::named(TOPIC_NAME).unwrap();
let partitioning = Partitioning::partition_id(PARTITION_ID);
@@ -160,11 +169,15 @@ async fn given_a_client_producing_when_its_primary_is_killed_should_resume_witho
assert!(
resumed,
"a client pinned to a killed primary must complete its next operation against \
- the surviving quorum within {RESUME_BUDGET:?}, but the SDK has no \
- multi-endpoint failover: it holds only the dead node's server_address, its \
- fail-fast gate (auto-login disabled) surfaces errors without reconnecting, \
- and the leader redirect that could reroute it needs a live connection to \
- read the roster, so every retry redialed the dead endpoint \
+ the surviving quorum within {RESUME_BUDGET:?}: the roster learned while the \
+ connection was healthy names the survivors, and the credentials the sign-in \
+ succeeded with re-establish the session on whichever one answers \
({attempt} attempts, last error: {last_error:?})"
);
+ assert_ne!(
+ producer.get_connection_info().await.server_address,
+ primary_endpoint,
+ "the send that resumed must have landed on a survivor, so the client has to \
+ have moved off the killed primary's endpoint"
+ );
}
diff --git a/core/sdk/src/client_provider.rs b/core/sdk/src/client_provider.rs
index 423b7048e..7fbe919dd 100644
--- a/core/sdk/src/client_provider.rs
+++ b/core/sdk/src/client_provider.rs
@@ -132,6 +132,9 @@ impl ClientProviderConfig {
TransportProtocol::Tcp => {
config.tcp = Some(Arc::new(TcpClientConfig {
server_address: args.tcp_server_address,
+ // Command-line arguments name a single server; the roster
+ // is learned once the client is connected.
+ failover_addresses: Vec::new(),
tls_enabled: args.tcp_tls_enabled,
tls_domain: args.tcp_tls_domain,
tls_ca_file: args.tcp_tls_ca_file,
diff --git a/core/sdk/src/leader_aware.rs b/core/sdk/src/leader_aware.rs
index 35ffce9fc..bd99d2e32 100644
--- a/core/sdk/src/leader_aware.rs
+++ b/core/sdk/src/leader_aware.rs
@@ -18,7 +18,7 @@
use iggy_binary_protocol::codes::GET_CLUSTER_METADATA_CODE;
use iggy_common::ClusterClient;
use iggy_common::{
- ClusterMetadata, ClusterNodeRole, ClusterNodeStatus, IggyError, TransportProtocol,
+ ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, IggyError, TransportProtocol,
};
use std::net::SocketAddr;
use std::str::FromStr;
@@ -38,12 +38,33 @@ pub(crate) fn is_unauthenticated_metadata_probe(code: u32, error: &IggyError) ->
code == GET_CLUSTER_METADATA_CODE && matches!(error, IggyError::Unauthenticated)
}
+/// What one leader check learned from the cluster roster.
+pub struct LeaderCheck {
+ /// The leader's address, when it is not the node the client is on.
+ pub redirect: Option<String>,
+ /// Every endpoint the roster named for this transport. A client keeps
+ /// them as failover candidates: the address it was configured with dies
+ /// with its node, and the roster is unreachable exactly when it is
+ /// needed, so it has to be remembered while the connection is healthy.
+ pub endpoints: Vec<String>,
+}
+
+impl LeaderCheck {
+ /// A check that learned nothing: stay where we are, remember no endpoint.
+ fn inconclusive() -> Self {
+ Self {
+ redirect: None,
+ endpoints: Vec::new(),
+ }
+ }
+}
+
/// Check if we need to redirect to leader and return the leader address if redirection is needed
pub async fn check_and_redirect_to_leader<C: ClusterClient>(
client: &C,
current_address: &str,
transport: TransportProtocol,
-) -> Result<Option<String>, IggyError> {
+) -> Result<LeaderCheck, IggyError> {
debug!("Checking cluster metadata for leader detection");
// A cluster can be transiently leaderless: a restarted node cedes the
@@ -60,15 +81,31 @@ pub async fn check_and_redirect_to_leader<C: ClusterClient>(
metadata.nodes.len(),
metadata.name
);
+ let endpoints = transport_endpoints(&metadata, transport);
match process_cluster_metadata(&metadata, current_address, transport) {
- Outcome::Redirect(address) => return Ok(Some(address)),
- Outcome::LeaderIsCurrent => return Ok(None),
+ Outcome::Redirect(address) => {
+ return Ok(LeaderCheck {
+ redirect: Some(address),
+ endpoints,
+ });
+ }
+ Outcome::LeaderIsCurrent => {
+ return Ok(LeaderCheck {
+ redirect: None,
+ endpoints,
+ });
+ }
Outcome::NoLeader => {
if tokio::time::Instant::now() >= deadline {
warn!(
"No active leader found in cluster metadata within {LEADERLESS_WAIT_BUDGET:?}, connection will continue on server node {current_address}",
);
- return Ok(None);
+ // A leaderless roster still names where the nodes
+ // are, and that is what failover needs.
+ return Ok(LeaderCheck {
+ redirect: None,
+ endpoints,
+ });
}
tokio::time::sleep(LEADERLESS_POLL_INTERVAL).await;
}
@@ -82,14 +119,14 @@ pub async fn check_and_redirect_to_leader<C: ClusterClient>(
debug!(
"Cluster metadata answered Unauthenticated; the session is gone, connection will continue on server node {current_address}"
);
- return Ok(None);
+ return Ok(LeaderCheck::inconclusive());
}
Err(e) => {
warn!(
"Failed to get cluster metadata: {}, connection will continue on server node {}",
e, current_address
);
- return Ok(None);
+ return Ok(LeaderCheck::inconclusive());
}
}
}
@@ -110,6 +147,29 @@ enum Outcome {
NoLeader,
}
+/// Every node's address for `transport`, in roster order. A node that does
+/// not expose the transport reports port 0 and is skipped: dialing it would
+/// burn a failover attempt on an endpoint that cannot answer.
+fn transport_endpoints(metadata: &ClusterMetadata, transport: TransportProtocol) -> Vec<String> {
+ metadata
+ .nodes
+ .iter()
+ .filter_map(|node| {
+ let port = transport_port(node, transport);
+ (port != 0).then(|| format!("{}:{port}", node.ip))
+ })
+ .collect()
+}
+
+fn transport_port(node: &ClusterNode, transport: TransportProtocol) -> u16 {
+ match transport {
+ TransportProtocol::Tcp => node.endpoints.tcp,
+ TransportProtocol::Quic => node.endpoints.quic,
+ TransportProtocol::Http => node.endpoints.http,
+ TransportProtocol::WebSocket => node.endpoints.websocket,
+ }
+}
+
/// Process cluster metadata and determine if redirection is needed
fn process_cluster_metadata(
metadata: &ClusterMetadata,
@@ -132,12 +192,7 @@ fn process_cluster_metadata(
match leader {
Some(leader_node) => {
- let leader_port = match transport {
- TransportProtocol::Tcp => leader_node.endpoints.tcp,
- TransportProtocol::Quic => leader_node.endpoints.quic,
- TransportProtocol::Http => leader_node.endpoints.http,
- TransportProtocol::WebSocket => leader_node.endpoints.websocket,
- };
+ let leader_port = transport_port(leader_node, transport);
let leader_address = format!("{}:{}", leader_node.ip, leader_port);
info!(
@@ -162,7 +217,7 @@ fn process_cluster_metadata(
/// Check if two addresses refer to the same endpoint
/// Handles various formats like 127.0.0.1:8090 vs localhost:8090
-fn is_same_address(addr1: &str, addr2: &str) -> bool {
+pub(crate) fn is_same_address(addr1: &str, addr2: &str) -> bool {
match (parse_address(addr1), parse_address(addr2)) {
(Some(sock1), Some(sock2)) => sock1.ip() == sock2.ip() && sock1.port() == sock2.port(),
_ => normalize_address(addr1) == normalize_address(addr2),
@@ -245,6 +300,36 @@ mod tests {
));
}
+ fn node(name: &str, ip: &str, tcp: u16, role: ClusterNodeRole) -> ClusterNode {
+ ClusterNode {
+ name: name.to_string(),
+ ip: ip.to_string(),
+ endpoints: iggy_common::TransportEndpoints::new(tcp, 0, 3000, 3001),
+ role,
+ status: ClusterNodeStatus::Healthy,
+ }
+ }
+
+ #[test]
+ fn the_roster_names_every_node_that_exposes_the_transport() {
+ let metadata = ClusterMetadata {
+ name: "iggy".to_string(),
+ nodes: vec![
+ node("iggy-1", "10.0.0.1", 8090, ClusterNodeRole::Leader),
+ node("iggy-2", "10.0.0.2", 8090, ClusterNodeRole::Follower),
+ node("iggy-3", "10.0.0.3", 8090, ClusterNodeRole::Follower),
+ ],
+ };
+
+ assert_eq!(
+ transport_endpoints(&metadata, TransportProtocol::Tcp),
+ vec!["10.0.0.1:8090", "10.0.0.2:8090", "10.0.0.3:8090"]
+ );
+ // A node that does not expose the transport reports port 0; dialing
+ // it would burn a failover attempt on an endpoint that cannot answer.
+ assert!(transport_endpoints(&metadata, TransportProtocol::Quic).is_empty());
+ }
+
#[test]
fn test_is_same_address() {
assert!(is_same_address("127.0.0.1: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 921ef5d3d..255e473e4 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -508,12 +508,15 @@ impl QuicClient {
/// 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(
self,
¤t_address,
iggy_common::TransportProtocol::Quic,
)
- .await?;
+ .await?
+ .redirect;
if let Some(new_leader_address) = leader_address {
let mut redirection_state = self.leader_redirection_state.lock().await;
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index 2c09e3f2e..0b8e42b48 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -16,7 +16,8 @@
// under the License.
use crate::leader_aware::{
- LeaderRedirectionState, check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
+ LeaderRedirectionState, check_and_redirect_to_leader, is_same_address,
+ is_unauthenticated_metadata_probe,
};
use crate::prelude::Client;
use crate::prelude::TcpClientConfig;
@@ -36,6 +37,7 @@ use iggy_common::{
use iggy_common::{BinaryClient, BinaryTransport, PersonalAccessTokenClient, UserClient};
use rustls::pki_types::{CertificateDer, ServerName, pem::PemObject};
use secrecy::ExposeSecret;
+use std::io;
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
@@ -68,6 +70,13 @@ const NOT_READY_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_
/// overall.
const TRANSIENT_FAILOVER_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);
+/// Bound on one dial while the client has other endpoints to try. A host
+/// that drops the SYN -- powered off, or partitioned away -- takes the OS
+/// connect timeout to fail, which is minutes, and every other endpoint waits
+/// behind it. A client that knows a single endpoint has nothing to starve, so
+/// its dial stays unbounded.
+const FAILOVER_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
+
/// TCP client for interacting with the Iggy API.
/// It requires a valid server address.
#[derive(Debug)]
@@ -80,6 +89,15 @@ pub struct TcpClient {
pub(crate) connected_at: Mutex<Option<IggyTimestamp>>,
leader_redirection_state: Mutex<LeaderRedirectionState>,
pub(crate) current_server_address: Mutex<String>,
+ /// Every endpoint the cluster roster named, refreshed on each leader
+ /// check. A node dies together with its address, and the roster is
+ /// unreachable exactly when it is needed, so the client has to have
+ /// remembered it while the connection was still healthy.
+ roster_endpoints: Mutex<Vec<String>>,
+ /// Credentials a sign-in on this client succeeded with, so a reconnect --
+ /// onto this node or, after a failover, another one -- can re-establish
+ /// the session instead of surfacing `Unauthenticated`. Cleared on logout.
+ session_credentials: Mutex<Option<Credentials>>,
// `std::sync::Mutex` (not `tokio::sync::Mutex`): the critical section
// is `encode_request_header`, which is pure CPU and never awaits. The
// tokio variant would pay a waker alloc + internal semaphore on
@@ -159,9 +177,10 @@ impl BinaryTransport for TcpClient {
return Err(IggyError::Disconnected);
}
- if matches!(self.config.auto_login, AutoLogin::Disabled) && !is_login_register_code(code) {
- // Without auto-login a reconnect cannot re-establish the session,
- // so non-login requests fail fast. Login/register itself is the
+ if !is_login_register_code(code) && self.sign_in_credentials().await.is_none() {
+ // With no credentials -- neither configured nor remembered from a
+ // sign-in -- a reconnect cannot re-establish the session, so
+ // non-login requests fail fast. Login/register itself is the
// exception: the server stays deliberately silent on transient
// register failures (the server `surface_login_failure`) and
// relies on the client timing out and replaying the request.
@@ -230,6 +249,14 @@ impl iggy_common::VsrSessionControl for TcpClient {
Ok(())
}
+ async fn remember_session_credentials(&self, credentials: Credentials) {
+ self.session_credentials.lock().await.replace(credentials);
+ }
+
+ async fn forget_session_credentials(&self) {
+ self.session_credentials.lock().await.take();
+ }
+
fn sdk_version(&self) -> &'static str {
crate::SDK_VERSION
}
@@ -292,6 +319,8 @@ impl TcpClient {
connected_at: Mutex::new(None),
leader_redirection_state: Mutex::new(LeaderRedirectionState::new()),
current_server_address: Mutex::new(server_address),
+ roster_endpoints: Mutex::new(Vec::new()),
+ session_credentials: Mutex::new(None),
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
skip_auto_login_once: Mutex::new(false),
consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
@@ -320,19 +349,20 @@ impl TcpClient {
}
self.set_state(ClientState::Connecting).await;
- if let Some(connected_at) = self.connected_at.lock().await.as_ref() {
- let now = IggyTimestamp::now();
- let elapsed = now.as_micros() - connected_at.as_micros();
- let interval = self.config.reconnection.reestablish_after.as_micros();
- trace!(
- "Elapsed time since last connection: {}",
- IggyDuration::from(elapsed)
- );
- if elapsed < interval {
- let remaining = IggyDuration::from(interval - elapsed);
- info!("Trying to connect to the server in: {remaining}",);
- sleep(remaining.get_duration()).await;
- }
+ let candidates = self.dial_candidates().await;
+ // The reestablish delay paces reconnects to the one endpoint a
+ // single-address client has. With other endpoints known there is
+ // somewhere else to go, and pausing first only pushes the
+ // failover past the window the caller is willing to wait; the
+ // retry interval still paces the loop.
+ let reestablish_wait = if candidates.len() > 1 {
+ None
+ } else {
+ self.reestablish_wait().await
+ };
+ if let Some(remaining) = reestablish_wait {
+ info!("Trying to connect to the server in: {remaining}",);
+ sleep(remaining.get_duration()).await;
}
let tls_enabled = self.config.tls_enabled;
@@ -340,14 +370,15 @@ impl TcpClient {
let connection_stream: ConnectionStreamKind;
let remote_address;
let client_address;
+ let mut candidate = 0;
loop {
- let server_address = self.current_server_address.lock().await.clone();
+ let server_address = candidates[candidate].clone();
info!(
"{NAME} client is connecting to server: {}...",
server_address
);
- let connection = TcpStream::connect(&server_address).await;
+ let connection = self.dial(&server_address, candidates.len() > 1).await;
if let Err(err) = &connection {
error!(
"Failed to connect to server: {}. Error: {}",
@@ -358,6 +389,15 @@ impl TcpClient {
return Err(IggyError::CannotEstablishConnection);
}
+ // Every other endpoint gets its turn before the retry
+ // interval: the node just lost may be gone for good, and
+ // pausing on it helps nothing.
+ candidate += 1;
+ if candidate < candidates.len() {
+ continue;
+ }
+ candidate = 0;
+
let unlimited_retries = self.config.reconnection.max_retries.is_none();
let max_retries = self.config.reconnection.max_retries.unwrap_or_default();
let max_retries_str =
@@ -387,6 +427,10 @@ impl TcpClient {
error!("Failed to establish TCP connection to the server: {error}",);
IggyError::CannotEstablishConnection
})?;
+ // The endpoint that answered is where this client now lives:
+ // the leader check compares against it, and the next
+ // reconnect starts from it.
+ *self.current_server_address.lock().await = server_address.clone();
client_address = stream.local_addr().map_err(|error| {
error!("Failed to get the local address of the client: {error}",);
IggyError::CannotEstablishConnection
@@ -485,9 +529,9 @@ impl TcpClient {
};
// Handle auto-login
- let should_redirect = match &self.config.auto_login {
- AutoLogin::Disabled => {
- info!("Automatic sign-in is disabled.");
+ let should_redirect = match self.sign_in_credentials().await {
+ None => {
+ info!("No credentials to sign in with.");
// Only `IggyClient` redirects after a manual sign-in, so
// a raw transport can stay on a backup: its first
// replicated write gets `TransientNotAccepted`, the
@@ -495,14 +539,14 @@ impl TcpClient {
// `Unauthenticated` until the caller signs in again.
false
}
- AutoLogin::Enabled(credentials) => {
+ Some(credentials) => {
if skip_auto_login {
info!("Skipping automatic sign-in for a retried login/register request.");
false
} else {
info!("{NAME} client: {client_address} is signing in...");
self.set_state(ClientState::Authenticating).await;
- match credentials {
+ match &credentials {
Credentials::UsernamePassword(username, password) => {
self.login_user(username, password.expose_secret()).await?;
info!(
@@ -540,14 +584,22 @@ impl TcpClient {
/// 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();
- let leader_address = check_and_redirect_to_leader(
+ let leader_check = check_and_redirect_to_leader(
self,
¤t_address,
iggy_common::TransportProtocol::Tcp,
)
.await?;
- if let Some(new_leader_address) = leader_address {
+ // Replaced wholesale rather than merged: the roster is the cluster's
+ // own answer about where its nodes are, so a node it dropped should
+ // stop being dialed. The configured seeds are kept separately and
+ // outlive it.
+ if !leader_check.endpoints.is_empty() {
+ *self.roster_endpoints.lock().await = leader_check.endpoints;
+ }
+
+ if let Some(new_leader_address) = leader_check.redirect {
let mut redirection_state = self.leader_redirection_state.lock().await;
if !redirection_state.can_redirect() {
warn!("Maximum leader redirections reached, continuing with current connection");
@@ -573,6 +625,69 @@ impl TcpClient {
}
}
+ /// Credentials to sign in with after connecting: the configured ones, or
+ /// else the ones a manual sign-in on this client succeeded with. A manual
+ /// sign-in is otherwise less reconnectable than a configured one, which
+ /// is a surprising difference between two ways of doing the same thing.
+ async fn sign_in_credentials(&self) -> Option<Credentials> {
+ match &self.config.auto_login {
+ AutoLogin::Enabled(credentials) => Some(credentials.clone()),
+ AutoLogin::Disabled => self.session_credentials.lock().await.clone(),
+ }
+ }
+
+ /// Endpoints to dial for one connect, likeliest first: where the client
+ /// currently is, then the roster it learned while connected, then the
+ /// configured seeds.
+ async fn dial_candidates(&self) -> Vec<String> {
+ let mut candidates = vec![self.current_server_address.lock().await.clone()];
+ let roster = self.roster_endpoints.lock().await.clone();
+ for endpoint in roster.iter().chain(self.config.failover_addresses.iter()) {
+ if !candidates
+ .iter()
+ .any(|candidate| is_same_address(candidate, endpoint))
+ {
+ candidates.push(endpoint.clone());
+ }
+ }
+ candidates
+ }
+
+ /// Dial one endpoint, bounding the wait while other endpoints are queued
+ /// behind it (see `FAILOVER_DIAL_TIMEOUT`).
+ async fn dial(&self, server_address: &str, bounded: bool) -> io::Result<TcpStream> {
+ if !bounded {
+ return TcpStream::connect(server_address).await;
+ }
+
+ match tokio::time::timeout(FAILOVER_DIAL_TIMEOUT, TcpStream::connect(server_address)).await
+ {
+ Ok(connection) => connection,
+ Err(_elapsed) => Err(io::Error::new(
+ io::ErrorKind::TimedOut,
+ format!("dialing {server_address} took longer than {FAILOVER_DIAL_TIMEOUT:?}"),
+ )),
+ }
+ }
+
+ /// What is left of the `reestablish_after` window since the last
+ /// successful connection, if any.
+ async fn reestablish_wait(&self) -> Option<IggyDuration> {
+ let connected_at = self
+ .connected_at
+ .lock()
+ .await
+ .as_ref()
+ .map(IggyTimestamp::as_micros)?;
+ let elapsed = IggyTimestamp::now().as_micros() - connected_at;
+ let interval = self.config.reconnection.reestablish_after.as_micros();
+ trace!(
+ "Elapsed time since last connection: {}",
+ IggyDuration::from(elapsed)
+ );
+ (elapsed < interval).then(|| IggyDuration::from(interval - elapsed))
+ }
+
async fn disconnect(&self) -> Result<(), IggyError> {
if self.get_state().await == ClientState::Disconnected {
return Ok(());
@@ -871,6 +986,92 @@ const fn is_login_register_code(code: u32) -> bool {
mod tests {
use super::*;
+ fn client_with(server_address: &str, failover_addresses: Vec<String>) -> TcpClient {
+ TcpClient::create(Arc::new(TcpClientConfig {
+ server_address: server_address.to_string(),
+ failover_addresses,
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client")
+ }
+
+ #[tokio::test]
+ async fn dial_candidates_lead_with_the_current_endpoint_and_name_each_other_one_once() {
+ let client = client_with(
+ "127.0.0.1:8090",
+ vec!["127.0.0.1:8092".to_string(), "localhost:8090".to_string()],
+ );
+ *client.roster_endpoints.lock().await = vec![
+ "127.0.0.1:8090".to_string(),
+ "127.0.0.1:8091".to_string(),
+ "127.0.0.1:8092".to_string(),
+ ];
+
+ // The current endpoint leads, the roster follows, and neither the
+ // roster's copy of the current endpoint nor a seed that only spells
+ // the same endpoint differently earns a second dial.
+ assert_eq!(
+ client.dial_candidates().await,
+ vec![
+ "127.0.0.1:8090".to_string(),
+ "127.0.0.1:8091".to_string(),
+ "127.0.0.1:8092".to_string(),
+ ]
+ );
+ }
+
+ #[tokio::test]
+ async fn a_client_that_learned_no_roster_still_dials_its_configured_seeds() {
+ let client = client_with("127.0.0.1:8090", vec!["127.0.0.1:8091".to_string()]);
+
+ assert_eq!(
+ client.dial_candidates().await,
+ vec!["127.0.0.1:8090".to_string(), "127.0.0.1:8091".to_string()]
+ );
+ }
+
+ #[tokio::test]
+ async fn a_sign_in_makes_a_client_without_auto_login_reconnectable() {
+ let client = client_with("127.0.0.1:8090", Vec::new());
+ assert!(client.sign_in_credentials().await.is_none());
+
+ client
+ .remember_session_credentials(Credentials::UsernamePassword(
+ "iggy".to_string(),
+ "iggy".into(),
+ ))
+ .await;
+ assert!(client.sign_in_credentials().await.is_some());
+
+ // An explicit logout leaves no session to restore, and a reconnect
+ // must not resurrect one.
+ client.forget_session_credentials().await;
+ assert!(client.sign_in_credentials().await.is_none());
+ }
+
+ #[tokio::test]
+ async fn configured_credentials_outrank_the_ones_a_sign_in_remembered() {
+ let client = TcpClient::create(Arc::new(TcpClientConfig {
+ auto_login: AutoLogin::Enabled(Credentials::UsernamePassword(
+ "configured".to_string(),
+ "iggy".into(),
+ )),
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client");
+ client
+ .remember_session_credentials(Credentials::UsernamePassword(
+ "signed-in".to_string(),
+ "iggy".into(),
+ ))
+ .await;
+
+ match client.sign_in_credentials().await {
+ Some(Credentials::UsernamePassword(username, _)) => assert_eq!(username, "configured"),
+ other => panic!("expected the configured credentials, got {other:?}"),
+ }
+ }
+
#[test]
fn should_fail_with_empty_connection_string() {
let value = "";
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index c439a38f1..362a2aa95 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -533,12 +533,15 @@ 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(
self,
¤t_address,
iggy_common::TransportProtocol::WebSocket,
)
- .await?;
+ .await?
+ .redirect;
if let Some(new_leader_address) = leader_address {
let mut redirection_state = self.leader_redirection_state.lock().await;
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
index 1885c69fa..17d524365 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
@@ -424,6 +424,25 @@ public sealed partial class TcpMessageStream
return true;
}
+ /// <summary>
+ /// Keeps every node the roster names as a dial candidate. Replaced wholesale rather than merged: the
+ /// roster is the cluster's own answer about where its nodes are, so a node it dropped stops being dialed.
+ /// The configured address is kept separately and outlives it. A node that does not expose the tcp
+ /// transport reports port 0 and is skipped, since dialing it would burn an attempt on an endpoint that
+ /// cannot answer.
+ /// </summary>
+ private void RememberRoster(ClusterMetadata clusterMetadata)
+ {
+ var endpoints = clusterMetadata.Nodes
+ .Where(node => node.Endpoints.Tcp != 0)
+ .Select(node => ServerAddress.HostPort(node.Ip, node.Endpoints.Tcp))
+ .ToArray();
+ if (endpoints.Length > 0)
+ {
+ _rosterAddresses = endpoints;
+ }
+ }
+
private async Task<ClusterNode?> GetCurrentLeaderNodeAsync(CancellationToken token)
{
var leaderlessDeadline = Environment.TickCount64 + VsrLeaderlessWaitMs;
@@ -438,6 +457,8 @@ public sealed partial class TcpMessageStream
return null;
}
+ RememberRoster(clusterMetadata);
+
if (clusterMetadata.Nodes.Count() == 1)
{
return null;
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index 32ea6e03e..043a50809 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
@@ -83,6 +83,17 @@ public sealed partial class TcpMessageStream : IIggyClient
private DateTimeOffset _lastConnectionTime;
private int _leaderRedirectCount;
+ // Every node the roster named on the last read, kept as dial candidates. A node dies together with its
+ // address, and the roster is unreachable exactly when it is needed, so the client has to have remembered it
+ // while the connection was still healthy. Written by the leader probe, read by the connect loop.
+ private string[] _rosterAddresses = [];
+
+ // The credentials a sign-in succeeded with, so a reconnect - on this node or, after a failover, another one -
+ // can re-establish the session instead of leaving every later request unauthenticated. A caller that signs in
+ // by hand is otherwise less reconnectable than one that configures auto login, which is a surprising
+ // difference between two ways of doing the same thing. Cleared on sign-out.
+ private AutoLoginSettings? _rememberedLogin;
+
// Both are written by the connect and redirect paths, which do not hold the sending semaphore the request
// paths read them under, so they are accessed through Interlocked rather than as plain fields. Losing an
// update to the skip flag leaves a connection reporting Connected that never authenticated; losing one to
@@ -671,7 +682,7 @@ public sealed partial class TcpMessageStream : IIggyClient
if (_configuration.ReconnectionSettings.Enabled && !_configuration.AutoLoginSettings.Enabled)
{
_logger.LogWarning(
- "Reconnection is enabled without auto login: a lost session cannot be restored, requests will fail until the client logs in again");
+ "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);
@@ -814,8 +825,14 @@ public sealed partial class TcpMessageStream : IIggyClient
throw new NotConnectedException();
}
- return await LoginRegisterAsync(CommandCodes.LOGIN_REGISTER_CODE,
+ var identity = await LoginRegisterAsync(CommandCodes.LOGIN_REGISTER_CODE,
LoginRegister.Serialize(userName, password), token);
+ _rememberedLogin = new AutoLoginSettings
+ {
+ Enabled = true, Username = userName, Password = password
+ };
+
+ return identity;
}
/// <inheritdoc />
@@ -833,6 +850,9 @@ public sealed partial class TcpMessageStream : IIggyClient
{
await ResetConsensusSessionAsync();
+ // An explicit sign-out leaves no session to restore, and a reconnect must not resurrect one.
+ _rememberedLogin = null;
+
if (_state == ConnectionState.Authenticated)
{
SetConnectionStateAsync(ConnectionState.Connected);
@@ -889,8 +909,11 @@ public sealed partial class TcpMessageStream : IIggyClient
/// <inheritdoc />
public async Task<AuthResponse?> LoginWithPersonalAccessTokenAsync(string token, CancellationToken ct = default)
{
- return await LoginRegisterAsync(CommandCodes.LOGIN_REGISTER_WITH_PAT_CODE,
+ var identity = await LoginRegisterAsync(CommandCodes.LOGIN_REGISTER_WITH_PAT_CODE,
LoginRegister.SerializeWithPersonalAccessToken(token), ct);
+ _rememberedLogin = new AutoLoginSettings { Enabled = true, PersonalAccessToken = token };
+
+ return identity;
}
/// <summary>
@@ -921,7 +944,10 @@ public sealed partial class TcpMessageStream : IIggyClient
return;
}
- if (_lastConnectionTime != DateTimeOffset.MinValue)
+ // The initial delay paces reconnects to the one endpoint a single-address client has. With other
+ // endpoints known there is somewhere else to go, and pausing first only pushes the failover past the
+ // window the caller is willing to wait; the dial loop's own delay still paces the retries.
+ if (_lastConnectionTime != DateTimeOffset.MinValue && DialCandidates().Length == 1)
{
await Task.Delay(_configuration.ReconnectionSettings.InitialDelay, token);
}
@@ -984,7 +1010,7 @@ public sealed partial class TcpMessageStream : IIggyClient
// the ping is what brings an idle client back.
var unrecoverable = _state is ConnectionState.Disconnected or ConnectionState.Connecting
&& !(_configuration.ReconnectionSettings.Enabled
- && _configuration.AutoLoginSettings.Enabled);
+ && SignInSettings() != null);
if (IsConnecting || unrecoverable)
{
continue;
@@ -1130,15 +1156,18 @@ public sealed partial class TcpMessageStream : IIggyClient
var retryCount = 0;
var redirects = 0;
var delay = _configuration.ReconnectionSettings.InitialDelay;
+
+ if (string.IsNullOrEmpty(_currentAddress))
+ {
+ _currentAddress = _configuration.BaseAddress;
+ }
+
+ var candidates = DialCandidates();
+ var candidate = 0;
do
{
await DropStreamAsync();
- if (string.IsNullOrEmpty(_currentAddress))
- {
- _currentAddress = _configuration.BaseAddress;
- }
-
if (!ServerAddress.TryParse(_currentAddress, out var host, out var port))
{
throw new InvalidBaseAddressException();
@@ -1191,9 +1220,9 @@ public sealed partial class TcpMessageStream : IIggyClient
// No pre-login roster read: the server auth-gates cluster metadata, so leadership settles after
// a sign-in binds a session. A login dialed at a backup still succeeds because the server
// forwards the register to the primary.
- if (autoLogin && _configuration.AutoLoginSettings.Enabled && !ConsumeSkipAutoLogin())
+ if (autoLogin && SignInSettings() is { } signInSettings && !ConsumeSkipAutoLogin())
{
- await AutoLoginAsync(token);
+ await AutoLoginAsync(signInSettings, token);
if (await RedirectAsync(token))
{
@@ -1233,6 +1262,17 @@ public sealed partial class TcpMessageStream : IIggyClient
throw;
}
+ // Every other endpoint gets its turn before the retry delay: the node just lost may be gone for
+ // good, and pausing on it helps nothing.
+ if (++candidate < candidates.Length)
+ {
+ _currentAddress = candidates[candidate];
+ continue;
+ }
+
+ candidate = 0;
+ _currentAddress = candidates[0];
+
retryCount++;
if (_configuration.ReconnectionSettings.UseExponentialBackoff)
{
@@ -1267,6 +1307,10 @@ public sealed partial class TcpMessageStream : IIggyClient
_logger.LogInformation("Following leader redirect {Redirect} to {Address}", redirects, _currentAddress);
+ // The redirect moved the client, so the endpoint it moved to leads the next dial.
+ candidates = DialCandidates();
+ candidate = 0;
+
await Task.Delay(delay, token);
}
}
@@ -1294,20 +1338,64 @@ public sealed partial class TcpMessageStream : IIggyClient
}
}
- private async Task AutoLoginAsync(CancellationToken token)
+ private string[] DialCandidates()
+ {
+ return DialCandidates(_currentAddress, _configuration.BaseAddress, _rosterAddresses);
+ }
+
+ /// <summary>
+ /// The endpoints one connect dials, likeliest first: where the client currently is, the address it was
+ /// configured with, then the roster it learned while connected. Duplicates are dropped, so an endpoint the
+ /// roster merely spells differently does not earn a second attempt.
+ /// </summary>
+ internal static string[] DialCandidates(string currentAddress, string baseAddress, string[] rosterAddresses)
+ {
+ var candidates = new List<string>();
+ if (!string.IsNullOrEmpty(currentAddress))
+ {
+ candidates.Add(currentAddress);
+ }
+
+ foreach (var endpoint in rosterAddresses.Prepend(baseAddress))
+ {
+ if (!string.IsNullOrEmpty(endpoint) &&
+ !candidates.Exists(known => ServerAddress.IsSame(known, endpoint)))
+ {
+ candidates.Add(endpoint);
+ }
+ }
+
+ return candidates.ToArray();
+ }
+
+ private async Task AutoLoginAsync(AutoLoginSettings settings, CancellationToken token)
{
- var settings = _configuration.AutoLoginSettings;
if (!string.IsNullOrEmpty(settings.PersonalAccessToken))
{
- _logger.LogInformation("Auto login enabled. Trying to login with a personal access token");
+ _logger.LogInformation("Signing in with a personal access token");
await LoginWithPersonalAccessTokenAsync(settings.PersonalAccessToken, token);
return;
}
- _logger.LogInformation("Auto login enabled. Trying to login with credentials: {Username}", settings.Username);
+ _logger.LogInformation("Signing in with credentials: {Username}", settings.Username);
await LoginUserAsync(settings.Username, settings.Password, token);
}
+ /// <summary>
+ /// The credentials a connect signs in with: the configured ones, or else the ones a sign-in on this client
+ /// succeeded with. Null when nothing has ever signed in, which is when a reconnect cannot restore a
+ /// session at all.
+ /// </summary>
+ private AutoLoginSettings? SignInSettings()
+ {
+ if (_configuration.AutoLoginSettings.Enabled)
+ {
+ return _configuration.AutoLoginSettings;
+ }
+
+ return _rememberedLogin;
+ }
+
/// <summary>
/// Whether this connect was triggered by a login or register request that will re-authenticate itself,
/// so the auto-login must sit this one out. Consumes the flag.
@@ -1360,11 +1448,12 @@ public sealed partial class TcpMessageStream : IIggyClient
throw;
}
- // Without auto login a reconnect cannot re-establish the session, so the request would only come
- // back unauthenticated. Login and register are the exception: they re-authenticate themselves.
- if (!_configuration.AutoLoginSettings.Enabled && Volatile.Read(ref _skipAutoLoginOnce) == 0)
+ // With no credentials - neither configured nor remembered from a sign-in - a reconnect cannot
+ // re-establish the session, so the request would only come back unauthenticated. Login and register
+ // are the exception: they re-authenticate themselves.
+ if (SignInSettings() == null && Volatile.Read(ref _skipAutoLoginOnce) == 0)
{
- _logger.LogWarning("Auto login is disabled, the session cannot be re-established");
+ _logger.LogWarning("No credentials to sign in with, the session cannot be re-established");
SetConnectionStateAsync(ConnectionState.Disconnected);
throw;
}
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/DialCandidatesTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/DialCandidatesTests.cs
new file mode 100644
index 000000000..7fdaf9907
--- /dev/null
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/DialCandidatesTests.cs
@@ -0,0 +1,64 @@
+// 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.
+
+using Apache.Iggy.IggyClient.Implementations;
+
+namespace Apache.Iggy.Tests.VsrTests;
+
+/// <summary>
+/// Mirrors the Rust SDK's <c>dial_candidates</c>: a client that loses the node it is on has to dial the rest
+/// of the cluster, and the two SDKs have to agree on which endpoints those are and in what order.
+/// </summary>
+public sealed class DialCandidatesTests
+{
+ [Fact]
+ public void LeadsWithTheCurrentEndpointThenNamesEachOtherOneOnce()
+ {
+ var candidates = TcpMessageStream.DialCandidates(
+ "127.0.0.1:8090",
+ "localhost:8090",
+ ["127.0.0.1:8090", "127.0.0.1:8091", "127.0.0.1:8092"]);
+
+ // Neither the roster's copy of the current endpoint nor a configured address that only spells the same
+ // endpoint differently earns a second dial.
+ Assert.Equal(["127.0.0.1:8090", "127.0.0.1:8091", "127.0.0.1:8092"], candidates);
+ }
+
+ [Fact]
+ public void KeepsTheConfiguredAddressWhenNoRosterWasLearned()
+ {
+ var candidates = TcpMessageStream.DialCandidates("127.0.0.1:8091", "127.0.0.1:8090", []);
+
+ Assert.Equal(["127.0.0.1:8091", "127.0.0.1:8090"], candidates);
+ }
+
+ [Fact]
+ public void FallsBackToTheConfiguredAddressBeforeTheFirstConnect()
+ {
+ var candidates = TcpMessageStream.DialCandidates(string.Empty, "127.0.0.1:8090", []);
+
+ Assert.Equal(["127.0.0.1:8090"], candidates);
+ }
+
+ [Fact]
+ public void DialsOneEndpointWhenNothingElseIsKnown()
+ {
+ var candidates = TcpMessageStream.DialCandidates("127.0.0.1:8090", "127.0.0.1:8090", []);
+
+ Assert.Equal(["127.0.0.1:8090"], candidates);
+ }
+}
diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
new file mode 100644
index 000000000..3bc704e06
--- /dev/null
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs
@@ -0,0 +1,346 @@
+// 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.
+
+using System.Buffers.Binary;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Contracts.Tcp;
+using Apache.Iggy.Enums;
+using Apache.Iggy.IggyClient.Implementations;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace Apache.Iggy.Tests.VsrTests;
+
+/// <summary>
+/// The node a client signed in on dies; its next request has to complete on a survivor the roster named,
+/// under a session established there. Mirrors
+/// <c>core/integration/tests/cluster/failover_client_continuity.rs</c>.
+/// </summary>
+public sealed class EndpointFailoverTests
+{
+ private const int HeaderSize = 256;
+ private const int SizeOffset = 48;
+ private const int CommandOffset = 60;
+ private const int RequestIdOffset = 168;
+ private const int RequestOperationOffset = 176;
+ private const int RequestReservedOffset = 196;
+ private const int ReplyRequestIdOffset = 200;
+ private const int ReplyOperationOffset = 208;
+ private const int ReplyStatusOffset = 216;
+
+ private const byte CommandReply = 8;
+ private const byte OperationRegister = 1;
+ private const byte OperationNonReplicated = 2;
+ private const int GetClusterMetadataCode = 12;
+ private const int PingCode = 1;
+
+ [Fact]
+ public async Task ResumesOnASurvivorAfterTheSignedInNodeDies()
+ {
+ using var primary = new MockNode();
+ using var survivor = new MockNode();
+
+ // The primary leads, so the sign-in settles there and the roster is only remembered - not acted on -
+ // until the node dies.
+ primary.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(primary.Port, survivor.Port, primary.Port))
+ : Answer(request));
+ survivor.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(primary.Port, survivor.Port, survivor.Port))
+ : Answer(request));
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{primary.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true, MaxRetries = 4, InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration, NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ // No auto login: the credentials come from the caller's own sign-in, which is the shape that could not
+ // reconnect at all before.
+ await client.LoginUserAsync("iggy", "iggy", TestContext.Current.CancellationToken);
+ await client.PingAsync(TestContext.Current.CancellationToken);
+ Assert.Equal(1, primary.Pings);
+
+ primary.Kill();
+
+ // The request in flight when the node died is allowed to fail; what is not allowed is never completing
+ // one, which is what a client that only knows the dead endpoint does.
+ var (resumed, lastError) = await ResumedWithin(client, TimeSpan.FromSeconds(10));
+ Assert.True(resumed,
+ $"the client has to resume on the survivor the roster named ({lastError}, survivor saw " +
+ $"{survivor.Registrations} registrations and {survivor.Pings} pings)");
+ Assert.True(survivor.Registrations >= 1, "the remembered credentials signed in again on the survivor");
+ Assert.True(survivor.Pings >= 1, "the request landed on the survivor");
+ }
+
+ [Fact]
+ public async Task FailsFastWhenNothingEverSignedIn()
+ {
+ using var node = new MockNode();
+ node.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(node.Port, node.Port, node.Port))
+ : Answer(request));
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{node.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true, MaxRetries = 2, InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration, NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.PingAsync(TestContext.Current.CancellationToken);
+
+ node.Kill();
+
+ var (resumed, _) = await ResumedWithin(client, TimeSpan.FromSeconds(2));
+ Assert.False(resumed, "a client that never signed in cannot restore a session by reconnecting");
+ }
+
+ private static async Task<(bool Resumed, string LastError)> ResumedWithin(TcpMessageStream client,
+ TimeSpan budget)
+ {
+ var deadline = DateTimeOffset.UtcNow + budget;
+ var lastError = "none";
+ var attempts = 0;
+ while (DateTimeOffset.UtcNow < deadline)
+ {
+ attempts++;
+ try
+ {
+ await client.PingAsync(TestContext.Current.CancellationToken);
+
+ return (true, lastError);
+ }
+ catch (Exception error)
+ {
+ lastError = $"{attempts} attempts, last: {error.GetType().Name}: {error.Message}";
+ await Task.Delay(50, TestContext.Current.CancellationToken);
+ }
+ }
+
+ return (false, lastError);
+ }
+
+ /// <summary>A reply for anything the roster read does not claim: a register, or an empty read.</summary>
+ private static byte[] Answer(MockRequest request)
+ {
+ return request.Operation == OperationRegister
+ ? Reply(OperationRegister, RegisterBody(session: 128))
+ : Reply(OperationNonReplicated, []);
+ }
+
+ private static byte[] Reply(byte operation, byte[] body)
+ {
+ 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);
+ body.CopyTo(frame.AsSpan(HeaderSize));
+
+ return frame;
+ }
+
+ /// <summary>
+ /// A register reply carries a committed result section, so its four leading zero bytes announce zero
+ /// entries and the typed payload starts right after them. A non-replicated read carries none.
+ /// </summary>
+ private static byte[] RegisterBody(ulong session)
+ {
+ var serverVersion = Encoding.UTF8.GetBytes("0.0.0");
+ var body = new byte[4 + 17 + serverVersion.Length];
+ var payload = body.AsSpan(4);
+ BinaryPrimitives.WriteUInt32LittleEndian(payload[..4], 7);
+ BinaryPrimitives.WriteUInt64LittleEndian(payload[4..12], session);
+ BinaryPrimitives.WriteUInt32LittleEndian(payload[12..16], 11 << 10);
+ payload[16] = (byte)serverVersion.Length;
+ serverVersion.CopyTo(payload[17..]);
+
+ return body;
+ }
+
+ private static byte[] ClusterMetadata(ushort primaryPort, ushort survivorPort, ushort leaderPort)
+ {
+ var body = new List<byte>();
+ WriteString(body, "test-cluster");
+ body.AddRange(BitConverter.GetBytes(2u));
+ WriteNode(body, "primary", primaryPort, primaryPort == leaderPort);
+ WriteNode(body, "survivor", survivorPort, survivorPort == leaderPort);
+
+ return body.ToArray();
+ }
+
+ private static void WriteNode(List<byte> body, string name, ushort port, bool leader)
+ {
+ WriteString(body, name);
+ WriteString(body, "127.0.0.1");
+ body.AddRange(BitConverter.GetBytes(port));
+ body.AddRange(BitConverter.GetBytes((ushort)0));
+ body.AddRange(BitConverter.GetBytes((ushort)0));
+ body.AddRange(BitConverter.GetBytes((ushort)0));
+ body.Add(leader ? (byte)0 : (byte)1);
+ body.Add(0);
+ }
+
+ private static void WriteString(List<byte> body, string value)
+ {
+ var bytes = Encoding.UTF8.GetBytes(value);
+ body.AddRange(BitConverter.GetBytes((uint)bytes.Length));
+ body.AddRange(bytes);
+ }
+
+ private readonly record struct MockRequest(byte Operation, int Code, ulong RequestId);
+
+ /// <summary>
+ /// A loopback VSR node. Killing it drops the live sockets and stops accepting, so a redial is refused the
+ /// way a dead process refuses one.
+ /// </summary>
+ private sealed class MockNode : IDisposable
+ {
+ private readonly TcpListener _listener;
+ private readonly List<TcpClient> _accepted = [];
+ private volatile bool _killed;
+ private int _pings;
+ private int _registrations;
+
+ public MockNode()
+ {
+ _listener = new TcpListener(IPAddress.Loopback, 0);
+ _listener.Start();
+ Port = (ushort)((IPEndPoint)_listener.LocalEndpoint).Port;
+ }
+
+ public ushort Port { get; }
+
+ public int Pings => Volatile.Read(ref _pings);
+
+ public int Registrations => Volatile.Read(ref _registrations);
+
+ public void Serve(Func<MockRequest, byte[]> handler)
+ {
+ _ = Task.Run(async () =>
+ {
+ while (!_killed)
+ {
+ TcpClient connection;
+ try
+ {
+ connection = await _listener.AcceptTcpClientAsync();
+ }
+ catch (Exception)
+ {
+ return;
+ }
+
+ lock (_accepted)
+ {
+ _accepted.Add(connection);
+ }
+
+ _ = Task.Run(() => Exchange(connection, handler));
+ }
+ });
+ }
+
+ public void Kill()
+ {
+ _killed = true;
+ lock (_accepted)
+ {
+ foreach (var connection in _accepted)
+ {
+ connection.Close();
+ }
+
+ _accepted.Clear();
+ }
+
+ _listener.Stop();
+ }
+
+ public void Dispose()
+ {
+ Kill();
+ }
+
+ private async Task Exchange(TcpClient connection, Func<MockRequest, byte[]> handler)
+ {
+ try
+ {
+ await using var stream = connection.GetStream();
+ var header = new byte[HeaderSize];
+ while (!_killed)
+ {
+ await ReadExactly(stream, header);
+ var size = BinaryPrimitives.ReadUInt32LittleEndian(header.AsSpan(SizeOffset, 4));
+ var body = new byte[size - HeaderSize];
+ await ReadExactly(stream, body);
+
+ var request = new MockRequest(header[RequestOperationOffset],
+ BinaryPrimitives.ReadInt32LittleEndian(header.AsSpan(RequestReservedOffset, 4)),
+ BinaryPrimitives.ReadUInt64LittleEndian(header.AsSpan(RequestIdOffset, 8)));
+ if (request.Operation == OperationRegister)
+ {
+ Interlocked.Increment(ref _registrations);
+ }
+ else if (request.Code == PingCode)
+ {
+ Interlocked.Increment(ref _pings);
+ }
+
+ var reply = handler(request);
+ BinaryPrimitives.WriteUInt64LittleEndian(reply.AsSpan(ReplyRequestIdOffset, 8),
+ request.RequestId);
+ await stream.WriteAsync(reply);
+ await stream.FlushAsync();
+ }
+ }
+ catch (Exception)
+ {
+ // A killed node and a client that went away look the same here.
+ }
+ }
+
+ private static async Task ReadExactly(NetworkStream stream, byte[] buffer)
+ {
+ var read = 0;
+ while (read < buffer.Length)
+ {
+ var chunk = await stream.ReadAsync(buffer.AsMemory(read));
+ if (chunk == 0)
+ {
+ throw new EndOfStreamException("Connection closed");
+ }
+
+ read += chunk;
+ }
+ }
+ }
+}
diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go
index fe662af53..5b9a46527 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -85,6 +85,14 @@ type IggyTcpClient struct {
// loggedOut records an explicit sign-out, so a reconnect's automatic
// sign-in does not silently reverse it; guarded by c.mtx.
loggedOut bool
+ // rememberedLogin holds the credentials a manual sign-in succeeded with,
+ // so a reconnect -- on this node or, after a failover, another one -- can
+ // re-establish the session instead of surfacing an unauthenticated error.
+ // A caller that signs in by hand is otherwise less reconnectable than one
+ // that configures auto-login, which is a surprising difference between
+ // two ways of doing the same thing. Cleared on sign-out; guarded by
+ // c.mtx.
+ rememberedLogin AutoLogin
// groups caches the consumer-group assignments this client polls with.
groups groupAssignmentCache
// topics caches what a send needs to resolve a partition locally.
@@ -480,12 +488,13 @@ func (c *IggyTcpClient) exchange(ctx context.Context, code uint32, frame []byte)
return nil, err
}
- // Without auto-login a reconnect cannot restore the session, so anything
- // but a sign-in fails here instead of replaying unauthenticated. The
- // sign-in itself is the exception: the server stays silent on a transient
+ // With no credentials -- neither configured nor remembered from a
+ // sign-in -- a reconnect cannot restore the session, so anything but a
+ // sign-in fails here instead of replaying unauthenticated. The sign-in
+ // itself is the exception: the server stays silent on a transient
// register failure and expects the client to replay it.
login := isRegisterCode(code)
- if !c.config.autoLogin.enabled && !login {
+ if _, ok := c.signInCredentials(); !ok && !login {
return nil, err
}
c.mtx.Lock()
@@ -890,8 +899,13 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
connectedAt := c.connectedAt
c.mtx.Unlock()
- // handle reestablish interval
- if !connectedAt.IsZero() {
+ candidates := c.connectionCandidates()
+
+ // The reestablish interval paces reconnects to the one endpoint a
+ // single-address client has. With other endpoints known there is somewhere
+ // else to go, and pausing first only pushes the failover past the window
+ // the caller is willing to wait; the retry interval still paces the loop.
+ if !connectedAt.IsZero() && len(candidates) == 1 {
now := time.Now()
elapsed := now.Sub(connectedAt)
reestablishAfter := c.config.reconnection.reestablishAfter
@@ -909,10 +923,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
attempts = uint(c.config.reconnection.maxRetries)
interval = c.config.reconnection.interval
}
-
- candidates := c.connectionCandidates()
var conn net.Conn
- var candidateIndex int
if err := retry.New(
retry.Context(ctx),
retry.Attempts(attempts),
@@ -923,46 +934,23 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
}),
).Do(
func() error {
- address := candidates[candidateIndex%len(candidates)]
- candidateIndex++
- c.logger.Info("Iggy client is connecting to server...", slog.String("server_address", address))
- connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", address)
- if err != nil {
- c.logger.Error("Failed to establish TCP connection to the server", slog.Any("error", err))
- return ierror.ErrCannotEstablishConnection
- }
-
- tc := connection.(*net.TCPConn)
- if err := tc.SetNoDelay(c.config.noDelay); err != nil {
- c.logger.Error("Failed to set the nodelay option on the client, continuing...", slog.Any("error", err))
- }
-
- c.mtx.Lock()
- c.clientAddress = tc.LocalAddr().String()
- c.currentServerAddress = address
- c.mtx.Unlock()
+ // Every endpoint gets its turn inside one attempt, so a full pass
+ // over the cluster costs one retry rather than one per endpoint:
+ // a pass that stopped at the first refusal would never reach the
+ // survivors of a client configured for a single retry.
+ var lastErr error
+ for _, address := range candidates {
+ connection, err := c.dialCandidate(ctx, address)
+ if err != nil {
+ lastErr = err
+ continue
+ }
- if !c.config.tlsEnabled {
conn = connection
return nil
}
- // TLS logic
- tlsConfig, err := c.createTLSConfig()
- if err != nil {
- _ = connection.Close()
- return err
- }
-
- tlsConn := tls.Client(connection, tlsConfig)
- if err := tlsConn.HandshakeContext(ctx); err != nil {
- c.logger.Error("Failed to establish a TLS connection to the server", slog.Any("error", err))
- _ = connection.Close()
- return fmt.Errorf("TLS handshake failed: %w", err)
- }
-
- conn = tlsConn
- return nil
+ return lastErr
}); err != nil {
c.mtx.Lock()
c.transportState = iggcon.TransportStateDisconnected
@@ -994,6 +982,47 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
return nil
}
+// dialCandidate opens one connection, wrapping it in TLS when configured, and
+// records the endpoint that answered: the leader check compares against it and
+// the next reconnect starts from it.
+func (c *IggyTcpClient) dialCandidate(ctx context.Context, address string) (net.Conn, error) {
+ c.logger.Info("Iggy client is connecting to server...", slog.String("server_address", address))
+ connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", address)
+ if err != nil {
+ c.logger.Error("Failed to establish TCP connection to the server", slog.Any("error", err))
+ return nil, ierror.ErrCannotEstablishConnection
+ }
+
+ tc := connection.(*net.TCPConn)
+ if err := tc.SetNoDelay(c.config.noDelay); err != nil {
+ c.logger.Error("Failed to set the nodelay option on the client, continuing...", slog.Any("error", err))
+ }
+
+ c.mtx.Lock()
+ c.clientAddress = tc.LocalAddr().String()
+ c.currentServerAddress = address
+ c.mtx.Unlock()
+
+ if !c.config.tlsEnabled {
+ return connection, nil
+ }
+
+ tlsConfig, err := c.createTLSConfig()
+ if err != nil {
+ _ = connection.Close()
+ return nil, err
+ }
+
+ tlsConn := tls.Client(connection, tlsConfig)
+ if err := tlsConn.HandshakeContext(ctx); err != nil {
+ c.logger.Error("Failed to establish a TLS connection to the server", slog.Any("error", err))
+ _ = connection.Close()
+ return nil, fmt.Errorf("TLS handshake failed: %w", err)
+ }
+
+ return tlsConn, nil
+}
+
func (c *IggyTcpClient) connectionCandidates() []string {
c.mtx.Lock()
defer c.mtx.Unlock()
@@ -1025,8 +1054,9 @@ func (c *IggyTcpClient) connectionCandidates() []string {
// backup: once the caller signs in, the first replicated request fails over
// through the transient-deny path.
func (c *IggyTcpClient) establishSession(ctx context.Context, skipAutoLogin bool) error {
- if !c.config.autoLogin.enabled {
- c.logger.Info("Automatic sign-in is disabled.")
+ credentials, ok := c.signInCredentials()
+ if !ok {
+ c.logger.Info("No credentials to sign in with.")
return nil
}
if skipAutoLogin {
@@ -1034,7 +1064,6 @@ func (c *IggyTcpClient) establishSession(ctx context.Context, skipAutoLogin bool
return nil
}
- credentials := c.config.autoLogin.credentials
if credentials.personalAccessToken != "" {
_, err := c.LoginWithPersonalAccessToken(ctx, credentials.personalAccessToken)
return err
@@ -1043,6 +1072,32 @@ func (c *IggyTcpClient) establishSession(ctx context.Context, skipAutoLogin bool
return err
}
+// signInCredentials reports the credentials a reconnect signs in with: the
+// configured ones, or else the ones a manual sign-in succeeded with.
+func (c *IggyTcpClient) signInCredentials() (Credentials, bool) {
+ if c.config.autoLogin.enabled {
+ return c.config.autoLogin.credentials, true
+ }
+ c.mtx.Lock()
+ defer c.mtx.Unlock()
+ return c.rememberedLogin.credentials, c.rememberedLogin.enabled
+}
+
+// rememberLogin keeps the credentials a sign-in just succeeded with.
+func (c *IggyTcpClient) rememberLogin(credentials Credentials) {
+ c.mtx.Lock()
+ defer c.mtx.Unlock()
+ c.rememberedLogin = NewAutoLogin(credentials)
+}
+
+// forgetLogin drops them: after an explicit sign-out there is no session to
+// restore, and a reconnect must not resurrect one.
+func (c *IggyTcpClient) forgetLogin() {
+ c.mtx.Lock()
+ defer c.mtx.Unlock()
+ c.rememberedLogin = AutoLogin{}
+}
+
func (c *IggyTcpClient) createTLSConfig() (*tls.Config, error) {
tlsConfig := &tls.Config{
InsecureSkipVerify: !c.config.tls.tlsValidateCertificate,
diff --git a/foreign/go/client/tcp/tcp_failover_test.go b/foreign/go/client/tcp/tcp_failover_test.go
new file mode 100644
index 000000000..c5735517c
--- /dev/null
+++ b/foreign/go/client/tcp/tcp_failover_test.go
@@ -0,0 +1,155 @@
+// 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.
+
+package tcp
+
+import (
+ "context"
+ "log/slog"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/apache/iggy/foreign/go/internal/command"
+ "github.com/apache/iggy/foreign/go/internal/vsr"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// The node a client signed in on dies; its next request has to complete on a
+// survivor the roster named, under the identity a fresh sign-in binds there.
+// Mirrors `core/integration/tests/cluster/failover_client_continuity.rs`.
+func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) {
+ var survivor *testListener
+ var primary *testListener
+ var primaryDead atomic.Bool
+
+ survivor = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 1, primary.address(), survivor.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 512)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ primary = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ // A dead node answers nothing; returning nil drops the connection the
+ // way a killed process does.
+ if primaryDead.Load() {
+ return nil
+ }
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ // The primary leads, so the sign-in settles here and the roster is
+ // only remembered -- not acted on -- until the node dies.
+ return clusterMetadataFrame(t, 0, primary.address(), survivor.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 128)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ // No auto-login: the credentials come from the caller's own sign-in, which
+ // is the shape that could not reconnect at all before.
+ client := newDialingClient(t, primary.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+ require.NoError(t, client.Ping(ctx), "the live primary answers")
+ require.Equal(t, primary.address(), client.currentServerAddress)
+
+ primaryDead.Store(true)
+ require.NoError(t, primary.listener.Close(), "stop accepting, so a redial is refused")
+
+ require.NoError(t, client.Ping(ctx),
+ "the client has to resume on the survivor the roster named")
+
+ assert.Equal(t, survivor.address(), client.currentServerAddress,
+ "the client moved off the dead endpoint")
+ assert.True(t, client.session.Bound(), "the session was re-established")
+
+ var registers int
+ for _, read := range survivor.recorded() {
+ if read.operation() == vsr.OperationRegister {
+ registers++
+ }
+ }
+ assert.Equal(t, 1, registers,
+ "the remembered credentials signed in again on the survivor")
+}
+
+// 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) {
+ var server *testListener
+ var dead atomic.Bool
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if dead.Load() {
+ return nil
+ }
+ return singleNodeHandler(t, func() string { return server.address() })(0, 0, read)
+ })
+
+ client := newDialingClient(t, server.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ require.NoError(t, client.Ping(ctx))
+
+ dead.Store(true)
+ require.NoError(t, server.listener.Close())
+
+ assert.Error(t, client.Ping(ctx),
+ "a client that never signed in cannot restore a session by reconnecting")
+}
+
+// An explicit sign-out is caller intent: the reconnect must not sign back in
+// with the credentials the earlier sign-in used.
+func TestFailover_DoesNotResurrectASignedOutSession(t *testing.T) {
+ var server *testListener
+ server = listenVSR(t, nil, singleNodeHandler(t, func() string { return server.address() }))
+
+ client := newDialingClient(t, server.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+ require.NoError(t, client.LogoutUser(ctx))
+
+ credentials, ok := client.signInCredentials()
+ assert.False(t, ok, "the sign-out forgot them")
+ assert.Empty(t, credentials.username)
+}
+
+func TestFailover_LeavesTheReestablishPauseToSingleEndpointClients(t *testing.T) {
+ client := NewIggyTcpClient(slog.New(slog.DiscardHandler),
+ WithServerAddress("127.0.0.1:8090"))
+ client.config.reconnection.reestablishAfter = time.Minute
+ client.connectedAt = time.Now()
+
+ // One endpoint: the pause is the only thing keeping a reconnect from
+ // hammering the node it just lost.
+ require.Len(t, client.connectionCandidates(), 1)
+
+ client.knownServerAddresses = []string{"127.0.0.1:8091"}
+ require.Len(t, client.connectionCandidates(), 2,
+ "with somewhere else to go the pause only delays the failover")
+}
diff --git a/foreign/go/client/tcp/tcp_session_credentials_test.go b/foreign/go/client/tcp/tcp_session_credentials_test.go
new file mode 100644
index 000000000..c219d4f6a
--- /dev/null
+++ b/foreign/go/client/tcp/tcp_session_credentials_test.go
@@ -0,0 +1,66 @@
+// 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.
+
+package tcp
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func newCredentialClient(autoLogin AutoLogin) *IggyTcpClient {
+ config := defaultTcpClientConfig()
+ config.autoLogin = autoLogin
+ return &IggyTcpClient{config: config}
+}
+
+func TestSignInCredentials_AreAbsentUntilSomethingSignsIn(t *testing.T) {
+ client := newCredentialClient(AutoLogin{})
+
+ _, ok := client.signInCredentials()
+ assert.False(t, ok)
+}
+
+func TestSignInCredentials_ComeFromAManualSignInWithoutAutoLogin(t *testing.T) {
+ client := newCredentialClient(AutoLogin{})
+
+ client.rememberLogin(NewUsernamePasswordCredentials("iggy", "secret"))
+
+ credentials, ok := client.signInCredentials()
+ require.True(t, ok)
+ assert.Equal(t, "iggy", credentials.username)
+ assert.Equal(t, "secret", credentials.password)
+
+ // An explicit sign-out leaves no session to restore, and a reconnect must
+ // not resurrect one.
+ client.forgetLogin()
+ _, ok = client.signInCredentials()
+ assert.False(t, ok)
+}
+
+func TestSignInCredentials_PreferTheConfiguredOnes(t *testing.T) {
+ client := newCredentialClient(NewAutoLogin(NewUsernamePasswordCredentials("configured", "secret")))
+
+ client.rememberLogin(NewPersonalAccessTokenCredentials("signed-in-token"))
+
+ credentials, ok := client.signInCredentials()
+ require.True(t, ok)
+ assert.Equal(t, "configured", credentials.username)
+ assert.Empty(t, credentials.personalAccessToken)
+}
diff --git a/foreign/go/client/tcp/tcp_session_management.go b/foreign/go/client/tcp/tcp_session_management.go
index d409f7d9a..b6a91e30e 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -33,7 +33,12 @@ func (c *IggyTcpClient) LoginUser(ctx context.Context, username string, password
if err != nil {
return nil, err
}
- return c.register(ctx, uint32(command.LoginRegisterCode), body)
+ identity, err := c.register(ctx, uint32(command.LoginRegisterCode), body)
+ if err != nil {
+ return nil, err
+ }
+ c.rememberLogin(NewUsernamePasswordCredentials(username, password))
+ return identity, nil
}
func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx context.Context, token string) (*iggcon.IdentityInfo, error) {
@@ -41,7 +46,12 @@ func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx context.Context, token
if err != nil {
return nil, err
}
- return c.register(ctx, uint32(command.LoginRegisterWithPATCode), body)
+ identity, err := c.register(ctx, uint32(command.LoginRegisterWithPATCode), body)
+ if err != nil {
+ return nil, err
+ }
+ c.rememberLogin(NewPersonalAccessTokenCredentials(token))
+ return identity, nil
}
// register runs the sign-in handshake, binds the session the server assigned,
@@ -188,6 +198,7 @@ func (c *IggyTcpClient) LogoutUser(ctx context.Context) error {
c.groups.clear()
c.topics.clearCounts()
c.mtx.Unlock()
+ c.forgetLogin()
return nil
}
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 e2673d965..93f463e72 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
@@ -49,6 +49,8 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -58,6 +60,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
+import java.util.stream.Stream;
/**
* Async TCP client for Apache Iggy message streaming, built on Netty.
@@ -129,10 +132,38 @@ public class AsyncIggyTcpClient {
private final Optional<File> tlsCertificate;
private final TcpConnectionPoolConfig poolConfig;
private final ClientRoutingState routingState = new ClientRoutingState();
+ private final LoginRoutingHook loginRoutingHook = new LoginRoutingHook() {
+
+ @Override
+ public CompletableFuture<IdentityInfo> loginOnLeader(Supplier<CompletableFuture<IdentityInfo>> loginAttempt) {
+ return AsyncIggyTcpClient.this.loginOnLeader(loginAttempt);
+ }
+
+ @Override
+ public void forgetLogin() {
+ rememberedLogin = null;
+ }
+ };
private final AtomicReference<AsyncTcpConnection> connection = new AtomicReference<>();
private final AtomicReference<CompletableFuture<Void>> loginChain =
new AtomicReference<>(CompletableFuture.completedFuture(null));
private volatile ConnectionInfo connectionInfo;
+ /**
+ * Every node the roster named on the last leader check, kept as redial
+ * candidates. A node dies together with its address, and the roster is
+ * unreachable exactly when it is needed, so it has to have been
+ * remembered while the connection was still healthy.
+ */
+ private volatile List<ConnectionInfo> rosterTargets = List.of();
+ /**
+ * The login a successful sign-in ran, replayed after a redial so the
+ * session is re-established on whichever node answers. The supplier
+ * already carries the credentials it signed in with, so nothing new is
+ * stored. Cleared on an explicit sign-out, which leaves no session to
+ * restore.
+ */
+ private volatile Supplier<CompletableFuture<IdentityInfo>> rememberedLogin;
+
private volatile boolean closed;
private MessagesClient messagesClient;
private ConsumerGroupsClient consumerGroupsClient;
@@ -235,9 +266,9 @@ public class AsyncIggyTcpClient {
consumerOffsetsClient = new ConsumerOffsetsTcpClient(currentConnection);
streamsClient = new StreamsTcpClient(currentConnection);
topicsClient = new TopicsTcpClient(currentConnection);
- usersClient = new UsersTcpClient(currentConnection, this::loginOnLeader);
+ usersClient = new UsersTcpClient(currentConnection, loginRoutingHook);
systemClient = new SystemTcpClient(currentConnection);
- personalAccessTokensClient = new PersonalAccessTokensTcpClient(currentConnection, this::loginOnLeader);
+ personalAccessTokensClient = new PersonalAccessTokensTcpClient(currentConnection, loginRoutingHook);
partitionsClient = new PartitionsTcpClient(currentConnection);
});
}
@@ -585,7 +616,7 @@ public class AsyncIggyTcpClient {
log.error("Redial gave up after {} attempts, next request will fail fast", policy.getMaxRetries());
return CompletableFuture.completedFuture(null);
}
- ConnectionInfo target = ReconnectPlan.target(connectionInfo, seedConnectionInfo, attempt);
+ ConnectionInfo target = ReconnectPlan.target(redialCandidates(), attempt);
Duration delay = ReconnectPlan.delay(policy, attempt);
Executor delayedExecutor = CompletableFuture.delayedExecutor(delay.toMillis(), TimeUnit.MILLISECONDS);
return CompletableFuture.supplyAsync(() -> null, delayedExecutor).thenCompose(ignored -> {
@@ -617,6 +648,12 @@ public class AsyncIggyTcpClient {
* again before Register when the redialed node is not the leader.
*/
private CompletableFuture<Void> replayLogin() {
+ Supplier<CompletableFuture<IdentityInfo>> replay = rememberedLogin;
+ if (replay != null) {
+ // Runs through loginOnLeader, so a redial that landed on a backup
+ // still settles on the leader before the session is used.
+ return loginOnLeader(replay).thenApply(identity -> null);
+ }
if (username.isEmpty() || password.isEmpty() || usersClient == null) {
return CompletableFuture.completedFuture(null);
}
@@ -638,6 +675,9 @@ public class AsyncIggyTcpClient {
CompletableFuture<IdentityInfo> callerFuture = new CompletableFuture<>();
transaction.whenComplete((identity, error) -> {
gate.complete(null);
+ if (error == null) {
+ rememberedLogin = loginAttempt;
+ }
if (error != null) {
callerFuture.completeExceptionally(error);
} else {
@@ -732,7 +772,33 @@ public class AsyncIggyTcpClient {
if (currentSystemClient == null) {
return CompletableFuture.completedFuture(Optional.empty());
}
- return LeaderAwareness.findLeaderElsewhere(currentSystemClient::getClusterMetadata, currentTarget);
+ return LeaderAwareness.findLeaderElsewhere(currentSystemClient::getClusterMetadata, currentTarget)
+ .thenApply(lookup -> {
+ // Replaced wholesale rather than merged: the roster is the
+ // cluster's own answer about where its nodes are, so a node
+ // it dropped stops being dialed. The configured seed is
+ // kept separately and outlives it.
+ if (!lookup.endpoints().isEmpty()) {
+ rosterTargets = lookup.endpoints();
+ }
+ return lookup.redirect();
+ });
+ }
+
+ /**
+ * Endpoints a redial rotates through, likeliest first: where the client
+ * currently is, the address it was configured with, then the roster it
+ * learned while connected. Duplicates are dropped, so an endpoint the
+ * roster merely spells differently does not earn a second attempt.
+ */
+ private List<ConnectionInfo> redialCandidates() {
+ List<ConnectionInfo> candidates = new ArrayList<>();
+ candidates.add(connectionInfo);
+ Stream.concat(Stream.of(seedConnectionInfo), rosterTargets.stream())
+ .filter(endpoint ->
+ candidates.stream().noneMatch(candidate -> LeaderAwareness.isSameAddress(candidate, endpoint)))
+ .forEach(candidates::add);
+ return List.copyOf(candidates);
}
CompletableFuture<Void> retarget(ConnectionInfo newTarget) {
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 f541e9de7..3b258b7ef 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
@@ -72,12 +72,12 @@ final class LeaderAwareness {
* exceptionally, so the redirection path cannot fail the login that
* triggered it.
*/
- static CompletableFuture<Optional<ConnectionInfo>> findLeaderElsewhere(
+ static CompletableFuture<LeaderLookup> findLeaderElsewhere(
Supplier<CompletableFuture<ClusterMetadata>> fetchMetadata, ConnectionInfo currentTarget) {
return findLeaderElsewhere(fetchMetadata, currentTarget, LEADERLESS_WAIT_BUDGET, LEADERLESS_POLL_INTERVAL);
}
- static CompletableFuture<Optional<ConnectionInfo>> findLeaderElsewhere(
+ static CompletableFuture<LeaderLookup> findLeaderElsewhere(
Supplier<CompletableFuture<ClusterMetadata>> fetchMetadata,
ConnectionInfo currentTarget,
Duration leaderlessWaitBudget,
@@ -87,7 +87,7 @@ final class LeaderAwareness {
fetchMetadata, currentTarget, leaderlessWaitBudget, leaderlessPollInterval, electionDeadlineNanos);
}
- private static CompletableFuture<Optional<ConnectionInfo>> pollForLeader(
+ private static CompletableFuture<LeaderLookup> pollForLeader(
Supplier<CompletableFuture<ClusterMetadata>> fetchMetadata,
ConnectionInfo currentTarget,
Duration leaderlessWaitBudget,
@@ -99,16 +99,18 @@ final class LeaderAwareness {
} catch (RuntimeException fetchError) {
fetched = CompletableFuture.failedFuture(fetchError);
}
- return fetched.<CompletableFuture<Optional<ConnectionInfo>>>handleAsync((metadata, error) -> {
+ return fetched.<CompletableFuture<LeaderLookup>>handleAsync((metadata, error) -> {
if (error != null) {
log.warn(
"Failed to get cluster metadata: {}, connection will continue on server node {}",
error.getMessage(),
currentTarget.serverAddress());
- return CompletableFuture.completedFuture(Optional.empty());
+ return CompletableFuture.completedFuture(LeaderLookup.inconclusive());
}
LeaderCheck check;
+ List<ConnectionInfo> endpoints;
try {
+ endpoints = nodeTargets(metadata);
check = checkLeader(metadata, currentTarget);
} catch (RuntimeException selectionError) {
log.warn(
@@ -116,10 +118,11 @@ final class LeaderAwareness {
+ " on server node {}",
selectionError.getMessage(),
currentTarget.serverAddress());
- return CompletableFuture.completedFuture(Optional.empty());
+ return CompletableFuture.completedFuture(LeaderLookup.inconclusive());
}
if (check instanceof LeaderCheck.Redirect redirect) {
- return CompletableFuture.completedFuture(Optional.of(redirect.target()));
+ return CompletableFuture.completedFuture(
+ new LeaderLookup(Optional.of(redirect.target()), endpoints));
}
if (check instanceof LeaderCheck.NoLeader) {
if (System.nanoTime() >= electionDeadlineNanos) {
@@ -128,7 +131,9 @@ final class LeaderAwareness {
+ " continue on server node {}",
leaderlessWaitBudget,
currentTarget.serverAddress());
- return CompletableFuture.completedFuture(Optional.empty());
+ // A leaderless roster still names where the nodes
+ // are, and that is what a redial needs.
+ return CompletableFuture.completedFuture(new LeaderLookup(Optional.empty(), endpoints));
}
Executor retryAfterInterval = CompletableFuture.delayedExecutor(
leaderlessPollInterval.toMillis(), TimeUnit.MILLISECONDS);
@@ -142,11 +147,23 @@ final class LeaderAwareness {
retryAfterInterval)
.thenCompose(Function.identity());
}
- return CompletableFuture.completedFuture(Optional.empty());
+ return CompletableFuture.completedFuture(new LeaderLookup(Optional.empty(), endpoints));
})
.thenCompose(Function.identity());
}
+ /**
+ * Every node's target for the tcp transport, in roster order. A node that
+ * does not expose the transport reports port 0 and is skipped: dialing it
+ * would burn a redial attempt on an endpoint that cannot answer.
+ */
+ static List<ConnectionInfo> nodeTargets(ClusterMetadata metadata) {
+ return metadata.nodes().stream()
+ .filter(node -> node.endpoints().tcp() != 0)
+ .map(node -> new ConnectionInfo(node.ip(), node.endpoints().tcp()))
+ .toList();
+ }
+
/**
* One leader-check verdict from a cluster-metadata snapshot.
*/
@@ -248,6 +265,24 @@ final class LeaderAwareness {
/**
* One leader-check verdict from a cluster-metadata snapshot.
*/
+ /**
+ * What one leader check learned from the roster: where to go, and every
+ * node the cluster named for this transport. A client keeps the latter as
+ * redial candidates, because the address it was configured with dies with
+ * its node and the roster is unreachable exactly when it is needed.
+ */
+ record LeaderLookup(Optional<ConnectionInfo> redirect, List<ConnectionInfo> endpoints) {
+
+ LeaderLookup {
+ endpoints = List.copyOf(endpoints);
+ }
+
+ /** A check that learned nothing: stay put, remember no endpoint. */
+ static LeaderLookup inconclusive() {
+ return new LeaderLookup(Optional.empty(), List.of());
+ }
+ }
+
sealed interface LeaderCheck {
/** A healthy leader with an enabled tcp transport lives elsewhere; reconnect to it. */
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LoginRoutingHook.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LoginRoutingHook.java
index a6acb7dc0..904b492e2 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LoginRoutingHook.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LoginRoutingHook.java
@@ -41,4 +41,10 @@ interface LoginRoutingHook {
* @return the identity returned by the successful Register response
*/
CompletableFuture<IdentityInfo> loginOnLeader(Supplier<CompletableFuture<IdentityInfo>> loginAttempt);
+
+ /**
+ * Drops any login kept for replay. Called on an explicit sign-out: there
+ * is no session left to restore, and a redial must not resurrect one.
+ */
+ default void forgetLogin() {}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ReconnectPlan.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ReconnectPlan.java
index ae596731f..3e9f45bf8 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ReconnectPlan.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ReconnectPlan.java
@@ -23,6 +23,7 @@ import org.apache.iggy.client.ConnectionInfo;
import org.apache.iggy.config.RetryPolicy;
import java.time.Duration;
+import java.util.List;
/**
* Pure redial planning: which address to dial on a given reconnect attempt
@@ -33,16 +34,18 @@ final class ReconnectPlan {
private ReconnectPlan() {}
/**
- * Alternates reconnect dials between the current endpoint and the
- * configured seed. After a leader redirect the current endpoint may die
- * with the leader, and the seed is the way back to the rest of the
- * cluster. Attempts are 1-based; odd attempts dial the current endpoint.
+ * Rotates reconnect dials through every endpoint the client knows, in the
+ * order the candidate list gives them. After a leader redirect the current
+ * endpoint may die with the leader, and the rest of the list -- the
+ * configured seed and the roster learned while connected -- is the way
+ * back to the rest of the cluster. Attempts are 1-based; the first dials
+ * the head of the list.
*/
- static ConnectionInfo target(ConnectionInfo current, ConnectionInfo seed, int attempt) {
- if (current.equals(seed)) {
- return current;
+ static ConnectionInfo target(List<ConnectionInfo> candidates, int attempt) {
+ if (candidates.isEmpty()) {
+ throw new IllegalArgumentException("a redial needs at least one candidate endpoint");
}
- return attempt % 2 == 1 ? current : seed;
+ return candidates.get(Math.floorMod(attempt - 1, candidates.size()));
}
/**
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java
index b6e7ab194..1bf1e7ced 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java
@@ -190,6 +190,7 @@ public class UsersTcpClient implements UsersClient {
return connection().send(CommandCode.User.LOGOUT.getValue(), payload).thenAccept(response -> {
response.release();
+ routingHook.forgetLogin();
log.debug("Logged out successfully");
});
}
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEndpointFailoverTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEndpointFailoverTest.java
new file mode 100644
index 000000000..f541701b9
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEndpointFailoverTest.java
@@ -0,0 +1,307 @@
+/*
+ * 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.
+ */
+
+package org.apache.iggy.client.async.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.iggy.config.RetryPolicy;
+import org.junit.jupiter.api.Test;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The node a client signed in on dies; its next request has to complete on a
+ * survivor the roster named, under a session established there. Mirrors
+ * {@code core/integration/tests/cluster/failover_client_continuity.rs}. The
+ * mock VSR framing matches {@link AsyncIggyTcpClientTransientFailoverTest},
+ * kept separate so a death mid-connection cannot disturb that suite's server.
+ */
+class AsyncIggyTcpClientEndpointFailoverTest {
+ private static final int HEADER_SIZE = 256;
+ private static final int SIZE_OFFSET = 48;
+ private static final int COMMAND_OFFSET = 60;
+ private static final int REQUEST_ID_OFFSET = 168;
+ private static final int REQUEST_OPERATION_OFFSET = 176;
+ private static final int REQUEST_CODE_OFFSET = 196;
+ private static final int REPLY_REQUEST_ID_OFFSET = 200;
+ private static final int REPLY_OPERATION_OFFSET = 208;
+ private static final int REPLY_STATUS_OFFSET = 216;
+
+ private static final int COMMAND_REPLY = 8;
+ private static final int OPERATION_REGISTER = 1;
+ private static final int OPERATION_NON_REPLICATED = 2;
+ private static final int GET_CLUSTER_METADATA_CODE = 12;
+ private static final int PING_CODE = 1;
+
+ @Test
+ void shouldResumeOnASurvivorAfterTheSignedInNodeDies() throws Exception {
+ InetAddress loopback = InetAddress.getLoopbackAddress();
+ try (ServerSocket primarySocket = new ServerSocket(0, 4, loopback);
+ ServerSocket survivorSocket = new ServerSocket(0, 4, loopback)) {
+ int primaryPort = primarySocket.getLocalPort();
+ int survivorPort = survivorSocket.getLocalPort();
+ AtomicInteger survivorRegistrations = new AtomicInteger();
+ AtomicInteger survivorPings = new AtomicInteger();
+
+ // The primary leads, so the sign-in settles there and the roster is
+ // only remembered -- not acted on -- until the node dies.
+ MockNode primary = MockNode.serve(primarySocket, request -> {
+ if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) {
+ return Response.success(
+ OPERATION_NON_REPLICATED, clusterMetadata(primaryPort, survivorPort, primaryPort));
+ }
+ if (request.operation() == OPERATION_REGISTER) {
+ return Response.success(OPERATION_REGISTER, registerBody(1));
+ }
+ return Response.success(OPERATION_NON_REPLICATED, Unpooled.EMPTY_BUFFER);
+ });
+ MockNode survivor = MockNode.serve(survivorSocket, request -> {
+ if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) {
+ return Response.success(
+ OPERATION_NON_REPLICATED, clusterMetadata(primaryPort, survivorPort, survivorPort));
+ }
+ if (request.operation() == OPERATION_REGISTER) {
+ survivorRegistrations.incrementAndGet();
+ return Response.success(OPERATION_REGISTER, registerBody(2));
+ }
+ if (request.is(PING_CODE, OPERATION_NON_REPLICATED)) {
+ survivorPings.incrementAndGet();
+ }
+ return Response.success(OPERATION_NON_REPLICATED, Unpooled.EMPTY_BUFFER);
+ });
+
+ AsyncIggyTcpClient client = AsyncIggyTcpClient.builder()
+ .host(loopback.getHostAddress())
+ .port(primaryPort)
+ .credentials("iggy", "iggy")
+ .requestTimeout(Duration.ofSeconds(5))
+ // A redial rotates one endpoint per attempt, so the survivor
+ // is the second: keep the pacing short enough to observe.
+ .retryPolicy(RetryPolicy.fixedDelay(8, Duration.ofMillis(50)))
+ .build();
+ try {
+ client.connect().get(5, TimeUnit.SECONDS);
+ client.login().get(5, TimeUnit.SECONDS);
+ client.sendBinaryRequest(PING_CODE, new byte[0]).get(5, TimeUnit.SECONDS);
+ assertThat(client.getConnectionInfo().port()).isEqualTo(primaryPort);
+
+ primary.kill();
+
+ // The request in flight when the node died is allowed to fail;
+ // what is not allowed is never completing one, which is what a
+ // client that only knows the dead endpoint does.
+ assertThat(resumeWithin(client, Duration.ofSeconds(10)))
+ .as("the client has to resume on the survivor the roster named")
+ .isTrue();
+
+ assertThat(client.getConnectionInfo().port())
+ .as("the client moved off the dead endpoint")
+ .isEqualTo(survivorPort);
+ assertThat(survivorRegistrations)
+ .as("the login was replayed on the survivor")
+ .hasValueGreaterThanOrEqualTo(1);
+ assertThat(survivorPings)
+ .as("the request landed on the survivor")
+ .hasValueGreaterThanOrEqualTo(1);
+ } finally {
+ client.close().get(5, TimeUnit.SECONDS);
+ survivor.close();
+ }
+ }
+ }
+
+ /** Retries until one request completes, or the budget runs out. */
+ private static boolean resumeWithin(AsyncIggyTcpClient client, Duration budget) throws InterruptedException {
+ long deadline = System.nanoTime() + budget.toNanos();
+ while (System.nanoTime() < deadline) {
+ try {
+ client.sendBinaryRequest(PING_CODE, new byte[0]).get(2, TimeUnit.SECONDS);
+ return true;
+ } catch (ExecutionException | TimeoutException stillDown) {
+ Thread.sleep(50);
+ }
+ }
+ return false;
+ }
+
+ private static ByteBuf registerBody(long session) {
+ ByteBuf body = Unpooled.buffer();
+ body.writeIntLE(0);
+ body.writeIntLE(1);
+ body.writeLongLE(session);
+ body.writeIntLE(11 << 10);
+ body.writeByte(0);
+ return body;
+ }
+
+ private static ByteBuf clusterMetadata(int primaryPort, int survivorPort, int leaderPort) {
+ ByteBuf body = Unpooled.buffer();
+ writeString(body, "test-cluster");
+ body.writeIntLE(2);
+ writeNode(body, "primary", primaryPort, primaryPort == leaderPort);
+ writeNode(body, "survivor", survivorPort, survivorPort == leaderPort);
+ return body;
+ }
+
+ private static void writeNode(ByteBuf body, String name, int port, boolean leader) {
+ writeString(body, name);
+ writeString(body, InetAddress.getLoopbackAddress().getHostAddress());
+ body.writeShortLE(port);
+ body.writeShortLE(0);
+ body.writeShortLE(0);
+ body.writeShortLE(0);
+ body.writeByte(leader ? 0 : 1);
+ body.writeByte(0);
+ }
+
+ private static void writeString(ByteBuf body, String value) {
+ byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+ body.writeIntLE(bytes.length);
+ body.writeBytes(bytes);
+ }
+
+ /**
+ * A loopback VSR node that keeps serving every connection it accepts until
+ * it is killed, which drops the live sockets and stops accepting so a
+ * redial is refused the way a dead process refuses one.
+ */
+ private static final class MockNode {
+ private final ServerSocket server;
+ private final List<Socket> accepted = new CopyOnWriteArrayList<>();
+ private volatile boolean killed;
+
+ private MockNode(ServerSocket server) {
+ this.server = server;
+ }
+
+ static MockNode serve(ServerSocket server, RequestHandler handler) {
+ MockNode node = new MockNode(server);
+ CompletableFuture.runAsync(() -> {
+ while (!node.killed) {
+ try {
+ Socket socket = server.accept();
+ node.accepted.add(socket);
+ CompletableFuture.runAsync(() -> node.exchange(socket, handler));
+ } catch (IOException accepted) {
+ return;
+ }
+ }
+ });
+ return node;
+ }
+
+ private void exchange(Socket socket, RequestHandler handler) {
+ try (socket) {
+ InputStream input = socket.getInputStream();
+ OutputStream output = socket.getOutputStream();
+ Request request;
+ while (!killed && (request = readRequest(input)) != null) {
+ writeResponse(output, request, handler.handle(request));
+ }
+ } catch (IOException closed) {
+ // A killed node and a client that went away look the same here.
+ }
+ }
+
+ void kill() throws IOException {
+ killed = true;
+ for (Socket socket : accepted) {
+ socket.close();
+ }
+ server.close();
+ }
+
+ void close() throws IOException {
+ kill();
+ }
+ }
+
+ private static Request readRequest(InputStream input) throws IOException {
+ byte[] header = input.readNBytes(HEADER_SIZE);
+ if (header.length == 0) {
+ return null;
+ }
+ if (header.length != HEADER_SIZE) {
+ throw new EOFException("Truncated VSR request header");
+ }
+ ByteBuffer fields = ByteBuffer.wrap(header).order(ByteOrder.LITTLE_ENDIAN);
+ int size = fields.getInt(SIZE_OFFSET);
+ byte[] body = input.readNBytes(size - HEADER_SIZE);
+ if (body.length != size - HEADER_SIZE) {
+ throw new EOFException("Truncated VSR request body");
+ }
+ return new Request(
+ Byte.toUnsignedInt(header[REQUEST_OPERATION_OFFSET]),
+ fields.getInt(REQUEST_CODE_OFFSET),
+ fields.getLong(REQUEST_ID_OFFSET));
+ }
+
+ private static void writeResponse(OutputStream output, Request request, Response response) throws IOException {
+ byte[] body = new byte[response.body().readableBytes()];
+ response.body().readBytes(body);
+ response.body().release();
+ byte[] header = new byte[HEADER_SIZE];
+ ByteBuffer fields = ByteBuffer.wrap(header).order(ByteOrder.LITTLE_ENDIAN);
+ fields.putInt(SIZE_OFFSET, HEADER_SIZE + body.length);
+ header[COMMAND_OFFSET] = (byte) COMMAND_REPLY;
+ fields.putLong(REPLY_REQUEST_ID_OFFSET, request.requestId());
+ header[REPLY_OPERATION_OFFSET] = (byte) response.operation();
+ fields.putInt(REPLY_STATUS_OFFSET, 0);
+ output.write(header);
+ output.write(body);
+ output.flush();
+ }
+
+ private record Request(int operation, int commandCode, long requestId) {
+ boolean is(int expectedCode, int expectedOperation) {
+ return commandCode == expectedCode && operation == expectedOperation;
+ }
+ }
+
+ private record Response(int operation, ByteBuf body) {
+ static Response success(int operation, ByteBuf body) {
+ return new Response(operation, body);
+ }
+ }
+
+ @FunctionalInterface
+ private interface RequestHandler {
+ Response handle(Request request);
+ }
+}
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 00540cd23..b5b73cfb7 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
@@ -206,6 +206,10 @@ class LeaderAwarenessTest {
private final ConnectionInfo currentTarget = new ConnectionInfo("iggy-follower", 8092);
private Optional<ConnectionInfo> findLeader(Supplier<CompletableFuture<ClusterMetadata>> fetch) {
+ return lookUpLeader(fetch).redirect();
+ }
+
+ private LeaderAwareness.LeaderLookup lookUpLeader(Supplier<CompletableFuture<ClusterMetadata>> fetch) {
return LeaderAwareness.findLeaderElsewhere(fetch, currentTarget, BUDGET, INTERVAL)
.orTimeout(30, TimeUnit.SECONDS)
.join();
@@ -248,7 +252,8 @@ class LeaderAwarenessTest {
Duration.ofMillis(100),
INTERVAL)
.orTimeout(30, TimeUnit.SECONDS)
- .join();
+ .join()
+ .redirect();
assertThat(leader).isEmpty();
assertThat(fetchCount.get()).isGreaterThan(1);
@@ -276,6 +281,22 @@ class LeaderAwarenessTest {
assertThat(leader).isEmpty();
}
+ @Test
+ void shouldRememberEveryNodeTheRosterNamesEvenWhileLeaderless() {
+ var lookup = LeaderAwareness.findLeaderElsewhere(
+ () -> CompletableFuture.completedFuture(leaderlessCluster()),
+ currentTarget,
+ Duration.ofMillis(100),
+ INTERVAL)
+ .orTimeout(30, TimeUnit.SECONDS)
+ .join();
+
+ // A leaderless roster still names where the nodes are, and that is
+ // what a redial needs.
+ assertThat(lookup.redirect()).isEmpty();
+ assertThat(lookup.endpoints()).isNotEmpty();
+ }
+
@Test
void shouldStayWithoutPollingWhenAlreadyOnLeader() {
var fetchCount = new AtomicInteger();
@@ -289,7 +310,8 @@ class LeaderAwarenessTest {
BUDGET,
INTERVAL)
.orTimeout(30, TimeUnit.SECONDS)
- .join();
+ .join()
+ .redirect();
assertThat(leader).isEmpty();
assertThat(fetchCount).hasValue(1);
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/ReconnectPlanTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/ReconnectPlanTest.java
index ce833a58c..fff6114d3 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/ReconnectPlanTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/ReconnectPlanTest.java
@@ -24,26 +24,36 @@ import org.apache.iggy.config.RetryPolicy;
import org.junit.jupiter.api.Test;
import java.time.Duration;
+import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
class ReconnectPlanTest {
private final ConnectionInfo seed = new ConnectionInfo("seed-node", 8090);
private final ConnectionInfo current = new ConnectionInfo("leader-node", 8090);
+ private final ConnectionInfo survivor = new ConnectionInfo("survivor-node", 8090);
@Test
- void shouldAlternateBetweenCurrentAndSeed() {
- assertThat(ReconnectPlan.target(current, seed, 1)).isEqualTo(current);
- assertThat(ReconnectPlan.target(current, seed, 2)).isEqualTo(seed);
- assertThat(ReconnectPlan.target(current, seed, 3)).isEqualTo(current);
- assertThat(ReconnectPlan.target(current, seed, 4)).isEqualTo(seed);
+ void shouldRotateThroughEveryKnownEndpoint() {
+ var candidates = List.of(current, seed, survivor);
+
+ assertThat(ReconnectPlan.target(candidates, 1)).isEqualTo(current);
+ assertThat(ReconnectPlan.target(candidates, 2)).isEqualTo(seed);
+ assertThat(ReconnectPlan.target(candidates, 3)).isEqualTo(survivor);
+ assertThat(ReconnectPlan.target(candidates, 4)).isEqualTo(current);
}
@Test
void shouldDialOnlyOneAddressWhenNeverRedirected() {
- assertThat(ReconnectPlan.target(seed, seed, 1)).isEqualTo(seed);
- assertThat(ReconnectPlan.target(seed, seed, 2)).isEqualTo(seed);
+ assertThat(ReconnectPlan.target(List.of(seed), 1)).isEqualTo(seed);
+ assertThat(ReconnectPlan.target(List.of(seed), 2)).isEqualTo(seed);
+ }
+
+ @Test
+ void shouldRefuseToPlanARedialWithoutCandidates() {
+ assertThatThrownBy(() -> ReconnectPlan.target(List.of(), 1)).isInstanceOf(IllegalArgumentException.class);
}
@Test
diff --git a/foreign/node/src/client/client.connection.test.ts b/foreign/node/src/client/client.connection.test.ts
index 2564cfacf..afb034d09 100644
--- a/foreign/node/src/client/client.connection.test.ts
+++ b/foreign/node/src/client/client.connection.test.ts
@@ -394,6 +394,31 @@ describe('IggyConnection', () => {
}
);
+ it('rotates a redial through the roster it learned while connected',
+ async () => {
+ const seed = await startServer();
+ const seedPort = (seed.address() as AddressInfo).port;
+ const connection = new IggyConnection(connectionConfig(seed));
+ connection.on('error', () => undefined);
+ try {
+ 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 }
+ ]);
+ // The endpoint the client is on leads, the roster follows, and the
+ // roster's copy of that endpoint does not earn a second attempt.
+ assert.deepEqual(
+ connection._redialCandidates().map((options) => options.port),
+ [seedPort, seedPort + 1, seedPort + 2]
+ );
+ } finally {
+ connection._destroy();
+ await new Promise<void>((resolve) => seed.close(() => resolve()));
+ }
+ }
+ );
+
it('settles a dial in flight when a redirect replaces the socket',
async () => {
const seed = await startServer();
diff --git a/foreign/node/src/client/client.connection.ts b/foreign/node/src/client/client.connection.ts
index 10b68639d..0ef16326f 100644
--- a/foreign/node/src/client/client.connection.ts
+++ b/foreign/node/src/client/client.connection.ts
@@ -119,6 +119,13 @@ export class IggyConnection extends EventEmitter {
private reconnectPromise?: Promise<this>;
/** Endpoint the client was configured with, kept across leader redirects */
private readonly seedOptions: ClientConfig['options'];
+ /**
+ * Every node the roster named on the last read, kept as redial candidates.
+ * A node dies together with its address, and the roster is unreachable
+ * exactly when it is needed, so it has to have been remembered while the
+ * connection was still healthy.
+ */
+ private rosterEndpoints: { host: string, port: number }[];
/** Incremental response frame decoder */
private responseDecoder: ResponseFrameDecoder;
@@ -136,6 +143,7 @@ export class IggyConnection extends EventEmitter {
this.ending = false;
this.reconnectOption = { ...DefaultReconnectOption, ...config.reconnect };
this.seedOptions = { ...config.options };
+ this.rosterEndpoints = [];
this.reconnectCount = 0;
this.connectPromise = undefined;
this.reconnectPromise = undefined;
@@ -301,7 +309,6 @@ export class IggyConnection extends EventEmitter {
): Promise<this> {
let lastError = initialError;
let expectedSocket = this.socket;
- let attempt = 0;
while (enabled && this.reconnectCount < maxRetries) {
this.connecting = true;
this.reconnectCount += 1;
@@ -313,24 +320,28 @@ export class IggyConnection extends EventEmitter {
if (this.connected || this.socket !== expectedSocket)
return this.connect();
- const options = this._reconnectTarget(attempt);
- attempt += 1;
- const socket = this._installSocket(
- getTransport({ ...this.config, options })
- );
- this.socket = socket;
- expectedSocket = socket;
- try {
- await this._waitForConnection(socket);
- if (this.socket !== socket)
- return this.connect();
- this.config.options = options;
- return this;
- } catch (error) {
- lastError = error instanceof Error
- ? error
- : new Error(String(error));
- debug('reconnect attempt failed', lastError);
+ // Every endpoint gets its turn inside one attempt, so a full pass over
+ // the cluster costs one retry rather than one per endpoint: a pass that
+ // stopped at the first refusal would never reach the survivors of a
+ // client configured for a single retry.
+ for (const options of this._redialCandidates()) {
+ const socket = this._installSocket(
+ getTransport({ ...this.config, options })
+ );
+ this.socket = socket;
+ expectedSocket = socket;
+ try {
+ await this._waitForConnection(socket);
+ if (this.socket !== socket)
+ return this.connect();
+ this.config.options = options;
+ return this;
+ } catch (error) {
+ lastError = error instanceof Error
+ ? error
+ : new Error(String(error));
+ debug('reconnect attempt failed', lastError);
+ }
}
}
@@ -342,16 +353,43 @@ export class IggyConnection extends EventEmitter {
}
/**
- * Alternates reconnect dials between the current endpoint and the
- * configured seed. After a leader redirect the current endpoint may die
- * with the leader, and the seed is the way back to the rest of the cluster.
+ * Records the cluster roster as redial candidates.
+ *
+ * Replaced wholesale rather than merged: the roster is the cluster's own
+ * answer about where its nodes are, so a node it dropped stops being
+ * dialed. The configured seed is kept separately and outlives it.
+ */
+ rememberRoster(endpoints: { host: string, port: number }[]): void {
+ if (endpoints.length === 0)
+ return;
+ this.rosterEndpoints = endpoints;
+ }
+
+ /**
+ * Endpoints a redial rotates through, likeliest first: where the client
+ * currently is, the endpoint it was configured with, then the roster it
+ * learned while connected. After a leader redirect the current endpoint may
+ * die with the leader, and the rest of the list is the way back to the
+ * cluster. Duplicates are dropped, so an endpoint the roster merely spells
+ * differently does not earn a second attempt.
*/
- private _reconnectTarget(attempt: number): ClientConfig['options'] {
- const current = this.config.options;
- if (this.seedOptions.host === current.host &&
- this.seedOptions.port === current.port)
- return current;
- return attempt % 2 === 0 ? current : this.seedOptions;
+ _redialCandidates(): ClientConfig['options'][] {
+ const candidates = [this.config.options];
+ const known = [
+ this.seedOptions,
+ ...this.rosterEndpoints.map(
+ ({ host, port }) => ({ ...this.config.options, host, port })
+ )
+ ];
+ for (const candidate of known) {
+ const duplicate = candidates.some(
+ (known) => known.port === candidate.port &&
+ normalizeHost(known.host) === normalizeHost(candidate.host)
+ );
+ if (!duplicate)
+ candidates.push(candidate);
+ }
+ return candidates;
}
async redirect(host: string, port: number) {
diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts
index 0697c5f3d..07cbcecd4 100644
--- a/foreign/node/src/client/client.socket.test.ts
+++ b/foreign/node/src/client/client.socket.test.ts
@@ -505,6 +505,111 @@ describe('VSR client socket', () => {
}
});
+ // The node a client authenticated on dies; its next command has to complete
+ // on a survivor the roster named, under a session established there.
+ // Mirrors `core/integration/tests/cluster/failover_client_continuity.rs`.
+ it('resumes on a survivor after the node it authenticated on dies',
+ async () => {
+ const primarySockets = new Set<Socket>();
+ let primaryDead = false;
+
+ const survivor = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
+ if (code === COMMAND_CODE.GetClusterMetadata) {
+ // The survivor leads once the primary is gone.
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(primary.port, survivor.port)
+ ));
+ return;
+ }
+ socket.write(replyFrame(operation));
+ });
+
+ const primary = await startVsrServer((frame, socket) => {
+ primarySockets.add(socket);
+ if (primaryDead) {
+ socket.destroy();
+ return;
+ }
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
+ if (code === COMMAND_CODE.GetClusterMetadata) {
+ // The primary leads, so the login settles here and the roster is
+ // only remembered, not acted on, until the node dies.
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(survivor.port, primary.port)
+ ));
+ return;
+ }
+ socket.write(replyFrame(operation));
+ });
+
+ const config: ClientConfig = {
+ ...vsrConfig(primary.port),
+ reconnect: { enabled: true, interval: 1, maxRetries: 3 }
+ };
+ const client = new CommandResponseStream(config);
+ try {
+ await client.authenticate(config.credentials);
+ await client.sendCommand(60_021, Buffer.alloc(0));
+ assert.ok(
+ primary.frames.some(
+ (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_021
+ ),
+ 'the live primary answered the first command'
+ );
+
+ primaryDead = true;
+ for (const socket of primarySockets)
+ socket.destroy();
+ await primary.close();
+
+ // The attempt in flight when the socket died is allowed to fail; what
+ // is not allowed is never completing one, which is what a client that
+ // only knows the dead endpoint does.
+ let resumed = false;
+ let lastError: unknown;
+ for (let attempt = 0; attempt < 20 && !resumed; attempt += 1) {
+ try {
+ await client.sendCommand(60_021, Buffer.alloc(0));
+ resumed = true;
+ } catch (error) {
+ lastError = error;
+ await new Promise((resolve) => setTimeout(resolve, 10));
+ }
+ }
+ assert.ok(resumed, `the client never resumed: ${String(lastError)}`);
+
+ const operations = survivor.frames.map(
+ (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
+ );
+ assert.ok(
+ operations.includes(Operation.Register),
+ 'the client signed in again on the survivor'
+ );
+ assert.ok(
+ survivor.frames.some(
+ (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_021
+ ),
+ 'the command landed on the survivor the roster named'
+ );
+ } finally {
+ client.destroy();
+ await survivor.close();
+ }
+ });
+
it('keeps a single-node login on its node', async () => {
const server = await startVsrServer(
(frame, socket) => singleNodeHandler(server.port)(frame, socket)
diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts
index d1e5e4fe2..4d57387e7 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -479,6 +479,13 @@ export class CommandResponseStream extends EventEmitter {
{ last: false }
);
const metadata = GET_CLUSTER_METADATA.deserialize(response);
+ // Every read feeds the redial candidates, leaderless ones included: a
+ // roster with no leader still names where the nodes are.
+ this.connection.rememberRoster(
+ metadata.nodes
+ .filter((node) => node.endpoints.tcp !== 0)
+ .map((node) => ({ host: node.ip, port: node.endpoints.tcp }))
+ );
if (metadata.nodes.length <= 1)
return undefined;
const leader = metadata.nodes.find(