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

core/binary_protocol/src/consensus/error.rs

@@ -79,6 +79,12 @@ pub enum ConsensusError {
     #[error("invalid bit pattern in header (enum discriminant out of range)")]
     InvalidBitPattern,
 
+    #[error(
+        "operation {operation:#04x} is not known to this build; the sender runs a release that \
+         added a consensus operation, so this node cannot journal or ack the frame"
+    )]
+    UnsupportedOperation { operation: u8 },
+
     #[error("client-bound command {0:?} cannot be dispatched on inbound path")]
     ClientBoundCommand(Command),
 }

core/binary_protocol/src/consensus/header.rs

@@ -103,6 +103,15 @@ pub fn frame_checksum_bytes(header: &[u8; HEADER_SIZE]) -> u128 {
 pub trait ConsensusHeader: Sized + CheckedBitPattern + NoUninit {
     const COMMAND: Command;
 
+    /// Byte offset of this header's `operation` field, `None` when it carries
+    /// none.
+    ///
+    /// The typed decode reads the raw byte here after a failed checked cast,
+    /// so an operation a newer release added is reported as version skew
+    /// rather than corruption. An offset rather than a getter because the cast
+    /// has already failed by then, so no typed view of the header exists.
+    const OPERATION_OFFSET: Option<usize> = None;
+
     /// Whether a frame carrying `command` may be typed as this header.
     /// Defaults to an exact match; a header that serves several commands
     /// with one layout (e.g. `RepairDone` / `RangeEvicted`) widens it.
@@ -477,6 +486,7 @@ fn validate_request_fields(
 }
 
 impl ConsensusHeader for RoutedRequestHeader {
+    const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation));
     const COMMAND: Command = Command::Request;
     /// The client-wire [`RequestHeader`] this is promoted from is unsealed, and the
     /// promotion copies `checksum` verbatim, so there is nothing here to verify.
@@ -511,6 +521,7 @@ impl ConsensusHeader for RoutedRequestHeader {
 }
 
 impl ConsensusHeader for RequestHeader {
+    const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation));
     const COMMAND: Command = Command::Request;
     const FRAME_SEALED: bool = false;
 
@@ -621,6 +632,7 @@ impl Default for ReplyHeader {
 }
 
 impl ConsensusHeader for ReplyHeader {
+    const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation));
     const COMMAND: Command = Command::Reply;
     const FRAME_SEALED: bool = false;
 
@@ -971,6 +983,7 @@ impl Default for PrepareHeader {
 }
 
 impl ConsensusHeader for PrepareHeader {
+    const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation));
     const COMMAND: Command = Command::Prepare;
     const FRAME_SEALED: bool = false;
 
@@ -1178,6 +1191,7 @@ impl Default for PrepareOkHeader {
 }
 
 impl ConsensusHeader for PrepareOkHeader {
+    const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation));
     const FRAME_SEALED: bool = true;
 
     const COMMAND: Command = Command::PrepareOk;

core/binary_protocol/src/consensus/operation.rs

@@ -98,6 +98,17 @@ pub enum Operation {
 }
 
 impl Operation {
+    /// Whether `code` is a discriminant this build defines.
+    ///
+    /// The typed decode needs this to tell an operation a newer release added
+    /// from a corrupted header byte: bytemuck's checked cast rejects both with
+    /// one undifferentiated error, and only the former is fixable by upgrading
+    /// this node.
+    #[must_use]
+    pub fn is_known_code(code: u8) -> bool {
+        bytemuck::checked::try_cast::<u8, Self>(code).is_ok()
+    }
+
     pub const INTERNAL_START: u8 = Self::CreateTopicWithAssignments as u8;
     pub const METADATA_START: u8 = Self::CreateStream as u8;
     pub const PARTITION_START: u8 = Self::SendMessages as u8;

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,

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;

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.

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

@@ -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");