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
The following commit(s) were added to refs/heads/clients_table_v2 by this push:
new 05124f1f7 addres review comments
05124f1f7 is described below
commit 05124f1f7d967eef878d504511d62842745f5e8d
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Tue Jul 28 09:30:42 2026 +0200
addres review comments
---
core/binary_protocol/src/consensus/header.rs | 18 ++-
core/consensus/src/client_table.rs | 4 +-
core/consensus/src/metadata_helpers.rs | 105 ++++++++++--
.../tests/cluster/client_table_restart.rs | 27 +++-
core/metadata/src/impls/metadata.rs | 122 +++++++++-----
core/metadata/src/impls/recovery.rs | 94 ++++++++++-
core/sdk/src/vsr.rs | 18 ++-
core/server-ng/src/bootstrap.rs | 15 +-
core/server-ng/src/dispatch.rs | 152 ++++-------------
core/server-ng/src/http/error.rs | 53 +++++-
core/server-ng/src/http/session.rs | 16 +-
core/server-ng/src/http/state.rs | 113 +++++++++++--
core/server-ng/src/session_manager.rs | 162 ++-----------------
core/shard/src/coordinator.rs | 179 ++++++++++++++++++---
core/shard/src/lib.rs | 17 +-
15 files changed, 708 insertions(+), 387 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs
b/core/binary_protocol/src/consensus/header.rs
index 0114f8eb7..b819037a7 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -163,11 +163,19 @@ pub struct RequestHeader {
pub operation: Operation,
pub operation_padding: [u8; 7],
pub namespace: u64,
- /// Session fence epoch, handed to the client by its `Register` reply and
- /// echoed on every subsequent request. Bumped by the server on each
- /// committed re-register of the same `client`, so a request stamped with
an
- /// older value is a zombie from before a rebind and gets fenced. Unrelated
- /// to op numbers: nothing in it derives from the consensus log.
+ /// Session fence epoch: the commit op of the latest committed `Register`
+ /// for this `client`. Handed to the client by that register's reply and
+ /// echoed on every subsequent request.
+ ///
+ /// Every bind commits a `Register`, so each rebind of the same `client`
+ /// carries a strictly higher value, and a request stamped with an older
one
+ /// is a zombie from before that rebind and gets fenced. Being a log
+ /// position rather than a counter is what makes it non-regressing: it
+ /// cannot restart low after the server drops an entry and the client
+ /// registers again.
+ ///
+ /// Zero on `Register` itself (the client has no epoch to echo yet) and on
+ /// sessionless ops; header validation enforces both.
pub session: u64,
/// Acting user id, stamped by the metadata primary at admission for every
/// gated client op so the in-apply RBAC gate resolves the same identity on
diff --git a/core/consensus/src/client_table.rs
b/core/consensus/src/client_table.rs
index 0e3471f68..0fcc3fc30 100644
--- a/core/consensus/src/client_table.rs
+++ b/core/consensus/src/client_table.rs
@@ -237,7 +237,9 @@ impl ClientTable {
}
/// Check a request against the table. Epoch fence first, then the
- /// watermark. For Register, use [`Self::check_register`].
+ /// watermark. Register does not come through here: every bind proposes
+ /// unconditionally so its fence actually moves, see
+ /// [`Self::commit_register`].
///
/// `request_checksum` is the request's integrity stamp; zero (unstamped)
/// disables the reuse check.
diff --git a/core/consensus/src/metadata_helpers.rs
b/core/consensus/src/metadata_helpers.rs
index 7bb40edf3..065de7799 100644
--- a/core/consensus/src/metadata_helpers.rs
+++ b/core/consensus/src/metadata_helpers.rs
@@ -73,6 +73,21 @@ pub enum PreflightOutcome {
/// `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).
+///
+/// ## What the catch-up gate below does and does not establish
+///
+/// It says this replica has APPLIED everything it holds and its log suffix has
+/// re-earned quorum, which is what makes the eviction decisions below safe
+/// against a stale in-memory table on a freshly promoted primary.
+///
+/// It says nothing about how the table was BUILT. Every clause of
+/// `is_caught_up_primary` is about the log; recovery, meanwhile, replays from
+/// this node's snapshot floor, and checkpointing is node-local, so two
replicas
+/// reconstruct their tables from different starting points. Epochs survive
that
+/// (they are op-derived), watermarks do not -- see
+/// `metadata::impls::recovery`. So a caught-up primary is authoritative for
its
+/// own table, not for agreement with its peers' tables. Closing that needs the
+/// table in the snapshot, not a stronger gate here.
pub fn request_preflight<B, P>(
consensus: &VsrConsensus<B, P>,
client_table: &RefCell<ClientTable>,
@@ -126,11 +141,13 @@ where
RequestStatus::Duplicate(cached_reply) => {
PreflightOutcome::Replay(cached_reply.into_wire_bytes())
}
- // Session evicted under capacity pressure. SAFETY: catch-up gate makes
- // this replica authoritative for session truth.
+ // Session evicted under capacity pressure. The catch-up gate makes
this
+ // replica authoritative for its own committed session state, which is
+ // what an eviction frame reports.
RequestStatus::NoSession =>
PreflightOutcome::Evict(EvictionReason::NoSession),
- // Zombie holdover from before a re-register: terminal for that
- // holder. SAFETY: catch-up gate makes this replica authoritative.
+ // Zombie holdover from before a re-register: terminal for that holder.
+ // Sound on any caught-up replica because the fence is op-derived, so
+ // every replica that applied this register holds the same value.
RequestStatus::Fenced { current, received } => {
tracing::debug!(
client_id,
@@ -234,7 +251,12 @@ where
/// `true` -> dispatch. `false` -> absorbed (`AlreadyRegistered` replays cache;
/// in-flight register silently dropped).
#[allow(clippy::future_not_send)]
-pub fn register_preflight<B, P>(consensus: &VsrConsensus<B, P>, client_id:
u128) -> bool
+pub fn register_preflight<B, P>(
+ consensus: &VsrConsensus<B, P>,
+ client_table: &RefCell<ClientTable>,
+ client_id: u128,
+ user_id: u32,
+) -> bool
where
B: MessageBus,
P: Pipeline<Entry = PipelineEntry>,
@@ -250,9 +272,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 -> a second register commits -> the epoch bumps past the
- // one the first register's reply handed the client, fencing a live
+ // committed in WAL but not yet applied. Without the gate this dispatches a
+ // second register, which commits at a later op -> the entry's fence moves
+ // 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!(
@@ -267,6 +289,30 @@ where
return false;
}
+ // OWNERSHIP GATE. `commit_register`'s rebind branch overwrites the entry's
+ // `user_id`, and `resolve_acting_user_id` resolves authority for every
+ // replicated op from that field, so admitting a register for an entry
+ // another user owns would hand the caller that user's authority. Refuse by
+ // dropping it: this runs on the wire ingress and on the promotion of a
+ // queued register, neither of which has a caller to return a typed error
+ // to. The in-process submit checks the same condition first and does
+ // return one (`ClientIdOwnedByAnotherUser`), so a client that reaches here
+ // and is dropped retries into that terminal answer.
+ //
+ // Correct to decide here because the catch-up gate above already
+ // established this replica as authoritative for session truth.
+ if let Some(owner) = client_table.borrow().get_user_id(client_id)
+ && owner != user_id
+ {
+ tracing::warn!(
+ client_id,
+ presented_user = user_id,
+ entry_owner = owner,
+ "register_preflight: dropping register for an entry owned by
another user"
+ );
+ return false;
+ }
+
// Past the gates, every Register dispatches -- including one whose client
// already holds an entry. A bind is a fencing event: `commit_register`'s
// rebind branch bumps the entry's epoch, which is what fences the previous
@@ -502,7 +548,7 @@ mod tests {
client_table.borrow_mut().commit_reply(client_id, app_reply);
assert!(
- register_preflight(&consensus, client_id),
+ register_preflight(&consensus, &client_table, client_id,
ACTING_USER_ID),
"rebind must dispatch so commit_register bumps the fence epoch"
);
let sends = consensus.message_bus().client_sends.borrow();
@@ -714,6 +760,41 @@ mod tests {
);
}
+ // The promotion path and the wire ingress both land here, and neither has
+ // a caller to return a typed error to, so an entry owned by another user
is
+ // refused by dropping the register. Without this a register queued while
+ // the primary was catching up would commit at promotion and
+ // `commit_register` would overwrite the entry's `user_id`, handing the
+ // presenter that user's authority.
+ #[test]
+ fn register_preflight_drops_a_register_for_another_users_entry() {
+ const OWNER: u32 = ACTING_USER_ID;
+ const IMPOSTOR: u32 = ACTING_USER_ID + 1;
+
+ 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 initial_reply = synthesize_register_reply(&consensus, client_id,
17);
+ client_table
+ .borrow_mut()
+ .commit_register(client_id, OWNER, initial_reply);
+
+ assert!(
+ !register_preflight(&consensus, &client_table, client_id,
IMPOSTOR),
+ "a register for another user's entry must not dispatch"
+ );
+ assert!(
+ register_preflight(&consensus, &client_table, client_id, OWNER),
+ "the owner's own rebind must still dispatch"
+ );
+ assert_eq!(
+ client_table.borrow().get_user_id(client_id),
+ Some(OWNER),
+ "the refused register must not have touched the entry"
+ );
+ }
+
// Watermark jump: request numbers above the watermark dispatch even
// when non-contiguous (there is no RequestGap).
#[test]
@@ -754,9 +835,10 @@ mod tests {
consensus.advance_commit_max(5);
assert_ne!(consensus.commit_min(), consensus.commit_max());
+ let client_table = fresh_client_table();
let client_id: u128 = 0xC0DE;
- let result = register_preflight(&consensus, client_id);
+ let result = register_preflight(&consensus, &client_table, client_id,
ACTING_USER_ID);
assert!(!result, "register dispatch must short-circuit");
let sends = consensus.message_bus().client_sends.borrow();
@@ -794,9 +876,10 @@ mod tests {
let consensus = VsrConsensus::new(1, 0, 3, 0, ClientSpyBus::new(),
LocalPipeline::new());
consensus.init();
+ let client_table = fresh_client_table();
let client_id: u128 = 0xC0DE;
- let result = register_preflight(&consensus, client_id);
+ let result = register_preflight(&consensus, &client_table, client_id,
ACTING_USER_ID);
assert!(result, "New client proceeds through consensus");
let sends = consensus.message_bus().client_sends.borrow();
diff --git a/core/integration/tests/cluster/client_table_restart.rs
b/core/integration/tests/cluster/client_table_restart.rs
index 661aba302..750e83ce2 100644
--- a/core/integration/tests/cluster/client_table_restart.rs
+++ b/core/integration/tests/cluster/client_table_restart.rs
@@ -39,10 +39,12 @@
//! the pre-restart epoch are fenced zombies. There is deliberately no
//! credential-free rebind; `given_live_session_when_unauthenticated_peer_*`
//! pins that.
-//! 3. Sessions survive transport disconnect: `submit_disconnect_logout`
-//! tears down only consumer-group members (the group must rebalance off
-//! a dead consumer); everything else keeps its slot for resume until an
-//! explicit `Logout` or capacity eviction.
+//! 3. A crash leaves no `Logout` behind. Every transport disconnect releases
+//! its session (`submit_disconnect_logout`), so what makes the entry
+//! survivable is that the process died with the connection still open --
+//! hence these tests restart before closing the socket. Holding the slot
+//! open past a graceful disconnect would make resume work there too, but
+//! it needs a timer of its own; see that function's rustdoc.
//!
//! The Rust SDK cannot drive this yet: it resets its `ConsensusSession` on
//! every disconnect and re-registers under a fresh identity (the
@@ -120,9 +122,20 @@ async fn
given_committed_request_when_node_restarts_should_dedup_same_id_retry(
let (mut stream, session) = register(addr).await;
let create_stream = create_stream_payload("iggy137-dedup");
let committed = commit_request(&mut stream, session, 1,
&create_stream).await;
- drop(stream);
+ // Restart BEFORE dropping the socket, because the ordering is the
scenario.
+ // A crash takes the process down with the connection still open, so no
+ // `Logout` is committed and the session is still in the WAL for replay to
+ // rebuild. Closing first would model a graceful goodbye instead, and a
+ // graceful disconnect ends the session by design
+ // (`submit_disconnect_logout` releases the slot).
+ //
+ // Deterministic, not a race: the harness stops the node with SIGTERM, and
+ // the per-connection cleanup in the bus installer skips
`remove_client_meta`
+ // once the bus token is triggered -- so a shutdown fires no
+ // connection-lost callback for any still-open socket.
harness.restart_server().await.unwrap();
+ drop(stream);
// The reply for request 1 was already delivered, but the client cannot
// know that in the crash window; retrying the same id must converge on
@@ -147,9 +160,9 @@ async fn
given_bound_session_when_node_restarts_should_accept_next_request_id(
)
.await;
- drop(stream);
-
+ // Crash ordering, see the sibling test.
harness.restart_server().await.unwrap();
+ drop(stream);
// Continuation, not retry: the session advances to the next id. A node
// that forgot the watermark sees request 2 on an unknown session and
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index 403f0b7df..e40c27a0d 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -50,7 +50,7 @@ use iggy_common::variadic;
use journal::{Journal, JournalHandle};
use message_bus::MessageBus;
use server_common::Message;
-use server_common::iobuf::Frozen;
+use server_common::iobuf::{Frozen, Owned};
use std::cell::{Cell, RefCell};
use std::mem::size_of;
use std::path::Path;
@@ -689,13 +689,14 @@ where
let request = message.header().request;
let request_checksum = message.header().request_checksum;
let operation = message.header().operation;
+ let user_id = message.header().user_id;
// Preflight first: dedup, eviction sends, cached-reply replay all
// must run regardless of pipeline pressure. Wire-path ingress has no
// home-shard transport context, so resends fall back to the
// consensus-plane (best-effort by VSR id).
let dispatch = if operation == Operation::Register {
- register_preflight(consensus, client_id)
+ register_preflight(consensus, &self.client_table, client_id,
user_id)
} else {
let outcome = request_preflight(
consensus,
@@ -1088,6 +1089,12 @@ where
.as_ref()
.expect("submit_register_in_process: consensus only exists on
shard 0");
+ // Wrong node: waiting or queueing cannot fix that, the client must
+ // re-route to the primary.
+ if !(consensus.is_primary() && consensus.is_normal() &&
!consensus.is_syncing()) {
+ return Err(MetadataSubmitError::NotPrimary);
+ }
+
// OWNERSHIP GATE: the login frame's `client` field is caller-supplied,
// and `resolve_acting_user_id` resolves authority for every replicated
// op from this entry, so rebinding someone else's entry would run the
@@ -1095,7 +1102,17 @@ where
// its `user_id`). Refuse unless the authenticated user owns it.
// Terminal (see `ClientIdOwnedByAnotherUser`). An owned entry falls
// through: the rebind must commit so the epoch actually moves.
- {
+ //
+ // Only a CAUGHT-UP primary may issue it, like both sibling readers of
+ // this table (`request_preflight` and `register_preflight`, which gate
+ // the same way): the refusal is terminal, so a lagging or diverged
+ // replica answering it would deny a legitimate login off state it has
+ // not finished applying, and the client would never learn to redirect.
+ // Not caught up therefore SKIPS the check rather than refusing -- the
+ // register goes on to park in the request queue below, and
+ // `register_preflight` re-applies this gate when the commit path
+ // promotes it, by which point the table is authoritative.
+ if is_caught_up_primary(consensus) {
let table = self.client_table.borrow();
if let Some(owner) = table.get_user_id(client_id)
&& owner != user_id
@@ -1111,12 +1128,6 @@ where
}
}
- // Wrong node: waiting or queueing cannot fix that, the client must
- // re-route to the primary.
- if !(consensus.is_primary() && consensus.is_normal() &&
!consensus.is_syncing()) {
- return Err(MetadataSubmitError::NotPrimary);
- }
-
// 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
@@ -1140,14 +1151,20 @@ where
"build_register_request_message produced a header that fails
validate()"
);
- // Catch-up gate (Register only: admitting one while a committed op
- // 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
- // promotes it (`drain_request_queue_into_prepares`, which re-runs
- // `register_preflight`) as soon as the in-flight batch drains, and
- // the await below resolves exactly like the direct dispatch would.
+ // Fence floor, snapshotted BEFORE dispatch. This register's op is
+ // assigned above the journal tail, so it is strictly greater than
+ // `commit_max` is now -- which is what lets the cancel path below tell
+ // OUR fence from an older entry's that happened to survive.
+ let epoch_floor = consensus.commit_max();
+
+ // Not caught up (admitting a register while a committed op is still
+ // unapplied risks a double-register fence 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 promotes it (`drain_request_queue_into_prepares`, which
+ // re-runs `register_preflight` and so applies the ownership gate) as
+ // soon as the in-flight batch drains, and the await below resolves
+ // exactly like the direct dispatch would.
if !is_caught_up_primary(consensus) ||
consensus.pipeline().borrow().is_full() {
let (entry, receiver) =
consensus::RequestEntry::with_subscriber(request);
if consensus
@@ -1159,14 +1176,17 @@ where
// Both queues full: honest terminal backpressure.
return Err(MetadataSubmitError::PipelineFull);
}
- // Commit and cancel share one recovery: read the table. On
- // commit, `commit_register` moved the entry's fence (the reply
header
- // carries neither epoch nor watermark). On a view-change cancel
- // or promotion-time rejection the re-read is correct-by-VSR: an
- // inherited Register applied via `commit_journal` between cancel
- // and read stamps the same op-derived epoch on every replica.
- let _ = receiver.await;
- return self.bound_session(client_id);
+ return match receiver.await {
+ // The reply's `commit` IS the fence `commit_register` just
+ // stored (`build_reply_message` stamps it from the prepare's
+ // op), so take it from there rather than re-reading the table.
+ Ok(reply) => {
+ self.bound_session(client_id, Some(reply.header().commit),
epoch_floor)
+ }
+ // Entry dropped before commit: view-change reset, or a
+ // promotion-time preflight rejection.
+ Err(Canceled) => self.bound_session(client_id, None,
epoch_floor),
+ };
}
// `prepare_request` only fails on `!is_client_allowed`; Register is
// allowed, so unreachable. Panic loudly on regression instead of
@@ -1175,18 +1195,39 @@ where
.prepare_request(request)
.expect("Operation::Register is client-allowed; prepare projection
cannot fail");
- // Same commit/cancel collapse as the queued path above.
- let _ = self.dispatch_prepare_and_await(consensus, prepare).await;
- self.bound_session(client_id)
+ match self.dispatch_prepare_and_await(consensus, prepare).await {
+ Ok(reply) => self.bound_session(client_id,
Some(reply.header().commit), epoch_floor),
+ Err(Canceled) => self.bound_session(client_id, None, epoch_floor),
+ }
}
- /// Read a client's post-commit bind state (fence epoch + watermark) in one
- /// borrow. `Canceled` when the entry is absent: the register was canceled
- /// before commit, or the entry was evicted in between.
- fn bound_session(&self, client_id: u128) -> Result<BoundSession,
MetadataSubmitError> {
+ /// Assemble the bind result in one table borrow.
+ ///
+ /// `committed_epoch` is `Some` when this call's own Register committed, in
+ /// which case the fence comes from the reply that carries it. `None` is
the
+ /// view-change cancel path, where the fence has to be read back -- and is
+ /// only ours if it sits above `epoch_floor`. An entry at or below the
floor
+ /// predates this register, so returning its epoch would hand the caller a
+ /// fence that never moved, and nothing downstream would notice: a stale
+ /// epoch satisfies `check_request`'s equality test, so there is no
`Fenced`
+ /// and no `EpochAhead` to surface it. `Canceled` instead, and the retry
+ /// gets a real bind.
+ ///
+ /// `Canceled` also covers an absent entry (evicted between commit and
+ /// read).
+ fn bound_session(
+ &self,
+ client_id: u128,
+ committed_epoch: Option<u64>,
+ epoch_floor: u64,
+ ) -> Result<BoundSession, MetadataSubmitError> {
let table = self.client_table.borrow();
- table
- .get_epoch(client_id)
+ let epoch = committed_epoch.or_else(|| {
+ table
+ .get_epoch(client_id)
+ .filter(|&epoch| epoch > epoch_floor)
+ });
+ epoch
.zip(table.get_watermark(client_id))
.map(|(epoch, watermark)| BoundSession { epoch, watermark })
.ok_or(MetadataSubmitError::Canceled)
@@ -1217,13 +1258,11 @@ where
) {
return Some(Ok(refusal));
}
+ let owned =
+ Owned::<{ server_common::MESSAGE_ALIGN
}>::copy_from_slice(reply.as_slice());
Some(
- server_common::Message::<GenericHeader>::try_from(
- server_common::iobuf::Owned::<{
server_common::MESSAGE_ALIGN }>::copy_from_slice(
- reply.as_slice(),
- ),
- )
- .map_err(|_| MetadataSubmitError::Canceled),
+ Message::<GenericHeader>::try_from(owned)
+ .map_err(|_| MetadataSubmitError::Canceled),
)
}
PreflightOutcome::Evict(reason) => {
@@ -2095,12 +2134,13 @@ where
let request = req.message.header().request;
let request_checksum = req.message.header().request_checksum;
let operation = req.message.header().operation;
+ let user_id = req.message.header().user_id;
// If preflight or projection rejects below, dropping `req` (and
// the sender taken from it) wakes an in-process awaiter with
// `Canceled`; its submit path re-checks the client table.
let reply_sender = req.take_reply_sender();
let dispatch = if operation == Operation::Register {
- register_preflight(consensus, client_id)
+ register_preflight(consensus, &self.client_table, client_id,
user_id)
} else {
let outcome = request_preflight(
consensus,
diff --git a/core/metadata/src/impls/recovery.rs
b/core/metadata/src/impls/recovery.rs
index 37fd7355d..00ea8b800 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -213,8 +213,29 @@ where
// Register/Logout mutate the client table and skip the state
// machine, mirroring the commit paths (`on_ack` / `commit_journal`).
- // Replaying them from a fresh table in apply order re-mints the same
- // epochs every replica derived live.
+ //
+ // Epochs come back identical on every replica: they are the register's
+ // own commit op, so replay reads them out of the log rather than
+ // deriving them from replay order.
+ //
+ // Watermarks do NOT, and this is a known gap rather than an invariant.
+ // Replay starts at THIS node's snapshot floor, and checkpointing is
+ // node-local (`checkpoint_if_needed` fires on local journal
occupancy),
+ // so replicas cross the floor at different ops. If a client's earlier
+ // register fell below this node's floor while a later one survived,
+ // replay takes `commit_register`'s fresh-entry branch and the entry
+ // returns with `watermark = 0`, where a peer that replayed both took
+ // the rebind branch and kept it. The fence still passes, so nothing is
+ // evicted, and the same request id a peer answers `Duplicate` gets
+ // answered `New` here and re-executed -- exactly-once degrading to
+ // at-least-once, silently.
+ //
+ // Unobservable today (no shipping client re-presents a recovered
+ // `client_id`), and closed by putting the table in the snapshot so
+ // replay no longer has to reconstruct it. Until then a caught-up
+ // primary is authoritative for its OWN table only, which is why
+ // `request_preflight`'s catch-up gate cannot bridge this (see its
+ // rustdoc).
if header.operation == Operation::Register {
let reply = build_reply_message(header, &bytes::Bytes::new());
client_table.commit_register(header.client, header.user_id, reply);
@@ -493,6 +514,75 @@ mod tests {
);
}
+ // The watermark half of table recovery does NOT survive a checkpoint, and
+ // this pins that as a fact rather than a comment. Shape: a client
registers,
+ // commits request 1, a checkpoint lands past that register, then the
client
+ // rebinds. Replay starts above the floor, so it never sees the first
+ // register: `commit_register` takes its fresh-entry branch and the entry
+ // comes back with watermark 0, while a peer whose floor sat lower replayed
+ // both registers, took the rebind branch, and kept watermark 1.
+ //
+ // Consequence, once a client re-presents a recovered id: the same request
+ // that peer answers `Duplicate` is `New` here and gets re-executed. The
+ // fence is unaffected -- epochs are op-derived, so this node still returns
+ // the second register's op.
+ //
+ // Red/green for the table-in-snapshot work: when the table ships in the
+ // checkpoint, the watermark assertion below flips to `Some(1)`.
+ #[compio::test]
+ async fn
recover_loses_the_watermark_when_a_checkpoint_hides_the_first_register() {
+ const CLIENT: u128 = 0x1337;
+ const USER: u32 = 7;
+ const FLOOR: u64 = 2;
+
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+
+ // Ops 1..=2 are below the floor and are never replayed: the client's
+ // first Register and the request that advanced its watermark.
+ IggySnapshot::new(FLOOR)
+ .persist(&metadata_dir.join("snapshot.bin"))
+ .unwrap();
+
+ {
+ let journal =
PrepareJournal::open(&metadata_dir.join("journal.wal"), 0)
+ .await
+ .unwrap();
+ for entry in [
+ make_client_prepare(1, Operation::Register, CLIENT, USER, 0),
+ make_client_prepare(2, Operation::CreateStream, CLIENT, USER,
1),
+ // The rebind, above the floor, so replay does see this one.
+ make_client_prepare(3, Operation::Register, CLIENT, USER, 0),
+ ] {
+ journal.append(entry).await.unwrap();
+ }
+ journal.storage_ref().fsync().await.unwrap();
+ }
+
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ true,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ |_| {},
+ )
+ .await
+ .unwrap();
+
+ let table = &recovered.client_table;
+ assert_eq!(
+ table.get_epoch(CLIENT),
+ Some(3),
+ "the fence is op-derived, so it survives the checkpoint intact"
+ );
+ assert_eq!(
+ table.get_watermark(CLIENT),
+ Some(0),
+ "KNOWN GAP: the pre-floor watermark is lost, so request 1 reads as
New \
+ here while a lower-floor peer answers it as a duplicate"
+ );
+ }
+
// The IGGY-137 restart contract: a rebooted node must remember where
// each client left off. Replay re-mints the epoch, restores the
// watermark, and re-caches the reply so a retry of the last committed
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 671fbd5e9..90a10a552 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -47,10 +47,20 @@ const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> =
0..4;
// the retry from the current ConsensusSession. If disconnect created a fresh
VSR
// client/session, the retried request gets a new (client_id, request_id)
tuple, so
// server-side deduplication cannot match a mutation that may already have
committed
-// before the transport failure. The server now rebinds a transport that
presents
-// its old (client, session) identity (`try_resume_session`), so the fix here
is
-// to keep the ConsensusSession across reconnects and retry replicated writes
-// under the same (client_id, request_id) instead of re-registering fresh.
+// before the transport failure.
+//
+// The fix is to keep the ConsensusSession's client_id and request counter
across
+// reconnects and retry replicated writes under the same (client_id,
request_id)
+// instead of re-registering fresh. Resume happens through the LOGIN path: the
+// reconnecting client re-authenticates presenting its previous client_id, the
+// server verifies the authenticated user owns that entry, and the rebind
commits
+// a Register that adopts the entry with its watermark and reply ring intact.
Note
+// the epoch changes -- the rebind moves the fence to the new register's op --
so
+// the session field must be taken from the new login reply, not carried over.
+//
+// There is deliberately no credential-free rebind: presenting (client,
session)
+// on an unauthenticated transport is refused, since that pair is a dedup key
and
+// never a bearer token.
pub(crate) fn encode_contiguous_request(
session: &mut ConsensusSession,
code: u32,
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 9886321f8..e018c4e8b 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -1218,8 +1218,19 @@ async fn shard_main(
.expect("shard 0 always has a coordinator attached by the
builder");
// Reseed the client-id minter above every recovered entry before any
// listener accepts. The counter is per process; the table it must not
- // collide with was rebuilt from the previous boot's WAL.
-
coord.seed_client_sequence(shard.plane.metadata().client_table.borrow().client_ids());
+ // collide with was rebuilt from the previous boot's WAL. Keyed by view
+ // so a later promotion refolds the table (the minting path calls the
+ // same method, see `HttpInner::register_session_once`).
+ let boot_view = shard
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .map_or(0, consensus::VsrConsensus::view);
+ coord.seed_client_sequence(
+ boot_view,
+ shard.plane.metadata().client_table.borrow().client_ids(),
+ );
let on_client_request =
make_client_request_handler(&shard, &sessions,
Arc::clone(&config.system));
let (accepted_replica, dialed_replica) =
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 8ab85204d..8c4b4e241 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -112,17 +112,6 @@ use tracing::{debug, warn};
pub(crate) type ClientRequestQueues = Rc<RefCell<HashMap<u128,
VecDeque<Message<GenericHeader>>>>>;
pub(crate) type ActiveClientRequests = Rc<RefCell<HashSet<u128>>>;
-/// How long a disconnected session's client-table slot is held open for the
-/// client to reconnect and resume onto it before the sweeper reclaims it.
-///
-/// Resume runs through the login path, so the window only has to cover a
-/// reconnect plus a re-login -- seconds. It is generous by an order of
-/// magnitude so a client riding out a view change still lands inside it, and
-/// still bounded so cumulative connects cannot push the client table to its
-/// capacity-eviction point (every eviction silently erases a watermark).
-/// Swept by `run_heartbeat_verifier`, whose tick is far shorter than this.
-const SESSION_RECLAIM_GRACE: std::time::Duration =
std::time::Duration::from_mins(1);
-
pub(crate) fn make_client_request_handler<B, MJ, S>(
shard: &Rc<ShellShard<B, MJ, S>>,
sessions: &Rc<RefCell<SessionManager>>,
@@ -147,12 +136,7 @@ where
.borrow_mut()
.remove_connection(client_id)
{
- submit_disconnect_logout(
- Rc::clone(&shard_for_disconnect),
- &sessions_for_disconnect,
- vsr_client_id,
- session,
- );
+ submit_disconnect_logout(Rc::clone(&shard_for_disconnect),
vsr_client_id, session);
}
}));
Rc::new(move |client_id, message| {
@@ -480,7 +464,7 @@ where
.remove_connection(client_id)
&& let Some(shard) =
upgrade_shard_handle(&shard_handle_for_disconnect)
{
- submit_disconnect_logout(shard, &sessions_for_disconnect,
vsr_client_id, session);
+ submit_disconnect_logout(shard, vsr_client_id, session);
}
}));
Rc::new(move |client_id, message| {
@@ -540,7 +524,6 @@ where
.metadata()
.submit_register_in_process(vsr_client_id, user_id)
.await
- .ok()
.map(|bound| (bound.epoch, bound.watermark));
let _ = reply.try_send(bound);
}
@@ -1584,34 +1567,6 @@ pub(crate) async fn run_heartbeat_verifier<B, MJ, S>(
}
}
- // Reclaim sessions whose resume window elapsed. A disconnect defers
- // rather than logs out (see `submit_disconnect_logout`) so a
- // reconnecting client can resume onto its entry through the login
- // path; without this sweep those slots would accumulate per
- // cumulative connect and push the client table into capacity
- // eviction, which silently erases dedup watermarks.
- let reclaims = sessions
- .borrow_mut()
- .take_expired_reclaims(std::time::Instant::now());
- for (vsr_client_id, session) in reclaims {
- debug!(
- vsr_client_id,
- "session resume window elapsed; reclaiming client-table slot"
- );
- // Re-arm before submitting. The submit is spawned and can fail
- // transiently (view change, full pipeline), and
`take_expired_reclaims`
- // has already dropped the entry, so without this a failed submit
- // leaks the slot until capacity eviction. The spawned task clears
- // the entry once the `Logout` commits.
- sessions.borrow_mut().defer_reclaim(
- vsr_client_id,
- session,
- SESSION_RECLAIM_GRACE,
- std::time::Instant::now(),
- );
- submit_session_logout(Rc::clone(&shard), &sessions, vsr_client_id,
session);
- }
-
shard.bus.sleep(interval).await;
}
}
@@ -1632,7 +1587,7 @@ async fn evict_stale_client<B, MJ, S>(
{
let bound = sessions.borrow_mut().remove_connection(transport_client_id);
if let Some((vsr_client_id, session)) = bound {
- submit_disconnect_logout(Rc::clone(shard), sessions, vsr_client_id,
session);
+ submit_disconnect_logout(Rc::clone(shard), vsr_client_id, session);
}
let ctx = shard.plane.metadata().consensus.as_ref().map_or(
consensus::EvictionContext {
@@ -2170,15 +2125,18 @@ where
.submit_register_in_process(vsr_client_id, user_id)
.await;
}
- let (reply, rx) = shard::channel::<Option<(u64, u64)>>(1);
+ let (reply, rx) = shard::channel::<Result<(u64, u64),
MetadataSubmitError>>(1);
shard.forward_metadata_submit(shard::MetadataSubmit::Register {
vsr_client_id,
user_id,
reply,
});
match rx.recv().await {
- Ok(Some((epoch, watermark))) => Ok(BoundSession { epoch, watermark }),
- _ => Err(MetadataSubmitError::Canceled),
+ Ok(Ok((epoch, watermark))) => Ok(BoundSession { epoch, watermark }),
+ // The owner's error, preserved: `Canceled` is only for a dropped
+ // channel, where nothing came back to classify.
+ Ok(Err(error)) => Err(error),
+ Err(_) => Err(MetadataSubmitError::Canceled),
}
}
@@ -2452,62 +2410,26 @@ where
))
}
-/// Disconnect policy: reclaim now, or hold the slot open for a resume window.
+/// Release the client-table slot for a disconnected transport, cluster-wide.
///
-/// The local `SessionManager` connection is already dropped by the caller.
-/// A consumer-group member is logged out immediately -- the group must
-/// rebalance off a dead consumer without waiting. Anything else has its
-/// reclaim DEFERRED by [`SESSION_RECLAIM_GRACE`], because session resume runs
-/// through the login path: a reconnecting client re-authenticates under its
-/// previous `client_id` and the committed rebind adopts the existing entry
-/// while keeping its watermark and reply ring intact, which only works while
-/// that entry still exists.
+/// The local `SessionManager` connection is already dropped by the caller;
+/// this is what drops the replicated entry, so a peer replica does not keep an
+/// orphaned session until it evicts one under capacity pressure.
///
-/// The deferral is swept by `run_heartbeat_verifier`, which calls
-/// [`submit_session_logout`] directly once the window elapses. Holding slots
-/// indefinitely instead would move the client table's eviction point from
-/// concurrent connections to CUMULATIVE connects, and every capacity eviction
-/// silently erases a dedup watermark.
-#[allow(clippy::future_not_send)]
-fn submit_disconnect_logout<B, MJ, S>(
- shard: Rc<ShellShard<B, MJ, S>>,
- sessions: &Rc<RefCell<SessionManager>>,
- vsr_client_id: u128,
- session: u64,
-) where
- B: ShellBus,
- MJ: JournalHandle + 'static,
- MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
- S: 'static,
-{
- let is_group_member = !shard
- .plane
- .metadata()
- .mux_stm
- .streams()
- .consumer_group_memberships(vsr_client_id)
- .is_empty();
- if is_group_member {
- submit_session_logout(shard, sessions, vsr_client_id, session);
- return;
- }
- debug!(
- vsr_client_id,
- grace = ?SESSION_RECLAIM_GRACE,
- "transport disconnected; holding session for resume, then reclaiming"
- );
- sessions.borrow_mut().defer_reclaim(
- vsr_client_id,
- session,
- SESSION_RECLAIM_GRACE,
- std::time::Instant::now(),
- );
-}
-
-/// Submit a session-matched `Logout` unconditionally, releasing the
-/// `ClientTable` slot on every replica (shard 0 included, since shard 0 is
-/// itself a replica).
+/// Unconditional, and deliberately so. Holding the slot open for a grace
+/// window would let a reconnecting client resume onto its entry with its
+/// watermark and reply ring intact, but nothing in tree re-presents a
+/// `client_id` after a disconnect (the Rust SDK mints a fresh one on
+/// re-login), so the window buys nothing today and the slot it holds is not
+/// free: the client table's eviction point moves from concurrent connections
+/// to CUMULATIVE connects, and every capacity eviction silently erases a
+/// dedup watermark.
///
+/// A resume window becomes worth having once SDK-side identity stability
+/// lands, at which point it needs a timer of its own -- riding the heartbeat
+/// verifier is not an option, since that only runs when `heartbeat.enabled`
+/// is set and `collect_stale` keys off the heartbeat interval, so ungating it
+/// would mass-evict consumer-group members on a deployment that does not ping.
/// Deliberately does NOT drop the local `ClientTable` slot first:
/// `submit_logout_*` short-circuits when the slot is already gone, so a
/// pre-emptive local removal would suppress the `Logout` and leave peer
@@ -2516,9 +2438,8 @@ fn submit_disconnect_logout<B, MJ, S>(
/// shard 0 and forwards for peer-homed connections; its session guard drops a
/// stale logout for a reused client id.
#[allow(clippy::future_not_send)]
-fn submit_session_logout<B, MJ, S>(
+fn submit_disconnect_logout<B, MJ, S>(
shard: Rc<ShellShard<B, MJ, S>>,
- sessions: &Rc<RefCell<SessionManager>>,
vsr_client_id: u128,
session: u64,
) where
@@ -2534,23 +2455,16 @@ fn submit_session_logout<B, MJ, S>(
const DISCONNECT_LOGOUT_REQUEST_ID: u64 = u64::MAX;
let bus = shard.bus.clone();
- let sessions = Rc::clone(sessions);
bus.spawn(async move {
- match submit_logout_on_owner(&shard, vsr_client_id, session,
DISCONNECT_LOGOUT_REQUEST_ID)
- .await
+ if let Err(error) =
+ submit_logout_on_owner(&shard, vsr_client_id, session,
DISCONNECT_LOGOUT_REQUEST_ID)
+ .await
{
- // Slot released cluster-wide; drop the bookkeeping that would
- // otherwise re-submit on the next sweep.
- Ok(_) => sessions.borrow_mut().complete_reclaim(vsr_client_id),
- // Leave the re-armed reclaim in place so the sweeper retries. A
- // repeat is harmless: `submit_logout_*` short-circuits once the
- // slot is gone, and its session guard drops a stale logout for a
- // reused client id.
- Err(error) => warn!(
+ warn!(
vsr_client_id,
?error,
- "session logout submit failed; retrying after the reclaim
grace"
- ),
+ "disconnect logout submit failed; peer slots may linger until
eviction"
+ );
}
});
}
diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs
index 511a1dd22..0d294908a 100644
--- a/core/server-ng/src/http/error.rs
+++ b/core/server-ng/src/http/error.rs
@@ -160,6 +160,21 @@ pub(in crate::http) enum AuthError {
/// way.
SessionNotAccepted,
SessionUnavailable,
+ /// The `client_id` this gateway minted already has a committed session
+ /// owned by a DIFFERENT user, so the Register was refused terminally.
+ ///
+ /// Distinct from [`Self::SessionUnavailable`] because the status code is
+ /// the whole point: 503 is about the most auto-retried status there is and
+ /// no foreign SDK special-cases it, so rendering a permanent,
deterministic
+ /// refusal as 503 hands the caller's HTTP stack a retry loop it can never
+ /// escape. 409 says the id is taken and stops it.
+ SessionIdOwnedByAnotherUser,
+ /// The minted `client_id` already had a committed session for this SAME
+ /// user, so the Register rebound onto it instead of creating one. Internal
+ /// to the mint retry in `register_session` and never rendered: the caller
+ /// mints a different id. Present as a variant so the retry cannot confuse
+ /// it with a terminal cross-user refusal.
+ SessionIdTaken,
}
impl From<IggyError> for AuthError {
@@ -186,7 +201,18 @@ impl IntoResponse for AuthError {
// Transient server condition -> 503, retryable by the CLIENT only
// (a forwarder must not re-issue an unknown-outcome Register under
// this node's session budget on the caller's behalf).
- Self::SessionUnavailable => service_unavailable(),
+ // `SessionIdTaken` only escapes the mint retry when every attempt
+ // collided, which means the minter is wrong rather than unlucky --
+ // same unknown-outcome answer as a canceled Register.
+ Self::SessionUnavailable | Self::SessionIdTaken =>
service_unavailable(),
+ // Terminal: retrying cannot change the answer, and admitting it
+ // would run this caller's replicated ops under the entry owner's
+ // authority.
+ Self::SessionIdOwnedByAnotherUser => (
+ StatusCode::CONFLICT,
+ Json(ErrorResponse::from_error(&IggyError::InvalidClientId)),
+ )
+ .into_response(),
}
}
}
@@ -634,4 +660,29 @@ mod tests {
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
assert!(!response.headers().contains_key(RETRY_AFTER));
}
+
+ // The ownership refusal is permanent and deterministic. Rendering it as
+ // 503 would hand the caller's HTTP stack a retry loop it can never escape
+ // (no foreign SDK special-cases 503), so the status is load-bearing.
+ #[test]
+ fn owned_client_id_renders_as_terminal_conflict() {
+ let response = AuthError::SessionIdOwnedByAnotherUser.into_response();
+ assert_eq!(response.status(), StatusCode::CONFLICT);
+ assert!(
+ response.headers().get(RETRY_AFTER).is_none(),
+ "a terminal refusal must not advertise a retry"
+ );
+ }
+
+ // Its siblings stay retryable, so the split is visible in one place.
+ #[test]
+ fn unknown_outcome_registers_stay_retryable() {
+ for error in [AuthError::SessionUnavailable,
AuthError::SessionNotAccepted] {
+ let status = error.into_response().status();
+ assert!(
+ status.is_server_error(),
+ "an unknown commit outcome must stay retryable, got {status}"
+ );
+ }
+ }
}
diff --git a/core/server-ng/src/http/session.rs
b/core/server-ng/src/http/session.rs
index acc6527e6..8b1540f66 100644
--- a/core/server-ng/src/http/session.rs
+++ b/core/server-ng/src/http/session.rs
@@ -40,13 +40,19 @@ use tokio::sync::Mutex;
/// which evicts the oldest-committed client when full. Capping HTTP at half
/// that bound keeps this plane from crowding the others out and keeps the
/// combined steady state under the shared bound, so a live idle HTTP session
-/// is not routinely evicted consensus-side. The other half stays honest
-/// because a dropped connection releases its slot once the reclaim grace
-/// expires, rather than holding it for the process lifetime. The residual
eviction race (both planes
-/// busy) degrades gracefully: an evicted session's next control write is
-/// classified as an eviction and re-registers (see
[`HttpInner::forget_session`]).
+/// is not routinely evicted consensus-side. The non-HTTP half tracks live
+/// connections because every transport disconnect logs its session out
+/// (`submit_disconnect_logout`), so occupancy there is concurrent, not
+/// cumulative. The residual eviction race (both planes busy) degrades
+/// gracefully: an evicted session's next control write is classified as an
+/// eviction and re-registers (see [`HttpInner::forget_session`]).
pub(in crate::http) const MAX_HTTP_SESSIONS: usize = CLIENTS_TABLE_MAX / 2;
+/// Watermark a brand-new client-table entry carries: no application request
+/// has committed under it yet. A fresh mint that comes back with anything else
+/// bound to an entry that already existed (see `HttpInner::register_session`).
+pub(in crate::http) const FRESH_ENTRY_WATERMARK: u64 = 0;
+
/// First per-session request id the write path hands out. VSR request numbers
/// are 1-based and strictly increasing within a session.
pub(in crate::http) const FIRST_REQUEST_ID: u64 = 1;
diff --git a/core/server-ng/src/http/state.rs b/core/server-ng/src/http/state.rs
index 81519ff1e..933d6c6d7 100644
--- a/core/server-ng/src/http/state.rs
+++ b/core/server-ng/src/http/state.rs
@@ -43,8 +43,8 @@ use crate::http::error::{AuthError, ReadError,
primary_redirect_location};
use crate::http::forward::ForwardState;
use crate::http::jwt::JwtManager;
use crate::http::session::{
- BarrierEntry, FIRST_REQUEST_ID, HttpSession, MAX_HTTP_SESSIONS,
RegistrationBarrier,
- forget_if_same, live_entry, sweep_expired,
+ BarrierEntry, FIRST_REQUEST_ID, FRESH_ENTRY_WATERMARK, HttpSession,
MAX_HTTP_SESSIONS,
+ RegistrationBarrier, forget_if_same, live_entry, sweep_expired,
};
/// Response header carrying the current VSR view number. Stamped by
@@ -194,18 +194,82 @@ impl HttpInner {
live_entry(&self.sessions.borrow(), key, now_secs)
}
- /// Mint a shard-0 client id and run the VSR `Register` for a fresh
session.
- /// Holds no table borrow; the caller inserts the result under `key`.
+ /// Mint a shard-0 client id and run the VSR `Register` for a fresh
session,
+ /// retrying on a fresh id if the minted one turns out to be taken.
+ ///
+ /// The minter is a per-process counter reseeded from the client table at
+ /// boot, so a fresh mint normally lands on a free id. Two situations break
+ /// that, and neither is predictable from here: a promoted primary mints
+ /// from a counter with no relationship to the ids its predecessor
+ /// committed, and in a cluster every node counts independently. Landing on
+ /// an occupied entry is therefore reactive to detect and cheap to fix --
+ /// mint again. Bounded, because a run of collisions means the counter is
+ /// wrong rather than unlucky, and looping would hide that.
+ ///
+ /// The two collision signals are asymmetric. A different owner is refused
+ /// terminally by the register ownership gate. The SAME user is not refused
+ /// at all -- it rebinds, silently inheriting a watermark written by
another
+ /// of that user's sessions, which would make this session's first writes
+ /// read as duplicates and answer them from the other session's cache. A
+ /// non-zero watermark on what should be a brand-new session is exactly
that
+ /// tell.
async fn register_session(
&self,
key: String,
user_id: u32,
expiry: u64,
+ ) -> Result<Rc<HttpSession>, AuthError> {
+ /// Enough to ride out a promotion-era counter overlap; beyond this the
+ /// minter is misconfigured and the 503 is the honest answer.
+ const MINT_ATTEMPTS: u8 = 3;
+
+ for attempt in 1..=MINT_ATTEMPTS {
+ match self
+ .register_session_once(key.clone(), user_id, expiry)
+ .await
+ {
+ Ok(session) => return Ok(session),
+ Err(AuthError::SessionIdOwnedByAnotherUser |
AuthError::SessionIdTaken)
+ if attempt < MINT_ATTEMPTS =>
+ {
+ warn!(
+ attempt,
+ "server-ng HTTP: minted client id was already
registered; re-minting"
+ );
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ Err(AuthError::SessionUnavailable)
+ }
+
+ /// One mint-and-Register attempt. `SessionIdTaken` means the id was live
+ /// under this same user, so the caller should mint a different one.
+ async fn register_session_once(
+ &self,
+ key: String,
+ user_id: u32,
+ expiry: u64,
) -> Result<Rc<HttpSession>, AuthError> {
let coordinator = self
.shard
.coordinator()
.ok_or(AuthError::SessionUnavailable)?;
+ // Refold the client table into the minter if this is the first mint of
+ // the current view. Cheap and skipped within a view, and it is what
+ // stops a PROMOTED primary from minting against ids its predecessor
+ // committed from an unrelated counter -- the table is replicated, the
+ // counter is per process. Boot does the same call (`bootstrap`); this
+ // one covers every later view.
+ {
+ let metadata = self.shard.plane.metadata();
+ if let Some(consensus) = metadata.consensus.as_ref() {
+ coordinator.seed_client_sequence(
+ consensus.view(),
+ metadata.client_table.borrow().client_ids(),
+ );
+ }
+ }
// Reuse the TCP accept path's minter: it draws from the same shard-0
// `client_seq`, so an HTTP session id can never collide with a TCP
// virtual client's and the shard-0 tag (top 16 bits == 0) is
preserved.
@@ -250,26 +314,43 @@ impl HttpInner {
MetadataSubmitError::NotPrimary
| MetadataSubmitError::NotCaughtUp
| MetadataSubmitError::PipelineFull =>
AuthError::SessionNotAccepted,
- _ => AuthError::SessionUnavailable,
+ // Terminal, and the only variant here that is: retrying
+ // anywhere cannot make the id free. Kept off the 503 path
+ // so the caller's HTTP stack does not auto-retry forever.
+ MetadataSubmitError::ClientIdOwnedByAnotherUser => {
+ AuthError::SessionIdOwnedByAnotherUser
+ }
+ // `InProgress` / `Canceled` mean a prepare may still
+ // commit cluster-wide, so the outcome is unknown rather
+ // than terminal. A future variant lands here too: 503 is
+ // the safe default, since it never asserts a refusal the
+ // server did not make.
+ MetadataSubmitError::InProgress |
MetadataSubmitError::Canceled | _ => {
+ AuthError::SessionUnavailable
+ }
}
})?;
- // Number from above the entry's watermark, not from 1. The gateway is
- // the one client whose request counter lives in the process that
- // restarts: after a reboot the id minter can re-mint this client id
- // for the same user, binding to the recovered entry -- numbering from
- // FIRST_REQUEST_ID would then read as duplicates and answer this
- // session's writes with the previous boot's cached replies. A fresh
- // entry has watermark 0, so this is FIRST_REQUEST_ID there.
- let next_request = bound.watermark + 1;
- debug_assert!(next_request >= FIRST_REQUEST_ID);
+ // A fresh mint must land on a fresh entry, so a watermark it did not
+ // write means the id was already registered to this same user (see
+ // `register_session`). Rebinding onto it would inherit that session's
+ // dedup history; hand the id back instead and let the caller re-mint.
+ if bound.watermark != FRESH_ENTRY_WATERMARK {
+ warn!(
+ client_id,
+ user_id,
+ watermark = bound.watermark,
+ "server-ng HTTP: minted client id already had a committed
session for this user"
+ );
+ return Err(AuthError::SessionIdTaken);
+ }
Ok(Rc::new(HttpSession {
key,
client_id,
session: bound.epoch,
user_id,
expiry,
- gate: Mutex::new(next_request),
- data_request: Cell::new(next_request),
+ gate: Mutex::new(FIRST_REQUEST_ID),
+ data_request: Cell::new(FIRST_REQUEST_ID),
registry_token: Cell::new(None),
in_flight_writes: Cell::new(0),
}))
diff --git a/core/server-ng/src/session_manager.rs
b/core/server-ng/src/session_manager.rs
index 4d6c6b43a..9033aa98b 100644
--- a/core/server-ng/src/session_manager.rs
+++ b/core/server-ng/src/session_manager.rs
@@ -99,9 +99,6 @@ pub struct SessionManager {
/// Reverse index: `client_id` → `connection_id` for fast lookup when
/// a consensus reply arrives and needs routing to the right connection.
client_to_connection: HashMap<u128, u128>,
- /// Disconnected sessions awaiting reclaim, `client_id` -> deadline.
- /// See [`Self::defer_reclaim`].
- pending_reclaims: HashMap<u128, PendingReclaim>,
/// This shard's copy of the configured cluster roster, served by the
/// pre-auth `GetClusterMetadata` read. Lives here because it is the
/// per-shard context already threaded to the non-replicated read path;
@@ -109,24 +106,12 @@ pub struct SessionManager {
cluster_roster: Rc<ClusterRoster>,
}
-/// A disconnected session's slot, held open just long enough for the client
-/// to come back and resume onto it.
-#[derive(Debug, Clone, Copy)]
-struct PendingReclaim {
- /// Fence epoch the disconnected connection was bound to; the reclaiming
- /// `Logout` is session-matched against it so it cannot wipe a newer
- /// registration under the same id.
- session: u64,
- deadline: Instant,
-}
-
impl SessionManager {
#[must_use]
pub fn new() -> Self {
Self {
connections: HashMap::new(),
client_to_connection: HashMap::new(),
- pending_reclaims: HashMap::new(),
cluster_roster: Rc::new(ClusterRoster::disabled()),
}
}
@@ -181,66 +166,6 @@ impl SessionManager {
.collect()
}
- /// Hold a disconnected session's client-table slot open for `grace`, then
- /// let the sweeper reclaim it.
- ///
- /// Session resume runs through the login path: a reconnecting client
- /// re-authenticates under its previous `client_id` and the register fast
- /// path binds it back to the existing entry, watermark and reply ring
- /// intact. That only works while the entry still exists, so a disconnect
- /// cannot reclaim it immediately -- but leaving it forever moves the
- /// client table's eviction point from concurrent connections to CUMULATIVE
- /// connects, and every capacity eviction silently erases a watermark
- /// (turning that client's next retry back into a re-execution). The grace
- /// window is the compromise: resume within it, reclaim after.
- ///
- /// Consumer-group members skip this entirely and are logged out at once --
- /// the group must rebalance off a dead consumer without waiting.
- pub fn defer_reclaim(&mut self, client_id: u128, session: u64, grace:
Duration, now: Instant) {
- self.pending_reclaims.insert(
- client_id,
- PendingReclaim {
- session,
- deadline: now + grace,
- },
- );
- }
-
- /// Drop a pending reclaim because the client came back. Called from
- /// [`Self::bind_session`], so a resumed session is never reclaimed out
- /// from under its new connection.
- fn cancel_reclaim(&mut self, client_id: u128) {
- self.pending_reclaims.remove(&client_id);
- }
-
- /// Drop a pending reclaim because its `Logout` committed and the slot is
- /// released. Separate from [`Self::cancel_reclaim`] only in intent: the
- /// reclaim submit is fire-and-forget and can fail transiently, so the
- /// sweeper re-arms the deadline before submitting and the entry is cleared
- /// here on success. Without that pairing a failed submit would drop the
- /// bookkeeping and leak the slot until capacity eviction, which silently
- /// erases the client's dedup watermark.
- pub fn complete_reclaim(&mut self, client_id: u128) {
- self.pending_reclaims.remove(&client_id);
- }
-
- /// Take every reclaim whose grace has elapsed. The caller submits a
- /// session-matched `Logout` for each, which releases the client-table slot
- /// on every replica.
- #[must_use]
- pub fn take_expired_reclaims(&mut self, now: Instant) -> Vec<(u128, u64)> {
- let mut reclaimed = Vec::new();
- self.pending_reclaims.retain(|&client_id, pending| {
- if now >= pending.deadline {
- reclaimed.push((client_id, pending.session));
- false
- } else {
- true
- }
- });
- reclaimed
- }
-
/// The consensus client id a connection is bound to, if any. The heartbeat
/// verifier reads it to look up consumer-group membership before deciding
/// whether an eviction would actually release anything.
@@ -342,10 +267,6 @@ impl SessionManager {
old_conn.state = ConnectionState::Connected;
}
- // The client came back and re-authenticated onto this session, so a
- // reclaim deferred by its previous disconnect must not fire.
- self.cancel_reclaim(client_id);
-
// Now mutate the target connection.
self.connections.get_mut(&connection_id).unwrap().state =
ConnectionState::Bound {
user_id,
@@ -627,83 +548,30 @@ mod tests {
assert!(mgr.get_session(c1).is_none());
assert_eq!(mgr.get_session(c2), Some((200, 20)));
}
- // A disconnected non-group session holds its client-table slot only for
- // the grace window: resume within it, reclaim after. Without the reclaim
- // the slot leaks per cumulative connect and pushes the client table into
- // capacity eviction, which silently erases dedup watermarks.
+ // Every disconnect releases its consensus session, group member or not.
+ // Holding the slot open for a resume window instead leaked it: the sweep
+ // that would have collected it rides the heartbeat verifier, which only
+ // runs when `heartbeat.enabled` is set, and that ships false -- so the
+ // slot survived for the process lifetime and pushed the client table
+ // toward capacity eviction, which silently erases dedup watermarks.
#[test]
- fn deferred_reclaim_expires_after_its_grace_window() {
+ fn disconnect_releases_the_bound_session_for_logout() {
let mut mgr = SessionManager::new();
- let now = Instant::now();
- let grace = Duration::from_mins(1);
-
- mgr.defer_reclaim(100, 7, grace, now);
- assert!(
- mgr.take_expired_reclaims(now + Duration::from_secs(59))
- .is_empty(),
- "inside the window the slot is held for resume"
- );
- assert_eq!(
- mgr.take_expired_reclaims(now + Duration::from_secs(61)),
- vec![(100, 7)],
- "past the window the slot is reclaimed with its bound epoch"
- );
- assert!(
- mgr.take_expired_reclaims(now + Duration::from_mins(10))
- .is_empty(),
- "a reclaim is taken exactly once"
- );
- }
-
- // A client that comes back and re-authenticates onto its session must not
- // have it reclaimed out from under the new connection.
- #[test]
- fn rebinding_cancels_a_pending_reclaim() {
- let mut mgr = SessionManager::new();
- let now = Instant::now();
- mgr.defer_reclaim(100, 7, Duration::from_mins(1), now);
-
let conn = 1;
mgr.ensure_connection(conn, addr(5100), ClientTransportKind::Tcp);
mgr.login(conn, 3).unwrap();
mgr.bind_session(conn, 100, 7).unwrap();
- assert!(
- mgr.take_expired_reclaims(now + Duration::from_mins(10))
- .is_empty(),
- "the resumed session must survive its old disconnect's reclaim"
- );
- assert_eq!(mgr.get_session(conn), Some((100, 7)));
- }
-
- // The reclaim submit is spawned and can fail transiently, so the sweeper
- // re-arms before submitting and clears only on a committed `Logout`. This
- // pins both halves: a failed submit comes back, a committed one does not.
- #[test]
- fn re_armed_reclaim_retries_until_completed() {
- let mut mgr = SessionManager::new();
- let now = Instant::now();
- let grace = Duration::from_mins(1);
-
- mgr.defer_reclaim(100, 7, grace, now);
- let swept = mgr.take_expired_reclaims(now + grace);
- assert_eq!(swept, vec![(100, 7)]);
-
- // Submit failed: the sweeper's re-arm is what makes it retryable.
- mgr.defer_reclaim(100, 7, grace, now + grace);
assert_eq!(
- mgr.take_expired_reclaims(now + grace + grace),
- vec![(100, 7)],
- "a failed submit must be retried, not leaked"
+ mgr.remove_connection(conn),
+ Some((100, 7)),
+ "the disconnect must hand back (client_id, epoch) so the caller
can log it out"
);
-
- // Submit committed on the retry.
- mgr.defer_reclaim(100, 7, grace, now + grace + grace);
- mgr.complete_reclaim(100);
- assert!(
- mgr.take_expired_reclaims(now + Duration::from_mins(10))
- .is_empty(),
- "a completed reclaim must not be re-submitted"
+ assert!(mgr.get_session(conn).is_none());
+ assert_eq!(
+ mgr.remove_connection(conn),
+ None,
+ "a second disconnect has nothing left to release"
);
}
}
diff --git a/core/shard/src/coordinator.rs b/core/shard/src/coordinator.rs
index 21e2ff821..f85daba69 100644
--- a/core/shard/src/coordinator.rs
+++ b/core/shard/src/coordinator.rs
@@ -62,6 +62,12 @@ use tracing::warn;
/// bits carry the shard, the bottom 112 the mint sequence.
const CLIENT_ID_SHARD_SHIFT: u32 = 112;
+/// Sequence half of a minted client id. The tag above it is the inter-shard
+/// reply routing key (`message_bus::client_id_owning_shard`), so a sequence
+/// that bled past this mask would silently re-route a connection's replies to
+/// another shard for the process lifetime.
+const CLIENT_SEQUENCE_MASK: u128 = (1 << CLIENT_ID_SHARD_SHIFT) - 1;
+
/// Coordinator owned by shard 0 only.
///
/// Wrapped in `Rc` by the bootstrap and shared with the replica listener,
@@ -83,6 +89,10 @@ pub struct ShardZeroCoordinator {
replica_rr: Cell<u16>,
client_rr: Cell<u16>,
client_seq: Cell<u128>,
+ /// View the mint counter was last seeded for. `None` until the first seed.
+ /// Tracked so [`Self::seed_client_sequence`] can be called on the minting
+ /// path and fold the table exactly once per view instead of per mint.
+ seeded_view: Cell<Option<u32>>,
}
impl ShardZeroCoordinator {
@@ -115,6 +125,7 @@ impl ShardZeroCoordinator {
replica_rr: Cell::new(0),
client_rr: Cell::new(0),
client_seq: Cell::new(1),
+ seeded_view: Cell::new(None),
})
}
@@ -144,32 +155,67 @@ impl ShardZeroCoordinator {
/// Mint a client id encoding `target_shard` in the top 16 bits and a
/// monotonic per-coordinator counter in the bottom 112 bits.
+ ///
+ /// The sequence is masked into its half of the id. Counting there cannot
+ /// reach 2^112, but the counter is also SEEDED from recovered client-table
+ /// ids ([`Self::seed_client_sequence`]), and those include values a client
+ /// supplied on the wire -- so without the mask a chosen id near the top of
+ /// the range would, after a restart, push the next mint's carry into the
+ /// shard tag and misroute that connection's replies. Zero is skipped
+ /// because the wire header rejects `client == 0`.
fn mint_client_id(&self, target_shard: u16) -> u128 {
- let seq = self.client_seq.get();
- self.client_seq.set(seq.wrapping_add(1));
+ let mut seq = self.client_seq.get() & CLIENT_SEQUENCE_MASK;
+ if seq == 0 {
+ seq = 1;
+ }
+ self.client_seq
+ .set(seq.wrapping_add(1) & CLIENT_SEQUENCE_MASK);
(u128::from(target_shard) << CLIENT_ID_SHARD_SHIFT) | seq
}
- /// Reseed the mint counter above every sequence in `recovered_ids`.
+ /// Reseed the mint counter above every sequence in `recovered_ids`, once
+ /// per `view`.
///
- /// The counter is per process and starts at 1, while a restarted node
- /// recovers client-table entries keyed by the previous boot's ids. Left
- /// alone, the first logins after a restart re-mint ids that are already
- /// taken: a different user's login is then refused outright (the register
- /// ownership gate), and the same user's inherits a watermark it never
- /// wrote. Seeding past the recovered high-water mark makes the collision
- /// unreachable instead of handled.
+ /// The counter is per process and starts at 1, while the client table it
+ /// must not collide with is REPLICATED -- so it holds ids minted by other
+ /// nodes' counters, and by this node's previous boot. Left alone, a mint
+ /// lands on an id that is already taken: a different user's login is then
+ /// refused outright (the register ownership gate), and the same user's
+ /// rebinds and inherits a watermark it never wrote. Seeding past the
+ /// table's high-water mark makes that unreachable instead of handled.
///
- /// Boot-time only, on shard 0, before any listener accepts. Never lowers
- /// the counter.
- pub fn seed_client_sequence(&self, recovered_ids: impl Iterator<Item =
u128>) {
- const SEQUENCE_MASK: u128 = (1 << CLIENT_ID_SHARD_SHIFT) - 1;
-
- let highest = recovered_ids.map(|id| id & SEQUENCE_MASK).max();
+ /// Two moments need it, which is why `view` is the key rather than "call
+ /// this at boot": startup, where the table was rebuilt from the previous
+ /// boot's WAL, and PROMOTION, where a node starts minting against ids its
+ /// predecessor committed from an unrelated counter. A view change is the
+ /// observable edge for the second, and re-seeding for a view already
+ /// covered is skipped, so this is cheap enough to call from the minting
+ /// path.
+ ///
+ /// Never lowers the counter, so a redundant call cannot hand back an id
+ /// this process already minted.
+ ///
+ /// Only ids whose tag names a shard of this node are folded in. The table
+ /// also holds ids a client chose for itself (the TCP login path takes
+ /// `client` straight off the wire), and those carry arbitrary tags;
folding
+ /// them would let one login steer this node's counter. A chosen id with a
+ /// *plausible* tag can still move it, which is why the mint masks -- the
+ /// worst case is then a counter that wraps low and collides with live
+ /// entries, which the register ownership gate refuses rather than
+ /// mis-serves.
+ pub fn seed_client_sequence(&self, view: u32, recovered_ids: impl
Iterator<Item = u128>) {
+ if self.seeded_view.get() == Some(view) {
+ return;
+ }
+ self.seeded_view.set(Some(view));
+ let highest = recovered_ids
+ .filter(|id| (id >> CLIENT_ID_SHARD_SHIFT) <
u128::from(self.total_shards))
+ .map(|id| id & CLIENT_SEQUENCE_MASK)
+ .max();
let Some(highest) = highest else {
return;
};
- let next = highest.saturating_add(1);
+ let next = highest.saturating_add(1) & CLIENT_SEQUENCE_MASK;
if next > self.client_seq.get() {
self.client_seq.set(next);
}
@@ -535,7 +581,8 @@ mod tests {
(3u128 << CLIENT_ID_SHARD_SHIFT) | 0x2a,
9,
];
- coord.seed_client_sequence(recovered.into_iter());
+ // total_shards is 4, so tags 1 and 3 are this node's and fold in.
+ coord.seed_client_sequence(0, recovered.into_iter());
let id = coord.mint_client_id(2);
assert_eq!(
@@ -562,8 +609,8 @@ mod tests {
for _ in 0..5 {
let _ = coord.mint_client_id(0);
}
- coord.seed_client_sequence(std::iter::empty());
- coord.seed_client_sequence([1u128, 2].into_iter());
+ coord.seed_client_sequence(0, std::iter::empty());
+ coord.seed_client_sequence(1, [1u128, 2].into_iter());
let id = coord.mint_client_id(0);
assert_eq!(
@@ -573,6 +620,98 @@ mod tests {
);
}
+ // The sequence must never carry into the shard tag: that tag is the
+ // inter-shard reply routing key, so a carry would send a live connection's
+ // replies to a different shard for the process lifetime. Reachable only
+ // because the counter is seeded from recovered ids, and the TCP login path
+ // takes `client` straight off the wire -- so a client can choose one near
+ // the ceiling and a restart folds it in.
+ #[test]
+ fn mint_never_carries_the_sequence_into_the_shard_tag() {
+ let senders = build_senders(4);
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 4,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
+
+ // A chosen id with a plausible tag, sequence at the ceiling.
+ coord.seed_client_sequence(0, std::iter::once(CLIENT_SEQUENCE_MASK));
+ for _ in 0..4 {
+ let id = coord.mint_shard_zero_client_id();
+ assert_eq!(
+ id >> CLIENT_ID_SHARD_SHIFT,
+ 0,
+ "shard-0 mint must keep tag 0, got id {id:#x}"
+ );
+ assert_ne!(id, 0, "the wire header rejects client == 0");
+ }
+ }
+
+ // Ids a client chose for itself carry arbitrary tags. Folding those into
+ // this node's counter would let one login steer it.
+ #[test]
+ fn seed_ignores_ids_tagged_outside_this_node() {
+ let senders = build_senders(2);
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 2,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
+
+ // Tag 9 is not a shard of a 2-shard node: caller-supplied, ignored.
+ coord.seed_client_sequence(
+ 0,
+ std::iter::once((9u128 << CLIENT_ID_SHARD_SHIFT) | 0x0f_ff_ff),
+ );
+ assert_eq!(
+ coord.mint_shard_zero_client_id() & CLIENT_SEQUENCE_MASK,
+ 1,
+ "an id tagged outside this node must not move the counter"
+ );
+ }
+
+ // Promotion is the second moment the counter can be wrong: the new primary
+ // mints against ids its predecessor committed from an unrelated counter.
+ // A view change is the observable edge, and re-seeding within one view is
+ // skipped so this is cheap enough to sit on the minting path.
+ #[test]
+ fn seed_refolds_the_table_once_per_view() {
+ let senders = build_senders(2);
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 2,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
+
+ coord.seed_client_sequence(4, std::iter::once(0x10));
+ assert_eq!(
+ coord.mint_shard_zero_client_id() & CLIENT_SEQUENCE_MASK,
+ 0x11
+ );
+
+ // Same view: the table is not refolded, so a later entry is ignored.
+ coord.seed_client_sequence(4, std::iter::once(0x80));
+ assert_eq!(
+ coord.mint_shard_zero_client_id() & CLIENT_SEQUENCE_MASK,
+ 0x12
+ );
+
+ // Promotion: new view, so the predecessor's high-water mark is folded.
+ coord.seed_client_sequence(5, std::iter::once(0x80));
+ assert_eq!(
+ coord.mint_shard_zero_client_id() & CLIENT_SEQUENCE_MASK,
+ 0x81,
+ "a promoted primary must mint above what its predecessor committed"
+ );
+ }
+
#[test]
fn shard_zero_local_and_delegated_ids_never_collide() {
let senders = build_senders(4);
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index b042d556c..74e48c91e 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -50,6 +50,7 @@ use message_bus::fd_transfer::DupedFd;
use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
use message_bus::replica::listener::MessageHandler;
use metadata::IggyMetadata;
+use metadata::MetadataSubmitError;
use metadata::impls::metadata::StreamsFrontend;
use metadata::stm::StateMachine;
use partitions::{IggyPartition, IggyPartitions, PollFragments, PollingArgs,
PollingConsumer};
@@ -161,16 +162,20 @@ pub fn channel<T: Send + 'static>(capacity: usize) ->
(Sender<T>, Receiver<T>) {
/// connection homes on a peer shard, that shard verifies credentials and
/// owns the session locally, but the consensus proposal (`Register` /
/// `Logout`) must execute on shard 0. The peer hands just that step here
-/// and awaits the committed op number over `reply` (`None` = transient
-/// submit failure; all `MetadataSubmitError` variants are transient by
-/// contract, so the caller retries rather than distinguishing them).
+/// and awaits the outcome over `reply`. `Register` carries the submit error
+/// verbatim because one variant
+/// (`MetadataSubmitError::ClientIdOwnedByAnotherUser`) is terminal and must
+/// not be retried; the remaining variants are transient by contract.
pub enum MetadataSubmit {
Register {
vsr_client_id: u128,
user_id: u32,
- /// `(epoch, watermark)` of the committed bind; `None` on a transient
- /// submit failure.
- reply: Sender<Option<(u64, u64)>>,
+ /// `Ok((epoch, watermark))` for a committed bind, or the submit error
+ /// verbatim. The error must survive the hop: the ownership refusal is
+ /// TERMINAL, and flattening it into "no reply" makes the login look
+ /// transient, which costs the client a retry storm of full password
+ /// verifications.
+ reply: Sender<Result<(u64, u64), MetadataSubmitError>>,
},
Logout {
vsr_client_id: u128,
