feat(sdk): fail over to a surviving node when the current one dies

bdd/rust/tests/helpers/cluster.rs

@@ -17,6 +17,7 @@
 
 use iggy::prelude::*;
 use std::env;
+use std::net::{SocketAddr, ToSocketAddrs};
 use std::sync::Arc;
 
 /// Resolves server address based on role and port, checking environment variables first
@@ -49,6 +50,24 @@ pub async fn create_and_connect_client(addr: &str) -> IggyClient {
     IggyClient::create(ClientWrapper::Tcp(client), None, None)
 }
 
+/// Whether two `host:port` spellings name the same endpoint.
+///
+/// A client that never redirected still holds the address it was given (a
+/// host name, in the BDD compose network), while a redirected one holds the
+/// address the roster published (an IP). Both name the same node, so they
+/// are compared once resolved, like the Go and Java suites do.
+pub fn is_same_endpoint(left: &str, right: &str) -> Result<bool, String> {
+    let resolve = |address: &str| -> Result<Vec<SocketAddr>, String> {
+        address
+            .to_socket_addrs()
+            .map(Iterator::collect)
+            .map_err(|error| format!("Failed to resolve server address {address}: {error}"))
+    };
+    let left = resolve(left)?;
+    let right = resolve(right)?;
+    Ok(left.iter().any(|candidate| right.contains(candidate)))
+}
+
 /// Verifies that a client is connected to the expected port
 pub async fn verify_client_connection(
     client: &IggyClient,

bdd/rust/tests/steps/leader_redirection.rs

@@ -288,10 +288,16 @@ async fn then_both_use_same_server(world: &mut LeaderContext) {
     let conn_info_a = client_a.get_connection_info().await;
     let conn_info_b = client_b.get_connection_info().await;
 
-    // Verify both clients are connected to the same server
-    assert_eq!(
-        conn_info_a.server_address, conn_info_b.server_address,
-        "Both clients should be connected to the same server"
+    // Verify both clients are connected to the same server. Client A holds
+    // the address it was configured with and client B the one the roster
+    // published for the leader, so the spellings differ even when the node
+    // is the same.
+    assert!(
+        cluster::is_same_endpoint(&conn_info_a.server_address, &conn_info_b.server_address)
+            .expect("Server addresses should resolve"),
+        "Both clients should be connected to the same server, got {} and {}",
+        conn_info_a.server_address,
+        conn_info_b.server_address
     );
 
     // Verify both can communicate

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,11 @@ 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())),
+            wire_resp.user_id,
+        )
+        .await;
         self.publish_event(DiagnosticEvent::SignedIn).await;
         Ok(IdentityInfo {
             user_id: wire_resp.user_id,

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;
@@ -174,6 +174,7 @@ impl<B: BinaryClient> UserClient for B {
             .to_bytes(),
         )
         .await?;
+        self.refresh_session_password(user_id, new_password).await;
         Ok(())
     }
 
@@ -218,6 +219,14 @@ 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()),
+            ),
+            wire_resp.user_id,
+        )
+        .await;
         self.publish_event(DiagnosticEvent::SignedIn).await;
         Ok(IdentityInfo {
             user_id: wire_resp.user_id,
@@ -229,6 +238,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;

core/common/src/traits/binary_transport.rs

@@ -15,7 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::{ClientState, DiagnosticEvent, IggyError, NonZeroIggyDuration};
+use crate::{
+    ClientState, Credentials, DiagnosticEvent, Identifier, IggyError, NonZeroIggyDuration,
+};
 use async_trait::async_trait;
 use bytes::Bytes;
 use std::sync::Arc;
@@ -51,6 +53,26 @@ 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, _user_id: u32) {}
+    /// 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) {}
+    /// A committed password change for `user`: when it is the signed-in user,
+    /// the credentials the next reconnect signs in with switch to the new
+    /// password, or that reconnect would replay the old one and fail an
+    /// unrelated request with `InvalidCredentials`. Other users' changes are
+    /// ignored.
+    ///
+    /// This covers a configured `AutoLogin` as well as a sign-in the caller
+    /// ran: the configured credentials still decide *who* the client signs in
+    /// as, and a committed change decides what that user's password is.
+    async fn refresh_session_password(&self, _user: &Identifier, _new_password: &str) {}
     /// 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.

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(),

core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs

@@ -23,6 +23,7 @@ use crate::{
 /// 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.
@@ -44,6 +45,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;
@@ -108,6 +116,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)
     }
@@ -185,6 +197,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");

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"
+    );
 }