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 07ed42fa8a087df2e0d43593ad706e019f378dc5 Author: Hubert Gruszecki <[email protected]> AuthorDate: Mon Oct 5 16:16:02 2026 +0200 fix(consensus): preserve revoked sessions through recovery and replay Revoked sessions could invalidate snapshots, skip group cleanup during state transfer, and retain HTTP access through unexpired JWTs. Preserve retry receipts, replay against historical session state, and reject inactive principals before serving reads or attaching write sessions. Handle SDK lock poisoning and request exhaustion without losing session identity. Add regressions and correct the reviewed protocol and storage documentation. --- core/consensus/src/client_table.rs | 128 +++++- core/consensus/src/metadata_helpers.rs | 16 +- .../tests/cluster/client_table_adversarial.rs | 8 +- .../tests/cluster/client_table_restart.rs | 12 +- core/integration/tests/cluster/partition_dedup.rs | 16 +- .../tests/server/descriptor_exhaustion_vsr.rs | 15 +- core/integration/tests/server/http_rbac.rs | 91 ++++ core/message_bus/src/replica/handshake.rs | 48 +- core/metadata/src/impls/metadata.rs | 509 +++++++++++++-------- core/metadata/src/stm/snapshot.rs | 10 +- core/partitions/src/iggy_partition.rs | 29 +- core/sdk/src/quic/quic_client.rs | 73 ++- core/sdk/src/tcp/tcp_client.rs | 73 ++- core/sdk/src/websocket/websocket_client.rs | 67 ++- core/server/src/http/extractor.rs | 1 + core/server/src/http/state.rs | 19 +- core/shard/src/lib.rs | 22 - foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs | 5 + foreign/csharp/Iggy_SDK/Vsr/VsrError.cs | 1 + .../VsrTests/ConsensusSessionTests.cs | 42 ++ .../iggy/client/async/tcp/vsr/VsrLoginCodec.java | 5 +- foreign/swift/Tools/golden-vectors/src/main.rs | 1 - 22 files changed, 875 insertions(+), 316 deletions(-) diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index dabf85f73..ba5f94a5b 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -1215,6 +1215,13 @@ impl ClientTable { return Ok(cached.clone()); } let cached = CachedReply::from(reply); + // A lower request does not acknowledge the watermark's result. + if let Some(pinned) = entry.ring.back() + && header.request < entry.watermark + { + entry.latest_commit = pinned.header().commit; + return Ok(cached); + } entry.ring.clear(); entry.latest_commit = cached.header().commit; entry.ring.push_back(cached.clone()); @@ -1826,8 +1833,12 @@ impl ClientEntry { if header.request != self.watermark || header.commit != self.latest_commit || self.ended_op.is_some_and(|ended| { - ended != self.latest_commit - || header.operation != iggy_binary_protocol::Operation::Logout + ended < self.latest_commit + || (ended == self.latest_commit + && !matches!( + header.operation, + Operation::Logout | Operation::UpdateUser | Operation::DeleteUser + )) }) { return false; @@ -2174,6 +2185,7 @@ mod tests { "future epoch", "watermark", "end marker", + "regressing end marker", "foreign receipt", ] { let mut snapshot = valid.clone(); @@ -2183,6 +2195,7 @@ mod tests { "future epoch" => entry.epoch = 31, "watermark" => entry.watermark = 6, "end marker" => entry.ended_op = Some(30), + "regressing end marker" => entry.ended_op = Some(29), "foreign receipt" => { entry.reply = make_reply_with_checksum(2, 5, 30, 0xbeef) .as_slice() @@ -2201,6 +2214,65 @@ mod tests { assert!(ClientTable::from_snapshot(valid).is_ok()); } + #[test] + fn given_revoked_user_when_recovered_should_preserve_tombstones_and_receipts() { + const CLIENT: u128 = 7; + const EPOCH: u64 = 10; + const END: u64 = 40; + let secret = [0x5a; BIND_SECRET_BYTES]; + for self_revocation in [ + None, + Some(Operation::UpdateUser), + Some(Operation::DeleteUser), + ] { + let mut table = ClientTable::new(1); + table + .commit_register( + CLIENT, + TEST_USER_ID, + bind_verifier(CLIENT, TEST_USER_ID, &secret), + make_register_reply(CLIENT, EPOCH), + ) + .unwrap(); + let attachment = table.attach_session(CLIENT, EPOCH, TEST_USER_ID).unwrap(); + let mut receipt = make_reply_for(CLIENT, 5, END - 1); + assert_eq!( + table.commit_reply(CLIENT, TEST_USER_ID, receipt.clone()), + CommitReply::Cached + ); + let ended = table.end_user_sessions(TEST_USER_ID, END); + assert_eq!(ended.len(), 1); + if let Some(operation) = self_revocation { + receipt = make_reply_for(CLIENT, 6, END).transmute_header(|old, header| { + *header = old; + header.operation = operation; + }); + assert_eq!( + table.commit_reply(CLIENT, TEST_USER_ID, receipt.clone()), + CommitReply::Cached + ); + } + assert!(!attachment.is_valid()); + for mut restored in [ + ClientTable::from_snapshot(table.to_snapshot()).unwrap(), + ClientTable::decode(&table.encode()).unwrap(), + ] { + assert_eq!(restored.ended_sessions().collect::<Vec<_>>(), ended); + assert_eq!( + restored.get_reply(CLIENT).unwrap().as_bytes(), + receipt.as_slice() + ); + assert!(restored.bind_session(CLIENT, EPOCH, &secret).is_err()); + assert!(matches!( + restored.check_request(CLIENT, EPOCH, 7, Operation::SendMessages), + RequestStatus::NoSession + )); + assert!(restored.finalize_session(ended[0])); + assert_eq!(restored.count(), 0); + } + } + } + /// Register client 1 (register commit stamped at op 10). Returns /// (table, epoch=1). fn table_with_client() -> (ClientTable, u64) { @@ -2944,6 +3016,58 @@ mod tests { )); } + #[test] + fn given_pinned_receipt_when_lower_request_commits_should_preserve_retry_result() { + const CLIENT: u128 = 7; + const SESSION: u64 = 42; + const PINNED_REQUEST: u64 = 5; + const PINNED_COMMIT: u64 = 100; + const REORDERED_REQUEST: u64 = 3; + let mut table = slice(1); + let pinned = make_reply_with_checksum(CLIENT, PINNED_REQUEST, PINNED_COMMIT, 0xabcd); + table + .commit_partition_reply(TEST_USER_ID, SESSION, pinned.clone()) + .unwrap(); + let reordered = + make_reply_with_checksum(CLIENT, REORDERED_REQUEST, PINNED_COMMIT + 1, 0xef01); + for _ in 0..2 { + let committed = table + .commit_partition_reply(TEST_USER_ID, SESSION, reordered.clone()) + .unwrap(); + assert_eq!(committed.as_bytes(), reordered.as_slice()); + assert_eq!(table.watermarks_sorted()[0].latest_commit, PINNED_COMMIT); + let mut restored = slice(1); + restored + .install_watermarks(1, table.watermarks_sorted()) + .unwrap(); + for retained in [&table, &restored] { + let RequestStatus::Duplicate(cached) = retained.check_partition_request( + CLIENT, + TEST_USER_ID, + SESSION, + PINNED_REQUEST, + Operation::SendMessages, + ) else { + panic!("the higher request lost its pinned receipt"); + }; + assert_eq!(cached.as_bytes(), pinned.as_slice()); + assert!(matches!( + retained.check_partition_request( + CLIENT, + TEST_USER_ID, + SESSION, + REORDERED_REQUEST, + Operation::SendMessages, + ), + RequestStatus::AlreadyApplied { + request: REORDERED_REQUEST, + watermark: PINNED_REQUEST + } + )); + } + } + } + #[test] fn given_missing_session_when_binding_should_distinguish_registration_uncertainty() { const CLIENT: u128 = 7; diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs index b0699de6c..8b1e2e34b 100644 --- a/core/consensus/src/metadata_helpers.rs +++ b/core/consensus/src/metadata_helpers.rs @@ -53,14 +53,10 @@ pub enum PreflightOutcome { /// not-caught-up primary). The caller sends a `TransientNotCommitted` reply /// so the client replays immediately instead of waiting out its read-timeout. NotReady, - /// Absorbed with nothing to send: stale/gap retry or a client-bug newer - /// session. Replaying the same `request_id` cannot help, so stay silent. - Drop, /// Terminal: the request will never be executed and no cached reply /// exists, so the caller answers with this `IggyError` code. Distinct from - /// [`Self::Drop`] in that the client learns immediately instead of waiting - /// out its read timeout, and from [`Self::NotReady`] in that a replay of - /// the same `request_id` cannot change the answer. + /// [`Self::NotReady`] in that replaying the same `request_id` cannot change + /// the answer. Reject(u32), } @@ -147,7 +143,7 @@ where RequestStatus::Duplicate(cached_reply) => { PreflightOutcome::Replay(cached_reply.into_wire_bytes()) } - // Session evicted under capacity pressure. The catch-up gate makes this + // Unknown client id or restarted session. The catch-up gate makes this // replica authoritative for its own committed session state, which is // what an eviction frame reports. RequestStatus::NoSession => PreflightOutcome::Evict(EvictionReason::NoSession), @@ -238,13 +234,11 @@ where } // The wire-ingress plane has no per-request transport context to build a // correlated `TransientNotCommitted` reply (that lives on the in-process - // home-shard path); stay silent here as before. NotReady and Drop both - // mean "do not dispatch"; the difference (explicit retry frame) only - // applies where the request header is in scope. + // home-shard path); stay silent here as before. // `Reject` needs the request header to build a correlated reply, which // only the home-shard path holds; it degrades to silence here for the // same reason NotReady does. - PreflightOutcome::NotReady | PreflightOutcome::Drop | PreflightOutcome::Reject(_) => false, + PreflightOutcome::NotReady | PreflightOutcome::Reject(_) => false, } } diff --git a/core/integration/tests/cluster/client_table_adversarial.rs b/core/integration/tests/cluster/client_table_adversarial.rs index 9b933bc72..ee6fbf44a 100644 --- a/core/integration/tests/cluster/client_table_adversarial.rs +++ b/core/integration/tests/cluster/client_table_adversarial.rs @@ -21,10 +21,10 @@ //! the table's resource edges, the third across the request-id gaps a client //! that also produces leaves behind. //! -//! 1. Capacity: a full table evicts the entry with the oldest commit. Eviction -//! keeps that client's request watermark (and the watermark's reply when the -//! ring still held it), so a client that was merely quiet re-registers and -//! its retry of an already-committed request id is answered, not re-executed. +//! 1. Capacity: a full table refuses new clients without evicting existing +//! entries. A live client's watermark and cached replies survive registry +//! pressure, and matching re-registration preserves its epoch so committed +//! request retries replay instead of executing again. //! 2. Reply retention: an entry keeps `REPLY_RING_CAPACITY` replies whatever //! they weigh and older ones for as long as they fit //! `REPLY_RING_RETENTION_BYTES`, so a retry arriving more commits late than diff --git a/core/integration/tests/cluster/client_table_restart.rs b/core/integration/tests/cluster/client_table_restart.rs index 9b9423e3c..c5339b219 100644 --- a/core/integration/tests/cluster/client_table_restart.rs +++ b/core/integration/tests/cluster/client_table_restart.rs @@ -437,11 +437,9 @@ async fn commit_request_header( /// keep presenting the old identity, round-robin across every node, until one /// commits (or serves the cached reply for) the request. /// -/// The re-login is the resume, and it is a rebind: the ownership-gated -/// Register commits, the recovered entry's fence moves to that register's op, -/// the login reply hands back the NEW epoch. Continuation frames must stamp -/// it -- the pre-restart epoch is a fenced zombie from here on. Watermark and -/// reply ring survive the rebind, which is what the dedup assertion rests on. +/// A matching Register authenticates the replacement connection while +/// preserving the recovered epoch, watermark, and reply ring. Continuation +/// frames keep that epoch so retries still find the original receipts. /// There is deliberately NO way to rebind without credentials -- an unbound /// transport that merely presents `(client, session)` gets the empty-reply /// fail-fast (see `given_unauthenticated_resume_*`). @@ -467,9 +465,7 @@ pub(super) async fn resume_request( sleep(RETRY_PAUSE).await; continue; }; - // Re-authenticate on the fresh connection. The rebind commits a - // Register, so the epoch strictly advances past the pre-restart one - // (op-derived; regression would mean the fence can be replayed into). + // Keep the epoch so continuation retries use the recovered receipt identity. let resumed = match login_on(&mut stream, CLIENT_ID).await { Some(resumed) => { assert!( diff --git a/core/integration/tests/cluster/partition_dedup.rs b/core/integration/tests/cluster/partition_dedup.rs index 7b8c49839..d3b4f909a 100644 --- a/core/integration/tests/cluster/partition_dedup.rs +++ b/core/integration/tests/cluster/partition_dedup.rs @@ -49,7 +49,7 @@ use iggy_binary_protocol::{ }; use integration::harness::TestHarness; use integration::iggy_harness; -use journal::partition_journal::PARTITION_WAL_BLOCK_SIZE; +use journal::partition_journal::{FRONTIER_FILE_NAME, PARTITION_WAL_BLOCK_SIZE}; use secrecy::SecretString; use std::mem::offset_of; use std::net::SocketAddr; @@ -319,6 +319,8 @@ async fn given_reclaimed_wal_when_recovered_quorum_replays_should_return_the_ori const CHECKPOINT_OP: u64 = 2; const MESSAGES_PER_BATCH: u32 = 33; const PAYLOAD_BYTES: usize = 1024 * 1024; + const WAL_FRONTIER_SLOTS: usize = 2; + const WAL_FRONTIER_MAGIC: &[u8; 8] = b"IGGYWAL3"; const WAL_CHECKPOINT_OFFSET: usize = 48; const WAL_HEAD_OFFSET: usize = 72; let observer = harness.root_client_for_node(0).await.unwrap(); @@ -347,12 +349,22 @@ async fn given_reclaimed_wal_when_recovered_quorum_replays_should_return_the_ori .node(node) .data_path() .join("streams/0/topics/0/partitions/0/prepares-1"); - std::fs::read(directory.join("frontier")).is_ok_and(|bytes| { + std::fs::read(directory.join(FRONTIER_FILE_NAME)).is_ok_and(|bytes| { + assert_eq!( + bytes.len(), + WAL_FRONTIER_SLOTS * PARTITION_WAL_BLOCK_SIZE, + "unexpected partition WAL frontier size on node {node}" + ); bytes .as_chunks::<PARTITION_WAL_BLOCK_SIZE>() .0 .iter() .any(|slot| { + assert_eq!( + &slot[..WAL_FRONTIER_MAGIC.len()], + WAL_FRONTIER_MAGIC, + "unsupported partition WAL frontier format on node {node}" + ); u64::from_le_bytes( slot[WAL_CHECKPOINT_OFFSET..WAL_CHECKPOINT_OFFSET + size_of::<u64>()] .try_into() diff --git a/core/integration/tests/server/descriptor_exhaustion_vsr.rs b/core/integration/tests/server/descriptor_exhaustion_vsr.rs index a6c0d49a5..ca7b2ab82 100644 --- a/core/integration/tests/server/descriptor_exhaustion_vsr.rs +++ b/core/integration/tests/server/descriptor_exhaustion_vsr.rs @@ -51,7 +51,9 @@ const WRITE_TIMEOUT: Duration = Duration::from_secs(5); /// that frees up goes to a waiting socket between two writes. const WRITE_PAUSE: Duration = Duration::from_millis(500); const EXIT_TIMEOUT: Duration = Duration::from_secs(60); -/// The head field of a frontier slot in `journal::partition_journal::JournalState`. +const WAL_FRONTIER_SLOTS: usize = 2; +const WAL_FRONTIER_MAGIC: &[u8; 8] = b"IGGYWAL3"; +/// The head field in the checked `IGGYWAL3` frontier slot layout. const WAL_HEAD_OFFSET: usize = 72; /// How a test writes the offset of a new consumer. @@ -137,13 +139,22 @@ async fn stop_on_offset_write_without_descriptor(harness: &mut TestHarness, writ .expect("send messages"); let frontier = partition_frontier(harness.server(), stream.id, topic.id); let before = std::fs::read(&frontier).expect("read the WAL frontier after sending"); - assert_eq!(before.len(), 2 * PARTITION_WAL_BLOCK_SIZE); + assert_eq!( + before.len(), + WAL_FRONTIER_SLOTS * PARTITION_WAL_BLOCK_SIZE, + "unexpected partition WAL frontier size" + ); assert!( before .as_chunks::<PARTITION_WAL_BLOCK_SIZE>() .0 .iter() .all(|slot| { + assert_eq!( + &slot[..WAL_FRONTIER_MAGIC.len()], + WAL_FRONTIER_MAGIC, + "unsupported partition WAL frontier format" + ); u64::from_le_bytes( slot[WAL_HEAD_OFFSET..WAL_HEAD_OFFSET + size_of::<u64>()] .try_into() diff --git a/core/integration/tests/server/http_rbac.rs b/core/integration/tests/server/http_rbac.rs index 1a5728ba3..c8f4ef59d 100644 --- a/core/integration/tests/server/http_rbac.rs +++ b/core/integration/tests/server/http_rbac.rs @@ -49,6 +49,7 @@ const PARTITION_ID: u32 = 0; const CONSUMER_ID: u32 = 1; // `IggyError` discriminants (== the numeric `id` in the JSON error body). +const UNAUTHENTICATED_ID: u64 = 40; const UNAUTHORIZED_ID: u64 = 41; const INVALID_CREDENTIALS_ID: u64 = 42; const CANNOT_DELETE_USER_ID: u64 = 48; @@ -56,6 +57,7 @@ const CANNOT_CHANGE_PERMISSIONS_ID: u64 = 49; // The apply's `ChangePasswordResult::UserNotFound` maps to `ResourceNotFound`. const RESOURCE_NOT_FOUND_ID: u64 = 20; // Snake-case `IggyError` names (== the `code` string in the JSON error body). +const UNAUTHENTICATED_CODE: &str = "unauthenticated"; const UNAUTHORIZED_CODE: &str = "unauthorized"; const INVALID_CREDENTIALS_CODE: &str = "invalid_credentials"; const CANNOT_DELETE_USER_CODE: &str = "cannot_delete_user"; @@ -328,6 +330,95 @@ async fn given_http_requests_when_denied_should_carry_typed_status_and_body(harn .await; } +#[iggy_harness] +async fn given_unexpired_jwts_when_user_revoked_should_reject_reads_and_writes( + harness: &TestHarness, +) { + let root = HttpClient::login_root(harness).await; + let permissions = Permissions { + global: GlobalPermissions { + manage_streams: true, + ..Default::default() + }, + ..Default::default() + }; + let created = root + .create_user( + "revoked-http-user", + "revoked-http-password", + serde_json::to_value(permissions).unwrap(), + ) + .await; + assert_eq!(created.status(), StatusCode::OK, "create active user"); + let details: Value = created.json().await.expect("created user JSON"); + let user_id = details["id"].as_u64().expect("created user id"); + let user_path = format!("/users/{user_id}"); + let established = root + .login("revoked-http-user", "revoked-http-password") + .await; + let unused_for_writes = root + .login("revoked-http-user", "revoked-http-password") + .await; + assert_ne!(established.token, unused_for_writes.token); + + assert_eq!( + established + .post_json("/streams", &json!({ "name": "before-revocation" })) + .await + .status(), + StatusCode::OK, + "active user must establish a write session" + ); + for credential in [&established, &unused_for_writes] { + assert_eq!( + credential.get("/streams").await.status(), + StatusCode::OK, + "both JWTs must authorize reads before revocation" + ); + } + + for revoked_state in ["inactive", "deleted"] { + let revoked = if revoked_state == "inactive" { + root.put_json(&user_path, &json!({ "status": "inactive" })) + .await + } else { + root.delete(&user_path).await + }; + assert_eq!( + revoked.status(), + StatusCode::NO_CONTENT, + "mark user {revoked_state}" + ); + for (kind, credential) in [ + ("established", &established), + ("unused-for-writes", &unused_for_writes), + ] { + let context = format!("{kind} JWT for {revoked_state} user"); + let read = credential.get("/streams").await; + let write = credential + .post_json( + "/streams", + &json!({ "name": format!("rejected-{revoked_state}-{kind}") }), + ) + .await; + for (operation, response) in [("read", read), ("write", write)] { + assert_eq!( + response.status(), + StatusCode::UNAUTHORIZED, + "{context}: {operation} must be rejected" + ); + assert_error_body( + response, + UNAUTHENTICATED_ID, + UNAUTHENTICATED_CODE, + &format!("{context}: {operation}"), + ) + .await; + } + } + } +} + // === Authorized requests carry the SDK-hidden success status codes ========== /// Root exercises the happy path; the SDK hides these exact status codes behind diff --git a/core/message_bus/src/replica/handshake.rs b/core/message_bus/src/replica/handshake.rs index 89db00473..0ccd0489a 100644 --- a/core/message_bus/src/replica/handshake.rs +++ b/core/message_bus/src/replica/handshake.rs @@ -36,8 +36,8 @@ //! cannot MAC one id and announce another), but the MAC key is a single //! cluster-wide PSK-derived subkey, not a per-replica key, so any PSK //! holder can mint a valid MAC for any (smaller) replica id - there is -//! no anti-Sybil guarantee. With auth disabled the acceptor stays in -//! legacy mode: the `ReplicaHello` id is trusted unverified. +//! no anti-Sybil guarantee. With auth disabled the `ReplicaHello` id is +//! trusted unverified after the build-identity check. //! //! Enabling auth is a coordinated-restart change, and not the only one: //! the consensus `cluster_id` is derived from the cluster name @@ -114,23 +114,22 @@ pub struct ReplicaTlsCtx { } /// Run the acceptor half on a delegated inbound stream and return the -/// verified peer id. +/// compatible peer id. /// -/// Reads the 256 B `ReplicaHello` frame, enforces command + cluster -/// match and the directional rule, then (when `auth` is configured) -/// runs the acceptor side of the mutual MAC handshake with `binding` +/// Reads the 256 B `ReplicaHello` frame, enforces command, cluster and +/// build-identity matches and the directional rule, then (when `auth` is +/// configured) runs the acceptor side of the mutual MAC handshake with `binding` /// folded into every MAC (the TLS exporter on a TLS link, see -/// [`ChannelBinding`]). With `auth = None` the acceptor stays in legacy -/// mode and returns the announced id unverified. +/// [`ChannelBinding`]). With `auth = None` it sends a build-identity +/// challenge and returns the announced id without authenticating it. /// /// The dialer (the peer) holds the strictly lower id; this acceptor /// holds the higher id (see the directional rule). The transcript /// therefore binds `dialer_id = peer_id`, `acceptor_id = self_id`. /// -/// On a rejection an authenticated, still-waiting dialer is answered -/// with a nonzero-status `build_challenge_message` (see `reject`) so -/// it learns the cause from its own logs rather than seeing a bare -/// connection close. +/// Command, cluster, build-identity and direction rejections send a +/// nonzero-status `ReplicaChallenge`, including when auth is disabled. +/// Missing-nonce and post-challenge failures close without a response. /// /// # Errors /// @@ -150,7 +149,8 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( let header = msg.header(); let peer_id = header.replica; let has_nonce = auth::has_nonce(&header.reserved_command); - // Both authenticated and unauthenticated peers read the build-gate response. + // Modern peers read the build-gate response even with auth disabled. + // Legacy unauthenticated peers discard the challenge in consensus dispatch. if header.command != Command::ReplicaHello { return reject( @@ -215,9 +215,6 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( }; if !has_nonce { - // Auth is enabled and enforced: a legacy (no-nonce) peer is rejected. - // No reject frame: a legacy dialer delegates its fd without reading a - // response. return reject( stream, our_cluster, @@ -297,9 +294,9 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( /// /// Logs and returns `Err(())` on any failure; the caller drops the /// stream and the shard-0 periodic sweep retries once the pending-dial -/// entry clears. With `auth = None` it sends a single plaintext -/// `ReplicaHello`. `binding` is folded into every MAC; both ends must -/// derive it from the same channel (see [`ChannelBinding`]). +/// entry clears. With `auth = None` it sends `ReplicaHello` and waits +/// for a compatible build-identity challenge. `binding` is folded into every +/// MAC; both ends must derive it from the same channel (see [`ChannelBinding`]). /// /// The dialer holds the strictly lower id, the acceptor the higher; the /// transcript binds `dialer_id = self_id`, `acceptor_id = peer_id`. If @@ -418,10 +415,11 @@ pub(crate) async fn dialer_handshake<S: AsyncRead + AsyncWrite>( /// Log a rejected handshake with its `reason` and, when `nack` is set, send a /// best-effort reject [`build_challenge_message`] (a `ReplicaChallenge` with a -/// nonzero status and no nonce/MAC) so an authenticated, still-waiting dialer -/// learns the cause. `nack` is false for legacy (no-nonce) and post-challenge -/// rejects, whose peer is not reading a response (a frame would reach its VSR -/// reader instead). Always returns `Err` so callers `return`. +/// nonzero status and no nonce/MAC) so a waiting dialer learns the cause. +/// Early validation rejects send a frame even to legacy unauthenticated +/// peers, whose consensus dispatch safely discards the unexpected challenge. +/// Missing-nonce and post-challenge rejects set `nack` to false. +/// Always returns `Err` so callers `return`. #[allow(clippy::future_not_send, clippy::similar_names)] async fn reject<S: AsyncRead + AsyncWrite>( stream: &mut S, @@ -481,8 +479,8 @@ fn build_challenge_message( /// Build a `ReplicaHello` frame announcing this replica's id and `cluster_id`. /// When `nonce_d` is set it is placed in `reserved_command[0..32]` to open -/// the authenticated handshake; otherwise the frame is the legacy plaintext -/// announce. +/// the authenticated handshake; otherwise the caller adds only the build +/// identity for the unauthenticated exchange. fn build_hello_message( cluster_id: u128, replica_id: u8, diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index abc106522..f334a74d7 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -608,16 +608,13 @@ fn require_shard_zero<'a, T>( /// exact session and rebalances its memberships; `FinalizeSession` removes it after /// partition retirement. Every other op applies to the state /// machine and caches the reply for at-most-once dedup. `fire_notifier` runs the -/// post-commit hook (a no-op during reconstruction, before it is wired). Does not -/// advance `commit_min`; the caller owns that counter. +/// post-commit hook (a no-op during reconstruction, before it is wired). Returns +/// the committed reply without advancing `commit_min`; the caller owns that counter. /// -/// `table_mutations_allowed` gates the CLIENT-TABLE half only; the state -/// machine always applies. After a state transfer the two artifacts sit at -/// different frontiers -- the snapshot at `S`, the transferred table at -/// `C >= S` -- so the tail replay over `(S, commit_max]` must run every state -/// machine effect (they are above the snapshot) while skipping table effects at -/// or below `C` (they are already in the transferred table). Pass `true` -/// wherever no transfer is in play. +/// `table_mutations_allowed` gates table updates only. Passing `false` requires +/// the table to retain the session identities needed by state-machine effects. +/// Live state transfer instead passes its separate replay table with `true`: +/// the newer protection table may already have finalized those identities. /// /// # Panics /// If a committed op fails to apply, which is a decode/corruption bug, since a @@ -630,7 +627,8 @@ pub fn apply_committed_prepare<M>( table_mutations_allowed: bool, fire_notifier: impl Fn(Operation), prepare: Message<PrepareHeader>, -) where +) -> Message<ReplyHeader> +where M: StreamsFrontend + StateMachine< Input = Message<PrepareHeader>, @@ -682,7 +680,7 @@ pub fn apply_committed_prepare<M>( mux_stm .streams() .refresh_consumer_group_session(header.client, epoch); - return; + return build_reply_message(&header, &bytes::Bytes::new()); } if header.operation == Operation::Logout { let ended = if table_mutations_allowed { @@ -709,7 +707,7 @@ pub fn apply_committed_prepare<M>( iggy_common::IggyTimestamp::from(header.timestamp), ); } - return; + return build_reply_message(&header, &bytes::Bytes::new()); } if header.operation == Operation::FinalizeSession { let identity = SessionIdentity::decode_from(prepare.body()).unwrap_or_else(|_| { @@ -721,7 +719,7 @@ pub fn apply_committed_prepare<M>( if table_mutations_allowed { client_table.borrow_mut().finalize_session(identity); } - return; + return build_reply_message(&header, &bytes::Bytes::new()); } // Normal op: apply, build the reply. `Err` is decode/corruption only; a // business rejection commits as a deterministic no-op whose code rides @@ -732,16 +730,14 @@ pub fn apply_committed_prepare<M>( header.op ); }); - if table_mutations_allowed && let Some(user_id) = apply.revoked_user { - for identity in client_table - .borrow_mut() - .end_user_sessions(user_id, header.op) - { - mux_stm.streams().remove_consumer_group_member( - identity.client_id, - iggy_common::IggyTimestamp::from(header.timestamp), - ); - } + if let Some(user_id) = apply.revoked_user { + apply_user_revocation( + mux_stm, + client_table, + table_mutations_allowed, + user_id, + &header, + ); } fire_notifier(header.operation); let reply = build_reply_message_with(&header, apply.reply_body_len(), |dst| { @@ -749,13 +745,42 @@ pub fn apply_committed_prepare<M>( }); if table_mutations_allowed && header.client != RESERVED_CLIENT_ID + && !message_bus::is_auto_commit_client(header.client) && header.operation != Operation::CompleteConsumerGroupRevocation { - let outcome = client_table - .borrow_mut() - .commit_reply(header.client, header.user_id, reply); + let outcome = + client_table + .borrow_mut() + .commit_reply(header.client, header.user_id, reply.clone()); log_commit_reply_outcome(outcome, header.client, header.op); } + reply +} + +fn apply_user_revocation<M: StreamsFrontend>( + mux_stm: &M, + client_table: &RefCell<ClientTable>, + table_mutations_allowed: bool, + user_id: u32, + header: &PrepareHeader, +) { + if table_mutations_allowed { + client_table + .borrow_mut() + .end_user_sessions(user_id, header.op); + } + // A transferred table may already contain this end marker while the + // streams snapshot still needs the matching membership release. + let table = client_table.borrow(); + for identity in table.ended_sessions().filter(|identity| { + identity.metadata_watermark == header.op + && table.get_user_id(identity.client_id) == Some(user_id) + }) { + mux_stm.streams().remove_consumer_group_member( + identity.client_id, + iggy_common::IggyTimestamp::from(header.timestamp), + ); + } } /// Late-bound callback invoked after every committed op on shard 0's metadata @@ -858,9 +883,13 @@ pub struct IggyMetadata<C, J, S, M, SB = PingPongSuperblock> { commit_notifier: RefCell<Option<CommitNotifier>>, /// Client-table mutations at or below this op are already reflected in a /// state-transferred table, so the tail-repair commit walk must skip - /// them (re-running `commit_register` would double-bump epochs). `0` + /// them to preserve newer receipts and incarnations. `0` /// outside state transfer (no op is skipped). Monotone per install. client_table_frontier: Cell<u64>, + /// Session history at the state machine's replay frontier. The transferred + /// protection table can have already forgotten sessions whose membership + /// effects still need replaying. Dropped once replay reaches that table. + replay_client_table: RefCell<Option<RefCell<ClientTable>>>, /// Last built [`StateTransferOffer`], shared by every requester of the same /// snapshot generation. Rebuilding per request re-reads and re-decodes the /// whole snapshot on shard 0's pump, and hands each requester its own @@ -943,6 +972,7 @@ where client_table: RefCell::new(ClientTable::new(CLIENTS_TABLE_MAX)), commit_notifier: RefCell::new(None), client_table_frontier: Cell::new(0), + replay_client_table: RefCell::new(None), transfer_offer_cache: RefCell::new(None), prepare_gap_drops: Cell::new(0), partitions_max: Cell::new(0), @@ -1068,13 +1098,49 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> { } /// Client-table mutations at or below the frontier are already in the - /// state-transferred table; the commit walk skips them (re-running - /// `commit_register` would double-bump epochs). STM effects still apply - /// -- the frontier fences the TABLE only. + /// state-transferred protection table. Replay uses a separate table at the + /// state machine's frontier until it catches up. const fn client_table_mutation_allowed(&self, op: u64) -> bool { op > self.client_table_frontier.get() } + fn apply_prepare(&self, prepare: Message<PrepareHeader>) -> Message<ReplyHeader> + where + M: StreamsFrontend + + StateMachine< + Input = Message<PrepareHeader>, + Output = crate::stm::result::ApplyReply, + Error = IggyError, + >, + { + let op = prepare.header().op; + if self.client_table_mutation_allowed(op) { + return apply_committed_prepare( + &*self.mux_stm, + &self.client_table, + true, + |operation| self.fire_commit_notifier(operation), + prepare, + ); + } + let history = self.replay_client_table.borrow(); + let table = history + .as_ref() + .expect("transferred history must retain its replay table"); + let reply = apply_committed_prepare( + &*self.mux_stm, + table, + true, + |operation| self.fire_commit_notifier(operation), + prepare, + ); + drop(history); + if op == self.client_table_frontier.get() { + self.replay_client_table.borrow_mut().take(); + } + reply + } + /// Raise the forced-checkpoint margin to cover a configured /// prepare-queue depth (`[metadata] prepare_queue_depth`). Clamped to /// the built-in floor by the coordinator; no-op on shards without a @@ -1873,7 +1939,7 @@ where // built from one caught-up-primary read in `state_transfer_offer`. // Refuse rather than install, which drops the caller back to journal // repair with the local state untouched. - if commit_op < snapshot_seq || table_frontier > commit_op { + if commit_op < snapshot_seq || table_frontier < snapshot_seq || table_frontier > commit_op { tracing::error!( snapshot_seq, commit_op, @@ -1944,6 +2010,34 @@ where let local_applied = consensus.commit_min(); let snapshot_ahead = snapshot_seq > local_applied; + let snapshot_table = if snapshot_ahead { + let table = snapshot + .snapshot() + .client_table + .clone() + .ok_or(SnapshotError::MissingClientTable)?; + Some(ClientTable::from_snapshot(table).map_err(SnapshotError::ClientTable)?) + } else { + None + }; + let protection_frontier = table_frontier.max(self.client_table_frontier.get()); + let replay_table = if snapshot_ahead && protection_frontier > snapshot_seq { + snapshot_table + } else if !snapshot_ahead + && table_frontier > local_applied + && self.replay_client_table.borrow().is_none() + { + let local_table = self.client_table.borrow(); + Some(if local_table.capacity_committed() { + ClientTable::from_snapshot(local_table.to_snapshot()) + .map_err(SnapshotError::ClientTable)? + } else { + ClientTable::new(local_table.capacity()) + }) + } else { + None + }; + if snapshot_ahead && let Some(journal) = &self.journal { // Discard the WAL suffix above the incoming floor BEFORE anything // installs: the commit walk matches entries by op number alone, so @@ -2027,6 +2121,10 @@ where ); } + if snapshot_ahead || replay_table.is_some() { + *self.replay_client_table.borrow_mut() = replay_table.map(RefCell::new); + } + // 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()); @@ -2237,9 +2335,6 @@ where code, ) .into_generic())), - // Client-bug shapes (future epoch, id reused for a different - // operation): replaying cannot help, so the home shard stays silent. - PreflightOutcome::Drop => Some(Err(MetadataSubmitError::Canceled)), } } @@ -3097,20 +3192,6 @@ where if !head_is_ours { continue; } - if self.client_table_mutation_allowed(prepare_header.op) - && let Err(error) = self - .client_table - .borrow_mut() - .commit_capacity(prepare_header.retry_capacity as usize) - { - fatal( - FatalReason::UnreconcilableLogFrontier, - &format!( - "invalid committed metadata retry capacity at op={}: {error}", - prepare_header.op - ), - ); - } let mut entry = consensus .pop_committed_prepare() @@ -3136,65 +3217,7 @@ where // Sync-only — this is what makes pop/apply/advance atomic on // the single-threaded shard and keeps the head revalidation // sound. - let reply = if matches!( - prepare_header.operation, - Operation::Register | Operation::Logout | Operation::FinalizeSession - ) { - apply_committed_prepare( - &*self.mux_stm, - &self.client_table, - self.client_table_mutation_allowed(prepare_header.op), - |_| {}, - prepare, - ); - build_reply_message(&prepare_header, &bytes::Bytes::new()) - } else { - // Normal op: apply SM, commit_reply. `Err` is decode/corruption - // only; a business rejection commits as a deterministic no-op - // whose `code` rides the reply body, replayed on retry. - let apply = gated_apply(&*self.mux_stm, prepare).unwrap_or_else(|err| { - panic!( - "on_ack: committed metadata op={} failed to apply: {err}", - prepare_header.op - ); - }); - if self.client_table_mutation_allowed(prepare_header.op) - && let Some(user_id) = apply.revoked_user - { - for identity in self - .client_table - .borrow_mut() - .end_user_sessions(user_id, prepare_header.op) - { - self.mux_stm.streams().remove_consumer_group_member( - identity.client_id, - iggy_common::IggyTimestamp::from(prepare_header.timestamp), - ); - } - } - // Post-commit notifier (e.g. partition reconciler - // wake-up). Filtering by operation is the - // recipient's responsibility. - self.fire_commit_notifier(prepare_header.operation); - let reply = - build_reply_message_with(&prepare_header, apply.reply_body_len(), |dst| { - apply.write_reply_body(dst); - }); - // Transferred protection already covers ops at or below its - // frontier; newer client receipts must be retained before reply. - if self.client_table_mutation_allowed(prepare_header.op) - && prepare_header.client != RESERVED_CLIENT_ID - && prepare_header.operation != Operation::CompleteConsumerGroupRevocation - { - let outcome = self.client_table.borrow_mut().commit_reply( - prepare_header.client, - prepare_header.user_id, - reply.clone(), - ); - log_commit_reply_outcome(outcome, prepare_header.client, prepare_header.op); - } - reply - }; + let reply = self.apply_prepare(prepare); consensus.advance_commit_min(prepare_header.op); // Paired with the counter bump, and before the reply leaves: a // client that holds this reply may re-home onto any shard and read, @@ -4000,17 +4023,9 @@ where // after replicated commits, not only quorum-acked ones reached via // `on_ack` on the primary. // - // Table mutations are skipped at or below the state-transfer - // frontier: those ops are already reflected in the transferred - // table, while their state-machine effects still have to replay - // (the snapshot sits at a lower op). - apply_committed_prepare( - &*self.mux_stm, - &self.client_table, - self.client_table_mutation_allowed(header.op), - |operation| self.fire_commit_notifier(operation), - prepare, - ); + // The replay table follows the snapshot while the transferred table + // retains newer protection for client admission. + self.apply_prepare(prepare); consensus.advance_commit_min(op); self.advance_applied_frontier(op); debug!("commit_journal: committed op={op}"); @@ -4717,24 +4732,16 @@ mod tests { ); } - /// Minimal committed `Register` reply for `ClientTable::commit_register`, - /// which reads only `client` and `commit` (the assigned session). fn register_reply(client: u128, session: u64) -> Message<ReplyHeader> { - let header_size = size_of::<ReplyHeader>(); - let mut reply = Message::<ReplyHeader>::new(header_size); - let header = bytemuck::checked::try_from_bytes_mut::<ReplyHeader>( - &mut reply.as_mut_slice()[..header_size], + build_reply_message( + &PrepareHeader { + client, + op: session, + operation: Operation::Register, + ..Default::default() + }, + &bytes::Bytes::new(), ) - .expect("zeroed bytes are a valid ReplyHeader"); - *header = ReplyHeader { - client, - request: 0, - commit: session, - command: Command::Reply, - operation: Operation::Register, - ..Default::default() - }; - reply } #[test] @@ -6720,56 +6727,136 @@ mod tests { async fn given_user_revocation_when_committed_should_release_group_memberships() { const CLIENT: u128 = 1; const USER: u32 = 7; - let (_directory, metadata) = metadata_with_group_member(CLIENT).await; - metadata.mux_stm.users().ensure_root_user("root", "hash"); - let consensus = metadata.consensus.as_ref().unwrap(); - let request = RoutedRequestHeader::default(); - for user_id in 1..=USER { - let create = CreateUserRequest { - username: WireName::new(format!("user-{user_id}")).unwrap(), - password: "hash".to_owned(), - status: UserStatus::Active.as_code(), - permissions: None, - options: WireOptions::empty(), - }; - metadata - .mux_stm - .update(build_prepare_message( + for (transferred, finalized, primary, snapshot_ahead) in [ + (false, false, false, false), + (true, false, false, false), + (true, false, true, false), + (true, true, false, false), + (true, true, true, false), + (true, true, false, true), + (true, true, true, true), + ] { + let (_directory, mut metadata) = metadata_with_group_member(CLIENT).await; + metadata.mux_stm.users().ensure_root_user("root", "hash"); + let consensus = metadata.consensus.as_ref().unwrap(); + let request = RoutedRequestHeader::default(); + for user_id in 1..=USER { + let create = CreateUserRequest { + username: WireName::new(format!("user-{user_id}")).unwrap(), + password: "hash".to_owned(), + status: UserStatus::Active.as_code(), + permissions: None, + options: WireOptions::empty(), + }; + metadata + .mux_stm + .update(build_prepare_message( + consensus, + &request, + Operation::CreateUser, + &create.to_bytes(), + )) + .unwrap(); + } + let mut snapshot = + <IggySnapshot as Snapshot>::create(&*metadata.mux_stm, consensus.commit_min(), 1) + .unwrap(); + snapshot.snapshot_mut().client_table = + Some(metadata.client_table.borrow().to_snapshot()); + let deletion = consensus::seal_prepare_capacity( + build_prepare_message( consensus, &request, - Operation::CreateUser, - &create.to_bytes(), - )) - .unwrap(); - } - let deletion = consensus::seal_prepare_capacity( - build_prepare_message( - consensus, - &request, - Operation::DeleteUser, - &DeleteUserRequest { - user_id: WireIdentifier::numeric(USER), + Operation::DeleteUser, + &DeleteUserRequest { + user_id: WireIdentifier::numeric(USER), + } + .to_bytes(), + ), + metadata.client_table_capacity(), + ); + let mut tail = vec![deletion]; + if transferred { + let mut table = + ClientTable::decode(&metadata.client_table.borrow().encode()).unwrap(); + let ended = table.end_user_sessions(USER, tail[0].header().op); + if finalized { + assert!(table.finalize_session(ended[0])); + consensus.sequencer().set_sequence(tail[0].header().op); + tail.push(consensus::seal_prepare_capacity( + build_prepare_message( + consensus, + &request, + Operation::FinalizeSession, + &ended[0].to_bytes(), + ), + metadata.client_table_capacity(), + )); } - .to_bytes(), - ), - metadata.client_table_capacity(), - ); - apply_committed_prepare( - &*metadata.mux_stm, - &metadata.client_table, - true, - |_| {}, - deletion, - ); - assert_eq!(metadata.client_table.borrow().ended_sessions().count(), 1); - assert!( - metadata - .mux_stm - .streams() - .consumer_group_memberships(CLIENT) - .is_empty(), - "revoked sessions must release their consumer assignments" - ); + let frontier = tail.last().unwrap().header().op; + if snapshot_ahead { + let journal = metadata.journal.take(); + metadata = metadata_plane(); + metadata.journal = journal; + } + metadata + .install_state_transfer(&snapshot.encode().unwrap(), table, frontier, frontier) + .await + .unwrap(); + } + let transferred_bytes = transferred.then(|| metadata.client_table.borrow().encode()); + let consensus = metadata.consensus.as_ref().unwrap(); + for prepare in tail { + let op = prepare.header().op; + if primary { + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + metadata + .journal + .as_ref() + .unwrap() + .handle() + .append(prepare) + .await + .unwrap(); + metadata.resume_stranded_commits().await; + } else { + metadata + .journal + .as_ref() + .unwrap() + .handle() + .append(prepare) + .await + .unwrap(); + consensus.advance_commit_max(op); + metadata.commit_journal().await; + } + } + if let Some(expected) = transferred_bytes { + assert_eq!( + metadata.client_table.borrow().encode(), + expected, + "tail replay must preserve transferred retry protection" + ); + } + assert_eq!( + metadata.client_table.borrow().ended_sessions().count(), + usize::from(!finalized) + ); + assert!( + metadata.replay_client_table.borrow().is_none(), + "replay state must be released after catch-up" + ); + assert!( + metadata + .mux_stm + .streams() + .consumer_group_memberships(CLIENT) + .is_empty(), + "revoked sessions must release their consumer assignments: transferred={transferred} finalized={finalized} primary={primary} snapshot_ahead={snapshot_ahead} committed={}", + consensus.commit_min() + ); + } } #[test] @@ -7193,10 +7280,8 @@ mod tests { 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 snapshot = + <IggySnapshot as Snapshot>::create(&TestMux::default(), SESSION, 1).unwrap(); let mut older_table = None; for (client, epoch) in [ (CLIENT, SESSION), @@ -7216,10 +7301,14 @@ mod tests { table .commit_register(client, USER, [0x5a; 32], reply) .unwrap(); + if epoch == SESSION { + snapshot.snapshot_mut().client_table = Some(table.to_snapshot()); + } if epoch == OLDER_FRONTIER { older_table = Some(ClientTable::decode(&table.encode()).unwrap()); } } + let snapshot = snapshot.encode().unwrap(); let expected_table = table.encode(); metadata .install_state_transfer(&snapshot, table, TABLE_FRONTIER, TABLE_FRONTIER) @@ -7442,19 +7531,43 @@ mod tests { // A donor mux fills the snapshot the way a serving primary would, so // the restore path sees populated sections rather than a bare envelope. - let snapshot_bytes = - <IggySnapshot as Snapshot>::create(&TestMux::default(), SNAPSHOT_SEQ, 1) - .expect("donor snapshot builds") - .encode() - .expect("donor snapshot encodes"); - md.install_state_transfer( - &snapshot_bytes, - ClientTable::new(CLIENTS_TABLE_MAX), - 0, - SNAPSHOT_SEQ, - ) - .await - .expect("install succeeds"); + for missing in [true, false] { + for frontier in [SNAPSHOT_SEQ, SNAPSHOT_SEQ + 1] { + let mut invalid_snapshot = + <IggySnapshot as Snapshot>::create(&TestMux::default(), SNAPSHOT_SEQ, 1) + .unwrap(); + if !missing { + let mut table = ClientTable::new(CLIENTS_TABLE_MAX).to_snapshot(); + table.capacity = 0; + invalid_snapshot.snapshot_mut().client_table = Some(table); + } + assert!(matches!( + md.install_state_transfer( + &invalid_snapshot.encode().unwrap(), + ClientTable::new(CLIENTS_TABLE_MAX), + frontier, + frontier + ) + .await, + Err(SnapshotError::MissingClientTable | SnapshotError::ClientTable(_)) + )); + assert_eq!( + journal_handle.last_op(), + Some(3), + "invalid protection must not truncate the WAL" + ); + assert_eq!(consensus.commit_min(), 0); + assert_eq!(md.client_table.borrow().get_epoch(CLIENT), Some(SESSION)); + } + } + let table = ClientTable::new(CLIENTS_TABLE_MAX); + let mut snapshot = <IggySnapshot as Snapshot>::create(&TestMux::default(), SNAPSHOT_SEQ, 1) + .expect("donor snapshot builds"); + snapshot.snapshot_mut().client_table = Some(table.to_snapshot()); + let snapshot_bytes = snapshot.encode().expect("donor snapshot encodes"); + md.install_state_transfer(&snapshot_bytes, table, SNAPSHOT_SEQ, SNAPSHOT_SEQ) + .await + .expect("install succeeds"); assert_eq!( journal_handle.last_op(), diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs index ce1712fa9..02f1fe75f 100644 --- a/core/metadata/src/stm/snapshot.rs +++ b/core/metadata/src/stm/snapshot.rs @@ -80,6 +80,10 @@ pub enum SnapshotError { Serialize(rmp_serde::encode::Error), /// Deserialization failed. Deserialize(rmp_serde::decode::Error), + /// The embedded client table contains invalid sessions or retry receipts. + ClientTable(consensus::client_table::ClientTableDecodeError), + /// Committed retry protection is absent from the snapshot. + MissingClientTable, /// I/O error during snapshot persist/load. Io(std::io::Error), /// I/O error during a specific stage of snapshot persistence. @@ -105,7 +109,7 @@ pub enum SnapshotError { UnsupportedFormatVersion { found: u32, expected: u32 }, /// A state-transfer descriptor's frontiers contradict its artifacts: a /// `commit_op` below the snapshot the same offer ships, or a client-table - /// frontier above that commit point. Both are impossible from a + /// frontier outside the snapshot-to-commit range. Both are impossible from a /// caught-up-primary offer, so the install is refused and the receiver falls /// back to journal repair rather than moving its frontiers off a bad /// manifest. @@ -138,6 +142,8 @@ impl fmt::Display for SnapshotError { match self { Self::Serialize(e) => write!(f, "snapshot serialization failed: {e}"), Self::Deserialize(e) => write!(f, "snapshot deserialization failed: {e}"), + Self::ClientTable(e) => write!(f, "snapshot client table is invalid: {e}"), + Self::MissingClientTable => write!(f, "snapshot client table is missing"), Self::Io(e) => write!(f, "snapshot I/O error: {e}"), Self::Persist { stage, source } => { write!(f, "snapshot persist failed at {stage:?} stage: {source}") @@ -182,8 +188,10 @@ impl std::error::Error for SnapshotError { match self { Self::Serialize(e) => Some(e), Self::Deserialize(e) => Some(e), + Self::ClientTable(e) => Some(e), Self::Io(e) | Self::Persist { source: e, .. } => Some(e), Self::ChecksumMismatch { .. } + | Self::MissingClientTable | Self::InvalidTrailer | Self::Truncated { .. } | Self::UnsupportedFormatVersion { .. } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index de5a2e1f7..fe210cb82 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -898,6 +898,15 @@ where .await } + /// Open prepare history for every disk partition, including replicated topics. + /// + /// `prepares-{created_revision}/` holds `prepares-{generation}.wal`, the + /// published `frontier`, and `segment-{generation}-{start_offset}.log` hard + /// links retaining message bodies. Replicated sends still incur buffered WAL + /// writes and link bookkeeping, but do not require an fsync before acknowledgement. + /// Persisted messages, persisted consumer offsets and session retirement request + /// sync barriers; checkpoints also sync before reclaiming history. + /// /// A supplied WAL has already had its receipt checkpoint restored before /// segment recovery; opening it again must not repeat that disk scan. /// @@ -910,12 +919,12 @@ where group_commit_delay: std::time::Duration, recovered: Option<(Rc<PartitionPersistence>, Vec<Message<PrepareHeader>>)>, ) -> Result<(), IggyError> { - let directory = self + let partition_directory = self .partition_dir .as_ref() .ok_or(IggyError::CannotReadFile)?; - let directory = - std::path::Path::new(directory).join(format!("prepares-{}", self.created_revision)); + let directory = std::path::Path::new(partition_directory) + .join(format!("prepares-{}", self.created_revision)); if recovered.is_none() && self.consensus.replica_count() > 1 && !self.consensus.recovery_election_allowed() @@ -927,13 +936,13 @@ where IggyError::CannotSyncFile })?; } - if let Some(directory) = &self.partition_dir { - self.materialization_missing = - crate::state_transfer::materialization_is_missing(directory, self.created_revision) - .await - .map_err(|_| IggyError::CannotReadFile)?; - self.ensure_materialization_recovery(); - } + self.materialization_missing = crate::state_transfer::materialization_is_missing( + partition_directory, + self.created_revision, + ) + .await + .map_err(|_| IggyError::CannotReadFile)?; + self.ensure_materialization_recovery(); let restore_receipts = recovered.is_none(); let (persistence, prepares) = if let Some(recovered) = recovered { recovered diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 5dda4d6fb..32efb1538 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -483,7 +483,7 @@ impl iggy_common::VsrSessionControl for QuicClient { *self .consensus_session .lock() - .expect("consensus session mutex poisoned") = ConsensusSession::new(); + .map_err(|_| IggyError::InvalidConfiguration)? = ConsensusSession::new(); self.consumer_group_state.clear_session_scoped(); self.poll_router.clear_session(); Ok(()) @@ -567,9 +567,8 @@ impl QuicClient { }) } - /// Whether an `AutoLogin` is configured on this client, which makes the - /// session after any connect the configured user's rather than whoever - /// signed in by hand. + /// Whether fallback `AutoLogin` credentials are configured. A remembered + /// successful sign-in takes precedence after reconnecting. pub(crate) fn auto_login_configured(&self) -> bool { matches!(self.config.auto_login, AutoLogin::Enabled(_)) } @@ -1096,8 +1095,8 @@ impl QuicClient { } None => { let mut session = consensus_session - .lock() - .expect("consensus session mutex poisoned"); + .lock() + .map_err(|_| IggyError::InvalidConfiguration)?; crate::vsr::encode_request_header(&mut session, code, &payload)?.0 } }; @@ -1319,6 +1318,68 @@ mod tests { send.stopped().await.unwrap(); } + #[tokio::test] + async fn given_poisoned_session_when_sending_or_resetting_should_return_configuration_error() { + const NO_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(50); + 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 consensus_session = Arc::clone(&client.consensus_session); + assert!( + std::panic::catch_unwind(|| { + let _session = consensus_session.lock().unwrap(); + panic!("poison the session for the regression test"); + }) + .is_err() + ); + + let mut header = None; + assert_eq!( + client + .send_raw_request( + SEND_MESSAGES_CODE, + Bytes::from_static(b"send"), + false, + &mut header, + ) + .await, + Err(IggyError::InvalidConfiguration) + ); + assert!(header.is_none()); + assert_eq!( + client.reset_vsr_session().await, + Err(IggyError::InvalidConfiguration) + ); + { + let session = consensus_session.lock().unwrap_err().into_inner(); + assert_eq!(session.session(), Some(1)); + assert_eq!(session.current_request_id(), 1); + } + assert!( + tokio::time::timeout(NO_REQUEST_TIMEOUT, peer.accept_bi()) + .await + .is_err() + ); + client.disconnect_transport().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); diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index 4d448007b..68f28826d 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -395,7 +395,7 @@ impl iggy_common::VsrSessionControl for TcpClient { *self .consensus_session .lock() - .expect("consensus session mutex poisoned") = ConsensusSession::new(); + .map_err(|_| IggyError::InvalidConfiguration)? = ConsensusSession::new(); self.consumer_group_state.clear_session_scoped(); self.poll_router.clear_session(); Ok(()) @@ -1044,21 +1044,15 @@ impl TcpClient { Ok(()) } - /// Whether an `AutoLogin` is configured on this client, which makes the - /// session after any connect the configured user's rather than whoever - /// signed in by hand. + /// Whether fallback `AutoLogin` credentials are configured. A remembered + /// successful sign-in takes precedence after reconnecting. pub(crate) fn auto_login_configured(&self) -> bool { matches!(self.config.auto_login, AutoLogin::Enabled(_)) } - /// Credentials to sign in with after connecting: the configured ones, or - /// else the ones a manual sign-in on this client succeeded with. A manual - /// sign-in is otherwise less reconnectable than a configured one, which - /// is a surprising difference between two ways of doing the same thing. - /// - /// A password change this client committed for the configured user is - /// applied on top: the configured password will never work again, and every - /// later reconnect would otherwise fail `InvalidCredentials`. + /// Credentials from the last successful sign-in, falling back to configured + /// `AutoLogin` when none are remembered. Committed username and password + /// changes are applied to the configured fallback. async fn sign_in_credentials(&self) -> Option<Credentials> { // The sign-in that last succeeded, whoever ran it: a client is whoever // it last signed in as, so a reconnect restores the session the caller @@ -1590,9 +1584,9 @@ impl TcpClient { } None => { let encoded = { - let mut consensus_session = consensus_session - .lock() - .expect("consensus session mutex poisoned"); + let Ok(mut consensus_session) = consensus_session.lock() else { + return (None, Err(IggyError::InvalidConfiguration)); + }; crate::vsr::encode_request_header(&mut consensus_session, code, &payload) }; match encoded { @@ -1681,7 +1675,7 @@ impl TcpClient { match crate::vsr::decode_response_split(&response_header, body) .map_err(|error| retry_outcome.observe(error)) { - Err(error) if !retry_transient || code == BIND_SESSION_CODE => return Err(error), + Err(error) if !retry_transient => return Err(error), // `TransientNotCommitted`: the op's outcome is unknown // (e.g. a view change canceled it in flight) -- ONLY a // same-session replay of the same request id is safe @@ -2017,6 +2011,53 @@ mod tests { } } + #[tokio::test] + async fn given_poisoned_session_when_sending_or_resetting_should_return_configuration_error() { + const TEST_BUDGET: std::time::Duration = std::time::Duration::from_secs(1); + let (listener, address) = live_endpoint().await; + let client = client_with(&address); + Client::connect(&client).await.unwrap(); + client.bind_vsr_session(1).await.unwrap(); + let (peer, _) = listener.accept().await.unwrap(); + let consensus_session = Arc::clone(&client.consensus_session); + assert!( + std::panic::catch_unwind(|| { + let _session = consensus_session.lock().unwrap(); + panic!("poison the session for the regression test"); + }) + .is_err() + ); + + let deadline = tokio::time::Instant::now() + TEST_BUDGET; + let (header, result) = client + .send_raw_vsr_attempt( + SEND_MESSAGES_CODE, + Bytes::from_static(b"send"), + None, + deadline, + deadline, + false, + ) + .await; + assert!(header.is_none()); + assert_eq!(result, Err(IggyError::InvalidConfiguration)); + assert_eq!( + client.reset_vsr_session().await, + Err(IggyError::InvalidConfiguration) + ); + { + let session = consensus_session.lock().unwrap_err().into_inner(); + assert_eq!(session.session(), Some(1)); + assert_eq!(session.current_request_id(), 1); + } + let mut byte = [0]; + assert_eq!( + peer.try_read(&mut byte).unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + client.disconnect_transport().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); diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs index f3e54e7ca..c158d2c61 100644 --- a/core/sdk/src/websocket/websocket_client.rs +++ b/core/sdk/src/websocket/websocket_client.rs @@ -477,7 +477,7 @@ impl iggy_common::VsrSessionControl for WebSocketClient { *self .consensus_session .lock() - .expect("consensus session mutex poisoned") = ConsensusSession::new(); + .map_err(|_| IggyError::InvalidConfiguration)? = ConsensusSession::new(); self.consumer_group_state.clear_session_scoped(); self.poll_router.clear_session(); Ok(()) @@ -555,9 +555,8 @@ impl WebSocketClient { }) } - /// Whether an `AutoLogin` is configured on this client, which makes the - /// session after any connect the configured user's rather than whoever - /// signed in by hand. + /// Whether fallback `AutoLogin` credentials are configured. A remembered + /// successful sign-in takes precedence after reconnecting. pub(crate) fn auto_login_configured(&self) -> bool { matches!(self.config.auto_login, AutoLogin::Enabled(_)) } @@ -1242,7 +1241,7 @@ impl WebSocketClient { let request = { let mut session = consensus_session .lock() - .expect("consensus session mutex poisoned"); + .map_err(|_| IggyError::InvalidConfiguration)?; crate::vsr::encode_contiguous_request(&mut session, code, &payload, &mut used_header)? }; trace!( @@ -1403,6 +1402,64 @@ mod tests { stream.send(Message::Binary(frame.into())).await.unwrap(); } + #[tokio::test] + async fn given_poisoned_session_when_sending_or_resetting_should_return_configuration_error() { + const NO_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(50); + 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 consensus_session = Arc::clone(&client.consensus_session); + assert!( + std::panic::catch_unwind(|| { + let _session = consensus_session.lock().unwrap(); + panic!("poison the session for the regression test"); + }) + .is_err() + ); + + let mut header = None; + assert_eq!( + client + .send_raw_request( + SEND_MESSAGES_CODE, + Bytes::from_static(b"send"), + false, + &mut header, + ) + .await, + Err(IggyError::InvalidConfiguration) + ); + assert!(header.is_none()); + assert_eq!( + client.reset_vsr_session().await, + Err(IggyError::InvalidConfiguration) + ); + { + let session = consensus_session.lock().unwrap_err().into_inner(); + assert_eq!(session.session(), Some(1)); + assert_eq!(session.current_request_id(), 1); + } + assert!( + tokio::time::timeout(NO_REQUEST_TIMEOUT, peer.next()) + .await + .is_err() + ); + client.disconnect_transport().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); diff --git a/core/server/src/http/extractor.rs b/core/server/src/http/extractor.rs index 7a8a26e53..9401536e8 100644 --- a/core/server/src/http/extractor.rs +++ b/core/server/src/http/extractor.rs @@ -181,6 +181,7 @@ pub(in crate::http) async fn resolve_credential( if let Ok(claims) = state.jwt.decode(bearer).await && let Ok(user_id) = claims.sub.parse::<u32>() { + state.ensure_active_user(user_id)?; return Ok(( format!("{JWT_KEY_PREFIX}{}", claims.jti), user_id, diff --git a/core/server/src/http/state.rs b/core/server/src/http/state.rs index 7bcdab17b..2b8ecbb04 100644 --- a/core/server/src/http/state.rs +++ b/core/server/src/http/state.rs @@ -30,9 +30,10 @@ use axum::response::Response; use configs::server::ServerConfig; use consensus::{MetadataHandle, VsrConsensus}; use futures::channel::oneshot; -use iggy_common::{ClusterMetadata, IggyTimestamp}; +use iggy_common::{ClusterMetadata, IggyError, IggyTimestamp, UserStatus}; use message_bus::InstanceToken; use metadata::MetadataSubmitError; +use metadata::impls::metadata::StreamsFrontend; use send_wrapper::SendWrapper; use tokio::sync::Mutex; use tracing::warn; @@ -291,6 +292,7 @@ impl HttpInner { expiry: u64, ) -> Result<Rc<HttpSession>, AuthError> { loop { + self.ensure_active_user(user_id)?; let now = IggyTimestamp::now().to_secs(); if let Some(session) = self.live_session(&key, now) { return Ok(session); @@ -340,6 +342,19 @@ impl HttpInner { } } + pub(in crate::http) fn ensure_active_user(&self, user_id: u32) -> Result<(), AuthError> { + let active = self.shard.plane.metadata().mux_stm.users().read(|users| { + users + .items + .get(user_id as usize) + .is_some_and(|user| user.status == UserStatus::Active) + }); + if !active { + return Err(IggyError::Unauthenticated.into()); + } + Ok(()) + } + /// Highest metadata op `user_id` was told committed here; see /// [`MetadataWatermarks`] for why the floor is per user and lives outside /// the session table. @@ -485,6 +500,8 @@ impl HttpInner { ) .await .ok_or(AuthError::SessionUnavailable)?; + // Register and catch-up may yield to a user revocation. + self.ensure_active_user(user_id)?; let attachment = metadata .client_table .borrow_mut() diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 77952c7e4..eff6cf5da 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -4090,28 +4090,6 @@ where true } - /// Admit a `PartitionSubmit`: same gates as the [`MessageBag::Request`] - /// arm, but every refusal answers on `reply` instead of the bus, and the - /// admitted request carries an in-process reply channel down to the - /// pipeline entry so its committed reply comes back here rather than - /// being routed by `header.client`. - #[allow(clippy::future_not_send)] - pub async fn on_partition_submit( - &self, - request: Message<RoutedRequestHeader>, - reply: Sender<Option<Message<GenericHeader>>>, - ) where - B: MessageBus + 'static, - MJ: JournalHandle, - <MJ as JournalHandle>::Target: - Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, - M: RestorableMetadataStm, - T: ShardsTable, - { - self.dispatch_partition_submit(request, reply, None, None) - .await; - } - /// Single entry point for first and re-delivered partition submits. /// Session and operation checks precede parking. Partition readiness and /// metadata fences are rechecked when the owning partition materializes. diff --git a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs index cc2120385..905e334fa 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs @@ -243,6 +243,11 @@ internal sealed class ConsensusSession var sessionId = _session ?? throw VsrError.Exception(VsrError.UNAUTHENTICATED, "A replicated request requires a bound consensus session."); + if (_requestCounter == ulong.MaxValue) + { + throw VsrError.Exception(VsrError.REQUEST_ID_EXHAUSTED, "Request ID counter exhausted."); + } + // Partition ops consume an id too: dedup needs each send to carry a distinct number, and the // metadata watermark tolerates the gaps. var requestId = _requestCounter; diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs index 87a6dd445..f0d7cbeeb 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/VsrError.cs @@ -46,6 +46,7 @@ internal static class VsrError internal const int CONSUMER_GROUP_PARTITION_NOT_OWNED = 5009; internal const int INCOMPATIBLE_PROTOCOL_VERSION = 14003; internal const int SESSION_MISMATCH = 14004; + internal const int REQUEST_ID_EXHAUSTED = 14005; /// <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/ConsensusSessionTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs index f55d0c9ad..30eb8cddc 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +using System.Reflection; using Apache.Iggy.Exceptions; using Apache.Iggy.Vsr; @@ -118,6 +119,47 @@ public sealed class ConsensusSessionTests Assert.Equal(VsrError.UNAUTHENTICATED, error.StatusCode); } + [Theory] + [InlineData((byte)VsrOperation.CreateStream)] + [InlineData((byte)VsrOperation.SendMessages)] + public void NextRequestId_ExhaustionPreservesIdentityAndRejectsLocally(byte operation) + { + var bindSecret = Enumerable.Range(0, LoginRegister.BIND_SECRET_BYTES) + .Select(value => (byte)value).ToArray(); + var session = new ConsensusSession(1, 10, bindSecret); + var clientId = session.ClientId; + var generation = session.Generation; + var counter = typeof(ConsensusSession).GetField("_requestCounter", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(counter); + counter.SetValue(session, ulong.MaxValue - 1); + + var lastFrame = session.Resolve((VsrOperation)operation); + Assert.Equal(new SessionFrame(clientId, ulong.MaxValue - 1, 10), lastFrame); + Assert.Equal(ulong.MaxValue, session.RequestCounter); + + for (var attempt = 0; attempt < 2; attempt++) + { + SessionFrame? issued = null; + var error = Assert.Throws<IggyInvalidStatusCodeException>(() => + { + issued = session.Resolve((VsrOperation)operation); + }); + + Assert.Null(issued); + Assert.Equal(VsrError.REQUEST_ID_EXHAUSTED, error.StatusCode); + Assert.False(error.FromServer); + Assert.Equal(ulong.MaxValue, session.RequestCounter); + Assert.Equal(clientId, session.ClientId); + Assert.Equal(10UL, session.Session); + Assert.Equal(generation, session.Generation); + Assert.Equal(bindSecret, session.BindSecret); + } + + Assert.Equal(new SessionFrame(clientId, ulong.MaxValue, 10), session.Resolve(VsrOperation.NonReplicated)); + Assert.Equal(ulong.MaxValue, session.RequestCounter); + } + [Fact] public void Resolve_DoesNotConsumeAnIdForNonReplicatedOps() { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java index ba80e077f..219e8fa5b 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java @@ -41,8 +41,9 @@ final class VsrLoginCodec { * Packed wire protocol version this codec targets: * {@code major << 20 | minor << 10 | patch}, 10 bits per field. * Keep in sync with {@code core/binary_protocol/src/version.rs}, independently - * of the crate release. The server enforces its minimum wire version and rejects - * newer major.minor versions. + * of the crate release. The server accepts the inclusive range from its minimum + * protocol version through its current version, including patch components. + * Versions outside this range, including newer patches, are rejected. */ static final int PROTOCOL_VERSION = (11 << 10) | 1; // 0.11.1 diff --git a/foreign/swift/Tools/golden-vectors/src/main.rs b/foreign/swift/Tools/golden-vectors/src/main.rs index 20df2f8ed..6f122e03b 100644 --- a/foreign/swift/Tools/golden-vectors/src/main.rs +++ b/foreign/swift/Tools/golden-vectors/src/main.rs @@ -52,7 +52,6 @@ use iggy_binary_protocol::requests::personal_access_tokens::*; use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; use iggy_binary_protocol::requests::streams::*; use iggy_binary_protocol::requests::system::*; -use iggy_binary_protocol::requests::system::{BindSessionRequest, SessionIdentity}; use iggy_binary_protocol::requests::topics::{ CreateTopicRequest, DeleteTopicRequest, GetTopicRequest, GetTopicsRequest, PurgeTopicRequest, UpdateTopicRequest,
