You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/07/27 08:49:20 UTC
(iggy) 01/03: temp
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 <g....@gmail.com>
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));
}
}