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