This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/durable-protocol-and-sessions in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 7ae28b83864f01cdb7108e592f5eb81fc817bc9e Author: Hubert Gruszecki <[email protected]> AuthorDate: Sun Oct 4 11:36:47 2026 +0200 fix(sdk): resume retained sessions and preserve uncertain outcomes Transport loss reset established sessions in Go, Java and C#, leaving old protection slots reserved and obscuring retry identity. Java, Node and C# also let later refusals overwrite an unknown mutation outcome. Resume established sessions with their original proof and permit fresh registration only after a definitive bind refusal. Preserve uncertainty across later refusals, fence retained Rust frames after transport locking, and restore retries for proven unsent attempts. Document session and API contracts and synchronize error catalogs. --- core/common/src/traits/binary_transport.rs | 24 ++- core/metadata/src/impls/metadata.rs | 219 +++++++++++++++++++-- core/sdk/README.md | 13 +- core/sdk/src/clients/client.rs | 7 +- core/sdk/src/clients/producer.rs | 14 +- core/sdk/src/poll_routing.rs | 10 +- core/sdk/src/quic/quic_client.rs | 65 +++++- core/sdk/src/session.rs | 21 +- core/sdk/src/tcp/tcp_client.rs | 7 +- core/sdk/src/vsr.rs | 20 +- core/sdk/src/websocket/websocket_client.rs | 53 +++++ .../Implementations/TcpMessageStream.Vsr.cs | 62 +++++- .../IggyClient/Implementations/TcpMessageStream.cs | 10 +- foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs | 2 +- foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs | 9 +- foreign/csharp/Iggy_SDK/Vsr/VsrError.cs | 2 + .../VsrTests/EndpointFailoverTests.cs | 140 +++++++++++-- foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs | 5 + .../Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs | 5 +- foreign/csharp/README.md | 12 +- foreign/go/README.md | 11 +- foreign/go/client/tcp/tcp_connect_test.go | 47 +++-- foreign/go/client/tcp/tcp_core.go | 15 +- foreign/go/client/tcp/tcp_core_test.go | 6 +- foreign/go/client/tcp/tcp_failover_test.go | 99 +++++++++- foreign/go/client/tcp/tcp_poll_routing_test.go | 4 +- foreign/go/client/tcp/tcp_session_management.go | 59 +++++- foreign/go/errors/errors.yaml | 3 + foreign/go/errors/errors_gen.go | 15 ++ foreign/go/internal/vsr/session.go | 5 +- foreign/java/README.md | 6 + .../iggy/client/async/tcp/AsyncTcpConnection.java | 78 +++++++- .../client/async/tcp/vsr/ConsensusSession.java | 36 ++-- .../client/async/tcp/vsr/VsrRequestEncoder.java | 12 ++ .../client/async/tcp/vsr/VsrResponseHandler.java | 1 - .../org/apache/iggy/exception/IggyErrorCode.java | 1 + .../AsyncIggyTcpClientTransientFailoverTest.java | 129 ++++++++++++ .../async/tcp/vsr/VsrRequestEncoderTest.java | 15 +- .../async/tcp/vsr/VsrResponseHandlerTest.java | 7 +- foreign/node/README.md | 4 + foreign/node/src/client/client.socket.test.ts | 21 +- foreign/node/src/client/client.socket.ts | 8 + foreign/node/src/wire/error.code.ts | 1 + foreign/python/README.md | 14 +- 44 files changed, 1117 insertions(+), 180 deletions(-) diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs index b4cd8c42d..cfbf3cfee 100644 --- a/core/common/src/traits/binary_transport.rs +++ b/core/common/src/traits/binary_transport.rs @@ -79,8 +79,26 @@ mod vsr_session_sealed { /// mutation corrupts the dedup counter or silently breaks at-most-once. #[async_trait] pub trait VsrSessionControl: vsr_session_sealed::Sealed + BinaryTransport { + /// Internal bearer proof. Never log or expose this value to application users. + #[doc(hidden)] async fn session_bind_secret(&self) -> Result<BindSecret, IggyError>; async fn session_identity(&self) -> Result<SessionIdentity, IggyError>; + #[doc(hidden)] + fn decode_session_binding( + &self, + identity: SessionIdentity, + response: &[u8], + ) -> Result<LoginRegisterResponse, IggyError> { + let bound = + LoginRegisterResponse::decode_from(response).map_err(|_| IggyError::InvalidFormat)?; + if bound.session == 0 { + return Err(IggyError::InvalidSession(bound.session)); + } + if identity.session != 0 && bound.session != identity.session { + return Err(IggyError::SessionMismatch(identity.session, bound.session)); + } + Ok(bound) + } async fn resume_vsr_session(&self) -> Result<(), IggyError> where Self: Sync, @@ -97,11 +115,7 @@ pub trait VsrSessionControl: vsr_session_sealed::Sealed + BinaryTransport { request.to_bytes(), ) .await?; - let bound = - LoginRegisterResponse::decode_from(&response).map_err(|_| IggyError::InvalidFormat)?; - if bound.session != identity.session { - return Err(IggyError::SessionMismatch(identity.session, bound.session)); - } + let bound = self.decode_session_binding(identity, &response)?; self.bind_vsr_session(bound.session).await?; self.set_state(ClientState::Authenticated).await; self.publish_event(DiagnosticEvent::SignedIn).await; diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 82ba93859..b559bfe78 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -1796,9 +1796,9 @@ where Ok(offer) } - /// Install a fetched state transfer: persist + restore the snapshot, - /// replace the client table, and jump the commit state to the snapshot - /// floor so the tail repair takes over from there. + /// Install a fetched state transfer without rewinding either the state + /// machine or client table, then let tail repair continue above the local + /// applied frontier. /// /// Ordering: persist FIRST (a crash mid-install must reboot from the /// transferred state, not the pre-transfer one), then the in-place STM @@ -1884,9 +1884,8 @@ where // commit walk never revisits ops it already counted as applied, so // the rewound-over effects would be lost until the next transfer. // Keep the local STM (it is a superset) and let tail repair cover - // `(commit_min, commit_op]`. The client table still installs below: - // it comes from the serving primary's LIVE state at `table_frontier - // == commit_op`, which is never behind this replica. + // `(commit_min, commit_op]`. The table is checked separately below: + // a cached offer can also lag the local table's committed protection. // Preliminary read, only to decide whether the gates are needed; the // binding decision is re-derived under them below. let snapshot_ahead = snapshot_seq > consensus.commit_min(); @@ -2013,20 +2012,31 @@ where snapshot_seq, local_applied, "transferred snapshot at or below the local applied frontier; \ - keeping the local state machine and installing the table only" + keeping the local state machine" ); } - let configured_capacity = self.client_table.borrow().capacity(); - if client_table.capacity_committed() && configured_capacity != client_table.capacity() { - warn!( - configured_capacity, - committed_capacity = client_table.capacity(), - "metadata retry capacity is fixed by committed state; ignoring local configuration" + // Commits can advance while the install awaits. A prior transfer may + // also have installed a table ahead of the local state machine. + let local_table_frontier = consensus.commit_min().max(self.client_table_frontier.get()); + if table_frontier >= local_table_frontier { + let configured_capacity = self.client_table.borrow().capacity(); + if client_table.capacity_committed() && configured_capacity != client_table.capacity() { + warn!( + configured_capacity, + committed_capacity = client_table.capacity(), + "metadata retry capacity is fixed by committed state; ignoring local configuration" + ); + } + *self.client_table.borrow_mut() = client_table; + self.client_table_frontier.set(table_frontier); + } else { + tracing::info!( + table_frontier, + local_table_frontier, + "transferred client table below the local table frontier; keeping the local client table" ); } - *self.client_table.borrow_mut() = client_table; - self.client_table_frontier.set(table_frontier); if snapshot_ahead { self.applied_frontier .record_recovery_revision(self.mux_stm.streams().read(|inner| inner.revision)); @@ -6900,6 +6910,185 @@ mod tests { ); } + #[compio::test] + async fn given_newer_local_state_when_installing_stale_transfer_should_preserve_sessions() { + const CLIENT: u128 = 9; + const NEW_CLIENT: u128 = 10; + const USER: u32 = 0; + const BIND_VERIFIER: [u8; 32] = [0x5a; 32]; + let directory = tempfile::tempdir().unwrap(); + let mut metadata = metadata_plane(); + metadata.journal = Some( + PrepareJournal::open(&directory.path().join("journal.wal"), 0) + .await + .unwrap(), + ); + metadata.mux_stm.users().ensure_root_user("root", "hash"); + let original_session = metadata + .submit_register_in_process(CLIENT, USER, BIND_VERIFIER) + .await + .unwrap(); + let consensus = metadata.consensus.as_ref().unwrap(); + let transfer_op = consensus.commit_min(); + let snapshot = <IggySnapshot as Snapshot>::create(&*metadata.mux_stm, transfer_op, 1) + .unwrap() + .encode() + .unwrap(); + let transferred_table = + ClientTable::decode(&metadata.client_table.borrow().encode()).unwrap(); + + let original_reply = metadata + .submit_request_in_process(create_stream_request(CLIENT, 1, "original")) + .await + .unwrap(); + assert_eq!( + iggy_binary_protocol::result_code(original_reply.body()), + Some(0), + ); + let new_session = metadata + .submit_register_in_process(NEW_CLIENT, USER, BIND_VERIFIER) + .await + .unwrap(); + let local_applied = consensus.commit_min(); + let local_table = metadata.client_table.borrow().encode(); + let local_snapshot = + <IggySnapshot as Snapshot>::create(&*metadata.mux_stm, local_applied, 1) + .unwrap() + .encode() + .unwrap(); + + let outcome = metadata + .install_state_transfer(&snapshot, transferred_table, transfer_op, transfer_op) + .await + .unwrap(); + + assert_eq!(outcome.installed_frontier, local_applied); + assert_eq!(consensus.commit_min(), local_applied); + assert_eq!(consensus.commit_max(), local_applied); + assert_eq!( + metadata.client_table.borrow().get_epoch(NEW_CLIENT), + Some(new_session.epoch), + "a stale transfer must preserve registrations above its frontier", + ); + assert_eq!( + metadata.client_table.borrow().encode(), + local_table, + "a stale transfer must preserve committed registrations and receipts", + ); + assert_eq!( + <IggySnapshot as Snapshot>::create(&*metadata.mux_stm, local_applied, 1) + .unwrap() + .encode() + .unwrap(), + local_snapshot, + ); + assert_eq!( + metadata + .bound_session(CLIENT, USER, BIND_VERIFIER) + .unwrap() + .epoch, + original_session.epoch, + ); + assert_eq!( + metadata + .bound_session(NEW_CLIENT, USER, BIND_VERIFIER) + .unwrap() + .epoch, + new_session.epoch, + ); + let replay = metadata + .submit_request_in_process(create_stream_request(CLIENT, 1, "original")) + .await + .unwrap(); + assert_eq!(replay.as_slice(), original_reply.as_slice()); + assert_eq!(consensus.commit_min(), local_applied); + + let mut request = create_stream_request(NEW_CLIENT, 1, "after-transfer"); + bytemuck::checked::from_bytes_mut::<RoutedRequestHeader>( + &mut request.as_mut_slice()[..size_of::<RoutedRequestHeader>()], + ) + .session = new_session.epoch; + let prepare = metadata.prepare_request(request).unwrap(); + let op = prepare.header().op; + metadata + .journal + .as_ref() + .unwrap() + .handle() + .append(prepare) + .await + .unwrap(); + consensus.advance_commit_max(op); + metadata.commit_journal().await; + assert_eq!(consensus.commit_min(), op); + assert_eq!( + metadata.client_table.borrow().get_watermark(NEW_CLIENT), + Some(1) + ); + } + + #[compio::test] + async fn given_ahead_client_table_when_installing_older_transfer_should_preserve_its_frontier() + { + const CLIENT: u128 = 9; + const OTHER_CLIENT: u128 = 10; + const NEW_CLIENT: u128 = 11; + const USER: u32 = 7; + const SESSION: u64 = 1; + const TABLE_FRONTIER: u64 = 3; + const OLDER_FRONTIER: u64 = 2; + let metadata = metadata_plane(); + let mut table = ClientTable::new(CLIENTS_TABLE_MAX); + table.commit_capacity(CLIENTS_TABLE_MAX).unwrap(); + let snapshot = <IggySnapshot as Snapshot>::create(&TestMux::default(), SESSION, 1) + .unwrap() + .encode() + .unwrap(); + let mut older_table = None; + for (client, epoch) in [ + (CLIENT, SESSION), + (OTHER_CLIENT, OLDER_FRONTIER), + (NEW_CLIENT, TABLE_FRONTIER), + ] { + let reply = build_reply_message( + &PrepareHeader { + client, + user_id: USER, + op: epoch, + operation: Operation::Register, + ..Default::default() + }, + &bytes::Bytes::new(), + ); + table + .commit_register(client, USER, [0x5a; 32], reply) + .unwrap(); + if epoch == OLDER_FRONTIER { + older_table = Some(ClientTable::decode(&table.encode()).unwrap()); + } + } + let expected_table = table.encode(); + metadata + .install_state_transfer(&snapshot, table, TABLE_FRONTIER, TABLE_FRONTIER) + .await + .unwrap(); + metadata + .install_state_transfer( + &snapshot, + older_table.unwrap(), + OLDER_FRONTIER, + OLDER_FRONTIER, + ) + .await + .unwrap(); + + assert_eq!(metadata.consensus.as_ref().unwrap().commit_min(), SESSION); + assert_eq!(metadata.client_table.borrow().encode(), expected_table); + assert_eq!(metadata.client_table_frontier.get(), TABLE_FRONTIER); + assert!(!metadata.client_table_mutation_allowed(TABLE_FRONTIER)); + assert!(metadata.client_table_mutation_allowed(TABLE_FRONTIER + 1)); + } + #[compio::test] async fn transferred_client_table_waits_for_applied_frontier_before_checkpoint() { const CLIENT: u128 = 9; diff --git a/core/sdk/README.md b/core/sdk/README.md index 826d93b25..ef4dd1e02 100644 --- a/core/sdk/README.md +++ b/core/sdk/README.md @@ -46,7 +46,7 @@ Run from your application crate. Use a release compatible with your server; for unreleased changes, build the SDK and server from the same source checkout. Cluster auto-commit polling requires servers that support consumer session -attachment and primary poll routing (binary commands 14, 103 and 104). +attachment and primary poll routing (binary commands 15, 103 and 104). Rust manual and interval offset writes also require command 123. It discovers the primary while allowing final commits for partitions awaiting handoff; stores and deletes retain their existing wire formats and deduplication keys. Pause @@ -62,6 +62,17 @@ cargo add iggy All four transports are included; this crate declares no optional Cargo features. +The durable session API adds required `session_identity` and `session_bind_secret` +methods to `VsrSessionControl`. External transport implementations must provide +them, retain their logical identity and bind proof across transport reconnects, +and clear them on explicit logout. `VsrSessionSealed` is an implementation marker, +not a restriction on external implementations. The bind proof is a bearer +credential for internal session control and must never be logged or exposed. + +An established session has its own lifetime. Password or PAT changes and PAT +expiry do not end it. Logout, session lease expiry, or user deactivation end the +session and prevent its bind proof from restoring it. + ## Quick start Start a source server from the repository root in a separate terminal: diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs index c116a3e12..9b05f8d38 100644 --- a/core/sdk/src/clients/client.rs +++ b/core/sdk/src/clients/client.rs @@ -122,8 +122,9 @@ const SESSION_CONTROL_CODES: [u32; 5] = [ /// [`IggyProducer`] so that _background_ producers flush the latest state. Finally, /// call [`shutdown()`] on the [`IggyClient`] which closes the connection. /// Use [`disconnect()`] rather than [`shutdown()`] to close the connection but keep the client usable, as a -/// client that has been shut down cannot reconnect. Explicit disconnect clears sign-in; -/// call [`connect()`] and [`login_user()`] again before issuing requests. +/// client that has been shut down cannot reconnect. Explicit disconnect clears the remembered sign-in. +/// Configured auto-login still authenticates on the next connection; otherwise call +/// [`connect()`] and [`login_user()`] again before issuing requests. /// /// # Examples /// @@ -757,7 +758,7 @@ impl IggyClient { /// /// let producer = client /// .producer("stream_name", "topic_name")? - /// .topic_durability(iggy::prelude::Durability::Persisted) // returns IggyProducerBuilder from IggyClient + /// .topic_durability(iggy::prelude::Durability::Persisted) /// .partitioning(Partitioning::balanced()) /// .send_retries(Some(3), Some(NonZeroIggyDuration::ONE_SECOND)) /// .create_topic_if_not_exists( diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs index 01ef9a6a4..d19defcf2 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs @@ -61,10 +61,15 @@ pub(crate) fn no_confirmations() -> SendMessagesResponse { } } -/// A producer retry creates another request identity. Only an explicit -/// non-admission permits that; ambiguous retries belong to the transport. +/// A producer retry creates another request identity. Only a refusal or +/// failure before submission permits that; ambiguous retries belong to the transport. const fn stops_send_retries(error: &IggyError) -> bool { - !matches!(error, IggyError::TransientNotAccepted) + !matches!( + error, + IggyError::TransientNotAccepted + | IggyError::NotConnected + | IggyError::CannotEstablishConnection + ) } pub struct ProducerCore { @@ -683,6 +688,7 @@ unsafe impl Sync for IggyProducer {} /// | --- | --- | --- | /// | the client is not signed in, so nothing may be sent yet | yes | the send goes ahead as soon as the client is signed in, or fails with [`IggyError::CannotSendMessagesDueToClientDisconnection`] once the retry budget is spent | /// | the binary transport reports that the server refused admission ([`IggyError::TransientNotAccepted`]) | yes, as a new request | the confirmation of the admitted attempt, or the final refusal | +/// | the binary transport cannot connect before submission ([`IggyError::NotConnected`] or [`IggyError::CannotEstablishConnection`]) | yes, as a new request | the confirmation of the submitted attempt, or the final connection error | /// | an HTTP request fails, including a server admission refusal | no producer retry | the HTTP error; HTTP transport retries are configured separately below | /// | another request error or an ambiguous failure | no | the cause and unconfirmed batch; the caller decides whether to resend | /// | the batch committed, but its confirmation could not be read ([`IggyError::InvalidBytesResponse`] or [`IggyError::InvalidJsonResponse`], raised on the HTTP transport only) | no | the write did happen and retrying would duplicate it on purpose | @@ -1110,6 +1116,8 @@ mod tests { ); } assert!(!stops_send_retries(&IggyError::TransientNotAccepted)); + assert!(!stops_send_retries(&IggyError::NotConnected)); + assert!(!stops_send_retries(&IggyError::CannotEstablishConnection)); } #[tokio::test] diff --git a/core/sdk/src/poll_routing.rs b/core/sdk/src/poll_routing.rs index 749489ba6..6bad4aba5 100644 --- a/core/sdk/src/poll_routing.rs +++ b/core/sdk/src/poll_routing.rs @@ -33,7 +33,6 @@ use iggy_binary_protocol::requests::messages::PollMessagesRequest; use iggy_binary_protocol::requests::system::{BindSessionRequest, SessionIdentity}; use iggy_binary_protocol::responses::messages::PollRoutingResponse; use iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse; -use iggy_binary_protocol::responses::users::LoginRegisterResponse; use iggy_binary_protocol::{WireDecode, WireEncode}; use iggy_common::ClientState; use iggy_common::{ @@ -389,14 +388,7 @@ impl<T: PollTransport> PollRouter<T> { ) .await .map_err(unaccepted_data_error)?; - let bound = LoginRegisterResponse::decode_from(&response) - .map_err(|_| IggyError::InvalidFormat)?; - if bound.session != route.consumer_session.session { - return Err(IggyError::SessionMismatch( - route.consumer_session.session, - bound.session, - )); - } + coordinator.decode_session_binding(route.consumer_session, &response)?; connection .client .set_state(ClientState::Authenticated) diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index db9ffa8bd..5dda4d6fb 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -1087,7 +1087,13 @@ impl QuicClient { }; let request_header = match preencoded { - Some(header) => header, + Some(header) => { + let session = consensus_session + .lock() + .map_err(|_| IggyError::InvalidConfiguration)?; + crate::vsr::validate_retained_header(&header, &session)?; + header + } None => { let mut session = consensus_session .lock() @@ -1313,6 +1319,63 @@ mod tests { send.stopped().await.unwrap(); } + #[tokio::test] + async fn a_retained_write_waiting_for_the_connection_must_recheck_its_session() { + const TEST_BUDGET: std::time::Duration = std::time::Duration::from_secs(1); + let _ = rustls::crypto::ring::default_provider().install_default(); + let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap(); + let config = quinn::ServerConfig::with_single_cert( + vec![certified.cert.der().clone()], + rustls::pki_types::PrivatePkcs8KeyDer::from(certified.signing_key.serialize_der()) + .into(), + ) + .unwrap(); + let endpoint = quinn::Endpoint::server(config, "127.0.0.1:0".parse().unwrap()).unwrap(); + let client = QuicClient::create(Arc::new(QuicClientConfig { + server_address: endpoint.local_addr().unwrap().to_string(), + ..Default::default() + })) + .unwrap(); + let (peer, connected) = tokio::join!( + async { endpoint.accept().await.unwrap().await.unwrap() }, + Client::connect(&client) + ); + connected.unwrap(); + client.bind_vsr_session(1).await.unwrap(); + let payload = Bytes::from_static(b"uncertain-send"); + let retained = crate::vsr::encode_request_header( + &mut client.consensus_session.lock().unwrap(), + SEND_MESSAGES_CODE, + &payload, + ) + .unwrap() + .0; + let connection_guard = client.connection.lock().await; + let mut retained = Some(retained); + let attempt = client.send_raw_request(SEND_MESSAGES_CODE, payload, false, &mut retained); + tokio::pin!(attempt); + tokio::select! { + biased; + _ = &mut attempt => panic!("the exchange must wait for the connection lock"), + () = tokio::task::yield_now() => {} + } + client.reset_vsr_session().await.unwrap(); + client.bind_vsr_session(2).await.unwrap(); + drop(connection_guard); + assert_eq!( + tokio::time::timeout(TEST_BUDGET, &mut attempt) + .await + .unwrap(), + Err(IggyError::TransientNotCommitted) + ); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(50), peer.accept_bi()) + .await + .is_err() + ); + Client::shutdown(&client).await.unwrap(); + } + #[tokio::test] async fn a_lost_reply_resumes_and_replays_only_the_original_session() { const TEST_USER_ID: u32 = 7; diff --git a/core/sdk/src/session.rs b/core/sdk/src/session.rs index ecd22d3d7..0ab8d96f4 100644 --- a/core/sdk/src/session.rs +++ b/core/sdk/src/session.rs @@ -21,12 +21,13 @@ //! //! # Lifecycle //! -//! Create one session per logical login. `begin_register` preserves its client -//! identity and secret when registration is retried. `bind` accepts only a +//! Create one session per logical login. Registration preserves its client +//! identity and secret when retried. `bind` accepts only a //! nonzero epoch matching any previous binding. Application request IDs are //! available only after binding; exhaustion fails instead of wrapping. //! Both `bind` and `next_request_id` return `Result` and callers must handle -//! failure. Explicit disconnect clears sign-in and requires a new login. +//! failure. Explicit disconnect clears remembered sign-in; configured AutoLogin +//! still authenticates the next connection. use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_common::IggyError; @@ -86,7 +87,8 @@ impl ConsensusSession { self.session.is_some() } - /// Registration proof retained across reconnects and session bindings. + /// Internal bearer proof retained across reconnects. Never log or expose it. + #[doc(hidden)] pub fn bind_secret(&self) -> BindSecret { self.bind_secret.clone() } @@ -111,11 +113,6 @@ impl ConsensusSession { 0 } - /// Start or retry registration without resetting the identity or counter. - pub const fn begin_register(&self) -> u64 { - self.register_request_id() - } - /// Get the next application request ID and advance the counter. /// /// Returns 1, 2, 3, ... (request 0 is reserved for register). @@ -174,8 +171,8 @@ mod tests { fn lost_registration_reply_preserves_identity_and_secret() { let session = ConsensusSession::with_client_id(7); let secret = session.bind_secret(); - assert_eq!(session.begin_register(), 0); - assert_eq!(session.begin_register(), 0); + assert_eq!(session.register_request_id(), 0); + assert_eq!(session.register_request_id(), 0); assert_eq!(session.client_id(), 7); assert_eq!( session.bind_secret().expose_secret(), @@ -189,7 +186,7 @@ mod tests { let mut session = ConsensusSession::with_client_id(7); session.bind(42).unwrap(); assert_eq!(session.next_request_id().unwrap(), 1); - assert_eq!(session.begin_register(), 0); + assert_eq!(session.register_request_id(), 0); session.bind(42).unwrap(); assert_eq!(session.client_id(), 7); assert_eq!(session.session(), Some(42)); diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index a61f92450..4d448007b 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -1583,11 +1583,8 @@ impl TcpClient { let Ok(session) = consensus_session.lock() else { return (Some(header), Err(IggyError::InvalidConfiguration)); }; - if header.client != session.client_id() - || (header.operation != iggy_binary_protocol::Operation::Register - && header.session != session.session().unwrap_or(0)) - { - return (Some(header), Err(IggyError::TransientNotCommitted)); + if let Err(error) = crate::vsr::validate_retained_header(&header, &session) { + return (Some(header), Err(error)); } header } diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs index 84f275360..131bec936 100644 --- a/core/sdk/src/vsr.rs +++ b/core/sdk/src/vsr.rs @@ -68,7 +68,10 @@ pub(crate) fn encode_contiguous_request( retained_header: &mut Option<RequestHeader>, ) -> Result<Bytes, IggyError> { let header = match *retained_header { - Some(header) => header, + Some(header) => { + validate_retained_header(&header, session)?; + header + } None => encode_request_header(session, code, payload)?.0, }; *retained_header = Some(header); @@ -86,7 +89,7 @@ pub(crate) fn encode_request_header( ) -> Result<(RequestHeader, usize), IggyError> { let (operation, request_id, session_id) = match code { LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE => { - (Operation::Register, session.begin_register(), 0) + (Operation::Register, session.register_request_id(), 0) } _ => { let operation = operation_for_code(code); @@ -189,6 +192,19 @@ pub(crate) fn retain_replay_header( } } +pub(crate) fn validate_retained_header( + header: &RequestHeader, + session: &ConsensusSession, +) -> Result<(), IggyError> { + if header.client != session.client_id() + || (header.operation != Operation::Register + && header.session != session.session().unwrap_or(0)) + { + return Err(IggyError::TransientNotCommitted); + } + Ok(()) +} + pub(crate) fn response_size(header: &[u8]) -> Result<usize, IggyError> { let size = read_size_field(header).ok_or(IggyError::InvalidCommand)? as usize; if size < HEADER_SIZE { diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs index d83eb7029..f3e54e7ca 100644 --- a/core/sdk/src/websocket/websocket_client.rs +++ b/core/sdk/src/websocket/websocket_client.rs @@ -1403,6 +1403,59 @@ mod tests { stream.send(Message::Binary(frame.into())).await.unwrap(); } + #[tokio::test] + async fn a_retained_write_waiting_for_the_stream_must_recheck_its_session() { + const TEST_BUDGET: std::time::Duration = std::time::Duration::from_secs(1); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = WebSocketClient::create(Arc::new(WebSocketClientConfig { + server_address: listener.local_addr().unwrap().to_string(), + ..Default::default() + })) + .unwrap(); + let (mut peer, connected) = tokio::join!( + async { + accept_async(listener.accept().await.unwrap().0) + .await + .unwrap() + }, + Client::connect(&client) + ); + connected.unwrap(); + client.bind_vsr_session(1).await.unwrap(); + let payload = Bytes::from_static(b"uncertain-send"); + let retained = crate::vsr::encode_request_header( + &mut client.consensus_session.lock().unwrap(), + SEND_MESSAGES_CODE, + &payload, + ) + .unwrap() + .0; + let stream_guard = client.stream.lock().await; + let mut retained = Some(retained); + let attempt = client.send_raw_request(SEND_MESSAGES_CODE, payload, false, &mut retained); + tokio::pin!(attempt); + tokio::select! { + biased; + _ = &mut attempt => panic!("the exchange must wait for the stream lock"), + () = tokio::task::yield_now() => {} + } + client.reset_vsr_session().await.unwrap(); + client.bind_vsr_session(2).await.unwrap(); + drop(stream_guard); + assert_eq!( + tokio::time::timeout(TEST_BUDGET, &mut attempt) + .await + .unwrap(), + Err(IggyError::TransientNotCommitted) + ); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(50), peer.next()) + .await + .is_err() + ); + Client::shutdown(&client).await.unwrap(); + } + #[tokio::test] async fn a_lost_reply_resumes_and_replays_only_the_original_session() { const TEST_USER_ID: u32 = 7; diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs index 4f0780729..9f90119c9 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs @@ -16,6 +16,7 @@ // under the License. using System.Buffers; +using System.Buffers.Binary; using System.IO.Hashing; using System.Runtime.ExceptionServices; using Apache.Iggy.Contracts; @@ -184,6 +185,10 @@ public sealed partial class TcpMessageStream : ISessionGenerationProvider else if (await RedirectAsync(token)) { await ConnectAsync(false, token); + if (await ResumeSessionAsync(token)) + { + return authResponse; + } continue; } @@ -198,6 +203,54 @@ public sealed partial class TcpMessageStream : ISessionGenerationProvider } } + private async Task<bool> ResumeSessionAsync(CancellationToken token) + { + await _sendingSemaphore.WaitAsync(token); + try + { + if (_consensusSession.Session is not { } session || _connection is not { } connection) + { + return false; + } + var identity = new byte[ConsumerSessionSize]; + BinaryPrimitives.WriteUInt128LittleEndian(identity, _consensusSession.ClientId); + BinaryPrimitives.WriteUInt64LittleEndian(identity.AsSpan(ConsumerSessionEpochOffset), session); + lock (_pollRoutingGate) + { + BinaryPrimitives.WriteUInt64LittleEndian(identity.AsSpan(ConsumerSessionWatermarkOffset), _metadataWatermark); + } + var deadline = Environment.TickCount64 + VsrRequestTimeoutMs; + var attempt = await connection.SendAttemptAsync(CommandCodes.BIND_SESSION_CODE, + LoginRegister.SerializeBindSession(identity, _consensusSession.BindSecret), deadline, deadline, + false, token, retryTransient: false); + if (attempt.Error is IggyInvalidStatusCodeException + { + FromServer: true, + StatusCode: VsrError.UNAUTHENTICATED or VsrError.STALE_CLIENT + }) + { + ResetConsensusSession(); + return false; + } + if (attempt.Error is { } error) + { + ExceptionDispatchInfo.Throw(error); + } + using var response = attempt.Response!; + var bound = LoginRegister.Deserialize(response.Memory.Span); + if (bound.Session != session) + { + throw VsrError.Exception(VsrError.SESSION_MISMATCH, "Bind reply changed the logical session."); + } + SetConnectionState(ConnectionState.Authenticated); + return true; + } + finally + { + _sendingSemaphore.Release(); + } + } + /// <summary> /// Whether the partitioning has to be resolved to an explicit partition id before the request is framed. /// This client resolves balanced and message-key kinds using its partition cache and cursor. @@ -805,13 +858,13 @@ public sealed partial class TcpMessageStream : ISessionGenerationProvider } /// <summary> - /// Drops the connection along with the session. A late or half-read reply would desync the framing of the + /// Drops the connection while retaining the session. A late or half-read reply would desync the framing of the /// next request, so the socket cannot be reused. The caller must hold <see cref="_sendingSemaphore" />, /// which owns every write to <see cref="_connection" />. /// </summary> /// <param name="connection"> - /// The connection the caller was using. A reconnect that completed in the meantime already closed it and - /// re-armed the session, so dropping anything but the live one would tear down a healthy replacement. + /// The connection the caller was using. A reconnect that completed in the meantime already closed it, + /// so dropping anything but the live one would tear down a healthy replacement. /// </param> private void DropVsrConnectionLocked(VsrConnection? connection) { @@ -820,7 +873,8 @@ public sealed partial class TcpMessageStream : ISessionGenerationProvider return; } - ResetConsensusSession(); + _groupState.ClearSessionScoped(); + ClearPollSession(); _connection = null; SetConnectionState(ConnectionState.Disconnected); connection.Dispose(); diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs index 21b09f395..09b0c93c2 100644 --- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs +++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs @@ -1140,7 +1140,10 @@ public sealed partial class TcpMessageStream : IIggyClient // forwards the register to the primary. if (autoLogin && SignInSettings() is { } signInSettings) { - await AutoLoginAsync(signInSettings, token); + if (!await ResumeSessionAsync(token)) + { + await AutoLoginAsync(signInSettings, token); + } if (settleOnLeader && await RedirectAsync(token)) { @@ -1259,7 +1262,7 @@ public sealed partial class TcpMessageStream : IIggyClient } /// <summary> - /// Closes the current connection and forgets the consensus session bound to it. Takes the sending + /// Closes the current connection while retaining its logical session. Takes the sending /// semaphore, which owns every write to <see cref="_connection" />, so an in-flight request never /// observes the field changing between its write and its reply. Never cancellable: a caller giving up is /// exactly when the connection has to be released. @@ -1272,7 +1275,8 @@ public sealed partial class TcpMessageStream : IIggyClient _connection?.Dispose(); _connection = null; - ResetConsensusSession(); + _groupState.ClearSessionScoped(); + ClearPollSession(); } finally { diff --git a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs index a23e1497a..cc2120385 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs @@ -204,7 +204,7 @@ internal sealed class ConsensusSession } } - /// <summary>Forget the binding and the client id, e.g. after an eviction or a torn connection.</summary> + /// <summary>Forget the binding and client id after logout or a terminal bind refusal.</summary> internal void Reset() { lock (_gate) diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs index 7e79131c4..214839d9b 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs @@ -38,7 +38,7 @@ internal sealed class VsrConnection : IDisposable private readonly long _maxResponseFrameSize; /// <summary> - /// Tears this connection down at the transport level - session reset, state event - when a frame-level + /// Tears this connection down at the transport level when a frame-level /// failure makes the socket unusable. Runs under the sending lock the caller holds, and it is the /// transport's job to ignore the call when a reconnect already replaced this connection. /// </summary> @@ -143,7 +143,12 @@ internal sealed class VsrConnection : IDisposable catch (IggyInvalidStatusCodeException e) { uncertain |= e is { FromServer: true, StatusCode: VsrError.TRANSIENT_NOT_COMMITTED }; - var verdict = uncertain && e is { FromServer: true, StatusCode: VsrError.TRANSIENT_NOT_ACCEPTED } + var verdict = uncertain && e is + { + FromServer: true, + StatusCode: VsrError.TRANSIENT_NOT_ACCEPTED or VsrError.UNAUTHORIZED or + VsrError.UNAUTHENTICATED or VsrError.STALE_CLIENT + } ? VsrError.FromServer(VsrError.TRANSIENT_NOT_COMMITTED, "A later refusal does not resolve the original request outcome.") : e; diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs index 6d01e66f0..87a6dd445 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs @@ -32,6 +32,7 @@ internal static class VsrError internal const int INVALID_IDENTIFIER = 6; internal const int STALE_CLIENT = 30; internal const int UNAUTHENTICATED = 40; + internal const int UNAUTHORIZED = 41; internal const int INVALID_CREDENTIALS = 42; internal const int INVALID_USERNAME = 43; internal const int INVALID_PASSWORD = 44; @@ -44,6 +45,7 @@ internal static class VsrError internal const int CONSUMER_GROUP_MEMBER_NOT_FOUND = 5006; internal const int CONSUMER_GROUP_PARTITION_NOT_OWNED = 5009; internal const int INCOMPATIBLE_PROTOCOL_VERSION = 14003; + internal const int SESSION_MISMATCH = 14004; /// <summary>A failure the client raised itself, before or instead of a server verdict.</summary> internal static IggyInvalidStatusCodeException Exception(int code, string message) diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs index 4bcc888d3..fd897528f 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs @@ -40,8 +40,12 @@ namespace Apache.Iggy.Tests.VsrTests; /// </summary> public sealed class EndpointFailoverTests { - [Fact] - public async Task LaterRefusalsKeepTheIdentityOfAnUncertainMetadataWrite() + [Theory] + [InlineData(VsrError.TRANSIENT_NOT_ACCEPTED)] + [InlineData(VsrError.UNAUTHORIZED)] + [InlineData(VsrError.UNAUTHENTICATED)] + [InlineData(VsrError.STALE_CLIENT)] + public async Task LaterRefusalsKeepTheIdentityOfAnUncertainMetadataWrite(int refusal) { const int RefusalCount = 50; using var node = new MockNode(); @@ -54,7 +58,7 @@ public sealed class EndpointFailoverTests var status = attempts.Count switch { 1 => VsrError.TRANSIENT_NOT_COMMITTED, - <= RefusalCount => VsrError.TRANSIENT_NOT_ACCEPTED, + <= RefusalCount => refusal, _ => 0 }; return Reply(request.Operation, status == 0 ? new byte[4] : [], (uint)status); @@ -92,7 +96,6 @@ public sealed class EndpointFailoverTests { using var primary = new MockNode(); using var survivor = new MockNode(); - var survivorRegistrations = new ConcurrentQueue<MockRequest>(); // The primary leads, so the sign-in settles there and the roster is only remembered - not acted on - // until the node dies. @@ -101,10 +104,6 @@ public sealed class EndpointFailoverTests : Answer(request)); survivor.Serve(request => { - if (request.Operation == OPERATION_REGISTER) - { - survivorRegistrations.Enqueue(request); - } return request.Code == GET_CLUSTER_METADATA_CODE ? Reply(OPERATION_NON_REPLICATED, ClusterMetadata(primary.Port, survivor.Port, survivor.Port)) : Answer(request); @@ -144,12 +143,114 @@ public sealed class EndpointFailoverTests 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.Equal(0, survivor.Registrations); + Assert.True(survivor.Requests(BIND_SESSION_CODE) >= 1, "the retained session resumed on the survivor"); Assert.True(survivor.Pings >= 1, "the request landed on the survivor"); - Assert.NotEmpty(survivorRegistrations); - Assert.All(survivorRegistrations, request => Assert.Equal( - LoginRegister.Serialize(changeCredentials ? "renamed" : "iggy", changeCredentials ? "new-secret" : "iggy", - LoginRegister.ReadBindSecret(request.Body)), request.Body)); + } + + [Theory] + [InlineData(0)] + [InlineData(VsrError.UNAUTHENTICATED)] + [InlineData(VsrError.STALE_CLIENT)] + [InlineData(VsrError.TRANSIENT_NOT_COMMITTED)] + [InlineData(VsrError.TRANSIENT_NOT_ACCEPTED)] + [InlineData(VsrError.SESSION_MISMATCH)] + public async Task ReconnectUsesBindBeforeReplacingALogicalSession(int bindStatus) + { + const ulong FirstSession = 128; + using var node = new MockNode(); + var registrations = new ConcurrentQueue<MockRequest>(); + var bindings = new ConcurrentQueue<MockRequest>(); + var writes = new ConcurrentQueue<MockRequest>(); + var pings = new ConcurrentQueue<MockRequest>(); + node.Serve(request => + { + if (request.Operation == OPERATION_REGISTER) + { + registrations.Enqueue(request); + return Reply(request.Operation, RegisterBody(FirstSession + (ulong)registrations.Count - 1)); + } + if (request.Code == BIND_SESSION_CODE) + { + bindings.Enqueue(request); + if (bindings.Count == 1 && bindStatus == VsrError.SESSION_MISMATCH) + { + return Reply(request.Operation, RegisterBody(request.Session + 1)[4..]); + } + return bindings.Count == 1 && bindStatus != 0 + ? Reply(request.Operation, [], (uint)bindStatus) + : Answer(request); + } + if (request.Operation == (byte)VsrOperation.CreateStream) + { + writes.Enqueue(request); + throw new IOException("The connection lost the committed write reply."); + } + if (request.Code == PING_CODE) + { + pings.Enqueue(request); + } + return request.Code == GET_CLUSTER_METADATA_CODE + ? Reply(OPERATION_NON_REPLICATED, ClusterMetadata(node.Port, node.Port, node.Port)) + : Answer(request); + }); + using var client = new TcpMessageStream(new IggyClientConfigurator + { + BaseAddress = $"127.0.0.1:{node.Port}", + Protocol = Protocol.Tcp, + HeartbeatInterval = TimeSpan.FromHours(1), + ReconnectionSettings = new ReconnectionSettings + { + Enabled = true, + MaxRetries = 1, + InitialDelay = TimeSpan.Zero, + WaitAfterReconnect = TimeSpan.Zero + } + }, NullLoggerFactory.Instance); + await client.ConnectAsync(TestContext.Current.CancellationToken); + await client.LoginUserAsync("iggy", "iggy", TestContext.Current.CancellationToken); + var generation = ((ISessionGenerationProvider)client).SessionGeneration; + var registration = Assert.Single(registrations); + + await Assert.ThrowsAsync<VsrRequestOutcomeUnknownException>(() => + client.CreateStreamAsync("lost-reply", token: TestContext.Current.CancellationToken)); + var write = Assert.Single(writes); + Assert.Equal(registration.ClientId, write.ClientId); + + if (bindStatus is VsrError.TRANSIENT_NOT_COMMITTED or VsrError.TRANSIENT_NOT_ACCEPTED or VsrError.SESSION_MISMATCH) + { + var refusal = await Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() => + client.PingAsync(TestContext.Current.CancellationToken)); + Assert.Equal(bindStatus, refusal.StatusCode); + Assert.Single(registrations); + Assert.Equal(generation, ((ISessionGenerationProvider)client).SessionGeneration); + } + await client.PingAsync(TestContext.Current.CancellationToken); + + Assert.Single(writes); + Assert.NotEmpty(bindings); + Assert.All(bindings, binding => + { + Assert.Equal(registration.ClientId, binding.ClientId); + Assert.Equal(write.Session, binding.Session); + Assert.Equal(LoginRegister.ReadBindSecret(registration.Body), + binding.Body.AsSpan(binding.Body.Length - LoginRegister.BIND_SECRET_BYTES).ToArray()); + }); + var ping = Assert.Single(pings); + if (bindStatus is VsrError.UNAUTHENTICATED or VsrError.STALE_CLIENT) + { + Assert.Equal(2, registrations.Count); + Assert.NotEqual(registration.ClientId, ping.ClientId); + Assert.NotEqual(write.Session, ping.Session); + Assert.NotEqual(generation, ((ISessionGenerationProvider)client).SessionGeneration); + } + else + { + Assert.Single(registrations); + Assert.Equal(write.ClientId, ping.ClientId); + Assert.Equal(write.Session, ping.Session); + Assert.Equal(generation, ((ISessionGenerationProvider)client).SessionGeneration); + } } [Fact] @@ -265,6 +366,11 @@ public sealed class EndpointFailoverTests return Reply(OPERATION_REGISTER, RegisterBody(session: 128)); } + if (request.Code == BIND_SESSION_CODE) + { + return Reply(OPERATION_NON_REPLICATED, [], VsrError.UNAUTHENTICATED); + } + if (evict) { evict = false; @@ -321,6 +427,11 @@ public sealed class EndpointFailoverTests return Reply(OPERATION_REGISTER, RegisterBody(session: 128)); } + if (request.Code == BIND_SESSION_CODE) + { + return Reply(OPERATION_NON_REPLICATED, [], VsrError.UNAUTHENTICATED); + } + if (evict) { evict = false; @@ -532,7 +643,8 @@ public sealed class EndpointFailoverTests await comesUp; Assert.NotNull(survivor); - Assert.True(survivor!.Registrations >= 1, "the session was re-established on the survivor"); + Assert.Equal(0, survivor!.Registrations); + Assert.True(survivor.Requests(BIND_SESSION_CODE) >= 1, "the session resumed on the survivor"); } finally { diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs index f740a833d..d857ca729 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs @@ -43,12 +43,17 @@ internal static class MockFrames internal const byte OPERATION_REGISTER = 1; internal const byte OPERATION_NON_REPLICATED = 2; internal const int GET_CLUSTER_METADATA_CODE = 12; + internal const int BIND_SESSION_CODE = 15; internal const int PING_CODE = 1; internal const uint TRANSIENT_NOT_ACCEPTED = 58; /// <summary>A reply for anything the roster read does not claim: a register, or an empty read.</summary> internal static byte[] Answer(MockRequest request) { + if (request.Code == BIND_SESSION_CODE) + { + return Reply(OPERATION_NON_REPLICATED, RegisterBody(request.Session)[4..]); + } return request.Operation == OPERATION_REGISTER ? Reply(OPERATION_REGISTER, RegisterBody(session: 128)) : Reply(OPERATION_NON_REPLICATED, []); diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs index d4e2818c2..d04f3ce1c 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs @@ -389,8 +389,9 @@ public sealed class PrimaryPollingTests await PollAsync(client); Assert.Equal(2, routes); - Assert.Equal(2, cluster.Coordinator.Registrations); - Assert.NotEqual(generation, ((ISessionGenerationProvider)client).SessionGeneration); + Assert.Equal(1, cluster.Coordinator.Registrations); + Assert.Equal(1, cluster.Coordinator.Requests(CommandCodes.BIND_SESSION_CODE)); + Assert.Equal(generation, ((ISessionGenerationProvider)client).SessionGeneration); Assert.Equal(1, cluster.Primaries[0].Requests(CommandCodes.POLL_MESSAGES_ON_PRIMARY_CODE)); } diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md index ba795a8a8..d2a9fb5d5 100644 --- a/foreign/csharp/README.md +++ b/foreign/csharp/README.md @@ -144,9 +144,8 @@ using var client = IggyClientFactory.CreateClient(new IggyClientConfigurator BackoffMultiplier = 2.0 }, - // Auto-login after connection. Optional for reconnection: a client that signs in with - // LoginUserAsync has that sign-in replayed on a reconnect too. Without either, a reconnect - // cannot restore the session and a lost connection fails the request + // Auto-login after connection. Reconnect first binds the retained session; configured or + // remembered LoginUserAsync credentials allow a fresh login after that session ends AutoLoginSettings = AutoLoginSettings.For("iggy", "iggy"), // or AutoLoginSettings.ForPersonalAccessToken("your_token") @@ -186,8 +185,11 @@ await client.ConnectAsync(); ### What changes under VSR - **Login registers a shared session in this checkout.** `LoginUserAsync` / `LoginWithPersonalAccessTokenAsync` - authenticate Register. Auxiliary connections bind the parent's identity and proof without another Register. - Disconnecting removes only that binding; Logout or committed expiry ends the session. An unresolved mutation + authenticate Register. Auxiliary and reconnected transports bind the retained identity and proof without another Register. + Disconnecting removes only that binding; Logout, committed lease expiry or user deactivation ends the session. + Password changes and PAT changes or expiry do not end an established session. + A terminal BindSession refusal (`Unauthenticated` or `StaleClient`) permits a fresh login. A transient refusal + retains the identity and surfaces to the caller, allowing a later reconnect to retry the bind. An unresolved mutation cannot be replayed under a fresh session. The older 0.9.0 connection-lifetime rule does not apply to this checkout. - **Leader redirection is automatic.** The client reads the cluster roster, follows the current leader and re-checks it when a request is refused because the node stopped being primary. diff --git a/foreign/go/README.md b/foreign/go/README.md index aeeb1afd5..2328477ef 100644 --- a/foreign/go/README.md +++ b/foreign/go/README.md @@ -50,9 +50,14 @@ Disable the ones you do not need so they cannot race with another process. `SendMessages` returns any placements the server reports. A send whose reply is lost to a dropped connection returns `ErrDisconnected` without a replay. -A reconnect registers a new client identity, so a caller retry can append -the batch twice. Consumers must handle duplicates through idempotent -processing or application-level deduplication. +A reconnect retains the client identity and authenticates it with `BindSession`. +Only a terminal bind refusal permits a new registration. An application resend +still creates a new request and can append the batch twice. Consumers must handle +duplicates through idempotent processing or application-level deduplication. + +An established session has its own lifetime. Password or PAT changes and PAT +expiry do not end it. Logout, session lease expiry, or user deactivation end the +session and prevent its bind proof from restoring it. Crash durability follows the topic's `durability` policy: `replicated` confirms replication, while `persisted` also waits for the required replicas diff --git a/foreign/go/client/tcp/tcp_connect_test.go b/foreign/go/client/tcp/tcp_connect_test.go index 12fcd615a..7176cfcb1 100644 --- a/foreign/go/client/tcp/tcp_connect_test.go +++ b/foreign/go/client/tcp/tcp_connect_test.go @@ -170,6 +170,8 @@ func singleNodeHandler(t *testing.T, address func() string) func(int, int, reque t.Helper() return func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 0, address()) case read.operation() == vsr.OperationRegister: @@ -241,6 +243,8 @@ func TestConnect_RedirectsToTheLeaderAfterSigningIn(t *testing.T) { // the primary) and only then reports the leader elsewhere. follower := listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 1, "127.0.0.1:1", leader.address()) case read.operation() == vsr.OperationRegister: @@ -262,11 +266,11 @@ func TestConnect_RedirectsToTheLeaderAfterSigningIn(t *testing.T) { onLeader := leader.recorded() require.Len(t, onLeader, 2, "the sign-in replay is followed by a leader re-check") - assert.Equal(t, vsr.OperationRegister, onLeader[0].operation()) + assert.Equal(t, uint32(command.BindSessionCode), onLeader[0].code()) assert.Equal(t, uint32(command.GetClusterMetadataCode), onLeader[1].code()) assert.True(t, client.session.Bound()) - assert.Equal(t, uint64(128), client.session.SessionID(), - "the leader's session superseded the follower's") + assert.Equal(t, uint64(512), client.session.SessionID(), + "the leader resumes the original session") } func TestConnect_RechecksLeadershipAfterTheRedirectedSignIn(t *testing.T) { @@ -276,6 +280,8 @@ func TestConnect_RechecksLeadershipAfterTheRedirectedSignIn(t *testing.T) { var intermediate *testListener intermediate = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 1, intermediate.address(), leader.address()) case read.operation() == vsr.OperationRegister: @@ -288,6 +294,8 @@ func TestConnect_RechecksLeadershipAfterTheRedirectedSignIn(t *testing.T) { var follower *testListener follower = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 1, follower.address(), intermediate.address()) case read.operation() == vsr.OperationRegister: @@ -319,10 +327,12 @@ func TestConnect_KeepsTheConnectionWhenTheRosterHasOneNode(t *testing.T) { assert.Equal(t, 1, server.connections(), "a one-node roster never redirects") } -func TestExchange_ReSignsInBeforeReplayingANonReplicatedRequest(t *testing.T) { +func TestExchange_ResumesBeforeReplayingANonReplicatedRequest(t *testing.T) { var server *testListener server = listenVSR(t, nil, func(connection, index int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 0, server.address()) case read.operation() == vsr.OperationRegister: @@ -346,20 +356,22 @@ func TestExchange_ReSignsInBeforeReplayingANonReplicatedRequest(t *testing.T) { require.Len(t, recorded, 6) assert.Equal(t, uint32(command.PingCode), recorded[2].code(), "the dropped attempt") - assert.Equal(t, vsr.OperationRegister, recorded[3].operation(), - "the replay signs in again before it repeats the request") + assert.Equal(t, uint32(command.BindSessionCode), recorded[3].code(), + "the replay resumes before it repeats the request") assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[4].code()) assert.Equal(t, uint32(command.PingCode), recorded[5].code()) - assert.NotEqual(t, recorded[2].clientID(), recorded[5].clientID(), - "a fresh connection registers a new client identity") - assert.Equal(t, uint64(129), client.session.SessionID()) + assert.Equal(t, recorded[2].clientID(), recorded[5].clientID(), + "a replacement connection preserves the client identity") + assert.Equal(t, uint64(128), client.session.SessionID()) } func TestExchange_RefusesToReplayAReplicatedRequestWithAnUnknownOutcome(t *testing.T) { var server *testListener server = listenVSR(t, nil, func(connection, index int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 0, server.address()) case read.operation() == vsr.OperationRegister: @@ -392,6 +404,8 @@ func TestExchange_ReconnectsThroughAKnownFollowerAfterTheLeaderDies(t *testing.T var follower, leader *testListener follower = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): leaderIndex := 1 if leaderFailed.Load() { @@ -406,6 +420,8 @@ func TestExchange_ReconnectsThroughAKnownFollowerAfterTheLeaderDies(t *testing.T }) leader = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 1, follower.address(), leader.address()) case read.operation() == vsr.OperationRegister: @@ -431,8 +447,8 @@ func TestExchange_ReconnectsThroughAKnownFollowerAfterTheLeaderDies(t *testing.T assert.Equal(t, vsr.OperationRegister, recorded[0].operation(), "the dialed follower completed the initial sign-in") assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[1].code()) - assert.Equal(t, vsr.OperationRegister, recorded[2].operation(), - "the reconnect signed in again on the surviving node") + assert.Equal(t, uint32(command.BindSessionCode), recorded[2].code(), + "the reconnect resumed the retained session on the surviving node") assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[3].code()) assert.Equal(t, uint32(command.PingCode), recorded[4].code()) } @@ -461,6 +477,8 @@ func TestExchange_DoesNotPreemptAReplayedSignIn(t *testing.T) { var mtx sync.Mutex server = listenVSR(t, nil, func(connection, index int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 0, server.address()) case read.operation() == vsr.OperationRegister: @@ -497,7 +515,7 @@ func TestExchange_DoesNotPreemptAReplayedSignIn(t *testing.T) { assert.Equal(t, 3, signIns) // The suppression rides the replay's own context, so nothing about it - // outlives that call: the next Connect signs in again. + // outlives that call: the next Connect resumes the bound session. require.NoError(t, client.disconnect()) require.NoError(t, client.Connect(context.Background())) signIns = 0 @@ -506,7 +524,8 @@ func TestExchange_DoesNotPreemptAReplayedSignIn(t *testing.T) { signIns++ } } - assert.Equal(t, 4, signIns, "the suppression leaked past the call that meant it") + assert.Equal(t, 3, signIns, "resume must not register another session") + assert.Equal(t, 1, requestCount(server.recorded(), command.BindSessionCode)) } func TestExchange_FailsFastWhenAutoLoginIsOff(t *testing.T) { @@ -531,6 +550,8 @@ func TestLogoutUser_ResetsTheSessionSoTheNextSignInIsANewIdentity(t *testing.T) var server *testListener server = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 0, server.address()) case read.operation() == vsr.OperationRegister: diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go index 3871d09ff..b6b65a17f 100644 --- a/foreign/go/client/tcp/tcp_core.go +++ b/foreign/go/client/tcp/tcp_core.go @@ -1021,7 +1021,6 @@ func (c *IggyTcpClient) invalidateConnLocked() { _ = c.closeConnLocked() c.transportState = iggcon.TransportStateDisconnected c.sessionState = iggcon.SessionStateUnauthenticated - c.session.Reset() c.clearPollSession() c.groups.clear() c.topics.clearCounts() @@ -1223,9 +1222,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) (err error) { c.connGeneration++ c.transportState = iggcon.TransportStateConnected c.connectedAt = time.Now() - // The server fence does not survive the old socket, so the new connection - // starts from a fresh client identity. - c.session.Reset() + // The logical identity survives the socket and is authenticated by BindSession. c.clearPollSession() clientAddress := c.clientAddress serverAddress := c.currentServerAddress @@ -1385,6 +1382,15 @@ func (c *IggyTcpClient) establishSession(ctx context.Context, skipAutoLogin bool c.logger.Info("Skipping the automatic sign-in for a replayed login.") return nil } + c.registerMtx.Lock() + identity, resumeErr := c.resumeSession(ctx) + if resumeErr == nil && identity != nil { + _, resumeErr = c.settleOnLeader(ctx, uint32(command.BindSessionCode), nil, [vsr.BindSecretBytes]byte{}) + } + c.registerMtx.Unlock() + if resumeErr != nil || identity != nil { + return resumeErr + } if credentials.personalAccessToken != "" { _, err := c.LoginWithPersonalAccessToken(ctx, credentials.personalAccessToken) @@ -1520,7 +1526,6 @@ func (c *IggyTcpClient) disconnectLocked() error { c.logger.Info("Iggy client is disconnecting from server...", slog.String("client_address", c.clientAddress)) c.transportState = iggcon.TransportStateDisconnected c.sessionState = iggcon.SessionStateUnauthenticated - c.session.Reset() c.clearPollSession() c.groups.clear() c.topics.clearCounts() diff --git a/foreign/go/client/tcp/tcp_core_test.go b/foreign/go/client/tcp/tcp_core_test.go index 5a085222c..f94fe296c 100644 --- a/foreign/go/client/tcp/tcp_core_test.go +++ b/foreign/go/client/tcp/tcp_core_test.go @@ -296,7 +296,7 @@ func TestExchange_EscalatesNotAcceptedToALeaderRecheck(t *testing.T) { assert.True(t, sawMetadata, "the client re-checked leadership") } -func TestExchange_ResetsTheSessionOnAnEviction(t *testing.T) { +func TestExchange_RetainsTheSessionUntilBindConfirmsAnEviction(t *testing.T) { client, serverConn := newPipeClient(t) client.config.reconnection.enabled = false before := client.session.ClientID() @@ -309,8 +309,8 @@ func TestExchange_ResetsTheSessionOnAnEviction(t *testing.T) { var eviction *vsr.EvictionError assert.ErrorAs(t, err, &eviction) - assert.False(t, client.session.Bound(), "the fence no longer holds") - assert.NotEqual(t, before, client.session.ClientID(), "a new identity is minted") + assert.True(t, client.session.Bound(), "retain the proof until the resume verdict") + assert.Equal(t, before, client.session.ClientID(), "an eviction does not authorize replay under a new identity") assert.Equal(t, iggcon.SessionStateUnauthenticated, client.sessionState) } diff --git a/foreign/go/client/tcp/tcp_failover_test.go b/foreign/go/client/tcp/tcp_failover_test.go index 9a0ca663e..ae200bec8 100644 --- a/foreign/go/client/tcp/tcp_failover_test.go +++ b/foreign/go/client/tcp/tcp_failover_test.go @@ -35,8 +35,82 @@ import ( ) // 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. +// survivor the roster named, under the retained logical identity. // Mirrors `core/integration/tests/cluster/failover_client_continuity.rs`. +func TestReconnect_BindVerdictPreservesOrReplacesOnlyTheLogicalSession(t *testing.T) { + for _, refusal := range []uint32{0, 30, 40, 57, 58} { + t.Run(ierror.FromCode(ierror.Code(refusal)).Error(), func(t *testing.T) { + var server *testListener + var registrations atomic.Int32 + var writes atomic.Int32 + var bindStatus atomic.Uint32 + bindStatus.Store(refusal) + server = listenVSR(t, nil, func(_, _ int, read request) []byte { + switch { + case read.operation() == vsr.OperationRegister: + return registerReplyFrame(7, uint64(128+registrations.Add(1))) + case read.code() == uint32(command.BindSessionCode): + if status := bindStatus.Load(); status != 0 { + return statusReplyFrame(vsr.OperationNonReplicated, status, nil) + } + return bindReplyFrame(7, read.sessionID()) + case read.code() == uint32(command.GetClusterMetadataCode): + return clusterMetadataFrame(t, 0, server.address()) + case read.operation() == vsr.OperationCreateStream: + if writes.Add(1) == 1 { + return nil + } + return replyFrame(vsr.OperationCreateStream, resultSection()) + default: + return replyFrame(vsr.OperationNonReplicated, nil) + } + }) + client := newDialingClient(t, server.address(), + WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy"))) + ctx := context.Background() + require.NoError(t, client.Connect(ctx)) + original := client.session.ClientID() + secret := client.session.BindSecret() + _, err := client.do(ctx, &command.CreateStream{Name: "lost-reply"}) + require.Error(t, err) + if refusal == uint32(ierror.TransientNotCommittedCode) || refusal == uint32(ierror.TransientNotAcceptedCode) { + require.ErrorIs(t, client.Connect(ctx), ierror.FromCode(ierror.Code(refusal))) + assert.Equal(t, original, client.session.ClientID()) + assert.Equal(t, int32(1), registrations.Load()) + bindStatus.Store(0) + } + require.NoError(t, client.Connect(ctx)) + assert.Equal(t, int32(1), writes.Load(), "resume must not replay the uncertain mutation") + terminal := refusal == uint32(ierror.StaleClientCode) || refusal == uint32(ierror.UnauthenticatedCode) + if terminal { + assert.NotEqual(t, original, client.session.ClientID()) + assert.Equal(t, int32(2), registrations.Load()) + } else { + assert.Equal(t, original, client.session.ClientID()) + assert.Equal(t, secret, client.session.BindSecret()) + assert.Equal(t, int32(1), registrations.Load()) + } + _, err = client.do(ctx, &command.CreateStream{Name: "explicit-new-write"}) + require.NoError(t, err) + var mutations []request + for _, read := range server.recorded() { + if read.operation() == vsr.OperationCreateStream { + mutations = append(mutations, read) + } + } + require.Len(t, mutations, 2) + assert.NotEqual(t, mutations[0].payload, mutations[1].payload) + if terminal { + assert.NotEqual(t, mutations[0].clientID(), mutations[1].clientID()) + assert.Equal(t, uint64(1), mutations[1].requestID()) + } else { + assert.Equal(t, mutations[0].clientID(), mutations[1].clientID()) + assert.Equal(t, mutations[0].requestID()+1, mutations[1].requestID()) + } + }) + } +} + func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) { var survivor *testListener var primary *testListener @@ -44,6 +118,8 @@ func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) { survivor = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 1, primary.address(), survivor.address()) case read.operation() == vsr.OperationRegister: @@ -60,6 +136,8 @@ func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) { return nil } switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) 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. @@ -97,14 +175,17 @@ func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) { registers++ } } - assert.Equal(t, 1, registers, - "the remembered credentials signed in again on the survivor") + assert.Equal(t, 0, registers, "resume must not register another session") + assert.Equal(t, 1, requestCount(survivor.recorded(), command.BindSessionCode)) } func TestFailover_WalksPastTwoRefusingReplicasToThePartitionPrimary(t *testing.T) { var metadataLeader *testListener partitionPrimary := listenVSR(t, nil, func(_, _ int, read request) []byte { + if read.code() == uint32(command.BindSessionCode) { + return bindReplyFrame(7, read.sessionID()) + } switch read.operation() { case vsr.OperationRegister: return registerReplyFrame(7, 384) @@ -115,6 +196,9 @@ func TestFailover_WalksPastTwoRefusingReplicasToThePartitionPrimary(t *testing.T } }) follower := listenVSR(t, nil, func(_, _ int, read request) []byte { + if read.code() == uint32(command.BindSessionCode) { + return bindReplyFrame(7, read.sessionID()) + } if read.operation() == vsr.OperationRegister { return registerReplyFrame(7, 256) } @@ -123,6 +207,8 @@ func TestFailover_WalksPastTwoRefusingReplicasToThePartitionPrimary(t *testing.T }) metadataLeader = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 0, metadataLeader.address(), follower.address(), partitionPrimary.address()) @@ -195,6 +281,9 @@ func TestFailover_ServerEvictionReplaysTheRememberedSignIn(t *testing.T) { var evict atomic.Bool var registers atomic.Int32 server = listenVSR(t, nil, func(_, _ int, read request) []byte { + if read.code() == uint32(command.BindSessionCode) { + return statusReplyFrame(vsr.OperationNonReplicated, uint32(ierror.ErrUnauthenticated.Code()), nil) + } if read.operation() == vsr.OperationRegister { registers.Add(1) return registerReplyFrame(7, 128) @@ -325,6 +414,8 @@ func TestFailover_ReLoginSurvivesALogoutTheOldPrimaryRefused(t *testing.T) { // refuses it as not-admitted and points at the survivor. follower = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): if demoted.Load() { return clusterMetadataFrame(t, 1, follower.address(), leader.address()) @@ -342,6 +433,8 @@ func TestFailover_ReLoginSurvivesALogoutTheOldPrimaryRefused(t *testing.T) { }) leader = listenVSR(t, nil, func(_, _ int, read request) []byte { switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.GetClusterMetadataCode): return clusterMetadataFrame(t, 1, follower.address(), leader.address()) case read.operation() == vsr.OperationRegister: diff --git a/foreign/go/client/tcp/tcp_poll_routing_test.go b/foreign/go/client/tcp/tcp_poll_routing_test.go index 1f04987f7..1e311b63a 100644 --- a/foreign/go/client/tcp/tcp_poll_routing_test.go +++ b/foreign/go/client/tcp/tcp_poll_routing_test.go @@ -79,6 +79,8 @@ func newPrimaryPollFixture(t *testing.T, } } switch { + case read.code() == uint32(command.BindSessionCode): + return bindReplyFrame(7, read.sessionID()) case read.operation() == vsr.OperationRegister: return registerReplyFrame(7, uint64(100+connection)) case read.sessionID() == 0: @@ -630,7 +632,7 @@ func TestPrimaryPoll_ControlConnectionRecoversBeforeRouting(t *testing.T) { previous := fixture.client.session.ClientID() _, err := pollPrimaryPartition(context.Background(), fixture.client, 0) require.NoError(t, err) - assert.NotEqual(t, previous, fixture.client.session.ClientID()) + assert.Equal(t, previous, fixture.client.session.ClientID()) assert.Equal(t, 2, fixture.coordinator.connections()) assert.Equal(t, 1, fixture.primaries[0].connections()) } diff --git a/foreign/go/client/tcp/tcp_session_management.go b/foreign/go/client/tcp/tcp_session_management.go index 034695527..d5a2cb545 100644 --- a/foreign/go/client/tcp/tcp_session_management.go +++ b/foreign/go/client/tcp/tcp_session_management.go @@ -19,10 +19,12 @@ package tcp import ( "context" + "errors" "log/slog" "time" iggcon "github.com/apache/iggy/foreign/go/contracts" + ierror "github.com/apache/iggy/foreign/go/errors" "github.com/apache/iggy/foreign/go/internal/command" "github.com/apache/iggy/foreign/go/internal/util" "github.com/apache/iggy/foreign/go/internal/vsr" @@ -162,10 +164,9 @@ func (c *IggyTcpClient) signIn(ctx context.Context, code uint32, body []byte, se // auth-gated, so the topology cannot be inspected before a login binds a // session. A login dialed at a backup still succeeds (the server forwards the // register to the primary); this settlement decides where later requests -// land, not whether the sign-in works. The redirect drops the fresh session -// along with the socket, so the sign-in is replayed on the leader and its -// identity supersedes the dialed node's. Leadership can move between the -// roster read and the replay, so each freshly bound hop rechecks the roster +// land, not whether the sign-in works. The redirect retains the session and +// binds its proof on the leader. Leadership can move between the +// roster read and the bind, so each authenticated hop rechecks the roster // under the shared redirect budget. // // Returns nil when the client stays where it is. @@ -198,13 +199,61 @@ func (c *IggyTcpClient) settleOnLeader(ctx context.Context, code uint32, body [] if err := c.Connect(suppressAutoLogin(ctx)); err != nil { return nil, err } - settled, err = c.signIn(ctx, code, body, secret) + settled, err = c.resumeSession(ctx) + if err == nil && settled == nil { + if code == uint32(command.BindSessionCode) { + return nil, ierror.ErrUnauthenticated + } + settled, err = c.signIn(ctx, code, body, secret) + } if err != nil { return nil, err } } } +// resumeSession preserves the request sequence while authenticating a replacement socket. +// Only a terminal bind refusal permits a new registration. +func (c *IggyTcpClient) resumeSession(ctx context.Context) (*iggcon.IdentityInfo, error) { + c.mtx.Lock() + if !c.session.Bound() { + c.mtx.Unlock() + return nil, nil + } + identity := consumerSession{client: c.session.ClientID(), session: c.session.SessionID(), watermark: c.metadataWatermark.Load()} + secret := c.session.BindSecret() + c.mtx.Unlock() + body, err := vsr.SerializeBindSession(identity.bytes(), iggcon.Version, secret) + if err != nil { + return nil, err + } + frame := append(reserveHeader(nil), body...) + response, _, err := c.sendFrame(ctx, uint32(command.BindSessionCode), frame) + if errors.Is(err, ierror.ErrUnauthenticated) || errors.Is(err, ierror.ErrStaleClient) { + c.mtx.Lock() + c.session.Reset() + c.mtx.Unlock() + return nil, nil + } + if err != nil { + return nil, err + } + bound, err := vsr.DecodeLoginRegister(response) + if err != nil { + return nil, err + } + c.mtx.Lock() + defer c.mtx.Unlock() + if bound.Session != identity.session || c.session.ClientID() != identity.client || c.session.SessionID() != identity.session { + return nil, ierror.ErrSessionMismatch + } + c.sessionState = iggcon.SessionStateAuthenticated + c.sessionUserID = bound.UserID + c.loggedOut = false + c.publishPollSession() + return &iggcon.IdentityInfo{UserId: bound.UserID}, nil +} + // endBoundSession logs out a live session before a re-login, so the server // drops its client-table entry instead of leaving it to be fenced. // diff --git a/foreign/go/errors/errors.yaml b/foreign/go/errors/errors.yaml index cd1261675..7a6b588d2 100644 --- a/foreign/go/errors/errors.yaml +++ b/foreign/go/errors/errors.yaml @@ -1311,3 +1311,6 @@ type: uint64 - name: Bound type: uint64 +- name: RequestIdExhausted + code: 14005 + format: "vsr request id exhausted" diff --git a/foreign/go/errors/errors_gen.go b/foreign/go/errors/errors_gen.go index 413aececf..ce151e65e 100644 --- a/foreign/go/errors/errors_gen.go +++ b/foreign/go/errors/errors_gen.go @@ -2690,6 +2690,15 @@ func (e SessionMismatch) Is(target error) bool { return ok } +type RequestIdExhausted struct{} + +func (e RequestIdExhausted) Error() string { return "vsr request id exhausted" } +func (e RequestIdExhausted) Code() Code { return 14005 } +func (e RequestIdExhausted) Is(target error) bool { + _, ok := target.(RequestIdExhausted) + return ok +} + var ( ErrError = Error{} ErrInvalidConfiguration = InvalidConfiguration{} @@ -2933,6 +2942,7 @@ var ( ErrInvalidSession = InvalidSession{} ErrIncompatibleProtocolVersion = IncompatibleProtocolVersion{} ErrSessionMismatch = SessionMismatch{} + ErrRequestIdExhausted = RequestIdExhausted{} ) type Code uint32 @@ -3180,6 +3190,7 @@ const ( InvalidSessionCode Code = 14001 IncompatibleProtocolVersionCode Code = 14003 SessionMismatchCode Code = 14004 + RequestIdExhaustedCode Code = 14005 ) func (c Code) String() string { @@ -3668,6 +3679,8 @@ func (c Code) String() string { return "IncompatibleProtocolVersion" case SessionMismatchCode: return "SessionMismatch" + case RequestIdExhaustedCode: + return "RequestIdExhausted" default: return "Unknown error code" } @@ -4159,6 +4172,8 @@ func FromCode(code Code) IggyError { return ErrIncompatibleProtocolVersion case SessionMismatchCode: return ErrSessionMismatch + case RequestIdExhaustedCode: + return ErrRequestIdExhausted default: return ErrError } diff --git a/foreign/go/internal/vsr/session.go b/foreign/go/internal/vsr/session.go index 8e149a6e8..58aed1298 100644 --- a/foreign/go/internal/vsr/session.go +++ b/foreign/go/internal/vsr/session.go @@ -142,9 +142,8 @@ func (s *Session) CurrentRequestID() uint64 { return s.requestCounter } -// Reset drops the session and mints a fresh client identifier. A disconnect -// invalidates the server-side fence, so the next Register must arrive under a -// new identity. +// Reset drops the session and mints a fresh client identifier after logout or +// a terminal bind refusal. Transport reconnects retain the logical identity. func (s *Session) Reset() { s.reset() } diff --git a/foreign/java/README.md b/foreign/java/README.md index f7f5d7a48..cec81a501 100644 --- a/foreign/java/README.md +++ b/foreign/java/README.md @@ -85,6 +85,12 @@ The source checkout uses protocol 0.11.1. Build it with a matching server. session's private bind secret. Auxiliary poll connections reuse that session instead of logging in independently. +Transport loss retains the logical identity and reconnect uses `BindSession`. +A definitive stale or unauthenticated bind response permits a fresh Register; +transient responses retain the identity. The bind proof is an independent +session credential: password/PAT changes or expiry do not end an established +session. Logout, lease expiry and user deactivation end session access. + ## Quick Start `Iggy.tcpClientBuilder()` handles the routing and the session for you. Over TCP and TCP/TLS, the client keeps the consumer group membership on the coordinator. It polls each partition through a separate connection to the primary of that partition. diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java index fc71be76d..7c3bbd803 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java @@ -103,6 +103,10 @@ public class AsyncTcpConnection { // The pool holds one channel, and one channel lives on one loop. static final int DEFAULT_IO_THREADS = 1; + private static final int STALE_CLIENT = 30; + private static final int UNAUTHENTICATED = 40; + private static final int UNAUTHORIZED = 41; + private static final int REGISTER_RESPONSE_MIN_BYTES = 17; private static final Logger log = LoggerFactory.getLogger(AsyncTcpConnection.class); private static final Duration DEFAULT_CONNECTION_TIMEOUT = Duration.ofMillis(3000); // A missing reply must not hold the single VSR-pinned channel forever. @@ -723,6 +727,36 @@ public class AsyncTcpConnection { private CompletableFuture<ByteBuf> sendAuthenticationFrame( Channel channel, ByteBuf payload, int commandCode, long requestDeadlineNanos) { + if (consensusSession.isBound()) { + long expectedSession = consensusSession.boundSession(); + ByteBuf binding = vsrEncoder.bindSession(channel.alloc()); + CompletableFuture<ByteBuf> bindingFuture = new CompletableFuture<>(); + sendFrame( + channel, binding, CommandCode.System.BIND_SESSION.getValue(), bindingFuture, requestDeadlineNanos); + return bindingFuture + .handle((response, error) -> { + if (error == null) { + payload.release(); + if (response.readableBytes() < REGISTER_RESPONSE_MIN_BYTES + || response.getLongLE(response.readerIndex() + Integer.BYTES) != expectedSession) { + response.release(); + throw new IggyNotConnectedException("BindSession returned a different session epoch"); + } + return CompletableFuture.completedFuture(response); + } + IggyServerException serverError = findServerError(error); + if (serverError != null + && (serverError.getRawErrorCode() == UNAUTHENTICATED + || serverError.getRawErrorCode() == STALE_CLIENT)) { + consensusSession.reset(); + onSessionEvicted(serverError.getRawErrorCode()); + return sendAuthenticationFrame(channel, payload, commandCode, requestDeadlineNanos); + } + payload.release(); + return CompletableFuture.<ByteBuf>failedFuture(error); + }) + .thenCompose(Function.identity()); + } CompletableFuture<ByteBuf> loginFuture = new CompletableFuture<>(); sendFrame(channel, payload, commandCode, loginFuture, requestDeadlineNanos); return loginFuture; @@ -825,7 +859,8 @@ public class AsyncTcpConnection { requestDeadlineNanos, deadlineNanos, notAcceptedDeadlineNanos, - commandCode); + commandCode, + false); } catch (RuntimeException e) { responseFuture.completeExceptionally(e); } finally { @@ -847,7 +882,8 @@ public class AsyncTcpConnection { long requestDeadlineNanos, long deadlineNanos, long notAcceptedDeadlineNanos, - int commandCode) { + int commandCode, + boolean uncertain) { if (requestDeadlineNanos - System.nanoTime() <= 0) { IggyTimeoutException timeout = responseTimeout(commandCode); handler.closeChannel(channel, timeout); @@ -873,7 +909,9 @@ public class AsyncTcpConnection { } }); attempt.whenComplete((response, error) -> { - if (shouldRetryTransient(commandCode, error, deadlineNanos, notAcceptedDeadlineNanos) + IggyServerException serverError = findServerError(error); + boolean nextUncertain = uncertain || isUncertainOutcome(serverError); + if (shouldRetryTransient(commandCode, error, deadlineNanos, notAcceptedDeadlineNanos, nextUncertain) && channel.isActive()) { try { channel.eventLoop() @@ -886,7 +924,8 @@ public class AsyncTcpConnection { requestDeadlineNanos, deadlineNanos, notAcceptedDeadlineNanos, - commandCode), + commandCode, + nextUncertain), TRANSIENT_RETRY_INTERVAL_MS, TimeUnit.MILLISECONDS); return; @@ -896,13 +935,32 @@ public class AsyncTcpConnection { } frame.release(); if (error != null) { - responseFuture.completeExceptionally(error); + responseFuture.completeExceptionally(preserveUncertainOutcome(error, serverError, nextUncertain)); } else { responseFuture.complete(response); } }); } + private static boolean isUncertainOutcome(IggyServerException error) { + return error != null && error.getRawErrorCode() == TRANSIENT_NOT_COMMITTED; + } + + private static Throwable preserveUncertainOutcome( + Throwable error, IggyServerException serverError, boolean uncertain) { + if (uncertain && serverError != null && isUnadmittedRefusal(serverError.getRawErrorCode())) { + return IggyServerException.fromTcpResponse(TRANSIENT_NOT_COMMITTED, new byte[0]); + } + return error; + } + + private static boolean isUnadmittedRefusal(long code) { + return code == TRANSIENT_NOT_ACCEPTED + || code == UNAUTHORIZED + || code == UNAUTHENTICATED + || code == STALE_CLIENT; + } + private static IggyTimeoutException responseTimeout(int commandCode) { return new IggyTimeoutException("Timed out waiting for a response to command code " + commandCode); } @@ -961,11 +1019,12 @@ public class AsyncTcpConnection { } private static boolean shouldRetryTransient( - int commandCode, Throwable error, long deadlineNanos, long notAcceptedDeadlineNanos) { + int commandCode, Throwable error, long deadlineNanos, long notAcceptedDeadlineNanos, boolean uncertain) { if (isPollRoutingCode(commandCode) || !(error instanceof IggyServerException serverError)) { return false; } - if (serverError.getRawErrorCode() == TRANSIENT_NOT_COMMITTED) { + if (serverError.getRawErrorCode() == TRANSIENT_NOT_COMMITTED + || (uncertain && serverError.getRawErrorCode() == TRANSIENT_NOT_ACCEPTED)) { return System.nanoTime() < deadlineNanos; } if (serverError.getRawErrorCode() == TRANSIENT_NOT_ACCEPTED) { @@ -996,9 +1055,8 @@ public class AsyncTcpConnection { /** * A server-side eviction unbinds the transport session and closes its - * channel. Bumping the generation makes the replacement channel re-run - * login and Register. The fresh session invalidates cached routing state - * such as consumer-group assignments. + * channel. Bumping the generation makes the replacement channel bind the + * retained session; only a definitive bind refusal starts a new Register. * * The reason travels to the listener so it can drop what belonged to the * evicted session and log what happened. The session itself is kept diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java index 00f22a1a9..1dfd54f32 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java @@ -45,7 +45,6 @@ public final class ConsensusSession { private Long session; private long requestCounter = 1; private long correlationCounter = 1; - private boolean registerConsumed; private long generation; private long metadataWatermark; private boolean shared; @@ -55,27 +54,13 @@ public final class ConsensusSession { } /** - * Arms a {@code Register}: on re-login (or a consumed one-shot register) - * the whole identity re-arms with a fresh client id so the server sees a - * brand-new registration. Returns the request id a Register carries, - * which is always zero. + * Returns the zero request id of Register without changing its identity. + * Repeating a registration after a lost reply must resolve the same epoch. * - * <p>The request counter is deliberately not rewound. This SDK multiplexes - * a single pinned channel and correlates replies by (operation, request - * id), so a send still in flight when a re-login re-arms would share its - * key with the first send of the new session: the correlation map would - * refuse the second one and a late reply for the first could be handed to - * it. A re-arm registers a fresh client id, which the server admits at - * watermark zero and which accepts any id above it, so carrying the - * counter forward costs nothing on the wire. + * <p>The request counter survives explicit resets too, so late replies + * cannot collide with pending requests from a replacement identity. */ synchronized long beginRegister() { - if (registerConsumed || session != null) { - generation++; - regenerateClientId(); - session = null; - } - registerConsumed = true; return 0; } @@ -84,6 +69,9 @@ public final class ConsensusSession { if (sessionEpoch == 0) { throw new IllegalStateException("Register reply carried a zero session epoch"); } + if (session != null && session != sessionEpoch) { + throw new IllegalStateException("Register reply changed the bound session epoch"); + } this.session = sessionEpoch; generation++; } @@ -126,20 +114,22 @@ public final class ConsensusSession { return session == null ? 0 : session; } - synchronized long boundSession() { + public synchronized long boundSession() { if (session == null) { throw new IggyNotConnectedException("Not authenticated, call login first"); } return session; } - synchronized boolean isBound() { + public synchronized boolean isBound() { return session != null; } - /** Clears the bound epoch (logout / eviction); next login re-registers. */ - synchronized void reset() { + /** Ends the local identity after explicit logout or a refused session bind. */ + public synchronized void reset() { session = null; + metadataWatermark = 0; + regenerateClientId(); generation++; } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java index 9962a3636..1163a53b4 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java @@ -48,6 +48,18 @@ public final class VsrRequestEncoder { return VsrLoginCodec.bindSession(alloc, clientLow, clientHigh, session, watermark, bindSecret); } + public ByteBuf bindSession(ByteBufAllocator alloc) { + synchronized (session) { + return bindSession( + alloc, + session.clientIdLow(), + session.clientIdHigh(), + session.boundSession(), + session.metadataWatermark(), + session.bindSecret()); + } + } + /** * Builds the full request frame. The caller keeps ownership of * {@code payload}; its reader index is not advanced. diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java index 65e77149e..c83c117de 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java @@ -168,7 +168,6 @@ public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> { private void handleEviction(ChannelHandlerContext ctx, ByteBuf frame) { IggyServerException error = VsrHeaders.evictionToException(frame); - session.reset(); try { // The reason travels with the notification so the listener can drop // what belonged to the evicted session and say which eviction it diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java index d72c29258..752402570 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java @@ -127,6 +127,7 @@ public enum IggyErrorCode { // VSR protocol errors INCOMPATIBLE_PROTOCOL_VERSION(14003), SESSION_MISMATCH(14004), + REQUEST_ID_EXHAUSTED(14005), // Unknown error code UNKNOWN(-1); diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java index be731adf8..8f49eb034 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientTransientFailoverTest.java @@ -23,6 +23,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.apache.iggy.consumergroup.Consumer; import org.apache.iggy.exception.IggyClientException; +import org.apache.iggy.exception.IggyConnectionException; import org.apache.iggy.exception.IggyErrorCode; import org.apache.iggy.exception.IggyServerException; import org.apache.iggy.identifier.ConsumerId; @@ -99,6 +100,122 @@ class AsyncIggyTcpClientTransientFailoverTest { private static final Duration REFUSAL_HOLD_WINDOW = Duration.ofSeconds(1); private static final Duration REFUSAL_HELD_AFTER = TEST_POLL_TIMEOUT.minus(REFUSAL_HOLD_WINDOW); + @Test + void shouldRetainUncertainMutationOutcomeAfterLaterRefusal() throws Exception { + for (int refusal : new int[] { + IggyErrorCode.STALE_CLIENT.getCode(), + IggyErrorCode.UNAUTHENTICATED.getCode(), + IggyErrorCode.UNAUTHORIZED.getCode(), + TRANSIENT_NOT_ACCEPTED + }) { + InetAddress loopback = InetAddress.getLoopbackAddress(); + try (ServerSocket socket = new ServerSocket(0, 1, loopback)) { + AtomicInteger attempts = new AtomicInteger(); + List<Long> requestIds = new CopyOnWriteArrayList<>(); + CompletableFuture<Void> server = serve(socket, request -> { + if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) { + return Response.success(OPERATION_NON_REPLICATED, singleNodeMetadata(socket.getLocalPort())); + } + if (request.operation() == OPERATION_REGISTER) { + return Response.success(OPERATION_REGISTER, registerBody(1)); + } + if (request.operation() == OPERATION_CREATE_STREAM) { + requestIds.add(request.requestId()); + int attempt = attempts.incrementAndGet(); + return Response.error( + OPERATION_CREATE_STREAM, + attempt == 1 + ? TRANSIENT_NOT_COMMITTED + : attempt == 2 ? refusal : IggyErrorCode.UNAUTHORIZED.getCode()); + } + throw new IllegalStateException("Unexpected request: " + request); + }); + AsyncIggyTcpClient client = client(socket); + try { + client.connect().get(5, TimeUnit.SECONDS); + client.users().login("iggy", "iggy").get(5, TimeUnit.SECONDS); + assertThatThrownBy(() -> client.sendBinaryRequest(CREATE_STREAM_CODE, new byte[0]) + .get(5, TimeUnit.SECONDS)) + .hasCauseInstanceOf(IggyServerException.class) + .satisfies(error -> assertThat(((IggyServerException) error.getCause()).getRawErrorCode()) + .isEqualTo(TRANSIENT_NOT_COMMITTED)); + assertThat(requestIds).hasSizeGreaterThanOrEqualTo(2).containsOnly(requestIds.get(0)); + } finally { + client.close().get(5, TimeUnit.SECONDS); + } + server.get(5, TimeUnit.SECONDS); + } + } + } + + @Test + void shouldBindRetainedSessionAfterTransportLoss() throws Exception { + assertReconnectBinding(false); + } + + @Test + void shouldRegisterFreshOnlyAfterRetainedSessionBindIsRefused() throws Exception { + assertReconnectBinding(true); + } + + private static void assertReconnectBinding(boolean refuseBinding) throws Exception { + InetAddress loopback = InetAddress.getLoopbackAddress(); + try (ServerSocket socket = new ServerSocket(0, 2, loopback)) { + List<Request> registrations = new CopyOnWriteArrayList<>(); + List<Request> bindings = new CopyOnWriteArrayList<>(); + AtomicBoolean disconnectNextRead = new AtomicBoolean(); + CompletableFuture<Void> server = serve(socket, 2, request -> { + if (request.operation() == OPERATION_REGISTER) { + registrations.add(request); + return Response.success(OPERATION_REGISTER, registerBody(registrations.size())); + } + if (request.is(BIND_SESSION_CODE, OPERATION_NON_REPLICATED)) { + bindings.add(request); + ByteBuffer identity = bindIdentity(request.body()); + assertThat(identity.getLong(0)) + .isEqualTo(registrations.get(0).clientLow()); + assertThat(identity.getLong(8)) + .isEqualTo(registrations.get(0).clientHigh()); + assertThat(identity.getLong(16)).isEqualTo(1); + ByteBuf reply = registerBody(1); + reply.skipBytes(Integer.BYTES); + if (refuseBinding) { + reply.release(); + return Response.error(OPERATION_NON_REPLICATED, IggyErrorCode.UNAUTHENTICATED.getCode()); + } + return Response.success(OPERATION_NON_REPLICATED, reply); + } + if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) { + return disconnectNextRead.compareAndSet(true, false) + ? Response.disconnect() + : Response.success(OPERATION_NON_REPLICATED, singleNodeMetadata(socket.getLocalPort())); + } + throw new IllegalStateException("Unexpected request: " + request); + }); + AsyncIggyTcpClient client = client(socket); + try { + client.connect().get(5, TimeUnit.SECONDS); + client.users().login("iggy", "iggy").get(5, TimeUnit.SECONDS); + disconnectNextRead.set(true); + assertThatThrownBy(() -> client.sendBinaryRequest(GET_CLUSTER_METADATA_CODE, new byte[0]) + .get(5, TimeUnit.SECONDS)) + .hasCauseInstanceOf(IggyConnectionException.class); + client.sendBinaryRequest(GET_CLUSTER_METADATA_CODE, new byte[0]).get(5, TimeUnit.SECONDS); + assertThat(bindings).hasSize(1); + assertThat(registrations).hasSize(refuseBinding ? 2 : 1); + if (refuseBinding) { + Request first = registrations.get(0); + Request second = registrations.get(registrations.size() - 1); + assertThat(second.clientLow() != first.clientLow() || second.clientHigh() != first.clientHigh()) + .isTrue(); + } + } finally { + client.close().get(5, TimeUnit.SECONDS); + } + server.get(5, TimeUnit.SECONDS); + } + } + @Test void shouldCancelAnAutoCommitWaitingForTopologyWithoutClosingCoordinator() throws Exception { InetAddress loopback = InetAddress.getLoopbackAddress(); @@ -321,6 +438,9 @@ class AsyncIggyTcpClientTransientFailoverTest { throw new IllegalStateException("Unexpected old-leader request: " + request); }); CompletableFuture<Void> newLeader = serve(newSocket, 2, request -> { + if (request.is(BIND_SESSION_CODE, OPERATION_NON_REPLICATED)) { + return Response.error(OPERATION_NON_REPLICATED, IggyErrorCode.UNAUTHENTICATED.getCode()); + } if (request.operation() == OPERATION_REGISTER) { logins.add(request.bodyAsText()); return Response.success(OPERATION_REGISTER, registerBody(logins.size() + 1)); @@ -1198,6 +1318,9 @@ class AsyncIggyTcpClientTransientFailoverTest { AtomicInteger registrations = new AtomicInteger(); AtomicInteger mutations = new AtomicInteger(); CompletableFuture<Void> server = serve(serverSocket, 2, request -> { + if (request.is(BIND_SESSION_CODE, OPERATION_NON_REPLICATED)) { + return Response.error(OPERATION_NON_REPLICATED, IggyErrorCode.UNAUTHENTICATED.getCode()); + } if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) { return Response.success(OPERATION_NON_REPLICATED, singleNodeMetadata(serverSocket.getLocalPort())); } @@ -1310,6 +1433,9 @@ class AsyncIggyTcpClientTransientFailoverTest { List<String> registeredLogins = new CopyOnWriteArrayList<>(); AtomicBoolean evict = new AtomicBoolean(true); CompletableFuture<Void> server = serve(serverSocket, 6, request -> { + if (request.is(BIND_SESSION_CODE, OPERATION_NON_REPLICATED)) { + return Response.error(OPERATION_NON_REPLICATED, IggyErrorCode.UNAUTHENTICATED.getCode()); + } if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) { return Response.success(OPERATION_NON_REPLICATED, singleNodeMetadata(serverSocket.getLocalPort())); } @@ -1379,6 +1505,9 @@ class AsyncIggyTcpClientTransientFailoverTest { List<String> registeredLogins = new CopyOnWriteArrayList<>(); AtomicBoolean evict = new AtomicBoolean(true); CompletableFuture<Void> server = serve(serverSocket, 6, request -> { + if (request.is(BIND_SESSION_CODE, OPERATION_NON_REPLICATED)) { + return Response.error(OPERATION_NON_REPLICATED, IggyErrorCode.UNAUTHENTICATED.getCode()); + } if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) { return Response.success(OPERATION_NON_REPLICATED, singleNodeMetadata(serverSocket.getLocalPort())); } diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java index 0d1a9d7bd..e86055b75 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java @@ -169,12 +169,13 @@ class VsrRequestEncoderTest { } @Test - void shouldReArmWithFreshClientIdOnSecondLogin() { + void shouldRetainClientIdAndSecretWhenRegistrationIsRepeated() { ByteBuf firstLogin = loginUserPayload(); encoder.encode(alloc, LOGIN_USER_CODE, firstLogin).release(); firstLogin.release(); long firstLow = session.clientIdLow(); long firstHigh = session.clientIdHigh(); + byte[] firstSecret = session.bindSecret(); session.bind(7); long firstGeneration = session.generation(); @@ -182,13 +183,11 @@ class VsrRequestEncoderTest { encoder.encode(alloc, LOGIN_USER_CODE, secondLogin).release(); secondLogin.release(); - assertThat(session.isBound()).isFalse(); - assertThat(session.clientIdLow() != firstLow || session.clientIdHigh() != firstHigh) - .isTrue(); - assertThat(session.generation()).isGreaterThan(firstGeneration); - session.bind(7); - assertThatThrownBy(() -> session.bindSecret(firstLow, firstHigh, 7)) - .isInstanceOf(IggyNotConnectedException.class); + assertThat(session.isBound()).isTrue(); + assertThat(session.clientIdLow()).isEqualTo(firstLow); + assertThat(session.clientIdHigh()).isEqualTo(firstHigh); + assertThat(session.bindSecret()).isEqualTo(firstSecret); + assertThat(session.generation()).isEqualTo(firstGeneration); } @Test diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java index 3159d3eac..6601ce0a1 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java @@ -174,7 +174,7 @@ class VsrResponseHandlerTest { } @Test - void shouldMapEvictionReasonAndResetSession() { + void shouldMapEvictionReasonAndRetainSessionUntilBindIsRefused() { session.beginRegister(); session.bind(42); CompletableFuture<ByteBuf> future = enqueue(VsrOperation.NON_REPLICATED, 7); @@ -184,7 +184,7 @@ class VsrResponseHandlerTest { channel.writeInbound(frame); assertThat(rawErrorCode(future)).isEqualTo(42); - assertThat(session.isBound()).isFalse(); + assertThat(session.isBound()).isTrue(); assertThat(evictions).hasValue(1); // The reason reaches the listener, which has to tell an eviction the // server decided on from a transport-shaped one. @@ -199,7 +199,7 @@ class VsrResponseHandlerTest { channel.writeInbound(frame); - assertThat(session.isBound()).isFalse(); + assertThat(session.isBound()).isTrue(); assertThat(evictions).hasValue(1); assertThat(lastEvictionReason).hasValue(VsrHeaders.ERROR_STALE_CLIENT); assertThat(frame.refCnt()).isZero(); @@ -288,6 +288,7 @@ class VsrResponseHandlerTest { CompletableFuture<ByteBuf> inFlight = new CompletableFuture<>(); long beforeLoginId = registerEncodedSend(encoder, inFlight); + session.reset(); ByteBuf loginPayload = loginUserPayload(); encoder.encode(channel.alloc(), LOGIN_USER_CODE, loginPayload).release(); loginPayload.release(); diff --git a/foreign/node/README.md b/foreign/node/README.md index f302b1adf..12776f7d6 100644 --- a/foreign/node/README.md +++ b/foreign/node/README.md @@ -65,6 +65,10 @@ a confirmed ended session requires a new registration. Logout forgets the local session. Transient not-committed responses retry the exact encoded request within one bounded deadline. A disconnected mutation is never replayed under a new session. +An established session has its own lifetime. Password or PAT changes and PAT +expiry do not end it. Logout, session lease expiry, or user deactivation end the +session and prevent its bind proof from restoring it. + The client pings every `heartbeatInterval` milliseconds, 5000 by default, which keeps an idle session alive when the server's `[heartbeat]` eviction is enabled. `heartbeatInterval` also accepts a duration expression such as `"10s"` or diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts index 9660b5228..125226aa9 100644 --- a/foreign/node/src/client/client.socket.test.ts +++ b/foreign/node/src/client/client.socket.test.ts @@ -945,7 +945,7 @@ describe('primary auto-commit polling', () => { }); describe('VSR client socket', () => { - for (const refusal of [0, 30, 58]) { + for (const refusal of [0, 30, 40, 57, 58]) { it(`resumes or replaces a lost connection only after bind verdict ${refusal}`, async () => { let registrations = 0; let sends = 0; @@ -973,25 +973,26 @@ describe('VSR client socket', () => { try { await client.authenticate(config.credentials); await assert.rejects(client.sendCommand(COMMAND_CODE.SendMessages, Buffer.from('lost-reply'))); - if (refusal === 58) { + if (refusal === 57 || refusal === 58) { await assert.rejects(client.authenticate(config.credentials), (error: unknown) => - error instanceof ResponseError && error.errorCode === 58); + error instanceof ResponseError && error.errorCode === refusal); assert.equal(registrations, 1, 'lagging metadata must not replace the logical session'); bindStatus = 0; } await client.authenticate(config.credentials); assert.equal(sends, 1, 'authentication must not replay the uncertain mutation'); - assert.equal(registrations, refusal === 30 ? 2 : 1); + const terminal = refusal === 30 || refusal === 40; + assert.equal(registrations, terminal ? 2 : 1); const binding = server.frames.find((frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === COMMAND_CODE.BindSession)!; assertSharedBinding(server.frames[0], binding); await client.sendCommand(COMMAND_CODE.SendMessages, Buffer.from('explicit-new-write')); const writes = server.frames.filter((frame) => frame.readUInt8(REQUEST_OFFSET.operation) === Operation.SendMessages); assert.equal(writes.length, 2); - assert.equal(writes[1].readBigUInt64LE(REQUEST_OFFSET.session), refusal === 30 ? TEST_SESSION + 1n : TEST_SESSION); - assert.equal(writes[1].readBigUInt64LE(REQUEST_OFFSET.request), refusal === 30 ? 1n : 2n); + assert.equal(writes[1].readBigUInt64LE(REQUEST_OFFSET.session), terminal ? TEST_SESSION + 1n : TEST_SESSION); + assert.equal(writes[1].readBigUInt64LE(REQUEST_OFFSET.request), terminal ? 1n : 2n); assert.equal(writes[0].subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16) - .equals(writes[1].subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16)), refusal !== 30); + .equals(writes[1].subarray(REQUEST_OFFSET.client, REQUEST_OFFSET.client + 16)), !terminal); } finally { client.destroy(); await server.close(); @@ -2251,7 +2252,8 @@ describe('VSR client socket', () => { } ); - it('keeps an uncertain write uncertain after a later refusal', async () => { + for (const refusal of [58, 41, 40, 30]) { + it(`keeps an uncertain write uncertain after a later refusal ${refusal}`, async () => { const EXPIRED_REQUEST_TIME = 30_001; let now = 0; let attempts = 0; @@ -2260,7 +2262,7 @@ describe('VSR client socket', () => { attempts += 1; now = attempts === 1 ? 1 : EXPIRED_REQUEST_TIME; socket.write(replyFrame( - Operation.SendMessages, Buffer.alloc(0), attempts === 1 ? 57 : 58 + Operation.SendMessages, Buffer.alloc(0), attempts === 1 ? 57 : refusal )); return; } @@ -2293,6 +2295,7 @@ describe('VSR client socket', () => { await server.close(); } }); + } it('shares token authentication between concurrent callers', async () => { const server = await startVsrServer( diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts index d9c16ce0d..e60b83474 100644 --- a/foreign/node/src/client/client.socket.ts +++ b/foreign/node/src/client/client.socket.ts @@ -57,6 +57,7 @@ const MAX_LEADER_REDIRECTS = 3; const TRANSIENT_NOT_COMMITTED = 57; const TRANSIENT_NOT_ACCEPTED = 58; const UNAUTHENTICATED = 40; +const UNAUTHORIZED = 41; const STALE_CLIENT = 30; const FEATURE_UNAVAILABLE = 5; const MAX_POLL_ROUTES = 4096; @@ -938,6 +939,13 @@ export class CommandResponseStream extends EventEmitter { parsed = decodeVsrResponse(response, command); break; } catch (error) { + if (error instanceof ResponseError && + lastTransientError?.errorCode === TRANSIENT_NOT_COMMITTED && + (error.errorCode === TRANSIENT_NOT_ACCEPTED || + error.errorCode === UNAUTHENTICATED || + error.errorCode === STALE_CLIENT || + error.errorCode === UNAUTHORIZED)) + error = responseError(command, TRANSIENT_NOT_COMMITTED); if (!(error instanceof ResponseError) || !isTransientVsrError(error.errorCode)) throw error; diff --git a/foreign/node/src/wire/error.code.ts b/foreign/node/src/wire/error.code.ts index 880dc1ff7..d05fc9d8e 100644 --- a/foreign/node/src/wire/error.code.ts +++ b/foreign/node/src/wire/error.code.ts @@ -271,6 +271,7 @@ export const translateErrorCode = (code: number): string => { case '14001': return "VSR session value {0} is invalid (must be non-zero)"; case '14003': return "Incompatible binary protocol version: client {}, server accepts [{}, {}]"; case '14004': return "VSR session mismatch: requested {0}, server bound {1}"; + case '14005': return "VSR request id exhausted"; default: return 'error'; } diff --git a/foreign/python/README.md b/foreign/python/README.md index 63b2b61f5..8847e05e4 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -179,6 +179,11 @@ asyncio.run(main()) transport. `examples/python/getting-started/producer.py` shows each swap in context. +Established binary sessions survive transport loss. Password/PAT changes and +PAT expiry do not end these sessions; logout, session lease expiry and account +deactivation do. Reconnection binds the retained session before attempting a +fresh login. + `HttpConfig` differs from TCP in two ways. There is no reconnection policy and no `AutoLogin`: `connect()` does not dial over HTTP, but it does start the heartbeat that `heartbeat_interval` configures, so call it and then @@ -272,15 +277,18 @@ exchange. Persisted sends also provide crash-safe retry receipts; weaker topic policies can lose data and receipts on a crash. HTTP NoAck confirms dispatch only. An application resend creates a new request and can duplicate messages. -Producer retries start a new attempt only after an explicit -`TransientNotAccepted` refusal. Missing destinations and other terminal errors +Producer retries start a new attempt after an explicit `TransientNotAccepted` +refusal or a connection failure before submission (`NotConnected` or +`CannotEstablishConnection`). Missing destinations and other terminal errors stop immediately. `send_retries` counts retries after the initial attempt. The first permitted retry runs immediately, and `send_retry_interval` delays only later retries. Set `send_retries` to `None` or `0` to disable producer retries. Set `send_retry_interval` to `None` to run permitted retries without a delay. A zero interval raises `ValueError`. -Transport retries are separate from producer retries. For example, the default +HTTP send errors stop producer retries immediately, so `send_retries` does not +retry a failed HTTP send. Transport retries are separate from producer retries. +For example, the default `HttpConfig(retries=3)` gives each producer attempt up to four HTTP attempts. A failed direct send raises `ProducerSendError`, which is a `RuntimeError`
