numinnex commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4174400955
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -872,7 +866,27 @@ export class CommandResponseStream extends EventEmitter {
handleResp: boolean,
deadline: number
): Promise<CommandResponse> {
- await this._processVsr(LOGOUT.code, LOGOUT.serialize(), true, deadline);
+ if (this.isAuthenticated) {
+ await this._processVsr(LOGOUT.code, LOGOUT.serialize(), true, deadline);
+ } else {
+ const resume = this.vsrSession.resumePayload(this.metadataWatermark);
+ if (resume) {
+ const remembered = this.rememberedCredentials;
+ const sameCredentials = remembered && ('token' in remembered
+ ? command === LOGIN_WITH_TOKEN.code &&
payload.equals(LOGIN_WITH_TOKEN.serialize(remembered))
+ : command === LOGIN.code &&
payload.equals(LOGIN.serialize(remembered)));
+ if (sameCredentials) {
+ try {
+ return await this._processVsr(COMMAND_CODE.BindSession, resume,
handleResp, deadline);
+ } catch (error) {
+ if (!(error instanceof ResponseError) ||
+ (error.errorCode !== UNAUTHENTICATED && error.errorCode !==
STALE_CLIENT))
Review Comment:
**critical:** the fallback to a fresh Register only covers
`UNAUTHENTICATED`/`STALE_CLIENT`. The server answers BindSession for a retired
or forgotten client with `TransientNotCommitted` (client_table.rs:622/629).
`vsrSession` survives `_resetSession`, so every later login retries
BindSession and fails the same way, and login stays wedged until `destroy()`.
Rust also falls back on 57 here. Needs a test for BindSession -> 57.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -682,8 +693,22 @@ where
/// Record the metadata incarnation this partition was built for. Must run
/// BEFORE [`Self::hydrate_applied_purge_generation`], which keys the
/// durable record on it.
- pub const fn set_created_revision(&mut self, created_revision: u64) {
+ pub fn set_created_revision(&mut self, created_revision: u64) {
self.created_revision = created_revision;
+ self.required_metadata_frontier =
self.required_metadata_frontier.max(created_revision);
+ }
+
+ /// A report covers only an applied barrier on a serving replica. WAL
replay
+ /// or the paired checkpoint must restore this floor before admission.
+ #[must_use]
+ pub fn session_retired(&self, identity: SessionIdentity) -> bool {
+ self.fatal.is_none()
+ && !self.requires_state_transfer()
+ && self.consensus.is_normal()
+ && !self.consensus.is_transferring()
Review Comment:
**critical:** a retired receipt can be resurrected. `RetireSession` runs as
`AUTO_COMMIT_CLIENT_ID`, so `pending_request(X)` never sees it, and the
frontier is raised only at commit.
Request R (X, S) prepared at N+1 behind RetireSession at N: the sweep reads
retired at commit N, then N+1 commits and `commit_partition_reply` recreates
the entry. FinalizeSession runs and the slot leaks permanently. The receipt
checkpoint carries it across restart.
A read-side check can't close this, because backup reporters have an empty
pipeline. Exclusion needs to happen at primary admission: no RetireSession(X)
while X is pending, and no X admitted while its RetireSession is queued.
##########
core/server/src/partition_helpers.rs:
##########
@@ -656,6 +664,22 @@ async fn load_partition(
// Before the hydrate: the durable record is keyed by incarnation, so a
// `purge.gen` left behind by a previous life of this namespace reads 0.
partition.set_created_revision(partition_metadata.created_revision);
+ partition
+ .restore_retry_checkpoint(recovered_persistence.0.checkpoint_op())
+ .await
+ .map_err(|source| ServerError::PartitionPrepareWalIo {
Review Comment:
**warning:** a missing or corrupt receipt checkpoint maps to
`PartitionPrepareWalIo`, which is in neither the tombstone nor the Refused arm,
so `boot/recovery.rs` aborts the whole node.
Map InvalidData/UnexpectedEof/NotFound to `PartitionRecovery::Refused`
(rejoin/state transfer on replicas, tombstone on a single node). Keep the node
abort for real I/O errors.
##########
core/server/src/http/submit.rs:
##########
@@ -343,49 +340,58 @@ pub(in crate::http) async fn logout_session(state:
&HttpInner, session: &Rc<Http
Ok(Ok(_)) => {}
Ok(Err(error)) => warn!(
?error,
- "server HTTP: VSR Logout submit failed; slot lingers until
eviction"
+ "server HTTP: VSR Logout submit failed; session remains until
lease expiry"
),
Err(_canceled) => warn!(
- "server HTTP: VSR Logout task dropped before replying; slot
lingers until eviction"
+ "server HTTP: VSR Logout task dropped before replying; session
remains until lease expiry"
),
}
state.forget_session(session);
}
-/// Run one awaited partition write (produce / consumer-offset write) end to
-/// end: admit against the in-flight caps, install the reply slot, dispatch
-/// into the partition plane, and wait (bounded) for the committed reply.
-///
-/// Slot-before-dispatch is load-bearing: every pre-dispatch gate failure
-/// inside [`dispatch_partition_request`] replies through `send_to_client`,
-/// which fires an installed slot, so one slot catches every exit. The slot
-/// guard borrows the registry, which is why this whole future runs inside
-/// the caller's `SendWrapper` on shard 0.
-///
-/// Hands back the graded reply frame and the header grading read, so a caller
-/// that renders a committed payload can bound the body without parsing the
-/// header again; the offset writes answer 204 and drop both.
pub(in crate::http) async fn partition_write_replicated(
state: &HttpInner,
session: &HttpSession,
operation: Operation,
body: &[u8],
) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> {
- // Admission sits here rather than before body decode: axum's extractors
- // already buffered and deserialized the body (bounded by the router-wide
- // `DefaultBodyLimit`) before the handler ran, so the caps gate what is
- // actually unbounded - the slot install, the dispatch, and the parked
- // reply await that pins this request's buffers for up to the reply
- // timeout. Held across every exit below; released by `Drop`.
+ partition_write(state, session, operation, body, &mut None).await
+}
+
+async fn partition_write(
+ state: &HttpInner,
+ session: &HttpSession,
+ operation: Operation,
+ body: &[u8],
+ dispatched: &mut Option<oneshot::Sender<Result<(), PartitionWriteError>>>,
+) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> {
+ let deadline = Instant::now() + PARTITION_WRITE_REPLY_TIMEOUT;
let _in_flight = admit_partition_write(&session.in_flight_writes,
&state.in_flight_writes)?;
+ let namespace = crate::namespace::resolve_partition_request_namespace(
Review Comment:
**warning:** the namespace is resolved here for the gate and again in
`dispatch_partition_request`. With `Balanced`, `next_balanced_partition`
advances the round-robin on each call, so the gate locks partition A while the
write goes to B.
Concurrent same-session writes then hit the one-unresolved rule and get
spurious `TransientNotAccepted`, which is the default HTTP producer path.
Resolve once and pass the namespace through.
##########
core/common/src/error/iggy_error.rs:
##########
@@ -573,6 +573,10 @@ pub enum IggyError {
iggy_binary_protocol::ProtocolVersion(*.2)
)]
IncompatibleProtocolVersion(u32, u32, u32) = 14003,
+ #[error("VSR session mismatch: requested {0}, server bound {1}")]
+ SessionMismatch(u64, u64) = 14004,
+ #[error("VSR request IDs exhausted; start a new logical session")]
+ RequestIdExhausted = 14005,
Review Comment:
**warning:** Swift is synced now, but 14005 is still missing from Go
`errors.yaml`, Java `IggyErrorCode.java` and Node `error.code.ts`, all of which
mirror 14004.
##########
core/sdk/src/clients/client.rs:
##########
@@ -122,8 +122,8 @@ const SESSION_CONTROL_CODES: [u32; 5] = [
/// [`IggyProducer`] so that _background_ producers flush the latest state.
Finally,
/// call [`shutdown()`] on the [`IggyClient`] which closes the connection.
/// Use [`disconnect()`] rather than [`shutdown()`] to close the connection
but keep the client usable, as a
-/// client that has been shut down cannot reconnect. Note, if `auto-login`
is configured, the client
-/// will reconnect automatically and undo the disconnect.
+/// client that has been shut down cannot reconnect. Explicit disconnect
clears sign-in;
Review Comment:
**warning:** with `AutoLogin` configured, `sign_in_credentials()` falls back
to the config (tcp_client.rs:1073, same in QUIC/WS), so `connect()` or the
reconnect path signs in again silently after an explicit disconnect.
Either restore the AutoLogin caveat or suppress config credentials until the
next explicit connect.
##########
foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs:
##########
@@ -139,10 +140,18 @@ internal async ValueTask<VsrAttempt> SendAttemptAsync(int
code, ReadOnlyMemory<b
return VsrAttempt.Ok(response, this);
}
- catch (IggyInvalidStatusCodeException e) when (retryTransient
&& IsReplayableTransient(e, transientDeadline,
-
readDeadline))
+ catch (IggyInvalidStatusCodeException e)
{
- var governingDeadline = e.StatusCode ==
VsrError.TRANSIENT_NOT_COMMITTED
+ uncertain |= e is { FromServer: true, StatusCode:
VsrError.TRANSIENT_NOT_COMMITTED };
Review Comment:
**warning:** the rule that a later refusal can't resolve an earlier
uncertain attempt is applied unevenly. Rust (`vsr.rs:44-57`) and Go remap
`TransientNotAccepted|Unauthorized|Unauthenticated|StaleClient` after 57. C#
here and Node map only `TransientNotAccepted`, and Java
(`AsyncTcpConnection.shouldRetryTransient`) does no sticky tracking.
A caller that sees a definitive refusal resends, which can duplicate.
Suggest one rule plus a parity test per SDK.
##########
core/consensus/src/client_table.rs:
##########
@@ -1277,62 +1134,136 @@ impl ClientTable {
return SliceRequestStatus::New;
};
let entry = self.slots[slot_idx].as_ref().expect("index/slot
mismatch");
- if entry.user_id != user_id || request > entry.watermark {
+ if entry.user_id != user_id {
return SliceRequestStatus::New;
}
- let below = entry.watermark - request;
- if below >= COMMITTED_WINDOW_BITS {
- return SliceRequestStatus::AgedOut;
+ entry.check_slice_request(request)
+ }
+
+ #[must_use]
+ /// # Panics
+ /// Only if the private index refers to an unoccupied slot.
+ pub fn check_partition_request(
+ &self,
+ client_id: u128,
+ user_id: u32,
+ session: u64,
+ request: u64,
+ operation: Operation,
+ ) -> RequestStatus {
+ let Some(&slot) = self.index.get(&client_id) else {
+ return RequestStatus::New;
+ };
+ let entry = self.slots[slot].as_ref().expect("index/slot mismatch");
+ if entry.user_id != user_id || entry.epoch != session {
+ return RequestStatus::Fenced {
+ current: entry.epoch,
+ received: session,
+ };
}
- if entry.committed_window & (1 << below) == 0 {
- SliceRequestStatus::New
- } else {
- SliceRequestStatus::Committed
+ match entry.check_slice_request(request) {
+ SliceRequestStatus::New => RequestStatus::New,
+ SliceRequestStatus::Committed => match entry.find_cached(request) {
+ Some(reply) if reply.header().operation != operation => {
+ RequestStatus::OperationMismatch { request }
+ }
+ Some(reply) => RequestStatus::Duplicate(reply.clone()),
+ None => RequestStatus::AlreadyApplied {
+ request,
+ watermark: entry.watermark,
+ },
+ },
+ SliceRequestStatus::AgedOut => RequestStatus::AlreadyApplied {
+ request,
+ watermark: entry.watermark,
+ },
}
}
- /// Record a committed request without a reply to cache.
+ /// Retain a partition receipt and return a shared handle to its first
committed result.
///
- /// The entry point for a plane that runs
- /// [`ClientTableMode::PartitionSlice`]: there is no register to create the
- /// entry, so the first committed request creates it, and there is no
result
- /// section worth retaining. [`Self::check_slice_request`] proves retained
- /// duplicates and the caller synthesizes their reply.
- ///
- /// Idempotent and order-insensitive for one user: the watermark only rises
- /// and the window only gains bits, so replaying an already-folded op is a
- /// no-op and a state-transfer install followed by a re-walk of the same
- /// commits converges. A commit above the watermark shifts the window up by
- /// the gap (ids that age out read as [`SliceRequestStatus::AgedOut`]);
- /// one below it sets its bit. A commit by a DIFFERENT user under the same
- /// client id replaces the entry outright: nothing observes a logout here,
so this is
- /// what stops the next holder of a reused id from having its first
- /// writes absorbed by the previous holder's watermark.
+ /// # Errors
+ /// Refuses invalid or inconsistent receipts and exhausted protection
capacity.
///
/// # Panics
- /// If called on a table whose mode caches replies -- that plane must go
- /// through [`Self::commit_reply`] so the ring stays populated.
- pub fn commit_request(&mut self, client_id: u128, user_id: u32, request:
u64, commit_op: u64) {
- debug_assert!(
- !self.mode.cache_replies(),
- "commit_request: a reply-caching table must use commit_reply"
- );
- // Zero is the reserved client id, refused at every ingress (wire
- // validation, the HTTP minter, the auto-commit guard at the call
- // sites). Kept as a return rather than an assert so that an artifact
- // slipping past the decoder degrades to no dedup for that entry
instead
+ /// Only if the private index refers to an unoccupied slot.
+ pub fn commit_partition_reply(
+ &mut self,
+ user_id: u32,
+ session: u64,
+ reply: Message<ReplyHeader>,
+ ) -> Result<CachedReply, ClientTableWireError> {
+ let header = *reply.header();
+ if header.client == 0
+ || session == 0
+ || header.commit == 0
+ || !matches!(
+ header.operation,
+ iggy_binary_protocol::Operation::SendMessages
+ | iggy_binary_protocol::Operation::StoreConsumerOffset
+ | iggy_binary_protocol::Operation::DeleteConsumerOffset
+ )
+ {
+ return Err(ClientTableWireError::InvalidReply);
+ }
+ if let Some(&slot) = self.index.get(&header.client)
+ && self.slots[slot]
+ .as_ref()
+ .is_some_and(|entry| entry.epoch != session)
+ {
+ return Err(ClientTableWireError::InvalidWatermark {
+ client_id: header.client,
+ });
+ }
+ self.commit_request(header.client, user_id, header.request,
header.commit)?;
+ let slot = self.index[&header.client];
+ let entry = self.slots[slot].as_mut().expect("index/slot mismatch");
+ entry.epoch = session;
+ if let Some(cached) = entry.find_cached(header.request) {
+ return Ok(cached.clone());
+ }
+ let cached = CachedReply::from(reply);
Review Comment:
**critical:** partition slices now keep a reply ring per (client,
partition). `trim_ring` only drops when `len > 5 && bytes > 8 KiB`, counting
logical `byte_len`, but each `CachedReply` is a 4 KiB-aligned `Frozen`.
That keeps ~29 SendMessages receipts at ~5.4 KiB RSS each, so ~155 KiB per
(client, partition) and ~620 MiB per partition at `dedup_clients_max = 4096`,
on every replica.
With one unresolved request per session/group, the latest receipt is enough,
and the checkpoint already keeps only the latest. Keep one compact receipt per
entry.
##########
core/server/src/consumer_group/liveness.rs:
##########
@@ -233,6 +248,236 @@ where
.unwrap_or(Pass::Drained)
}
+#[allow(clippy::too_many_lines)]
+async fn retire_sessions<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ liveness: &RefCell<ConsumerGroupLiveness>,
+ stop: &Receiver<()>,
+ shutdown: &ShutdownToken,
+ report_interval: Duration,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let metadata = shard.plane.metadata();
+ let Some(consensus) = metadata.consensus.as_ref() else {
+ return;
+ };
+ if !consensus.is_normal()
+ || consensus.is_transferring()
+ || consensus.commit_min() != consensus.commit_max()
+ {
+ return;
+ }
+ let view = consensus.view();
+ let identity = metadata
+ .client_table
+ .borrow()
+ .ended_sessions()
+ .min_by_key(|identity| (identity.metadata_watermark,
identity.client_id));
Review Comment:
**critical:** retirement is O(ended sessions x partitions) of fsynced
`RetireSession` ops, one identity at a time, including partitions the session
never touched.
`session_retired` needs `commit_min == commit_max`, so busy partitions keep
failing, the cursor resets and the sweep resends every ~200ms. A fatal or
transferring partition on the primary node, or a TruncatePartition revision
bump, stalls every finalize.
Tables never evict, so churn fills them and logins are refused cluster-wide.
Suggest one frontier op per partition covering all sessions ended at <= W,
skipping tombstoned or fatal namespaces, and counting the primary like any
other reporter.
##########
core/server/src/partition_helpers.rs:
##########
@@ -612,7 +603,37 @@ async fn load_partition(
dir: directory.clone(),
source,
},
- })?;
+ };
+ if replica_count == 1
Review Comment:
**warning:** a single-node boot refuses any partition without a published
frontier, even when the dir is absent (crash after CreateTopic commit, before
materialization or mid first `install_frontier`).
That becomes InvalidData -> Refused -> permanent tombstone of a provably
empty partition. When there are no segment bytes and no frontier, init a fresh
WAL instead.
##########
core/binary_protocol/src/version.rs:
##########
@@ -114,17 +110,15 @@ pub const IGGY_PROTOCOL_VERSION: u32 =
parse_packed_semver(env!("CARGO_PKG_VERSI
// minor/patch = compatible, so MIN derives from the current major instead
// of the current minor. Under 0.x a minor bump may break the wire, so the
// minor-scoped window is correct.
-pub const IGGY_PROTOCOL_VERSION_MIN: u32 = IGGY_PROTOCOL_VERSION & !PATCH_MASK;
-// A 0.0.x crate version would pack MIN to 0, which `EvictionHeader::validate`
-// rejects (window requires min >= 1) -- the server would emit eviction frames
-// that fail its own validation.
-const _: () = assert!(IGGY_PROTOCOL_VERSION_MIN > 0);
+pub const IGGY_PROTOCOL_VERSION_MIN: u32 = pack_protocol_version(0, 11, 1);
Review Comment:
**warning:** the login layout change (32-byte bind secret) ships as 0.11.1,
but master already advertises `0.11.1-edge.1` with the old layout. MIN=0.11.1
admits master edge clients and misparses their body. Old 0.11.0 servers ignore
patch on the upper bound and misparse new clients.
One number now covers two layouts, so both sides get confusing errors
instead of `IncompatibleProtocol`. Suggest 0.12.0 for both current and MIN.
##########
core/consensus/src/client_table.rs:
##########
@@ -765,20 +585,196 @@ impl ClientTable {
/// # Panics
/// If the table already holds a client.
pub fn set_capacity(&mut self, max_clients: usize) {
+ if self.capacity_committed {
+ return;
+ }
assert!(
self.index.is_empty(),
"set_capacity must run before any client registers"
);
*self = Self::with_mode(max_clients, self.mode);
}
- /// Snapshot the table for the metadata checkpoint: every occupied slot
with its
- /// index, so positions (and with them deterministic eviction order)
survive, plus
- /// each entry's `client_id` so the index rebuilds on decode.
+ #[must_use]
+ pub const fn capacity_committed(&self) -> bool {
+ self.capacity_committed
+ }
+
+ #[must_use]
+ pub fn contains(&self, client_id: u128) -> bool {
+ self.index.contains_key(&client_id)
+ }
+
+ /// Verify a binding credential, resolving a zero epoch to its original
registration.
+ ///
+ /// # Errors
+ /// Returns `TransientNotCommitted` while registration is not visible,
+ /// `Unauthenticated` for a bad proof or ended session, and
`InvalidSession`
+ /// for a different nonzero epoch.
+ pub fn bind_session(
+ &mut self,
+ client_id: u128,
+ session: u64,
+ secret: &[u8; BIND_SECRET_BYTES],
+ ) -> Result<(u32, u64, SessionAttachment), IggyError> {
+ let user_id = self
+ .get_user_id(client_id)
+ .ok_or(IggyError::TransientNotCommitted)?;
Review Comment:
**warning:** BindSession uses 57 (`TransientNotCommitted`, which means
"outcome unknown, replay") for "session gone". This is what wedges the Node
SDK, and it lets an unauthenticated caller tell known client ids from unknown
ones (unknown -> 57, known -> `Unauthenticated`).
Suggest: if `session != 0` and the entry is absent after the frontier wait,
return `Unauthenticated`, the same as a bad secret. If `session == 0`, return
the retryable 58.
##########
core/server/src/partition_helpers.rs:
##########
@@ -612,7 +603,37 @@ async fn load_partition(
dir: directory.clone(),
source,
},
- })?;
+ };
+ if replica_count == 1
+ &&
!journal::PartitionPrepareJournal::has_published_frontier(&directory)
+ .await
+ .map_err(&wal_error)?
+ {
+ return Err(wal_error(std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ "existing singleton partition is missing its prepare WAL
frontier",
+ )));
+ }
+ if replica_count > 1 && !allow_bootstrap {
Review Comment:
**warning:** whole-cluster power loss after CreateTopic commits but before
any replica materializes leaves every replica fenced by
`fence_missing_history`. All are `requires_state_transfer` with no source to
transfer from, so the group is dark forever.
"Never materialized" is not "history lost". Fence only when segment/WAL
bytes or prior frontier evidence exist.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -256,6 +255,7 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>(
// gate above failed closed on `None`, so `0` (unattributed) is only a
// type-level fallback here.
new_header.user_id = acting_user_id.unwrap_or(0);
+ new_header.metadata_watermark =
shard.plane.metadata().applied_frontier().get();
Review Comment:
**critical:** `metadata_watermark` is stamped after the
`wait_for_partition_routable` await, but the binding/attachment was validated
before it.
If Logout applies during the wait, the stamp is >= ended_op, the frontier
gate passes, and an ended session's write commits after retirement. This is a
second path into the resurrection leak.
Capture the frontier at the binding check, before any await.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -788,6 +793,60 @@ where
.and_then(std::rc::Weak::upgrade)
}
+#[allow(clippy::future_not_send)]
+async fn complete_session_binding<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ identity: SessionIdentity,
+ secret: BindSecret,
+) -> Result<(u32, u64), IggyError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let watermark = identity.metadata_watermark.max(identity.session);
Review Comment:
**warning:** BindSession is a credential-free resume, and nothing here
re-checks that the user still exists, is `Active`, or has an unexpired PAT. No
delete/deactivate/password-change path ends sessions either, so a secret holder
can keep binding after revocation.
The comment at L103-139 still says resume happens only through the login
path and that a credential-free rebind was removed, which contradicts this code.
##########
core/message_bus/src/replica/handshake.rs:
##########
@@ -471,3 +523,45 @@ fn build_finish_message(
},
)
}
+
+fn with_identity(
+ message: Message<GenericHeader>,
+ identity: &[u8; auth::IDENTITY_LEN],
+) -> Message<GenericHeader> {
+ message.transmute_header(|old, new: &mut GenericHeader| {
+ *new = old;
+ new.reserved_command[auth::IDENTITY_OFFSET..auth::IDENTITY_OFFSET +
auth::IDENTITY_LEN]
+ .copy_from_slice(identity);
+ })
+}
+
+/// Exact release, protocol, and executable identity for coordinated
deployment.
+/// # Errors
+/// Returns an error if the running executable cannot be opened or read.
+pub fn binary_identity(release: &str) -> std::io::Result<[u8;
auth::IDENTITY_LEN]> {
+ static ARTIFACT: OnceLock<[u8; auth::IDENTITY_LEN]> = OnceLock::new();
+ const READ_BYTES: usize = 32 * 1024;
+ let artifact = if let Some(artifact) = ARTIFACT.get() {
+ *artifact
+ } else {
+ let mut file = std::fs::File::open(std::env::current_exe()?)?;
Review Comment:
**warning:** peer identity hashes the whole executable, so a same-release
rebuild (strip, distro package, container vs native) can't join and a rolling
replace is impossible, with no escape hatch.
It also runs a blocking `std::fs` read in async `shard_main` on every shard,
and the OnceLock get-then-set race means N concurrent hashes. Suggest protocol
+ storage format + release, with the exe hash opt-in and computed once in main.
##########
core/consensus/src/client_table.rs:
##########
@@ -1352,65 +1283,102 @@ impl ClientTable {
entry.latest_commit = entry.latest_commit.max(commit_op);
}
}
- return;
+ return Ok(());
}
- let freed = if self.index.len() >= self.clients_max {
- self.evict_oldest()
- } else {
- None
- };
- let Some(slot_idx) = freed.or_else(|| self.first_free_slot()) else {
- // Only reachable at a zero cap, which config validation rejects.
- return;
+ if self.index.len() >= self.clients_max {
Review Comment:
**warning:** eviction is gone and the partition cap is 4096 (half the
metadata cap). Slots are freed only by retirement, which is slow (see
liveness.rs). A hot partition with >4096 live or unretired sessions refuses new
writers with `TransientNotAccepted` until retirement catches up.
Raising the cap can't help existing groups (frozen capacity) and doubles
ring memory. `impls.rs:189` still says the oldest client is evicted.
##########
core/server/src/http/submit.rs:
##########
@@ -343,49 +340,58 @@ pub(in crate::http) async fn logout_session(state:
&HttpInner, session: &Rc<Http
Ok(Ok(_)) => {}
Ok(Err(error)) => warn!(
?error,
- "server HTTP: VSR Logout submit failed; slot lingers until
eviction"
+ "server HTTP: VSR Logout submit failed; session remains until
lease expiry"
),
Err(_canceled) => warn!(
- "server HTTP: VSR Logout task dropped before replying; slot
lingers until eviction"
+ "server HTTP: VSR Logout task dropped before replying; session
remains until lease expiry"
),
}
state.forget_session(session);
}
-/// Run one awaited partition write (produce / consumer-offset write) end to
-/// end: admit against the in-flight caps, install the reply slot, dispatch
-/// into the partition plane, and wait (bounded) for the committed reply.
-///
-/// Slot-before-dispatch is load-bearing: every pre-dispatch gate failure
-/// inside [`dispatch_partition_request`] replies through `send_to_client`,
-/// which fires an installed slot, so one slot catches every exit. The slot
-/// guard borrows the registry, which is why this whole future runs inside
-/// the caller's `SendWrapper` on shard 0.
-///
-/// Hands back the graded reply frame and the header grading read, so a caller
-/// that renders a committed payload can bound the body without parsing the
-/// header again; the offset writes answer 204 and drop both.
pub(in crate::http) async fn partition_write_replicated(
state: &HttpInner,
session: &HttpSession,
operation: Operation,
body: &[u8],
) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> {
- // Admission sits here rather than before body decode: axum's extractors
- // already buffered and deserialized the body (bounded by the router-wide
- // `DefaultBodyLimit`) before the handler ran, so the caps gate what is
- // actually unbounded - the slot install, the dispatch, and the parked
- // reply await that pins this request's buffers for up to the reply
- // timeout. Held across every exit below; released by `Drop`.
+ partition_write(state, session, operation, body, &mut None).await
+}
+
+async fn partition_write(
+ state: &HttpInner,
+ session: &HttpSession,
+ operation: Operation,
+ body: &[u8],
+ dispatched: &mut Option<oneshot::Sender<Result<(), PartitionWriteError>>>,
+) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> {
+ let deadline = Instant::now() + PARTITION_WRITE_REPLY_TIMEOUT;
let _in_flight = admit_partition_write(&session.in_flight_writes,
&state.in_flight_writes)?;
+ let namespace = crate::namespace::resolve_partition_request_namespace(
+ &state.shard,
+ operation,
+ body,
+ session.client_id,
+ )
+ .map_err(PartitionWriteError::Rejected)?;
+ let gate = session.partition_gate(namespace);
Review Comment:
**warning:** the gate is now held through the commit wait, including
`?ack=none`, so same-session writes to one partition run at 1/(RTT+fsync) and
ack=none waits for the previous commit. That behaviour change isn't documented.
`admit_partition_write` runs before the gate, so writers queued on one
partition use up the session budget for other partitions. Take admission after
the gate and document the ack=none change. Moving the gate out would hit the
one-unresolved rule.
##########
core/consensus/src/client_table.rs:
##########
@@ -765,20 +585,196 @@ impl ClientTable {
/// # Panics
/// If the table already holds a client.
pub fn set_capacity(&mut self, max_clients: usize) {
+ if self.capacity_committed {
+ return;
+ }
assert!(
self.index.is_empty(),
"set_capacity must run before any client registers"
);
*self = Self::with_mode(max_clients, self.mode);
}
- /// Snapshot the table for the metadata checkpoint: every occupied slot
with its
- /// index, so positions (and with them deterministic eviction order)
survive, plus
- /// each entry's `client_id` so the index rebuilds on decode.
+ #[must_use]
+ pub const fn capacity_committed(&self) -> bool {
+ self.capacity_committed
+ }
+
+ #[must_use]
+ pub fn contains(&self, client_id: u128) -> bool {
+ self.index.contains_key(&client_id)
+ }
+
+ /// Verify a binding credential, resolving a zero epoch to its original
registration.
+ ///
+ /// # Errors
+ /// Returns `TransientNotCommitted` while registration is not visible,
+ /// `Unauthenticated` for a bad proof or ended session, and
`InvalidSession`
+ /// for a different nonzero epoch.
+ pub fn bind_session(
+ &mut self,
+ client_id: u128,
+ session: u64,
+ secret: &[u8; BIND_SECRET_BYTES],
+ ) -> Result<(u32, u64, SessionAttachment), IggyError> {
+ let user_id = self
+ .get_user_id(client_id)
+ .ok_or(IggyError::TransientNotCommitted)?;
+ let epoch = self
+ .registered_session(
+ client_id,
+ user_id,
+ bind_verifier(client_id, user_id, secret),
+ )?
+ .ok_or(IggyError::TransientNotCommitted)?;
+ if session != 0 && session != epoch {
+ return Err(IggyError::InvalidSession(session));
+ }
+ let attachment = self
+ .attach_session(client_id, epoch, user_id)
+ .ok_or(IggyError::Unauthenticated)?;
+ Ok((user_id, epoch, attachment))
+ }
+
+ /// Resolve an idempotent registration for the same principal and verifier.
+ ///
+ /// # Errors
+ /// Returns `Unauthenticated` when a registered identity has different
credentials.
+ ///
+ /// # Panics
+ /// Only if the private index refers to an unoccupied slot.
+ pub fn registered_session(
+ &self,
+ client_id: u128,
+ user_id: u32,
+ verifier: [u8; BIND_SECRET_BYTES],
+ ) -> Result<Option<u64>, IggyError> {
+ let Some(&slot) = self.index.get(&client_id) else {
+ return Ok(None);
+ };
+ let entry = self.slots[slot].as_ref().expect("index/slot mismatch");
+ if entry.user_id != user_id
+ || entry.ended_op.is_some()
+ || blake3::Hash::from_bytes(entry.bind_verifier) !=
blake3::Hash::from_bytes(verifier)
+ {
+ return Err(IggyError::Unauthenticated);
+ }
+ Ok(Some(entry.epoch))
+ }
+
+ pub fn end_session(
+ &mut self,
+ client_id: u128,
+ user_id: u32,
+ session: u64,
+ ended_op: u64,
+ ) -> bool {
+ if let Some(&slot) = self.index.get(&client_id)
+ && let Some(entry) = &mut self.slots[slot]
+ && entry.user_id == user_id
+ && entry.epoch == session
+ && entry.ended_op.is_none()
+ {
+ entry.ended_op = Some(ended_op);
+ entry.attachment = None;
+ return true;
+ }
+ false
+ }
+
+ /// Retain the exact Logout result before ending its session.
+ ///
+ /// # Errors
+ /// Returns an error if a new Logout receipt cannot be retained.
+ ///
+ /// # Panics
+ /// Only if the private index refers to an unoccupied slot.
+ pub fn commit_logout(
+ &mut self,
+ client_id: u128,
+ user_id: u32,
+ session: u64,
+ reply: Message<ReplyHeader>,
+ ) -> Result<bool, ClientTableWireError> {
+ let Some(&slot) = self.index.get(&client_id) else {
+ return Ok(false);
+ };
+ let entry = self.slots[slot].as_ref().expect("index/slot mismatch");
+ if entry.epoch != session
+ || entry.user_id != user_id
+ || entry.ended_op.is_some()
+ || (reply.header().request <= entry.watermark
+ && reply.header().request != EXPIRED_SESSION_REQUEST_ID)
+ {
+ return Ok(false);
+ }
+ let ended_op = reply.header().commit;
+ if self.commit_reply(client_id, user_id, reply) != CommitReply::Cached
{
+ return Err(ClientTableWireError::InvalidWatermark { client_id });
+ }
+ Ok(self.end_session(client_id, user_id, session, ended_op))
+ }
+
+ pub fn ended_sessions(
+ &self,
+ ) -> impl Iterator<Item =
iggy_binary_protocol::requests::system::SessionIdentity> + '_ {
+ self.slots.iter().flatten().filter_map(|entry| {
+ entry.ended_op.map(
+ |ended_op|
iggy_binary_protocol::requests::system::SessionIdentity {
+ client_id: entry.client_id,
+ session: entry.epoch,
+ metadata_watermark: ended_op,
+ },
+ )
+ })
+ }
+
+ pub fn forget_session(&mut self, client_id: u128, session: u64) -> bool {
+ let Some(&slot) = self.index.get(&client_id) else {
+ return false;
+ };
+ if self.slots[slot]
+ .as_ref()
+ .is_none_or(|entry| entry.epoch != session)
+ {
+ return false;
+ }
+ self.index.remove(&client_id);
+ self.slots[slot] = None;
+ true
+ }
+
+ pub fn finalize_session(
+ &mut self,
+ identity: iggy_binary_protocol::requests::system::SessionIdentity,
+ ) -> bool {
+ if !self.ended_sessions().any(|ended| ended == identity) {
+ return false;
+ }
+ self.forget_session(identity.client_id, identity.session)
+ }
+
+ /// Apply the limit carried by the ordered log, independent of local
configuration.
///
- /// Walks only occupied slots. This runs on shard 0's checkpoint task,
inside the
- /// section where that core is already blocked on the snapshot's fsyncs,
so it must
- /// not also allocate and serde-walk one element per configured slot.
+ /// # Errors
+ /// Refuses an invalid limit, a changed committed limit, or one below live
protection.
+ pub fn commit_capacity(&mut self, capacity: usize) -> Result<(),
ClientTableWireError> {
+ if capacity == 0
+ || capacity > CLIENTS_TABLE_SLOT_MAX
+ || self.count() > capacity
+ || (self.capacity_committed && self.clients_max != capacity)
Review Comment:
**warning:** capacity is frozen at the first commit, and restore/transfer
carry it forward, so changing `clients_table_max`/`dedup_clients_max` is
silently ignored for existing groups.
Either add a replicated capacity-change op or document it as create-time
only and warn at boot when config != committed.
##########
core/sdk/src/quic/quic_client.rs:
##########
@@ -1030,33 +1076,32 @@ impl QuicClient {
let consensus_session = self.consensus_session.clone();
let metadata_watermark =
Arc::clone(&self.poll_router.metadata_watermark);
// SAFETY: we run code holding the `connection` lock in a task so we
can't be cancelled while holding the lock.
- tokio::spawn(async move {
- let connection = connection.lock().await;
- let Some(connection) = connection.as_ref() else {
- error!("Cannot send data. Client is not connected.");
- return Err(IggyError::NotConnected);
- };
+ let preencoded = *header;
+ let (used_header, result) = tokio::spawn(async move {
+ let mut used_header = preencoded;
+ let result = async {
+ let connection = connection.lock().await;
+ let Some(connection) = connection.as_ref() else {
+ error!("Cannot send data. Client is not connected.");
+ return Err(IggyError::NotConnected);
+ };
- let (request_header, request_size) = {
- let mut consensus_session = consensus_session
+ let request_header = match preencoded {
+ Some(header) => header,
Review Comment:
**warning:** the retained header is sent without re-checking it against the
live session after the lock. TCP does that check (tcp_client.rs:1582-1594,
`a_retained_write_waiting_for_the_stream_must_recheck_its_session`). Same gap
in websocket_client.rs:1242.
A concurrent reset+rebind sends an old-identity header on a connection bound
to the new session. Suggest a shared helper in vsr.rs and porting the test.
##########
core/sdk/src/clients/producer.rs:
##########
@@ -61,17 +61,10 @@ pub(crate) fn no_confirmations() -> SendMessagesResponse {
}
}
-/// Why `error` ends the retries of a send, or `None` when the producer's
-/// at-least-once retry policy applies.
-const fn send_retry_stop_reason(error: &IggyError) -> Option<&'static str> {
- match error {
- // Raised while decoding an HTTP reply body, after a successful status.
- IggyError::InvalidBytesResponse | IggyError::InvalidJsonResponse => {
- Some("batch committed, confirmation unreadable")
- }
- IggyError::RequestTooOld => Some("outcome unknown, a later write
already committed"),
- _ => None,
- }
+/// A producer retry creates another request identity. Only an explicit
+/// non-admission permits that; ambiguous retries belong to the transport.
+const fn stops_send_retries(error: &IggyError) -> bool {
+ !matches!(error, IggyError::TransientNotAccepted)
Review Comment:
**warning:** this stops on everything except `TransientNotAccepted`,
including `NotConnected`/`CannotEstablishConnection`, which
`replay_after_session_reset_is_safe` treats as provably unsent. `send_retries`
is effectively 0 for those, and always 0 on HTTP, which the Python README
doesn't mention.
Retrying that set is safe in Rust because `RetryOutcome::observe` remaps
after an uncertain attempt. Foreign SDKs without the sticky rule shouldn't copy
it.
##########
core/common/src/traits/binary_transport.rs:
##########
@@ -75,6 +79,34 @@ mod vsr_session_sealed {
/// mutation corrupts the dedup counter or silently breaks at-most-once.
#[async_trait]
pub trait VsrSessionControl: vsr_session_sealed::Sealed + BinaryTransport {
+ async fn session_bind_secret(&self) -> Result<BindSecret, IggyError>;
Review Comment:
**warning:** these new required methods break external implementors. The
trait docs invite them, and `Sealed` is publicly re-exported. That break isn't
in the PR's listed API breaks.
`session_bind_secret` is also public on the TCP/QUIC/WS clients and hands
out the full bind proof. List the break and `#[doc(hidden)]` or restrict the
secret accessor.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -3997,6 +4120,211 @@ fn segment_manifest_digest(manifest:
&[consensus::StateArtifact]) -> u64 {
hasher.finish()
}
+fn checked_retry_table(
+ wire: &ConsumerOffsetsWire,
+ through_op: u64,
+) -> Result<ClientTable, consensus::client_table::ClientTableWireError> {
+ let mut table = ClientTable::with_mode(wire.dedup_capacity,
ClientTableMode::PartitionSlice);
+ table.install_watermarks(wire.dedup_capacity, wire.dedup.iter().cloned())?;
+ for entry in &wire.dedup {
+ if entry.latest_commit > through_op {
+ return
Err(consensus::client_table::ClientTableWireError::InvalidReply);
+ }
+ }
+ Ok(table)
+}
+
+async fn write_retry_checkpoint(
+ path: &Path,
+ group: u64,
+ incarnation: u64,
+ through_op: u64,
+ wire: &ConsumerOffsetsWire,
+) -> std::io::Result<()> {
+ let mut bytes = Vec::new();
+ bytes.extend_from_slice(&RETRY_CHECKPOINT_MAGIC);
+ bytes.extend_from_slice(&group.to_le_bytes());
+ bytes.extend_from_slice(&incarnation.to_le_bytes());
+ bytes.extend_from_slice(&through_op.to_le_bytes());
+ bytes.extend_from_slice(&wire.encode());
+ bytes.extend_from_slice(&state_artifact_checksum(&bytes).to_le_bytes());
+ let temporary =
path.with_extension(RETRY_CHECKPOINT_TEMP_SUFFIX.trim_start_matches('.'));
+ write_staging_file(&temporary, bytes).await?;
+ compio::fs::rename(&temporary, path).await?;
+ let parent = path.parent().ok_or_else(|| {
+ std::io::Error::new(
+ std::io::ErrorKind::InvalidInput,
+ "receipt checkpoint has no parent",
+ )
+ })?;
+ compio::fs::File::open(parent).await?.sync_all().await
+}
+
+impl<B, SB> IggyPartition<B, SB>
+where
+ B: MessageBus,
+ SB: SuperblockStore,
+{
+ fn retry_checkpoint_path(&self, through_op: u64) ->
std::io::Result<PathBuf> {
+ let directory = self.partition_dir.as_ref().ok_or_else(|| {
+ std::io::Error::new(
+ std::io::ErrorKind::NotFound,
+ "receipt checkpoint requires a partition directory",
+ )
+ })?;
+ Ok(Path::new(directory)
+ .join(format!("prepares-{}", self.created_revision))
+ .join(format!(
+
"{RETRY_CHECKPOINT_PREFIX}{through_op}{RETRY_CHECKPOINT_SUFFIX}"
+ )))
+ }
+
+ pub(crate) async fn persist_retry_checkpoint(
+ &self,
+ through_op: u64,
+ ) -> std::io::Result<PathBuf> {
+ if through_op != self.consensus().commit_min() {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ "receipt frontier differs from the applied frontier",
+ ));
+ }
+ let wire = self
+ .offsets_wire_snapshot(false)
Review Comment:
**warning:** the receipt checkpoint reuses `offsets_wire_snapshot(false)`,
so every WAL checkpoint encodes and fsyncs all consumer offsets inline on the
shard tick, while restore only reads dedup plus the frontier.
Transfer-only validation refusals also become partition-fatal here. Suggest
a dedicated receipt wire, keeping the order: receipt durable -> checkpoint
files -> reclaim after the published checkpoint_op.
--
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]
