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