diff --git a/core/binary_protocol/src/consensus/error.rs b/core/binary_protocol/src/consensus/error.rs index 02fb044e64..8b6a4ba289 100644 --- a/core/binary_protocol/src/consensus/error.rs +++ b/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), } diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs index 2f205df3d9..8fe0150cc9 100644 --- a/core/binary_protocol/src/consensus/header.rs +++ b/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 = 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 = 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 = 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 = 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 = 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 = Some(core::mem::offset_of!(Self, operation)); const FRAME_SEALED: bool = true; const COMMAND: Command = Command::PrepareOk; diff --git a/core/binary_protocol/src/consensus/operation.rs b/core/binary_protocol/src/consensus/operation.rs index 691bd9f1b7..95de1d4eed 100644 --- a/core/binary_protocol/src/consensus/operation.rs +++ b/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::(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; 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 7e1299f284..d46fef8ba0 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 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 eb785109a6..c199f76de9 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 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 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 d3c2bd2e3a..9c6a923b62 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 86997c6d7a..e3584130fc 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, /// 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> for TcpClientConfig { fn from(connection_string: ConnectionString) -> 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 6f665a5777..607573279d 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) -> 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 { 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/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index 1294558e44..506a61aa99 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -103,6 +103,34 @@ pub const CLIENTS_TABLE_SLOT_MAX: usize = 1 << 16; /// refcount bumps and are never persisted or transferred. pub const REPLY_RING_CAPACITY: usize = 5; +/// What capacity eviction keeps after reclaiming an entry's replies. +/// +/// At-most-once needs only the fence: the watermark says which request numbers +/// already committed, and the ring merely supplies the bytes to replay. Dropping +/// the whole entry made an evicted client's resume mint at watermark zero, so the +/// retry of a committed request re-executed. Keeping it lets the resume answer +/// from the fence instead: [`RequestStatus::Duplicate`] when the watermark's +/// reply was still ringed, [`RequestStatus::AlreadyApplied`] when it was not. +/// +/// In memory only, like the ring: a fence does not survive a checkpoint or a state +/// transfer, and losing one degrades a resume to the pre-fix behaviour rather than +/// corrupting anything. +#[derive(Debug)] +struct EvictedFence { + client_id: u128, + epoch: u64, + user_id: u32, + watermark: u64, + watermark_checksum: u128, + /// The watermark request's own reply when the ring still held it, so the + /// retry the resume contract prescribes replays its original bytes. `None` + /// when it had already aged out, or when a rebind left the register reply + /// as the newest entry: the resume then answers + /// [`RequestStatus::AlreadyApplied`], which still never re-executes. + /// One refcount bump, not a copy. + latest: Option, +} + /// Per-session entry: fence epoch + committed-request watermark + replies. /// /// The key (`client_id` today, the stable `session_id` once SDK identity @@ -407,6 +435,17 @@ pub struct ClientTable { slots: Vec>, /// `client_id` -> slot index. Rebuilt on decode. index: HashMap, + /// Fences of clients capacity eviction reclaimed, oldest at the front. + /// + /// Bounded by the slot count. A fence is the entry's header fields plus, at + /// most, the watermark request's own reply, so it costs a fraction of the + /// entry it replaces. Trimmed oldest-first. + /// + /// Replica-local best-effort, NOT replicated state: the bound is + /// `slots.len()`, which `from_snapshot` and `decode` size per node, and a + /// state transfer replaces the table wholesale. Losing a fence degrades a + /// resume to the pre-fence behaviour; it never makes one more permissive. + evicted_fences: VecDeque, } /// Whether two integrity stamps for the same request number disagree. @@ -426,6 +465,7 @@ impl ClientTable { Self { slots, index: HashMap::with_capacity(max_clients), + evicted_fences: VecDeque::new(), } } @@ -558,7 +598,12 @@ impl ClientTable { latest_commit, }); } - Ok(Self { slots, index }) + Ok(Self { + slots, + index, + // Fences are in-memory only; a restored table starts with none. + evicted_fences: VecDeque::new(), + }) } /// Check a request against the table. Epoch fence first, then the @@ -654,6 +699,11 @@ impl ClientTable { /// /// Full table evicts the oldest commit, see [`Self::evict_oldest`]. /// + /// A key this table evicted for capacity re-registers as a fresh entry that + /// RESTORES the evicted watermark (and the watermark reply when it survived) + /// from [`EvictedFence`], for the same `user_id` only. A committed `Logout` + /// forgets that fence, so a register after one starts clean. + /// /// # Panics /// If `client_id == 0` or `client_id != reply.header().client`. pub fn commit_register(&mut self, client_id: u128, user_id: u32, reply: Message) { @@ -689,6 +739,17 @@ impl ClientTable { .retain(|stored| stored.header().request != REGISTER_REQUEST_ID); entry.push_latest(cached); } else { + // A client this table evicted for capacity is resuming, not + // arriving: its committed request numbers must stay deduped, or the + // retry the resume contract prescribes re-executes. The watermark's + // own reply comes back with the fence when the ring still held it, + // so that retry replays its bytes; every other retry at or below the + // watermark answers `AlreadyApplied`, which also never re-executes. + // + // Same identity only: `client_id` is client-supplied, so a fence + // must never hand one user another user's dedup history, nor its + // cached reply bytes, merely because the key was reused. + let fence = self.take_fence(client_id, user_id); let freed = if self.index.len() >= self.slots.len() { self.evict_oldest() } else { @@ -697,16 +758,27 @@ impl ClientTable { let slot_idx = freed .or_else(|| self.first_free_slot()) .expect("eviction must free a slot"); + debug_assert!( + fence.as_ref().is_none_or(|fence| epoch > fence.epoch), + "commit_register: revived fence epoch regression" + ); let latest_commit = cached.header().commit; let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY); + // Oldest at the front: the retained reply committed before this + // register did, and `latest()` must stay the register's own reply. + if let Some(replay) = fence.as_ref().and_then(|fence| fence.latest.as_ref()) { + ring.push_back(replay.clone()); + } ring.push_back(cached); self.slots[slot_idx] = Some(ClientEntry { epoch, user_id, client_id, latest_commit, - watermark: REGISTER_REQUEST_ID, - watermark_checksum: 0, + watermark: fence + .as_ref() + .map_or(REGISTER_REQUEST_ID, |fence| fence.watermark), + watermark_checksum: fence.as_ref().map_or(0, |fence| fence.watermark_checksum), ring, }); self.index.insert(client_id, slot_idx); @@ -797,9 +869,14 @@ impl ClientTable { .find(|stored| stored.header().request == new_request) { *stored = cached; - // The watermark's reply is the ring's back, so replacing it in - // place moves the latest commit without a push. - entry.latest_commit = new_commit; + // Re-derived, not assigned from `new_commit`: the replaced entry + // is not necessarily the ring's back. A rebind pushes the + // register reply last, and a fence-revived entry carries the + // watermark's reply at the front, so assuming otherwise lets + // `latest_commit` disagree with what `decode` rebuilds from + // `ring.back()` -- and it is `evict_oldest`'s only ranking key, + // so the two would pick different victims from one log. + entry.latest_commit = entry.latest().header().commit; } else { entry.push_latest(cached); } @@ -830,6 +907,13 @@ impl ClientTable { /// /// [`Operation::Register`]: iggy_binary_protocol::Operation pub fn remove_client(&mut self, client_id: u128) -> bool { + // A committed Logout is the explicit end of the session, so it also + // forgets any fence a capacity eviction left for this key. Otherwise a + // Logout that commits after the eviction (its prepare predates it, so + // there is no entry left to drop) would strand a fence, and a later + // register would revive a watermark the client had already ended. + self.evicted_fences + .retain(|fence| fence.client_id != client_id); let Some(slot_idx) = self.index.remove(&client_id) else { return false; }; @@ -849,9 +933,14 @@ impl ClientTable { /// session that every backup drops. /// /// A client with an uncommitted prepare is therefore evictable. Its - /// commit lands as [`CommitReply::NoEntry`] -- the reply still ships, and - /// the client learns the session is gone on its next request (`NoSession` - /// -> eviction frame -> re-register). + /// commit lands as [`CommitReply::NoEntry`] -- the reply still ships, the + /// client learns the session is gone on its next request (`NoSession` -> + /// eviction frame -> re-register), and that commit reaches no fence, so a + /// resume can re-execute exactly that request. + /// + /// The evicted session's dedup fence survives via [`Self::remember_fence`] + /// unless it had committed nothing or the fence is later trimmed, so the + /// re-registering client is normally answered rather than re-executed. /// /// **Caveat**: eviction erases the evicted session's watermark, so its /// next retry is treated as `New` (re-executes). Bounded by table @@ -876,6 +965,10 @@ impl ClientTable { let (slot_idx, _) = evictee?; let entry = self.slots[slot_idx].take().expect("evictee must exist"); self.index.remove(&entry.client_id); + // Reclaim the replies, keep the fence: the evicted client's own resume + // must not read as a first-time register, or the retry of a committed + // request re-executes. + self.remember_fence(&entry); trace!( client_id = entry.client_id, "evict_oldest: removed client from session table" @@ -883,6 +976,56 @@ impl ClientTable { Some(slot_idx) } + /// Record an evicted entry's dedup fence, trimming oldest-first. + fn remember_fence(&mut self, entry: &ClientEntry) { + // Nothing committed under this session, so there is nothing to dedup. + // Worth skipping rather than storing: `evict_oldest` ranks on the oldest + // `latest_commit`, and a session idle since its register carries its own + // register op, which makes these the PREFERRED victims -- storing them + // would crowd real fences out of a store bounded by the slot count. + if entry.watermark == REGISTER_REQUEST_ID { + return; + } + // One fence per identity: a later eviction supersedes the earlier one, + // and two fences for one key would let the older (lower) watermark be + // found first and revive a stale one. + self.evicted_fences + .retain(|fence| fence.client_id != entry.client_id || fence.user_id != entry.user_id); + self.evicted_fences.push_back(EvictedFence { + client_id: entry.client_id, + epoch: entry.epoch, + user_id: entry.user_id, + watermark: entry.watermark, + watermark_checksum: entry.watermark_checksum, + latest: entry.find_cached(entry.watermark).cloned(), + }); + while self.evicted_fences.len() > self.slots.len() { + self.evicted_fences.pop_front(); + } + } + + /// Take back the fence a previous capacity eviction left for this + /// `(client_id, user_id)` pair. The identity half is the security-relevant + /// one: `client_id` arrives off the wire. + /// + /// Linear because it runs only on a register that missed the index, which is + /// a consensus commit and already far dearer than a scan of at most + /// `slots.len()` fences. + fn take_fence(&mut self, client_id: u128, user_id: u32) -> Option { + // Both fields in the predicate, not a client_id match with the identity + // checked afterwards: `client_id` is client-supplied, so the store can + // legitimately hold one fence per user for the same key. Matching on the + // id alone would let whichever fence sits nearer the front shadow the + // caller's own, handing it a fresh watermark and re-executing a request + // it had already committed. It also leaves another user's fence in place + // rather than consuming it. + let position = self + .evicted_fences + .iter() + .position(|fence| fence.client_id == client_id && fence.user_id == user_id)?; + self.evicted_fences.remove(position) + } + fn first_free_slot(&self) -> Option { self.slots.iter().position(Option::is_none) } @@ -1245,6 +1388,184 @@ mod tests { /// assert on it (see `register_stores_user_id` for the accessor check). const TEST_USER_ID: u32 = 7; + /// Capacity eviction reclaims an entry's replies but must not reset its + /// dedup fence: the evicted client's own resume is a rebind in everything + /// but bookkeeping, and the resume contract has it retry the request it + /// never saw answered. + #[test] + fn eviction_keeps_the_fence_so_a_resumed_client_is_not_re_executed() { + const CLIENT_A: u128 = 0xA11CE; + const CHURN: [u128; 2] = [0xB0B1, 0xB0B2]; + + let mut table = ClientTable::new(2); + table.commit_register(CLIENT_A, TEST_USER_ID, make_register_reply(CLIENT_A, 1)); + // Request 1 commits for A, so its watermark is 1. + table.commit_reply(CLIENT_A, make_reply_for(CLIENT_A, 1, 2)); + + // Two fresh registers fill the table and evict A (oldest commit). + for (offset, churn) in CHURN.iter().enumerate() { + let commit = 3 + offset as u64; + table.commit_register(*churn, TEST_USER_ID, make_register_reply(*churn, commit)); + } + assert!( + table.get_epoch(CLIENT_A).is_none(), + "the churn must have evicted A for this test to mean anything" + ); + + // A resumes: fresh register under the same id, then retries request 1. + table.commit_register(CLIENT_A, TEST_USER_ID, make_register_reply(CLIENT_A, 10)); + let resumed_epoch = table.get_epoch(CLIENT_A).expect("resume registered"); + + match table.check_request(CLIENT_A, resumed_epoch, 1, 0) { + RequestStatus::Duplicate(replayed) => { + assert_eq!( + replayed.header().request, + 1, + "the retained reply must be the watermark request's own" + ); + } + other => panic!( + "a committed request retried after capacity eviction must replay its cached \ + reply, not be executed a second time; got {other:?}" + ), + } + + // A request above the restored watermark is still new. + assert!(matches!( + table.check_request(CLIENT_A, resumed_epoch, 2, 0), + RequestStatus::New + )); + } + + /// `client_id` is client-supplied, so a fence belongs to the user that + /// earned it: a register under a different identity must neither inherit the + /// dedup history (it would be handed another user's cached reply bytes) nor + /// consume the fence (anyone could then erase another client's history just + /// by presenting its key). + #[test] + fn a_fence_is_neither_inherited_nor_consumed_by_a_different_user() { + const CLIENT_A: u128 = 0xA11CE; + const OTHER_USER: u32 = TEST_USER_ID + 1; + + let mut table = ClientTable::new(2); + table.commit_register(CLIENT_A, TEST_USER_ID, make_register_reply(CLIENT_A, 1)); + table.commit_reply(CLIENT_A, make_reply_for(CLIENT_A, 1, 2)); + for (offset, churn) in [0xB0B1u128, 0xB0B2].iter().enumerate() { + let commit = 3 + offset as u64; + table.commit_register(*churn, TEST_USER_ID, make_register_reply(*churn, commit)); + } + assert!( + table.get_epoch(CLIENT_A).is_none(), + "the churn must have evicted A, leaving its fence" + ); + + table.commit_register(CLIENT_A, OTHER_USER, make_register_reply(CLIENT_A, 10)); + let squatter_epoch = table.get_epoch(CLIENT_A).expect("registered"); + assert!( + matches!( + table.check_request(CLIENT_A, squatter_epoch, 1, 0), + RequestStatus::New + ), + "a different user must start at a fresh watermark, not inherit the fence" + ); + assert!( + table + .evicted_fences + .iter() + .any(|fence| fence.client_id == CLIENT_A && fence.user_id == TEST_USER_ID), + "the owner's fence must survive a register under another identity" + ); + } + + /// Two fences can share a `client_id` with different users (the key is + /// client-supplied). Lookup must find the caller's own fence rather than + /// stopping at whichever one happens to sit closer to the front. + #[test] + fn a_fence_is_found_behind_another_users_fence_for_the_same_client_id() { + const CLIENT_A: u128 = 0xA11CE; + const FIRST_USER: u32 = TEST_USER_ID; + const SECOND_USER: u32 = TEST_USER_ID + 1; + + let mut table = ClientTable::new(2); + for (user, watermark) in [(FIRST_USER, 1u64), (SECOND_USER, 4u64)] { + table.evicted_fences.push_back(EvictedFence { + client_id: CLIENT_A, + epoch: watermark, + user_id: user, + watermark, + watermark_checksum: 0, + latest: Some(CachedReply::from_message(make_reply_for( + CLIENT_A, watermark, watermark, + ))), + }); + } + + let fence = table + .take_fence(CLIENT_A, SECOND_USER) + .expect("the second user's own fence must be reachable behind the first user's"); + assert_eq!(fence.watermark, 4); + assert!( + table + .evicted_fences + .iter() + .any(|fence| fence.user_id == FIRST_USER), + "and taking it must leave the other user's fence in place" + ); + assert!( + table.take_fence(CLIENT_A, SECOND_USER).is_none(), + "a fence is consumed on the hit, so a re-minted key cannot revive a \ + stale watermark and swallow a fresh session's requests" + ); + } + + /// A committed `Logout` ends the session explicitly, so it must not leave a + /// fence behind for a later register to revive: the client asked to be + /// forgotten, and a Logout committing after its entry was evicted finds + /// nothing to drop. + #[test] + fn logout_forgets_an_evicted_fence() { + const CLIENT_A: u128 = 0xA11CE; + + let mut table = ClientTable::new(2); + table.commit_register(CLIENT_A, TEST_USER_ID, make_register_reply(CLIENT_A, 1)); + table.commit_reply(CLIENT_A, make_reply_for(CLIENT_A, 1, 2)); + for (offset, churn) in [0xB0B1u128, 0xB0B2].iter().enumerate() { + let commit = 3 + offset as u64; + table.commit_register(*churn, TEST_USER_ID, make_register_reply(*churn, commit)); + } + assert!(table.get_epoch(CLIENT_A).is_none(), "A must be evicted"); + + table.remove_client(CLIENT_A); + table.commit_register(CLIENT_A, TEST_USER_ID, make_register_reply(CLIENT_A, 10)); + let epoch = table.get_epoch(CLIENT_A).expect("registered again"); + assert!( + matches!( + table.check_request(CLIENT_A, epoch, 1, 0), + RequestStatus::New + ), + "a register after Logout must start fresh, not revive the ended session's watermark" + ); + } + + /// The fence store is bounded by the slot count, so a churn far longer than + /// the table cannot grow it without limit. + #[test] + fn evicted_fences_stay_bounded_by_the_slot_count() { + let mut table = ClientTable::new(2); + // Each client commits an app request, so eviction has a real fence to + // keep -- a register-only session is skipped on purpose. + for client in 1..=20u128 { + let commit = (client * 2) as u64; + table.commit_register(client, TEST_USER_ID, make_register_reply(client, commit)); + table.commit_reply(client, make_reply_for(client, 1, commit + 1)); + } + assert_eq!( + table.evicted_fences.len(), + table.slots.len(), + "fences must be trimmed to the slot count" + ); + } + #[allow(clippy::cast_possible_truncation)] fn make_register_reply(client: u128, commit: u64) -> Message { let header_size = std::mem::size_of::(); diff --git a/core/integration/tests/cluster/client_table_adversarial.rs b/core/integration/tests/cluster/client_table_adversarial.rs index 10af0e749b..90c1ae245c 100644 --- a/core/integration/tests/cluster/client_table_adversarial.rs +++ b/core/integration/tests/cluster/client_table_adversarial.rs @@ -17,14 +17,14 @@ //! Adversarial specs against the VSR client table's at-most-once guarantees. //! -//! Both tests are RED SPECS, expected to FAIL: they assert the dedup contract -//! a retrying client needs, and the current table cannot honour it at its two -//! resource edges. +//! Both assert the dedup contract a retrying client needs at the table's two +//! resource edges. The capacity one now passes; the reply-ring one is still a +//! RED SPEC, expected to FAIL. //! -//! 1. Capacity: a full table evicts the entry with the oldest commit, and the -//! eviction erases that client's request watermark. A client that was -//! merely quiet (not gone) re-registers and its retry of an -//! already-committed request id re-executes. +//! 1. Capacity: a full table evicts the entry with the oldest commit. Eviction +//! keeps that client's request watermark (and the watermark's reply when the +//! ring still held it), so a client that was merely quiet re-registers and +//! its retry of an already-committed request id is answered, not re-executed. //! 2. Reply ring: each entry retains only its `REPLY_RING_CAPACITY` most //! recent committed replies. A retry of a request whose reply aged out is //! refused with the terminal `RequestAlreadyApplied` and no result payload, @@ -77,25 +77,21 @@ const REPLY_WAIT: Duration = Duration::from_secs(5); const RETRY_PAUSE: Duration = Duration::from_millis(100); -/// RED SPEC, expected to FAIL: capacity eviction must not erase a live -/// client's dedup watermark. +/// Capacity eviction must not erase a live client's dedup watermark. /// /// With the table floored at two slots, three fresh registrations evict /// `CLIENT_A` (its commit is the oldest) while its connection is still open /// and its request 1 is committed. The client then does exactly what the /// resume contract tells a disconnected client to do: reconnect, /// re-authenticate under its own identity, and retry the request it never saw -/// answered. The register finds no entry to rebind, mints a fresh one at -/// watermark zero, and the retry of the committed request re-executes. +/// answered. The register finds no entry to rebind, so it restores the fence +/// eviction left and the retry is answered from it. /// -/// The proof of re-execution is the committed duplicate-name rejection: a -/// dedup hit replays the cached success bytes, so any committed rejection -/// means the state machine ran the operation a second time. At-most-once -/// holds only for clients the table happened not to evict. -// TODO(hubcio): fix this test -#[ignore = "capacity eviction erases a live client's dedup watermark; replay re-executes"] +/// A committed duplicate-name rejection is the proof of re-execution: a dedup +/// hit replays the cached success bytes, so any committed rejection means the +/// state machine ran the operation a second time. #[iggy_harness(cluster_nodes = 1, server(metadata.clients_table_max = "2"))] -async fn given_a_low_client_table_cap_when_connects_churn_should_erase_a_live_dedup_watermark( +async fn given_a_low_client_table_cap_when_connects_churn_should_keep_a_live_dedup_watermark( harness: &mut TestHarness, ) { let addr = tcp_addr(harness); @@ -141,10 +137,10 @@ async fn given_a_low_client_table_cap_when_connects_churn_should_erase_a_live_de other => panic!( "capacity eviction erased a live client's dedup watermark: request 1 was \ committed and its reply delivered, but after the table (capacity 2) evicted \ - the entry to admit churn registrations, the resume re-registered at watermark \ - zero and the retry of request 1 was re-executed by the state machine instead \ - of being answered from the dedup cache (at-most-once broken for any client \ - the table evicts while it is merely quiet); got {other:?}" + the entry to admit churn registrations, the resume did not restore the fence, \ + so the retry of request 1 was re-executed by the state machine instead of \ + being answered from the dedup cache (at-most-once broken for any client the \ + table evicts while it is merely quiet); got {other:?}" ), } } diff --git a/core/integration/tests/cluster/failover_client_continuity.rs b/core/integration/tests/cluster/failover_client_continuity.rs index ab4c93f82f..9eeed6c0dd 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 423b7048ea..7fbe919dd3 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 35ffce9fca..bd99d2e32a 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, + /// 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, +} + +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( client: &C, current_address: &str, transport: TransportProtocol, -) -> Result, IggyError> { +) -> Result { 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( 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( 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 { + 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 921ef5d3dc..255e473e47 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 { 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 2c09e3f2ef..0b8e42b48d 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>, leader_redirection_state: Mutex, pub(crate) current_server_address: Mutex, + /// 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>, + /// 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>, // `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 { 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 { + 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 { + 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 { + 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 { + 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) -> 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 c439a38f19..362a2aa953 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 { 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/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 4df3e54020..0ab7d1be87 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -2350,6 +2350,65 @@ mod tests { ); } + /// The mixed-version upgrade hole from IGGY-250, at the router seam this + /// time: a wire-valid consensus frame carrying an operation only a newer + /// release defines must leave an accounted, operator-visible trace instead + /// of a bare warn log, because the frame's group stops making progress + /// until this node is upgraded. + #[compio::test] + async fn given_an_unknown_operation_when_dispatched_should_account_an_upgrade_fence_drop() { + // Far past every discriminant this build defines. + const OPERATION_FROM_A_NEWER_RELEASE: u8 = 0xEE; + + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let shard = build_test_shard(0, &config, TestMux::default()); + + let mut owned = server_common::iobuf::Owned::<{ server_common::MESSAGE_ALIGN }>::zeroed( + iggy_binary_protocol::HEADER_SIZE, + ); + { + let frame = owned.as_mut_slice(); + let size_offset = std::mem::offset_of!(PrepareHeader, size); + let frame_size = + u32::try_from(iggy_binary_protocol::HEADER_SIZE).expect("header size fits in u32"); + frame[size_offset..size_offset + 4].copy_from_slice(&frame_size.to_le_bytes()); + frame[std::mem::offset_of!(PrepareHeader, command)] = Command::Prepare as u8; + frame[std::mem::offset_of!(PrepareHeader, operation)] = OPERATION_FROM_A_NEWER_RELEASE; + let header: &[u8; iggy_binary_protocol::HEADER_SIZE] = frame + [..iggy_binary_protocol::HEADER_SIZE] + .try_into() + .expect("frame spans a full header"); + let checksum = iggy_binary_protocol::frame_checksum_bytes(header); + frame[..size_of::()].copy_from_slice(&checksum.to_le_bytes()); + } + let message = Message::::try_from(owned) + .expect("a sealed Prepare frame is wire-valid in the generic view"); + + let before = shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::CONSENSUS, + shard::metrics::frame_drop_reason::UNSUPPORTED_OPERATION, + ); + shard.dispatch(message); + assert_eq!( + shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::CONSENSUS, + shard::metrics::frame_drop_reason::UNSUPPORTED_OPERATION, + ), + before + 1, + "an operation from a newer release must be accounted under its own reason, not \ + folded into the generic unparsable drop" + ); + assert_eq!( + shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::CONSENSUS, + shard::metrics::frame_drop_reason::UNPARSABLE, + ), + 0, + "version skew must not read as header corruption" + ); + } + /// Receive half of the purge gate in `on_repair_range_reply`: while a /// committed purge has not applied locally, a repair verdict must be /// deferred wholesale -- installing the peer's floor against pre-purge diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs index cd4cd91e19..391790d0ac 100644 --- a/core/server_common/src/consensus_message.rs +++ b/core/server_common/src/consensus_message.rs @@ -242,7 +242,7 @@ where let bytes = >::header_storage(&self.backing); let typed = bytemuck::checked::try_from_bytes::(&bytes[..size_of::()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::(bytes))?; // Before `validate`: a header that did not survive the link intact cannot // have any of its fields believed, and `validate` reads them. typed.verify_frame()?; @@ -281,7 +281,7 @@ where let bytes = >::header_storage(&self.backing); let typed = bytemuck::checked::try_from_bytes::(&bytes[..size_of::()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::(bytes))?; // Before `validate`: a header that did not survive the link intact cannot // have any of its fields believed, and `validate` reads them. typed.verify_frame()?; @@ -477,7 +477,7 @@ where } let header = bytemuck::checked::try_from_bytes::(&bytes[..size_of::()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::(bytes))?; header.validate()?; // `size` is the whole-frame length and must at least span the header, or @@ -524,7 +524,7 @@ where } let header = bytemuck::checked::try_from_bytes::(&first.as_slice()[..size_of::()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::(first.as_slice()))?; header.validate()?; // See `TryFrom`: `size` must at least span the header so a @@ -774,6 +774,25 @@ impl MessageBag { } } +/// Why `H`'s header bytes failed bytemuck's checked cast. +/// +/// An operation discriminant this build does not define means the sender runs a +/// newer release; the frame is wire-valid and the node needs upgrading, which is +/// a different operator action from the corrupted-header case. bytemuck reports +/// both as one error, so the operation byte is probed here to separate them. +fn classify_failed_cast(bytes: &[u8]) -> ConsensusError +where + H: ConsensusHeader, +{ + if let Some(offset) = H::OPERATION_OFFSET + && let Some(&code) = bytes.get(offset) + && !Operation::is_known_code(code) + { + return ConsensusError::UnsupportedOperation { operation: code }; + } + ConsensusError::InvalidBitPattern +} + impl TryFrom> for MessageBag where T: ConsensusHeader, diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 99f70a2240..6d2e725e6a 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -120,6 +120,14 @@ pub mod frame_drop_variant { /// replicated traffic has nobody to answer, so this is the only record the op /// was destroyed. pub mod frame_drop_reason { + /// Operation discriminant unknown to this build: the sender is newer. + /// + /// Distinct from `UNPARSABLE` because upgrading this node is the fix, and + /// until it is, the frame's consensus group gap-stops here. + pub const UNSUPPORTED_OPERATION: &str = "unsupported_operation"; + /// A consensus frame failed typed decode for any other reason (corrupt + /// header, bad size, client-bound command on the inbound path). + pub const UNPARSABLE: &str = "unparsable"; pub const FULL: &str = "full"; pub const DISCONNECTED: &str = "disconnected"; pub const UNROUTABLE: &str = "unroutable"; @@ -134,7 +142,7 @@ pub mod frame_drop_reason { // site actually produces it, so the unreachable corners of the 7 x 7 cross // product never appear as permanent zero-valued series. const VARIANT_COUNT: usize = 7; -const REASON_COUNT: usize = 7; +const REASON_COUNT: usize = 9; const VARIANTS: [&str; VARIANT_COUNT] = [ frame_drop_variant::CONSENSUS, @@ -147,6 +155,8 @@ const VARIANTS: [&str; VARIANT_COUNT] = [ ]; const REASONS: [&str; REASON_COUNT] = [ + frame_drop_reason::UNSUPPORTED_OPERATION, + frame_drop_reason::UNPARSABLE, frame_drop_reason::FULL, frame_drop_reason::DISCONNECTED, frame_drop_reason::UNROUTABLE, diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 4340761a00..d290c861ac 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -23,7 +23,7 @@ use crate::{IggyShard, LifecycleFrame, Receiver, RestorableMetadataStm, ShardFra use consensus::{MetadataHandle, PartitionsHandle}; use crossfire::TrySendError; use futures::FutureExt; -use iggy_binary_protocol::{GenericHeader, Operation, PrepareHeader}; +use iggy_binary_protocol::{ConsensusError, GenericHeader, Operation, PrepareHeader}; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; use message_bus::{ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn}; @@ -62,17 +62,33 @@ where pub fn dispatch(&self, message: Message) { let bag = match MessageBag::try_from(message) { Ok(bag) => bag, + Err(ConsensusError::UnsupportedOperation { operation }) => { + // Terminal for this consensus group, not a per-frame hiccup: + // the op is never journaled, never acked, and every later + // prepare dies on the resulting gap while quorum hides the + // outage. Repair wraps the same typed header, so it cannot + // rescue this node either -- only upgrading it can. Nothing + // fences the sending peer, so the log and the counter are the + // whole signal an operator gets. + self.metrics.record_frame_drop( + frame_drop_variant::CONSENSUS, + frame_drop_reason::UNSUPPORTED_OPERATION, + ); + tracing::error!( + shard = self.id, + operation = format_args!("{operation:#04x}"), + build_release = iggy_binary_protocol::IGGY_PROTOCOL_VERSION, + "consensus frame carries an operation this build does not know; the sender \ + runs a newer release. This node cannot journal or ack it, so its consensus \ + group stops making progress until this node is upgraded" + ); + return; + } Err(e) => { - // TODO(hubcio): this drop is the whole story for a consensus - // frame carrying an Operation this build does not know: no - // metric, no peer error, no eviction. An old node in a mixed - // cluster silently gap-stops the group here (never journals - // the op, never PrepareOks, every later prepare dies on the - // gap check) while quorum hides it, and repair wraps the same - // typed header so it cannot rescue. Rolling upgrades across - // consensus-op additions need a version fence (release_min / - // release_max bounds on the replica plane) before this arm is - // safe to hit. + self.metrics.record_frame_drop( + frame_drop_variant::CONSENSUS, + frame_drop_reason::UNPARSABLE, + ); tracing::warn!(shard = self.id, error = %e, "dropping unparsable consensus frame"); return; } @@ -804,25 +820,19 @@ mod tests { use server_common::{MESSAGE_ALIGN, Message, MessageBag}; use std::mem::offset_of; - /// RED SPEC, expected to FAIL: pins the mixed-cluster upgrade hole in - /// `dispatch`'s decode seam. - /// /// An `Operation` discriminant this build does not know, arriving on an /// otherwise wire-valid consensus frame (correct command, size, checksum: - /// exactly what a newer release sends after an op addition), decodes to - /// the same undifferentiated `ConsensusError::InvalidBitPattern` as random - /// memory corruption. `dispatch` answers both identically: a warn log and - /// a dropped frame. No metric, no peer error, no eviction, no version - /// fence. An old node in a mixed cluster therefore gap-stops its consensus - /// group silently (never journals the op, never sends a `PrepareOk`, every - /// later prepare dies on the gap check) while quorum hides the outage. + /// exactly what a newer release sends after an op addition), must decode to + /// its own error rather than the `InvalidBitPattern` that random memory + /// corruption produces. /// - /// Passes once the decode surfaces a dedicated unsupported-operation - /// signal the router can fence and account, instead of collapsing it into - /// the corruption error. + /// The two need different operator actions: version skew is fixed by + /// upgrading this node, and until it is, the frame's consensus group makes + /// no progress (the op is never journaled, never acked, and every later + /// prepare dies on the gap). `dispatch` splits its drop arms on this + /// distinction; the accounting half is pinned in the server crate by + /// `given_an_unknown_operation_when_dispatched_should_account_an_upgrade_fence_drop`. #[test] - // TODO(hubcio): fix this test - #[ignore = "unknown operation collapses into InvalidBitPattern; no upgrade fence"] fn given_an_unknown_operation_when_a_consensus_frame_decodes_should_surface_an_upgrade_fence_signal() { // Far past every defined Operation discriminant (the highest is 165). @@ -858,7 +868,12 @@ mod tests { }; assert!( - !matches!(error, ConsensusError::InvalidBitPattern), + matches!( + error, + ConsensusError::UnsupportedOperation { + operation: OPERATION_FROM_A_NEWER_RELEASE + } + ), "unknown operation {OPERATION_FROM_A_NEWER_RELEASE:#x} is silently dropped: the \ typed decode collapses a wire-valid frame from a newer release into the same \ InvalidBitPattern as corruption, and dispatch drops both with only a warn log, \ diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs index 1885c69faf..17d5243659 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 @@ private async Task RedirectAsync(CancellationToken token) return true; } + /// + /// 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. + /// + 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 GetCurrentLeaderNodeAsync(CancellationToken token) { var leaderlessDeadline = Environment.TickCount64 + VsrLeaderlessWaitMs; @@ -438,6 +457,8 @@ private async Task RedirectAsync(CancellationToken token) 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 32ea6e03ec..043a508092 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 Task ConnectAsync(CancellationToken token = default) 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 async Task ChangePasswordAsync(Identifier userId, string currentPassword, 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; } /// @@ -833,6 +850,9 @@ public async Task LogoutUserAsync(CancellationToken token = default) { 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 async Task DeletePersonalAccessTokenAsync(string name, CancellationToken /// public async Task 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; } /// @@ -921,7 +944,10 @@ or ConnectionState.Authenticating 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 @@ private async Task RunHeartbeatAsync(TimeSpan interval, CancellationToken token) // 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 @@ private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken 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 @@ private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken // 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 @@ private async Task TryEstablishConnectionAsync(bool autoLogin, CancellationToken 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 @@ async Task BackoffOrThrowAsync() _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 @@ private async Task DropStreamAsync() } } - private async Task AutoLoginAsync(CancellationToken token) + private string[] DialCandidates() + { + return DialCandidates(_currentAddress, _configuration.BaseAddress, _rosterAddresses); + } + + /// + /// 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. + /// + internal static string[] DialCandidates(string currentAddress, string baseAddress, string[] rosterAddresses) + { + var candidates = new List(); + 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); } + /// + /// 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. + /// + private AutoLoginSettings? SignInSettings() + { + if (_configuration.AutoLoginSettings.Enabled) + { + return _configuration.AutoLoginSettings; + } + + return _rememberedLogin; + } + /// /// 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 @@ private async Task> SendWithResponseAsync(ReadOnlyMemory +/// Mirrors the Rust SDK's dial_candidates: 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. +/// +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 0000000000..3bc704e06e --- /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; + +/// +/// 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 +/// core/integration/tests/cluster/failover_client_continuity.rs. +/// +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); + } + + /// A reply for anything the roster read does not claim: a register, or an empty read. + 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; + } + + /// + /// 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. + /// + 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(); + 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 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 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); + + /// + /// 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. + /// + private sealed class MockNode : IDisposable + { + private readonly TcpListener _listener; + private readonly List _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 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 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 fe662af53c..5b9a46527c 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 0000000000..c5735517cc --- /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 0000000000..c219d4f6a3 --- /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 d409f7d9af..b6a91e30ea 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 e2673d965a..93f463e723 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 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.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 tlsCertificate; private final TcpConnectionPoolConfig poolConfig; private final ClientRoutingState routingState = new ClientRoutingState(); + private final LoginRoutingHook loginRoutingHook = new LoginRoutingHook() { + + @Override + public CompletableFuture loginOnLeader(Supplier> loginAttempt) { + return AsyncIggyTcpClient.this.loginOnLeader(loginAttempt); + } + + @Override + public void forgetLogin() { + rememberedLogin = null; + } + }; private final AtomicReference connection = new AtomicReference<>(); private final AtomicReference> 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 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> rememberedLogin; + private volatile boolean closed; private MessagesClient messagesClient; private ConsumerGroupsClient consumerGroupsClient; @@ -235,9 +266,9 @@ public CompletableFuture connect() { 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 @@ private CompletableFuture redialAttempt(int attempt, RetryPolicy policy) { 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 @@ private CompletableFuture redialAttempt(int attempt, RetryPolicy policy) { * again before Register when the redialed node is not the leader. */ private CompletableFuture replayLogin() { + Supplier> 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 @@ CompletableFuture loginOnLeader(Supplier 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 @@ CompletableFuture> findLeaderElsewhere(ConnectionInfo c 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 redialCandidates() { + List 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 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 f541e9de7c..3b258b7ef4 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 @@ private LeaderAwareness() {} * exceptionally, so the redirection path cannot fail the login that * triggered it. */ - static CompletableFuture> findLeaderElsewhere( + static CompletableFuture findLeaderElsewhere( Supplier> fetchMetadata, ConnectionInfo currentTarget) { return findLeaderElsewhere(fetchMetadata, currentTarget, LEADERLESS_WAIT_BUDGET, LEADERLESS_POLL_INTERVAL); } - static CompletableFuture> findLeaderElsewhere( + static CompletableFuture findLeaderElsewhere( Supplier> fetchMetadata, ConnectionInfo currentTarget, Duration leaderlessWaitBudget, @@ -87,7 +87,7 @@ static CompletableFuture> findLeaderElsewhere( fetchMetadata, currentTarget, leaderlessWaitBudget, leaderlessPollInterval, electionDeadlineNanos); } - private static CompletableFuture> pollForLeader( + private static CompletableFuture pollForLeader( Supplier> fetchMetadata, ConnectionInfo currentTarget, Duration leaderlessWaitBudget, @@ -99,16 +99,18 @@ private static CompletableFuture> pollForLeader( } catch (RuntimeException fetchError) { fetched = CompletableFuture.failedFuture(fetchError); } - return fetched.>>handleAsync((metadata, error) -> { + return fetched.>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 endpoints; try { + endpoints = nodeTargets(metadata); check = checkLeader(metadata, currentTarget); } catch (RuntimeException selectionError) { log.warn( @@ -116,10 +118,11 @@ private static CompletableFuture> pollForLeader( + " 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 @@ private static CompletableFuture> pollForLeader( + " 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 @@ private static CompletableFuture> pollForLeader( 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 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 @@ private static boolean reachesOnlyLocalMachine(InetAddress[] addresses) { /** * 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 redirect, List 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 a6acb7dc09..904b492e2a 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 loginOnLeader(Supplier> 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 ae596731f4..3e9f45bf8b 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.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 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 b6e7ab1940..1bf1e7ceda 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 CompletableFuture logout() { 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 0000000000..f541701b9a --- /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 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 00540cd23c..b5b73cfb78 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 FindLeaderElsewhere { private final ConnectionInfo currentTarget = new ConnectionInfo("iggy-follower", 8092); private Optional findLeader(Supplier> fetch) { + return lookUpLeader(fetch).redirect(); + } + + private LeaderAwareness.LeaderLookup lookUpLeader(Supplier> fetch) { return LeaderAwareness.findLeaderElsewhere(fetch, currentTarget, BUDGET, INTERVAL) .orTimeout(30, TimeUnit.SECONDS) .join(); @@ -248,7 +252,8 @@ void shouldGiveUpOnLeaderlessClusterAfterBudget() { Duration.ofMillis(100), INTERVAL) .orTimeout(30, TimeUnit.SECONDS) - .join(); + .join() + .redirect(); assertThat(leader).isEmpty(); assertThat(fetchCount.get()).isGreaterThan(1); @@ -276,6 +281,22 @@ void shouldGiveUpWhenMetadataFetchThrowsSynchronously() { 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 @@ void shouldStayWithoutPollingWhenAlreadyOnLeader() { 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 ce833a58cb..fff6114d30 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.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 2564cfacfd..afb034d09b 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((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 10b68639d0..0ef16326f3 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; /** 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 { 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 0697c5f3d4..07cbcecd46 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(); + 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 d1e5e4fe2a..4d57387e79 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(