This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch clients_table_v2 in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c8ee030488938b1185a4fd983618cea675f41cf9 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Mon Jul 27 09:58:55 2026 +0200 temp --- core/consensus/src/client_table.rs | 693 ++++++++++++++++++++++----------- core/consensus/src/metadata_helpers.rs | 227 +++++++---- core/metadata/src/impls/metadata.rs | 161 +++++--- 3 files changed, 700 insertions(+), 381 deletions(-) diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index 6b0a8526f..09a87292e 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -17,7 +17,7 @@ use iggy_binary_protocol::ReplyHeader; use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen}; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::mem::size_of; use tracing::trace; @@ -71,19 +71,46 @@ impl CachedReply { /// Real requests start at 1 (header validation enforces `request > 0`). pub const REGISTER_REQUEST_ID: u64 = 0; -/// Per-client entry (VR paper §4, Fig. 2): session + latest committed reply. +/// Displaced replies retained per entry for below-watermark duplicate hits. /// -/// `session` is assigned at registration and fixed for the entry's lifetime. +/// The SDK enforces one request in flight per session, so the only reply a +/// live client can be waiting for is its latest (`request == watermark`). +/// The ring answers old retransmits and post-rebind stragglers with the +/// original bytes instead of a bare "already applied"; losing an entry +/// degrades the answer, never correctness. In-memory only: ring contents are +/// refcount bumps and are never persisted or transferred. +const REPLY_RING_CAPACITY: usize = 4; + +/// Per-session entry: fence epoch + committed-request watermark + replies. +/// +/// The key (`client_id` today, the stable `session_id` once SDK identity +/// stability lands) is client-supplied; `epoch` is the server-minted fence +/// that orders rebinds of that key. #[derive(Debug)] pub struct ClientEntry { - /// Session number = commit op of the register. Monotonic across - /// registrations; new register always gets a higher session. - pub session: u64, - /// Acting user id captured at register. Fixed for the entry's lifetime; - /// lets every replica resolve session -> user without a metadata lookup. - pub user_id: u32, - /// Cached reply for client's latest committed request. - pub reply: CachedReply, + /// Fence epoch: 1 at first register, +1 per committed re-register. + /// Minted here, in apply order, so every replica derives the same value. + /// Requests stamped with an older epoch are zombies and get fenced; + /// a newer epoch than minted is a protocol violation. + epoch: u64, + /// Acting user id captured at register (re-register refreshes it: the + /// rebind re-authenticated). Lets every replica resolve session -> user + /// without a metadata lookup. + user_id: u32, + /// Highest committed request number. `REGISTER_REQUEST_ID` (0) until the + /// first app op commits. Survives re-register: a resumed session keeps + /// its dedup history. + watermark: u64, + /// `request_checksum` of the watermark request; catches a client reusing + /// a request id for a different operation. Zero when unstamped (integrity + /// fields are zeroed on the wire today), which disables the comparison. + watermark_checksum: u128, + /// Latest committed reply (register or app op). + reply: CachedReply, + /// Displaced app replies, oldest at front, bounded by + /// [`REPLY_RING_CAPACITY`]. Register replies never enter (their + /// `request == REGISTER_REQUEST_ID` can never match a lookup). + ring: VecDeque<CachedReply>, } /// Result of checking a request against the client table. @@ -93,49 +120,71 @@ pub struct ClientEntry { /// committed state. #[derive(Debug)] pub enum RequestStatus { - /// Not seen; proceed with consensus. + /// Above the watermark; proceed with consensus. Jumps are allowed: the + /// watermark records the highest committed request, not a contiguous + /// sequence, so `watermark + k` for any `k >= 1` is new. New, - /// Exact request already committed; re-send cached reply. + /// At or below the watermark with the original reply still cached; + /// re-send it. Duplicate(CachedReply), - /// Older than client's latest committed request; drop silently. - Stale, - /// No session for this client; must register first. + /// At or below the watermark, original reply no longer cached. Applied + /// once already; must not re-execute, nothing to replay. + AlreadyApplied { request: u64, watermark: u64 }, + /// Request number matches the watermark but its `request_checksum` + /// differs: the client reused a request id for a different operation. + /// Returning the cached reply would answer the wrong request. + ChecksumMismatch { request: u64 }, + /// No entry for this client; must register first. NoSession, - /// Session number doesn't match the entry. - SessionMismatch { expected: u64, received: u64 }, - /// Request != `committed + 1`. Skipped numbers would be lost permanently. - RequestGap { expected: u64, received: u64 }, - /// Client already has a session. From `check_register`. + /// Stamped epoch is older than the entry's: a zombie holdover from + /// before a re-register. Terminal for that holder. + Fenced { current: u64, received: u64 }, + /// Stamped epoch is newer than any this table minted: client bug + /// (epochs are only handed out by register replies). + EpochAhead { current: u64, received: u64 }, + /// Client already has an entry. From `check_register`. AlreadyRegistered { - session: u64, + epoch: u64, cached_reply: CachedReply, }, } -/// VSR client-table: durable per-client session state. +/// VSR client table: per-session fence epoch + request-watermark dedup. /// /// Fixed-size slot array (source of truth) + `HashMap` index (O(1) lookup). /// -/// ## Plane: metadata-only +/// ## Semantics (v2) +/// +/// - **Epoch, not commit.** Session identity is the client-supplied key; +/// the entry's `epoch` is a plain counter minted at `commit_register` +/// (1, then +1 per rebind). No field derives from a commit op number, so +/// the same table logic serves any consensus group. +/// - **Watermark, not contiguity.** A request above the watermark executes +/// (gaps allowed); at or below is a duplicate. There is no `RequestGap`: +/// a client that jumps its counter loses nothing but the skipped ids. +/// - **Replies are volatile.** Latest reply plus a small ring of displaced +/// ones, all in-memory refcounts. A duplicate whose reply aged out is +/// still refused execution ([`RequestStatus::AlreadyApplied`]). /// -/// Backs Register session, request contiguity, metadata-retry dedup, and -/// `NoSession`/`SessionTooLow` eviction. Partition plane is at-least-once; -/// `SendMessages` retries can re-commit at a new offset and consumers -/// dedup via message ID (`server_common::MessageDeduplicator`). +/// ## Plane /// -/// Do not add per-partition `ClientTable` or `(client_id, request)` dedup -/// on the partition side, that flips iggy's contract toward at-most-once. -/// See project memory `project_vsr_clients_table_integration`. +/// Metadata-plane today. The design spans planes (one logical table, +/// group-resident slices); partition-plane integration arrives once +/// partition prepares carry real `(session_id, request)` instead of the +/// transport id (data-plane request numbering, IGGY-137). Until then the +/// partition plane stays at-least-once with no dedup. /// /// ## Tracking /// -/// Committed state only, latest reply per client. In-flight state -/// (acks, subscribers, in-progress dedup) lives on [`crate::PipelineEntry`]. -/// Updated by `commit_reply` / `commit_register`. +/// Committed state only. In-flight state (acks, subscribers, in-progress +/// dedup) lives on [`crate::PipelineEntry`]. Updated by `commit_reply` / +/// `commit_register` in the apply path, so every replica of the group +/// derives an identical table from the committed log. /// /// ## Known gaps /// -/// - **Checkpoint serialization**: slot layout deterministic, encode/decode TODO. +/// - **Serialization**: encode/decode for rejoin slice-fetch and state +/// transfer TODO (IGGY-137). #[derive(Debug)] pub struct ClientTable { /// `None` = free slot. Deterministic iteration for eviction + serialization. @@ -156,52 +205,70 @@ impl ClientTable { } } - /// Check request against table. Session first, then request progression. - /// For Register, use [`check_register`]. + /// Check a request against the table. Epoch fence first, then the + /// watermark. For Register, use [`Self::check_register`]. + /// + /// `request_checksum` is the request's integrity stamp; zero (unstamped) + /// disables the reuse check. /// /// # Panics /// If index points to empty slot (invariant violation). #[must_use] - pub fn check_request(&self, client_id: u128, session: u64, request: u64) -> RequestStatus { + pub fn check_request( + &self, + client_id: u128, + epoch: u64, + request: u64, + request_checksum: u128, + ) -> RequestStatus { assert!(client_id != 0, "client_id 0 is reserved for internal use"); // Header validation guarantees both > 0 at wire layer. - debug_assert!(session > 0, "check_request: session must be > 0"); + debug_assert!(epoch > 0, "check_request: epoch must be > 0"); debug_assert!(request > 0, "check_request: request must be > 0"); - // Session check before request: wrong-session must be rejected even if - // (client_id, request) matches a correct-session pending entry. + // Epoch check before request: a fenced zombie must be rejected even + // if its request number would read as a clean duplicate. let Some(&slot_idx) = self.index.get(&client_id) else { return RequestStatus::NoSession; }; let entry = self.slots[slot_idx].as_ref().expect("index/slot mismatch"); - if session != entry.session { - return RequestStatus::SessionMismatch { - expected: entry.session, - received: session, + if epoch < entry.epoch { + return RequestStatus::Fenced { + current: entry.epoch, + received: epoch, }; } - - let committed_request = entry.reply.header().request; - - if request < committed_request { - return RequestStatus::Stale; + if epoch > entry.epoch { + return RequestStatus::EpochAhead { + current: entry.epoch, + received: epoch, + }; } - if request == committed_request { - return RequestStatus::Duplicate(entry.reply.clone()); + + if request > entry.watermark { + return RequestStatus::New; } - if request != committed_request + 1 { - return RequestStatus::RequestGap { - expected: committed_request + 1, - received: request, - }; + + if request == entry.watermark + && entry.watermark_checksum != 0 + && request_checksum != 0 + && entry.watermark_checksum != request_checksum + { + return RequestStatus::ChecksumMismatch { request }; } - RequestStatus::New + match entry.find_cached(request) { + Some(cached) => RequestStatus::Duplicate(cached.clone()), + None => RequestStatus::AlreadyApplied { + request, + watermark: entry.watermark, + }, + } } - /// Check register. Valid without existing session; returns - /// `AlreadyRegistered { session, cached_reply }`. + /// Check register. Valid without existing entry; returns + /// `AlreadyRegistered { epoch, cached_reply }` otherwise. /// /// Caller does in-flight dedup via `pipeline.has_message_from_client`. /// @@ -216,33 +283,25 @@ impl ClientTable { }; let entry = self.slots[slot_idx].as_ref().expect("index/slot mismatch"); RequestStatus::AlreadyRegistered { - session: entry.session, + epoch: entry.epoch, cached_reply: entry.reply.clone(), } } - /// Record committed register; create or update session. + /// Record a committed register: create the entry at epoch 1, or bump the + /// existing entry's epoch (rebind). /// - /// Session = `reply.header().commit`. Monotonic, deterministic. - /// Idempotent on same-session WAL replay. - /// - /// # Session mismatch (no panic, log + skip) - /// - /// - `existing.session > new`: stale WAL replay; newer slot is authoritative. - /// - `existing.session < new`: duplicate Register at different ops, - /// protocol violation; keep existing (other replicas may have agreed on it). - /// - /// Was `assert_eq!` pre-fix. `commit_journal` runs without the - /// `is_caught_up_primary` gate (it's what opens the gate), so a - /// malformed WAL or capacity-evict-then-reregister race could reach - /// here and panic the shard pump. + /// The epoch is minted HERE, in apply order, so it is deterministic + /// across replicas without reading any commit number. A rebind refreshes + /// `user_id` (the bind re-authenticated), replaces the latest reply with + /// the register reply (the displaced app reply moves into the ring), and + /// preserves the watermark - session resume keeps dedup history. /// /// Full table evicts oldest commit; `in_flight` protects pipeline /// holders, see [`Self::evict_oldest`]. /// /// # Panics - /// If `client_id == 0`, `session == 0`, or `client_id != reply.header().client`. - /// Session mismatch does NOT panic. + /// If `client_id == 0` or `client_id != reply.header().client`. pub fn commit_register<F>( &mut self, client_id: u128, @@ -260,55 +319,37 @@ impl ClientTable { reply.header().client ); - let session = reply.header().commit; - assert!(session > 0, "commit_register: session must be > 0"); - - let existing = self.index.get(&client_id).copied(); - - // Mismatch on re-register: log + skip, not panic. See doc above. - if let Some(slot_idx) = existing { - let slot = self.slots[slot_idx].as_ref().expect("index/slot mismatch"); - if slot.session != session { - tracing::warn!( - client_id, - existing_session = slot.session, - replay_session = session, - "commit_register: session mismatch (stale WAL replay or \ - duplicate Register at different ops); skipping update" - ); - return; - } - } - // Freeze once; later dedup-hit clones Arc-bump. let cached: CachedReply = CachedReply::from_message(reply); - // Update in place on re-register, else new slot. Reply-delivery - // channel lives on popped `PipelineEntry`, fired by commit caller - // after this returns, slot-first ordering, see `commit_reply`. - if let Some(slot_idx) = existing { - self.slots[slot_idx] - .as_mut() - .expect("index/slot mismatch") - .reply = cached; + if let Some(&slot_idx) = self.index.get(&client_id) { + let entry = self.slots[slot_idx].as_mut().expect("index/slot mismatch"); + entry.epoch += 1; + entry.user_id = user_id; + let displaced = std::mem::replace(&mut entry.reply, cached); + entry.push_ring(displaced); } else { if self.index.len() >= self.slots.len() { self.evict_oldest(&in_flight); } let slot_idx = self.first_free_slot().expect("eviction must free a slot"); self.slots[slot_idx] = Some(ClientEntry { - session, + epoch: 1, user_id, + watermark: REGISTER_REQUEST_ID, + watermark_checksum: 0, reply: cached, + ring: VecDeque::with_capacity(REPLY_RING_CAPACITY), }); self.index.insert(client_id, slot_idx); } } - /// Record committed reply, update in place. Client must be registered. + /// Record a committed reply: advance the watermark, cache the reply, + /// move the displaced one into the ring. /// - /// `session` is asserted against stored session to guard WAL replay - /// from clobbering a newer entry. + /// `epoch` is asserted against the entry to guard a mis-attributed apply + /// from clobbering a rebound session's state. /// /// Reply delivery is caller's job, `Sender` lives on the popped /// `PipelineEntry` ([`crate::PipelineEntry::take_reply_sender`]), @@ -319,18 +360,23 @@ impl ClientTable { /// ships; cache skipped; client gets `NoSession` next request. /// /// # Panics - /// On session mismatch or commit/request regression. Missing client + /// On epoch mismatch or commit/watermark regression. Missing client /// does NOT panic. - pub fn commit_reply(&mut self, client_id: u128, session: u64, reply: Message<ReplyHeader>) { + pub fn commit_reply(&mut self, client_id: u128, epoch: u64, reply: Message<ReplyHeader>) { assert!(client_id != 0, "client_id 0 is reserved for internal use"); let new_header = reply.header(); let new_client = new_header.client; let new_request = new_header.request; let new_commit = new_header.commit; + let new_checksum = new_header.request_checksum; assert_eq!( client_id, new_client, "commit_reply: client_id mismatch (arg={client_id}, header={new_client})", ); + debug_assert!( + new_request > REGISTER_REQUEST_ID, + "commit_reply: register replies go through commit_register" + ); let Some(&slot_idx) = self.index.get(&client_id) else { // Evicted between prepare and commit (WAL replay or @@ -345,33 +391,40 @@ impl ClientTable { return; }; - let slot = self.slots[slot_idx].as_ref().expect("index/slot mismatch"); - let slot_header = slot.reply.header(); - let slot_commit = slot_header.commit; - let slot_request = slot_header.request; + let entry = self.slots[slot_idx].as_mut().expect("index/slot mismatch"); assert_eq!( - slot.session, session, - "commit_reply: session mismatch for client {client_id}: \ - entry={}, prepare={session}", - slot.session + entry.epoch, epoch, + "commit_reply: epoch mismatch for client {client_id}: \ + entry={}, prepare={epoch}", + entry.epoch ); + let latest_commit = entry.reply.header().commit; assert!( - new_commit >= slot_commit, - "commit_reply: commit regression for client {client_id}: {slot_commit} -> {new_commit}", + new_commit >= latest_commit, + "commit_reply: commit regression for client {client_id}: {latest_commit} -> {new_commit}", ); assert!( - new_request >= slot_request, - "commit_reply: request regression for client {client_id}: {slot_request} -> {new_request}", + new_request >= entry.watermark, + "commit_reply: watermark regression for client {client_id}: {} -> {new_request}", + entry.watermark ); // Freeze once; later dedup-hit clones Arc-bump. - self.slots[slot_idx] - .as_mut() - .expect("index/slot mismatch") - .reply = CachedReply::from_message(reply); + let cached = CachedReply::from_message(reply); + if new_request == entry.watermark { + // Same request re-committed (WAL replay shape): replace in + // place, never push the stale twin into the ring - two cached + // replies for one request number would make lookups ambiguous. + entry.reply = cached; + } else { + let displaced = std::mem::replace(&mut entry.reply, cached); + entry.push_ring(displaced); + entry.watermark = new_request; + } + entry.watermark_checksum = new_checksum; } - /// Remove a client session and cached reply. + /// Remove a client session and cached replies. /// /// **LOCAL ONLY -- does NOT replicate.** Two correct call sites: /// @@ -412,9 +465,9 @@ impl ClientTable { /// state -> identical choice. `commit_journal` catch-up has empty pipeline, /// so `in_flight` returns `false` everywhere, matches pre-fix policy. /// - /// **Metadata caveat**: pre-checkpoint, eviction breaks at-most-once - /// for the evicted client, next metadata retry treated as `New`. - /// Partition plane unaffected (at-least-once, doesn't use this table). + /// **Caveat**: eviction erases the evicted session's watermark, so its + /// next retry is treated as `New` (re-executes). Bounded by table + /// capacity; the op-TTL + slice persistence work (IGGY-137) shrinks it. fn evict_oldest<F>(&mut self, in_flight: &F) where F: Fn(u128) -> bool, @@ -453,7 +506,7 @@ impl ClientTable { self.slots.iter().position(Option::is_none) } - /// Cached reply for a client (duplicate re-sends). + /// Latest cached reply for a client. /// /// Borrow avoids Arc bump for header-only inspection. Wire-senders /// `.clone()` (Arc bump) then `.into_wire_bytes()`. @@ -463,11 +516,21 @@ impl ClientTable { self.slots[slot_idx].as_ref().map(|entry| &entry.reply) } - /// Session number for a registered client. + /// Fence epoch for a registered client. This is the u64 the register + /// reply hands the client and the wire `session` field carries back. #[must_use] - pub fn get_session(&self, client_id: u128) -> Option<u64> { + pub fn get_epoch(&self, client_id: u128) -> Option<u64> { let &slot_idx = self.index.get(&client_id)?; - self.slots[slot_idx].as_ref().map(|entry| entry.session) + self.slots[slot_idx].as_ref().map(|entry| entry.epoch) + } + + /// Committed-request watermark for a registered client. A (re)bind reply + /// surfaces this so a restarted client resumes numbering at + /// `watermark + 1` instead of silently colliding below it. + #[must_use] + pub fn get_watermark(&self, client_id: u128) -> Option<u64> { + let &slot_idx = self.index.get(&client_id)?; + self.slots[slot_idx].as_ref().map(|entry| entry.watermark) } /// Acting user id captured when the client registered. @@ -484,6 +547,33 @@ impl ClientTable { } } +impl ClientEntry { + /// Cached reply whose `request` matches, latest first then the ring + /// (newest displaced entries sit at the back; scan order is irrelevant + /// because request numbers in the ring are unique). + fn find_cached(&self, request: u64) -> Option<&CachedReply> { + if self.reply.header().request == request { + return Some(&self.reply); + } + self.ring + .iter() + .find(|cached| cached.header().request == request) + } + + /// Retain a displaced reply for below-watermark duplicates. Register + /// replies never enter: `request == REGISTER_REQUEST_ID` can never match + /// a `check_request` lookup (wire validation enforces `request > 0`). + fn push_ring(&mut self, displaced: CachedReply) { + if displaced.header().request == REGISTER_REQUEST_ID { + return; + } + if self.ring.len() == REPLY_RING_CAPACITY { + self.ring.pop_front(); + } + self.ring.push_back(displaced); + } +} + #[cfg(test)] mod tests { use super::*; @@ -512,6 +602,15 @@ mod tests { } fn make_reply_for(client: u128, request: u64, commit: u64) -> Message<ReplyHeader> { + make_reply_with_checksum(client, request, commit, 0) + } + + fn make_reply_with_checksum( + client: u128, + request: u64, + commit: u64, + request_checksum: u128, + ) -> Message<ReplyHeader> { let header_size = std::mem::size_of::<ReplyHeader>(); let mut msg = Message::<ReplyHeader>::new(header_size); let header = bytemuck::checked::try_from_bytes_mut::<ReplyHeader>( @@ -522,6 +621,7 @@ mod tests { client, request, commit, + request_checksum, command: Command2::Reply, operation: Operation::SendMessages, ..ReplyHeader::default() @@ -534,30 +634,59 @@ mod tests { |_| false } - /// Register client 1 at commit 10. Returns (table, session=10). + /// Register client 1 (register commit stamped at op 10). Returns + /// (table, epoch=1). fn table_with_client() -> (ClientTable, u64) { let mut table = ClientTable::new(10); - let session = 10; - table.commit_register( - 1, - TEST_USER_ID, - make_register_reply(1, session), - no_in_flight(), - ); - (table, session) + table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10), no_in_flight()); + (table, 1) } // Registration tests #[test] - fn register_creates_session() { + fn register_mints_epoch_one() { let mut table = ClientTable::new(10); table.commit_register(1, TEST_USER_ID, make_register_reply(1, 42), no_in_flight()); - assert_eq!(table.get_session(1), Some(42)); + assert_eq!(table.get_epoch(1), Some(1)); + assert_eq!(table.get_watermark(1), Some(0)); assert_eq!(table.get_user_id(1), Some(TEST_USER_ID)); assert_eq!(table.count(), 1); } + // Re-register = rebind: epoch bumps, watermark (dedup history) survives. + #[test] + fn reregister_bumps_epoch_and_preserves_watermark() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 5, 15)); + assert_eq!(table.get_watermark(1), Some(5)); + + table.commit_register(1, TEST_USER_ID, make_register_reply(1, 20), no_in_flight()); + assert_eq!(table.get_epoch(1), Some(2), "rebind mints the next epoch"); + assert_eq!( + table.get_watermark(1), + Some(5), + "session resume keeps dedup history" + ); + assert_eq!(table.count(), 1); + + // The displaced app reply moved into the ring: the watermark request + // still answers with its original bytes under the new epoch. + match table.check_request(1, 2, 5, 0) { + RequestStatus::Duplicate(cached) => assert_eq!(cached.header().request, 5), + other => panic!("expected Duplicate from ring, got {other:?}"), + } + } + + // A rebind re-authenticates; the fresh register's user wins. + #[test] + fn reregister_refreshes_user_id() { + let mut table = ClientTable::new(10); + table.commit_register(1, 11, make_register_reply(1, 10), no_in_flight()); + table.commit_register(1, 22, make_register_reply(1, 20), no_in_flight()); + assert_eq!(table.get_user_id(1), Some(22)); + } + // Each entry keeps the user id it registered with; lookups are per-client. #[test] fn register_stores_user_id() { @@ -581,16 +710,15 @@ mod tests { #[test] fn check_register_already_registered() { - let (table, session) = table_with_client(); + let (table, epoch) = table_with_client(); match table.check_register(1) { RequestStatus::AlreadyRegistered { - session: s, + epoch: e, cached_reply, } => { - assert_eq!(s, session); + assert_eq!(e, epoch); // Cached reply IS the register reply, preflight replays it. assert_eq!(cached_reply.header().request, REGISTER_REQUEST_ID); - assert_eq!(cached_reply.header().commit, session); } other => panic!("expected AlreadyRegistered, got {other:?}"), } @@ -598,17 +726,17 @@ mod tests { #[test] fn check_register_already_registered_after_progress() { - let (mut table, session) = table_with_client(); + let (mut table, epoch) = table_with_client(); // Client progresses past registration. - table.commit_reply(1, 10, make_reply_for(1, 1, 11)); - table.commit_reply(1, 10, make_reply_for(1, 2, 12)); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); + table.commit_reply(1, epoch, make_reply_for(1, 2, 12)); // Cached reply is now latest app reply; preflight must silent-drop. match table.check_register(1) { RequestStatus::AlreadyRegistered { - session: s, + epoch: e, cached_reply, } => { - assert_eq!(s, session); + assert_eq!(e, epoch); assert_eq!( cached_reply.header().request, 2, @@ -619,50 +747,124 @@ mod tests { } } - // Session validation tests + // Epoch fence tests #[test] fn check_request_no_session() { let table = ClientTable::new(10); - // Not registered: valid session/request but no entry. + // Not registered: valid epoch/request but no entry. assert!(matches!( - table.check_request(1, 99, 1), + table.check_request(1, 99, 1, 0), RequestStatus::NoSession )); } + // Zombie fencing: requests stamped with a pre-rebind epoch are terminal. #[test] - fn check_request_session_mismatch() { - let (table, session) = table_with_client(); - match table.check_request(1, session + 1, 1) { - RequestStatus::SessionMismatch { expected, received } => { - assert_eq!(expected, session); - assert_eq!(received, session + 1); + fn check_request_stale_epoch_is_fenced() { + let (mut table, _) = table_with_client(); + table.commit_register(1, TEST_USER_ID, make_register_reply(1, 20), no_in_flight()); + assert_eq!(table.get_epoch(1), Some(2)); + match table.check_request(1, 1, 1, 0) { + RequestStatus::Fenced { current, received } => { + assert_eq!(current, 2); + assert_eq!(received, 1); } - other => panic!("expected SessionMismatch, got {other:?}"), + other => panic!("expected Fenced, got {other:?}"), } } + // Epochs are only handed out by register replies; a newer-than-minted + // epoch is a client bug, distinct from the zombie case. #[test] - fn check_request_correct_session_new() { - let (mut table, session) = table_with_client(); - table.commit_reply(1, 10, make_reply_for(1, 1, 11)); + fn check_request_future_epoch_is_client_bug() { + let (table, epoch) = table_with_client(); + match table.check_request(1, epoch + 1, 1, 0) { + RequestStatus::EpochAhead { current, received } => { + assert_eq!(current, epoch); + assert_eq!(received, epoch + 1); + } + other => panic!("expected EpochAhead, got {other:?}"), + } + } + + // Watermark tests + + #[test] + fn check_request_above_watermark_is_new() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); + assert!(matches!( + table.check_request(1, epoch, 2, 0), + RequestStatus::New + )); + } + + // No contiguity requirement: a jump past the watermark executes. The + // watermark records the highest committed request, not a sequence. + #[test] + fn check_request_jump_above_watermark_is_new() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); assert!(matches!( - table.check_request(1, session, 2), + table.check_request(1, epoch, 9, 0), RequestStatus::New )); + // And committing the jump moves the watermark to it. + table.commit_reply(1, epoch, make_reply_for(1, 9, 12)); + assert_eq!(table.get_watermark(1), Some(9)); } #[test] - fn check_request_duplicate_after_commit() { - let (mut table, session) = table_with_client(); - table.commit_reply(1, 10, make_reply_for(1, 1, 11)); - match table.check_request(1, session, 1) { + fn check_request_duplicate_at_watermark() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); + match table.check_request(1, epoch, 1, 0) { RequestStatus::Duplicate(cached) => assert_eq!(cached.header().request, 1), other => panic!("expected Duplicate, got {other:?}"), } } + // Below-watermark duplicate with the original still in the ring answers + // with the original bytes. + #[test] + fn check_request_below_watermark_hits_ring() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); + table.commit_reply(1, epoch, make_reply_for(1, 2, 12)); + match table.check_request(1, epoch, 1, 0) { + RequestStatus::Duplicate(cached) => { + assert_eq!(cached.header().request, 1, "original reply, not latest"); + assert_eq!(cached.header().commit, 11, "original commit op"); + } + other => panic!("expected Duplicate from ring, got {other:?}"), + } + } + + // Below-watermark duplicate whose reply aged out of the ring is refused + // execution with nothing to replay. + #[test] + fn check_request_below_watermark_past_ring_is_already_applied() { + let (mut table, epoch) = table_with_client(); + // Requests 1..=6: request 1's reply is displaced beyond the ring + // (capacity 4 holds 2,3,4,5 once 6 is latest). + for request in 1..=6u64 { + table.commit_reply(1, epoch, make_reply_for(1, request, 10 + request)); + } + match table.check_request(1, epoch, 1, 0) { + RequestStatus::AlreadyApplied { request, watermark } => { + assert_eq!(request, 1); + assert_eq!(watermark, 6); + } + other => panic!("expected AlreadyApplied, got {other:?}"), + } + // The oldest retained entry still answers. + match table.check_request(1, epoch, 2, 0) { + RequestStatus::Duplicate(cached) => assert_eq!(cached.header().request, 2), + other => panic!("expected Duplicate, got {other:?}"), + } + } + // Dedup across view change. Backup inherits client_table via // commit_journal; on failover, retry must return ORIGINAL cached reply // (same request, same commit op), no re-execution. Pipeline state is @@ -670,10 +872,10 @@ mod tests { // Simulator test covers end-to-end; this is the unit invariant. #[test] fn duplicate_survives_view_change_reset() { - let (mut table, session) = table_with_client(); - table.commit_reply(1, session, make_reply_for(1, 1, 11)); + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); - match table.check_request(1, session, 1) { + match table.check_request(1, epoch, 1, 0) { RequestStatus::Duplicate(cached) => { assert_eq!(cached.header().client, 1, "original client_id"); assert_eq!(cached.header().request, 1, "ORIGINAL request, not re-issue"); @@ -687,50 +889,78 @@ mod tests { } } + // Checksum tests + + // Same request id, different request bytes: returning the cached reply + // would answer the wrong request. Refused loudly. #[test] - fn check_request_stale() { - let (mut table, session) = table_with_client(); - table.commit_reply(1, 10, make_reply_for(1, 5, 15)); + fn check_request_checksum_mismatch_at_watermark() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_with_checksum(1, 1, 11, 0xAA)); + match table.check_request(1, epoch, 1, 0xBB) { + RequestStatus::ChecksumMismatch { request } => assert_eq!(request, 1), + other => panic!("expected ChecksumMismatch, got {other:?}"), + } + // Matching stamp replays. assert!(matches!( - table.check_request(1, session, 3), - RequestStatus::Stale + table.check_request(1, epoch, 1, 0xAA), + RequestStatus::Duplicate(_) )); } + // Integrity fields are zeroed on the wire today; a zero on either side + // must not trip the mismatch (rollout compatibility). #[test] - fn check_request_gap_rejected() { - let (mut table, session) = table_with_client(); - table.commit_reply(1, 10, make_reply_for(1, 1, 11)); - // Skip from 1 to 3, reject. - match table.check_request(1, session, 3) { - RequestStatus::RequestGap { expected, received } => { - assert_eq!(expected, 2); - assert_eq!(received, 3); - } - other => panic!("expected RequestGap, got {other:?}"), - } + fn check_request_zero_checksum_disables_comparison() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_with_checksum(1, 1, 11, 0xAA)); + assert!(matches!( + table.check_request(1, epoch, 1, 0), + RequestStatus::Duplicate(_) + )); + + table.commit_reply(1, epoch, make_reply_for(1, 2, 12)); // stored zero + assert!(matches!( + table.check_request(1, epoch, 2, 0xBB), + RequestStatus::Duplicate(_) + )); } // Commit tests #[test] fn commit_caches_reply() { - let (mut table, _) = table_with_client(); - table.commit_reply(1, 10, make_reply_for(1, 1, 11)); + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); let cached = table.get_reply(1).expect("should have cached reply"); assert_eq!(cached.header().request, 1); } #[test] - fn commit_updates_preserves_session() { - let (mut table, session) = table_with_client(); - table.commit_reply(1, 10, make_reply_for(1, 1, 11)); - table.commit_reply(1, 10, make_reply_for(1, 2, 12)); + fn commit_updates_preserves_epoch() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); + table.commit_reply(1, epoch, make_reply_for(1, 2, 12)); assert_eq!(table.get_reply(1).unwrap().header().request, 2); - assert_eq!(table.get_session(1), Some(session)); + assert_eq!(table.get_epoch(1), Some(epoch)); assert_eq!(table.count(), 1); } + // Same request re-committed (WAL replay shape): replace in place, no + // ring push - two cached replies for one request number would make + // duplicate lookups ambiguous. + #[test] + fn commit_reply_same_request_replaces_in_place() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); + table.commit_reply(1, epoch, make_reply_for(1, 1, 11)); + assert_eq!(table.get_watermark(1), Some(1)); + match table.check_request(1, epoch, 1, 0) { + RequestStatus::Duplicate(cached) => assert_eq!(cached.header().request, 1), + other => panic!("expected Duplicate, got {other:?}"), + } + } + // Eviction tests #[test] @@ -872,62 +1102,53 @@ mod tests { // Edge cases - #[test] - fn commit_register_idempotent_on_replay() { - let mut table = ClientTable::new(10); - table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10), no_in_flight()); - // Same client_id + session = idempotent (WAL replay). - table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10), no_in_flight()); - assert_eq!(table.get_session(1), Some(10)); - assert_eq!(table.count(), 1); - } - - // Re-register with mismatched session must not panic shard pump. - // Stale WAL replay or duplicate Register at different ops; either way - // log + skip, existing slot stays authoritative. - #[test] - fn commit_register_different_session_logs_and_skips() { - let mut table = ClientTable::new(10); - table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10), no_in_flight()); - // existing=10, replay=20. - table.commit_register(1, TEST_USER_ID, make_register_reply(1, 20), no_in_flight()); - assert_eq!(table.get_session(1), Some(10), "first session stays"); - // Smaller replay session: same skip. - table.commit_register(1, TEST_USER_ID, make_register_reply(1, 5), no_in_flight()); - assert_eq!(table.get_session(1), Some(10)); - } - // commit_reply for unregistered/evicted client must not panic; // wire reply still ships, cache silently skipped. #[test] fn commit_reply_for_unregistered_client_is_noop() { let mut table = ClientTable::new(10); // No register: index has no entry. - table.commit_reply(1, 10, make_reply_for(1, 1, 10)); + table.commit_reply(1, 1, make_reply_for(1, 1, 10)); assert!(table.get_reply(1).is_none(), "no entry must be created"); assert_eq!(table.count(), 0); } #[test] - #[should_panic(expected = "session mismatch")] - fn commit_reply_wrong_session_panics() { - let (mut table, _session) = table_with_client(); - // Registered session=10, commit session=99. + #[should_panic(expected = "epoch mismatch")] + fn commit_reply_wrong_epoch_panics() { + let (mut table, _epoch) = table_with_client(); + // Entry epoch=1, commit claims epoch=99. table.commit_reply(1, 99, make_reply_for(1, 1, 11)); } #[test] - fn different_clients_independent_sessions() { + #[should_panic(expected = "watermark regression")] + fn commit_reply_watermark_regression_panics() { + let (mut table, epoch) = table_with_client(); + table.commit_reply(1, epoch, make_reply_for(1, 5, 15)); + table.commit_reply(1, epoch, make_reply_for(1, 3, 16)); + } + + #[test] + fn different_clients_independent_epochs() { let mut table = ClientTable::new(10); table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10), no_in_flight()); table.commit_register(2, TEST_USER_ID, make_register_reply(2, 20), no_in_flight()); - assert_eq!(table.get_session(1), Some(10)); - assert_eq!(table.get_session(2), Some(20)); - assert!(matches!(table.check_request(1, 10, 1), RequestStatus::New)); - assert!(matches!(table.check_request(2, 20, 1), RequestStatus::New)); + // Rebind client 2 only. + table.commit_register(2, TEST_USER_ID, make_register_reply(2, 30), no_in_flight()); + assert_eq!(table.get_epoch(1), Some(1)); + assert_eq!(table.get_epoch(2), Some(2)); + assert!(matches!( + table.check_request(1, 1, 1, 0), + RequestStatus::New + )); + assert!(matches!( + table.check_request(2, 2, 1, 0), + RequestStatus::New + )); assert!(matches!( - table.check_request(1, 20, 1), - RequestStatus::SessionMismatch { .. } + table.check_request(2, 1, 1, 0), + RequestStatus::Fenced { .. } )); } } diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs index 6191f0896..fdc3878bb 100644 --- a/core/consensus/src/metadata_helpers.rs +++ b/core/consensus/src/metadata_helpers.rs @@ -56,17 +56,23 @@ pub enum PreflightOutcome { Drop, } -/// Request preflight (metadata only): session validation, dedup, in-flight check. +/// Request preflight (metadata only): epoch fence, watermark dedup, +/// in-flight check. /// /// Pure decision -- emits no frames (see [`PreflightOutcome`]). Callers turn /// the outcome into a reply: the home-shard path resends by transport id, the /// message-plane paths fall back to [`apply_preflight_consensus_plane`]. +/// +/// `session` is the wire `session` field, which carries the entry's fence +/// epoch; `request_checksum` is the request's integrity stamp (zero = +/// unstamped, disables the reuse check). pub fn request_preflight<B, P>( consensus: &VsrConsensus<B, P>, client_table: &RefCell<ClientTable>, client_id: u128, session: u64, request: u64, + request_checksum: u128, ) -> PreflightOutcome where B: MessageBus, @@ -107,7 +113,7 @@ where let status = client_table .borrow() - .check_request(client_id, session, request); + .check_request(client_id, session, request, request_checksum); match status { // Frozen-backed cache -> refcount handoff to the home shard, no copy. RequestStatus::Duplicate(cached_reply) => { @@ -116,29 +122,45 @@ where // Session evicted under capacity pressure. SAFETY: catch-up gate makes // this replica authoritative for session truth. RequestStatus::NoSession => PreflightOutcome::Evict(EvictionReason::NoSession), - RequestStatus::SessionMismatch { expected, received } => { - // expected > received: stale session (rotated post-eviction) -> terminal eviction. - // expected < received: client bug; silent drop, log. - // SAFETY: catch-up gate makes this replica authoritative. - if expected > received { - PreflightOutcome::Evict(EvictionReason::SessionTooLow) - } else { - // Catch-up gate rules out network race; newer-than-issued - // session = client bug. Error log, no eviction (transient bug - // must not kill session), no rate limit (per-event). - tracing::error!( - client_id, - expected, - received, - "request_preflight: ignoring newer session (client bug)" - ); - PreflightOutcome::Drop - } + // Zombie holdover from before a re-register: terminal for that + // holder. SAFETY: catch-up gate makes this replica authoritative. + RequestStatus::Fenced { current, received } => { + tracing::debug!( + client_id, + current, + received, + "request_preflight: fencing stale-epoch request" + ); + PreflightOutcome::Evict(EvictionReason::SessionTooLow) + } + // Catch-up gate rules out network race; an epoch newer than any this + // table minted = client bug. Error log, no eviction (transient bug + // must not kill the session), no rate limit (per-event). + RequestStatus::EpochAhead { current, received } => { + tracing::error!( + client_id, + current, + received, + "request_preflight: ignoring future epoch (client bug)" + ); + PreflightOutcome::Drop + } + // Same request id, different request bytes: replaying the cached + // reply would answer the wrong request, and re-executing would + // double-apply. Loud drop; the client must fix its numbering. + RequestStatus::ChecksumMismatch { request } => { + tracing::error!( + client_id, + request, + "request_preflight: request id reused for a different operation (client bug)" + ); + PreflightOutcome::Drop + } + // Applied once, original reply aged out of the ring: refuse + // re-execution, nothing to replay. Silent drop. + RequestStatus::AlreadyApplied { .. } | RequestStatus::AlreadyRegistered { .. } => { + PreflightOutcome::Drop } - // Client bug; recovered by client retry. Silent drop. - RequestStatus::Stale - | RequestStatus::RequestGap { .. } - | RequestStatus::AlreadyRegistered { .. } => PreflightOutcome::Drop, RequestStatus::New => PreflightOutcome::Dispatch, } } @@ -213,8 +235,9 @@ where // Catch-up gate: new primary may have inherited Register(client, op=N) // committed in WAL but not yet applied. Without gate, check_register - // returns New -> fresh register -> two register entries -> commit_register's - // session-equality assert panics on replay. SDK retry recovers post-catch-up. + // returns New -> a second register commits -> the epoch bumps past the + // one the first register's reply handed the client, fencing a live + // client for no reason. SDK retry recovers post-catch-up. if !is_caught_up_primary(consensus) { tracing::debug!( client_id, @@ -231,7 +254,7 @@ where let status = client_table.borrow().check_register(client_id); match status { RequestStatus::AlreadyRegistered { - session, + epoch, cached_reply, } => { // cached.request == REGISTER_REQUEST_ID: replay cached bytes @@ -247,13 +270,13 @@ where .await; tracing::debug!( client_id, - session, + epoch, "register_preflight: replayed cached register reply" ); } else { tracing::debug!( client_id, - session, + epoch, cached_request = cached_reply.header().request, "register_preflight: retry past register, drop" ); @@ -368,8 +391,9 @@ fn build_eviction_from_header(header: EvictionHeader) -> Message<EvictionHeader> /// `NoSession`/`SessionTooLow` against stale table erases live clients. /// - **Dispatch**: primary with `commit_min < commit_max` may hold an /// inherited `Register(client, op=N)` in WAL but not yet applied. -/// Fresh `Register(client, op=M>N)` panics `commit_register`'s -/// session-equality assert on replay. +/// Admitting a fresh Register commits a second register, bumping the +/// epoch past the one the inherited register's reply handed the client +/// and fencing a live client for no reason. /// /// `false` -> caller silent-drops; client retry lands on peer or here /// post-catch-up. @@ -479,10 +503,10 @@ mod tests { let client_table = fresh_client_table(); let client_id: u128 = 0xBEEF; - let session: u64 = 17; + let register_commit: u64 = 17; // Cached reply IS the register reply (request == REGISTER_REQUEST_ID). - let initial_reply = synthesize_register_reply(&consensus, client_id, session); + let initial_reply = synthesize_register_reply(&consensus, client_id, register_commit); let original_checksum = initial_reply.header().checksum; client_table .borrow_mut() @@ -505,7 +529,7 @@ mod tests { "must be original cached bytes, not fresh synthesis" ); assert_eq!(header.request, REGISTER_REQUEST_ID); - assert_eq!(header.commit, session); + assert_eq!(header.commit, register_commit); } // Past-register retry: silent drop, no replay, no eviction. Read-timeout @@ -517,18 +541,18 @@ mod tests { let client_table = fresh_client_table(); let client_id: u128 = 0xBEEF; - let session: u64 = 17; + let epoch: u64 = 1; - let initial_reply = synthesize_register_reply(&consensus, client_id, session); + let initial_reply = synthesize_register_reply(&consensus, client_id, 17); client_table .borrow_mut() .commit_register(client_id, ACTING_USER_ID, initial_reply, |_| false); // SendMessages commits -> cached is no longer the register reply. - let app_reply = synthesize_send_messages_reply(&consensus, client_id, session, 1, 18); + let app_reply = synthesize_send_messages_reply(&consensus, client_id, 1, 18); client_table .borrow_mut() - .commit_reply(client_id, session, app_reply); + .commit_reply(client_id, epoch, app_reply); let result = futures::executor::block_on(register_preflight(&consensus, &client_table, client_id)); @@ -556,8 +580,9 @@ mod tests { &consensus, &client_table, client_id, - 10, // session + 10, // epoch (wire session field) 1, // request + 0, // request_checksum (unstamped) ), client_id, )); @@ -576,29 +601,33 @@ mod tests { assert_eq!(header.client, client_id); } - // Stale session (post capacity-evict + re-register): terminal SessionTooLow. + // Zombie fencing: a request stamped with a pre-rebind epoch gets a + // terminal SessionTooLow eviction. #[test] - fn request_preflight_session_too_low_evicts_client() { + fn request_preflight_stale_epoch_evicts_client() { let consensus = VsrConsensus::new(1, 0, 3, 0, ClientSpyBus::new(), LocalPipeline::new()); consensus.init(); let client_table = fresh_client_table(); let client_id: u128 = 0xBEEF; - let real_session: u64 = 99; - // Slot at session 99. - let initial_reply = synthesize_register_reply(&consensus, client_id, real_session); + // Register, then rebind: entry epoch is now 2. + let initial_reply = synthesize_register_reply(&consensus, client_id, 17); client_table .borrow_mut() .commit_register(client_id, ACTING_USER_ID, initial_reply, |_| false); + let rebind_reply = synthesize_register_reply(&consensus, client_id, 25); + client_table + .borrow_mut() + .commit_register(client_id, ACTING_USER_ID, rebind_reply, |_| false); - // Older retry (17 < 99): stale-session case. + // Zombie still stamping epoch 1: fenced. let result = futures::executor::block_on(apply_preflight_consensus_plane( &consensus, - request_preflight(&consensus, &client_table, client_id, 17, 1), + request_preflight(&consensus, &client_table, client_id, 1, 1, 0), client_id, )); - assert!(!result, "SessionMismatch short-circuits"); + assert!(!result, "Fenced short-circuits"); let sends = consensus.message_bus().client_sends.borrow(); assert_eq!(sends.len(), 1, "one Eviction"); @@ -611,36 +640,33 @@ mod tests { assert_eq!(header.client, client_id); } - // Newer-than-cluster session: sessions monotonic; healthy SDK can't reach - // this. Client bug -> silent drop, no eviction. + // Newer-than-minted epoch: epochs are only handed out by register + // replies; healthy SDK can't reach this. Client bug -> silent drop, no + // eviction. #[test] - fn request_preflight_session_too_high_is_silent_drop() { + fn request_preflight_future_epoch_is_silent_drop() { let consensus = VsrConsensus::new(1, 0, 3, 0, ClientSpyBus::new(), LocalPipeline::new()); consensus.init(); let client_table = fresh_client_table(); let client_id: u128 = 0xBEEF; - let real_session: u64 = 17; - // Slot at session 17. - let initial_reply = synthesize_register_reply(&consensus, client_id, real_session); + // Entry at epoch 1. + let initial_reply = synthesize_register_reply(&consensus, client_id, 17); client_table .borrow_mut() .commit_register(client_id, ACTING_USER_ID, initial_reply, |_| false); - // Client claims newer session (99 > 17), client bug. + // Client claims epoch 99 (> 1), client bug. let result = futures::executor::block_on(apply_preflight_consensus_plane( &consensus, - request_preflight(&consensus, &client_table, client_id, 99, 1), + request_preflight(&consensus, &client_table, client_id, 99, 1, 0), client_id, )); - assert!(!result, "SessionMismatch short-circuits"); + assert!(!result, "EpochAhead short-circuits"); let sends = consensus.message_bus().client_sends.borrow(); - assert!( - sends.is_empty(), - "newer-session mismatch must be silent drop" - ); + assert!(sends.is_empty(), "future epoch must be silent drop"); } // Backups never send NoSession: their ClientTable lags. Without gate, @@ -661,8 +687,9 @@ mod tests { &consensus, &client_table, client_id, - 10, // session + 10, // epoch (wire session field) 1, // request + 0, // request_checksum (unstamped) ), client_id, )); @@ -675,41 +702,75 @@ mod tests { ); } - // Stale + RequestGap: silent drop, no eviction. + // Below-watermark retry whose reply is still cached: replayed, not + // re-executed and not dropped. #[test] - fn request_preflight_stale_is_silent_drop() { + fn request_preflight_below_watermark_replays_ring_hit() { let consensus = VsrConsensus::new(1, 0, 3, 0, ClientSpyBus::new(), LocalPipeline::new()); consensus.init(); let client_table = fresh_client_table(); let client_id: u128 = 0xABCD; - let session: u64 = 5; + let epoch: u64 = 1; - let initial_reply = synthesize_register_reply(&consensus, client_id, session); + let initial_reply = synthesize_register_reply(&consensus, client_id, 5); client_table .borrow_mut() .commit_register(client_id, ACTING_USER_ID, initial_reply, |_| false); - // Cache at request 5 -> request 3 is stale. - let advanced = synthesize_send_messages_reply(&consensus, client_id, session, 5, 100); - client_table - .borrow_mut() - .commit_reply(client_id, session, advanced); + for (request, commit) in [(3u64, 98u64), (5, 100)] { + let reply = synthesize_send_messages_reply(&consensus, client_id, request, commit); + client_table + .borrow_mut() + .commit_reply(client_id, epoch, reply); + } let result = futures::executor::block_on(apply_preflight_consensus_plane( &consensus, - request_preflight(&consensus, &client_table, client_id, session, 3), // stale + request_preflight(&consensus, &client_table, client_id, epoch, 3, 0), client_id, )); - assert!(!result); + assert!(!result, "duplicate short-circuits"); let sends = consensus.message_bus().client_sends.borrow(); - assert!(sends.is_empty(), "stale = silent drop"); + assert_eq!(sends.len(), 1, "ring hit replays the original reply"); + let header = bytemuck::checked::try_from_bytes::<ReplyHeader>( + &sends[0].1.as_slice()[..HEADER_SIZE], + ) + .expect("valid ReplyHeader"); + assert_eq!(header.request, 3, "original reply for the retried request"); + } + + // Watermark jump: request numbers above the watermark dispatch even + // when non-contiguous (there is no RequestGap). + #[test] + fn request_preflight_jump_above_watermark_dispatches() { + let consensus = VsrConsensus::new(1, 0, 3, 0, ClientSpyBus::new(), LocalPipeline::new()); + consensus.init(); + let client_table = fresh_client_table(); + + let client_id: u128 = 0xABCD; + let epoch: u64 = 1; + + let initial_reply = synthesize_register_reply(&consensus, client_id, 5); + client_table + .borrow_mut() + .commit_register(client_id, ACTING_USER_ID, initial_reply, |_| false); + let advanced = synthesize_send_messages_reply(&consensus, client_id, 2, 99); + client_table + .borrow_mut() + .commit_reply(client_id, epoch, advanced); + + let outcome = request_preflight(&consensus, &client_table, client_id, epoch, 9, 0); + assert!( + matches!(outcome, PreflightOutcome::Dispatch), + "watermark jump must dispatch" + ); } - // Catch-up gate prevents WAL-replay race in commit_register: primary - // with commit_min < commit_max may hold inherited Register(client) in - // WAL not yet applied. Fresh Register would panic session-equality - // assert on replay. + // Catch-up gate prevents the WAL-replay race in commit_register: primary + // with commit_min < commit_max may hold an inherited Register(client) in + // WAL not yet applied. A fresh Register would commit a second register + // and bump the epoch past the inherited reply's, fencing a live client. #[test] fn register_preflight_silently_drops_when_behind_on_commits() { let consensus = VsrConsensus::new(1, 0, 3, 0, ClientSpyBus::new(), LocalPipeline::new()); @@ -775,11 +836,13 @@ mod tests { // Fixture: register reply mirroring `commit_register` storage. Test-only // production replays cached via `AlreadyRegistered { cached_reply }`. + // `register_commit` stamps the reply's op/commit (recency for eviction + // ordering); the entry's epoch is minted by the table, not read from it. #[allow(clippy::cast_possible_truncation)] fn synthesize_register_reply<B, P>( consensus: &VsrConsensus<B, P>, client_id: u128, - session: u64, + register_commit: u64, ) -> Message<ReplyHeader> where B: MessageBus, @@ -798,8 +861,8 @@ mod tests { command: Command2::Reply, replica: consensus.replica(), client: client_id, - op: session, - commit: session, + op: register_commit, + commit: register_commit, request: REGISTER_REQUEST_ID, operation: Operation::Register, ..ReplyHeader::default() @@ -807,12 +870,11 @@ mod tests { msg } - // SendMessages reply fixture: advances cached request number. + // SendMessages reply fixture: advances the cached watermark. #[allow(clippy::cast_possible_truncation)] fn synthesize_send_messages_reply<B, P>( consensus: &VsrConsensus<B, P>, client_id: u128, - session: u64, request: u64, commit: u64, ) -> Message<ReplyHeader> @@ -826,7 +888,6 @@ mod tests { &mut msg.as_mut_slice()[..header_size], ) .expect("zeroed bytes are valid"); - let _ = session; *header = ReplyHeader { cluster: consensus.cluster(), size: header_size as u32, diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 7ba69922b..e6caae450 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -396,7 +396,8 @@ pub enum MetadataSubmitError { NotPrimary, /// Primary but `commit_min < commit_max` (committed prefix not yet /// drained). Dispatching now would race ops inherited from a prior view; - /// for `Register` that trips `commit_register`'s session-eq assert. + /// for `Register` that double-commits a register and bumps the epoch + /// past the first reply's, fencing a live client. NotCaughtUp, /// Prepare queue full. PipelineFull, @@ -615,6 +616,7 @@ where let client_id = message.header().client; let session = message.header().session; let request = message.header().request; + let request_checksum = message.header().request_checksum; let operation = message.header().operation; // Preflight first: dedup, eviction sends, cached-reply replay all @@ -624,8 +626,14 @@ where let dispatch = if operation == Operation::Register { register_preflight(consensus, &self.client_table, client_id).await } else { - let outcome = - request_preflight(consensus, &self.client_table, client_id, session, request); + let outcome = request_preflight( + consensus, + &self.client_table, + client_id, + session, + request, + request_checksum, + ); apply_preflight_consensus_plane(consensus, outcome, client_id).await }; if !dispatch { @@ -995,9 +1003,12 @@ where .as_ref() .expect("submit_register_in_process: consensus only exists on shard 0"); - // Idempotent fast path: existing session skips pipeline + wire-reply. - if let Some(session) = self.client_table.borrow().get_session(client_id) { - return Ok(session); + // Idempotent fast path: existing entry skips pipeline + wire-reply. + // Returns the current epoch without bumping it; rebind-bumps-epoch + // (zombie fencing per reconnect) arrives with the stable-session-id + // auth flow, which commits a Register per bind. + if let Some(epoch) = self.client_table.borrow().get_epoch(client_id) { + return Ok(epoch); } // Wrong node: waiting or queueing cannot fix that, the client must @@ -1006,10 +1017,10 @@ where return Err(MetadataSubmitError::NotPrimary); } - // Mirror wire-path register_preflight: a racing second prepare fails - // check_register on commit. Surface pre-synthesis. Scans both the - // prepare queue and the request queue, so a register absorbed below - // dedups its own replays. + // Mirror wire-path register_preflight: a racing second prepare would + // commit a second register and bump the epoch past the first reply's. + // Surface pre-synthesis. Scans both the prepare queue and the request + // queue, so a register absorbed below dedups its own replays. if consensus .pipeline() .borrow() @@ -1030,7 +1041,7 @@ where ); // Catch-up gate (Register only: admitting one while a committed op - // is still unapplied races `commit_register`'s session-eq assert) or + // is still unapplied risks a double-register epoch bump) or // prepare queue full: absorb into the request queue instead of // bouncing with a transient error. The queued // entry carries this caller's reply subscriber; the commit path @@ -1049,14 +1060,20 @@ where return Err(MetadataSubmitError::PipelineFull); } return match receiver.await { - Ok(reply) => Ok(reply.header().commit), + // The commit's `commit_register` minted the epoch; read it + // from the table (the reply header does not carry it). + Ok(_reply) => self + .client_table + .borrow() + .get_epoch(client_id) + .ok_or(MetadataSubmitError::Canceled), // Entry dropped before commit: view-change reset or a // promotion-time preflight rejection. Same re-check as the // direct path's cancel arm below. Err(Canceled) => self .client_table .borrow() - .get_session(client_id) + .get_epoch(client_id) .ok_or(MetadataSubmitError::Canceled), }; } @@ -1068,18 +1085,24 @@ where .expect("Operation::Register is client-allowed; prepare projection cannot fail"); match self.dispatch_prepare_and_await(consensus, prepare).await { - Ok(reply) => Ok(reply.header().commit), + // The commit's `commit_register` minted the epoch; read it from + // the table (the reply header does not carry it). + Ok(_reply) => self + .client_table + .borrow() + .get_epoch(client_id) + .ok_or(MetadataSubmitError::Canceled), Err(Canceled) => { // View-change cancel. Re-check is correct-by-VSR: any // inherited Register applied via local commit_journal between - // cancel and read produces a cluster-authoritative session - // (`session = commit-op`, deterministic). Own surviving - // Register would have routed through `AlreadyRegistered` - // against the same entry, so no "this primary vs inherited - // primary" split. + // cancel and read mints the same epoch on every replica + // (`commit_register` counts in apply order, deterministic). + // Own surviving Register would have routed through + // `AlreadyRegistered` against the same entry, so no "this + // primary vs inherited primary" split. self.client_table .borrow() - .get_session(client_id) + .get_epoch(client_id) .ok_or(MetadataSubmitError::Canceled) } } @@ -1113,12 +1136,12 @@ where .as_ref() .expect("submit_logout_in_process: consensus only exists on shard 0"); - // Session guard: only propose a Logout when the slot still holds the - // exact session this logout targets. A late disconnect-logout for a - // reused client id (slot since rebound to a newer session) carries the - // stale session and is dropped here, so it can never wipe the fresh + // Epoch guard: only propose a Logout when the slot still holds the + // exact epoch this logout targets. A late disconnect-logout for a + // reused client id (slot since rebound to a newer epoch) carries the + // stale epoch and is dropped here, so it can never wipe the fresh // registration. A missing slot also fails the match and short-circuits. - if self.client_table.borrow().get_session(client_id) != Some(session) { + if self.client_table.borrow().get_epoch(client_id) != Some(session) { return Ok(consensus.commit_min()); } @@ -1165,7 +1188,7 @@ where return match receiver.await { Ok(reply) => Ok(reply.header().commit), Err(Canceled) => { - if self.client_table.borrow().get_session(client_id).is_none() { + if self.client_table.borrow().get_epoch(client_id).is_none() { Ok(consensus.commit_min()) } else { Err(MetadataSubmitError::Canceled) @@ -1180,7 +1203,7 @@ where match self.dispatch_prepare_and_await(consensus, prepare).await { Ok(reply) => Ok(reply.header().commit), Err(Canceled) => { - if self.client_table.borrow().get_session(client_id).is_none() { + if self.client_table.borrow().get_epoch(client_id).is_none() { Ok(consensus.commit_min()) } else { Err(MetadataSubmitError::Canceled) @@ -1295,7 +1318,7 @@ where /// /// No client session exists, so this skips `request_preflight` (like /// the logout precedent) and uses the reserved internal `client` id - /// `0`: never registered, so the commit path's `get_session(0)` is + /// `0`: never registered, so the commit path's `get_epoch(0)` is /// `None` and skips `commit_reply` (and its `assert!(client_id != 0)`), /// while the preflight and register asserts never run. Delete is /// idempotent, so the dropped dedup is harmless and a re-proposal on the @@ -1400,6 +1423,7 @@ where let client_id = request_header.client; let session = request_header.session; let request = request_header.request; + let request_checksum = request_header.request_checksum; let consensus = self .consensus @@ -1429,13 +1453,21 @@ where .into_generic()); } - // Dedup / session / eviction. shard 0 cannot route by the VSR + // Dedup / epoch fence / eviction. shard 0 cannot route by the VSR // consensus `client_id` (its top bits are random, not home-shard // routing), so a Replay/Evict/NotReady is returned to the home shard as // the reply -- `handle_client_request` writes it to the originating // socket by transport id, exactly like a fresh commit. Drop (client-bug - // stale/gap) surfaces as Canceled so the home shard stays silent. - match request_preflight(consensus, &self.client_table, client_id, session, request) { + // already-applied / future-epoch) surfaces as Canceled so the home + // shard stays silent. + match request_preflight( + consensus, + &self.client_table, + client_id, + session, + request, + request_checksum, + ) { PreflightOutcome::Dispatch => {} PreflightOutcome::Replay(reply) => { return server_common::Message::<GenericHeader>::try_from( @@ -1545,7 +1577,7 @@ where consensus.verify_pipeline(); let receiver = consensus.pipeline_message_with_subscriber(PlaneKind::Metadata, &prepare); // Register is the one op whose admission requires the catch-up gate - // (session-eq assert at commit); its submit path checks the gate and + // (double-register epoch bump); its submit path checks the gate and // the check-to-dispatch section is synchronous. Non-register ops // dispatch mid-window by design (they pipeline behind the in-flight // batch, like the wire path always has). @@ -1815,14 +1847,11 @@ where // Cache only if session exists. Client evicted between // prepare and commit: skip cache (`commit_reply` no-ops), // wire reply still ships. - let session = self - .client_table - .borrow() - .get_session(prepare_header.client); - if let Some(session) = session { + let epoch = self.client_table.borrow().get_epoch(prepare_header.client); + if let Some(epoch) = epoch { self.client_table.borrow_mut().commit_reply( prepare_header.client, - session, + epoch, reply.clone(), ); } else { @@ -1931,8 +1960,9 @@ where /// /// # Safety /// Re-preflight per iteration: `commit_journal` may have advanced the - /// client's request between push and drain (Stale / Duplicate / - /// `AlreadyRegistered`). Skipping produces a duplicate prepare and panics. + /// client's watermark between push and drain (Duplicate / AlreadyApplied + /// / `AlreadyRegistered`). Skipping produces a duplicate prepare and + /// panics. #[allow(clippy::future_not_send)] async fn drain_request_queue_into_prepares(&self) { let consensus = self.consensus.as_ref().unwrap(); @@ -1952,6 +1982,7 @@ where let client_id = req.message.header().client; let session = req.message.header().session; let request = req.message.header().request; + let request_checksum = req.message.header().request_checksum; let operation = req.message.header().operation; // If preflight or projection rejects below, dropping `req` (and // the sender taken from it) wakes an in-process awaiter with @@ -1960,8 +1991,14 @@ where let dispatch = if operation == Operation::Register { register_preflight(consensus, &self.client_table, client_id).await } else { - let outcome = - request_preflight(consensus, &self.client_table, client_id, session, request); + let outcome = request_preflight( + consensus, + &self.client_table, + client_id, + session, + request, + request_checksum, + ); apply_preflight_consensus_plane(consensus, outcome, client_id).await }; if !dispatch { @@ -2260,8 +2297,8 @@ where /// between. [`crate::metadata_helpers::is_caught_up_primary`] reads /// `commit_min == commit_max` as proof the table is caught up; an await /// here lets another task observe transient equality with stale table, - /// dispatch a fresh Register on an already-registered client, and panic - /// `commit_register`'s session-eq assert. + /// dispatch a fresh Register on an already-registered client, and bump + /// the epoch past the reply the live client holds. /// /// Inner block sync today. Future async state-machine must either: /// 1. Apply SM + bump `commit_min` in one `RefCell` borrow, or @@ -2331,11 +2368,11 @@ where }); // Cache only if session still exists. WAL replay may carry a // reply for a later-evicted client; `commit_reply` no-ops. - let session = self.client_table.borrow().get_session(header.client); - if let Some(session) = session { + let epoch = self.client_table.borrow().get_epoch(header.client); + if let Some(epoch) = epoch { self.client_table .borrow_mut() - .commit_reply(header.client, session, reply); + .commit_reply(header.client, epoch, reply); } else { tracing::trace!( client = header.client, @@ -3507,20 +3544,20 @@ mod tests { "resumed driver commits nothing new" ); assert_eq!( - md.client_table.borrow().get_session(CLIENT_A), + md.client_table.borrow().get_epoch(CLIENT_A), None, "session removed by the committed logout" ); } /// Register is the one op that still honors the catch-up gate (its - /// admission races `commit_register`'s session-eq assert against - /// committed-but-unapplied ops). New contract: a register arriving in - /// the mid-commit window is ABSORBED into the pipeline's request queue - /// with its reply subscriber attached, promoted by - /// the commit path once the batch drains, and the caller's await - /// resolves with the committed session — instead of the historical - /// `NotCaughtUp` bounce that one-shot CLI clients surfaced as + /// admission races a committed-but-unapplied register; a double commit + /// bumps the epoch past the first reply's and fences a live client). + /// New contract: a register arriving in the mid-commit window is + /// ABSORBED into the pipeline's request queue with its reply subscriber + /// attached, promoted by the commit path once the batch drains, and the + /// caller's await resolves with the committed epoch — instead of the + /// historical `NotCaughtUp` bounce that one-shot CLI clients surfaced as /// "Disconnected" login failures. #[compio::test] async fn register_in_mid_commit_window_is_queued_then_committed() { @@ -3634,13 +3671,13 @@ mod tests { } assert_eq!( outcome.expect("absorbed register must resolve"), - Ok(2), - "queued register commits with the next batch; session = commit op" + Ok(1), + "queued register commits with the next batch; first bind mints epoch 1" ); assert_eq!( - md.client_table.borrow().get_session(CLIENT_C), - Some(2), - "session created by the promoted register" + md.client_table.borrow().get_epoch(CLIENT_C), + Some(1), + "entry created by the promoted register" ); } @@ -3765,8 +3802,8 @@ mod tests { } compio::time::sleep(std::time::Duration::from_millis(1)).await; } - assert_eq!(outcome.expect("promoted register must resolve"), Ok(2)); - assert_eq!(md.client_table.borrow().get_session(CLIENT_C), Some(2)); + assert_eq!(outcome.expect("promoted register must resolve"), Ok(1)); + assert_eq!(md.client_table.borrow().get_epoch(CLIENT_C), Some(1)); assert!(is_caught_up_primary(consensus)); } }
