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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
