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