You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "hubcio (via GitHub)" <gi...@apache.org> on 2026/10/02 16:45:28 UTC
[PR] feat(consensus): persist retry receipts and shared sessions [iggy]
hubcio opened a new pull request, #4387:
URL: https://github.com/apache/iggy/pull/4387
Lost replies and capacity eviction could erase retry evidence and
permit duplicate mutations after promotion or restart.
Persist original results by session, group and request identity.
Replay the first receipt even when a retry changes its payload,
while preserving principal, session and client-operation guards.
Protect committed capacity and publish receipt checkpoints before
reclaiming their WAL. Refuse recovery when protection is absent or
corrupt.
Bind connections using client-owned proof, retain logical sessions
across reconnects, renew leases only on authenticated activity, and
retire protection across every allocated group before finalization.
Preserve uncertainty when a later attempt is refused.
Replay the exact encoded request after same-session resume. Refused
expired sessions can log in again without replaying an ambiguous
write under a new identity. Keep HTTP attachments invalidatable and
drain bounded retirement sweeps between heartbeat reports.
Retain attachments through materialization and keep awaited HTTP
mutations behind the existing session gate.
Restore recovered receipts before admission while allowing live
pipeline-backed commits. Allow intact empty WALs to elect, fence
missing history before replacement WAL creation, and compare
bootstrap revisions in the same domain.
Introduce protocol 0.11, mandatory storage admission and exact peer
build checks. Require explicit fresh-cluster initialization and
Persisted durability for explicit writes without changing defaults.
Align Go, Java, C# and Node login proofs and shared binding frames.
Cover changed-payload retries, reply loss, crash recovery and
reclamation, corruption, capacity pressure, lifecycle fences and
transport retries.
Document first-receipt integrity stamps and stable binding epochs.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4387:
URL: https://github.com/apache/iggy/pull/4387#issuecomment-5971136276
/skill team-review-slim
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4174005822
##########
core/sdk/src/session.rs:
##########
@@ -98,66 +86,45 @@ impl ConsensusSession {
self.session.is_some()
}
- /// Bind the session after register commits through consensus.
- /// The `session` value is the commit op number from the server's reply.
- ///
- /// # Panics
- /// Panics if already bound (drop and create a new session instead).
- pub fn bind(&mut self, session: u64) {
- assert!(
- self.session.is_none(),
- "session already bound (session={})",
- self.session.unwrap()
- );
- assert!(session > 0, "session must be > 0");
+ /// Registration proof retained across reconnects and session bindings.
+ pub fn bind_secret(&self) -> BindSecret {
+ self.bind_secret.clone()
+ }
+
+ /// Accept the original registration, including its replay after reconnect.
+ pub fn bind(&mut self, session: u64) -> Result<(), IggyError> {
+ if session == 0 || self.session.is_some_and(|bound| bound != session) {
Review Comment:
warning: `bind()` reports a conflicting nonzero epoch as `InvalidSession`, whose message says the value must be non-zero, so the caller reads the wrong cause. Return `SessionMismatch(requested, bound)` for a conflict, and give the counter-overflow arm its own error.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java:
##########
@@ -38,12 +38,13 @@
final class VsrLoginCodec {
/**
- * Packed semver of the {@code iggy_binary_protocol} crate this codec
- * targets: {@code major << 20 | minor << 10 | patch}, 10 bits per field.
- * Keep in sync with {@code core/binary_protocol/Cargo.toml}; the server
- * accepts any client whose major.minor is not newer than its own.
+ * Packed wire protocol version this codec targets:
+ * {@code major << 20 | minor << 10 | patch}, 10 bits per field.
+ * Keep in sync with {@code core/binary_protocol/src/version.rs}, independently
+ * of the crate release. The server enforces its minimum wire version and rejects
+ * newer major.minor versions.
*/
- static final int PROTOCOL_VERSION = (11 << 10); // 0.11.0
+ static final int PROTOCOL_VERSION = (11 << 10) | 1; // 0.11.1
Review Comment:
nit: The Java codec pins the packed protocol version by hand while Go, C#, and Node check theirs against `core/binary_protocol/src/version.rs`. Add a Java test that reads the Rust constant, so a bump cannot leave Java clients rejected at login.
##########
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 {
Review Comment:
nit: HTTP transport errors never match `TransientNotAccepted`, so `send_retries` never retries a send over HTTP, and the retry table does not say so. Add one line naming the HTTP transport exception.
##########
core/journal/src/partition_journal/segments.rs:
##########
@@ -666,6 +720,9 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
self.segment_files_dirty = true;
self.segment_links_dirty = true;
self.segment_files.insert(key, file);
+ // Establish the inode and links before buffered appends can be acknowledged.
+ // A later shutdown barrier must not need a new directory descriptor.
+ self.sync_segment_files().await?;
Review Comment:
nit: `open_segment_file` syncs every open segment file and both directories even when the retained file already existed and no link was created. Set the dirty flags and sync only in the branch that creates the file or link.
##########
core/message_bus/src/replica/handshake.rs:
##########
@@ -149,10 +152,8 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>(
let header = msg.header();
let peer_id = header.replica;
let has_nonce = auth::has_nonce(&header.reserved_command);
- // Only a peer that sent a nonce speaks the authenticated protocol and is
- // waiting to read our response; a legacy (no-nonce) dialer delegates its
- // fd without reading, so a reject frame would land in its VSR reader instead.
- let nackable = auth.is_some() && has_nonce;
+ // Both authenticated and unauthenticated peers read the build-gate response.
+ let nackable = true;
Review Comment:
simplification: `nackable` is now the constant `true`, and the `AuthRequired` reject passes `false` directly. Delete the binding and pass `true` at the four call sites. Saves about two lines and one misleading local.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4173794166
##########
core/server/src/http/submit.rs:
##########
@@ -406,75 +410,62 @@ pub(in crate::http) async fn partition_write_replicated(
);
PartitionWriteError::Unavailable
})?;
- dispatch_partition_request(
- &state.shard,
- message,
- session.client_id,
- session.session,
- session.client_id,
- Some(session.user_id),
- None,
- )
- .await;
- drop(next_data_request_id);
- let outcome = compio::time::timeout(PARTITION_WRITE_REPLY_TIMEOUT, receiver).await;
- // Removes the slot unless the reply already fired, so a late commit
- // reply after a timeout sheds at the bus instead of leaking a waiter.
+ let attachment = session.attachment.borrow().clone();
+ let outcome =
+ compio::time::timeout(deadline.saturating_duration_since(Instant::now()), async {
+ dispatch_partition_request(
+ &state.shard,
+ message,
+ session.client_id,
+ session.session,
+ session.client_id,
+ Some(session.user_id),
+ Some((session.client_id, attachment)),
+ )
+ .await;
+ if let Some(dispatched) = dispatched.take() {
+ let _ = dispatched.send(Ok(()));
+ }
+ receiver.await
+ })
+ .await;
drop(guard);
match outcome {
Ok(Ok(reply)) => {
let reply = reply.into_contiguous();
classify_partition_reply(&reply).map(|header| (reply, header))
}
- // Cancelled (reply target torn down by session eviction mid-wait) or
- // elapsed: same caller contract either way - outcome unknown, 504.
Ok(Err(_)) | Err(_) => Err(PartitionWriteError::Timeout(operation)),
}
}
-/// Fire-and-forget produce (`?ack=none`): installs no reply slot and never
-/// awaits a commit. The commit still happens; its reply (and any gate-failure
-/// reply) targets a request id with no slot installed and is shed at the bus by
-/// design.
-///
-/// Admitted against the in-flight caps exactly like the acked path
-/// ([`partition_write_replicated`]): being reply-less does not make it free.
-/// The request still parks inside [`dispatch_partition_request`] for the
-/// routable-wait budget while pinning its buffered body, so it must count
-/// against the per-session and shard-global budgets or it would be an uncapped
-/// bypass. The in-flight guard is held across dispatch and released by `Drop`.
-/// A refusal is a synchronous 429/503, which is honest for `?ack=none`:
-/// admission runs before dispatch, so it is not a commit reply the caller
-/// opted out of - it is the request being turned away up front.
+/// Return after dispatch while a detached task holds the partition gate until
+/// the reply or deadline. A 202 still carries no commit or admission guarantee.
pub(in crate::http) async fn produce_unacked(
- state: &HttpInner,
- session: &HttpSession,
- body: &[u8],
+ state: &Rc<HttpInner>,
+ session: &Rc<HttpSession>,
+ body: Bytes,
) -> Result<(), PartitionWriteError> {
- let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?;
- // Same gate as the acked path, for the same ordering reason.
- let mut next_data_request_id = session.data_gate.lock().await;
- let request_id = *next_data_request_id;
- *next_data_request_id += 1;
- let message = build_request_message(
- Operation::SendMessages,
- session.client_id,
- session.session,
- request_id,
- body,
- );
- dispatch_partition_request(
- &state.shard,
- message,
- session.client_id,
- session.session,
- session.client_id,
- Some(session.user_id),
- None,
- )
- .await;
- drop(next_data_request_id);
- Ok(())
+ let state = Rc::clone(state);
+ let session = Rc::clone(session);
+ let (sent, dispatched) = oneshot::channel();
+ compio::runtime::spawn(async move {
Review Comment:
warning: `produce_unacked()` now runs every `?ack=none` produce through `partition_write()`, so the detached task holds an in-flight permit and a reply slot until the commit reply or the `504` deadline. Return after dispatch on the unacked path, as the previous body did.
##########
core/server/src/http/submit.rs:
##########
@@ -343,49 +340,56 @@ 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);
+ let _partition_guard = compio::time::timeout(
Review Comment:
warning: `partition_write()` holds the new per-namespace gate through the commit reply, so one credential cannot keep two writes to the same partition in flight. Release the gate before the reply wait, or state why the ordering needs the full round trip.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -6226,7 +6437,45 @@ where
// repair ring - and not even there on a single replica, which keeps no
// ring at all. A miss degrades to a successful send carrying no
// confirmation, a legal answer no client can tell from a real one.
- let committed_batch_stats = self.resolve_committed_visible_offsets(&drained);
+ let committed_batch_stats = self.resolve_committed_batch_stats(&drained);
+ let retirements: Vec<_> = drained
Review Comment:
warning: The commit walk builds the `retirements` vector for every drained entry on every batch, though only `RetireSession` entries use it. Build the vector only when the batch holds a `RetireSession`.
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -976,8 +987,8 @@ export class CommandResponseStream extends EventEmitter {
if (error instanceof LeaderMovedError)
throw error;
// Once bytes were handed to the socket, a local transport or decode
- // failure leaves the request outcome ambiguous. Register a fresh session
- // rather than replaying that request under a different client identity.
+ // failure leaves the request outcome ambiguous. Drop the transport while
+ // retaining the logical session and its bind proof.
Review Comment:
warning: The Node README still says that a disconnect or eviction invalidates the session and that later work must register a new session, but this path retains the session and its bind proof. Update the session text in `foreign/node/README.md`.
##########
foreign/go/internal/command/code.go:
##########
@@ -25,7 +25,7 @@ const (
GetSnapshotFileCode Code = 11
GetClusterMetadataCode Code = 12
DescribeOptionsCode Code = 13
- AttachConsumerSessionCode Code = 14
+ BindSessionCode Code = 15
Review Comment:
nit: `foreign/go/README.md` still lists binary command 14 for consumer-session attachment. Change it to 15, the new `BindSessionCode`.
##########
foreign/go/client/tcp/tcp_poll_routing.go:
##########
@@ -288,7 +290,15 @@ func (c *IggyTcpClient) pollOnRoute(ctx context.Context, key string, payload []b
}
if !slot.attached || slot.parent.client != route.parent.client ||
slot.parent.session != route.parent.session || slot.parent.watermark < route.parent.watermark {
- if _, state, err := slot.client.sendPollRequest(exchangeCtx, uint32(command.AttachConsumerSessionCode), route.parent.bytes()); err != nil {
+ current := c.pollSession.Load()
+ if current == nil || current.parent.client != route.parent.client || current.parent.session != route.parent.session {
+ return nil, ierror.ErrTransientNotAccepted
+ }
+ binding, err := vsr.SerializeBindSession(route.parent.bytes(), iggcon.Version, current.bindSecret)
+ if err != nil {
+ return nil, err
+ }
+ if _, state, err := slot.client.sendPollRequest(exchangeCtx, uint32(command.BindSessionCode), binding); err != nil {
Review Comment:
nit: The Go poll path discards the `BindSession` reply, while the Rust SDK rejects a mismatched bound session. Compare the reply session with `route.parent.session` and fail the route on mismatch.
##########
core/consensus/src/impls.rs:
##########
@@ -419,6 +431,47 @@ where
}
}
+impl<B, P> VsrConsensus<B, P>
+where
+ B: MessageBus,
+ P: Pipeline<Entry = PipelineEntry>,
+{
+ #[must_use]
+ pub fn retry_capacity(&self, table: &crate::ClientTable) -> usize {
+ if table.capacity_committed() {
+ return table.capacity();
+ }
+ self.pipeline
+ .borrow()
+ .head()
+ .filter(|entry| entry.header.retry_capacity != 0)
+ .map_or_else(
+ || table.capacity(),
+ |entry| entry.header.retry_capacity as usize,
+ )
+ }
+
+ /// Both queues own reservations. Clearing or refusing an entry releases it
+ /// automatically, and a commit replaces its reservation with live protection.
+ #[must_use]
+ pub fn has_retry_capacity(&self, table: &crate::ClientTable, client: u128) -> bool {
+ if table.contains(client) || message_bus::is_auto_commit_client(client) {
+ return true;
+ }
+ let capacity = self.retry_capacity(table);
+ let pipeline = self.pipeline.borrow();
+ let pending: HashSet<_> = pipeline
Review Comment:
warning: `has_retry_capacity()` collects every pending client id into a new `HashSet` on each admission, and a client without a committed entry pays that build on every request. Track the pending distinct count in the pipeline instead.
##########
foreign/node/scripts/check-vsr-protocol.mjs:
##########
@@ -283,16 +295,16 @@ for (const [name, value] of rustOperations) {
}
const protocolVersion =
- rustProtocolCargo.match(/^version = "([0-9]+)\.([0-9]+)\.([0-9]+)/m);
-assert.ok(protocolVersion, 'binary protocol crate version is missing');
+ rustProtocolVersion.match(/pub const IGGY_PROTOCOL_VERSION: u32 = pack_protocol_version\(([0-9]+), ([0-9]+), ([0-9]+)\);/);
+assert.ok(protocolVersion, 'binary protocol wire version is missing');
const nodePackedVersion = nodeRegister.match(
/export const IGGY_PROTOCOL_VERSION =\s*\(([0-9]+) << 20\) \| \(([0-9]+) << 10\) \| ([0-9]+);/
);
assert.ok(nodePackedVersion, 'Node packed protocol version source changed');
assert.deepEqual(
[Number(nodePackedVersion[1]), Number(nodePackedVersion[2])],
[Number(protocolVersion[1]), Number(protocolVersion[2])],
- 'Node protocol major.minor differs from iggy_binary_protocol'
+ 'Node protocol major.minor differs from the Rust wire version'
Review Comment:
warning: The drift check compares only the major and minor fields, so a patch mismatch between the Node mirror and the Rust wire version passes. Compare `nodePackedVersion[3]` with the patch field too.
##########
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>;
+ async fn session_identity(&self) -> Result<SessionIdentity, IggyError>;
+ async fn resume_vsr_session(&self) -> Result<(), IggyError>
+ where
+ Self: Sync,
+ {
+ let identity = self.session_identity().await?;
+ let request = BindSessionRequest {
+ version_info: crate::rust_sdk_version_info(self.sdk_version())?,
+ identity,
+ bind_secret: self.session_bind_secret().await?,
+ };
+ let response = self
+ .send_raw_with_response(
+ iggy_binary_protocol::codes::BIND_SESSION_CODE,
+ request.to_bytes(),
+ )
+ .await?;
+ let bound =
+ LoginRegisterResponse::decode_from(&response).map_err(|_| IggyError::InvalidFormat)?;
+ if bound.session != identity.session {
+ return Err(IggyError::InvalidSession(bound.session));
Review Comment:
warning: This branch returns `IggyError::InvalidSession`, whose message says the value must be non-zero, though the real conflict is a different bound session. Return `IggyError::SessionMismatch(identity.session, bound.session)` as `poll_routing.rs` does.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -6226,7 +6437,45 @@ where
// repair ring - and not even there on a single replica, which keeps no
// ring at all. A miss degrades to a successful send carrying no
// confirmation, a legal answer no client can tell from a real one.
- let committed_batch_stats = self.resolve_committed_visible_offsets(&drained);
+ let committed_batch_stats = self.resolve_committed_batch_stats(&drained);
+ let retirements: Vec<_> = drained
+ .iter()
+ .map(|entry| {
+ if entry.header.operation != Operation::RetireSession {
+ return None;
+ }
+ let prepare = self.log.journal().inner.repair_entry(entry.header.op)?;
+ SessionIdentity::decode_from(
+ prepare
+ .as_slice()
+ .get(iggy_binary_protocol::HEADER_SIZE..)?,
+ )
+ .ok()
+ })
+ .collect();
+ for ((entry, batch_stats), retirement) in
+ drained.iter().zip(&committed_batch_stats).zip(&retirements)
+ {
+ if self
+ .dedup
+ .commit_capacity(entry.header.retry_capacity as usize)
+ .is_err()
+ || (entry.header.operation == Operation::SendMessages && batch_stats.is_none())
Review Comment:
warning: The comment above says a batch-stats miss degrades to a successful send, but this check turns a `SendMessages` miss into `FatalCommit` and stops the node. Align the comment with the check, or drop the check.
##########
foreign/node/src/wire/command.code.ts:
##########
@@ -23,7 +23,7 @@ export const COMMAND_CODE = {
GetSnapshot: 11, // @TODO GET_SNAPSHOT_FILE_CODE: u32 = 11
GetClusterMetadata: 12,
DescribeOptions: 13,
- AttachConsumerSession: 14,
+ BindSession: 15,
Review Comment:
nit: `foreign/node/README.md` still lists binary command 14 for consumer-session attachment. Change it to 15, the new `BindSession` code.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java:
##########
@@ -43,7 +43,7 @@ final class VsrLoginCodec {
* Keep in sync with {@code core/binary_protocol/Cargo.toml}; the server
* accepts any client whose major.minor is not newer than its own.
*/
- static final int PROTOCOL_VERSION = (11 << 10); // 0.11.0
+ static final int PROTOCOL_VERSION = (11 << 10) | 1; // 0.11.1
Review Comment:
nit: The javadoc for `PROTOCOL_VERSION` tells readers to keep the value in sync with `core/binary_protocol/Cargo.toml`, but the wire version now lives in `core/binary_protocol/src/version.rs`, independent of the crate release. Point the javadoc at that file.
##########
core/journal/src/partition_journal/segments.rs:
##########
@@ -143,6 +143,53 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
}
}
+ /// Move the physical append point over a reserved offset gap.
+ /// The complete preceding chain must already be durably checkpointed.
+ ///
+ /// # Errors
+ /// Returns an error for an uncheckpointed tail, a regressing offset, or a failed barrier.
+ pub async fn reanchor_segment_storage(&mut self, next_offset: u64) -> io::Result<()> {
+ self.ensure_healthy()?;
+ let mut segments = self
+ .state
+ .segment_storage
+ .ok_or_else(|| invalid("segment storage is not enabled"))?;
+ if self.state.checkpoint != self.state.head || self.durable_head != self.state.head {
+ return Err(invalid("cannot reanchor an uncheckpointed segment tail"));
+ }
+ if next_offset < segments.tail.position.next_offset {
+ return Err(invalid(
+ "reserved segment offset regresses the physical tail",
+ ));
+ }
+ if next_offset == segments.tail.position.next_offset {
+ return Ok(());
+ }
+ segments.reset_position(SegmentPosition {
+ start_offset: next_offset,
+ length: 0,
+ next_offset,
+ })?;
+ let state = JournalState {
+ segment_storage: Some(segments),
+ ..self.state
+ };
+ self.poisoned = true;
+ self.open_segment_file(
+ segments.tail.generation,
+ next_offset,
+ 0,
+ self.preallocate_segments.then_some(segments.max_size),
+ )
+ .await?;
+ self.sync_segment_files().await?;
Review Comment:
simplification: `open_segment_file()` already syncs these files and clears both dirty flags, and the fresh generation from `reset_position()` makes its early return impossible, so this call syncs nothing. Delete the redundant `sync_segment_files()` call.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4183150322
##########
core/server/src/http/submit.rs:
##########
@@ -406,75 +410,62 @@ pub(in crate::http) async fn partition_write_replicated(
);
PartitionWriteError::Unavailable
})?;
- dispatch_partition_request(
- &state.shard,
- message,
- session.client_id,
- session.session,
- session.client_id,
- Some(session.user_id),
- None,
- )
- .await;
- drop(next_data_request_id);
- let outcome = compio::time::timeout(PARTITION_WRITE_REPLY_TIMEOUT, receiver).await;
- // Removes the slot unless the reply already fired, so a late commit
- // reply after a timeout sheds at the bus instead of leaking a waiter.
+ let attachment = session.attachment.borrow().clone();
+ let outcome =
+ compio::time::timeout(deadline.saturating_duration_since(Instant::now()), async {
+ dispatch_partition_request(
+ &state.shard,
+ message,
+ session.client_id,
+ session.session,
+ session.client_id,
+ Some(session.user_id),
+ Some((session.client_id, attachment)),
+ )
+ .await;
+ if let Some(dispatched) = dispatched.take() {
+ let _ = dispatched.send(Ok(()));
+ }
+ receiver.await
+ })
+ .await;
drop(guard);
match outcome {
Ok(Ok(reply)) => {
let reply = reply.into_contiguous();
classify_partition_reply(&reply).map(|header| (reply, header))
}
- // Cancelled (reply target torn down by session eviction mid-wait) or
- // elapsed: same caller contract either way - outcome unknown, 504.
Ok(Err(_)) | Err(_) => Err(PartitionWriteError::Timeout(operation)),
}
}
-/// Fire-and-forget produce (`?ack=none`): installs no reply slot and never
-/// awaits a commit. The commit still happens; its reply (and any gate-failure
-/// reply) targets a request id with no slot installed and is shed at the bus by
-/// design.
-///
-/// Admitted against the in-flight caps exactly like the acked path
-/// ([`partition_write_replicated`]): being reply-less does not make it free.
-/// The request still parks inside [`dispatch_partition_request`] for the
-/// routable-wait budget while pinning its buffered body, so it must count
-/// against the per-session and shard-global budgets or it would be an uncapped
-/// bypass. The in-flight guard is held across dispatch and released by `Drop`.
-/// A refusal is a synchronous 429/503, which is honest for `?ack=none`:
-/// admission runs before dispatch, so it is not a commit reply the caller
-/// opted out of - it is the request being turned away up front.
+/// Return after dispatch while a detached task holds the partition gate until
+/// the reply or deadline. A 202 still carries no commit or admission guarantee.
pub(in crate::http) async fn produce_unacked(
- state: &HttpInner,
- session: &HttpSession,
- body: &[u8],
+ state: &Rc<HttpInner>,
+ session: &Rc<HttpSession>,
+ body: Bytes,
) -> Result<(), PartitionWriteError> {
- let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?;
- // Same gate as the acked path, for the same ordering reason.
- let mut next_data_request_id = session.data_gate.lock().await;
- let request_id = *next_data_request_id;
- *next_data_request_id += 1;
- let message = build_request_message(
- Operation::SendMessages,
- session.client_id,
- session.session,
- request_id,
- body,
- );
- dispatch_partition_request(
- &state.shard,
- message,
- session.client_id,
- session.session,
- session.client_id,
- Some(session.user_id),
- None,
- )
- .await;
- drop(next_data_request_id);
- Ok(())
+ let state = Rc::clone(state);
+ let session = Rc::clone(session);
+ let (sent, dispatched) = oneshot::channel();
+ compio::runtime::spawn(async move {
Review Comment:
The reason for keeping `?ack=none` inside `partition_write()` is clear from `core/server/src/http/submit.rs:377`, but the admission doc now contradicts it. `core/server/src/http/admission.rs:40` still says that `?ack=none` produces install no reply slot and never await a commit, while `produce_unacked` spawns `partition_write` at `core/server/src/http/submit.rs:458`, which takes the in-flight permit at `core/server/src/http/submit.rs:385`, installs a reply slot at `core/server/src/http/submit.rs:402` and awaits the reply at `core/server/src/http/submit.rs:432`, so an unacked write holds its budget until the commit or the reply timeout.
##########
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:
The busy-partition stall is fixed by the `pending_request` check at `core/partitions/src/iggy_partition.rs:709` and the backup-quorum finalize at `core/server/src/consumer_group/liveness.rs:290`, but two paths still block finalization. A tombstoned or fatal namespace never reads as retired (`core/shard/src/lib.rs:2107`, `core/shard/src/poll.rs:183`, `core/partitions/src/iggy_partition.rs:705`) and the sweep has no filter for it, so on a single node, where the primary counts itself only after a complete sweep at `core/server/src/consumer_group/liveness.rs:449`, the cursor resets every pass at `core/server/src/consumer_group/liveness.rs:422` and no ended session is finalized until the table fills and new sessions are refused. Every applied `TruncatePartition` also bumps the revision at `core/metadata/src/stm/stream.rs:972` even when the watermark stays put, and `core/server/src/consumer_group/lease.rs:176` then zeroes the cursor and clears the reporters, so steady segment deletion s
tarves finalization while the sweep still retires one identity at a time across every partition (`core/server/src/consumer_group/liveness.rs:281`).
##########
core/server/src/partition_helpers.rs:
##########
@@ -612,7 +603,37 @@ async fn load_partition(
dir: directory.clone(),
source,
},
- })?;
+ };
+ if replica_count == 1
Review Comment:
This still reproduces at bb42eda0. For a partition that metadata names but whose directory never existed, boot calls `load_partition_or_fence` at `core/server/src/boot/recovery.rs:209`, the loader takes the `NotFound` arm at `core/server/src/partition_helpers.rs:255` and creates the hierarchy, and then the single-node frontier check at `core/server/src/partition_helpers.rs:610` returns `InvalidData`, which `core/server/src/partition_helpers.rs:588` maps to `Refused` and `core/server/src/partition_helpers.rs:391` tombstones. The only fresh-build path for an absent directory is in the reconciler at `core/server/src/partition_reconciler.rs:792`, and its gate at `core/server/src/partition_reconciler.rs:656` refuses a namespace that boot already tombstoned. The fix commit only widened the error mapping, and the README paragraph at `core/server/README.md:154` predates this comment.
##########
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:
This is still open at bb42eda0. Both `IGGY_PROTOCOL_VERSION` and `IGGY_PROTOCOL_VERSION_MIN` remain `pack_protocol_version(0, 11, 1)` at `core/binary_protocol/src/version.rs:103` and `core/binary_protocol/src/version.rs:116`, so a pre-PR client built from `0.11.1-edge.1` still passes the gate at `core/server/src/dispatch/session_ops.rs:1114`, and `decode_after_prefix` reads its username, password and context bytes as the 32-byte bind secret at `core/binary_protocol/src/requests/users/login_register.rs:146`. That client gets `MalformedLogin`, or `InvalidCredentials` for a body of 32 bytes or more, and never `IncompatibleProtocol`. The notes added at `core/binary_protocol/src/version.rs:100` and `core/server/README.md:94` describe the clash, while `core/binary_protocol/src/requests/users/login_register.rs:88` itself requires a version bump for any wire-incompatible change.
##########
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:
This still reproduces at bb42eda0. Boot passes `allow_bootstrap = false` at `core/server/src/boot/recovery.rs:218`, so every multi-replica load reaches `fence_missing_history` at `core/server/src/partition_helpers.rs:620`, and `core/partitions/src/persistence.rs:504` marks materialization missing whenever no frontier is published, without reading any segment or WAL bytes. A fenced replica never raises `recovery_election_allowed` (`core/partitions/src/iggy_partition.rs:963`), only a caught-up primary builds a transfer offer at `core/partitions/src/state_transfer.rs:2037`, and the fence clears only after an install at `core/partitions/src/state_transfer.rs:2982`, so after a whole-cluster power loss before the first frontier no replica can elect or serve and the group stays dark. The README text at `core/server/README.md:157` predates this comment and still sends the operator to delete and recreate the partition.
##########
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:
The freeze is documented at `core/server/README.md:66`, but the operator configuration still says the opposite. `core/server/config.toml:863` tells operators that raising `clients_table_max` lifts the HTTP session cap too, which holds only for new groups, because the HTTP cap follows the configured value at `core/server/src/http/session.rs:51` while an existing metadata table keeps its committed size, and the same text sits in `core/configs/src/server_config/metadata.rs:113`. A partition that replays its capacity from the WAL at `core/partitions/src/iggy_partition.rs:6516` without a receipt checkpoint also adopts the committed `retry_capacity` with no mismatch warning, unlike the paths at `core/metadata/src/impls/metadata.rs:1048` and `core/partitions/src/state_transfer.rs:4158`.
##########
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:
The owner check at `core/server/src/dispatch/mod.rs:807` closes the missing-user case but not a reused id. `DeleteUser` at `core/metadata/src/stm/user.rs:575` frees the slab key without touching the client table, so the entry keeps the old `user_id` and its verifier, and `IdSlab::insert` at `core/metadata/src/stm/id_slab.rs:78` hands the lowest freed key to the next `CreateUser` at `core/metadata/src/stm/user.rs:474`. After a delete and a new create, the holder of the deleted user's bind secret passes `bind_session` at `core/consensus/src/client_table.rs:617` and the active-user check, and then acts with the new user's permissions. Deactivation only sets `user.status` at `core/metadata/src/stm/user.rs:547` and nothing after the bind checks it, so an already bound connection keeps passing `authorize_partition_op` at `core/server/src/dispatch/authz.rs:57`, which makes the statement at `core/server/README.md:64` that deactivation ends session access false.
##########
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:
The race and the blocking read are fixed, since `bootstrap` warms the identity at `core/server/src/boot/mod.rs:302` and `OnceLock::get_or_init` at `core/message_bus/src/replica/handshake.rs:544` hashes once. The executable hash itself is unchanged: `core/message_bus/src/replica/handshake.rs:545` reads the whole `current_exe`, `core/message_bus/src/replica/handshake.rs:564` mixes the digest into the peer identity unconditionally, and no config key, env var or flag turns it off, so a stripped, distro-packaged or rebuilt binary of the same release is still rejected with `IncompatibleBuild` at `core/message_bus/src/replica/handshake.rs:182`. The note at `core/server/README.md:92` documents the limit, but a rolling replace across build variants stays impossible.
##########
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:
The doc at `core/consensus/src/impls.rs:189` is fixed, but the configuration still promises the eviction that this PR removed. `core/server/config.toml:863` says that the client whose last commit is oldest is evicted and `core/server/config.toml:973` says that under-sizing degrades rather than breaks, with the same text in `core/configs/src/server_config/metadata.rs:113` and `core/configs/src/server_config/partition.rs:209`, while `commit_request` now returns `TooManyEntries` at `core/consensus/src/client_table.rs:1257` and admission refuses new writers through `has_retry_capacity` at `core/partitions/src/iggy_partition.rs:4680`. An operator who under-sizes `dedup_clients_max` gets `TransientNotAccepted` for new writers, and each freed slot waits on the serial retirement sweep at `core/server/src/consumer_group/liveness.rs:281`.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4174478262
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2312,18 +2406,24 @@ where
.then(|| self.consensus().last_prepare_checksum())
})
};
- let checkpoint_prepare = self
- .log
- .journal()
- .inner
- .repair_entry(commit_op)
- .map_or_else(Vec::new, |prepare| prepare.as_slice().to_vec());
- if self.persistence.is_some()
+ let checkpoint_prepare = if require_prepare {
+ self.log
+ .journal()
+ .inner
+ .repair_entry(commit_op)
+ .map_or_else(Vec::new, |prepare| prepare.as_slice().to_vec())
+ } else {
+ Vec::new()
+ };
+ if require_prepare
+ && self.persistence.is_some()
&& (prepare_checksum.is_none() || (commit_op > 0 && checkpoint_prepare.is_empty()))
{
return Err(PartitionTransferUnavailable::MissingPrepareChecksum { op: commit_op });
}
Ok(ConsumerOffsetsWire {
+ required_metadata_frontier: self.required_metadata_frontier,
+ dedup_capacity: self.dedup().capacity(),
Review Comment:
The transfer offer sends `self.dedup().capacity()` even when no prepare committed yet, so the receiver of a commit_op 0 transfer installs the sender's local `dedup_clients_max` as committed capacity through `install_watermarks`. If the first committed prepare nominates a different `retry_capacity`, the receiver's commit walk fails `commit_capacity` at `core/partitions/src/iggy_partition.rs:6453`, sets `FatalCommit`, and the node shuts down. This is a separate path from the ignored configuration change at `core/consensus/src/client_table.rs:765`, because here a node-local value fixes the capacity before any prepare does, and the mismatch is fatal.
##########
foreign/go/client/tcp/tcp_session_management.go:
##########
@@ -29,27 +29,31 @@ import (
)
func (c *IggyTcpClient) LoginUser(ctx context.Context, username string, password string) (*iggcon.IdentityInfo, error) {
- body, err := vsr.SerializeLoginRegister(username, password, iggcon.Version)
+ secret := vsr.NewBindSecret()
Review Comment:
Go, Java and C# register a fresh client id and secret on every transport loss (`foreign/go/client/tcp/tcp_core.go:1024`, `foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs:761`, Java `beginRegister`), while Rust and Node resume with `BindSession`. With eviction gone, each reconnect from these three SDKs leaves the old entry holding a slot until the 30 s session timeout, which adds SDK-driven churn on top of the slot pressure described at `core/consensus/src/client_table.rs:1289`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4380,80 +4528,103 @@ where
// earned, not the typed 404 the existence check would raise now
// that the offset is gone.
//
- // A replay racing its own in-flight original is absorbed here: the
- // slice only knows committed ops, so it cannot yet see the copy
- // still in the pipeline. Keyed on the exact `(client, request)`
- // for the transports that keep several writes in flight per
- // client (HTTP handlers on one session, the pipelining SDKs):
- // matching any request from the client would serialize them to
- // one in-flight write per group. A lockstep TCP connection never
- // has a second request here to begin with.
+ // One unresolved mutation per session/group protects the original
+ // receipt until the caller resolves it. An exact pending retry is
+ // still uncertain; a different request waits for that original
+ // instead of displacing its receipt.
//
- // Those same transports can deliver a client's ids out of order:
+ // HTTP handlers and pipelining SDKs can deliver a client's ids out of order:
// a write refused transiently here is replayed after its
// successors committed. The slice therefore keeps a committed-id
// window under the watermark (`consensus::COMMITTED_WINDOW_BITS`)
// and admits an unmarked id inside it instead of absorbing it.
if !is_auto_commit_client(client_id) {
- if consensus.pipeline_has_message_from_client_request(client_id, request) {
+ if let Some((pending_session, pending_request, pending_operation)) =
Review Comment:
The preflight replaced the exact-replay check `pipeline_has_message_from_client_request` with `pending_request`, so on the binary transports a pipelined second request from the same client in the group gets `TransientNotAccepted`. Pipelining clients such as the Java async SDK handle that code as a routing failure, retry on a timer and run leader rechecks for a lane that is only busy. The thread at `core/server/src/http/submit.rs:370` covers the HTTP side effect of the double resolve. This one hits any pipelining binary client.
##########
core/sdk/src/clients/client.rs:
##########
@@ -755,7 +756,8 @@ impl IggyClient {
/// client.connect().await?;
///
/// let producer = client
- /// .producer("stream_name", "topic_name")? // returns IggyProducerBuilder from IggyClient
+ /// .producer("stream_name", "topic_name")?
+ /// .topic_durability(iggy::prelude::Durability::Persisted) // returns IggyProducerBuilder from IggyClient
Review Comment:
The `returns IggyProducerBuilder from IggyClient` comment moved onto the `.topic_durability(...)` line, so it now labels the wrong call.
##########
core/metadata/src/impls/recovery.rs:
##########
@@ -642,19 +676,22 @@ 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).
- if client_table.get_epoch(header.client).is_some() {
+ // Every client commit must restore its original receipt before admission.
+ if header.client != consensus::client_table::RESERVED_CLIENT_ID
+ && header.operation != Operation::CompleteConsumerGroupRevocation
+ {
let cached = build_reply_message_with(header, reply.reply_body_len(), |dst| {
reply.write_reply_body(dst);
});
- // Skips (never panics) on a stale-request replay: capacity
- // eviction is replica-local and unlogged, so a WAL can legitimately
- // replay a lower request id onto a preserved watermark. Recovery
- // must boot from such a WAL, not refuse it.
- let _ = client_table.commit_reply(header.client, header.user_id, cached);
+ if client_table.commit_reply(header.client, header.user_id, cached)
+ != consensus::client_table::CommitReply::Cached
Review Comment:
Recovery now refuses to boot on any replay outcome other than `Cached`, but `checkpoint_if_needed` at `core/metadata/src/impls/metadata.rs:3547` can snapshot a client table that a state transfer installed ahead of `commit_min`. `install_state_transfer` sets the table at `table_frontier` while the state machine sits at `snapshot_seq`, so replaying ops in `(snap_op, table_frontier]` returns `SkippedRegression` for a client that committed twice there, or fails `commit_register` for a session that ended there, and the node cannot restart.
##########
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);
+ entry.push_latest(cached.clone());
+ Ok(cached)
+ }
+
+ /// Fold an AUTO offset request into the partition watermark without a receipt.
Review Comment:
The doc of `commit_request` says that it folds AUTO offset requests without a receipt, but the commit walk skips dedup for auto-commit ops at `core/partitions/src/iggy_partition.rs:6621`, and the only production caller is `commit_partition_reply`, although the method stays `pub`.
##########
core/sdk/src/session.rs:
##########
@@ -98,66 +86,50 @@ impl ConsensusSession {
self.session.is_some()
}
- /// Bind the session after register commits through consensus.
- /// The `session` value is the commit op number from the server's reply.
- ///
- /// # Panics
- /// Panics if already bound (drop and create a new session instead).
- pub fn bind(&mut self, session: u64) {
- assert!(
- self.session.is_none(),
- "session already bound (session={})",
- self.session.unwrap()
- );
- assert!(session > 0, "session must be > 0");
+ /// Registration proof retained across reconnects and session bindings.
+ pub fn bind_secret(&self) -> BindSecret {
+ self.bind_secret.clone()
+ }
+
+ /// Accept the original registration, including its replay after reconnect.
+ pub fn bind(&mut self, session: u64) -> Result<(), IggyError> {
+ if session == 0 {
+ return Err(IggyError::InvalidSession(session));
+ }
+ if let Some(bound) = self.session
+ && bound != session
+ {
+ return Err(IggyError::SessionMismatch(bound, session));
+ }
self.session = Some(session);
+ Ok(())
}
/// Returns the request ID for the register operation (always 0).
///
- /// Must be called exactly once, before [`bind`](Self::bind).
- ///
- /// # Panics
- /// Panics if called more than once or after the session is bound.
- pub fn register_request_id(&mut self) -> u64 {
- assert!(
- !self.register_consumed,
- "register_request_id already called"
- );
- assert!(!self.is_bound(), "register_request_id called after bind");
- self.register_consumed = true;
+ pub const fn register_request_id(&self) -> u64 {
Review Comment:
`register_request_id` and `begin_register` are now both `const fn` that return 0, and only `core/sdk/src/vsr.rs:89` calls one of them outside tests, so the second method is redundant public API.
##########
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 {
Review Comment:
`ClientTable::check_slice_request` is public but has no production caller since the partition preflight moved to `check_partition_request` at `core/partitions/src/iggy_partition.rs:4564`, so it and its wrapper-only tests are dead code.
##########
core/server/src/session_manager.rs:
##########
@@ -471,10 +380,10 @@ impl SessionManager {
pub fn get_user_id(&self, connection_id: u128) -> Option<u32> {
let conn = self.connections.get(&connection_id)?;
match conn.state {
- ConnectionState::Authenticated { user_id } | ConnectionState::Bound { user_id, .. } => {
+ ConnectionState::Bound { user_id, .. } if self.get_session(connection_id).is_some() => {
Review Comment:
`get_user_id` matches `Bound` and then calls `get_session`, which looks up the same connection again only to check that the attachment is valid on the `conn` already in hand.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -372,14 +374,33 @@ impl<T: PollTransport> PollRouter<T> {
|| session.session != route.consumer_session.session
|| session.metadata_watermark < route.consumer_session.metadata_watermark
}) {
- connection
+ let response = connection
Review Comment:
The poll path repeats the `BindSession` request encoding, reply decode and session check of `resume_vsr_session` at `core/common/src/traits/binary_transport.rs:84`, so the two copies of the handshake can drift.
##########
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!(
Review Comment:
The allowed partition receipt operations (`SendMessages | StoreConsumerOffset | DeleteConsumerOffset`) are matched in three places, here, in `install_watermarks` at `core/consensus/src/client_table.rs:1367` and at `core/shard/src/poll.rs:63`, so write, install and poll validation can drift apart.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4387:
URL: https://github.com/apache/iggy/pull/4387#issuecomment-5970589662
/skill team-review-slim
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4182072742
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -4380,80 +4528,103 @@ where
// earned, not the typed 404 the existence check would raise now
// that the offset is gone.
//
- // A replay racing its own in-flight original is absorbed here: the
- // slice only knows committed ops, so it cannot yet see the copy
- // still in the pipeline. Keyed on the exact `(client, request)`
- // for the transports that keep several writes in flight per
- // client (HTTP handlers on one session, the pipelining SDKs):
- // matching any request from the client would serialize them to
- // one in-flight write per group. A lockstep TCP connection never
- // has a second request here to begin with.
+ // One unresolved mutation per session/group protects the original
+ // receipt until the caller resolves it. An exact pending retry is
+ // still uncertain; a different request waits for that original
+ // instead of displacing its receipt.
//
- // Those same transports can deliver a client's ids out of order:
+ // HTTP handlers and pipelining SDKs can deliver a client's ids out of order:
// a write refused transiently here is replayed after its
// successors committed. The slice therefore keeps a committed-id
// window under the watermark (`consensus::COMMITTED_WINDOW_BITS`)
// and admits an unmarked id inside it instead of absorbing it.
if !is_auto_commit_client(client_id) {
- if consensus.pipeline_has_message_from_client_request(client_id, request) {
+ if let Some((pending_session, pending_request, pending_operation)) =
Review Comment:
will be fixxed in future PRs
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #4387:
URL: https://github.com/apache/iggy/pull/4387
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #4387:
URL: https://github.com/apache/iggy/pull/4387#discussion_r4183149407
##########
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:
This is still open at bb42eda0. Both `IGGY_PROTOCOL_VERSION` and `IGGY_PROTOCOL_VERSION_MIN` remain `pack_protocol_version(0, 11, 1)` at `core/binary_protocol/src/version.rs:103` and `core/binary_protocol/src/version.rs:116`, so a pre-PR client built from `0.11.1-edge.1` still passes the gate at `core/server/src/dispatch/session_ops.rs:1114`, and `decode_after_prefix` reads its username, password and context bytes as the 32-byte bind secret at `core/binary_protocol/src/requests/users/login_register.rs:146`. That client gets `MalformedLogin`, or `InvalidCredentials` for a body of 32 bytes or more, and never `IncompatibleProtocol`. The notes added at `core/binary_protocol/src/version.rs:100` and `core/server/README.md:94` describe the clash, while `core/binary_protocol/src/requests/users/login_register.rs:88` itself requires a version bump for any wire-incompatible change.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(consensus): persist retry receipts and shared sessions [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4387:
URL: https://github.com/apache/iggy/pull/4387#issuecomment-5958044913
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4387?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `70.54264%` with `38 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 86.64%. Comparing base ([`7c10d8a`](https://app.codecov.io/gh/apache/iggy/commit/7c10d8adbe846143234307c00efabec2f77ddb43?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`98a2f15`](https://app.codecov.io/gh/apache/iggy/commit/98a2f15e6b993b259e5e8587a454c17fac012d43?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4387?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [foreign/go/client/tcp/tcp\_poll\_routing.go](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fgo%2Fclient%2Ftcp%2Ftcp_poll_routing.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9jbGllbnQvdGNwL3RjcF9wb2xsX3JvdXRpbmcuZ28=) | 52.94% | [4 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/go/internal/vsr/register.go](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fgo%2Finternal%2Fvsr%2Fregister.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9pbnRlcm5hbC92c3IvcmVnaXN0ZXIuZ28=) | 70.37% | [6 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/csharp/Iggy\_SDK/Vsr/ConsensusSession.cs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fcsharp%2FIggy_SDK%2FVsr%2FConsensusSession.cs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9jc2hhcnAvSWdneV9TREsvVnNyL0NvbnNlbnN1c1Nlc3Npb24uY3M=) | 75.00% | [4 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/go/errors/errors\_gen.go](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fgo%2Ferrors%2Ferrors_gen.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9lcnJvcnMvZXJyb3JzX2dlbi5nbw==) | 50.00% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/go/client/tcp/tcp\_session\_management.go](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fgo%2Fclient%2Ftcp%2Ftcp_session_management.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9jbGllbnQvdGNwL3RjcF9zZXNzaW9uX21hbmFnZW1lbnQuZ28=) | 50.00% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/csharp/Iggy\_SDK/Vsr/LoginRegister.cs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fcsharp%2FIggy_SDK%2FVsr%2FLoginRegister.cs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9jc2hhcnAvSWdneV9TREsvVnNyL0xvZ2luUmVnaXN0ZXIuY3M=) | 89.28% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/go/internal/vsr/session.go](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fgo%2Finternal%2Fvsr%2Fsession.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9nby9pbnRlcm5hbC92c3Ivc2Vzc2lvbi5nbw==) | 33.33% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...SDK/IggyClient/Implementations/TcpMessageStream.cs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fcsharp%2FIggy_SDK%2FIggyClient%2FImplementations%2FTcpMessageStream.cs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9jc2hhcnAvSWdneV9TREsvSWdneUNsaWVudC9JbXBsZW1lbnRhdGlvbnMvVGNwTWVzc2FnZVN0cmVhbS5jcw==) | 50.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [foreign/csharp/Iggy\_SDK/Vsr/VsrOperation.cs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=foreign%2Fcsharp%2FIggy_SDK%2FVsr%2FVsrOperation.cs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9jc2hhcnAvSWdneV9TREsvVnNyL1Zzck9wZXJhdGlvbi5jcw==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4387 +/- ##
============================================
- Coverage 87.86% 86.64% -1.23%
+ Complexity 1579 1577 -2
============================================
Files 1290 1288 -2
Lines 230145 228528 -1617
Branches 193482 191968 -1514
============================================
- Hits 202228 198000 -4228
- Misses 23195 25910 +2715
+ Partials 4722 4618 -104
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.03% <ø> (-0.04%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.70% <ø> (-0.05%)` | :arrow_down: |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `59.95% <82.53%> (-17.74%)` | :arrow_down: |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.24% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.67% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.48% <ø> (-3.40%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4387/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `60.21% <59.09%> (-10.06%)` | :arrow_down: |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4387?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/binary\_protocol/src/codes.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fcodes.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvZGVzLnJz) | `100.00% <ø> (ø)` | |
| [core/binary\_protocol/src/consensus/command.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fconsensus%2Fcommand.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvbnNlbnN1cy9jb21tYW5kLnJz) | `100.00% <ø> (ø)` | |
| [.../binary\_protocol/src/consensus/consumer\_session.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fconsensus%2Fconsumer_session.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvbnNlbnN1cy9jb25zdW1lcl9zZXNzaW9uLnJz) | `96.85% <ø> (-0.03%)` | :arrow_down: |
| [core/binary\_protocol/src/consensus/header.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fconsensus%2Fheader.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvbnNlbnN1cy9oZWFkZXIucnM=) | `83.29% <ø> (-0.08%)` | :arrow_down: |
| [core/binary\_protocol/src/consensus/operation.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fconsensus%2Foperation.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2NvbnNlbnN1cy9vcGVyYXRpb24ucnM=) | `97.03% <ø> (-0.07%)` | :arrow_down: |
| [core/binary\_protocol/src/dispatch.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fdispatch.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL2Rpc3BhdGNoLnJz) | `89.43% <ø> (-0.15%)` | :arrow_down: |
| [...nary\_protocol/src/requests/users/login\_register.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Fusers%2Flogin_register.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3VzZXJzL2xvZ2luX3JlZ2lzdGVyLnJz) | `100.00% <ø> (ø)` | |
| [...ocol/src/requests/users/login\_register\_with\_pat.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Frequests%2Fusers%2Flogin_register_with_pat.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3JlcXVlc3RzL3VzZXJzL2xvZ2luX3JlZ2lzdGVyX3dpdGhfcGF0LnJz) | `100.00% <ø> (ø)` | |
| [...ry\_protocol/src/responses/messages/poll\_routing.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fresponses%2Fmessages%2Fpoll_routing.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3Jlc3BvbnNlcy9tZXNzYWdlcy9wb2xsX3JvdXRpbmcucnM=) | `100.00% <ø> (ø)` | |
| [...y\_protocol/src/responses/messages/send\_messages.rs](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree&filepath=core%2Fbinary_protocol%2Fsrc%2Fresponses%2Fmessages%2Fsend_messages.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9iaW5hcnlfcHJvdG9jb2wvc3JjL3Jlc3BvbnNlcy9tZXNzYWdlcy9zZW5kX21lc3NhZ2VzLnJz) | `97.85% <ø> (ø)` | |
| ... and [99 more](https://app.codecov.io/gh/apache/iggy/pull/4387?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [130 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4387/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org