You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "numinnex (via GitHub)" <gi...@apache.org> on 2026/07/27 08:49:42 UTC
[PR] feat(consensus): rework client table for session resume across restarts [iggy]
numinnex opened a new pull request, #3753:
URL: https://github.com/apache/iggy/pull/3753
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #3753:
URL: https://github.com/apache/iggy/pull/3753#issuecomment-5101249821
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3753:
URL: https://github.com/apache/iggy/pull/3753#discussion_r3658181796
##########
core/server-ng/src/dispatch.rs:
##########
@@ -578,11 +578,142 @@ where
.ok();
let _ = reply.try_send(commit);
}
+ shard::MetadataSubmit::ResumeLookup {
+ vsr_client_id,
+ reply,
+ } => {
+ let entry = lookup_resumable_session(&shard, vsr_client_id);
+ let _ = reply.try_send(entry);
+ }
}
});
})
}
+/// Shard 0's read side of a session-resume attempt: `(epoch, user_id)` for a
+/// registered client, `None` otherwise. Table reads are only authoritative on
+/// a caught-up primary (same rule as `request_preflight`'s gate); a stale
+/// answer here would rebind a transport to a session the cluster has already
+/// rotated, so fail closed and let the client retry.
+fn lookup_resumable_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ vsr_client_id: u128,
+) -> Option<(u64, u32)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ let metadata = shard.plane.metadata();
+ let consensus = metadata.consensus.as_ref()?;
+ if !consensus::is_caught_up_primary(consensus) {
+ debug!(
+ vsr_client_id,
+ is_primary = consensus.is_primary(),
+ commit_min = consensus.commit_min(),
+ commit_max = consensus.commit_max(),
+ "resume lookup refused: not a caught-up primary"
+ );
+ return None;
+ }
+ let table = metadata.client_table.borrow();
+ let entry = table
+ .get_epoch(vsr_client_id)
+ .zip(table.get_user_id(vsr_client_id));
+ if entry.is_none() {
+ debug!(vsr_client_id, "resume lookup: no table entry");
+ }
+ entry
+}
+
+/// Rebind a reconnecting transport that presents its pre-restart identity
+/// (IGGY-137 session resume, the implicit-rebind contract): a replicated
+/// request on an unbound transport whose `(client, session)` matches a live
+/// entry in the replicated client table binds this connection to that
+/// session and proceeds as if bound all along.
+///
+/// The `(client_id, epoch)` pair acts as a bearer token here: the client id
+/// is a client-generated random u128, so presenting it proves the caller is
+/// (or eavesdropped on) the original registrant. `bind_session` evicts any
+/// previous connection bound to the same client id, which is the conflict
+/// rule for one session arriving on two connections.
+///
+/// Returns the `(client_id, session)` binding on success, `None` when the
+/// identity does not check out (caller falls back to the unbound-transport
+/// reply and the client re-registers).
+#[allow(clippy::future_not_send)]
+async fn try_resume_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ header: &RequestHeader,
+) -> Option<(u128, u64)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ // A resumable frame carries the full old identity; anything less is a
+ // plain unbound request (e.g. a fresh SDK that has not registered yet).
+ if header.client == 0 || header.session == 0 || header.request == 0 {
+ return None;
+ }
+
+ let entry = if shard.id == 0 {
+ lookup_resumable_session(shard, header.client)
+ } else {
+ let (reply, rx) = shard::channel::<Option<(u64, u32)>>(1);
+ shard.forward_metadata_submit(shard::MetadataSubmit::ResumeLookup {
+ vsr_client_id: header.client,
+ reply,
+ });
+ rx.recv().await.ok().flatten()
+ };
+ let (epoch, user_id) = entry?;
+ if epoch != header.session {
+ // Stale epoch = zombie holdover (the preflight would fence it);
+ // future epoch = client bug. Neither may rebind.
+ warn!(
+ transport_client_id,
+ client = header.client,
+ presented = header.session,
+ current = epoch,
+ "refusing session resume with mismatched epoch"
+ );
+ return None;
+ }
+
+ {
+ let mut sessions = sessions.borrow_mut();
+ if let Err(error) = sessions.login(transport_client_id, user_id) {
Review Comment:
The first two paragraphs are right and are fixed: resume is deleted outright rather than gated, so there is no credential-free bind and no `(client_id, epoch)` bearer path. Both halves of the third paragraph need correcting though, because the deleted case does not hold.
The trace stops at identity resolution and does not follow the stamped `user_id` into the gate that consumes it. `DeleteUser`'s apply purges the permissioner:
```rust
// stm/user.rs
state.permissioner.delete_permissions_for_user(user_id as UserId);
```
and that purge is total: `permissioner/mod.rs` clears the global map, both all-streams sets and all three specific-streams indices. Every rule then fails closed on a missing entry, e.g. `permissioner_rules/streams.rs` is `if let Some(..) = users_permissions.get(&user_id) && .. { Ok(()) } else { Err(Unauthorized) }`, and `messages.rs::poll_messages` reaches the same `Err(Unauthorized)` after exhausting every path.
Both planes go through that same permissioner, metadata via `authz::authorize` -> `check`, partition via `dispatch/authz.rs::append_messages` and `dispatch.rs`'s `poll_messages`. So a deleted user's surviving entry holds a slot and a watermark and no authority: every gated op on either plane commits as an `Unauthorized` no-op. `remove_client` not firing on `DeleteUser` is not load-bearing, because deletion is already enforced per-op by the layer that actually decides.
The slab-reuse half narrows to almost nothing for the same reason. The freed id carries no inherited grants (the purge removed them), and a *reconnect* cannot exploit the reuse: the register fast path now requires the authenticated user to own the entry, and after reuse that id legitimately belongs to the new user, so whoever binds acts as themselves. What is left needs the deleted user's socket to stay open across the delete AND an admin creating a new user that lands on exactly the freed slab key. Root is structurally excluded, slab 0 is undeletable and `stm/user.rs` comments that exact reason.
The clause that does stand is the other one, "the authz layer never re-checks status", and it stands on its own without the session-purge framing. The distinction is which attributes the gate re-reads per op:
- grants: `UpdatePermissionsRequest::apply` calls `permissioner.update_permissions_for_user`, and `authorize` consults the permissioner on every op, so revocation is already live against bound sessions.
- existence: purged on delete, same per-op path, also live.
- status: `UpdateUserRequest::apply` sets `user.status` and never touches the permissioner, and `authz.rs` contains no `status` reference on this branch or on master.
So status is the single authorization-relevant attribute living outside the per-op path, which makes it the only one a bound session can outrun. That also points at a cheaper fix than purging sessions: check status inside `authorize` after the root short-circuit (`&Users` already exposes it in the same borrow), and in the partition path's `authorize_uid`. Deactivation then behaves exactly like permission revocation, which works today. Purging the client table instead would mean a `user_id -> client_ids` reverse scan over up to `CLIENTS_TABLE_MAX` slots inside shard 0's no-await commit region, which is the cost you object to on `evict_oldest`, and `SessionManager` is per-shard transport-local while `DeleteUser` applies on shard 0 only, so that half needs a cross-shard broadcast. Neither buys authority the permissioner does not already enforce.
This is also pre-existing rather than introduced here: master has the same two `remove_client` call sites, the same login-only status check in `auth.rs`, and no status check in the gate. Worth its own issue; happy to file it.
Unrelated to the above, your `dispatch.rs:2553` finding turned up one more thing while I was verifying this. The reclaim I added to fix the leak was itself leaky: `take_expired_reclaims` drains with `retain`, so the entry was gone before the spawned `Logout` submit ran, and a transient failure (`NotPrimary` mid view change, `PipelineFull`, `Canceled`) dropped it permanently, leaving exactly the slot-until-eviction case your comment describes. Now the sweeper re-arms the deadline before submitting and the entry is cleared only when the `Logout` commits, so a failed submit retries after the grace instead of leaking. A repeat is harmless since `submit_logout_*` short-circuits once the slot is gone. Pinned by a unit test covering both halves.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3753:
URL: https://github.com/apache/iggy/pull/3753#discussion_r3659194637
##########
core/server-ng/src/dispatch.rs:
##########
@@ -112,6 +112,17 @@ 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);
Review Comment:
this sweep never runs. `take_expired_reclaims` has exactly one production caller (`run_heartbeat_verifier`), which is only spawned when `config.heartbeat.enabled` is true - and that ships `false` in both `core/server-ng/config.toml` and `core/server/config.toml`. there is no `true` anywhere in the repo.
so the deferral is armed and never collected: `pending_reclaims` grows for the process lifetime and the client-table slot is never released.
##########
core/server-ng/src/http/state.rs:
##########
@@ -253,14 +253,23 @@ impl HttpInner {
_ => AuthError::SessionUnavailable,
Review Comment:
`ClientIdOwnedByAnotherUser` falls into the `_` arm here, so it becomes `SessionUnavailable` and then a bare 503.
worth flagging because it means fixing the `.ok()` at the shard boundary changes nothing for http - both paths independently downgrade a permanent, deterministic refusal to a retryable status. and 503 is about the most auto-retried status there is: none of the foreign sdks have any 503 handling, so whatever the integrator's http stack does by default takes over.
the diagnostic story is rough too - the actionable line is the `iggy.metadata.diag` warn in `submit_register_in_process`, while the default log view shows `Canceled` here, which reads as transient.
a dedicated arm mapping to a terminal 4xx would fix it - 409 fits, the id is taken.
##########
core/shard/src/coordinator.rs:
##########
@@ -143,7 +147,32 @@ impl ShardZeroCoordinator {
fn mint_client_id(&self, target_shard: u16) -> u128 {
let seq = self.client_seq.get();
self.client_seq.set(seq.wrapping_add(1));
- (u128::from(target_shard) << 112) | seq
+ (u128::from(target_shard) << CLIENT_ID_SHARD_SHIFT) | seq
+ }
+
+ /// Reseed the mint counter above every sequence in `recovered_ids`.
+ ///
+ /// 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.
+ ///
+ /// 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>) {
Review Comment:
this only runs at boot - `bootstrap.rs` is the sole call site. but the client table is replicated, so a node keeps accumulating entries minted by other nodes' counters long after its own seed ran, and nothing reseeds on promotion.
after a failover the new primary mints from a counter that has no relationship to the ids its predecessor committed. the two are uncorrelated, so they collide whenever the promoted node's counter is at or below the old primary's high-water mark. `MAX_HTTP_SESSIONS` is 4096, so a lot of consecutive http logins can land on an occupied entry.
different user hits the ownership gate and gets a terminal refusal for a perfectly legitimate login. same user falls through and rebinds, inheriting a watermark written by a different session of that user.
reseeding on promotion as well as at boot removes the trigger outright, and it's the smallest of the fixes in this area.
separately: the mask folds every recovered id including caller-supplied ones, and `mint_client_id` never bounds `seq`. one login with a chosen `client_id` near the top of the 112-bit range plus a restart overflows into the shard tag, and `client_id_owning_shard` is the inter-shard reply routing key - so replies for new connections would forward to the wrong shard for the process lifetime. seeding only from ids where `(id >> 112) < total_shards` and bounding `seq` covers it.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -2369,10 +2452,61 @@ where
))
}
-/// Disconnect cleanup: the local `SessionManager` connection is already
-/// dropped by the caller; this submits a session-matched `Logout` so the
-/// committed apply releases the `ClientTable` slot on every replica (shard 0
-/// included, since shard 0 is itself a replica).
+/// Disconnect policy: reclaim now, or hold the slot open for a resume window.
+///
+/// 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 refences the existing entry
+/// while keeping its watermark and reply ring intact, which only works while
+/// that entry still exists.
+///
+/// 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);
Review Comment:
this is a regression against master. on master `submit_disconnect_logout` had no `is_group_member` branch at all - it spawned the logout unconditionally, so every disconnect released the slot cluster-wide. now a non-consumer-group disconnect emits nothing, and the sweep meant to catch it is gated off by default (see the comment on `SESSION_RECLAIM_GRACE`).
with 8192 slots and one leaked per non-cg disconnect that reached `Bound`, a node fills up over hours to days depending on churn, and only a process restart resets it. after that every register evicts a live session, and each eviction silently erases that session's dedup watermark.
simplest fix is to delete the grace machinery and restore the unconditional logout - that also removes the `complete_reclaim` session-key bug and the demoted-primary retry loop below. worth a tracking note for when sdk-side stable client ids land, since the window becomes useful then and would need its own always-on timer rather than riding the heartbeat verifier.
please don't fix this by flipping `heartbeat.enabled` - `collect_stale` keys off the heartbeat interval, so ungating it would mass-evict every consumer-group member on a deployment that isn't pinging.
##########
core/server-ng/src/http/state.rs:
##########
@@ -253,14 +253,23 @@ impl HttpInner {
_ => 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.
Review Comment:
this can only ever be `FIRST_REQUEST_ID` on a single node, which makes the whole `BoundSession` watermark path dead there.
the comment describes resume as the minter re-minting a client id onto a recovered entry - but that was never resume, it was an accidental collision, and `seed_client_sequence` was added in this same commit specifically to make it unreachable ("seeding past the recovered high-water mark makes the collision unreachable instead of handled"). so a fresh mint is provably above every recovered id, every register takes `commit_register`'s fresh-entry branch, and `watermark` is `REGISTER_REQUEST_ID`.
net effect is that two fixes in this commit cancel: the watermark plumbing added for resume, and the reseed that removed the only mechanism producing a non-zero watermark.
the exception is a cluster - after a failover a promoted node can mint onto an existing entry and a same-user collision rebinds, so the value is non-zero but inherited from a different session of that user. that's worse than dead, not better.
worth deciding whether the resume machinery lands now alongside the sdk half, or comes out and the pr description is adjusted.
##########
core/consensus/src/client_table.rs:
##########
@@ -156,181 +236,190 @@ 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`].
Review Comment:
`check_register` was deleted in this commit, so this intra-doc link dangles. `cargo doc` in ci doesn't set `RUSTDOCFLAGS`, so it warns and still exits 0 - it'll just sit in the log.
there's a prose reference to `check_register` in `metadata_helpers.rs` too.
##########
core/integration/tests/cluster/client_table_restart.rs:
##########
@@ -235,22 +344,64 @@ async fn commit_request(stream: &mut TcpStream, session: u64, request: u64, body
}
}
-/// Post-restart continuation: keep presenting the old identity until the
-/// server commits (or serves the cached reply for) the request. Every
-/// attempt uses a fresh connection, both because the old one died with the
-/// node and so an unanswered frame cannot desync the next attempt. Panics
+/// Post-restart continuation: re-authenticate under the OLD `client_id`, then
+/// keep presenting the old identity, round-robin across every node, until one
+/// commits (or serves the cached reply for) the request.
+///
+/// The re-login is the resume, and it is a rebind: the ownership-gated
+/// Register commits, the recovered entry refences at that register's op, and
+/// the login reply hands back the NEW epoch. Continuation frames must stamp
+/// it -- the pre-restart epoch is a fenced zombie from here on. Watermark and
+/// reply ring survive the rebind, which is what the dedup assertion rests on.
+/// There is deliberately NO way to rebind without credentials -- an unbound
+/// transport that merely presents `(client, session)` gets the empty-reply
+/// fail-fast (see `given_unauthenticated_resume_*`).
+///
+/// Every attempt uses a fresh connection, both because the old one died with
+/// the node and so an unanswered frame cannot desync the next attempt. Panics
/// with the last observed failure mode when the budget runs out.
-async fn resume_request(addr: SocketAddr, session: u64, request: u64, body: &Bytes) {
- let header = request_header(Operation::CreateStream, session, request, body.len());
+async fn resume_request(
+ addrs: &[SocketAddr],
+ session: u64,
+ request: u64,
+ body: &Bytes,
+ expect_replay_of: Option<&CommittedReply>,
+) {
let deadline = Instant::now() + RESUME_BUDGET;
let mut last_failure = "the listener never came back".to_string();
+ let mut attempt = 0usize;
while Instant::now() < deadline {
+ let addr = addrs[attempt % addrs.len()];
+ attempt += 1;
let Ok(mut stream) = TcpStream::connect(addr).await else {
sleep(RETRY_PAUSE).await;
continue;
};
+ // Re-authenticate on the fresh connection. The rebind commits a
+ // Register, so the epoch strictly advances past the pre-restart one
+ // (op-derived; regression would mean the fence can be replayed into).
+ let resumed = match login_on(&mut stream).await {
+ Some(resumed) => {
+ assert!(
+ resumed > session,
+ "rebind must refence above the pre-restart epoch (old {session}, got {resumed})"
Review Comment:
looks like a botched line join - there's a run of spaces mid-message and the line runs past 100 columns.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -969,35 +1048,67 @@ where
/// Submit `Register` from in-process, await commit. Wire reply still fires
/// via `message_bus.send_to_client`; subscriber is additive.
///
+ /// Every bind proposes -- there is deliberately no fast path returning an
+ /// existing entry's state. A bind is a fencing event: only a committed
+ /// Register moves the entry's epoch (to the register's commit op), and
+ /// that bump is what fences the previous holder of this session
+ /// (`RequestStatus::Fenced`). Short-circuiting a rebind would leave two
+ /// live holders sharing one fence, the zombie scenario the epoch exists
+ /// to kill. Rebinding onto an existing entry preserves its watermark and
+ /// reply ring, which is how session resume works.
+ ///
/// # Returns
- /// Session number (= commit op). Idempotent: existing session short-circuits.
+ /// [`BoundSession`]: the fence epoch the client must stamp into `session`,
+ /// plus the entry's current watermark so a caller that lost its position
+ /// (the HTTP gateway after a restart) can resume numbering above it.
///
/// # Errors
- /// [`MetadataSubmitError`] (all transient): `NotPrimary`, `NotCaughtUp`,
- /// `PipelineFull`, `InProgress`, `Canceled`. `Canceled` dominates on view
- /// change; new primary inherits via `commit_journal`, SDK retries.
+ /// [`MetadataSubmitError`]. All transient except
+ /// `ClientIdOwnedByAnotherUser`, which is terminal: `NotPrimary`,
+ /// `NotCaughtUp`, `PipelineFull`, `InProgress`, `Canceled`. `Canceled`
+ /// dominates on view change; the new primary inherits via
+ /// `commit_journal` and the SDK retries.
///
/// # Panics
/// On `client_id == 0` or shard without consensus.
///
/// # Safety
- /// Catch-up gate load-bearing: dispatch with `commit_min < commit_max`
- /// produces two register entries and panics on replay.
+ /// Catch-up gate load-bearing: a Register dispatched with
+ /// `commit_min < commit_max` can double-commit against an inherited one,
+ /// fencing the live client's fresh reply for no reason.
#[allow(clippy::future_not_send)]
pub async fn submit_register_in_process(
&self,
client_id: u128,
user_id: u32,
- ) -> Result<u64, MetadataSubmitError> {
+ ) -> Result<BoundSession, MetadataSubmitError> {
assert!(client_id != 0, "client_id 0 is reserved for internal use");
let consensus = self
.consensus
.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);
+ // 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
+ // caller's ops under that user (and `commit_register` would clobber
+ // 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.
+ {
+ let table = self.client_table.borrow();
+ if let Some(owner) = table.get_user_id(client_id)
+ && owner != user_id
+ {
+ warn!(
+ target: "iggy.metadata.diag",
+ client_id,
+ authenticated_user = user_id,
+ entry_owner = owner,
+ "refusing register: client id is registered to a different user"
+ );
+ return Err(MetadataSubmitError::ClientIdOwnedByAnotherUser);
+ }
Review Comment:
this gate reads the replicated table authoritatively but sits above the primary check below it, and there's no `is_caught_up_primary` anywhere before it.
both sibling readers of this table gate first - `request_preflight` and `register_preflight` both call the catch-up gate, and `request_preflight` has a SAFETY comment justifying it as what makes the replica authoritative for session truth. this one consults the same table with neither guard, so a backup answers the ownership refusal instead of `NotPrimary` and the client never learns to redirect.
moving it below the catch-up check makes it consistent with the other two and stops a lagging or diverged replica from issuing a terminal refusal.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -519,13 +535,14 @@ where
user_id,
reply,
} => {
- let session = shard
+ let bound = shard
.plane
.metadata()
.submit_register_in_process(vsr_client_id, user_id)
.await
- .ok();
- let _ = reply.try_send(session);
+ .ok()
Review Comment:
`.ok()` throws the error away, so `ClientIdOwnedByAnotherUser` arrives as `None` and `submit_register_on_owner` turns it into `Canceled`. `LoginRegisterError::is_terminal` then reads `Canceled.is_transient()` as true and answers the login as transient.
that matters because login is special-cased twice in the sdk: `tcp_client.rs` gives it the full `RESPONSE_READ_TIMEOUT` instead of the 2 s transient window, and the failover arm explicitly excludes login codes. so the client replays against a 30 s budget at a 50 ms interval with no leader redirect - order of 600 attempts - and every one costs a full argon2 verify on the home shard's event loop (`Argon2::default()`, m=19456 KiB).
only shard 0 takes the in-process path, so this is every connection that round-robins elsewhere.
putting `BoundSession` plus the real error on this channel fixes it and drops the untyped `(u64, u64)` tuple at the same time.
##########
core/consensus/src/metadata_helpers.rs:
##########
@@ -116,29 +129,55 @@ where
// Session evicted under capacity pressure. SAFETY: catch-up gate makes
// this replica authoritative for session truth.
RequestStatus::NoSession => PreflightOutcome::Evict(EvictionReason::NoSession),
Review Comment:
this claim doesn't hold for a node whose table was rebuilt from a truncated wal.
every clause of `is_caught_up_primary` is about the log - applied what it holds, suffix re-earned quorum. none of it says anything about how the table was built. but `recover()` starts from the local snapshot floor, and checkpoints are node-local: `checkpoint_if_needed` fires on local journal occupancy, the floor is the local `commit_min`, and the margin is per-node config. so two replicas cross the boundary at different ops.
the apply paths are deterministic, the initial condition isn't, and this gate checks the wrong quantity to bridge them. either the comment needs correcting or the check needs adding - the second is the already-acknowledged table-in-snapshot work.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -1067,21 +1175,82 @@ where
.prepare_request(request)
.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),
- 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.
- self.client_table
- .borrow()
- .get_session(client_id)
- .ok_or(MetadataSubmitError::Canceled)
+ // Same commit/cancel collapse as the queued path above.
+ let _ = self.dispatch_prepare_and_await(consensus, prepare).await;
Review Comment:
`let _ =` discards the committed reply, and `bound_session` then re-reads the table. the reply header already carries this register's own epoch - `build_reply_message` stamps `commit` from the prepare's op.
this is a regression against master, which had `Ok(reply) => Ok(reply.header().commit)` on the queued arm and only re-read the table on cancel. the cancel path is where it bites: if the register is canceled by a view change but some earlier entry for that `client_id` survives, `bound_session` hands back the old epoch and the login succeeds with a fence that never moved.
not reachable today since nothing re-presents a client id, but it goes live the moment sdk identity stability lands, and it's silent when it fires - a stale epoch passes `check_request`'s equality test, so no `Fenced`, no `EpochAhead`, nothing.
taking `epoch` from `reply.header().commit` and reading only `watermark` from the table is the two-line fix.
##########
core/server-ng/src/http/session.rs:
##########
@@ -37,10 +37,12 @@ use tokio::sync::Mutex;
///
/// Bounded by the shared VSR client table: HTTP sessions and the TCP/QUIC/WS
/// virtual clients all Register into the one [`CLIENTS_TABLE_MAX`]-slot table,
-/// which LRU-evicts the oldest 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 residual eviction race (both planes
+/// 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
Review Comment:
this rationale doesn't hold as written. the added clause says a dropped connection releases its slot once the reclaim grace expires, but the sweep that would do that is behind `heartbeat.enabled`, which ships false - so a dropped non-consumer-group connection holds its slot for the process lifetime, which is the case the sentence says doesn't happen.
the line is also over 100 columns after the edit.
##########
core/metadata/src/impls/recovery.rs:
##########
@@ -200,6 +211,26 @@ where
continue;
}
Review Comment:
the epochs part is true - they're op-derived so they replay identically. the watermark isn't.
if an earlier register for a client fell below this node's snapshot floor but a later one survived, replay takes `commit_register`'s fresh-entry branch and the entry comes back with a valid epoch and `watermark = 0`, where a peer that took the rebind branch preserved it. the fence passes, so nothing is evicted, and the same request id a peer answers `Duplicate` gets answered `New` and re-executed.
no shipping client can observe this today since none re-presents a recovered id, but it becomes a silent exactly-once to at-least-once violation once identity stability lands. the comment asserting cross-replica determinism here is what would stop a reader from asking.
the test suite can't catch it either - `client_table_restart.rs` uses a two-op wal, so the snapshot floor is always 0. one test that forces a checkpoint before the restart would turn this into a red/green fact.
##########
core/sdk/src/vsr.rs:
##########
@@ -47,11 +47,10 @@ 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. TigerBeetle avoids session resume and relies on
-// idempotency for requests retried from a new client session. For Iggy, either stop
-// transparent retries for replicated mutations after VSR session reset, add explicit
-// session resume/rebind semantics, or add a protocol-level idempotency key that is
-// independent of (client_id, request_id).
+// 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
Review Comment:
this todo points the next implementer at `try_resume_session`, which this pr deletes as a pre-auth session takeover. so the guidance is now to build against a mechanism that was removed for security reasons.
awkward because the integration test module doc nominates "the `sdk/vsr.rs` retry todo" as the remaining client half - so the one file the pr names as the follow-up is the one file that misdirects it.
the advice underneath is still right (keep the session across reconnects, retry under the same identity), it just needs to say resume happens through the login path now.
##########
core/binary_protocol/src/consensus/header.rs:
##########
@@ -152,12 +152,22 @@ pub struct RequestHeader {
pub reserved_frame: [u8; 66],
pub client: u128,
+ /// Integrity stamp over the request payload, used by the client table to
+ /// catch a `request` number reused for a different operation: a retry that
+ /// disagrees with the stamp of the cached reply is refused rather than
+ /// answered with the wrong reply. Zero means unstamped, which disables the
+ /// comparison; the wire currently sends zero.
pub request_checksum: u128,
pub timestamp: u64,
pub request: u64,
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
Review Comment:
last sentence is no longer true - `commit_register` sets the epoch to `reply.header().commit`, which is the register's commit op, so it derives from the log directly. `client_table.rs` says so in this same commit ("op-derived fence", "the commit number becomes the session number").
also "bumped by the server on each committed re-register" reads as a counter, which it was before this pr and isn't now.
worth getting exact since this struct is the wire contract.
##########
core/server-ng/src/session_manager.rs:
##########
@@ -166,6 +181,66 @@ 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) {
Review Comment:
this removes by `client_id` alone, so it can delete a reclaim armed for a newer epoch.
the sequence: sweeper takes the reclaim for epoch A, re-arms, spawns the logout. client reconnects and `bind_session` cancels. client disconnects again, arming epoch B. the in-flight logout for A hits the epoch guard in `submit_logout_in_process`, which returns `Ok(commit_min)` - a success value for a logout that logged nothing out - and the spawned task calls `complete_reclaim`, dropping B's entry. slot then leaks until capacity eviction, which is exactly what the grace window exists to prevent.
not reachable while the sweep is gated off, but it goes live the moment that's fixed. compare-and-remove on `(client_id, session)` covers it.
same applies to the demoted-primary case - `submit_logout_in_process` returns `NotPrimary` there and the sweeper re-arms every tick forever, using the grace constant as the retry backoff.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex merged PR #3753:
URL: https://github.com/apache/iggy/pull/3753
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3753:
URL: https://github.com/apache/iggy/pull/3753#discussion_r3657986828
##########
core/consensus/src/client_table.rs:
##########
@@ -345,33 +401,47 @@ 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.latest().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,
Review Comment:
Remedy landed: neither apply nor replay asserts anymore. `commit_reply` returns a typed `CommitReply { Cached, NoEntry, SkippedRegression { stored, received } }` and the callers log it (`warn!` for the regression, trace for the missing entry). The wire reply ships either way, so a skip degrades one entry's dedup instead of taking down a shard pump or blocking boot.
The reachability chain does not hold though, which matters for whether this is a regression or a latent hazard.
The shape needs two `Register(X)` for the same X inside one replay window, and a second Register is only ever proposed when the table has no entry for X. That is the same fast path you point out on `submit_register_in_process`: entry present means the epoch is returned and nothing is proposed.
- Within one process, absence means `evict_oldest` dropped it, so X has to come back. The Rust SDK abandons the key on re-login (`*self = Self::new()`, as you note on dispatch.rs:2553) and the HTTP counter is monotonic per process, so no id repeats.
- Across a restart, absence means the entry was not recovered, which means `Register(X)@a` fell below the replay floor. Replay then never sees `@a`, so there is no rebind branch to take and the watermark starts at 0.
`R < 5` via HTTP renumbering needs the cross-restart case, which is the one that cannot also supply `@a`.
It is also not introduced here. On master the same WAL panics through a different door: `Register(X)@c` carries a different `commit`, so `commit_register` takes its session-mismatch warn-and-skip arm and leaves the old entry in place with `reply.request == 5`, and the next `commit_reply` for `R < 5` fails `assert!(new_request >= slot_request)`. Same input, same panic, before this PR.
Fixed regardless: asserting on state that diverges replica-locally was the wrong failure mode whether or not the current tree can produce the input, and the epoch param went with it (your client_table.rs:375 comment).
##########
core/consensus/src/metadata_helpers.rs:
##########
@@ -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 { .. } => {
Review Comment:
Payload is logged now: `warn!` with `client_id`, `request` and `watermark`, since a run of these means the reply ring is undersized for the client's retry latency.
The last line does not hold as scoped. A correct client has at most one request in flight per session, which is the invariant the ring capacity doc rests on, so the in-doubt id is by definition its own watermark. The watermark's reply is structurally the newest ring entry: `push_latest` pops the front, and only a strictly newer request triggers a pop. So the watermark's reply cannot age out while it is still the watermark, and retrying an in-doubt id resolves as `Duplicate`, never here. Reaching `AlreadyApplied` needs a request strictly below the watermark whose reply has aged out, which means either more than one op in flight or a return to an older id.
The path where a well-behaved client does land below an inherited watermark is the post-restart renumbering on recovery.rs:220. The cross-user half of that is closed now, `submit_register_in_process` requires the authenticated user to own the entry before binding. The same-user half remains: a fresh session that collides with its own prior entry inherits a watermark it has no way to read, which is the DX gap on client_table.rs:577.
Kept as a drop rather than an empty Reply. The requests that reach it did commit, so `InvalidCommand` would be a wrong answer rather than a fast one, and the clients that can reach it are misnumbering. Happy to emit the reply instead if you would rather have fail-fast there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3753:
URL: https://github.com/apache/iggy/pull/3753#issuecomment-5089638420
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3753?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 73.06%. Comparing base ([`c5fec1f`](https://app.codecov.io/gh/apache/iggy/commit/c5fec1ff646e7513619c771cea909188526a7141?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`9244af7`](https://app.codecov.io/gh/apache/iggy/commit/9244af7ae70a981ab153df23f0abc4e644387852?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3753 +/- ##
============================================
- Coverage 74.54% 73.06% -1.49%
Complexity 969 969
============================================
Files 1306 1304 -2
Lines 150028 149581 -447
Branches 125462 125172 -290
============================================
- Hits 111840 109288 -2552
- Misses 34686 36875 +2189
+ Partials 3502 3418 -84
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.81% <ø> (-0.03%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.71% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.28% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `92.27% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.52% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `80.77% <ø> (-11.48%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3753/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `18.18% <ø> (-24.91%)` | :arrow_down: |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3753?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/binary\_protocol/src/consensus/operation.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fconsensus%2Foperation.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvbnNlbnN1cy9vcGVyYXRpb24ucnM=) | `96.69% <ø> (ø)` | |
| [core/consensus/src/client\_table.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fconsensus%2Fsrc%2Fclient_table.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25zZW5zdXMvc3JjL2NsaWVudF90YWJsZS5ycw==) | `95.70% <ø> (-0.38%)` | :arrow_down: |
| [core/consensus/src/metadata\_helpers.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fconsensus%2Fsrc%2Fmetadata_helpers.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9jb25zZW5zdXMvc3JjL21ldGFkYXRhX2hlbHBlcnMucnM=) | `88.22% <ø> (-1.09%)` | :arrow_down: |
| [core/metadata/src/impls/metadata.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fimpls%2Fmetadata.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvaW1wbHMvbWV0YWRhdGEucnM=) | `69.21% <ø> (+0.03%)` | :arrow_up: |
| [core/metadata/src/impls/recovery.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fimpls%2Frecovery.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvaW1wbHMvcmVjb3ZlcnkucnM=) | `79.71% <ø> (ø)` | |
| [core/metadata/src/stm/user.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fmetadata%2Fsrc%2Fstm%2Fuser.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9tZXRhZGF0YS9zcmMvc3RtL3VzZXIucnM=) | `87.95% <ø> (ø)` | |
| [core/server-ng/src/bootstrap.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fserver-ng%2Fsrc%2Fbootstrap.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXItbmcvc3JjL2Jvb3RzdHJhcC5ycw==) | `11.74% <ø> (+0.02%)` | :arrow_up: |
| [core/server-ng/src/dispatch.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fserver-ng%2Fsrc%2Fdispatch.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXItbmcvc3JjL2Rpc3BhdGNoLnJz) | `33.38% <ø> (ø)` | |
| [core/server-ng/src/users.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fserver-ng%2Fsrc%2Fusers.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXItbmcvc3JjL3VzZXJzLnJz) | `65.43% <ø> (ø)` | |
| [core/shard/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree&filepath=core%2Fshard%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zaGFyZC9zcmMvbGliLnJz) | `54.14% <ø> (ø)` | |
| ... and [1 more](https://app.codecov.io/gh/apache/iggy/pull/3753?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [101 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3753/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3753:
URL: https://github.com/apache/iggy/pull/3753#discussion_r3658642751
##########
core/consensus/src/client_table.rs:
##########
@@ -260,55 +322,44 @@ 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;
+ // Drop the previous register reply (if still retained) before
+ // pushing the new one: only the newest rebind's reply is
+ // replayable, and two request-0 entries would break the ring's
+ // unique-request invariant.
+ entry
+ .ring
+ .retain(|stored| stored.header().request != REGISTER_REQUEST_ID);
+ entry.push_latest(cached);
} 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");
+ let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY);
+ ring.push_back(cached);
self.slots[slot_idx] = Some(ClientEntry {
- session,
+ epoch: 1,
Review Comment:
Correcting myself on both counts here: I declined the remedy, and I was wrong. It has landed.
`commit_register` now sets the entry's epoch to the register's commit op (`reply.header().commit`, stamped from the prepare's op by `build_reply_message`) instead of `1, +1 per rebind`.
What I missed when I argued for the counter: a counter resets. Drop an entry by capacity eviction and re-register the same key, and the fresh entry is back at epoch 1 -- indistinguishable from the zombie still holding epoch 1 from before the eviction, so the fence is blind exactly where it is most needed. An op-derived fence cannot regress, because ops do not. That is your "log-derived identity" point, and it is the property the counter was quietly trading away.
My "not comparable across planes" objection conflated the fence *value* with the table's *identity*. The key stays client-supplied; only the fence is op-derived. Register commits solely in the metadata group, so there is exactly one minting authority and the value is a plain `u64` for any group's slice to store and compare. Our own design note said as much before the counter drifted in ("Register is always a metadata-plane op, so there is exactly one minting authority. Monotonic within the metadata group = globally comparable everywhere") -- I argued against a decision that had already been made correctly.
The inertness half of your diagnosis is fixed alongside it, because deriving the value changes nothing if no bind ever commits. `submit_register_in_process` lost its fast-path short-circuit and `register_preflight` no longer consults the table (`check_register` and `RequestStatus::AlreadyRegistered` are deleted), so every bind proposes. The ownership gate stays where it was and still refuses another user's entry before dispatching; an owned entry now falls through to a real Register, whose rebind branch refences the entry while preserving its watermark and reply ring. `Fenced`/`EpochAhead` are reachable on the live path for the first time.
One thing that falls out for free: the reconnect-versus-reclaim-sweep race is closed. A sweep's in-flight `Logout` carries the pre-rebind epoch, and `submit_logout_in_process`'s epoch guard drops it rather than tearing down the session the client just rebound.
Cost is a consensus round trip per reconnect, which is the honest price of a bind being a logged event.
##########
core/consensus/src/client_table.rs:
##########
@@ -412,9 +482,9 @@ impl ClientTable {
/// state -> identical choice. `commit_journal` catch-up has empty pipeline,
Review Comment:
Withdrawing the second half of this reply. The determinism fix and the provenance note stand; the "it is an improvement, not a regression" argument does not, and it was reasoning from the wrong place.
The trace I gave was accurate as far as it went: master takes the "newer than issued (client bug)" arm and silently drops, this branch produces `Fenced` -> terminal eviction, and an eviction does recover faster than a read timeout. But comparing failure modes was the wrong exercise. The epoch was not supposed to be *diverging* on either branch. I used master's worse handling of a bad state to argue our bad state was fine, instead of asking why the state was reachable at all -- which is what your `client_table.rs:370` comment was pointing at, and which I also pushed back on there.
The actual answer is that a per-entry counter was the wrong derivation, and it is now the register's commit op (see the reply on client_table.rs:370). Under that scheme the spurious-fence case I was defending cannot arise: epochs come from log positions, so replicas that agree on the log agree on every entry's fence, whether or not they made the same local eviction decision. There is no divergent-epoch state left to characterize as better or worse.
The `in_flight` removal you asked for is what makes the victim choice agree in the first place, and that part shipped as described.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #3753:
URL: https://github.com/apache/iggy/pull/3753#issuecomment-5100413605
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3753:
URL: https://github.com/apache/iggy/pull/3753#discussion_r3656261037
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -995,9 +1011,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) {
Review Comment:
the fast path returns the existing entry's epoch without checking that the authenticated user matches `entry.user_id`, and it never proposes a Register, so `commit_register` - the only writer that refreshes `user_id` - never runs. the vsr key is caller-supplied on the wire (`let vsr_client_id = request.header().client;` in the login handler at dispatch.rs:2687), so a client authenticating with its own valid low-privilege credentials but presenting someone else's client id gets bound to that entry, and every later replicated op resolves its acting user from it. the frame is well formed (session 0, request 0) so header validation passes it cleanly.
scope is replicated metadata ops - partition ops resolve the user from `SessionManager` instead, so those stay on the attacker's own identity. it needs the victim's client id, which is a uuidv4 for sdk clients but is logged and sent cleartext on every frame, and is sequential for http.
the entry doc already claims "re-register refreshes it: the rebind re-authenticated", so the check looks intended and simply absent. requiring `entry.user_id == authenticated user_id` here closes this and the restart case below in one edit.
##########
core/metadata/src/impls/recovery.rs:
##########
@@ -200,6 +211,26 @@ where
continue;
}
+ // 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.
+ if header.operation == Operation::Register {
+ let reply = build_reply_message(header, &bytes::Bytes::new());
+ client_table.commit_register(header.client, header.user_id, reply, |_| false);
Review Comment:
this is the first change that makes the table survive a reboot: replay rebuilds entries keyed by the previous boot's client ids. but neither id minter is seeded from recovered state - `client_seq` in the shard coordinator is `Cell::new(1)` per process, and http uses that minted value directly as the vsr table key. both sequences restart at 1, so the first post-restart login can land on a live recovered entry owned by a different user, and the fast path above then hands it that entry without refreshing `user_id`.
two effects. replicated ops execute under the prior boot's authority, and that stamp goes into the replicated prepare header, so into the durable log and the next snapshot. and the new session numbers its requests from 1 against an inherited watermark, so its early writes read as duplicates and it gets the previous user's cached reply bytes decoded as an answer to a command it never sent, or `AlreadyApplied` and the writes silently vanish.
no attacker needed - this fires on the scenario the pr is built for. it needs the prior Register still inside the replayed window and a different login order across restarts. worth noting the exposure is inversely correlated with traffic: checkpointing is what drains the stale Register, so a quiet deployment stays exposed indefinitely.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -578,11 +578,142 @@ where
.ok();
let _ = reply.try_send(commit);
}
+ shard::MetadataSubmit::ResumeLookup {
+ vsr_client_id,
+ reply,
+ } => {
+ let entry = lookup_resumable_session(&shard, vsr_client_id);
+ let _ = reply.try_send(entry);
+ }
}
});
})
}
+/// Shard 0's read side of a session-resume attempt: `(epoch, user_id)` for a
+/// registered client, `None` otherwise. Table reads are only authoritative on
+/// a caught-up primary (same rule as `request_preflight`'s gate); a stale
+/// answer here would rebind a transport to a session the cluster has already
+/// rotated, so fail closed and let the client retry.
+fn lookup_resumable_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ vsr_client_id: u128,
+) -> Option<(u64, u32)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ let metadata = shard.plane.metadata();
+ let consensus = metadata.consensus.as_ref()?;
+ if !consensus::is_caught_up_primary(consensus) {
+ debug!(
+ vsr_client_id,
+ is_primary = consensus.is_primary(),
+ commit_min = consensus.commit_min(),
+ commit_max = consensus.commit_max(),
+ "resume lookup refused: not a caught-up primary"
+ );
+ return None;
+ }
+ let table = metadata.client_table.borrow();
+ let entry = table
+ .get_epoch(vsr_client_id)
+ .zip(table.get_user_id(vsr_client_id));
+ if entry.is_none() {
+ debug!(vsr_client_id, "resume lookup: no table entry");
+ }
+ entry
+}
+
+/// Rebind a reconnecting transport that presents its pre-restart identity
+/// (IGGY-137 session resume, the implicit-rebind contract): a replicated
+/// request on an unbound transport whose `(client, session)` matches a live
+/// entry in the replicated client table binds this connection to that
+/// session and proceeds as if bound all along.
+///
+/// The `(client_id, epoch)` pair acts as a bearer token here: the client id
+/// is a client-generated random u128, so presenting it proves the caller is
+/// (or eavesdropped on) the original registrant. `bind_session` evicts any
+/// previous connection bound to the same client id, which is the conflict
+/// rule for one session arriving on two connections.
+///
+/// Returns the `(client_id, session)` binding on success, `None` when the
+/// identity does not check out (caller falls back to the unbound-transport
+/// reply and the client re-registers).
+#[allow(clippy::future_not_send)]
+async fn try_resume_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ header: &RequestHeader,
+) -> Option<(u128, u64)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ // A resumable frame carries the full old identity; anything less is a
+ // plain unbound request (e.g. a fresh SDK that has not registered yet).
+ if header.client == 0 || header.session == 0 || header.request == 0 {
+ return None;
+ }
+
+ let entry = if shard.id == 0 {
+ lookup_resumable_session(shard, header.client)
+ } else {
+ let (reply, rx) = shard::channel::<Option<(u64, u32)>>(1);
+ shard.forward_metadata_submit(shard::MetadataSubmit::ResumeLookup {
+ vsr_client_id: header.client,
+ reply,
+ });
+ rx.recv().await.ok().flatten()
+ };
+ let (epoch, user_id) = entry?;
+ if epoch != header.session {
+ // Stale epoch = zombie holdover (the preflight would fence it);
+ // future epoch = client bug. Neither may rebind.
+ warn!(
+ transport_client_id,
+ client = header.client,
+ presented = header.session,
+ current = epoch,
+ "refusing session resume with mismatched epoch"
+ );
+ return None;
+ }
+
+ {
+ let mut sessions = sessions.borrow_mut();
+ if let Err(error) = sessions.login(transport_client_id, user_id) {
Review Comment:
resume binds an unbound transport to an existing table session and logs it in as the entry's cached `user_id` while taking no credential at all. no user-exists check, no `status == Active` (the normal login path does check it in auth.rs), no pat expiry. authority then resolves from the table, so the connection runs as the original registrant, and `bind_session` also evicts the legitimate holder.
the doc above treats `(client_id, epoch)` as a bearer token on the premise that the client id is a client-generated random u128. that holds for the rust sdk but not for http/quic/tcp-tls, which draw from the sequential per-process counter, and the epoch half contributes no entropy because nothing on the live path ever bumps it past 1. so for an http-originated entry the credential is `client=N, session=1`. the table also carries no plane or transport tag, so a raw tcp peer can bind to an http session.
separately exploitable with no guessing at all by a deactivated or deleted user: `remove_client` only runs on Logout apply and replay, so neither DeleteUser nor a status flip drops the session, and the authz layer never re-checks status. user slab ids are reused, so a stale entry later resolves to whoever occupies that index.
fix is to re-authenticate on resume and bind only when `entry.user_id` matches the authenticated user.
##########
core/consensus/src/metadata_helpers.rs:
##########
@@ -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) => {
Review Comment:
`CreatePersonalAccessToken` mints a fresh raw token and hash on every attempt, before any preflight (`rewrite_pat_request_for_user` in server-ng's pat.rs). when that request comes back a duplicate, this arm hands the cached committed reply to the caller and it flows straight into `build_raw_pat_reply`. both of that function's guards pass - the reply is a `Command2::Reply`, and a cached successful create carries result code 0 - so the newly minted raw token gets spliced into a reply whose committed hash belongs to the original token. the caller receives a credential that can never authenticate, and the real raw secret only ever existed at the original mint, so it is gone.
this was already reachable at `request == watermark` before, but the reply ring widens the replayable set from one id to every id the ring retains. fix is to carry a "this was a replay" flag out of the preflight and pass the cached reply through untouched - the same reasoning `build_raw_pat_reply` already applies to the eviction case.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -2397,6 +2542,21 @@ fn submit_disconnect_logout<B, MJ, S>(
// The logout apply keys on (client, session) only, so any non-zero id
// is valid here.
const DISCONNECT_LOGOUT_REQUEST_ID: u64 = u64::MAX;
+
+ let is_group_member = !shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(vsr_client_id)
+ .is_empty();
+ if !is_group_member {
Review Comment:
a non-group-member disconnect returns without submitting Logout, the heartbeat verifier only reaps group members, and there is no ttl (`ClientEntry` has no timestamp field; the table defers it). meanwhile the reference sdk abandons the key on every re-login - `begin_register` does `*self = Self::new()` and the sdk module doc spells out the consequence, that the old entry stays in the server `ClientTable` until evicted. before this change the disconnect Logout reclaimed the entry before the sdk abandoned the id; now the two halves compose and the slot leaks.
then `evict_oldest` ranks by last-write recency, which is anti-correlated with liveness: a live idle admin connection gets evicted in preference to a recently-leaked dead entry.
on its own this is bounded by the table size and an explicit Logout still frees a bound slot, so it is not an authz break by itself. what makes it matter is that it moves the eviction-per-register point from ~4096 simultaneous connections to ~4096 cumulative connects, which any cli-driven or health-probed cluster reaches on a timer - and eviction is exactly the unlogged drop that the findings above turn into re-execution and a replay panic.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -578,11 +578,142 @@ where
.ok();
let _ = reply.try_send(commit);
}
+ shard::MetadataSubmit::ResumeLookup {
+ vsr_client_id,
+ reply,
+ } => {
+ let entry = lookup_resumable_session(&shard, vsr_client_id);
+ let _ = reply.try_send(entry);
+ }
}
});
})
}
+/// Shard 0's read side of a session-resume attempt: `(epoch, user_id)` for a
+/// registered client, `None` otherwise. Table reads are only authoritative on
+/// a caught-up primary (same rule as `request_preflight`'s gate); a stale
+/// answer here would rebind a transport to a session the cluster has already
+/// rotated, so fail closed and let the client retry.
+fn lookup_resumable_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ vsr_client_id: u128,
+) -> Option<(u64, u32)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ let metadata = shard.plane.metadata();
+ let consensus = metadata.consensus.as_ref()?;
+ if !consensus::is_caught_up_primary(consensus) {
+ debug!(
+ vsr_client_id,
+ is_primary = consensus.is_primary(),
+ commit_min = consensus.commit_min(),
+ commit_max = consensus.commit_max(),
+ "resume lookup refused: not a caught-up primary"
+ );
+ return None;
+ }
+ let table = metadata.client_table.borrow();
+ let entry = table
+ .get_epoch(vsr_client_id)
+ .zip(table.get_user_id(vsr_client_id));
+ if entry.is_none() {
+ debug!(vsr_client_id, "resume lookup: no table entry");
+ }
+ entry
+}
+
+/// Rebind a reconnecting transport that presents its pre-restart identity
+/// (IGGY-137 session resume, the implicit-rebind contract): a replicated
+/// request on an unbound transport whose `(client, session)` matches a live
+/// entry in the replicated client table binds this connection to that
+/// session and proceeds as if bound all along.
+///
+/// The `(client_id, epoch)` pair acts as a bearer token here: the client id
+/// is a client-generated random u128, so presenting it proves the caller is
+/// (or eavesdropped on) the original registrant. `bind_session` evicts any
+/// previous connection bound to the same client id, which is the conflict
+/// rule for one session arriving on two connections.
+///
+/// Returns the `(client_id, session)` binding on success, `None` when the
+/// identity does not check out (caller falls back to the unbound-transport
+/// reply and the client re-registers).
+#[allow(clippy::future_not_send)]
+async fn try_resume_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ header: &RequestHeader,
+) -> Option<(u128, u64)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ // A resumable frame carries the full old identity; anything less is a
+ // plain unbound request (e.g. a fresh SDK that has not registered yet).
+ if header.client == 0 || header.session == 0 || header.request == 0 {
Review Comment:
this guard filters nothing. ingress validation already enforces `client != 0` for all ops and `session > 0 && request > 0` for every non-Register replicated op, so the only frame it can still catch is `Operation::Reserved` - and the comment's "a fresh sdk that has not registered yet" describes a class that cannot reach here at all, since such a client's `session == 0` is rejected at ingress.
so every real unbound replicated frame reaches the lookup, which on shards other than 0 is a cross-shard message plus an await on the metadata owner shard, before any credential check, with attacker-chosen ids and no memoization. the bind also happens before the `is_partition()` branch, so data-plane frames pay it too. it is bounded and fails closed (`try_send` drops), so this is cost and surprise rather than a break.
either delete the guard or reduce it to a `debug_assert`, and fix the comment.
##########
core/consensus/src/metadata_helpers.rs:
##########
@@ -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 { .. } => {
Review Comment:
`AlreadyApplied` maps to a silent drop and its `{ request, watermark }` payload is thrown away by `{ .. }`, while all three sibling arms log. no reply frame goes out on either plane.
the convention this file and the dispatch path already settled on is terminal -> reply, transient -> silence; the comment on the unbound-transport case says outright that the sdk blocked until socket timeout, so emit an empty Reply and let it fail fast. this is terminal. server-ng does log something, but a `warn!` that labels it transient and carries no request or watermark, so the one log that exists cannot diagnose it.
the silence predates this change; what is new is that a correct client retrying an in-doubt id can land here.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -1049,14 +1068,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
Review Comment:
after the rework both arms here are byte-identical, and so are both arms of the match below. `let _ = <fut>.await;` followed by a single `get_epoch` read says the same thing. the two rationale comments are load-bearing and should stay.
##########
core/consensus/src/client_table.rs:
##########
@@ -453,21 +524,31 @@ 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()`.
#[must_use]
pub fn get_reply(&self, client_id: u128) -> Option<&CachedReply> {
let &slot_idx = self.index.get(&client_id)?;
- self.slots[slot_idx].as_ref().map(|entry| &entry.reply)
+ self.slots[slot_idx].as_ref().map(ClientEntry::latest)
}
- /// 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> {
Review Comment:
`get_watermark` has no non-test callers, and the doc's promise that "a (re)bind reply surfaces this" is not implemented: `LoginRegisterResponse` is exactly `{user_id, session, server_protocol_version, server_version}` and `ReplyHeader.context` is hardcoded 0, so there is no side channel either.
that is the dx gap sitting under the whole feature - without the watermark in the bind reply, no sdk can implement the resume numbering this pr exists to enable. either wire it into the reply or drop the accessor and the doc claim.
##########
core/consensus/src/client_table.rs:
##########
@@ -412,9 +482,9 @@ impl ClientTable {
/// state -> identical choice. `commit_journal` catch-up has empty pipeline,
Review Comment:
this determinism claim does not hold. the `in_flight` input reads the local pipeline, which only a primary populates - `push_prepare_entry` asserts `is_primary()` and `has_message_from_client` scans only local queues. same committed log, different victim on primary vs backup, so table membership and epoch counters diverge. the next sentence concedes the input is pipeline state and then draws the opposite conclusion from it: the prepare queue holds uncommitted primary-local prepares, so it is not agreed-log-derived.
more precisely it is false on any replica whose pipeline is empty, which is wider than just backups - a demoted primary keeps its queue until view change completes.
this split is what makes the replay panic above reachable on a backup. the change also drops the `warn!` that used to make table mismatches visible.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -578,11 +578,142 @@ where
.ok();
let _ = reply.try_send(commit);
}
+ shard::MetadataSubmit::ResumeLookup {
+ vsr_client_id,
+ reply,
+ } => {
+ let entry = lookup_resumable_session(&shard, vsr_client_id);
+ let _ = reply.try_send(entry);
+ }
}
});
})
}
+/// Shard 0's read side of a session-resume attempt: `(epoch, user_id)` for a
+/// registered client, `None` otherwise. Table reads are only authoritative on
+/// a caught-up primary (same rule as `request_preflight`'s gate); a stale
+/// answer here would rebind a transport to a session the cluster has already
+/// rotated, so fail closed and let the client retry.
+fn lookup_resumable_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ vsr_client_id: u128,
+) -> Option<(u64, u32)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ let metadata = shard.plane.metadata();
+ let consensus = metadata.consensus.as_ref()?;
+ if !consensus::is_caught_up_primary(consensus) {
Review Comment:
this collapses the transient "not a caught-up primary" refusal into the same `None` as the terminal "no table entry", so the caller answers with the empty Reply that the sdk is documented (a few hundred lines down) to surface as `InvalidCommand` fail-fast rather than `TransientNotCommitted`.
on a multi-replica node the window is guaranteed, not incidental: `is_caught_up_primary` requires `commit_max >= recovery_barrier`, and bootstrap arms the barrier whenever `commit_watermark < restored_op`, which is always true post-restart in the non-solo branch because `commit_watermark` is folded from prepare `commit` stamps and so can never reach the journal head. so the client that can resume is told terminally that it cannot.
make it tri-state (found / no entry / refused) and emit `TransientNotCommitted` for refused.
##########
core/consensus/src/client_table.rs:
##########
@@ -260,55 +322,44 @@ 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;
+ // Drop the previous register reply (if still retained) before
+ // pushing the new one: only the newest rebind's reply is
+ // replayable, and two request-0 entries would break the ring's
+ // unique-request invariant.
+ entry
+ .ring
+ .retain(|stored| stored.header().request != REGISTER_REQUEST_ID);
+ entry.push_latest(cached);
} 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");
+ let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY);
+ ring.push_back(cached);
self.slots[slot_idx] = Some(ClientEntry {
- session,
+ epoch: 1,
Review Comment:
a fresh entry hardcodes `epoch: 1, watermark: REGISTER_REQUEST_ID`. so after any entry drop, a client re-registering under the same id gets a clean slate, and every request id it already committed is above 0, which means all of them read `New`, dispatch, and re-apply.
both pre-existing containments were removed together here: `session` used to be `prepare_header.op`, log-derived and non-regressing per key, and contiguity (`request == committed + 1`) capped how far a reset could reach. eviction and journal-window loss are local and unlogged, so nothing in the log tells a re-registering or recovering path what the entry used to hold.
the fence half is separately inert: no live proposing path bumps an epoch (the fast path returns early, and the preflight either replays or drops), so every entry a primary holds sits at epoch 1 and `Fenced`/`EpochAhead` never fire. zombie fencing, the stated design goal, does not happen in practice.
making the fence the register's op number restores a log-derived identity and fixes the replica-divergence problem below at the same time.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -578,11 +578,142 @@ where
.ok();
let _ = reply.try_send(commit);
}
+ shard::MetadataSubmit::ResumeLookup {
+ vsr_client_id,
+ reply,
+ } => {
+ let entry = lookup_resumable_session(&shard, vsr_client_id);
+ let _ = reply.try_send(entry);
+ }
}
});
})
}
+/// Shard 0's read side of a session-resume attempt: `(epoch, user_id)` for a
+/// registered client, `None` otherwise. Table reads are only authoritative on
+/// a caught-up primary (same rule as `request_preflight`'s gate); a stale
+/// answer here would rebind a transport to a session the cluster has already
+/// rotated, so fail closed and let the client retry.
+fn lookup_resumable_session<B, MJ, S>(
+ shard: &Rc<ShellShard<B, MJ, S>>,
+ vsr_client_id: u128,
+) -> Option<(u64, u32)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+{
+ let metadata = shard.plane.metadata();
+ let consensus = metadata.consensus.as_ref()?;
+ if !consensus::is_caught_up_primary(consensus) {
+ debug!(
+ vsr_client_id,
+ is_primary = consensus.is_primary(),
+ commit_min = consensus.commit_min(),
+ commit_max = consensus.commit_max(),
+ "resume lookup refused: not a caught-up primary"
+ );
+ return None;
+ }
+ let table = metadata.client_table.borrow();
+ let entry = table
+ .get_epoch(vsr_client_id)
+ .zip(table.get_user_id(vsr_client_id));
+ if entry.is_none() {
+ debug!(vsr_client_id, "resume lookup: no table entry");
+ }
+ entry
+}
+
+/// Rebind a reconnecting transport that presents its pre-restart identity
+/// (IGGY-137 session resume, the implicit-rebind contract): a replicated
+/// request on an unbound transport whose `(client, session)` matches a live
+/// entry in the replicated client table binds this connection to that
+/// session and proceeds as if bound all along.
+///
+/// The `(client_id, epoch)` pair acts as a bearer token here: the client id
+/// is a client-generated random u128, so presenting it proves the caller is
+/// (or eavesdropped on) the original registrant. `bind_session` evicts any
+/// previous connection bound to the same client id, which is the conflict
+/// rule for one session arriving on two connections.
+///
+/// Returns the `(client_id, session)` binding on success, `None` when the
+/// identity does not check out (caller falls back to the unbound-transport
+/// reply and the client re-registers).
+#[allow(clippy::future_not_send)]
+async fn try_resume_session<B, MJ, S>(
Review Comment:
resume skips the login preamble. `is_protocol_compatible` has exactly one call site, inside the login handler, and `record_sdk_info` two in auth.rs - resume calls neither. so after a server upgrade narrows the compatible window, a previously-registered client rebinds without re-presenting `ClientVersionInfo` and nothing else re-checks it. client listings also report sdk name and version as none for resumed sessions.
##########
core/consensus/src/client_table.rs:
##########
@@ -345,33 +401,47 @@ 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.latest().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,
Review Comment:
the rebind branch deliberately preserves the watermark, and a live capacity eviction leaves no log record (`evict_oldest` only mutates `slots`/`index`). so for a wal shaped `Register(X)@a, app(X,req=5)@b, [evict], Register(X)@c, app(X,req=R<5)@d` the live run is fine, because the eviction reset the watermark, but replay starts from an empty table, never evicts, takes the rebind branch, keeps watermark 5, and this assert fires inside `recover()`. deterministic on the same wal, so the node cannot boot without wal surgery, while every other replay failure surfaces as a typed `RecoveryError`.
there is a variant that needs no reboot: a backup that did not evict X hits the same sequence in `commit_journal` and panics its shard pump. the `evict_oldest` divergence noted below is what produces that primary/backup split. `R < 5` is reachable in-tree since http restarts numbering at 1.
apply and replay paths should log-and-skip or return a typed error here, not assert.
##########
core/metadata/src/impls/recovery.rs:
##########
@@ -188,6 +198,7 @@ where
.fold(snapshot_floor, u64::max)
};
+ let mut client_table = ClientTable::new(CLIENTS_TABLE_MAX);
Review Comment:
replay starts above the snapshot floor, `checkpoint` drains `0..=last_op`, the journal index is 1024 slots and its own doc warns that entries past a wrap become unreachable for recovery, and the table is not in the snapshot yet (serialization todo). so the advertised resume does not cover a session whose Register fell below the replay window.
the consequence here is a loud refusal rather than silent re-execution: a missing entry gives `NoSession`, which maps to `Evict(EvictionReason::NoSession)`. the silent part comes one step later, after the client re-registers and gets a fresh entry with watermark 0.
##########
core/consensus/src/client_table.rs:
##########
@@ -424,8 +494,9 @@ impl ClientTable {
for (idx, slot) in self.slots.iter().enumerate() {
let Some(entry) = slot else { continue };
- let commit = entry.reply.header().commit;
- let client_id = entry.reply.header().client;
+ let latest = entry.latest().header();
Review Comment:
this walks every slot inside shard 0's no-await commit region, and per occupied slot it does two `bytemuck::checked::try_from_bytes` header casts plus `in_flight`, which itself runs two linear scans over the prepare and request queues. then `first_free_slot` walks the array again. with an 8192-slot table that is a lot of synchronous work on the commit loop, and it compounds with the slot leak: once the table is full this runs on every register.
if you cache `commit`/`client_id` on the entry to avoid the casts, ship it together with the determinism fix above - caching alone would freeze the replica-dependent victim choice in place while making this function look reviewed.
##########
core/consensus/src/client_table.rs:
##########
@@ -156,52 +207,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
+ entry.find_cached(request).map_or(
Review Comment:
the checksum comparison above only runs for `request == entry.watermark`, so any below-watermark hit that finds a cached reply here returns it without ever comparing the stamp. a client that reused an old request id for a different operation gets the wrong operation's reply.
latent today rather than active: nothing writes a non-zero `request_checksum` - the sdk leaves it 0, `RequestHeader::default` zeroes it, and server-ng's wire layer says outright that the server does not validate it - so both `!= 0` conditions fail on every real request and the guard cannot fire at all. it becomes a real gap the moment integrity stamping lands, so worth closing the coverage now (compare at every `find_cached` hit) rather than after. keeping the field is the right call even though it is derivable from the cached reply - the ring is documented as lossy by design, and a correctness gate should not depend on a best-effort cache.
##########
core/consensus/src/client_table.rs:
##########
@@ -319,18 +370,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>) {
Review Comment:
the `epoch` param and its equality assert are unreachable by construction: all three production callers read the epoch from this same table one statement earlier with no await in between, on a single-threaded shard, and the boot installer cannot interleave. looking the epoch up inside `commit_reply` drops the param, the duplicated caller-side traces, one hash lookup and one RefCell borrow pair per committed metadata op, and one panic path off a `pub` api.
if a partition-plane caller later wants the tripwire back, it should return as an explicit skip rather than an assert.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -546,6 +547,14 @@ impl<C, J, S, M> IggyMetadata<C, J, S, M> {
*self.commit_notifier.borrow_mut() = notifier;
}
+ /// Install the client table rebuilt by WAL-replay recovery
+ /// ([`crate::impls::recovery::recover`]). Boot-time only, on the owning
+ /// shard, before it serves traffic - replacing a live table would drop
+ /// committed session state.
+ pub fn install_client_table(&self, client_table: ClientTable) {
Review Comment:
worth guarding this against a non-empty table (`count()` exists on `ClientTable`) so a stray second install cannot silently drop committed session state - the doc says boot-time only but nothing enforces it. prefer a `debug_assert` or a logged refusal over `assert!`, since a boot-path panic is a worse failure than a refusal. cheaper than threading the table through `IggyMetadata::new`, which has ten call sites.
##########
core/consensus/src/client_table.rs:
##########
@@ -71,19 +71,48 @@ 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.
+/// Committed replies retained per entry, newest at the back.
///
-/// `session` is assigned at registration and fixed for the entry's lifetime.
+/// The back is the latest committed reply and is structurally safe:
+/// eviction pops the front, and only pushing a newer reply triggers it.
+/// 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`).
+/// Older entries answer old retransmits and post-rebind stragglers with the
+/// original bytes instead of a bare "already applied"; losing one
+/// degrades the answer, never correctness. In-memory only: ring contents are
+/// refcount bumps and are never persisted or transferred.
+const REPLY_RING_CAPACITY: usize = 5;
+
+/// 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 {
Review Comment:
`ClientEntry` is `pub` with all six fields private and no reference outside this file, and `lib.rs` re-exports only `CachedReply` and `ClientTable`. it is still reachable as `consensus::client_table::ClientEntry` through the `pub mod`, so making it private is a real surface reduction for one keyword.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on PR #3753:
URL: https://github.com/apache/iggy/pull/3753#issuecomment-5093410562
Thanks for this review -- the identity model was the right thing to block on, and two of your findings turned out to be load-bearing in ways I argued against before coming around. Everything from the numbered blocker list is addressed; details are in the per-thread replies, and I have resolved the threads that are fully closed. Summary of what changed and what I deliberately left open.
## The identity model
**Fence is log-derived again.** `commit_register` sets the entry's epoch to the register's commit op rather than a per-entry counter. Your remedy #3, which I initially declined on the grounds that an op-derived value would not be comparable across planes -- that conflated the fence *value* with the table's *identity*. The key stays client-supplied; Register commits only in the metadata group, so there is one minting authority and any group's slice stores a plain `u64`. What the counter was quietly trading away is the thing you were pointing at: a counter *resets* when an entry is dropped by capacity eviction and re-created, so a zombie holding epoch 1 and a fresh incarnation holding epoch 1 are indistinguishable. Ops do not reset.
**Binds commit.** The fast-path short-circuit in `submit_register_in_process` is gone and `register_preflight` no longer consults the table (`check_register` and `RequestStatus::AlreadyRegistered` deleted), so every bind proposes and the epoch actually moves. `Fenced`/`EpochAhead` are reachable on the live path for the first time -- the inertness you flagged. The ownership gate stays ahead of the dispatch and still refuses another user's entry; an owned entry falls through to a real Register whose rebind branch preserves watermark and reply ring. Cost is one consensus round trip per reconnect, which is the price of a bind being a logged event.
**Credential-free resume is deleted, not gated.** `try_resume_session` and `MetadataSubmit::ResumeLookup` are gone; resume runs through the login path, which already did it correctly. That took findings #1, #2, #5, #6 and #9's first half with it.
**`client_seq` no longer starts fresh into a recovered table.** The coordinator reseeds the minter above the highest recovered client-id sequence at boot, before any listener accepts. This makes the restart re-mint collision *unreachable* rather than merely refused, which demotes the ownership gate to a backstop.
## Apply-path robustness
`commit_reply` returns a typed `CommitReply { Cached, NoEntry, SkippedRegression }` instead of asserting; callers log it. The wire reply ships either way, so a skip degrades one entry's dedup rather than taking down a shard pump or refusing to boot. The unreachable `epoch` param went with it. `install_client_table` is a logged refusal returning `bool`, not an `assert!` -- your point that a boot-path panic is worse than a refusal.
`evict_oldest` no longer consults pipeline state, so the victim choice depends only on committed state. It also no longer casts headers: `client_id` and `latest_commit` are denormalized onto the entry, and it returns the freed slot so `commit_register` skips the second array walk.
## Correctness
- PAT replay: a replayed `CreatePersonalAccessToken` is refused with `PersonalAccessTokenAlreadyExists` rather than served a cached success with a freshly minted secret spliced in. The complete fix is derivable secrets so a replay can reproduce the original token; that is a wire change and wants its own issue.
- HTTP request id is burned at stamp time, before any fallible work.
- `AlreadyApplied` now answers with a terminal `RequestAlreadyApplied` code instead of silence, so the client learns immediately instead of waiting out its read timeout. I pushed back on this initially; you were right that terminal deserves a frame.
- The checksum reuse check runs at every cached-reply hit, not just at the watermark.
- Slot reclaim: non-group-member disconnects defer reclaim by a grace window and a sweeper submits the `Logout`. Verifying that turned up a leak in my own fix -- the drained reclaim was dropped on a transient submit failure, leaving exactly the slot-until-eviction case your comment describes. The sweeper now re-arms before submitting and clears only on a committed `Logout`.
- `Operation::Reserved` is rejected at ingress, so the `Reserved, request = 0` path can no longer reach the dedup preflight ahead of the operation gate.
- Duplicate match arms collapsed; `ClientEntry` is private; `session`/`request_checksum` documented on all three headers; stale contiguity and `SessionMismatch` docs swept.
## Tests
Ingress validation is pinned (`RequestHeader::validate` through `try_into_typed`: the field rules table plus zero-client rejection), since several conclusions on both sides turned on whether it runs. The replay proof is now direct -- `commit_request` returns the committed reply and the retry is asserted byte-identical, with `op` equality being what rules out a re-apply; it no longer rests on `CreateStream` rejecting a duplicate name. Both restart specs run single-node and 3-node. New unit coverage for the reclaim retry, the minter reseed, the terminal refusal, and the denormalized eviction ranking.
## Open, by intent
- **Watermark surfaced to SDKs.** The gateway case is closed (the register submit returns `BoundSession { epoch, watermark }` and HTTP seeds its numbering above the inherited watermark), so no wire change was needed. An SDK resuming across its *own* process restart still needs it in the bind reply; that is tier-2 identity and wants the protocol discussion.
- **Sessions below the snapshot floor** (recovery.rs:201) -- needs the table in the checkpoint, which is the state-transfer work.
- **Status is not re-checked for bound sessions.** Deactivating a user does not stop its live sessions. Master has the identical gap (same two `remove_client` sites, same login-only check, no status reference in the gate), so it is not from this PR; deletion *is* enforced per-op because `DeleteUser` purges the permissioner and the rules fail closed. Filing separately, which is why I left that thread open.
- Replayed `Logout` group staleness, and `Reserved`'s post-preflight gate ordering beyond the ingress fix.
I also corrected two of my own earlier replies in this thread where the epoch rework made them wrong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): rework client table for session resume across restarts [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #3753:
URL: https://github.com/apache/iggy/pull/3753#discussion_r3658027139
##########
core/consensus/src/client_table.rs:
##########
@@ -412,9 +482,9 @@ impl ClientTable {
/// state -> identical choice. `commit_journal` catch-up has empty pipeline,
Review Comment:
The determinism claim was wrong exactly as you describe, and the `warn!` point is fair too. Fixed by deleting `in_flight` from `evict_oldest` and from `commit_register`'s signature rather than by caching the header fields: the victim is now oldest-commit with a lowest-slot tiebreak and nothing local feeds the choice. The doc now states the constraint positively, that pipeline state must not be consulted because only a primary populates it. A `warn!` also comes back on the other side of the same divergence, `commit_reply` now returns `SkippedRegression { stored, received }` and the caller logs it.
Two provenance corrections though.
`in_flight`, its two queue scans and the "pipeline state derives from the agreed log" sentence are all present on master, and `evict_oldest` diffs semantically clean against it. So the divergence is not introduced by this PR. Worth fixing here since the PR is what makes the table durable, but it is not a regression.
The consequence you draw from it is also better under this PR than before it, not worse. Take the same primary/backup split and follow it to what the client sees:
- master: the backup takes `commit_register`'s session-mismatch arm, skips the update and keeps `session = A`, while the client holds the newer `M` from the primary. On failover `check_request` sees `expected(A) < received(M)`, which is the "newer than issued, client bug" arm: error log, no eviction, no reply. The client waits out its read timeout before it learns anything.
- here: the backup bumps to epoch 2 while the client holds 1. On failover that is `Fenced`, which maps to a terminal `Evict(SessionTooLow)`. Eviction tells the client to re-register, so it recovers on the next round trip.
Same divergence, and the fence turns a silent wedge into a frame that names the problem. So the spurious-fence symptom is new, but it is an improvement over the silent drop it replaced.
##########
core/consensus/src/client_table.rs:
##########
@@ -260,55 +322,44 @@ 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;
+ // Drop the previous register reply (if still retained) before
+ // pushing the new one: only the newest rebind's reply is
+ // replayable, and two request-0 entries would break the ring's
+ // unique-request invariant.
+ entry
+ .ring
+ .retain(|stored| stored.header().request != REGISTER_REQUEST_ID);
+ entry.push_latest(cached);
} 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");
+ let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY);
+ ring.push_back(cached);
self.slots[slot_idx] = Some(ClientEntry {
- session,
+ epoch: 1,
Review Comment:
Diagnosis accepted, remedy declined.
The two observations hold. A fresh entry does reset epoch and watermark, and unlogged entry drops leave nothing in the log to reconstruct what the entry held. The fence is inert on the live path for the reason you give: the register fast path returns before proposing, so nothing bumps an epoch past 1 and `Fenced`/`EpochAhead` are unreachable there.
Making the fence the register's op number is the part I do not want. The table is deliberately free of any value derived from a commit op number, because one logical table spans both the metadata plane and the partition plane and those have independent op ranges. An op-derived fence is not comparable across them, so it buys log-derived identity for the metadata plane at the cost of the table no longer being one mechanism. That constraint is the reason `session` stopped being `prepare_header.op` in the first place, so restoring it is a design reversal rather than a fix.
The alternative that keeps the model is to make rebind actually propose a Register, i.e. delete the fast-path short-circuit, so the bump is real, logged and replica-identical without anything in the entry deriving from an op number. That is the change I would rather make, and it is not in this PR.
The "fixes the replica-divergence problem below at the same time" half is moot now regardless: the divergence is fixed at its source by removing pipeline state from the victim choice (see the reply on client_table.rs:482), so an op-derived fence is no longer load-bearing for it.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -995,9 +1011,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) {
Review Comment:
Fixed: the fast path now requires the authenticated user to own the entry, and a mismatch is a terminal `MetadataSubmitError::ClientIdOwnedByAnotherUser` so the SDK fails hard instead of spinning on it. The resume path that shared the hole is deleted outright rather than gated.
Two corrections to the framing.
The mechanism is not introduced here. Master's fast path is the same early return with no owner check (`get_session` where this branch reads `get_epoch`), and `resolve_acting_user_id` is byte-identical on master. What this PR changes is reachability, not the missing check: recovery now rebuilds the table across a restart, which is what makes the collision you describe on recovery.rs:220 a routine event rather than a nonexistent one. Both still needed fixing here, since this is the PR that makes them reachable.
The scope paragraph understates it in your own favour. "partition ops resolve the user from `SessionManager` instead, so those stay on the attacker's own identity" is true for the fast path, but false for resume: `try_resume_session` called `SessionManager::login` with the `user_id` it read out of the table, so `dispatch_partition_request` resolved the victim's identity too. The data plane was in scope for that path, not just replicated metadata ops.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org