This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch clients_table_v2 in repository https://gitbox.apache.org/repos/asf/iggy.git
commit d2461e340821784b7503cd740c177c99c75e4e9f Author: Grzegorz Koszyk <[email protected]> AuthorDate: Mon Jul 27 10:49:02 2026 +0200 feat(consensus): rework client table for session resume across restarts --- core/binary_protocol/src/consensus/operation.rs | 3 +- core/consensus/src/client_table.rs | 8 +- core/consensus/src/metadata_helpers.rs | 7 +- .../tests/cluster/client_table_restart.rs | 61 ++++--- core/metadata/src/impls/metadata.rs | 14 +- core/metadata/src/impls/recovery.rs | 178 ++++++++++++++++++++- core/metadata/src/stm/user.rs | 2 +- core/server-ng/src/bootstrap.rs | 17 +- core/server-ng/src/dispatch.rs | 178 +++++++++++++++++++-- core/server-ng/src/users.rs | 10 +- core/shard/src/lib.rs | 9 ++ 11 files changed, 422 insertions(+), 65 deletions(-) diff --git a/core/binary_protocol/src/consensus/operation.rs b/core/binary_protocol/src/consensus/operation.rs index 954623132..11d15a106 100644 --- a/core/binary_protocol/src/consensus/operation.rs +++ b/core/binary_protocol/src/consensus/operation.rs @@ -30,7 +30,8 @@ pub enum Operation { /// Register a client session with the cluster. Goes through the same /// consensus pipeline (prepare/replicate/commit) as normal operations /// but skips state machine dispatch at commit time, the metadata - /// plane calls `commit_register` directly. Session number = commit op. + /// plane calls `commit_register` directly, which mints the session's + /// fence epoch (1 at first register, +1 per rebind). Register = 1, /// Non-replicated client request carried in VSR framing. The concrete diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index 09a87292e..2a82fc14f 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -258,13 +258,13 @@ impl ClientTable { return RequestStatus::ChecksumMismatch { request }; } - match entry.find_cached(request) { - Some(cached) => RequestStatus::Duplicate(cached.clone()), - None => RequestStatus::AlreadyApplied { + entry.find_cached(request).map_or( + RequestStatus::AlreadyApplied { request, watermark: entry.watermark, }, - } + |cached| RequestStatus::Duplicate(cached.clone()), + ) } /// Check register. Valid without existing entry; returns diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs index fdc3878bb..1ffbe8691 100644 --- a/core/consensus/src/metadata_helpers.rs +++ b/core/consensus/src/metadata_helpers.rs @@ -733,10 +733,9 @@ mod tests { let sends = consensus.message_bus().client_sends.borrow(); assert_eq!(sends.len(), 1, "ring hit replays the original reply"); - let header = bytemuck::checked::try_from_bytes::<ReplyHeader>( - &sends[0].1.as_slice()[..HEADER_SIZE], - ) - .expect("valid ReplyHeader"); + let header = + bytemuck::checked::try_from_bytes::<ReplyHeader>(&sends[0].1.as_slice()[..HEADER_SIZE]) + .expect("valid ReplyHeader"); assert_eq!(header.request, 3, "original reply for the retried request"); } diff --git a/core/integration/tests/cluster/client_table_restart.rs b/core/integration/tests/cluster/client_table_restart.rs index 3eb6195ce..2e5789335 100644 --- a/core/integration/tests/cluster/client_table_restart.rs +++ b/core/integration/tests/cluster/client_table_restart.rs @@ -21,41 +21,40 @@ //! node crash must be able to continue: a retry of an already-committed //! request id must be answered from the dedup cache (never re-applied, //! never silently dropped), and the next request id must be admitted. -//! Today the table lives only in memory, so a rebooted node has no record -//! of the session or its request watermark and both scenarios fail. //! -//! The Rust SDK cannot drive this: it resets its `ConsensusSession` on every -//! disconnect and re-registers under a fresh identity. The frames are -//! therefore hand-crafted on a raw TCP socket, same technique as the -//! protocol-version gate tests. +//! Server-side this rests on three landed pieces: //! -//! What has to land for these tests to go green, in order: +//! 1. WAL-replay table recovery: `metadata::impls::recovery::recover` +//! replays registers (minting the same epochs) and re-caches committed +//! replies byte-identically, so a rebooted node remembers where each +//! client left off. Sessions whose register fell below the snapshot +//! floor are not recovered yet (no checkpoint artifact - IGGY-137 +//! remainder). +//! 2. Implicit rebind: a replicated request on an unbound transport whose +//! `(client, session)` matches a live table entry rebinds the transport +//! (`try_resume_session`); the same session arriving on two connections +//! evicts the older binding (`SessionManager::bind_session`). +//! 3. Sessions survive transport disconnect: `submit_disconnect_logout` +//! tears down only consumer-group members (the group must rebalance off +//! a dead consumer); everything else keeps its slot for resume until an +//! explicit `Logout` or capacity eviction. //! -//! 1. Persist the clients table (IGGY-137, standalone): include the -//! (client id, last request id, cached reply) entries in the checkpoint -//! and recover them on boot from WAL replay, so a rebooted node -//! remembers where each client left off. -//! 2. The client stops forgetting itself on disconnect: keep client id, -//! session id and request counter across reconnects and present the old -//! identity instead of a fresh Register. -//! 3. The server accepts a resumed identity: look the session up in the -//! replicated table, rebind the new transport to it, and answer with -//! the last committed request id so the client knows whether its -//! in-doubt request went through. Define the conflict rule for the same -//! session arriving on two connections (evict the older). -//! 4. SDK retry rule change: a replicated write may only be retried under -//! the same (client id, request id); the path that re-issues an -//! in-doubt write under a fresh session after failover goes away. +//! The Rust SDK cannot drive this yet: it resets its `ConsensusSession` on +//! every disconnect and re-registers under a fresh identity (the +//! `sdk/vsr.rs` retry TODO). The frames are therefore hand-crafted on a raw +//! TCP socket, same technique as the protocol-version gate tests. SDK-side +//! identity stability (keep client id + request counter across reconnects, +//! retry replicated writes only under the same identity) is the remaining +//! client half. //! -//! Steps 2+3 must ship together; 1 is standalone. An alternative to 2-4 is -//! a per-request idempotency key that is independent of the session, the -//! way TigerBeetle does it: no session resume at all, retries from a fresh -//! client session stay safe because dedup keys off the request, not the -//! (client, session) pair. +//! Single-node topology on purpose: the raw client pins one address, and a +//! follower cannot commit replicated TCP writes (no follower forwarding for +//! TCP yet), so post-failover resume against a 3-node cluster is future +//! work alongside that forwarding. //! //! These tests pin the implicit-rebind contract: a resumed client simply //! keeps sending under its old `(client, session)` on a fresh connection and -//! the server rebinds the transport from the persisted table. If the +//! the server rebinds the transport from the recovered table. If the //! session-resume work settles on an explicit resume handshake instead, //! adjust `resume_request` to speak it. @@ -102,8 +101,7 @@ const REPLY_WAIT: Duration = Duration::from_secs(5); const RETRY_PAUSE: Duration = Duration::from_millis(100); -#[iggy_harness] -#[ignore = "red until clients-table persistence + session resume land"] +#[iggy_harness(cluster_nodes = 1)] async fn given_committed_request_when_node_restarts_should_dedup_same_id_retry( harness: &mut TestHarness, ) { @@ -122,8 +120,7 @@ async fn given_committed_request_when_node_restarts_should_dedup_same_id_retry( resume_request(addr, session, 1, &create_stream).await; } -#[iggy_harness] -#[ignore = "red until clients-table persistence + session resume land"] +#[iggy_harness(cluster_nodes = 1)] async fn given_bound_session_when_node_restarts_should_accept_next_request_id( harness: &mut TestHarness, ) { diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index e6caae450..92ea95c5e 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -547,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) { + *self.client_table.borrow_mut() = client_table; + } + /// Install the resolved byte value used for `MaxTopicSize::ServerDefault`. /// Server-ng bootstrap calls this with `system.topic.max_size` on every /// shard (responses read it too); only shard 0's copy feeds admission. @@ -1960,9 +1968,9 @@ where /// /// # Safety /// Re-preflight per iteration: `commit_journal` may have advanced the - /// client's watermark between push and drain (Duplicate / AlreadyApplied - /// / `AlreadyRegistered`). Skipping produces a duplicate prepare and - /// panics. + /// client's watermark between push and drain (`Duplicate` / + /// `AlreadyApplied` / `AlreadyRegistered`). Skipping produces a duplicate + /// prepare and panics. #[allow(clippy::future_not_send)] async fn drain_request_queue_into_prepares(&self) { let consensus = self.consensus.as_ref().unwrap(); diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs index 88413b14d..51e3cf56f 100644 --- a/core/metadata/src/impls/recovery.rs +++ b/core/metadata/src/impls/recovery.rs @@ -19,7 +19,8 @@ use crate::impls::metadata::IggySnapshot; use crate::stm::StateMachine; use crate::stm::authz::GatedApply; use crate::stm::snapshot::{MetadataSnapshot, RestoreSnapshot, Snapshot, SnapshotError}; -use iggy_binary_protocol::consensus::PrepareHeader; +use consensus::{CLIENTS_TABLE_MAX, ClientTable, build_reply_message, build_reply_message_with}; +use iggy_binary_protocol::consensus::{Operation, PrepareHeader}; use iggy_common::IggyError; use journal::prepare_journal::{JournalError, PrepareJournal}; use server_common::Message; @@ -86,6 +87,13 @@ pub struct RecoveredMetadata<M> { pub journal: PrepareJournal, pub snapshot: Option<IggySnapshot>, pub mux_stm: M, + /// Client table rebuilt from the replayed committed prefix: registers + /// re-mint epochs in apply order, replies re-cache byte-identically + /// (`build_reply_message*` reads only the prepare header + deterministic + /// apply output). Sessions whose register fell below the snapshot floor + /// are NOT recovered - the table has no checkpoint artifact yet + /// (IGGY-137); those clients re-register and their epoch restarts at 1. + pub client_table: ClientTable, /// `None` means no snapshot existed and no journal entries were replayed. /// `Some(op)` is the highest op applied, either from the snapshot or journal replay. /// @@ -105,7 +113,9 @@ pub struct RecoveredMetadata<M> { /// 1. Load snapshot from `{data_dir}/metadata/snapshot.bin` if present /// 2. Restore state machine from snapshot, or initialize empty state /// 3. Open WAL at `{data_dir}/metadata/journal.wal`, scan and rebuild index -/// 4. Replay journal entries from the first post-snapshot op through the state machine +/// 4. Replay journal entries from the first post-snapshot op through the +/// state machine, rebuilding the client table alongside (registers mint +/// epochs, replies re-cache) exactly as the commit paths did live /// 5. Return the assembled `RecoveredMetadata` /// /// Only the owning shard (shard 0) should call this. Peer shards receive @@ -188,6 +198,7 @@ where .fold(snapshot_floor, u64::max) }; + let mut client_table = ClientTable::new(CLIENTS_TABLE_MAX); let mut last_applied_op: Option<u64> = None; let mut last_journaled_op: Option<u64> = None; for header in &headers_to_replay { @@ -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); + last_applied_op = Some(header.op); + continue; + } + if header.operation == Operation::Logout { + client_table.remove_client(header.client); + // TODO: the commit paths also run `remove_consumer_group_member` + // here; recovery has no `StreamsFrontend` bound, so replayed + // logouts leave stale group members (pre-existing, harmless for + // dead connections but a divergence from the live apply). + last_applied_op = Some(header.op); + continue; + } + let entry = journal.entry_at(header).await?.ok_or_else(|| { RecoveryError::Io(std::io::Error::new( std::io::ErrorKind::InvalidData, @@ -209,6 +240,16 @@ where // WAL replay must recompute authorization denials identically to the // primary/backup commit paths, so it goes through the same gate. let reply = mux_stm.gated_update(entry)?; + // Re-cache the reply exactly like the commit paths: same prepare + // header + deterministic apply output = the original bytes. Skipped + // when the session is absent (server-originated ops, or the client + // was evicted / registered below the snapshot floor). + if let Some(epoch) = client_table.get_epoch(header.client) { + let cached = build_reply_message_with(header, reply.reply_body_len(), |dst| { + reply.write_reply_body(dst); + }); + client_table.commit_reply(header.client, epoch, cached); + } tracing::debug!( target: "iggy.metadata.diag", op = header.op, @@ -224,6 +265,7 @@ where journal, snapshot, mux_stm, + client_table, last_applied_op, last_journaled_op, }) @@ -264,6 +306,31 @@ mod tests { Message::try_from(buffer).unwrap() } + /// A client-attributed prepare (Register / app op / Logout) as the + /// admission path stamps it. + fn make_client_prepare( + op: u64, + operation: Operation, + client: u128, + user_id: u32, + request: u64, + ) -> Message<PrepareHeader> { + let total_size = HEADER_SIZE; + let mut buffer = Owned::<4096>::zeroed(total_size); + let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>( + &mut buffer.as_mut_slice()[..HEADER_SIZE], + ); + header.size = total_size as u32; + header.command = Command2::Prepare; + header.op = op; + header.commit = op.saturating_sub(1); + header.operation = operation; + header.client = client; + header.user_id = user_id; + header.request = request; + Message::try_from(buffer).unwrap() + } + #[compio::test] async fn recover_empty_state() { let dir = tempdir().unwrap(); @@ -422,6 +489,113 @@ mod tests { ); } + // The IGGY-137 restart contract: a rebooted node must remember where + // each client left off. Replay re-mints the epoch, restores the + // watermark, and re-caches the reply so a retry of the last committed + // request id dedups instead of re-executing or silently dropping. + #[compio::test] + async fn recover_rebuilds_client_table_from_wal() { + use consensus::client_table::RequestStatus; + + const CLIENT: u128 = 0x1337; + const USER: u32 = 7; + + let dir = tempdir().unwrap(); + let metadata_dir = dir.path().join("metadata"); + std::fs::create_dir_all(&metadata_dir).unwrap(); + + { + let journal = PrepareJournal::open(&metadata_dir.join("journal.wal"), 0) + .await + .unwrap(); + journal + .append(make_client_prepare(1, Operation::Register, CLIENT, USER, 0)) + .await + .unwrap(); + journal + .append(make_client_prepare( + 2, + Operation::CreateStream, + CLIENT, + USER, + 1, + )) + .await + .unwrap(); + journal.storage_ref().fsync().await.unwrap(); + } + + // Solo: every journaled op is committed. + let recovered = recover::<TestStm>( + dir.path(), + true, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + |_| {}, + ) + .await + .unwrap(); + + let table = &recovered.client_table; + assert_eq!(table.get_epoch(CLIENT), Some(1), "register minted epoch 1"); + assert_eq!(table.get_user_id(CLIENT), Some(USER)); + assert_eq!( + table.get_watermark(CLIENT), + Some(1), + "committed request 1 restored the watermark" + ); + match table.check_request(CLIENT, 1, 1, 0) { + RequestStatus::Duplicate(cached) => { + assert_eq!(cached.header().request, 1, "retry replays the cached reply"); + } + other => panic!("expected Duplicate, got {other:?}"), + } + assert!( + matches!(table.check_request(CLIENT, 1, 2, 0), RequestStatus::New), + "the next request id is admitted" + ); + } + + // A replayed Logout removes the entry, mirroring the commit paths. + #[compio::test] + async fn recover_replays_logout_as_session_removal() { + const CLIENT: u128 = 0x1337; + const USER: u32 = 7; + + let dir = tempdir().unwrap(); + let metadata_dir = dir.path().join("metadata"); + std::fs::create_dir_all(&metadata_dir).unwrap(); + + { + let journal = PrepareJournal::open(&metadata_dir.join("journal.wal"), 0) + .await + .unwrap(); + journal + .append(make_client_prepare(1, Operation::Register, CLIENT, USER, 0)) + .await + .unwrap(); + journal + .append(make_client_prepare(2, Operation::Logout, CLIENT, USER, 1)) + .await + .unwrap(); + journal.storage_ref().fsync().await.unwrap(); + } + + let recovered = recover::<TestStm>( + dir.path(), + true, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + |_| {}, + ) + .await + .unwrap(); + assert_eq!( + recovered.client_table.get_epoch(CLIENT), + None, + "logged-out session must not be resurrected" + ); + assert_eq!(recovered.last_applied_op, Some(2)); + } + #[test] fn snapshot_persist_load_roundtrip() { let dir = tempdir().unwrap(); diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs index e7784a81e..c8ea95f9d 100644 --- a/core/metadata/src/stm/user.rs +++ b/core/metadata/src/stm/user.rs @@ -563,7 +563,7 @@ impl StateHandler for ChangePasswordRequest { // `verify_and_rewrite_change_password`): the accept path always // replicates a non-empty Argon2 hash, so this is unambiguous. Rejecting // here (rather than denying pre-consensus) commits the op as a no-op, - // keeping the client's request sequence contiguous in the ClientTable. + // recording the request id in the ClientTable so a retry of it dedups. if self.new_password.is_empty() { return ApplyReply::err(ChangePasswordResult::InvalidCredentials); } diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index b7d578a3f..592fbe8de 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -925,6 +925,7 @@ async fn shard_main( recovered.snapshot, recovered.last_applied_op, recovered.last_journaled_op, + recovered.client_table, )), ) } @@ -943,8 +944,10 @@ async fn shard_main( // Metadata consensus + journal + snapshot live only on shard 0. // `IggyShard::tick_metadata` short-circuits when `consensus.is_none()`, // so peer shards have no caller that reads `journal` or `snapshot`. - let (metadata_consensus, journal_for_metadata, snapshot_for_metadata) = - if let Some((journal, snapshot, last_applied_op, last_journaled_op)) = owner_state { + let (metadata_consensus, journal_for_metadata, snapshot_for_metadata, recovered_client_table) = + if let Some((journal, snapshot, last_applied_op, last_journaled_op, client_table)) = + owner_state + { let snapshot_floor = snapshot.as_ref().map_or(0, IggySnapshot::sequence_number); let commit_watermark = last_applied_op.unwrap_or(snapshot_floor); let restored_op = last_journaled_op.unwrap_or(snapshot_floor); @@ -959,9 +962,9 @@ async fn shard_main( config.metadata.prepare_queue_depth, cluster_heartbeat_ticks(config), ); - (Some(consensus), Some(journal), snapshot) + (Some(consensus), Some(journal), snapshot, Some(client_table)) } else { - (None, None, None) + (None, None, None, None) }; let metadata = ServerNgMetadata::new( metadata_consensus, @@ -970,6 +973,12 @@ async fn shard_main( mux_stm, Some(PathBuf::from(&config.system.path)), ); + // Reinstall the sessions the WAL replay rebuilt, so a rebooted node + // dedups retries and admits continuations from clients that kept their + // identity across the restart (IGGY-137). + if let Some(client_table) = recovered_client_table { + metadata.install_client_table(client_table); + } // Shard 0's copy resolves the `ServerDefault` sentinels (max topic size and // message expiry) at admission; every shard's copy backs the same resolution in responses. metadata.set_default_max_topic_size(config.system.topic.max_size.as_bytes_u64()); diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index 89160a63a..d02a3d1e9 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -107,7 +107,7 @@ use std::cell::RefCell; use std::collections::{HashMap, HashSet, VecDeque}; use std::rc::Rc; use std::sync::Arc; -use tracing::{debug, warn}; +use tracing::{debug, info, warn}; pub(crate) type ClientRequestQueues = Rc<RefCell<HashMap<u128, VecDeque<Message<GenericHeader>>>>>; pub(crate) type ActiveClientRequests = Rc<RefCell<HashSet<u128>>>; @@ -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) { + warn!( + transport_client_id, + error = %error, + "session resume: login transition failed" + ); + return None; + } + if let Err(error) = sessions.bind_session(transport_client_id, header.client, epoch) { + warn!( + transport_client_id, + error = %error, + "session resume: bind failed" + ); + return None; + } + } + info!( + transport_client_id, + client = header.client, + epoch, + user_id, + "rebound transport to resumed session" + ); + Some((header.client, epoch)) +} + fn enqueue_client_request<B, MJ, S>( shard: Rc<ShellShard<B, MJ, S>>, sessions: Rc<RefCell<SessionManager>>, @@ -748,6 +879,13 @@ async fn handle_client_request<B, MJ, S>( } let bound = sessions.borrow().get_session(transport_client_id); + // Unbound transport presenting a full old identity: session resume + // (IGGY-137). The table survived the restart via WAL-replay recovery; + // a matching `(client, epoch)` rebinds this connection in place. + let bound = match bound { + Some(bound) => Some(bound), + None => try_resume_session(shard, sessions, transport_client_id, &header).await, + }; if bound.is_none() { // Replicated request on an unbound transport. Without this short- // circuit, the rewrite below overwrites `header.client` with @@ -2168,11 +2306,10 @@ async fn handle_delete_segments_request<B, MJ, S>( // offset on the owning shard, then replicate a `TruncatePartition(offset)` // AS the client's own request through the standard owner path: the commit // records (client, session, request) in the `ClientTable` on every replica, - // keeping the sequence contiguous. Skipping the commit (or attributing it - // to an internal id) leaves a hole that fails the next metadata op's - // `request == committed + 1` preflight -> RequestGap -> silent drop -> the - // SDK blocks until timeout. A no-op delete still commits `up_to_offset = 0` - // (monotonic apply) for the same reason. + // advancing the watermark. Skipping the commit (or attributing it to an + // internal id) leaves this request id unrecorded, so the SDK's own retry + // of it would re-execute instead of deduping. A no-op delete still + // commits `up_to_offset = 0` (monotonic apply) for the same reason. let truncate = match resolve_delete_segments_truncate( shard, &header, @@ -2370,9 +2507,17 @@ 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). +/// dropped by the caller; for a consumer-group member 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) and the group rebalances off the dead consumer. +/// +/// A connection with NO group membership keeps its session: the client may +/// reconnect and resume under its old `(client, session)` identity +/// (IGGY-137, `try_resume_session`), which is exactly the crash-retry window +/// dedup exists for. Its slot is reclaimed by an explicit client `Logout` or +/// the table's capacity eviction. Same membership rule as the heartbeat +/// verifier, which deliberately reaps only group members. /// /// Deliberately does NOT drop the local `ClientTable` slot first: /// `submit_logout_*` short-circuits when the slot is already gone, so a @@ -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 { + debug!( + vsr_client_id, + "transport disconnected; keeping session for resume" + ); + return; + } let bus = shard.bus.clone(); bus.spawn(async move { if let Err(error) = diff --git a/core/server-ng/src/users.rs b/core/server-ng/src/users.rs index e5e08f2ca..2fb03af40 100644 --- a/core/server-ng/src/users.rs +++ b/core/server-ng/src/users.rs @@ -97,11 +97,11 @@ where /// /// When the target resolves (`stored_hash` is `Some`) the supplied /// `current_password` is verified against it first. A mismatch does NOT deny -/// pre-consensus: that would consume the client's request id without advancing -/// the replicated `ClientTable`, gapping the sequence so the next replicated op -/// is dropped (`RequestGap`) and the caller stalls. Instead the new password is -/// emptied, which signals the committed apply to reject with -/// `InvalidCredentials` (a committed no-op that keeps the sequence contiguous). +/// pre-consensus: that would consume the client's request id without recording +/// it in the replicated `ClientTable`, so a retry of that id would re-execute +/// instead of deduping. Instead the new password is emptied, which signals the +/// committed apply to reject with `InvalidCredentials` (a committed no-op that +/// advances the watermark). /// /// Either way the current password is stripped and the accepted new one hashed, /// so no plaintext credential ever enters consensus. An unresolved target keeps diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 60803f5b5..27b0b2a10 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -199,6 +199,15 @@ pub enum MetadataSubmit { partition_id: u32, reply: Sender<Option<u64>>, }, + /// A home shard asks shard 0 whether `vsr_client_id` has a live entry in + /// the replicated client table, to rebind a reconnecting transport that + /// presents its old identity (session resume, IGGY-137). Read-only. + /// `reply` carries `(epoch, user_id)` for a registered client, `None` + /// otherwise. + ResumeLookup { + vsr_client_id: u128, + reply: Sender<Option<(u64, u32)>>, + }, } /// Handler shard 0 runs for an inbound [`MetadataSubmit`].
