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 <[email protected]> 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(
