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_reconnection_config.rs

@@ -20,10 +20,43 @@ use std::str::FromStr;
 
 #[derive(Debug, Clone)]
 pub struct TcpClientReconnectionConfig {
+    /// Whether a lost connection is redialed at all. With this off the
+    /// endpoints the client knows still get one pass, since they were
+    /// configured to be tried, but nothing is retried after it.
     pub enabled: bool,
+    /// How many passes over the known endpoints *after the first*, or `None`
+    /// for unlimited. `Some(0)` still makes that one pass, since the endpoints
+    /// were configured to be tried.
+    ///
+    /// Passes, not dials: one pass tries the endpoint the client is on, the
+    /// addresses it was configured with, and every node the roster named, so a
+    /// survivor is reached inside the first pass rather than one delay per
+    /// endpoint.
+    ///
+    /// The number is not portable across SDKs. Each counts the same setting in
+    /// its own terms, and `0` means something different in every one of them, so
+    /// a deployment that runs several has to set this per SDK rather than copy
+    /// one value across:
+    ///
+    /// | SDK  | `N`                                    | `0`                                       | unlimited        |
+    /// | ---- | -------------------------------------- | ----------------------------------------- | ---------------- |
+    /// | Rust | `N` passes after a first, unpaced one   | that first pass alone                     | `None`           |
+    /// | C#   | as Rust                                | unlimited                                 | `0`              |
+    /// | Go   | `N` passes, the first one of them       | unlimited                                 | `0`              |
+    /// | Java | `N` passes, the first one of them       | one pass, and only with several endpoints  | a large `N`      |
+    /// | Node | `N` passes, the first one of them       | no pass at all                            | a large `N`      |
     pub max_retries: Option<u32>,
-    /// Delay between connection attempts.
+    /// Delay between passes. The first pass runs at once when the client knows
+    /// more than one endpoint.
     pub interval: NonZeroIggyDuration,
+    /// Cooldown before redialing the endpoint of the last successful
+    /// connection, measured from when that connection was established rather
+    /// than from when it was lost: a session that outlived this interval is
+    /// redialed with no wait at all, which is the point -- the pace limit is
+    /// there for connections that keep dropping straight away.
+    ///
+    /// Owed to that endpoint alone: the others are dialed without waiting, and
+    /// the paced one goes last in the pass.
     pub reestablish_after: IggyDuration,
 }
 

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

core/integration/tests/sdk/disconnect_relogin.rs

@@ -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.
+
+//! An explicit disconnect ends the session for good: the credentials a manual
+//! sign-in remembered (for reconnecting across involuntary drops and
+//! failovers) must not resurrect it. Pins at the Rust layer the contract the
+//! C++ e2e suite asserts through the FFI (`DisconnectThenReconnectWithoutRelogin`,
+//! `GetStatsBeforeLoginThrows`), so a regression fails here first instead of
+//! three suites downstream.
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+
+#[iggy_harness]
+async fn given_a_logged_in_client_when_explicitly_disconnected_should_require_a_fresh_login(
+    harness: &TestHarness,
+) {
+    let client = harness.new_client().await.unwrap();
+    client
+        .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+        .await
+        .unwrap();
+    client.get_me().await.expect("authenticated get_me works");
+
+    client.disconnect().await.unwrap();
+    client.connect().await.unwrap();
+    assert!(
+        matches!(client.get_me().await, Err(IggyError::Unauthenticated)),
+        "an explicit disconnect is caller intent, like a logout: the sign-in it ended \
+         must not be silently replayed by the reconnect, so the client's own \
+         authentication gate refuses the request before it is sent"
+    );
+
+    client.disconnect().await.unwrap();
+    assert!(
+        matches!(client.get_stats().await, Err(IggyError::NotConnected)),
+        "an operation after an explicit disconnect must fail on the dead transport \
+         instead of reconnecting into a resurrected session"
+    );
+
+    // The remembered sign-in exists for involuntary drops; a fresh manual
+    // login after the disconnect works exactly as before.
+    client.connect().await.unwrap();
+    client
+        .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+        .await
+        .unwrap();
+    client
+        .get_me()
+        .await
+        .expect("a fresh login restores service");
+}

core/integration/tests/sdk/mod.rs

@@ -18,6 +18,7 @@
 mod consumer_group;
 mod consumer_group_membership;
 mod consumer_offset;
+mod disconnect_relogin;
 mod hello_world;
 mod http_refresh;
 mod options;

core/sdk/src/clients/binary_personal_access_tokens.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::clients::redirect_login_settled;
 use crate::prelude::{ClientWrapper, IggyClient};
 use async_trait::async_trait;
 use iggy_common::locking::IggyRwLockFn;
@@ -77,6 +78,15 @@ impl PersonalAccessTokenClient for IggyClient {
         if should_redirect {
             info!("Redirected to leader, reconnecting and re-authenticating");
             self.connect().await?;
+            // The reconnect signs in with the credentials this very call just
+            // remembered, so on a client without a configured `AutoLogin` the
+            // session is already this user's: signing in again would cost a
+            // logout and a second login (an argon2 each, on the server) for
+            // nothing. With `AutoLogin::Enabled` the reconnect signed in the
+            // configured user, who may not be this one, so the login runs.
+            if redirect_login_settled(&*self.client.read().await).await {
+                return Ok(identity);
+            }
             self.login_with_personal_access_token(token).await
         } else {
             Ok(identity)