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