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 ae6dfb9c5989adf2ea095f04e19ca6886b20a414 Author: Hubert Gruszecki <[email protected]> AuthorDate: Sun Oct 4 11:35:52 2026 +0200 fix(consensus): fence retirement and preserve recovery frontiers Retirement could race with queued writes and reuse obsolete quorum reports. Transferred retry tables could also be checkpointed before their effects were applied, making restart recovery fail. Exclude writes while their session retirement is pending, qualify reports by view, and checkpoint only after applied state catches up. Retain one partition receipt, distinguish uncommitted transfer capacity, and isolate corrupt receipt recovery to its partition. Reuse resolved HTTP partitions and capture the admission frontier before routing waits. Warm peer identity outside shard runtimes. --- core/binary_protocol/src/version.rs | 3 + core/consensus/src/client_table.rs | 482 +++++++++++------------------ core/consensus/src/impls.rs | 38 +-- core/consensus/src/lib.rs | 16 +- core/message_bus/src/replica/handshake.rs | 17 +- core/metadata/src/impls/metadata.rs | 141 +++++++++ core/partitions/src/iggy_partition.rs | 334 ++++++++++++++++++-- core/partitions/src/state_transfer.rs | 89 ++++-- core/server/README.md | 22 ++ core/server/src/boot/mod.rs | 5 +- core/server/src/consumer_group/liveness.rs | 99 +++++- core/server/src/dispatch/mod.rs | 135 +++++--- core/server/src/dispatch/partition.rs | 181 ++++++++++- core/server/src/http/submit.rs | 3 +- core/server/src/partition_helpers.rs | 124 ++++++-- core/server/src/session_manager.rs | 13 +- core/shard/src/poll.rs | 9 +- core/simulator/src/lib.rs | 5 +- 18 files changed, 1210 insertions(+), 506 deletions(-) diff --git a/core/binary_protocol/src/version.rs b/core/binary_protocol/src/version.rs index 1473f644e..0dbcfe669 100644 --- a/core/binary_protocol/src/version.rs +++ b/core/binary_protocol/src/version.rs @@ -96,7 +96,10 @@ const COMPONENT_MAX: u32 = (1 << COMPONENT_BITS) - 1; const PATCH_MASK: u32 = COMPONENT_MAX; /// Current binary protocol version, independent of the crate's release version. +/// /// Version 0.11.1 requires the session-binding secret in login requests. +/// Intermediate development builds with this version may use incompatible layouts; +/// coordinated deployment requires matching clients and server binaries. pub const IGGY_PROTOCOL_VERSION: u32 = pack_protocol_version(0, 11, 1); /// Oldest protocol version this build accepts at login. diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index 865a61283..a16959c6b 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -101,6 +101,14 @@ pub const REGISTER_REQUEST_ID: u64 = 0; /// validation rejects this id, so no external client can claim it. pub const RESERVED_CLIENT_ID: u128 = 0; +#[must_use] +pub const fn is_partition_receipt_operation(operation: Operation) -> bool { + matches!( + operation, + Operation::SendMessages | Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset + ) +} + pub use iggy_binary_protocol::requests::users::login_register::BIND_SECRET_BYTES; const BIND_VERIFIER_CONTEXT: &str = "apache.iggy session bind verifier v1"; @@ -228,8 +236,8 @@ pub struct ClientEntrySnapshot { pub epoch: u64, pub user_id: u32, pub watermark: u64, - /// Wire bytes of the entry's latest committed reply, round-tripped through - /// `CachedReply::from_message`. Never empty: registration seeds the ring. + /// Wire bytes of the latest committed reply, validated as [`Message<ReplyHeader>`] + /// on restore. Never empty: registration seeds the ring. /// /// Serialized as a msgpack `bin` blob, not the integer array a plain `Vec<u8>` /// produces, which spends 2 bytes on every byte >= 0x80 and runs a checkpoint's @@ -401,7 +409,7 @@ pub enum RequestStatus { /// Committed-window verdict for a partition dedup slice. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SliceRequestStatus { +enum SliceRequestStatus { /// No retained commit mark for this request. New, /// A retained commit mark protects this request from execution. @@ -426,17 +434,13 @@ pub enum CommitReply { /// Which of the table's mechanisms an instance runs. /// -/// The metadata plane needs all of them. A partition group's slice needs only -/// the watermark: it has no register to mint an epoch from, no result section -/// worth caching, and one table per group rather than per node, so the -/// preallocated slot array would reserve ~384 KiB per partition before a single -/// client connects. The predicates below are the only two combinations that -/// exist, so the mode is an enum rather than three independent flags. +/// Metadata registers sessions and keeps a reply ring in preallocated slots. +/// Partition slices grow lazily and retain one session-qualified receipt per client. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ClientTableMode { /// Metadata plane: replies cached, epoch fenced, slots preallocated. Metadata, - /// One partition consensus group's slice: watermark only. + /// One partition consensus group's receipt protection. PartitionSlice, } @@ -448,9 +452,8 @@ impl ClientTableMode { matches!(self, Self::Metadata) } - /// Enforce the register-minted epoch fence. Off: entries carry no epoch, - /// `check_request` ignores the presented one, and a committed request may - /// create its own entry (there is no register to do it). + /// Enforce the metadata registration fence in `check_request`. + /// Partition admission checks its receipt's epoch independently. #[must_use] pub const fn fence_epoch(self) -> bool { matches!(self, Self::Metadata) @@ -608,25 +611,28 @@ impl ClientTable { /// Verify a binding credential, resolving a zero epoch to its original registration. /// /// # Errors - /// Returns `TransientNotCommitted` while registration is not visible, - /// `Unauthenticated` for a bad proof or ended session, and `InvalidSession` - /// for a different nonzero epoch. + /// Returns `TransientNotAccepted` while an uncertain registration is not visible, + /// `Unauthenticated` for an absent known session, bad proof or ended session, + /// and `InvalidSession` for a different nonzero epoch. pub fn bind_session( &mut self, client_id: u128, session: u64, secret: &[u8; BIND_SECRET_BYTES], ) -> Result<(u32, u64, SessionAttachment), IggyError> { - let user_id = self - .get_user_id(client_id) - .ok_or(IggyError::TransientNotCommitted)?; + let missing = if session == 0 { + IggyError::TransientNotAccepted + } else { + IggyError::Unauthenticated + }; + let user_id = self.get_user_id(client_id).ok_or(missing)?; let epoch = self .registered_session( client_id, user_id, bind_verifier(client_id, user_id, secret), )? - .ok_or(IggyError::TransientNotCommitted)?; + .ok_or(IggyError::Unauthenticated)?; if session != 0 && session != epoch { return Err(IggyError::InvalidSession(session)); } @@ -1110,36 +1116,6 @@ impl ClientTable { CommitReply::Cached } - /// Watermark-plus-window dedup check for a plane that mints no epoch. - /// - /// Only marked IDs in the retained window prove a committed request. - /// Unmarked IDs remain admissible within that window. Another user's - /// entry is not evidence about this caller and reads as new. - /// - /// # Panics - /// If called on a table that fences epochs, which must use - /// [`Self::check_request`]. - #[must_use] - pub fn check_slice_request( - &self, - client_id: u128, - user_id: u32, - request: u64, - ) -> SliceRequestStatus { - debug_assert!( - !self.mode.fence_epoch(), - "check_slice_request: an epoch-fencing table must use check_request" - ); - let Some(&slot_idx) = self.index.get(&client_id) else { - return SliceRequestStatus::New; - }; - let entry = self.slots[slot_idx].as_ref().expect("index/slot mismatch"); - if entry.user_id != user_id { - return SliceRequestStatus::New; - } - entry.check_slice_request(request) - } - #[must_use] /// # Panics /// Only if the private index refers to an unoccupied slot. @@ -1197,12 +1173,7 @@ impl ClientTable { if header.client == 0 || session == 0 || header.commit == 0 - || !matches!( - header.operation, - iggy_binary_protocol::Operation::SendMessages - | iggy_binary_protocol::Operation::StoreConsumerOffset - | iggy_binary_protocol::Operation::DeleteConsumerOffset - ) + || !is_partition_receipt_operation(header.operation) { return Err(ClientTableWireError::InvalidReply); } @@ -1223,11 +1194,13 @@ impl ClientTable { return Ok(cached.clone()); } let cached = CachedReply::from(reply); - entry.push_latest(cached.clone()); + entry.ring.clear(); + entry.latest_commit = cached.header().commit; + entry.ring.push_back(cached.clone()); Ok(cached) } - /// Fold an AUTO offset request into the partition watermark without a receipt. + /// Fold a partition receipt's identity into its committed request window. /// /// Replay is idempotent and order-insensitive within the retained window. /// The first committed request fixes its principal. Entries are never @@ -1240,7 +1213,7 @@ impl ClientTable { /// # Errors /// Returns an error for zero identities, a changed principal, or exhausted /// committed protection capacity. - pub fn commit_request( + pub(crate) fn commit_request( &mut self, client_id: u128, user_id: u32, @@ -1251,11 +1224,6 @@ impl ClientTable { !self.mode.cache_replies(), "commit_request: a reply-caching table must use commit_reply" ); - // Zero is the reserved client id, refused at every ingress (wire - // validation, the HTTP minter, the auto-commit guard at the call - // sites). Kept as a return rather than an assert so that an artifact - // slipping past the decoder degrades to no dedup for that entry instead - // of taking the replica down. if client_id == 0 { return Err(ClientTableWireError::InvalidWatermark { client_id }); } @@ -1364,12 +1332,7 @@ impl ClientTable { || header.commit != entry.latest_commit || header.commit == 0 || header.size as usize != entry.reply.len() - || !matches!( - header.operation, - iggy_binary_protocol::Operation::SendMessages - | iggy_binary_protocol::Operation::StoreConsumerOffset - | iggy_binary_protocol::Operation::DeleteConsumerOffset - ) + || !is_partition_receipt_operation(header.operation) { return Err(ClientTableWireError::InvalidReply); } @@ -1784,11 +1747,8 @@ impl ClientTable { Ok(table) } - /// Slot capacity, i.e. the largest table this one can absorb from a peer. - /// - /// Sized at construction from the configured cap, then raised by - /// [`Self::from_snapshot`] to cover any slot the checkpoint holds, so this - /// can exceed `[metadata] clients_table_max`. + /// Protection slots fixed by the first committed prepare's capacity. + /// Recovery and transfer preserve that limit over local configuration. #[must_use] pub const fn capacity(&self) -> usize { self.clients_max @@ -1796,10 +1756,8 @@ impl ClientTable { } impl ClientEntry { - /// Entry for a plane that mints no epoch and caches no reply: the fields a - /// [`ClientTableMode::PartitionSlice`] table never reads stay at their - /// zero values. Bit 0 of the window is forced on: the watermark itself is - /// committed by definition. + /// Partition identity initialized before its session and receipt are installed. + /// Bit 0 is always set because the watermark itself is committed. const fn watermark_only( client_id: u128, user_id: u32, @@ -2583,7 +2541,6 @@ mod tests { // partition plane actually relies on. const SLICE_USER: u32 = 3; - const OTHER_USER: u32 = 4; fn slice(clients_max: usize) -> ClientTable { ClientTable::with_mode(clients_max, ClientTableMode::PartitionSlice) @@ -2629,68 +2586,6 @@ mod tests { .collect() } - #[test] - fn given_partition_slice_when_empty_should_admit_and_not_preallocate() { - // The reason this plane cannot use the metadata mode: one table per - // group, so preallocating the cap would reserve hundreds of KiB per - // partition before a single client connects. - let table = slice(4096); - assert_eq!(table.count(), 0); - assert_eq!(table.slots.len(), 0, "slots must grow on demand"); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 1), - SliceRequestStatus::New - )); - } - - #[test] - fn given_partition_slice_when_request_replayed_should_report_duplicate() { - // Only what committed is a duplicate: an id below the watermark that - // never committed is a reordered arrival and still executes. - let mut table = slice(4); - table.commit_request(7, SLICE_USER, 5, 100).unwrap(); - - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 5), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 4), - SliceRequestStatus::New - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 6), - SliceRequestStatus::New - )); - } - - #[test] - fn given_partition_slice_when_request_id_gaps_should_accept_the_jump() { - // One client counter feeds several groups, so a slice legitimately sees - // only a subset of the ids that client mints; the skipped ids stay - // admissible in case they were routed here late rather than elsewhere. - let mut table = slice(4); - table.commit_request(7, SLICE_USER, 5, 100).unwrap(); - table.commit_request(7, SLICE_USER, 9, 101).unwrap(); - - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 5), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 9), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 7), - SliceRequestStatus::New - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 10), - SliceRequestStatus::New - )); - } - #[test] fn given_partition_slice_when_commit_replayed_should_be_idempotent() { let mut table = slice(4); @@ -2701,99 +2596,6 @@ mod tests { assert_eq!(table.watermarks_sorted(), vec![watermark(7, 5, 100)]); } - #[test] - fn given_partition_slice_when_lower_id_commits_late_should_admit_then_absorb() { - // A pipelining client had request 2 refused transiently and replays it - // after 3 committed: the replay is a new write, not a duplicate, and - // only once it commits does it read as one. - let mut table = slice(4); - table.commit_request(7, SLICE_USER, 1, 100).unwrap(); - table.commit_request(7, SLICE_USER, 3, 101).unwrap(); - - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 2), - SliceRequestStatus::New - )); - table.commit_request(7, SLICE_USER, 2, 102).unwrap(); - - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 2), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 1), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 3), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 4), - SliceRequestStatus::New - )); - assert_eq!( - table.watermarks_sorted(), - vec![DedupWatermark { - client: 7, - user_id: SLICE_USER, - watermark: 3, - latest_commit: 102, - committed_window: 0b111, - session: 0, - reply: Vec::new(), - }] - ); - } - - #[test] - fn given_partition_slice_when_id_ages_out_of_window_should_report_unknown_outcome() { - let mut table = slice(4); - table.commit_request(7, SLICE_USER, 1, 100).unwrap(); - table - .commit_request(7, SLICE_USER, 1 + COMMITTED_WINDOW_BITS + 10, 101) - .unwrap(); - - for request in [1, 2, 11] { - assert_eq!( - table.check_slice_request(7, SLICE_USER, request), - SliceRequestStatus::AgedOut, - "aged-out request {request} must have an unknown outcome" - ); - } - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 12), - SliceRequestStatus::New - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 1 + COMMITTED_WINDOW_BITS), - SliceRequestStatus::New - )); - } - - #[test] - fn given_partition_slice_when_watermark_jumps_should_shift_the_window() { - let mut table = slice(4); - table.commit_request(7, SLICE_USER, 1, 100).unwrap(); - table.commit_request(7, SLICE_USER, 2, 101).unwrap(); - table.commit_request(7, SLICE_USER, 5, 102).unwrap(); - - // 5 (bit 0), 2 (bit 3), 1 (bit 4) committed; 3 and 4 did not. - assert_eq!(table.watermarks_sorted()[0].committed_window, 0b11001); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 3), - SliceRequestStatus::New - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 4), - SliceRequestStatus::New - )); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 2), - SliceRequestStatus::Committed - )); - } - #[test] fn given_partition_slice_when_filled_to_cap_should_grow_without_holes() { // The lazily grown array must hand out every index once and never @@ -2810,35 +2612,6 @@ mod tests { assert!(table.slots.iter().all(Option::is_some)); } - #[test] - fn given_partition_slice_when_watermarks_installed_should_replace_not_merge() { - let mut table = slice(4); - table.commit_request(9, SLICE_USER, 3, 1).unwrap(); - table - .install_watermarks( - 4, - [receipt_watermark(1, 4, 50), receipt_watermark(2, 7, 60)], - ) - .unwrap(); - - assert_eq!(table.count(), 2); - assert!( - matches!( - table.check_slice_request(9, SLICE_USER, 3), - SliceRequestStatus::New - ), - "install replaces rather than merges" - ); - assert!(matches!( - table.check_slice_request(1, SLICE_USER, 4), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(2, SLICE_USER, 8), - SliceRequestStatus::New - )); - } - #[test] fn given_partition_slice_when_install_exceeds_local_cap_should_preserve_committed_limit() { let mut table = slice(2); @@ -2856,29 +2629,6 @@ mod tests { assert_eq!(clients_of(&table), vec![1, 2, 3]); } - #[test] - fn given_partition_slice_when_install_carries_user_should_keep_it() { - let mut table = slice(4); - table - .install_watermarks( - 4, - [DedupWatermark { - user_id: OTHER_USER, - ..receipt_watermark(1, 4, 50) - }], - ) - .unwrap(); - - assert!(matches!( - table.check_slice_request(1, OTHER_USER, 4), - SliceRequestStatus::Committed - )); - assert!(matches!( - table.check_slice_request(1, SLICE_USER, 4), - SliceRequestStatus::New - )); - } - #[test] fn given_partition_slice_when_exported_should_sort_ascending_by_client() { let mut table = slice(8); @@ -2891,19 +2641,6 @@ mod tests { assert_eq!(clients_of(&table), vec![10, 20, 30]); } - #[test] - fn given_partition_slice_when_cleared_should_admit_everything() { - let mut table = slice(4); - table.commit_request(7, SLICE_USER, 5, 100).unwrap(); - table.install_watermarks(4, std::iter::empty()).unwrap(); - - assert_eq!(table.count(), 0); - assert!(matches!( - table.check_slice_request(7, SLICE_USER, 5), - SliceRequestStatus::New - )); - } - #[test] fn given_partition_slice_when_client_is_reserved_zero_should_record_nothing() { // Zero is reserved cluster-wide and refused at every ingress; a commit @@ -2921,13 +2658,6 @@ mod tests { assert!(clients_of(&table).is_empty()); } - #[cfg(debug_assertions)] - #[test] - #[should_panic(expected = "an epoch-fencing table must use check_request")] - fn given_metadata_table_when_is_duplicate_called_should_panic() { - let _ = ClientTable::new(4).check_slice_request(7, SLICE_USER, 1); - } - #[cfg(debug_assertions)] #[test] #[should_panic(expected = "a reply-caching table must use commit_reply")] @@ -3149,6 +2879,144 @@ mod tests { } } + #[test] + fn given_partition_receipts_when_requests_commit_should_retain_only_the_latest() { + const CLIENT: u128 = 7; + const SESSION: u64 = 42; + const REQUESTS: u64 = 40; + let mut table = slice(1); + for request in 1..=REQUESTS { + table + .commit_partition_reply( + TEST_USER_ID, + SESSION, + make_reply_for(CLIENT, request, request), + ) + .unwrap(); + } + let entry = table.slots[table.index[&CLIENT]].as_ref().unwrap(); + assert_eq!( + entry.ring.len(), + 1, + "only the unresolved result needs bytes" + ); + assert_eq!(entry.latest().header().request, REQUESTS); + assert!(matches!( + table.check_partition_request( + CLIENT, + TEST_USER_ID, + SESSION, + REQUESTS - 1, + Operation::SendMessages, + ), + RequestStatus::AlreadyApplied { .. } + )); + assert!(matches!( + table.check_partition_request( + CLIENT, + TEST_USER_ID, + SESSION, + REQUESTS, + Operation::SendMessages, + ), + RequestStatus::Duplicate(_) + )); + } + + #[test] + fn given_missing_session_when_binding_should_distinguish_registration_uncertainty() { + const CLIENT: u128 = 7; + const SESSION: u64 = 42; + let secret = [0x5a; BIND_SECRET_BYTES]; + let mut table = ClientTable::new(1); + assert!(matches!( + table.bind_session(CLIENT, SESSION, &secret), + Err(IggyError::Unauthenticated) + )); + assert!(matches!( + table.bind_session(CLIENT, 0, &secret), + Err(IggyError::TransientNotAccepted) + )); + table + .commit_register( + CLIENT, + TEST_USER_ID, + bind_verifier(CLIENT, TEST_USER_ID, &secret), + make_register_reply(CLIENT, SESSION), + ) + .unwrap(); + assert!(matches!( + table.bind_session(CLIENT, SESSION, &[0x6a; BIND_SECRET_BYTES]), + Err(IggyError::Unauthenticated) + )); + } + + #[test] + fn given_partition_receipts_when_request_ids_skip_should_protect_only_committed_ids() { + const CLIENT: u128 = 7; + const SESSION: u64 = 42; + let mut table = slice(1); + for (request, commit) in [(1, 100), (5, 101)] { + table + .commit_partition_reply( + TEST_USER_ID, + SESSION, + make_reply_for(CLIENT, request, commit), + ) + .unwrap(); + } + assert!(matches!( + table.check_partition_request( + CLIENT, + TEST_USER_ID, + SESSION, + 3, + Operation::SendMessages, + ), + RequestStatus::New + )); + assert!(matches!( + table.check_partition_request( + CLIENT, + TEST_USER_ID, + SESSION, + 1, + Operation::SendMessages, + ), + RequestStatus::AlreadyApplied { + request: 1, + watermark: 5 + } + )); + table + .commit_partition_reply( + TEST_USER_ID, + SESSION, + make_reply_for(CLIENT, COMMITTED_WINDOW_BITS + 10, 102), + ) + .unwrap(); + assert!(matches!( + table.check_partition_request( + CLIENT, + TEST_USER_ID, + SESSION, + 3, + Operation::SendMessages, + ), + RequestStatus::AlreadyApplied { .. } + )); + assert!(matches!( + table.check_partition_request( + CLIENT, + TEST_USER_ID + 1, + SESSION, + COMMITTED_WINDOW_BITS + 11, + Operation::SendMessages, + ), + RequestStatus::Fenced { .. } + )); + } + #[test] fn partition_pressure_and_transfer_preserve_the_exact_receipt() { let mut table = slice(1); diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 974496254..cd17643cc 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -186,7 +186,7 @@ pub const PREPARE_QUEUE_CEILING: usize = DVC_HEADERS_MAX - 1; pub const PROBE_ATTEMPTS_MAX: u32 = 5; /// Maximum number of clients tracked in the clients table. -/// When exceeded, the client with the oldest committed request is evicted. +/// New sessions are refused at capacity; ended sessions retain protection until retirement. pub const CLIENTS_TABLE_MAX: usize = 8192; /// Default live dedup entries per PARTITION consensus group. @@ -621,6 +621,10 @@ impl LocalPipeline { self.request_queue.pop_front() } + pub fn pending_requests(&self) -> impl Iterator<Item = &RequestEntry> { + self.request_queue.iter() + } + /// True iff `prepare_queue` is full (NOT including `request_queue`). /// Callers branch on this between direct push and [`Self::push_request`]. #[must_use] @@ -785,24 +789,6 @@ impl LocalPipeline { .any(|r| r.message.header().client == client) } - /// True if either queue already holds this exact `(client, request)`. - /// - /// The partition-plane in-flight check. Narrower than - /// [`Self::has_message_from_client`] on purpose: the partition pipeline is - /// depth-`prepare_queue_depth` by design, so blocking every concurrent - /// request from one client would serialize it to one in-flight write per - /// group. Only an exact replay needs absorbing. - #[must_use] - pub fn has_message_from_client_request(&self, client: u128, request: u64) -> bool { - self.prepare_queue - .iter() - .any(|p| p.header.client == client && p.header.request == request) - || self.request_queue.iter().any(|r| { - let header = r.message.header(); - header.client == client && header.request == request - }) - } - /// Verify pipeline invariants. /// /// # Panics @@ -958,10 +944,6 @@ impl Pipeline for LocalPipeline { Self::has_message_from_client(self, client_id) } - fn has_message_from_client_request(&self, client_id: u128, request: u64) -> bool { - Self::has_message_from_client_request(self, client_id, request) - } - fn cancel_all_subscribers(&mut self) { Self::cancel_all_subscribers(self); } @@ -2082,16 +2064,6 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { self.pipeline.borrow().pending_request(client_id) } - /// True iff this exact `(client, request)` is already in flight. The - /// partition plane's in-flight dedup: absorbs a replay without serializing - /// a client's pipeline depth. - #[must_use] - pub fn pipeline_has_message_from_client_request(&self, client_id: u128, request: u64) -> bool { - self.pipeline - .borrow() - .has_message_from_client_request(client_id, request) - } - /// Header of the oldest in-flight prepare. #[must_use] pub fn pipeline_head_header(&self) -> Option<PrepareHeader> { diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs index 304c97a43..1de069145 100644 --- a/core/consensus/src/lib.rs +++ b/core/consensus/src/lib.rs @@ -70,22 +70,12 @@ pub trait Pipeline { fn verify(&self); /// True iff either queue carries `client_id`. Used by metadata-plane - /// preflight for in-flight dedup; the partition plane uses the narrower - /// [`Self::has_message_from_client_request`] instead, to keep a client's - /// pipeline depth. Default `false`; falls through to slot dedup in - /// `check_request`. + /// preflight for in-flight dedup. Default `false` falls through to the + /// committed receipt check. fn has_message_from_client(&self, _client_id: u128) -> bool { false } - /// True iff either queue carries this exact `(client, request)`. The - /// partition-plane in-flight dedup check: narrow on purpose, so a client - /// keeps its pipeline depth and only an exact replay is absorbed. - /// Default `false`. - fn has_message_from_client_request(&self, _client_id: u128, _request: u64) -> bool { - false - } - fn pending_client_ids(&self) -> impl Iterator<Item = u128>; /// The unresolved session, request number, and operation. fn pending_request( @@ -179,7 +169,7 @@ pub mod le_cursor; pub use client_table::{ CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableMode, ClientTableSnapshot, ClientTableWireError, CommitReply, DedupWatermark, - EXPIRED_SESSION_REQUEST_ID, RESERVED_CLIENT_ID, + EXPIRED_SESSION_REQUEST_ID, RESERVED_CLIENT_ID, is_partition_receipt_operation, }; pub mod state_manifest; pub use state_manifest::{ diff --git a/core/message_bus/src/replica/handshake.rs b/core/message_bus/src/replica/handshake.rs index c14003e6a..786a3b4d9 100644 --- a/core/message_bus/src/replica/handshake.rs +++ b/core/message_bus/src/replica/handshake.rs @@ -539,11 +539,9 @@ fn with_identity( /// # Errors /// Returns an error if the running executable cannot be opened or read. pub fn binary_identity(release: &str) -> std::io::Result<[u8; auth::IDENTITY_LEN]> { - static ARTIFACT: OnceLock<[u8; auth::IDENTITY_LEN]> = OnceLock::new(); + static ARTIFACT: OnceLock<std::io::Result<[u8; auth::IDENTITY_LEN]>> = OnceLock::new(); const READ_BYTES: usize = 32 * 1024; - let artifact = if let Some(artifact) = ARTIFACT.get() { - *artifact - } else { + let artifact = ARTIFACT.get_or_init(|| { let mut file = std::fs::File::open(std::env::current_exe()?)?; let mut hasher = blake3::Hasher::new(); let mut bytes = vec![0; READ_BYTES]; @@ -554,14 +552,15 @@ pub fn binary_identity(release: &str) -> std::io::Result<[u8; auth::IDENTITY_LEN } hasher.update(&bytes[..count]); } - let artifact = *hasher.finalize().as_bytes(); - let _ = ARTIFACT.set(artifact); - artifact - }; + Ok(*hasher.finalize().as_bytes()) + }); + let artifact = artifact + .as_ref() + .map_err(|error| std::io::Error::new(error.kind(), error.to_string()))?; let mut hasher = blake3::Hasher::new_derive_key("apache.iggy replica identity v1"); hasher.update(&iggy_binary_protocol::IGGY_PROTOCOL_VERSION.to_le_bytes()); hasher.update(&(release.len() as u64).to_le_bytes()); hasher.update(release.as_bytes()); - hasher.update(&artifact); + hasher.update(artifact); Ok(*hasher.finalize().as_bytes()) } diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 0fe021b77..82ba93859 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -1045,6 +1045,13 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> { ); return false; } + if client_table.capacity_committed() && current.capacity() != client_table.capacity() { + warn!( + configured_capacity = current.capacity(), + committed_capacity = client_table.capacity(), + "metadata retry capacity is fixed by committed state; ignoring local configuration" + ); + } *current = client_table; true } @@ -2010,6 +2017,14 @@ where ); } + 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); if snapshot_ahead { @@ -3529,6 +3544,11 @@ where // between commit_min+1 and commit_max haven't been applied to the // state machine yet, draining them would lose data on crash. let snap_op = consensus.commit_min(); + // A transferred table may include replies above the restored STM floor. + // Recovery must never replay that interval against its later protection. + if snap_op < self.client_table_frontier.get() { + return; + } // Stamp created_at from the injected consensus clock (seed-derived // under the simulator), not the wall clock, so replayed snapshots are // byte-identical. @@ -6880,6 +6900,127 @@ mod tests { ); } + #[compio::test] + async fn transferred_client_table_waits_for_applied_frontier_before_checkpoint() { + const CLIENT: u128 = 9; + const USER: u32 = 7; + const SNAPSHOT_OP: u64 = 1; + const TABLE_FRONTIER: u64 = 3; + const JOURNAL_SLOTS: usize = 16; + let directory = tempfile::tempdir().unwrap(); + let metadata_directory = directory.path().join(crate::impls::METADATA_DIR); + std::fs::create_dir_all(&metadata_directory).unwrap(); + let journal = PrepareJournal::open(&metadata_directory.join("journal.wal"), 0) + .await + .unwrap(); + let consensus = VsrConsensus::new( + 1, + 0, + 1, + server_common::sharding::METADATA_GROUP, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + let metadata: IggyMetadata<_, PrepareJournal, (), TestMux> = IggyMetadata::new( + Some(consensus), + Some(journal), + None, + None, + TestMux::default(), + Some(directory.path().to_path_buf()), + ); + let mut transferred_table = ClientTable::new(CLIENTS_TABLE_MAX); + transferred_table + .commit_capacity(CLIENTS_TABLE_MAX) + .unwrap(); + transferred_table + .commit_register( + CLIENT, + USER, + [0x5a; 32], + register_reply(CLIENT, SNAPSHOT_OP), + ) + .unwrap(); + let mut snapshot = + <IggySnapshot as Snapshot>::create(&TestMux::default(), SNAPSHOT_OP, 1).unwrap(); + snapshot.snapshot_mut().client_table = Some(transferred_table.to_snapshot()); + for request in 1..TABLE_FRONTIER { + let reply = build_reply_message( + &PrepareHeader { + client: CLIENT, + user_id: USER, + session: SNAPSHOT_OP, + request, + op: request + SNAPSHOT_OP, + operation: Operation::CreateStream, + ..Default::default() + }, + &bytes::Bytes::new(), + ); + assert_eq!( + transferred_table.commit_reply(CLIENT, USER, reply), + CommitReply::Cached, + ); + } + metadata + .install_state_transfer( + &snapshot.encode().unwrap(), + transferred_table, + TABLE_FRONTIER, + TABLE_FRONTIER, + ) + .await + .unwrap(); + let consensus = metadata.consensus.as_ref().unwrap(); + let journal = metadata.journal.as_ref().unwrap(); + let coordinator = metadata.coordinator.as_ref().unwrap(); + metadata.set_checkpoint_margin(usize::MAX); + metadata.checkpoint_if_needed(consensus, journal).await; + let persisted = IggySnapshot::load(&coordinator.snapshot_path()).unwrap().0; + assert_eq!(persisted.sequence_number(), SNAPSHOT_OP); + assert_eq!( + persisted.snapshot().client_table.as_ref().unwrap().slots[0] + .1 + .watermark, + 0, + "ahead retry protection cannot be persisted at an earlier state-machine floor", + ); + drop(persisted); + for (request, name) in [(1, "s1"), (2, "s2")] { + let prepare = metadata + .prepare_request(create_stream_request(CLIENT, request, name)) + .unwrap(); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + journal.handle().append(prepare).await.unwrap(); + } + metadata.resume_stranded_commits().await; + assert_eq!(consensus.commit_min(), TABLE_FRONTIER); + metadata.checkpoint_if_needed(consensus, journal).await; + let persisted = IggySnapshot::load(&coordinator.snapshot_path()).unwrap().0; + assert_eq!(persisted.sequence_number(), TABLE_FRONTIER); + drop(persisted); + drop(metadata); + let recovered = crate::impls::recovery::recover::<TestMux>( + directory.path(), + crate::impls::recovery::ReplicaIdentity { + cluster: 1, + replica_id: 0, + replica_count: 1, + }, + JOURNAL_SLOTS, + CLIENTS_TABLE_MAX, + |_| {}, + |_, _| {}, + ) + .await + .unwrap(); + assert_eq!( + recovered.client_table.get_watermark(CLIENT), + Some(TABLE_FRONTIER - 1) + ); + } + /// The commit loop and the promotion of queued requests run at the tail /// of `on_ack`, inside whichever future delivered the quorum ack. Drop /// that future mid-commit and — on an idle server — nothing re-drives diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index bb5f1d746..e7d88ab86 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -727,11 +727,42 @@ where && !self.requires_state_transfer() && self.consensus.is_normal() && !self.consensus.is_transferring() - && self.consensus.commit_min() == self.consensus.commit_max() + && self.consensus.pending_request(identity.client_id).is_none() && self.required_metadata_frontier >= identity.metadata_watermark && self.dedup.get_epoch(identity.client_id) != Some(identity.session) } + fn pending_session_retirement(&self, client_id: u128, session: u64) -> bool { + let matches_identity = |body: &[u8]| { + SessionIdentity::decode_from(body).is_ok_and(|identity| { + identity.client_id == client_id && identity.session == session + }) + }; + self.consensus.with_pipeline(|pipeline| { + pipeline.pending_requests().any(|entry| { + entry.message.header().operation == Operation::RetireSession + && matches_identity(entry.message.body()) + }) || pipeline.prepare_head().is_some_and(|head| { + let tail = pipeline.prepare_tail().expect("pipeline head implies tail"); + (head.header.op..=tail.header.op).any(|op| { + pipeline.message_by_op(op).is_some_and(|entry| { + entry.header.operation == Operation::RetireSession + && self + .log + .journal() + .inner + .repair_entry(op) + .is_none_or(|prepare| { + matches_identity( + &prepare.as_slice()[iggy_binary_protocol::HEADER_SIZE..], + ) + }) + }) + }) + }) + }) + } + /// Seed [`Self::applied_purge_generation`] from the partition dir's /// `purge.gen` file at build time (both fresh create and recovery walk /// this). Absent file reads 0, so a partition that never purged and a @@ -4490,6 +4521,38 @@ where return; } + // Retirement and the retiring identity must never occupy the same + // pipeline. Otherwise a later commit could recreate its receipt. + let retirement_conflict = if message.header().operation == Operation::RetireSession { + let Ok(identity) = SessionIdentity::decode_from(message.body()) else { + Self::send_partition_deny_or_log( + consensus, + message.header(), + IggyError::InvalidCommand.as_code(), + "invalid retirement reply failed", + reply.take(), + ) + .await; + return; + }; + consensus.pending_request(identity.client_id).is_some() + || self.pending_session_retirement(identity.client_id, identity.session) + } else { + !is_auto_commit_client(client_id) + && self.pending_session_retirement(client_id, message.header().session) + }; + if retirement_conflict { + Self::send_partition_deny_or_log( + consensus, + message.header(), + IggyError::TransientNotAccepted.as_code(), + "session retirement admission reply failed", + reply.take(), + ) + .await; + return; + } + // Parse once for both the delete-existence check and AckLevel dispatch. let consumer_offset = match message.header().operation { Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset => { @@ -9443,6 +9506,10 @@ mod tests { .set_last_prepare_checksum(inflight.header().checksum); origin.consensus().advance_commit_max(1); origin.consensus().advance_commit_min(1); + origin + .dedup_mut() + .commit_capacity(usize::try_from(committed.header().retry_capacity).unwrap()) + .unwrap(); // Retention can remove all polled segments while the repair ring still // serves the checkpoint prepare and the primary keeps accepting writes. origin.offset_space.committed_seeded = true; @@ -9454,6 +9521,7 @@ mod tests { let offer = origin.state_transfer_offer(&repair_config()).await.unwrap(); let offsets = crate::state_transfer::ConsumerOffsetsWire::decode(&offer.offsets.1).unwrap(); + assert_eq!(offsets.dedup_capacity, origin.dedup().capacity()); assert_eq!(offsets.prepare_checksum, Some(committed.header().checksum)); assert_eq!(offsets.checkpoint_prepare, committed.as_slice()); let (mut receiver, _) = recording_partition_at(1, 3); @@ -10319,6 +10387,29 @@ mod tests { assert!(partition.fatal.is_none(), "{:?}", partition.fatal); } + fn retire_session_request( + namespace: IggyNamespace, + identity: SessionIdentity, + ) -> Message<RoutedRequestHeader> { + let body = identity.to_bytes(); + let size = size_of::<RoutedRequestHeader>() + body.len(); + let mut request = Message::<RoutedRequestHeader>::new(size).transmute_header( + |_, header: &mut RoutedRequestHeader| { + header.command = Command::Request; + header.operation = Operation::RetireSession; + header.cluster = TEST_CLUSTER; + header.client = message_bus::AUTO_COMMIT_CLIENT_ID; + header.session = 1; + header.request = identity.metadata_watermark; + header.metadata_watermark = identity.metadata_watermark; + header.group = namespace.inner(); + header.size = u32::try_from(size).unwrap(); + }, + ); + request.as_mut_slice()[size_of::<RoutedRequestHeader>()..].copy_from_slice(&body); + request + } + #[compio::test] async fn given_interleaved_sends_and_retirements_when_committing_should_preserve_live_receipts() { @@ -10335,44 +10426,46 @@ mod tests { send.as_mut_slice()[size_of::<RoutedRequestHeader>()..] .copy_from_slice(&build_segment_record(partition.namespace(), 0)); partition.on_request(send, None).await; - if client_id == CLIENTS { - continue; - } + } + partition + .consensus + .advance_commit_max(u64::try_from(CLIENTS).unwrap()); + partition.commit_journal(&repair_config()).await; + for client_id in 1..CLIENTS { let identity = SessionIdentity { client_id, session: 1, metadata_watermark: ENDING_FRONTIER + u64::try_from(client_id).unwrap(), }; - let body = identity.to_bytes(); - let size = size_of::<RoutedRequestHeader>() + body.len(); - let mut retire = Message::<RoutedRequestHeader>::new(size).transmute_header( - |_, header: &mut RoutedRequestHeader| { - header.command = Command::Request; - header.operation = Operation::RetireSession; - header.cluster = TEST_CLUSTER; - header.client = message_bus::AUTO_COMMIT_CLIENT_ID; - header.session = 1; - header.request = identity.metadata_watermark; - header.metadata_watermark = identity.metadata_watermark; - header.group = partition.consensus.group(); - header.size = u32::try_from(size).unwrap(); - }, - ); - retire.as_mut_slice()[size_of::<RoutedRequestHeader>()..].copy_from_slice(&body); - partition.on_request(retire, None).await; + partition + .on_request( + retire_session_request(partition.namespace(), identity), + None, + ) + .await; + if client_id == 1 { + let mut send = checksumless_send_request(partition.namespace(), 1) + .transmute_header(|header, next: &mut RoutedRequestHeader| { + *next = header; + next.client = CLIENTS; + next.request = 2; + }); + send.as_mut_slice()[size_of::<RoutedRequestHeader>()..] + .copy_from_slice(&build_segment_record(partition.namespace(), 0)); + partition.on_request(send, None).await; + } } - let committed_op = u64::try_from(CLIENTS * 2 - 1).unwrap(); - assert_eq!( - partition.consensus.pipeline_len(), - usize::try_from(committed_op).unwrap() - ); + let committed_op = u64::try_from(CLIENTS * 2).unwrap(); partition.consensus.advance_commit_max(committed_op); partition.commit_journal(&repair_config()).await; assert!(partition.fatal.is_none(), "{:?}", partition.fatal); assert_eq!(partition.consensus.commit_min(), committed_op); assert_eq!(partition.dedup.count(), 1); assert!(partition.dedup.contains(CLIENTS)); - assert_eq!(replies.borrow().len(), usize::try_from(CLIENTS).unwrap()); + assert_eq!( + replies.borrow().len(), + usize::try_from(CLIENTS + 1).unwrap() + ); for client_id in 1..CLIENTS { assert!(partition.session_retired(SessionIdentity { client_id, @@ -10382,6 +10475,121 @@ mod tests { } } + #[compio::test] + async fn retirement_excludes_pending_target_writes_and_reports_during_unrelated_commits() { + const ENDING_FRONTIER: u64 = 10; + let (mut partition, replies) = recording_partition_at(0, 3); + let identity = SessionIdentity { + client_id: 1, + session: 1, + metadata_watermark: ENDING_FRONTIER, + }; + let mut send = checksumless_send_request(partition.namespace(), 1); + send.as_mut_slice()[size_of::<RoutedRequestHeader>()..] + .copy_from_slice(&build_segment_record(partition.namespace(), 0)); + partition.on_request(send.clone(), None).await; + partition + .on_request( + retire_session_request(partition.namespace(), identity), + None, + ) + .await; + assert_eq!( + partition.consensus.sequencer().current_sequence(), + 1, + "retirement must wait for the target write" + ); + partition.consensus.advance_commit_max(1); + partition.commit_journal(&repair_config()).await; + partition + .on_request( + retire_session_request(partition.namespace(), identity), + None, + ) + .await; + let retry = send + .clone() + .transmute_header(|header, next: &mut RoutedRequestHeader| { + *next = header; + next.request = 2; + }); + partition.on_request(retry, None).await; + assert_eq!( + partition.consensus.sequencer().current_sequence(), + 2, + "target write must wait for its retirement" + ); + let rejected = *bytemuck::checked::from_bytes::<ReplyHeader>( + &replies.borrow().last().unwrap().1.as_slice()[..size_of::<ReplyHeader>()], + ); + assert_eq!(rejected.status, IggyError::TransientNotAccepted.as_code()); + partition.consensus.advance_commit_max(2); + partition.commit_journal(&repair_config()).await; + assert_eq!(partition.dedup.count(), 0); + let unrelated = send.transmute_header(|header, next: &mut RoutedRequestHeader| { + *next = header; + next.client = 2; + next.metadata_watermark = ENDING_FRONTIER; + }); + partition.on_request(unrelated, None).await; + partition.consensus.advance_commit_max(3); + assert_eq!(partition.consensus.commit_min(), 2); + assert!( + partition.session_retired(identity), + "unrelated commits must not block the completed retirement" + ); + partition.commit_journal(&repair_config()).await; + assert_eq!(partition.dedup.count(), 1); + assert!(!partition.dedup.contains(identity.client_id)); + } + + #[compio::test] + async fn queued_retirement_excludes_a_later_target_write() { + let (mut partition, replies) = + recording_partition_with_pipeline(0, 3, LocalPipeline::with_capacities(1, 4)); + let mut unrelated = checksumless_send_request(partition.namespace(), 1).transmute_header( + |header, next: &mut RoutedRequestHeader| { + *next = header; + next.client = 2; + }, + ); + unrelated.as_mut_slice()[size_of::<RoutedRequestHeader>()..] + .copy_from_slice(&build_segment_record(partition.namespace(), 0)); + partition.on_request(unrelated, None).await; + let identity = SessionIdentity { + client_id: 1, + session: 1, + metadata_watermark: 10, + }; + partition + .on_request( + retire_session_request(partition.namespace(), identity), + None, + ) + .await; + assert_eq!( + partition + .consensus + .with_pipeline(LocalPipeline::request_queue_len), + 1 + ); + let mut target = checksumless_send_request(partition.namespace(), 1); + target.as_mut_slice()[size_of::<RoutedRequestHeader>()..] + .copy_from_slice(&build_segment_record(partition.namespace(), 0)); + partition.on_request(target, None).await; + assert_eq!( + partition + .consensus + .with_pipeline(LocalPipeline::request_queue_len), + 1, + "target cannot queue behind its retirement" + ); + let rejected = *bytemuck::checked::from_bytes::<ReplyHeader>( + &replies.borrow()[0].1.as_slice()[..size_of::<ReplyHeader>()], + ); + assert_eq!(rejected.status, IggyError::TransientNotAccepted.as_code()); + } + #[compio::test] async fn certified_empty_wal_without_superblock_preserves_its_log_view() { const CREATED_VIEW: u32 = 2; @@ -11065,6 +11273,78 @@ mod tests { checkpoint.persist_retry_checkpoint(through).await.unwrap(); } + #[compio::test] + async fn receipt_checkpoint_excludes_consumer_offsets_and_restores_retry_protection() { + const CHECKPOINT_OP: u64 = 1; + const METADATA_FRONTIER: u64 = 17; + const CLIENT: u128 = 42; + const SESSION: u64 = 4; + const CONSUMER_ID: u32 = 7; + let directory = tempfile::tempdir().unwrap(); + let mut partition = test_partition(); + partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); + std::fs::create_dir(directory.path().join("prepares-0")).unwrap(); + partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, CONSUMER_ID, 10, 10); + partition.consumer_offsets.pin().insert( + CONSUMER_ID as usize, + ConsumerOffset::new(ConsumerKind::Consumer, CONSUMER_ID, 10, String::new()), + ); + partition.seed_recovered_consumer_offset(ConsumerKind::ConsumerGroup, CONSUMER_ID, 11, 11); + partition.consumer_group_offsets.pin().insert( + ConsumerGroupId(CONSUMER_ID as usize), + ConsumerOffset::new(ConsumerKind::ConsumerGroup, CONSUMER_ID, 11, String::new()), + ); + partition.required_metadata_frontier = METADATA_FRONTIER; + partition + .dedup_mut() + .commit_capacity(consensus::PARTITION_DEDUP_CLIENTS_MAX) + .unwrap(); + let reply = consensus::build_reply_message( + &PrepareHeader { + client: CLIENT, + request: 1, + session: SESSION, + op: CHECKPOINT_OP, + operation: Operation::StoreConsumerOffset, + ..Default::default() + }, + &Bytes::new(), + ); + partition + .dedup_mut() + .commit_partition_reply(1, SESSION, reply) + .unwrap(); + partition + .consensus() + .restore_commit_state(CHECKPOINT_OP, CHECKPOINT_OP); + let path = partition + .persist_retry_checkpoint(CHECKPOINT_OP) + .await + .unwrap(); + let bytes = std::fs::read(path).unwrap(); + let content = consensus::le_cursor::split_verified_trailer(&bytes).unwrap(); + let mut cursor = consensus::le_cursor::LeCursor::new(content); + cursor.take(size_of::<[u8; 4]>()).unwrap(); + assert_eq!(cursor.u64().unwrap(), partition.consensus().group()); + assert_eq!(cursor.u64().unwrap(), partition.created_revision); + assert_eq!(cursor.u64().unwrap(), CHECKPOINT_OP); + let wire = crate::state_transfer::ConsumerOffsetsWire::decode(cursor.remaining()).unwrap(); + assert!(wire.consumers.is_empty() && wire.groups.is_empty()); + assert!(wire.checkpoint_prepare.is_empty()); + assert_eq!(wire.required_metadata_frontier, METADATA_FRONTIER); + let mut recovered = test_partition(); + recovered.set_partition_dir(directory.path().to_string_lossy().into_owned()); + recovered + .restore_retry_checkpoint(CHECKPOINT_OP) + .await + .unwrap(); + assert_eq!( + recovered.dedup().watermarks_sorted(), + partition.dedup().watermarks_sorted() + ); + assert_eq!(recovered.required_metadata_frontier, METADATA_FRONTIER); + } + #[compio::test] async fn receipt_checkpoint_recovery_refuses_missing_corrupt_and_foreign_artifacts() { const CHECKPOINT_OP: u64 = 7; diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index 9173a1e2b..268671639 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -447,7 +447,7 @@ impl ConsumerOffsetsWire { let group_count = cursor.u32()?; let external_group_count = cursor.u32()?; let dedup_count = cursor.u32()?; - if dedup_capacity == 0 || dedup_capacity > CLIENTS_TABLE_SLOT_MAX { + if dedup_capacity > CLIENTS_TABLE_SLOT_MAX { return Err(ConsumerOffsetsWireError::InvalidCapacity); } if dedup_count as usize > dedup_capacity { @@ -828,6 +828,28 @@ mod tests { } } + #[test] + fn given_zero_frontier_transfer_when_capacity_is_uncommitted_should_keep_local_limit() { + const LOCAL_CAPACITY: usize = 2; + const FIRST_COMMITTED_CAPACITY: usize = 3; + let wire = ConsumerOffsetsWire { + dedup_capacity: 0, + ..Default::default() + }; + let decoded = ConsumerOffsetsWire::decode(&wire.encode()).unwrap(); + let mut restored = checked_retry_table(&decoded, 0, LOCAL_CAPACITY).unwrap(); + assert!(!restored.capacity_committed()); + assert_eq!(restored.capacity(), LOCAL_CAPACITY); + restored.commit_capacity(FIRST_COMMITTED_CAPACITY).unwrap(); + assert_eq!(restored.capacity(), FIRST_COMMITTED_CAPACITY); + assert!(checked_retry_table(&decoded, 1, LOCAL_CAPACITY).is_err()); + let with_receipt = ConsumerOffsetsWire { + dedup: vec![dedup_entry(1, 1, 1)], + ..wire + }; + assert!(ConsumerOffsetsWire::decode(&with_receipt.encode()).is_err()); + } + #[test] fn given_offset_table_when_encoded_should_round_trip() { let encoded = table().encode(); @@ -2314,7 +2336,7 @@ where // frontier 0 claim a history that neither replica holds. // `install_state_transfer`'s `purge_advances` check re-decides the purge // against the metadata plane and refuses the rest. - let offsets_wire = self.offsets_wire_snapshot(true)?; + let offsets_wire = self.offsets_wire_snapshot()?; if segments.is_empty() && offsets_wire.next_offset == 0 && offsets_wire.purge_generation == 0 @@ -2496,10 +2518,7 @@ where /// Snapshot committed durable offsets only. Eager auto-commit progress and /// follower-local cursor entries stay in the live maps until a replicated /// store commits them, so neither can be promoted by state transfer. - fn offsets_wire_snapshot( - &self, - require_prepare: bool, - ) -> Result<ConsumerOffsetsWire, PartitionTransferUnavailable> { + fn offsets_wire_snapshot(&self) -> Result<ConsumerOffsetsWire, PartitionTransferUnavailable> { self.validate_consumer_offset_transfer_counts()?; let consumer_map = self.consumer_offsets.pin(); let consumers = self.snapshot_offset_kind(ConsumerKind::Consumer, |id| { @@ -2535,24 +2554,24 @@ where .then(|| self.consensus().last_prepare_checksum()) }) }; - let checkpoint_prepare = if require_prepare { - self.log - .journal() - .inner - .repair_entry(commit_op) - .map_or_else(Vec::new, |prepare| prepare.as_slice().to_vec()) - } else { - Vec::new() - }; - if require_prepare - && self.persistence.is_some() + let checkpoint_prepare = self + .log + .journal() + .inner + .repair_entry(commit_op) + .map_or_else(Vec::new, |prepare| prepare.as_slice().to_vec()); + if self.persistence.is_some() && (prepare_checksum.is_none() || (commit_op > 0 && checkpoint_prepare.is_empty())) { return Err(PartitionTransferUnavailable::MissingPrepareChecksum { op: commit_op }); } Ok(ConsumerOffsetsWire { required_metadata_frontier: self.required_metadata_frontier, - dedup_capacity: self.dedup().capacity(), + dedup_capacity: if self.dedup().capacity_committed() { + self.dedup().capacity() + } else { + 0 + }, checkpoint_prepare, prepare_checksum, purge_generation: self.applied_purge_generation, @@ -2591,7 +2610,7 @@ where pub(crate) fn offsets_wire_snapshot_for_test( &self, ) -> Result<Vec<(u32, u64)>, PartitionTransferUnavailable> { - self.offsets_wire_snapshot(true).map(|wire| wire.consumers) + self.offsets_wire_snapshot().map(|wire| wire.consumers) } /// Validate one completed `SEGMENT_LOG` artifact and spill it to staging @@ -2843,7 +2862,7 @@ where }); } let offsets_wire = ConsumerOffsetsWire::decode(offsets_bytes)?; - let retry_table = checked_retry_table(&offsets_wire, commit_op) + let retry_table = checked_retry_table(&offsets_wire, commit_op, self.dedup().capacity()) .map_err(PartitionInstallError::RetryProtection)?; if self.persistence.is_some() && (offsets_wire.prepare_checksum.is_none() @@ -4256,7 +4275,26 @@ fn segment_manifest_digest(manifest: &[consensus::StateArtifact]) -> u64 { fn checked_retry_table( wire: &ConsumerOffsetsWire, through_op: u64, + configured_capacity: usize, ) -> Result<ClientTable, consensus::client_table::ClientTableWireError> { + if wire.dedup_capacity == 0 { + if through_op != 0 || !wire.dedup.is_empty() { + return Err( + consensus::client_table::ClientTableWireError::InvalidCapacity { capacity: 0 }, + ); + } + return Ok(ClientTable::with_mode( + configured_capacity, + ClientTableMode::PartitionSlice, + )); + } + if wire.dedup_capacity != configured_capacity { + tracing::warn!( + configured_capacity, + committed_capacity = wire.dedup_capacity, + "partition retry capacity is fixed by committed state; ignoring local configuration" + ); + } let mut table = ClientTable::with_mode(wire.dedup_capacity, ClientTableMode::PartitionSlice); table.install_watermarks(wire.dedup_capacity, wire.dedup.iter().cloned())?; for entry in &wire.dedup { @@ -4322,9 +4360,12 @@ where "receipt frontier differs from the applied frontier", )); } - let wire = self - .offsets_wire_snapshot(false) - .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?; + let wire = ConsumerOffsetsWire { + dedup_capacity: self.dedup().capacity(), + required_metadata_frontier: self.required_metadata_frontier, + dedup: self.dedup().watermarks_sorted(), + ..Default::default() + }; let path = self.retry_checkpoint_path(through_op)?; write_retry_checkpoint( &path, @@ -4399,7 +4440,7 @@ where } let wire = ConsumerOffsetsWire::decode(cursor.remaining()) .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?; - let table = checked_retry_table(&wire, through_op) + let table = checked_retry_table(&wire, through_op, self.dedup().capacity()) .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?; *self.dedup_mut() = table; self.required_metadata_frontier = self diff --git a/core/server/README.md b/core/server/README.md index 26543e2e2..300dee6d7 100644 --- a/core/server/README.md +++ b/core/server/README.md @@ -62,6 +62,24 @@ External group offsets belong to groups managed outside Iggy, such as a Kafka gateway. They require no Iggy group membership and can exceed the partition's message-offset range. +The bind proof is an independent session credential. Password changes and PAT +revocation or expiry do not end an established session. Binding checks that the +owner still exists and is active; current permissions still govern each request. +Explicit logout, lease expiry and user deactivation end session access. + +`clients_table_max` and `dedup_clients_max` nominate immutable limits when the +first operation commits in each metadata or partition group. Recovery and state +transfer preserve those committed limits. Configuration changes affect groups +that have not committed a limit yet; existing groups log a mismatch and retain +their committed capacity. At capacity, new sessions or writers are refused until +ordered retirement releases slots. Live retry protection is never evicted. + +HTTP writes using one session and partition serialize through the previous +write's reply or bounded deadline. With `ack=none`, a request returns 202 after +its dispatch, while the next request to that partition waits for the previous +write to settle. Waiters acquire the in-flight permit after the partition gate, +so they do not consume the session's budget for other partitions. + Durability defaults remain `Replicated`, and ordinary SDK sends and explicit offset writes support the configured policy. Crash-safe send retries require `Persisted`; crash-safe explicit offset retries require `Persisted` and @@ -75,6 +93,10 @@ completion contracts. This release changes protocol and storage formats. Servers and compatible clients deploy together; mixed versions and rolling upgrades are unsupported. Peers verify protocol, release and executable identity before admission. +Deploy the same executable artifact to every replica; stripped or independently +rebuilt binaries have different identities even when their release matches. +Protocol 0.11.1 is the coordinated release boundary. Intermediate development +builds advertising that version are not a compatibility guarantee. An unsupported data directory is refused before WAL scanning or file changes. Replacing the binary does not migrate old data. Restore or migrate retained data only through a separately verified procedure; copying a format marker is not diff --git a/core/server/src/boot/mod.rs b/core/server/src/boot/mod.rs index 57e8e34b4..f864ed422 100644 --- a/core/server/src/boot/mod.rs +++ b/core/server/src/boot/mod.rs @@ -273,7 +273,7 @@ async fn wipe_system_path(config: &ServerConfig) -> Result<(), ServerError> { /// # Errors /// /// Returns an error if shard allocation fails, the inbox capacity is -/// invalid, or any OS thread fails to spawn. Per-shard recovery / +/// invalid, the executable cannot be read, or any OS thread fails to spawn. Per-shard recovery / /// listener / consensus failures surface through the per-thread `Result` /// the caller observes on `.join()`. /// @@ -298,6 +298,9 @@ pub fn bootstrap( install_default_crypto_provider(); validate_root_credentials_env(&config)?; warm_dummy_password_hash(); + // Hash the executable before shard runtimes start doing asynchronous work. + message_bus::replica::handshake::binary_identity(crate::VERSION) + .map_err(ServerError::BinaryIdentity)?; // The sync GetStats read path has no access to server config, so capture // the data directory here for its disk-usage reporting. crate::sysinfo_probe::init_stats_data_path(config.get_system_path().into()); diff --git a/core/server/src/consumer_group/liveness.rs b/core/server/src/consumer_group/liveness.rs index d29e8b4eb..44b867754 100644 --- a/core/server/src/consumer_group/liveness.rs +++ b/core/server/src/consumer_group/liveness.rs @@ -273,6 +273,7 @@ async fn retire_sessions<B, MJ, S, SB>( return; } let view = consensus.view(); + liveness.borrow_mut().observe_view(Some(view)); let identity = metadata .client_table .borrow() @@ -286,6 +287,24 @@ async fn retire_sessions<B, MJ, S, SB>( liveness .borrow_mut() .reconcile_retirement(Some(identity), revision); + let remote_quorum = consensus.is_primary() + && liveness + .borrow() + .retirement + .as_ref() + .is_some_and(|progress| progress.reporters.len() >= consensus.quorum_replication()); + if remote_quorum { + // The metadata primary need not belong to the common reporting quorum. + // Every reporter already proved retirement in every allocated group. + if let Err(error) = metadata.submit_session_finalization(identity).await { + trace!( + ?error, + client_id = identity.client_id, + "session finalization deferred" + ); + } + return; + } let (cursor, reported) = { let tracker = liveness.borrow(); let Some(progress) = &tracker.retirement else { @@ -591,8 +610,8 @@ mod tests { use crate::dispatch::test_support::{ SpyBus, TestShard, prepare_message, register_reply, test_shard, }; - use consensus::Sequencer; use consensus::client_table::ClientTable; + use consensus::{LocalPipeline, Sequencer, ViewRestore, VsrConsensus}; use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment; use iggy_binary_protocol::requests::consumer_groups::CreateConsumerGroupRequest; use iggy_binary_protocol::requests::streams::CreateStreamRequest; @@ -1274,6 +1293,84 @@ mod tests { assert!(tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT, TIMEOUT)); } + #[compio::test] + async fn retirement_sweep_discards_quorum_from_the_previous_view() { + const END_OP: u64 = SESSION + 1; + let (_directory, mut shard) = shard_with_members(1).await; + let mut consensus = VsrConsensus::new( + 1, + 0, + 3, + METADATA_GROUP, + shard.bus.clone(), + LocalPipeline::new(), + ); + consensus.restore_view(ViewRestore { + durable_view: Some((VIEW, VIEW)), + view_fallback: None, + seed_view: None, + }); + consensus.init(); + consensus.restore_commit_state(END_OP, END_OP); + consensus.sequencer().set_sequence(END_OP); + Rc::get_mut(&mut shard) + .unwrap() + .plane + .metadata_mut() + .consensus = Some(consensus); + let metadata = shard.plane.metadata(); + assert!( + metadata + .client_table + .borrow_mut() + .end_session(CLIENT, 0, SESSION, END_OP) + ); + let identity = metadata + .client_table + .borrow() + .ended_sessions() + .next() + .unwrap(); + let revision = metadata.mux_stm.streams().read(|inner| inner.revision); + let tracker = RefCell::new(ConsumerGroupLiveness::default()); + { + let mut tracker = tracker.borrow_mut(); + tracker.observe_view(Some(VIEW - 1)); + tracker.reconcile_retirement(Some(identity), revision); + let progress = tracker.retirement.as_mut().unwrap(); + progress.reported = true; + progress.reporters.extend([0, 1]); + } + let (stop, receiver) = shard::channel(1); + stop.try_send(()).unwrap(); + let (_signal, shutdown) = Shutdown::new(); + let sweep = Box::pin(retire_sessions( + &shard, &tracker, &receiver, &shutdown, TIMEOUT, + )); + assert!( + futures::poll!(sweep).is_ready(), + "old-view quorum must not start a session finalization" + ); + let tracker = tracker.borrow(); + let progress = tracker.retirement.as_ref().unwrap(); + assert!(!progress.reported); + assert!(progress.reporters.is_empty()); + assert_eq!(progress.cursor, 0); + assert_eq!( + metadata + .consensus + .as_ref() + .unwrap() + .sequencer() + .current_sequence(), + END_OP + ); + assert_eq!( + metadata.client_table.borrow().ended_sessions().next(), + Some(identity) + ); + } + #[test] fn retirement_quorum_discards_reports_from_an_older_namespace_revision() { let identity = iggy_binary_protocol::requests::system::SessionIdentity { diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs index 8b2863326..91099ba7a 100644 --- a/core/server/src/dispatch/mod.rs +++ b/core/server/src/dispatch/mod.rs @@ -70,9 +70,10 @@ use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_binary_protocol::{ EvictionReason, GenericHeader, Operation, RequestHeader, RoutedRequestHeader, }; -use iggy_common::IggyError; +use iggy_common::{IggyError, UserStatus}; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; +use metadata::impls::metadata::StreamsFrontend; use secrecy::ExposeSecret; use server_common::Message; use std::cell::RefCell; @@ -100,43 +101,9 @@ type ClientRequestQueues = Rc<RefCell<AHashMap<u128, VecDeque<Message<GenericHea const MAX_QUEUED_CLIENT_REQUESTS: usize = 1024; type ActiveClientRequests = Rc<RefCell<AHashSet<u128>>>; -// Session resume is performed BY THE LOGIN PATH, not by a separate -// credential-free rebind. -// -// A reconnecting client re-authenticates on the new connection and presents -// its previous `client_id` in the login frame; `submit_register_in_process` -// finds the existing table entry, verifies the authenticated user owns it, -// and returns its epoch, so `bind_session` binds the new transport to the -// old entry with its watermark and reply ring intact. That IS the resume. -// -// An earlier revision instead rebound an *unbound* transport straight from -// the table whenever a replicated frame carried a matching -// `(client, session)`, treating that pair as a bearer token. That was wrong -// in four ways, and the combination was a pre-auth session takeover: -// -// - it called `SessionManager::login` itself, so no credential was ever -// presented, and the connection was logged in as the entry's cached -// `user_id`; authority for replicated ops then resolves from the table -// (`resolve_acting_user_id`) and for partition ops from the session -// manager, so BOTH planes ran as the original registrant; -// - the pair carries far less entropy than "client-generated random -// u128" implies: HTTP mints `client_id` from a per-boot nonce and the -// shard-0 sequential counter (`mint_shard_zero_client_id`) and the epoch -// is a metadata commit position, so neither value is an authentication -// secret; -// - `ClientEntry` carries no transport or plane tag, so a raw TCP peer -// could bind an HTTP-originated session; -// - `bind_session` demotes the evicted holder to `Connected`, the one -// state `login` accepts, so the loser's next replicated frame -// re-resumed and stole the session back, unbounded and with no eviction -// frame either way. -// -// Routing resume through login also restores the checks that path owns: -// password / PAT verification, `UserStatus::Active`, PAT expiry, the -// protocol-version gate, and SDK-info recording. -// -// An unbound transport sending a replicated frame therefore gets the typed -// `Eviction(NoSession)` fail-fast below and must log in. +// BindSession authenticates with the registered secret, never with the +// public client id and epoch alone. It retains the original retry identity; +// unbound replicated frames still fail with Eviction(NoSession). #[allow(clippy::too_many_arguments)] fn enqueue_client_request<B, MJ, S, SB>( @@ -643,6 +610,7 @@ async fn handle_client_request<B, MJ, S, SB>( transport_client_id, user_id, consumer_session, + None, ) .await; } @@ -836,6 +804,15 @@ where Either::Right(_) => return Err(IggyError::TransientNotAccepted), } }; + let active_user = 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_user { + return Err(IggyError::Unauthenticated); + } sessions.borrow_mut().bind_authenticated_connection( transport_client_id, identity.client_id, @@ -852,15 +829,22 @@ mod tests { use super::*; use crate::cluster_meta::ClusterRoster; use crate::dispatch::host::ServerHost; - use crate::dispatch::test_support::{FIRST_BOOT, SpyBus, TestMux, TestShard, test_shard}; + use crate::dispatch::test_support::{ + FIRST_BOOT, SpyBus, TestMux, TestShard, prepare_message, register_reply, test_shard, + }; + use consensus::client_table::bind_verifier; use iggy_binary_protocol::Command; use iggy_binary_protocol::codes::{ GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE, POLL_MESSAGES_CODE, }; + use iggy_binary_protocol::requests::users::UpdateUserRequest; use iggy_binary_protocol::{EvictionHeader, ReplyHeader}; + use iggy_binary_protocol::{WireEncode, WireIdentifier, WireOptions}; use journal::prepare_journal::PrepareJournal; + use message_bus::installer::conn_info::ClientTransportKind; use metadata::IggyMetadata; use metadata::impls::metadata::IggySnapshot; + use metadata::stm::StateMachine; use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig}; use server_common::MESSAGE_ALIGN; use server_common::sharding::ShardId; @@ -874,6 +858,79 @@ mod tests { use std::mem::size_of; use std::sync::atomic::AtomicBool; + #[compio::test] + async fn bind_session_rechecks_the_owner_exists_and_is_active() { + const CLIENT: u128 = 1; + const TRANSPORT: u128 = 2; + const SESSION: u64 = 1; + const SECRET: [u8; 32] = [0x5a; 32]; + for status in [None, Some(UserStatus::Inactive), Some(UserStatus::Active)] { + let bus = SpyBus::default(); + let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); + let metadata = shard.plane.metadata(); + if let Some(status) = status { + metadata.mux_stm.users().ensure_root_user("iggy", "hash"); + metadata + .mux_stm + .update(prepare_message( + Operation::UpdateUser, + CLIENT, + SESSION, + &UpdateUserRequest { + user_id: WireIdentifier::numeric(0), + username: None, + status: Some(status.as_code()), + options: WireOptions::empty(), + } + .to_bytes(), + )) + .unwrap(); + } + metadata + .client_table + .borrow_mut() + .commit_register( + CLIENT, + 0, + bind_verifier(CLIENT, 0, &SECRET), + register_reply(CLIENT, SESSION), + ) + .unwrap(); + metadata.applied_frontier().advance(SESSION); + let sessions = Rc::new(RefCell::new(SessionManager::new())); + sessions.borrow_mut().ensure_connection( + TRANSPORT, + "127.0.0.1:34567".parse().unwrap(), + ClientTransportKind::Tcp, + ); + let outcome = complete_session_binding( + &shard, + &sessions, + TRANSPORT, + SessionIdentity { + client_id: CLIENT, + session: SESSION, + metadata_watermark: SESSION, + }, + BindSecret::new(Box::new(SECRET)), + ) + .await; + if status == Some(UserStatus::Active) { + assert_eq!(outcome.unwrap(), (0, SESSION)); + assert_eq!( + sessions.borrow().get_session(TRANSPORT), + Some((CLIENT, SESSION)) + ); + } else { + assert!( + matches!(outcome, Err(IggyError::Unauthenticated)), + "{status:?}: {outcome:?}" + ); + assert!(sessions.borrow().get_session(TRANSPORT).is_none()); + } + } + } + /// A test shard wired to its own lanes (the held sender feeds them), /// for the reply-lane pump tests below. fn reply_lane_test_shard(name: &str) -> (SpyBus, shard::TaggedSender, Rc<TestShard>) { diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index 15e6f310b..d67b106b9 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -88,7 +88,11 @@ use tracing::{debug, warn}; /// /// `vsr_client_id` keys the consumer-group offset fence (the member id), /// not the transport id stamped into the partition-op header. -#[allow(clippy::future_not_send, clippy::too_many_lines)] +#[allow( + clippy::future_not_send, + clippy::too_many_arguments, + clippy::too_many_lines +)] pub async fn dispatch_partition_request<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, request: Message<RoutedRequestHeader>, @@ -97,6 +101,7 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>( transport_client_id: u128, acting_user_id: Option<u32>, consumer_session: Option<(u128, SessionAttachment)>, + resolved_namespace: Option<u64>, ) where B: ShellBus, MJ: JournalHandle + 'static, @@ -105,13 +110,32 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>( SB: SuperblockStore + 'static, { let header = *request.header(); - let namespace = match resolve_partition_request_namespace( - shard, - header.operation, - request_body(&request), - consumer_session - .as_ref() - .map_or(vsr_client_id, |(parent, _)| *parent), + let metadata_watermark = shard.plane.metadata().applied_frontier().get(); + if consumer_session + .as_ref() + .is_some_and(|(_, attachment)| !attachment.is_valid()) + { + send_deny_reply( + shard, + transport_client_id, + &header, + IggyError::StaleClient.as_code(), + ) + .await; + return; + } + let namespace = match resolved_namespace.map_or_else( + || { + resolve_partition_request_namespace( + shard, + header.operation, + request_body(&request), + consumer_session + .as_ref() + .map_or(vsr_client_id, |(parent, _)| *parent), + ) + }, + Ok, ) { Ok(namespace) => namespace, Err(error) => { @@ -258,7 +282,7 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>( // gate above failed closed on `None`, so `0` (unattributed) is only a // type-level fallback here. new_header.user_id = acting_user_id.unwrap_or(0); - new_header.metadata_watermark = shard.plane.metadata().applied_frontier().get(); + new_header.metadata_watermark = metadata_watermark; }); if attachment.is_none() && vsr_client_id == transport_client_id { shard.dispatch(request.into_generic()); @@ -1342,6 +1366,131 @@ mod tests { assert_eq!(persistence.checkpoint_op(), 0); } + #[compio::test] + async fn pre_resolved_balanced_write_keeps_its_namespace_and_admission_frontier() { + const CLIENT: u128 = 1; + const ADMISSION_FRONTIER: u64 = 10; + const LATER_FRONTIER: u64 = 20; + let bus = SpyBus::default(); + let mut shard = test_shard(&bus, 0, 1, 1); + let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1); + shard.attach_senders(vec![sender]); + let shard = Rc::new(shard); + let metadata = shard.plane.metadata(); + metadata.mux_stm.users().ensure_root_user("iggy", "hash"); + metadata + .mux_stm + .update(prepare_message( + Operation::CreateStream, + CLIENT, + 1, + &CreateStreamRequest { + name: WireName::new("stream").unwrap(), + options: WireOptions::empty(), + } + .to_bytes(), + )) + .unwrap(); + metadata + .mux_stm + .update(prepare_message( + Operation::CreateTopicWithAssignments, + CLIENT, + 2, + &CreateTopicWithAssignmentsRequest { + request: CreateTopicRequest { + stream_id: WireIdentifier::numeric(0), + partitions_count: 2, + name: WireName::new("topic").unwrap(), + options: WireOptions::empty(), + }, + derived_options: WireOptions::empty(), + partitions: vec![ + CreatedPartitionAssignment { + partition_id: 0, + consensus_group_id: 1, + }, + CreatedPartitionAssignment { + partition_id: 1, + consensus_group_id: 2, + }, + ], + created_view: 0, + } + .to_bytes(), + )) + .unwrap(); + metadata.applied_frontier().advance(ADMISSION_FRONTIER); + let wire_header = SendMessagesHeader { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partitioning: WirePartitioning::Balanced, + messages_count: 1, + } + .to_bytes(); + let mut body = Vec::new(); + body.extend_from_slice(&u32::try_from(wire_header.len()).unwrap().to_le_bytes()); + body.extend_from_slice(&wire_header); + let namespace = + resolve_partition_request_namespace(&shard, Operation::SendMessages, &body, CLIENT) + .unwrap(); + let delayed_shard = Rc::clone(&shard); + let updater = compio::runtime::spawn(async move { + compio::time::sleep(std::time::Duration::from_millis(1)).await; + delayed_shard + .plane + .metadata() + .applied_frontier() + .advance(LATER_FRONTIER); + for partition_id in 0..2 { + let namespace = delayed_shard + .plane + .metadata() + .mux_stm + .streams() + .namespace_from_partition( + &WireIdentifier::numeric(0), + &WireIdentifier::numeric(0), + partition_id, + ) + .unwrap(); + delayed_shard + .shards_table() + .insert(namespace, PartitionLocation::new(ShardId::new(0), 0)); + } + }); + dispatch_partition_request( + &shard, + request_message(Operation::SendMessages, CLIENT, 1, 1, &body), + CLIENT, + 1, + CLIENT, + Some(DEFAULT_ROOT_USER_ID), + None, + Some(namespace), + ) + .await; + let _ = updater.await; + let ShardFrame::Consensus { + message: MessageBag::Request(request), + .. + } = owner_inbox.try_recv().unwrap() + else { + panic!("the dispatcher must submit one partition write"); + }; + assert_eq!( + request.header().group, + namespace, + "Balanced must advance once for the gate and write" + ); + assert_eq!( + request.header().metadata_watermark, + ADMISSION_FRONTIER, + "routing waits cannot promote an old binding past Logout" + ); + assert_eq!(metadata.applied_frontier().get(), LATER_FRONTIER); + } + #[compio::test] async fn given_invalid_partition_writes_when_resolving_should_preserve_offset_error_codes() { const VSR_CLIENT: u128 = 1; @@ -1468,8 +1617,17 @@ mod tests { )); for (index, (operation, body, expected)) in cases.into_iter().enumerate() { let request = request_message(operation, 1, 1, index as u64 + 1, &body); - dispatch_partition_request(&shard, request, 1, 1, 91, Some(DEFAULT_ROOT_USER_ID), None) - .await; + dispatch_partition_request( + &shard, + request, + 1, + 1, + 91, + Some(DEFAULT_ROOT_USER_ID), + None, + None, + ) + .await; let replies = bus.client_replies.borrow(); assert_eq!( replies.len(), @@ -1977,6 +2135,7 @@ mod tests { TRANSPORT, Some(DEFAULT_ROOT_USER_ID), None, + None, ) .await; diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs index dd2342c4b..ae9c642fb 100644 --- a/core/server/src/http/submit.rs +++ b/core/server/src/http/submit.rs @@ -366,7 +366,6 @@ async fn partition_write( dispatched: &mut Option<oneshot::Sender<Result<(), PartitionWriteError>>>, ) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> { let deadline = Instant::now() + PARTITION_WRITE_REPLY_TIMEOUT; - let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?; let namespace = crate::namespace::resolve_partition_request_namespace( &state.shard, operation, @@ -383,6 +382,7 @@ async fn partition_write( ) .await .map_err(|_| PartitionWriteError::Rejected(IggyError::TransientNotAccepted))?; + let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?; ensure_in_process_reply_target(state, session); let request_id = { let mut next_id = session.data_gate.lock().await; @@ -423,6 +423,7 @@ async fn partition_write( session.client_id, Some(session.user_id), Some((session.client_id, attachment)), + Some(namespace), ) .await; if let Some(dispatched) = dispatched.take() { diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index d2dbaa604..f0715e28a 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -583,27 +583,30 @@ async fn load_partition( // recovered message timestamp here, or an NTP rewind across a restart could // regress persisted `base_timestamp`. + let wal_directory = + Path::new(&partition_dir).join(format!("prepares-{}", partition_metadata.created_revision)); + let wal_error = |source: std::io::Error| match source.kind() { + std::io::ErrorKind::InvalidData + | std::io::ErrorKind::UnexpectedEof + | std::io::ErrorKind::NotFound => { + ServerError::PartitionRecovery(PartitionRecoveryError::Refused { + dir: PathBuf::from(&partition_dir), + stream_id: namespace.stream_id(), + topic_id: namespace.topic_id(), + partition_id: namespace.partition_id(), + reason: PartitionRecoveryRefusal::PrepareWal { + directory: wal_directory.clone(), + source, + }, + }) + } + _ => ServerError::PartitionPrepareWalIo { + dir: wal_directory.clone(), + source, + }, + }; let recovered_persistence = { - let directory = Path::new(&partition_dir) - .join(format!("prepares-{}", partition_metadata.created_revision)); - let wal_error = |source: std::io::Error| match source.kind() { - std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof => { - ServerError::PartitionRecovery(PartitionRecoveryError::Refused { - dir: PathBuf::from(&partition_dir), - stream_id: namespace.stream_id(), - topic_id: namespace.topic_id(), - partition_id: namespace.partition_id(), - reason: PartitionRecoveryRefusal::PrepareWal { - directory: directory.clone(), - source, - }, - }) - } - _ => ServerError::PartitionPrepareWalIo { - dir: directory.clone(), - source, - }, - }; + let directory = wal_directory.clone(); if replica_count == 1 && !journal::PartitionPrepareJournal::has_published_frontier(&directory) .await @@ -633,7 +636,7 @@ async fn load_partition( .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS), ) .await - .map_err(wal_error)?; + .map_err(&wal_error)?; if let Some(log_view) = persistence.certified_log_view() && log_view != 0 && log_view < partition_metadata.created_view @@ -667,11 +670,7 @@ async fn load_partition( partition .restore_retry_checkpoint(recovered_persistence.0.checkpoint_op()) .await - .map_err(|source| ServerError::PartitionPrepareWalIo { - dir: Path::new(&partition_dir) - .join(format!("prepares-{}", partition_metadata.created_revision)), - source, - })?; + .map_err(wal_error)?; let recovered_segments = recover_partition_segments( partitions_config, namespace, @@ -1162,7 +1161,8 @@ mod tests { use configs::server::ServerConfig; use consensus::Sequencer; use iggy_binary_protocol::batch::BATCH_HEADER_SIZE; - use iggy_binary_protocol::{Command, Operation, PrepareHeader}; + use iggy_binary_protocol::requests::system::SessionIdentity; + use iggy_binary_protocol::{Command, Operation, PrepareHeader, WireEncode}; use journal::DurableAppend; use journal::superblock::SuperblockStore; use partitions::PartitionPathLayout; @@ -1706,6 +1706,76 @@ mod tests { } } + #[compio::test] + async fn missing_or_corrupt_retry_checkpoints_fence_only_the_partition() { + let namespace = IggyNamespace::new(1, 1, 0); + for checkpoint in [None, Some(b"corrupt receipt checkpoint".as_slice())] { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + drop(build_solo_partition(&config).await.unwrap()); + let directory = PathBuf::from(config.get_partition_path(1, 1, 0)); + let wal = directory.join("prepares-0"); + let mut journal = journal::PartitionPrepareJournal::open(&wal, namespace.inner(), 0) + .await + .unwrap(); + let identity = SessionIdentity { + client_id: 1, + session: 1, + metadata_watermark: 2, + }; + let body = identity.to_bytes(); + let total = size_of::<PrepareHeader>() + body.len(); + let mut prepare = Message::<PrepareHeader>::new(total); + prepare.as_mut_slice()[size_of::<PrepareHeader>()..].copy_from_slice(&body); + let prepare = prepare.transmute_header(|_, header: &mut PrepareHeader| { + header.command = Command::Prepare; + header.operation = Operation::RetireSession; + header.cluster = CLUSTER; + header.group = namespace.inner(); + header.op = 1; + header.retry_capacity = u32::try_from(config.partition.dedup_clients_max).unwrap(); + header.client = message_bus::AUTO_COMMIT_CLIENT_ID; + header.session = 1; + header.request = identity.metadata_watermark; + header.size = u32::try_from(total).unwrap(); + header.checksum = header.identity_checksum(); + }); + journal.append(prepare.into_frozen()).await.unwrap(); + journal.checkpoint(1).await.unwrap(); + drop(journal); + if let Some(checkpoint) = checkpoint { + std::fs::write(wal.join("receipts-1.checkpoint"), checkpoint).unwrap(); + } + let partitions = solo_partitions(&config); + let outcome = load_partition_or_fence( + &config, + namespace, + Arc::new(PartitionStats::default()), + &Partition::new(0, namespace.inner(), IggyTimestamp::now(), 0, 0), + TopicRuntimeOptions::default(), + CLUSTER, + 0, + 1, + false, + Rc::new(IggyMessageBus::new(0)), + &partitions, + ) + .await + .unwrap(); + assert!( + outcome.is_none(), + "unusable retry protection must refuse this partition" + ); + assert!(wal.exists(), "refusal must retain the WAL evidence"); + if let Some(checkpoint) = checkpoint { + assert_eq!( + std::fs::read(wal.join("receipts-1.checkpoint")).unwrap(), + checkpoint + ); + } + } + } + #[compio::test] async fn given_solo_segments_without_wal_when_reloaded_should_refuse_and_preserve_bytes() { let root = tempfile::tempdir().unwrap(); diff --git a/core/server/src/session_manager.rs b/core/server/src/session_manager.rs index b80dc1894..fd684ce83 100644 --- a/core/server/src/session_manager.rs +++ b/core/server/src/session_manager.rs @@ -379,11 +379,16 @@ impl SessionManager { #[must_use] pub fn get_user_id(&self, connection_id: u128) -> Option<u32> { let conn = self.connections.get(&connection_id)?; + if conn + .consumer_session + .as_ref() + .is_some_and(|(_, attachment)| !attachment.is_valid()) + { + return None; + } match conn.state { - ConnectionState::Bound { user_id, .. } if self.get_session(connection_id).is_some() => { - Some(user_id) - } - _ => None, + ConnectionState::Bound { user_id, .. } => Some(user_id), + ConnectionState::Connected => None, } } diff --git a/core/shard/src/poll.rs b/core/shard/src/poll.rs index 4a6dcdad4..a19d88b08 100644 --- a/core/shard/src/poll.rs +++ b/core/shard/src/poll.rs @@ -31,7 +31,7 @@ use crate::shards_table::ShardsTable; use crate::{IggyShard, PartitionRead, PartitionReadReply, Sender}; use consensus::client_table::SessionAttachment; -use consensus::{Consensus, MetadataHandle, PartitionsHandle}; +use consensus::{Consensus, MetadataHandle, PartitionsHandle, is_partition_receipt_operation}; use iggy_binary_protocol::{Operation, RoutedRequestHeader}; use iggy_common::IggyError; use journal::superblock::SuperblockStore; @@ -60,12 +60,7 @@ pub struct ConsumerAttachment { impl ConsumerAttachment { pub(crate) fn validate_write(&self, operation: Operation) -> Result<(), IggyError> { - if !matches!( - operation, - Operation::SendMessages - | Operation::StoreConsumerOffset - | Operation::DeleteConsumerOffset - ) { + if !is_partition_receipt_operation(operation) { return Err(IggyError::InvalidCommand); } if !self.session.is_valid() { diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index cb8854d0a..36895ed86 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -3276,9 +3276,10 @@ mod tests { expected.commit_offset, 1, "both reordered requests must append" ); - for replay in [delayed, later] { + for (replay, expected_status) in [(delayed, 0), (later, IggyError::RequestTooOld.as_code())] + { let duplicate = submit_and_wait_for_reply(&mut sim, client.client_id(), 0, replay); - assert_eq!(duplicate.header().status, 0); + assert_eq!(duplicate.header().status, expected_status); assert_eq!( sim.offsets(0, namespace), Some(expected),
