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/09/02 14:23:27 UTC
[PR] refactor(server): make dispatch routing and failure paths explicit [iggy]
hubcio opened a new pull request, #4036:
URL: https://github.com/apache/iggy/pull/4036
The TCP dispatch funnel hid its rules in control flow: seven
ordered if-return probes where the order was the only spec, two
rewrite chains written out inline in two places, twenty send
sites each picking a failure channel by whichever builder they
called, two client-request handlers on shard 0 with separate
queues, and a read gate that let unknown codes fall through to a
catch-all.
Each is now written down once. classify() returns a RequestClass
that handle_client_request matches on, so the order of checks
inside classify is the routing. rewrite.rs holds both rewrite
chains as pure functions. dispatch/failure.rs holds the failure
channel table and the one exit every host-built frame takes. Each
shard builds one client-request handler that shard 0's transports
share, dropping the duplicate queues, the double hook install and
a reference cycle that leaked the shard. The read gate now
decides every code in the command table instead of passing
unlisted ones to the builder's empty-ok. Tests pin the probe
order, each channel's bytes, and every table entry.
Removing the cycle exposed a shutdown bug it had hidden: shard 0
owns the only write handle to the metadata state machine and
could drop it while peers were still reading, panicking their
pumps. Peers now count themselves out once their runtime is gone,
and shard 0 waits for that, bounded by shutdown_join_timeout.
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4036:
URL: https://github.com/apache/iggy/pull/4036#issuecomment-5512195177
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4036?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 84.65%. Comparing base ([`e6ec869`](https://app.codecov.io/gh/apache/iggy/commit/e6ec869500c3b362bf7f4b85650b7fc6df9a5fa1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`f5da637`](https://app.codecov.io/gh/apache/iggy/commit/f5da63735445203f55f86f8767fee8ae37598a20?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4036 +/- ##
============================================
- Coverage 85.12% 84.65% -0.48%
Complexity 1402 1402
============================================
Files 1236 1236
Lines 181607 180869 -738
Branches 147900 147162 -738
============================================
- Hits 154596 153112 -1484
- Misses 22957 23764 +807
+ Partials 4054 3993 -61
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4036/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/4036/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.05% <ø> (+<0.01%)` | :arrow_up: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4036/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `67.29% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4036/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.39% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4036/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.06% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4036/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.65% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4036/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.24% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4036/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.05% <ø> (-13.24%)` | :arrow_down: |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4036?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/server/src/boot/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fboot%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Jvb3QvbW9kLnJz) | `86.75% <ø> (-0.15%)` | :arrow_down: |
| [core/server/src/boot/recovery.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fboot%2Frecovery.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Jvb3QvcmVjb3ZlcnkucnM=) | `95.93% <ø> (-0.04%)` | :arrow_down: |
| [core/server/src/boot/threads.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fboot%2Fthreads.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Jvb3QvdGhyZWFkcy5ycw==) | `76.55% <ø> (ø)` | |
| [core/server/src/dispatch/authz.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Fauthz.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL2F1dGh6LnJz) | `95.90% <ø> (+1.13%)` | :arrow_up: |
| [core/server/src/dispatch/mod.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL21vZC5ycw==) | `91.77% <ø> (-0.51%)` | :arrow_down: |
| [core/server/src/dispatch/partition.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Fpartition.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL3BhcnRpdGlvbi5ycw==) | `89.64% <ø> (+3.70%)` | :arrow_up: |
| [core/server/src/dispatch/reads.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Freads.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL3JlYWRzLnJz) | `90.07% <ø> (+0.88%)` | :arrow_up: |
| [core/server/src/dispatch/session\_ops.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Fsession_ops.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL3Nlc3Npb25fb3BzLnJz) | `93.33% <ø> (+1.32%)` | :arrow_up: |
| [core/server/src/dispatch/test\_support.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Ftest_support.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL3Rlc3Rfc3VwcG9ydC5ycw==) | `82.94% <ø> (+0.48%)` | :arrow_up: |
| [core/server/src/http/handlers.rs](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fhttp%2Fhandlers.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2h0dHAvaGFuZGxlcnMucnM=) | `79.53% <ø> (ø)` | |
| ... and [1 more](https://app.codecov.io/gh/apache/iggy/pull/4036?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [40 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4036/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
Re: [PR] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4036:
URL: https://github.com/apache/iggy/pull/4036#discussion_r3926682550
##########
core/server/src/rewrite.rs:
##########
@@ -0,0 +1,534 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The two pre-consensus request-rewrite chains, side by side.
+//!
+//! [`tcp_chain`] serves the TCP funnel, [`http_chain`] the HTTP submit. Each
+//! step rewrites or validates a request BEFORE consensus, so a rejected
+//! request burns no replicated log entry and no plaintext secret enters
+//! consensus. The chains enter the PAT rewrite through different functions
+//! on purpose: TCP resolves the acting user from the transport
+//! `SessionManager` ([`maybe_rewrite_pat_request`]), HTTP authenticates
+//! against its own session table and passes the resolved `user_id`
+//! ([`rewrite_pat_request_for_user`]).
+//!
+//! Three more links complete the chains but live in their spines, because
+//! they fire partition-read mesh RPCs or are plane-specific. Two are async:
+//! the consumer-group Join/Leave enrichment ([`crate::consumer_group`], the
+//! TCP funnel calls it after [`tcp_chain`]) and the `DeleteSegments` ->
+//! `TruncatePartition` resolution
+//! (`dispatch::partition::resolve_delete_segments_truncate`, called by both
+//! spines). The third, the consumer-offset rewrite on the partition path
+//! (`consumer_group::maybe_rewrite_consumer_offset_request`, called by
+//! `dispatch::partition`), is synchronous.
+
+use crate::pat::{maybe_rewrite_pat_request, rewrite_pat_request_for_user};
+use crate::segment_cleaner::UNENFORCEABLE_TOPIC_SIZE_WARN;
+use crate::session_manager::SessionManager;
+use crate::shell::{ShellBus, ShellShard};
+use crate::users::maybe_rewrite_user_password_request;
+use crate::wire::request_body;
+use consensus::MetadataHandle;
+use iggy_binary_protocol::requests::partitions::{
+ CreatePartitionsRequest, DeletePartitionsRequest,
+};
+use iggy_binary_protocol::requests::streams::{CreateStreamRequest, UpdateStreamRequest};
+use iggy_binary_protocol::requests::topics::{CreateTopicRequest, UpdateTopicRequest};
+use iggy_binary_protocol::requests::users::{CreateUserRequest, UpdateUserRequest};
+use iggy_binary_protocol::{
+ MAX_PARTITIONS_PER_REQUEST, Operation, PrepareHeader, RoutedRequestHeader, WireDecode,
+ WireIdentifier, WireOptions,
+};
+use iggy_common::{
+ IggyByteSize, IggyError, MaxTopicSize, TopicCreateOptions, UPDATABLE_STREAM_OPTION_KEYS,
+ UPDATABLE_TOPIC_OPTION_KEYS, UPDATABLE_USER_OPTION_KEYS, validate_preallocated_topic_bytes,
+ validate_topic_segment_size,
+};
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use metadata::impls::metadata::StreamsFrontend;
+use metadata::stm::stream::Streams;
+use server_common::Message;
+use std::cell::RefCell;
+use std::rc::Rc;
+use tracing::warn;
+
+/// A staged pre-consensus rejection: `stage` labels the chain step for the
+/// deny log line, `error` is the typed code the deny reply carries.
+pub struct RewriteDeny {
Review Comment:
**nit** — the set of pre-consensus stages is split between this struct and a call-site literal. Three live in `stage` (`:110` `"personal-access-token"`, `:123` `"user-password"`, `:127` `"static-bounds"`); the fourth is a bare `"consumer-group"` at `dispatch/mod.rs:640`, fed to the same `context` parameter by the same `send_pre_consensus_deny`. So the type does not own the set it appears to define, and a fifth stage has two plausible homes. One enum (or one const set) that `mod.rs:640` must also name would close it. (`error` is already a typed `IggyError`, so this is about cohesion, not stringly-typed errors.)
##########
core/server/src/rewrite.rs:
##########
@@ -0,0 +1,534 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The two pre-consensus request-rewrite chains, side by side.
+//!
+//! [`tcp_chain`] serves the TCP funnel, [`http_chain`] the HTTP submit. Each
+//! step rewrites or validates a request BEFORE consensus, so a rejected
+//! request burns no replicated log entry and no plaintext secret enters
+//! consensus. The chains enter the PAT rewrite through different functions
+//! on purpose: TCP resolves the acting user from the transport
+//! `SessionManager` ([`maybe_rewrite_pat_request`]), HTTP authenticates
+//! against its own session table and passes the resolved `user_id`
+//! ([`rewrite_pat_request_for_user`]).
+//!
+//! Three more links complete the chains but live in their spines, because
+//! they fire partition-read mesh RPCs or are plane-specific. Two are async:
+//! the consumer-group Join/Leave enrichment ([`crate::consumer_group`], the
+//! TCP funnel calls it after [`tcp_chain`]) and the `DeleteSegments` ->
+//! `TruncatePartition` resolution
+//! (`dispatch::partition::resolve_delete_segments_truncate`, called by both
+//! spines). The third, the consumer-offset rewrite on the partition path
+//! (`consumer_group::maybe_rewrite_consumer_offset_request`, called by
+//! `dispatch::partition`), is synchronous.
+
+use crate::pat::{maybe_rewrite_pat_request, rewrite_pat_request_for_user};
+use crate::segment_cleaner::UNENFORCEABLE_TOPIC_SIZE_WARN;
+use crate::session_manager::SessionManager;
+use crate::shell::{ShellBus, ShellShard};
+use crate::users::maybe_rewrite_user_password_request;
+use crate::wire::request_body;
+use consensus::MetadataHandle;
+use iggy_binary_protocol::requests::partitions::{
+ CreatePartitionsRequest, DeletePartitionsRequest,
+};
+use iggy_binary_protocol::requests::streams::{CreateStreamRequest, UpdateStreamRequest};
+use iggy_binary_protocol::requests::topics::{CreateTopicRequest, UpdateTopicRequest};
+use iggy_binary_protocol::requests::users::{CreateUserRequest, UpdateUserRequest};
+use iggy_binary_protocol::{
+ MAX_PARTITIONS_PER_REQUEST, Operation, PrepareHeader, RoutedRequestHeader, WireDecode,
+ WireIdentifier, WireOptions,
+};
+use iggy_common::{
+ IggyByteSize, IggyError, MaxTopicSize, TopicCreateOptions, UPDATABLE_STREAM_OPTION_KEYS,
+ UPDATABLE_TOPIC_OPTION_KEYS, UPDATABLE_USER_OPTION_KEYS, validate_preallocated_topic_bytes,
+ validate_topic_segment_size,
+};
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use metadata::impls::metadata::StreamsFrontend;
+use metadata::stm::stream::Streams;
+use server_common::Message;
+use std::cell::RefCell;
+use std::rc::Rc;
+use tracing::warn;
+
+/// A staged pre-consensus rejection: `stage` labels the chain step for the
+/// deny log line, `error` is the typed code the deny reply carries.
+pub struct RewriteDeny {
+ pub stage: &'static str,
+ pub error: IggyError,
+}
+
+/// The TCP funnel's pre-consensus rewrite chain, in order: the PAT rewrite
+/// (resolving the acting user from the transport `sessions` binding), the
+/// password rewrite, then the static bounds gate. Mirrors [`http_chain`]
+/// plus that bounds step, which the binary wire needs because it has no
+/// `command.validate()` layer. Returns the rewritten request and the raw
+/// PAT token the funnel substitutes into the committed reply; a rejection
+/// names the failing stage for the deny log line.
+pub fn tcp_chain<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ max_tokens_per_user: u32,
+ request: Message<RoutedRequestHeader>,
+) -> Result<(Message<RoutedRequestHeader>, Option<String>), RewriteDeny>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let (request, raw_pat_token) = maybe_rewrite_pat_request(
+ sessions,
+ transport_client_id,
+ max_tokens_per_user,
+ |user_id| {
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .users()
+ .read(|users| users.pat_count_of(user_id))
+ },
+ request,
+ )
+ // Token cap reached, malformed body, or a lost session binding.
+ .map_err(|error| RewriteDeny {
+ stage: "personal-access-token",
+ error,
+ })?;
+ // Hash raw passwords and, for ChangePassword, verify the current password
+ // on the primary before replication; see `crate::users`. Replicas store the
+ // hash directly. A wrong current password is not denied here: it rides
+ // consensus and applies as a committed InvalidCredentials no-op, so the only
+ // Err returned is a malformed body.
+ let request =
+ maybe_rewrite_user_password_request(shard, request).map_err(|error| RewriteDeny {
+ stage: "user-password",
+ error,
+ })?;
+ static_bounds(shard, &request).map_err(|error| RewriteDeny {
+ stage: "static-bounds",
+ error,
+ })?;
+ Ok((request, raw_pat_token))
+}
+
+/// The HTTP submit's pre-consensus rewrite chain: the PAT rewrite for the
+/// already-authenticated `user_id`, then the password rewrite. Mirrors
+/// [`tcp_chain`] minus the session lookup (the HTTP listener authenticates
+/// against its own session table and resolves the acting user itself) and
+/// minus the static bounds step: HTTP enforces the same bounds in its
Review Comment:
**nit** — this sentence newly asserts cross-transport bounds parity, and the two caps it rests on are defined twice with no cross-check: `MAX_PARTITIONS_COUNT = 1000` (`common/src/lib.rs:162`, used only by the HTTP DTO validators) and `MAX_PARTITIONS_PER_REQUEST = 1000` (`binary_protocol/src/lib.rs:107`, used only on the binary path at `:182`). Equivalent today, so no live divergence — but the parity is now documented and nothing enforces it. This repo's own idiom for exactly this drift class is a `const _: () = assert!(..)` (see `boot/recovery.rs:301-350`).
##########
core/integration/tests/server/raw_tcp.rs:
##########
@@ -0,0 +1,192 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Raw TCP framing for the server suites that hand-craft client frames the
+//! SDK cannot emit: connect to the harness server, write one request frame,
+//! read the header the server answers with, and register root so a frame
+//! can ride a bound session.
+
+use std::mem::offset_of;
+use std::time::Duration;
+
+use iggy::prelude::*;
+use iggy_binary_protocol::codec::{WireDecode, WireEncode};
+use iggy_binary_protocol::consensus::{
+ Command, Operation, ReplyHeader, RequestHeader, read_size_field, result_code,
+ result_section_len,
+};
+use iggy_binary_protocol::requests::users::LoginRegisterRequest;
+use iggy_binary_protocol::responses::users::LoginRegisterResponse;
+use iggy_binary_protocol::{
+ ClientVersionInfo, EvictionHeader, HEADER_SIZE, IGGY_PROTOCOL_VERSION, WireName,
+};
+use integration::harness::TestHarness;
+use secrecy::SecretString;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::time::{Instant, sleep, timeout};
+
+/// Per-frame reply wait. A server that drops the frame answers nothing at
+/// all, so an unanswered read is a verdict, not a reason to wait longer.
+const REPLY_WAIT: Duration = Duration::from_secs(5);
+
+/// Budget for the register to commit: right after boot the single node may
+/// still be electing itself and answers transient rejections meanwhile.
+const COMMIT_BUDGET: Duration = Duration::from_secs(15);
+
+const RETRY_PAUSE: Duration = Duration::from_millis(100);
+
+pub(crate) async fn connect(harness: &TestHarness) -> TcpStream {
+ let addr = harness
+ .server()
+ .tcp_addr()
+ .expect("server must expose a TCP address");
+ TcpStream::connect(addr).await.unwrap()
+}
+
+pub(crate) fn request_header(
+ operation: Operation,
+ client: u128,
+ session: u64,
+ request: u64,
+ body_len: usize,
+) -> RequestHeader {
+ RequestHeader {
+ command: Command::Request,
+ operation,
+ size: u32::try_from(HEADER_SIZE + body_len).unwrap(),
+ client,
+ session,
+ request,
+ ..Default::default()
+ }
+}
+
+/// A header-only `NonReplicated` frame; the command code travels in the
+/// first 4 reserved bytes.
+pub(crate) fn non_replicated_header(
+ client: u128,
+ session: u64,
+ request: u64,
+ code: u32,
+) -> RequestHeader {
+ let mut header = request_header(Operation::NonReplicated, client, session, request, 0);
+ header.reserved[..4].copy_from_slice(&code.to_le_bytes());
+ header
+}
+
+pub(crate) async fn write_frame(stream: &mut TcpStream, header: &RequestHeader, body: &[u8]) {
+ stream.write_all(bytemuck::bytes_of(header)).await.unwrap();
+ if !body.is_empty() {
+ stream.write_all(body).await.unwrap();
+ }
+}
+
+/// Read the header of the next server frame, within [`REPLY_WAIT`].
+pub(crate) async fn read_frame_header(stream: &mut TcpStream) -> [u8; HEADER_SIZE] {
+ let mut header = [0u8; HEADER_SIZE];
+ timeout(REPLY_WAIT, stream.read_exact(&mut header))
+ .await
+ .expect("server must answer within the reply wait, not stall")
+ .expect("reply header read failed");
+ header
+}
+
+/// Write one frame and read one Reply off the lockstep connection, the body
+/// sized by the reply's size field.
+pub(crate) async fn exchange(
+ stream: &mut TcpStream,
+ header: &RequestHeader,
+ body: &[u8],
+) -> ([u8; HEADER_SIZE], Vec<u8>) {
+ write_frame(stream, header, body).await;
+ let reply_header = read_frame_header(stream).await;
+ let command = frame_command(&reply_header);
+ assert_eq!(
+ command,
+ Command::Reply as u8,
+ "expected a Reply frame, got command byte {command} (an Eviction carries reason {})",
+ reply_header[offset_of!(EvictionHeader, reason)]
+ );
+
+ let total_size = read_size_field(&reply_header).expect("reply size field") as usize;
+ let mut reply_body = vec![0u8; total_size - HEADER_SIZE];
Review Comment:
**warning** — `total_size - HEADER_SIZE` has no floor, so a reply whose size field is under `HEADER_SIZE` fails in a way that hides the regression under test. `read_size_field` (`binary_protocol/src/consensus/header.rs:73-78`) is a bare 4-byte LE read with no `>= HEADER_SIZE` guard — the SDK adds its own at `sdk/src/vsr.rs:181-183`, which is why this helper cannot inherit it. Debug builds panic "attempt to subtract with overflow"; release test runs have no `overflow-checks` (`[profile.release]` sets only `lto` and `codegen-units`), so it wraps to a ~1.8e19 `vec![0u8; n]` and aborts on allocation failure. Assert `total_size >= HEADER_SIZE` with a message naming the frame.
##########
core/integration/tests/server/legacy_login_vsr.rs:
##########
@@ -60,39 +59,20 @@ async fn given_legacy_pat_login_code_when_sent_raw_should_evict_malformed_login(
/// eviction. The reject runs before the session gate, so this unbound socket
/// exercises the same path a bound connection would.
async fn assert_legacy_login_code_evicted(harness: &TestHarness, code: u32) {
- let mut header = RequestHeader {
- command: Command::Request,
- operation: Operation::NonReplicated,
- size: u32::try_from(HEADER_SIZE).unwrap(),
- // NonReplicated leaves session / request unchecked, but the header
- // validator still requires a nonzero client id.
- client: 0xC0FFEE,
- session: 0,
- request: 0,
- ..Default::default()
- };
- // A non-replicated command code travels in the first 4 reserved bytes.
- header.reserved[..4].copy_from_slice(&code.to_le_bytes());
+ // NonReplicated leaves session / request unchecked, but the header
+ // validator still requires a nonzero client id.
+ let header = non_replicated_header(0xC0FFEE, 0, 0, code);
- let addr = harness
- .server()
- .tcp_addr()
- .expect("server must expose a TCP address");
- let mut stream = TcpStream::connect(addr).await.unwrap();
- stream.write_all(bytemuck::bytes_of(&header)).await.unwrap();
+ let mut stream = connect(harness).await;
+ write_frame(&mut stream, &header, &[]).await;
- // Eviction is header-only: exactly 256 bytes. The timeout makes the
+ // Eviction is header-only: exactly 256 bytes. The bounded read makes the
// fail-fast contract explicit -- a regression that silently drops the
// frame trips this instead of hanging until the test wall clock.
- let mut reply = [0u8; HEADER_SIZE];
- timeout(Duration::from_secs(5), stream.read_exact(&mut reply))
- .await
- .expect("server must answer a legacy login code within 5s, not stall")
- .expect("reading the eviction frame must succeed");
+ let reply = read_frame_header(&mut stream).await;
- let command_offset = offset_of!(RequestHeader, command);
assert_eq!(
- reply[command_offset],
+ frame_command(&reply),
Review Comment:
**nit** — this file still reads the reason byte as `reply[HEADER_SIZE - 1]` against a hardcoded `EVICTION_REASON_MALFORMED_LOGIN: u8 = 15`, while the new harness next door reads the same field as `offset_of!(EvictionHeader, reason)` (`raw_tcp.rs:123`). Both are correct today — `header.rs:755` statically asserts `reason` is the last header byte — but renumber `EvictionReason` and the hardcoded `15` silently checks a different reason while still passing. Export an `eviction_reason()` accessor from `raw_tcp` and compare against `EvictionReason::MalformedLogin as u8`.
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4036:
URL: https://github.com/apache/iggy/pull/4036#discussion_r3926680721
##########
core/server/src/dispatch/partition.rs:
##########
@@ -509,11 +511,13 @@ pub(in crate::dispatch) async fn handle_poll_messages<B, MJ, S, SB>(
{
let Ok(wire) = PollMessagesRequest::decode_from(request_body(request)) else {
// Undecodable poll: keep the fail-fast empty-poll shape.
+ let (body, channel) = empty_poll_fallback(0);
Review Comment:
**warning** — this is the last fail-open of exactly the class the PR set out to close, and it is on the hottest read path. An undecodable `POLL_MESSAGES` body answers status 0 plus a *valid* 16-byte `PolledMessages` body, which every SDK decodes as a successful 0-message poll, so a version-skewed consumer loops forever with no error — strictly worse than a wedge, which at least surfaces a timeout. Every sibling undecodable-body path in this same diff denies typed (`rewrite.rs` `static_bounds` -> `InvalidCommand`; `mod.rs:640`, whose commit is titled "deny malformed consumer-group requests instead of dropping").
No SDK breaks on the typed deny, and it is not a new wire shape: the server already sends nonzero-status `POLL_MESSAGES` replies at `:539` (authz) and `:566-577` (`PartitionNotFound`/`StreamIdNotFound`/`TopicIdNotFound`), pinned end-to-end in `integration/tests/server/poll_semantics_vsr.rs:70-73,113,145,188-192`. All five native SDKs peek `status` before any body decode — Rust `sdk/src/vsr.rs:204,234`, Go `reply.go:81`, Node `reply.ts:79`, Java `VsrResponseHandler.java:204`, .NET `VsrReplyDecoder.cs:60` — and Python/C++/PHP are FFI over Rust.
The rustdoc at `:496-497` ("empty body so the SDK fails fast on decode") argues for the current shape but contradicts its own function: `:578-580` explains the body is a valid 16-byte poll precisely so the decoder does *not* fail. And `failure.rs:562-593` never enters `handle_poll_messages`, so this change leaves that snapshot byte-identical.
Fix: `send_non_replicated_deny(shard, request, transport_client_id, IggyError::InvalidCommand.as_code())`. Scope it to this undecodable-body branch only — leave the `ConsumerGroupPartitionNotOwned` resync sentinel at `:585-593` alone.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -642,6 +695,7 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
request,
transport_client_id,
Bytes::new(),
+ FrameChannel::Reply,
Review Comment:
**nit** — same shape as the undecodable poll: a malformed request body gets a success-shaped status-0 empty frame, which the SDK reads as "no stored offset". It is also labelled `FrameChannel::Reply` with the identical `"get_consumer_offset"` context as the legitimate no-offset success at `:757`, so the two are indistinguishable in the send-failure log by every field. Deny typed here too, or at minimum give this branch its own `context`.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -154,17 +158,14 @@ pub(in crate::dispatch) async fn handle_non_replicated_request<B, MJ, S, SB>(
request.header().session,
commit,
);
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- "failed to send non-replicated ping reply"
- );
- }
+ send_host_frame(
Review Comment:
**nit** — `record_heartbeat` was already called unconditionally in the funnel (`dispatch/mod.rs:457`) before `classify` routed here, so the PING arm's own call at `:153` is redundant: two `Instant::now()` and two `get_mut` per ping. The funnel call strictly covers it. Pre-existing, not introduced here.
##########
core/server/src/dispatch/reads.rs:
##########
@@ -324,18 +333,14 @@ async fn handle_default_non_replicated<B, MJ, S, SB>(
request.header().session,
commit,
);
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- code,
- error = %error,
- "failed to send non-replicated VSR reply"
- );
- }
+ send_host_frame(
Review Comment:
**warning** — the gate-deny branch at `:308-311` emits no log at all, while this builder-`Err` branch 25 lines below `warn!`s with `code` and `error`. `send_non_replicated_deny` only logs on *send* failure, so a successful refusal leaves zero server-side breadcrumb — and the refusal this PR introduces (every armless or unknown non-replicated code, via `authorize_default_read`'s new tail) is exactly the one an operator would need to see. Add a `warn!` with `code` + `error` on the gate-deny branch.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -615,45 +457,34 @@ async fn handle_client_request<B, MJ, S, SB>(
sessions.borrow_mut().record_heartbeat(transport_client_id);
let header = *request.header();
- if header.operation == Operation::NonReplicated {
- // Auth bypass guard: `PING`, the liveness probe, is the only pre-auth
- // code, on every roster shape. `GET_CLUSTER_METADATA` describes the
- // private replica network and is not something an unauthenticated
- // caller gets to read; a client that dialed a backup no longer needs
- // it to find the leader, because the backup authenticates the login
- // locally and forwards only the consensus proposal
- // (`submit_register_local_or_forward`). Every other non-replicated
- // code MUST go through Register first, which binds the acting user
- // the per-op authz gates resolve.
- let nr_code = u32::from_le_bytes(request.header().reserved[..4].try_into().unwrap());
- // Legacy (pre-register) login codes. The server authenticates only via
- // the Register handshake (LOGIN_REGISTER / LOGIN_REGISTER_WITH_PAT,
- // Operation::Register); the vsr SDK funnels both logins there and never
- // emits these. Reject them uniformly with a typed MalformedLogin (the
- // SDK maps it to InvalidFormat) before the session gate, so a legacy or
- // foreign client fails fast instead of getting the generic
- // Unauthenticated deny the pre-auth guard would send unbound, or the
- // silent empty-ok Reply the bound non-replicated path would send.
- if matches!(
- nr_code,
- LOGIN_USER_CODE | LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE
- ) {
+ let bound = sessions.borrow().get_session(transport_client_id);
+ match classify(&header, bound.is_some()) {
+ RequestClass::LegacyLogin => {
+ // Legacy (pre-register) login codes. The server authenticates only via
+ // the Register handshake (LOGIN_REGISTER / LOGIN_REGISTER_WITH_PAT,
+ // Operation::Register); the vsr SDK funnels both logins there and never
+ // emits these. Reject them uniformly with a typed MalformedLogin (the
+ // SDK maps it to InvalidFormat) before the session gate, so a legacy or
+ // foreign client fails fast instead of getting the generic
+ // Unauthenticated deny the pre-auth guard would send unbound, or the
+ // silent empty-ok Reply the bound non-replicated path would send.
+ let nr_code = non_replicated_code(&header);
warn!(
transport_client_id,
code = nr_code,
"rejecting legacy login code; server requires the register handshake"
);
- send_login_eviction(
+ send_eviction(
shard,
transport_client_id,
header.client,
EvictionReason::MalformedLogin,
+ "legacy login rejection",
)
.await;
- return;
}
- let allowed_pre_auth = nr_code == PING_CODE;
- if !allowed_pre_auth && sessions.borrow().get_session(transport_client_id).is_none() {
+ RequestClass::UnauthenticatedRead => {
+ let nr_code = non_replicated_code(&header);
Review Comment:
**warning** — the rationale just below (`:488-490`, "Foreign SDKs still probe `GET_CLUSTER_METADATA` before login until they are fixed") is stale for every SDK in tree, so a genuine unauthenticated read of the private replica topology now hides at `debug` (`:491-495`).
All seven read the roster only *after* login: Node `client.socket.ts:773` (comment says "deliberately absent: the server auth-gates it"), Go `internal/util/leader_aware.go:52-88`, .NET `TcpMessageStream.Vsr.cs:489`, Java `AsyncIggyTcpClient.java:854-856` (`loginAndSettleOnLeader` composes the roster read onto the *completed* login future), and Python/C++/PHP inherit Rust. Java's poll is the most aggressive of the seven — `LEADERLESS_WAIT_BUDGET = 5s` at `LEADERLESS_POLL_INTERVAL = 250ms` (`LeaderAwareness.java:60-62`), up to 20 reads per login — but all 20 are authenticated, and this arm fires only for an unbound transport, so dropping the special case cannot cause a warn flood.
Fix: drop the special case and let it `warn!` like every other pre-auth denial.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -681,353 +512,179 @@ async fn handle_client_request<B, MJ, S, SB>(
IggyError::Unauthenticated.as_code(),
)
.await;
- return;
}
- handle_non_replicated_request(shard, sessions, system_config, transport_client_id, request)
- .await;
- return;
- }
-
- if header.operation == Operation::Register && header.session == 0 && header.request == 0 {
- handle_login_register_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- if header.operation == Operation::Logout {
- handle_logout_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- let bound = sessions.borrow().get_session(transport_client_id);
- if bound.is_none() {
- // Replicated request on an unbound transport. Without this short-
- // circuit, the rewrite below overwrites `header.client` with
- // `transport_client_id` and dispatches; the request_preflight then
- // rejects with `NoSession`/`Fenced` and the failure disappears
- // silently, wedging the SDK until the socket timeout. A typed
- // `Eviction(NoSession)` is right here, unlike the pre-auth read
- // guard above: a replicated request implies the client believes it
- // has a session, and that session is gone, so it must register
- // again. An empty status-0 Reply is not safe here, because
- // SendMessages is the one replicated operation without a result
- // section, and its decoder would read the empty body as a
- // successful send.
- warn!(
- transport_client_id,
- operation = ?header.operation,
- "rejecting replicated request from unbound transport with Eviction(NoSession)"
- );
- send_unauthenticated_eviction(shard, transport_client_id).await;
- return;
- }
-
- // DeleteSegments is neither a partition nor a metadata consensus op: the
- // owning shard resolves the requested count to a concrete offset, then a
- // `TruncatePartition` is replicated through metadata (Option A). Each
- // replica's reconciler trims to the committed watermark. Handle it here,
- // ahead of the partition/metadata routing below.
- if header.operation == Operation::DeleteSegments {
- handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
- return;
- }
-
- if header.operation.is_partition() {
- // `bound` is Some here (unbound transports returned above).
- let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
- // `get_session` discards the acting user id the partition gate needs;
- // resolve it from the same bound connection. A bound transport always
- // has one, but the gate fails closed on `None` rather than trust that.
- let acting_user_id = sessions.borrow().get_user_id(transport_client_id);
- dispatch_partition_request(
- shard,
- request,
- vsr_client_id,
- bound_session,
- transport_client_id,
- acting_user_id,
- )
- .await;
- return;
- }
-
- let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
- *new_header = header;
- // Metadata-plane ops route by operation: stamp the sentinel group.
- new_header.group = server_common::sharding::METADATA_GROUP;
- // `bound` is always Some here (unbound transports early-return above);
- // this sets the consensus client id + session for the replicated op.
- if let Some((bound_client_id, bound_session)) = bound {
- new_header.client = bound_client_id;
- new_header.session = bound_session;
- }
- });
- let (request, raw_pat_token) = match maybe_rewrite_pat_request(
- sessions,
- transport_client_id,
- max_tokens_per_user,
- |user_id| {
- shard
- .plane
- .metadata()
- .mux_stm
- .users()
- .read(|users| users.pat_count_of(user_id))
- },
- request,
- ) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Token cap reached, malformed body, or a lost session binding.
- send_pre_consensus_deny(
+ RequestClass::NonReplicatedRead => {
+ // The auth-bypass guard is `classify`'s `UnauthenticatedRead` class:
+ // `PING`, the liveness probe, is the only pre-auth code, on every
+ // roster shape. `GET_CLUSTER_METADATA` describes the private replica
+ // network and is not something an unauthenticated caller gets to
+ // read; a client that dialed a backup no longer needs it to find the
+ // leader, because the backup authenticates the login locally and
+ // forwards only the consensus proposal
+ // (`submit_register_local_or_forward`). Every other non-replicated
+ // code MUST go through Register first, which binds the acting user
+ // the per-op authz gates resolve.
+ handle_non_replicated_request(
shard,
- &header,
+ sessions,
+ system_config,
transport_client_id,
- &error,
- "personal-access-token",
+ request,
)
.await;
- return;
}
- };
- // Hash raw passwords and, for ChangePassword, verify the current password
- // on the primary before replication; see `crate::users`. Replicas store the
- // hash directly. A wrong current password is not denied here: it rides
- // consensus and applies as a committed InvalidCredentials no-op, so the only
- // Err returned is a malformed body.
- let request = match maybe_rewrite_user_password_request(shard, request) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Malformed body: deny fast with InvalidCommand.
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "user-password")
- .await;
- return;
+ RequestClass::LoginRegister => {
+ handle_login_register_request(shard, sessions, transport_client_id, request).await;
}
- };
- // Static bounds run pre-consensus so a rejected request burns no
- // replicated log entry; HTTP covers the same bounds via
- // `command.validate()`. A body that fails to decode denies typed too
- // (`InvalidCommand`), instead of riding consensus just to fail there.
- let bounds = match header.operation {
- Operation::CreateTopic => CreateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_topic| {
- // `parse` doubles as the catalog gate: an unknown key or a
- // malformed value denies typed here, pre-consensus.
- let options = TopicCreateOptions::parse(&create_topic.options)?;
- if let Some(segment_size) = options.segment_size {
- validate_topic_segment_size(
- segment_size.as_bytes_u64(),
- iggy_common::MAX_TOPIC_SEGMENT_SIZE,
- )?;
- }
- let segment_size = options.segment_size.map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- if options
- .preallocate_segments
- .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS)
- {
- validate_preallocated_topic_bytes(segment_size, create_topic.partitions_count)?;
- }
- let max_topic_size = options
- .max_topic_size
- .unwrap_or(MaxTopicSize::ServerDefault);
- validate_topic_bounds(create_topic.partitions_count, max_topic_size, segment_size)?;
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- create_topic.partitions_count,
- );
- Ok(())
- }),
- Operation::CreatePartitions => CreatePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_partitions| {
- validate_partitions_change_count(create_partitions.partitions_count)?;
- let metadata = shard.plane.metadata();
- warn_unenforceable_topic_size_on_partition_add(
- metadata.mux_stm.streams(),
- &create_partitions.stream_id,
- &create_partitions.topic_id,
- shard.bus_max_message_size(),
- create_partitions.partitions_count,
- );
- Ok(())
- }),
- Operation::DeletePartitions => DeletePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|delete_partitions| {
- validate_partitions_change_count(delete_partitions.partitions_count)
- }),
- // Only the updatable subset: the create-time knobs are pushed to
- // partitions when the topic is built and nothing re-pushes them, so
- // accepting one here would store a value no partition ever sees.
- Operation::UpdateTopic => UpdateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_topic| {
- validate_option_keys(&update_topic.options, UPDATABLE_TOPIC_OPTION_KEYS)?;
- let options = TopicCreateOptions::parse(&update_topic.options)?;
- let Some(max_topic_size) = options.max_topic_size else {
- return Ok(());
- };
- // An update can lower the cap below one segment just as a
- // create can, and the stored map would then report a size the
- // topic can never enforce. The floor is this topic's own
- // segment size, since that key is create-only.
- let metadata = shard.plane.metadata();
- let streams = metadata.mux_stm.streams();
- let segment_size = streams
- .topic_segment_size(&update_topic.stream_id, &update_topic.topic_id)
- .map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- validate_topic_size_floor(max_topic_size, segment_size)?;
- let partitions_count = streams
- .topic_partitions_count(&update_topic.stream_id, &update_topic.topic_id)
- .unwrap_or(0);
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- u32::try_from(partitions_count).unwrap_or(u32::MAX),
- );
- Ok(())
- }),
- Operation::UpdateStream => UpdateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_stream| {
- validate_option_keys(&update_stream.options, UPDATABLE_STREAM_OPTION_KEYS)
- }),
- Operation::UpdateUser => UpdateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_user| {
- validate_option_keys(&update_user.options, UPDATABLE_USER_OPTION_KEYS)
- }),
- Operation::CreateStream => CreateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_stream| validate_option_keys(&create_stream.options, &[])),
- Operation::CreateUser => CreateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_user| validate_option_keys(&create_user.options, &[])),
- _ => Ok(()),
- };
- if let Err(error) = bounds {
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "static-bounds").await;
- return;
- }
- // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
- // partition count for Join) before replication; see `crate::consumer_group`.
- let request = match maybe_rewrite_consumer_group_request(shard, request).await {
- Ok(rewritten) => rewritten,
- Err(error) => {
+ RequestClass::Logout => {
+ handle_logout_request(shard, sessions, transport_client_id, request).await;
+ }
+ RequestClass::UnboundReplicated => {
+ // Replicated request on an unbound transport. Without this short-
+ // circuit, the rewrite below overwrites `header.client` with
+ // `transport_client_id` and dispatches; the request_preflight then
+ // rejects with `NoSession`/`Fenced` and the failure disappears
+ // silently, wedging the SDK until the socket timeout. A typed
+ // `Eviction(NoSession)` is right here, unlike the plain deny of
+ // `UnauthenticatedRead`: a replicated request implies the client
+ // believes it has a session, and that session is gone, so it must
+ // register again. An empty status-0 Reply is not safe here, because
+ // SendMessages is the one replicated operation without a result
+ // section, and its decoder would read the empty body as a
+ // successful send.
warn!(
transport_client_id,
- error = %error,
operation = ?header.operation,
- "dropping consumer-group request with invalid payload"
+ "rejecting replicated request from unbound transport with Eviction(NoSession)"
);
- return;
+ // The eviction context is best-effort off the metadata consensus
+ // (peer shards have none; zeroes are cosmetic -- the SDK only
+ // reads the reason), and the evicted id is the transport id: no
+ // VSR session exists to name.
+ send_eviction(
+ shard,
+ transport_client_id,
+ transport_client_id,
+ EvictionReason::NoSession,
+ "unbound replicated request",
+ )
+ .await;
}
- };
- let request_header = *request.header();
- // Replicated request: run consensus on the metadata owner (shard 0) and
- // bring the committed reply back here. This shard owns the connection,
- // so it writes the reply to the socket via the transport client id --
- // shard 0 can't route by the consensus client id (no home-shard bits).
- match submit_client_request_on_owner(shard, request).await {
- Some(reply) => {
- // The raw PAT token never enters consensus (it is non-deterministic
- // and secret), so the committed reply body is empty. Substitute the
- // raw-token response here, on the minting client's home shard, using
- // the confirmed commit position from the committed reply.
- let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) {
- Ok(reply) => reply,
+ RequestClass::DeleteSegments => {
+ // DeleteSegments is neither a partition nor a metadata consensus op: the
+ // owning shard resolves the requested count to a concrete offset, then a
+ // `TruncatePartition` is replicated through metadata (Option A). Each
+ // replica's reconciler trims to the committed watermark. `classify`
+ // names it ahead of the partition and metadata classes.
+ handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
+ }
+ RequestClass::Partition => {
+ // `bound` is Some here: `classify` sends unbound transports to
+ // `UnboundReplicated`.
+ let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
+ // `get_session` discards the acting user id the partition gate needs;
+ // resolve it from the same bound connection. A bound transport always
+ // has one, but the gate fails closed on `None` rather than trust that.
+ let acting_user_id = sessions.borrow().get_user_id(transport_client_id);
+ dispatch_partition_request(
+ shard,
+ request,
+ vsr_client_id,
+ bound_session,
+ transport_client_id,
+ acting_user_id,
+ )
+ .await;
+ }
+ RequestClass::ReplicatedMetadata => {
+ let request =
+ request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
+ *new_header = header;
+ // Metadata-plane ops route by operation: stamp the sentinel group.
+ new_header.group = server_common::sharding::METADATA_GROUP;
+ // `bound` is always Some here (`classify` sends unbound transports to
+ // `UnboundReplicated`); this sets the consensus client id + session
+ // for the replicated op.
+ if let Some((bound_client_id, bound_session)) = bound {
+ new_header.client = bound_client_id;
+ new_header.session = bound_session;
+ }
+ });
+ let (request, raw_pat_token) = match tcp_chain(
+ shard,
+ sessions,
+ transport_client_id,
+ max_tokens_per_user,
+ request,
+ ) {
+ Ok(rewritten) => rewritten,
+ Err(RewriteDeny { stage, error }) => {
+ send_pre_consensus_deny(shard, transport_client_id, &header, &error, stage)
+ .await;
+ return;
+ }
+ };
+ // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
+ // partition count for Join) before replication; see `crate::consumer_group`.
+ let request = match maybe_rewrite_consumer_group_request(shard, request).await {
+ Ok(rewritten) => rewritten,
Err(error) => {
- warn!(
+ // The rewrite only ever fails on an undecodable body, so a
+ // replay cannot help: deny typed instead of leaving the
+ // lockstep connection to its read timeout.
+ send_pre_consensus_deny(
+ shard,
transport_client_id,
- error = %error,
- "failed to build raw PAT reply"
- );
+ &header,
+ &error,
+ "consumer-group",
+ )
+ .await;
return;
}
};
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- operation = ?header.operation,
- "failed to deliver committed reply to client"
- );
+ let request_header = *request.header();
+ // Replicated request: run consensus on the metadata owner (shard 0) and
+ // bring the committed reply back here. This shard owns the connection,
+ // so it writes the reply to the socket via the transport client id --
+ // shard 0 can't route by the consensus client id (no home-shard bits).
+ match submit_client_request_on_owner(shard, request).await {
+ Some(reply) => {
+ // The raw PAT token never enters consensus (it is non-deterministic
+ // and secret), so the committed reply body is empty. Substitute the
+ // raw-token response here, on the minting client's home shard, using
+ // the confirmed commit position from the committed reply.
+ let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) {
+ Ok(reply) => reply,
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to build raw PAT reply"
Review Comment:
**nit** — this `Err` returns with no frame at all, *after* the metadata op already committed. The lockstep connection then wedges to its read timeout on an op that succeeded server-side, and `failure.rs:31` declares exactly two deliberate silences, of which this is not one.
Both `Err` arms look practically unreachable (`responses.rs:1566` already gates `command == Command::Reply` on a server-built frame, and `WireName::new(raw)` caps at 255 bytes against the fixed short token from `pat.rs:131`), so this is cheap either way: it is reachable only for a PAT mint, an admin-frequency path. Send the typed deny and add the row.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -681,353 +512,179 @@ async fn handle_client_request<B, MJ, S, SB>(
IggyError::Unauthenticated.as_code(),
)
.await;
- return;
}
- handle_non_replicated_request(shard, sessions, system_config, transport_client_id, request)
- .await;
- return;
- }
-
- if header.operation == Operation::Register && header.session == 0 && header.request == 0 {
- handle_login_register_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- if header.operation == Operation::Logout {
- handle_logout_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- let bound = sessions.borrow().get_session(transport_client_id);
- if bound.is_none() {
- // Replicated request on an unbound transport. Without this short-
- // circuit, the rewrite below overwrites `header.client` with
- // `transport_client_id` and dispatches; the request_preflight then
- // rejects with `NoSession`/`Fenced` and the failure disappears
- // silently, wedging the SDK until the socket timeout. A typed
- // `Eviction(NoSession)` is right here, unlike the pre-auth read
- // guard above: a replicated request implies the client believes it
- // has a session, and that session is gone, so it must register
- // again. An empty status-0 Reply is not safe here, because
- // SendMessages is the one replicated operation without a result
- // section, and its decoder would read the empty body as a
- // successful send.
- warn!(
- transport_client_id,
- operation = ?header.operation,
- "rejecting replicated request from unbound transport with Eviction(NoSession)"
- );
- send_unauthenticated_eviction(shard, transport_client_id).await;
- return;
- }
-
- // DeleteSegments is neither a partition nor a metadata consensus op: the
- // owning shard resolves the requested count to a concrete offset, then a
- // `TruncatePartition` is replicated through metadata (Option A). Each
- // replica's reconciler trims to the committed watermark. Handle it here,
- // ahead of the partition/metadata routing below.
- if header.operation == Operation::DeleteSegments {
- handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
- return;
- }
-
- if header.operation.is_partition() {
- // `bound` is Some here (unbound transports returned above).
- let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
- // `get_session` discards the acting user id the partition gate needs;
- // resolve it from the same bound connection. A bound transport always
- // has one, but the gate fails closed on `None` rather than trust that.
- let acting_user_id = sessions.borrow().get_user_id(transport_client_id);
- dispatch_partition_request(
- shard,
- request,
- vsr_client_id,
- bound_session,
- transport_client_id,
- acting_user_id,
- )
- .await;
- return;
- }
-
- let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
- *new_header = header;
- // Metadata-plane ops route by operation: stamp the sentinel group.
- new_header.group = server_common::sharding::METADATA_GROUP;
- // `bound` is always Some here (unbound transports early-return above);
- // this sets the consensus client id + session for the replicated op.
- if let Some((bound_client_id, bound_session)) = bound {
- new_header.client = bound_client_id;
- new_header.session = bound_session;
- }
- });
- let (request, raw_pat_token) = match maybe_rewrite_pat_request(
- sessions,
- transport_client_id,
- max_tokens_per_user,
- |user_id| {
- shard
- .plane
- .metadata()
- .mux_stm
- .users()
- .read(|users| users.pat_count_of(user_id))
- },
- request,
- ) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Token cap reached, malformed body, or a lost session binding.
- send_pre_consensus_deny(
+ RequestClass::NonReplicatedRead => {
+ // The auth-bypass guard is `classify`'s `UnauthenticatedRead` class:
+ // `PING`, the liveness probe, is the only pre-auth code, on every
+ // roster shape. `GET_CLUSTER_METADATA` describes the private replica
+ // network and is not something an unauthenticated caller gets to
+ // read; a client that dialed a backup no longer needs it to find the
+ // leader, because the backup authenticates the login locally and
+ // forwards only the consensus proposal
+ // (`submit_register_local_or_forward`). Every other non-replicated
+ // code MUST go through Register first, which binds the acting user
+ // the per-op authz gates resolve.
+ handle_non_replicated_request(
shard,
- &header,
+ sessions,
+ system_config,
transport_client_id,
- &error,
- "personal-access-token",
+ request,
)
.await;
- return;
}
- };
- // Hash raw passwords and, for ChangePassword, verify the current password
- // on the primary before replication; see `crate::users`. Replicas store the
- // hash directly. A wrong current password is not denied here: it rides
- // consensus and applies as a committed InvalidCredentials no-op, so the only
- // Err returned is a malformed body.
- let request = match maybe_rewrite_user_password_request(shard, request) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Malformed body: deny fast with InvalidCommand.
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "user-password")
- .await;
- return;
+ RequestClass::LoginRegister => {
+ handle_login_register_request(shard, sessions, transport_client_id, request).await;
}
- };
- // Static bounds run pre-consensus so a rejected request burns no
- // replicated log entry; HTTP covers the same bounds via
- // `command.validate()`. A body that fails to decode denies typed too
- // (`InvalidCommand`), instead of riding consensus just to fail there.
- let bounds = match header.operation {
- Operation::CreateTopic => CreateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_topic| {
- // `parse` doubles as the catalog gate: an unknown key or a
- // malformed value denies typed here, pre-consensus.
- let options = TopicCreateOptions::parse(&create_topic.options)?;
- if let Some(segment_size) = options.segment_size {
- validate_topic_segment_size(
- segment_size.as_bytes_u64(),
- iggy_common::MAX_TOPIC_SEGMENT_SIZE,
- )?;
- }
- let segment_size = options.segment_size.map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- if options
- .preallocate_segments
- .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS)
- {
- validate_preallocated_topic_bytes(segment_size, create_topic.partitions_count)?;
- }
- let max_topic_size = options
- .max_topic_size
- .unwrap_or(MaxTopicSize::ServerDefault);
- validate_topic_bounds(create_topic.partitions_count, max_topic_size, segment_size)?;
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- create_topic.partitions_count,
- );
- Ok(())
- }),
- Operation::CreatePartitions => CreatePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_partitions| {
- validate_partitions_change_count(create_partitions.partitions_count)?;
- let metadata = shard.plane.metadata();
- warn_unenforceable_topic_size_on_partition_add(
- metadata.mux_stm.streams(),
- &create_partitions.stream_id,
- &create_partitions.topic_id,
- shard.bus_max_message_size(),
- create_partitions.partitions_count,
- );
- Ok(())
- }),
- Operation::DeletePartitions => DeletePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|delete_partitions| {
- validate_partitions_change_count(delete_partitions.partitions_count)
- }),
- // Only the updatable subset: the create-time knobs are pushed to
- // partitions when the topic is built and nothing re-pushes them, so
- // accepting one here would store a value no partition ever sees.
- Operation::UpdateTopic => UpdateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_topic| {
- validate_option_keys(&update_topic.options, UPDATABLE_TOPIC_OPTION_KEYS)?;
- let options = TopicCreateOptions::parse(&update_topic.options)?;
- let Some(max_topic_size) = options.max_topic_size else {
- return Ok(());
- };
- // An update can lower the cap below one segment just as a
- // create can, and the stored map would then report a size the
- // topic can never enforce. The floor is this topic's own
- // segment size, since that key is create-only.
- let metadata = shard.plane.metadata();
- let streams = metadata.mux_stm.streams();
- let segment_size = streams
- .topic_segment_size(&update_topic.stream_id, &update_topic.topic_id)
- .map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- validate_topic_size_floor(max_topic_size, segment_size)?;
- let partitions_count = streams
- .topic_partitions_count(&update_topic.stream_id, &update_topic.topic_id)
- .unwrap_or(0);
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- u32::try_from(partitions_count).unwrap_or(u32::MAX),
- );
- Ok(())
- }),
- Operation::UpdateStream => UpdateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_stream| {
- validate_option_keys(&update_stream.options, UPDATABLE_STREAM_OPTION_KEYS)
- }),
- Operation::UpdateUser => UpdateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_user| {
- validate_option_keys(&update_user.options, UPDATABLE_USER_OPTION_KEYS)
- }),
- Operation::CreateStream => CreateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_stream| validate_option_keys(&create_stream.options, &[])),
- Operation::CreateUser => CreateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_user| validate_option_keys(&create_user.options, &[])),
- _ => Ok(()),
- };
- if let Err(error) = bounds {
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "static-bounds").await;
- return;
- }
- // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
- // partition count for Join) before replication; see `crate::consumer_group`.
- let request = match maybe_rewrite_consumer_group_request(shard, request).await {
- Ok(rewritten) => rewritten,
- Err(error) => {
+ RequestClass::Logout => {
+ handle_logout_request(shard, sessions, transport_client_id, request).await;
+ }
+ RequestClass::UnboundReplicated => {
+ // Replicated request on an unbound transport. Without this short-
+ // circuit, the rewrite below overwrites `header.client` with
+ // `transport_client_id` and dispatches; the request_preflight then
+ // rejects with `NoSession`/`Fenced` and the failure disappears
+ // silently, wedging the SDK until the socket timeout. A typed
+ // `Eviction(NoSession)` is right here, unlike the plain deny of
+ // `UnauthenticatedRead`: a replicated request implies the client
+ // believes it has a session, and that session is gone, so it must
+ // register again. An empty status-0 Reply is not safe here, because
+ // SendMessages is the one replicated operation without a result
+ // section, and its decoder would read the empty body as a
+ // successful send.
warn!(
transport_client_id,
- error = %error,
operation = ?header.operation,
- "dropping consumer-group request with invalid payload"
+ "rejecting replicated request from unbound transport with Eviction(NoSession)"
);
- return;
+ // The eviction context is best-effort off the metadata consensus
+ // (peer shards have none; zeroes are cosmetic -- the SDK only
+ // reads the reason), and the evicted id is the transport id: no
+ // VSR session exists to name.
+ send_eviction(
+ shard,
+ transport_client_id,
+ transport_client_id,
+ EvictionReason::NoSession,
+ "unbound replicated request",
+ )
+ .await;
}
- };
- let request_header = *request.header();
- // Replicated request: run consensus on the metadata owner (shard 0) and
- // bring the committed reply back here. This shard owns the connection,
- // so it writes the reply to the socket via the transport client id --
- // shard 0 can't route by the consensus client id (no home-shard bits).
- match submit_client_request_on_owner(shard, request).await {
- Some(reply) => {
- // The raw PAT token never enters consensus (it is non-deterministic
- // and secret), so the committed reply body is empty. Substitute the
- // raw-token response here, on the minting client's home shard, using
- // the confirmed commit position from the committed reply.
- let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) {
- Ok(reply) => reply,
+ RequestClass::DeleteSegments => {
+ // DeleteSegments is neither a partition nor a metadata consensus op: the
+ // owning shard resolves the requested count to a concrete offset, then a
+ // `TruncatePartition` is replicated through metadata (Option A). Each
+ // replica's reconciler trims to the committed watermark. `classify`
+ // names it ahead of the partition and metadata classes.
+ handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
+ }
+ RequestClass::Partition => {
+ // `bound` is Some here: `classify` sends unbound transports to
+ // `UnboundReplicated`.
+ let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
+ // `get_session` discards the acting user id the partition gate needs;
+ // resolve it from the same bound connection. A bound transport always
+ // has one, but the gate fails closed on `None` rather than trust that.
+ let acting_user_id = sessions.borrow().get_user_id(transport_client_id);
+ dispatch_partition_request(
+ shard,
+ request,
+ vsr_client_id,
+ bound_session,
+ transport_client_id,
+ acting_user_id,
+ )
+ .await;
+ }
+ RequestClass::ReplicatedMetadata => {
+ let request =
+ request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
+ *new_header = header;
+ // Metadata-plane ops route by operation: stamp the sentinel group.
+ new_header.group = server_common::sharding::METADATA_GROUP;
+ // `bound` is always Some here (`classify` sends unbound transports to
+ // `UnboundReplicated`); this sets the consensus client id + session
+ // for the replicated op.
+ if let Some((bound_client_id, bound_session)) = bound {
+ new_header.client = bound_client_id;
+ new_header.session = bound_session;
+ }
+ });
+ let (request, raw_pat_token) = match tcp_chain(
+ shard,
+ sessions,
+ transport_client_id,
+ max_tokens_per_user,
+ request,
+ ) {
+ Ok(rewritten) => rewritten,
+ Err(RewriteDeny { stage, error }) => {
+ send_pre_consensus_deny(shard, transport_client_id, &header, &error, stage)
+ .await;
+ return;
+ }
+ };
+ // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
+ // partition count for Join) before replication; see `crate::consumer_group`.
+ let request = match maybe_rewrite_consumer_group_request(shard, request).await {
+ Ok(rewritten) => rewritten,
Err(error) => {
- warn!(
+ // The rewrite only ever fails on an undecodable body, so a
Review Comment:
**nit** — worth stating what the reader needs here: the reason a replay cannot help is that both of `maybe_rewrite_consumer_group_request`'s own errors are `InvalidCommand` decode failures (`consumer_group.rs:80`, `:94`). Its third error path, `rewrite_request_body`'s `InvalidConfiguration` (`wire.rs:76-77`), needs a body past `u32::MAX` against a 64 MiB `MAX_MESSAGE_SIZE`, so it is unreachable for any client frame — the deny is correct for all three regardless. Fine as-is if you would rather not expand it; flagging only because the sentence is load-bearing for why this is terminal rather than transient.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -615,45 +457,34 @@ async fn handle_client_request<B, MJ, S, SB>(
sessions.borrow_mut().record_heartbeat(transport_client_id);
let header = *request.header();
- if header.operation == Operation::NonReplicated {
- // Auth bypass guard: `PING`, the liveness probe, is the only pre-auth
- // code, on every roster shape. `GET_CLUSTER_METADATA` describes the
- // private replica network and is not something an unauthenticated
- // caller gets to read; a client that dialed a backup no longer needs
- // it to find the leader, because the backup authenticates the login
- // locally and forwards only the consensus proposal
- // (`submit_register_local_or_forward`). Every other non-replicated
- // code MUST go through Register first, which binds the acting user
- // the per-op authz gates resolve.
- let nr_code = u32::from_le_bytes(request.header().reserved[..4].try_into().unwrap());
- // Legacy (pre-register) login codes. The server authenticates only via
- // the Register handshake (LOGIN_REGISTER / LOGIN_REGISTER_WITH_PAT,
- // Operation::Register); the vsr SDK funnels both logins there and never
- // emits these. Reject them uniformly with a typed MalformedLogin (the
- // SDK maps it to InvalidFormat) before the session gate, so a legacy or
- // foreign client fails fast instead of getting the generic
- // Unauthenticated deny the pre-auth guard would send unbound, or the
- // silent empty-ok Reply the bound non-replicated path would send.
- if matches!(
- nr_code,
- LOGIN_USER_CODE | LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE
- ) {
+ let bound = sessions.borrow().get_session(transport_client_id);
Review Comment:
**nit (perf)** — `get_session` is now unconditional here, where pre-PR it ran only inside the `NonReplicated` non-pre-auth branch and again after the Register/Logout returns. PING, Register and Logout each pay one extra `RefCell` borrow plus a map lookup. This is the only per-frame cost this diff adds, and it is small — PING is per-heartbeat-interval, Register/Logout are once per connection — while the diff removes more than it adds (five `Rc`/`Arc` clones now sit behind the `active.insert` guard instead of ahead of it, one fewer checked header cast, one bus clone moved to handler-build time). Net per-frame cost is negative. Not blocking; noting so it is a deliberate trade.
Two pre-existing items on the same line region, for a follow-up rather than this PR: `let header = *request.header()` at `:459` copies 256 bytes on every frame when only the `ReplicatedMetadata` arm needs the pre-`transmute_header` snapshot — bind the *reference* once here and keep the by-value copy inside that arm (do not replace it with per-arm `request.header()` calls: `RequestBacking::header` runs `bytemuck::checked::try_from_bytes` + `.expect` per call at `server_common/src/consensus_message.rs:159-163`, so N accessor calls can cost more than the one memcpy — same reason `failure.rs:168-170`, `:325-327` and `reads.rs:157-160` should bind once). And five same-key lookups happen per data-plane frame (`bus.client_meta` plus an `Rc` clone at `:451`, `ensure_connection`, `record_heartbeat` at `:457`, this `get_session`, then `get_user_id`/`read_context`) where `ensure_transport_connection` is a no-op after frame 1, so the `client_meta` lookup is dead work — one `connections`
walk mirroring the existing `read_context` would collapse them.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -299,8 +230,19 @@ fn enqueue_client_request<B, MJ, S, SB>(
return;
}
- let bus = shard.bus.clone();
+ let shard_handle = Rc::clone(shard_handle);
+ let sessions = Rc::clone(sessions);
+ let system_config = Arc::clone(system_config);
+ let queues = Rc::clone(queues);
+ let active = Rc::clone(active);
bus.spawn(async move {
+ // The handle is set once the shard is built. A frame that beats
+ // it stays queued and the client's next frame drains both, so the
+ // active slot must be released here or that next frame never spawns.
+ let Some(shard) = upgrade_shard_handle(&shard_handle) else {
+ active.borrow_mut().remove(&client_id);
Review Comment:
**nit** — releasing the active slot here is right (otherwise the client's next frame finds the slot taken and nothing ever drains), but the queued frame stays in `queues[client_id]` forever if the client sends nothing more: the connection-lost hook clears neither `queues` nor `active`. Unreachable in production, since listeners bind after the backfill (`boot/mod.rs:912-920` asserts it).
Three pre-existing items share this state machine and want one fix in the connection-lost hook: a panic inside `handle_client_request` leaves `client_id` in `active` forever (compio catches the detached task's panic), so every later frame for that client queues and never drains; `pop_next_client_request` removes the map entry whenever a queue drains to empty (`:303`), so a lockstep SDK pays a `HashMap` insert plus a `VecDeque` alloc/free per request — keeping the entry is safe, `pop_next_client_request` already handles empty-but-present correctly; and the per-client `VecDeque` is unbounded with no end-to-end backpressure, because the bounded socket->dispatch channel drains at spin speed into it (`enqueue_client_request` is synchronous). Also consider `AHashMap` for `queues`/`active` and `SessionManager`'s maps — ids are server-minted so there is no HashDoS surface, `ahash` is already a direct dep, and `message_bus/src/lib.rs:702` sets the precedent (check `iter_clients` orderi
ng first).
##########
core/server/src/dispatch/mod.rs:
##########
@@ -366,193 +308,93 @@ fn pop_next_client_request(
message
}
-/// Per-request partitions-count cap, shared by create-topic, create-partitions
-/// and delete-partitions admission. Runs pre-consensus like
-/// [`validate_topic_bounds`]: an oversized count must not burn a replicated
-/// log entry (create-partitions admission would also allocate that many
-/// consensus-group ids before replicating).
-///
-/// Zero passes here because a zero-partition TOPIC is legal (legacy
-/// `create_topic` admits `0..=MAX`); the add/remove requests reject it in
-/// [`validate_partitions_change_count`].
-const fn validate_partitions_count(partitions_count: u32) -> Result<(), IggyError> {
- if partitions_count > MAX_PARTITIONS_PER_REQUEST {
- return Err(IggyError::TooManyPartitions);
- }
- Ok(())
+/// Where the funnel routes a client request. Derived by [`classify`], which
+/// IS the routing: [`handle_client_request`] matches on its result. The
+/// variant ORDER mirrors the order of the checks inside [`classify`], and
+/// that order is semantics (documented there).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RequestClass {
Review Comment:
**nit** — `RequestClass` and `classify` (`:362`) are `pub` but referenced only inside this file, its own tests included; sibling dispatch items use `pub(in crate::dispatch)` and the submodules are private `mod`. Tighten both.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -172,6 +114,12 @@ where
})
}
+/// Build the shard's one client-request handler: per-client FIFO queues
+/// drained one task per client, and the bus connection-lost hook that
+/// logs a dropped connection out. Every transport on the shard must
+/// share the instance (shard 0 hands it to its local QUIC, TCP-TLS and
+/// WSS listeners as well), or a client's ordering guarantee and the
+/// disconnect hook split by transport.
Review Comment:
**nit** — the connection-lost hook below puts `remove_connection(client_id)` as the *first* operand of the `&&` chain, so it mutates before the weak upgrade is attempted: a failed upgrade strips the `SessionManager` entry without submitting the replicated `Logout`, leaking the `ClientTable` entry and its consumer-group memberships. The `&&` chain is unchanged from pre-PR, but shard 0 only now reaches it — the deleted `make_client_request_handler` held a strong `Rc<shard>` whose upgrade could not fail. Window is pre-build / post-runtime-drop only, so effectively unreachable, and it self-heals via `recover()`'s `remove_consumer_group_member` on next boot. Cheap fix: upgrade first, then `remove_connection`.
##########
core/server/src/dispatch/session_ops.rs:
##########
@@ -552,35 +475,18 @@ async fn evict_stale_client<B, MJ, S, SB>(
if let Some((vsr_client_id, session)) = bound {
submit_disconnect_logout(Rc::clone(shard), vsr_client_id, session);
}
- let ctx = shard.plane.metadata().consensus.as_ref().map_or(
- consensus::EvictionContext {
- cluster: 0,
- view: 0,
- replica: 0,
- },
- consensus::EvictionContext::from_consensus,
- );
- let eviction = consensus::build_eviction_message(
- ctx,
+ warn!(
Review Comment:
**nit** — this `warn!` now runs unconditionally *before* `send_eviction`, where pre-PR it sat in the `else` of the send result and fired only on success. The log now asserts a delivery it did not observe, and the contradicting line says only "failed to send host frame" with `context = "stale-client eviction"`.
##########
core/server/src/dispatch/session_ops.rs:
##########
@@ -331,58 +331,28 @@ async fn surface_login_failure<B, MJ, S, SB>(
SB: SuperblockStore + 'static,
{
if error.is_terminal() {
- send_login_eviction(
+ send_eviction(
shard,
transport_client_id,
request_header.client,
eviction_reason_for(error),
+ "login rejection",
Review Comment:
**nit** — this label is passed at four more sites (`:1186`, `:1204`, `:1302`, `:1317`) covering four distinct `EvictionReason`s, so `context` cannot distinguish them; see the `send_host_frame` comment about the dropped `reason` field. More broadly, the one `context` field now carries three naming conventions — snake_case opcodes (`reads.rs:124` `"get_me"`, `partition.rs:551` `"poll_messages"`), kebab-case stages (`rewrite.rs:110` `"personal-access-token"`), prose (`failure.rs:117` `"request denial"`, this line) and hybrids (`partition.rs:1088` `"delete_segments reply"`) — which makes filtering by it impossible. Repo rule 7 wants labels matching literal API names; pick one shape.
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #4036:
URL: https://github.com/apache/iggy/pull/4036
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4036:
URL: https://github.com/apache/iggy/pull/4036#discussion_r3928026755
##########
core/server/src/boot/mod.rs:
##########
@@ -627,6 +645,18 @@ async fn shard_main(
))
.await?;
+ // The shard above owns the metadata state machine's only write handle,
+ // and the peers read through it until their runtimes are gone. Declared
Review Comment:
Fixed, with one deviation from the suggested order: the wait is armed immediately **before** `broadcast_metadata_bundle`, not after it. The broadcast can fail after serving some peers, and by then those peers already hold read handles, so arming after it leaves exactly the window you describe one level up.
`IggyMetadata::mux_stm` is now `Rc<M>`, with `new` taking `impl Into<Rc<M>>` so the 21 existing call sites are untouched. `shard_main` keeps a clone in a binding declared before the wait, and the broadcast moved out of the `MetadataHandoff::Owner` arm to just after it. A `recover()` failure inside that arm needs no wait at all: no peer holds a handle yet, they are still parked in `await_metadata_bundle`.
Drop order is now shard, then `metadata` and its `Rc` clone, then the wait, then `shard_main`'s own clone. So the writer outlives every reader on each exit from the broadcast onward, the `?` returns inside the callee and a failed `.build()` included.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
+//! | [`FrameChannel::Reply`] | status-0 success frame | host-built success replies: login/register, ping, logout, non-replicated read bodies, committed metadata replies |
+//! | silent drop | no frame | deliberate only where a reply would be wrong: an undecodable header (nothing to echo), a transient consensus submit failure (the SDK read-timeout replays) |
Review Comment:
The certification is gone. The row now names the transient consensus submit failure as the one deliberate silence, and calls the `RequestHeader::validate` drop what it is: a gap, not a contract. It says why that matters too, that the fields decode so a deny could be echoed under the transport id, and that the client instead waits out its read timeout.
I did not turn it into a deny in this PR. At that point `try_into_typed::<RequestHeader>` has already failed, so only a `GenericHeader` exists: building the reply would mean reading `operation` and `request` out of a header the server just refused to interpret, and `into_routed()` has not run. That wants its own change with its own test rather than a rider here.
A `Scope:` paragraph under the table also states that it covers `crate::dispatch` only, and names the two neighbours that answer on their own paths, including `stage_transient_deny` shedding the frame when its lifecycle queue is full.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
Review Comment:
Taken the third way. Instead of adding a row for "undecodable client body answered success-shaped", both sites now deny typed, so nothing of that shape rides `EmptyFrame` any more. The row says so: a permanent client error never rides this channel, because there is nothing to retry.
While there, `handle_delete_segments_request` had the same shape one function down and was not in the list: an undecodable body acked status 0 with an empty body, which reads as a completed trim. It denies typed now as well, and the `Err(_) => None` arm plus the `if let Some(truncate)` wrapper collapsed into one `Err(error)` arm that carries the real code.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
+//! | [`FrameChannel::Reply`] | status-0 success frame | host-built success replies: login/register, ping, logout, non-replicated read bodies, committed metadata replies |
+//! | silent drop | no frame | deliberate only where a reply would be wrong: an undecodable header (nothing to echo), a transient consensus submit failure (the SDK read-timeout replays) |
+//! | HTTP status | HTTP status code | the HTTP spine maps the same rejections in `crate::http::error`; it never rides these frames |
+//!
+//! The last two send nothing, so [`FrameChannel`] has no variant for them.
+//! This exit covers HOST-built frames only: the partitions engine builds and
+//! sends produce/poll replies on its own path by design.
+
+use crate::responses::{
+ NonReplicatedResponse, build_deny_reply, build_empty_reply, current_metadata_commit,
+};
+use crate::shell::{ShellBus, ShellShard};
+use bytes::Bytes;
+use consensus::{
+ EvictionContext, MetadataHandle, build_eviction_message,
+ build_incompatible_protocol_eviction_message, build_result_rejection_reply,
+};
+use iggy_binary_protocol::{EvictionReason, PrepareHeader, RoutedRequestHeader};
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use message_bus::BusMessage;
+use server_common::Message;
+use std::rc::Rc;
+use tracing::warn;
+
+/// Labels the channel a host-built frame rides, for the send-failure log.
+/// The taxonomy, including the two channels that never construct a frame,
+/// is on the module doc.
+#[derive(Clone, Copy, Debug)]
+pub(in crate::dispatch) enum FrameChannel {
+ TypedDeny,
+ Eviction,
+ ResyncSentinel,
+ EmptyFrame,
+ Reply,
+}
+
+/// The one send exit for host-built client frames. Best-effort: a failed
+/// send means the connection is gone (or its queue is full), and there is
+/// nothing left to reply on, so the error is logged and dropped. `frame` is
+/// any bus message: a contiguous frozen frame or the vectored poll reply.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_host_frame<B: ShellBus>(
+ bus: &B,
+ transport_client_id: u128,
+ frame: impl Into<BusMessage>,
+ channel: FrameChannel,
+ context: &'static str,
+) {
+ if let Err(send_error) = bus.send_to_client(transport_client_id, frame).await {
+ warn!(
+ transport_client_id,
+ error = %send_error,
+ channel = ?channel,
+ context,
Review Comment:
Fixed, by a different route than a `reason` field on the send-failure line: `FrameChannel::Eviction` now carries the `EvictionReason`, and `FrameChannel` grew a `Display` that renders it, so the line reads `channel=eviction(MalformedLogin)`.
That keeps one shape for the log line while making the reason part of the channel's identity, which is what the taxonomy already claims the channel is. It also means `send_host_frame` needs no eviction-specific parameter, and the five `"login_rejection"` sites become distinguishable without touching their context labels.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -681,353 +512,179 @@ async fn handle_client_request<B, MJ, S, SB>(
IggyError::Unauthenticated.as_code(),
)
.await;
- return;
}
- handle_non_replicated_request(shard, sessions, system_config, transport_client_id, request)
- .await;
- return;
- }
-
- if header.operation == Operation::Register && header.session == 0 && header.request == 0 {
- handle_login_register_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- if header.operation == Operation::Logout {
- handle_logout_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- let bound = sessions.borrow().get_session(transport_client_id);
- if bound.is_none() {
- // Replicated request on an unbound transport. Without this short-
- // circuit, the rewrite below overwrites `header.client` with
- // `transport_client_id` and dispatches; the request_preflight then
- // rejects with `NoSession`/`Fenced` and the failure disappears
- // silently, wedging the SDK until the socket timeout. A typed
- // `Eviction(NoSession)` is right here, unlike the pre-auth read
- // guard above: a replicated request implies the client believes it
- // has a session, and that session is gone, so it must register
- // again. An empty status-0 Reply is not safe here, because
- // SendMessages is the one replicated operation without a result
- // section, and its decoder would read the empty body as a
- // successful send.
- warn!(
- transport_client_id,
- operation = ?header.operation,
- "rejecting replicated request from unbound transport with Eviction(NoSession)"
- );
- send_unauthenticated_eviction(shard, transport_client_id).await;
- return;
- }
-
- // DeleteSegments is neither a partition nor a metadata consensus op: the
- // owning shard resolves the requested count to a concrete offset, then a
- // `TruncatePartition` is replicated through metadata (Option A). Each
- // replica's reconciler trims to the committed watermark. Handle it here,
- // ahead of the partition/metadata routing below.
- if header.operation == Operation::DeleteSegments {
- handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
- return;
- }
-
- if header.operation.is_partition() {
- // `bound` is Some here (unbound transports returned above).
- let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
- // `get_session` discards the acting user id the partition gate needs;
- // resolve it from the same bound connection. A bound transport always
- // has one, but the gate fails closed on `None` rather than trust that.
- let acting_user_id = sessions.borrow().get_user_id(transport_client_id);
- dispatch_partition_request(
- shard,
- request,
- vsr_client_id,
- bound_session,
- transport_client_id,
- acting_user_id,
- )
- .await;
- return;
- }
-
- let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
- *new_header = header;
- // Metadata-plane ops route by operation: stamp the sentinel group.
- new_header.group = server_common::sharding::METADATA_GROUP;
- // `bound` is always Some here (unbound transports early-return above);
- // this sets the consensus client id + session for the replicated op.
- if let Some((bound_client_id, bound_session)) = bound {
- new_header.client = bound_client_id;
- new_header.session = bound_session;
- }
- });
- let (request, raw_pat_token) = match maybe_rewrite_pat_request(
- sessions,
- transport_client_id,
- max_tokens_per_user,
- |user_id| {
- shard
- .plane
- .metadata()
- .mux_stm
- .users()
- .read(|users| users.pat_count_of(user_id))
- },
- request,
- ) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Token cap reached, malformed body, or a lost session binding.
- send_pre_consensus_deny(
+ RequestClass::NonReplicatedRead => {
+ // The auth-bypass guard is `classify`'s `UnauthenticatedRead` class:
+ // `PING`, the liveness probe, is the only pre-auth code, on every
+ // roster shape. `GET_CLUSTER_METADATA` describes the private replica
+ // network and is not something an unauthenticated caller gets to
+ // read; a client that dialed a backup no longer needs it to find the
+ // leader, because the backup authenticates the login locally and
+ // forwards only the consensus proposal
+ // (`submit_register_local_or_forward`). Every other non-replicated
+ // code MUST go through Register first, which binds the acting user
+ // the per-op authz gates resolve.
+ handle_non_replicated_request(
shard,
- &header,
+ sessions,
+ system_config,
transport_client_id,
- &error,
- "personal-access-token",
+ request,
)
.await;
- return;
}
- };
- // Hash raw passwords and, for ChangePassword, verify the current password
- // on the primary before replication; see `crate::users`. Replicas store the
- // hash directly. A wrong current password is not denied here: it rides
- // consensus and applies as a committed InvalidCredentials no-op, so the only
- // Err returned is a malformed body.
- let request = match maybe_rewrite_user_password_request(shard, request) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Malformed body: deny fast with InvalidCommand.
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "user-password")
- .await;
- return;
+ RequestClass::LoginRegister => {
+ handle_login_register_request(shard, sessions, transport_client_id, request).await;
}
- };
- // Static bounds run pre-consensus so a rejected request burns no
- // replicated log entry; HTTP covers the same bounds via
- // `command.validate()`. A body that fails to decode denies typed too
- // (`InvalidCommand`), instead of riding consensus just to fail there.
- let bounds = match header.operation {
- Operation::CreateTopic => CreateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_topic| {
- // `parse` doubles as the catalog gate: an unknown key or a
- // malformed value denies typed here, pre-consensus.
- let options = TopicCreateOptions::parse(&create_topic.options)?;
- if let Some(segment_size) = options.segment_size {
- validate_topic_segment_size(
- segment_size.as_bytes_u64(),
- iggy_common::MAX_TOPIC_SEGMENT_SIZE,
- )?;
- }
- let segment_size = options.segment_size.map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- if options
- .preallocate_segments
- .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS)
- {
- validate_preallocated_topic_bytes(segment_size, create_topic.partitions_count)?;
- }
- let max_topic_size = options
- .max_topic_size
- .unwrap_or(MaxTopicSize::ServerDefault);
- validate_topic_bounds(create_topic.partitions_count, max_topic_size, segment_size)?;
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- create_topic.partitions_count,
- );
- Ok(())
- }),
- Operation::CreatePartitions => CreatePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_partitions| {
- validate_partitions_change_count(create_partitions.partitions_count)?;
- let metadata = shard.plane.metadata();
- warn_unenforceable_topic_size_on_partition_add(
- metadata.mux_stm.streams(),
- &create_partitions.stream_id,
- &create_partitions.topic_id,
- shard.bus_max_message_size(),
- create_partitions.partitions_count,
- );
- Ok(())
- }),
- Operation::DeletePartitions => DeletePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|delete_partitions| {
- validate_partitions_change_count(delete_partitions.partitions_count)
- }),
- // Only the updatable subset: the create-time knobs are pushed to
- // partitions when the topic is built and nothing re-pushes them, so
- // accepting one here would store a value no partition ever sees.
- Operation::UpdateTopic => UpdateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_topic| {
- validate_option_keys(&update_topic.options, UPDATABLE_TOPIC_OPTION_KEYS)?;
- let options = TopicCreateOptions::parse(&update_topic.options)?;
- let Some(max_topic_size) = options.max_topic_size else {
- return Ok(());
- };
- // An update can lower the cap below one segment just as a
- // create can, and the stored map would then report a size the
- // topic can never enforce. The floor is this topic's own
- // segment size, since that key is create-only.
- let metadata = shard.plane.metadata();
- let streams = metadata.mux_stm.streams();
- let segment_size = streams
- .topic_segment_size(&update_topic.stream_id, &update_topic.topic_id)
- .map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- validate_topic_size_floor(max_topic_size, segment_size)?;
- let partitions_count = streams
- .topic_partitions_count(&update_topic.stream_id, &update_topic.topic_id)
- .unwrap_or(0);
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- u32::try_from(partitions_count).unwrap_or(u32::MAX),
- );
- Ok(())
- }),
- Operation::UpdateStream => UpdateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_stream| {
- validate_option_keys(&update_stream.options, UPDATABLE_STREAM_OPTION_KEYS)
- }),
- Operation::UpdateUser => UpdateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_user| {
- validate_option_keys(&update_user.options, UPDATABLE_USER_OPTION_KEYS)
- }),
- Operation::CreateStream => CreateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_stream| validate_option_keys(&create_stream.options, &[])),
- Operation::CreateUser => CreateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_user| validate_option_keys(&create_user.options, &[])),
- _ => Ok(()),
- };
- if let Err(error) = bounds {
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "static-bounds").await;
- return;
- }
- // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
- // partition count for Join) before replication; see `crate::consumer_group`.
- let request = match maybe_rewrite_consumer_group_request(shard, request).await {
- Ok(rewritten) => rewritten,
- Err(error) => {
+ RequestClass::Logout => {
+ handle_logout_request(shard, sessions, transport_client_id, request).await;
+ }
+ RequestClass::UnboundReplicated => {
+ // Replicated request on an unbound transport. Without this short-
+ // circuit, the rewrite below overwrites `header.client` with
+ // `transport_client_id` and dispatches; the request_preflight then
+ // rejects with `NoSession`/`Fenced` and the failure disappears
+ // silently, wedging the SDK until the socket timeout. A typed
+ // `Eviction(NoSession)` is right here, unlike the plain deny of
+ // `UnauthenticatedRead`: a replicated request implies the client
+ // believes it has a session, and that session is gone, so it must
+ // register again. An empty status-0 Reply is not safe here, because
+ // SendMessages is the one replicated operation without a result
+ // section, and its decoder would read the empty body as a
+ // successful send.
warn!(
transport_client_id,
- error = %error,
operation = ?header.operation,
- "dropping consumer-group request with invalid payload"
+ "rejecting replicated request from unbound transport with Eviction(NoSession)"
);
- return;
+ // The eviction context is best-effort off the metadata consensus
+ // (peer shards have none; zeroes are cosmetic -- the SDK only
+ // reads the reason), and the evicted id is the transport id: no
+ // VSR session exists to name.
+ send_eviction(
+ shard,
+ transport_client_id,
+ transport_client_id,
+ EvictionReason::NoSession,
+ "unbound replicated request",
+ )
+ .await;
}
- };
- let request_header = *request.header();
- // Replicated request: run consensus on the metadata owner (shard 0) and
- // bring the committed reply back here. This shard owns the connection,
- // so it writes the reply to the socket via the transport client id --
- // shard 0 can't route by the consensus client id (no home-shard bits).
- match submit_client_request_on_owner(shard, request).await {
- Some(reply) => {
- // The raw PAT token never enters consensus (it is non-deterministic
- // and secret), so the committed reply body is empty. Substitute the
- // raw-token response here, on the minting client's home shard, using
- // the confirmed commit position from the committed reply.
- let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) {
- Ok(reply) => reply,
+ RequestClass::DeleteSegments => {
+ // DeleteSegments is neither a partition nor a metadata consensus op: the
+ // owning shard resolves the requested count to a concrete offset, then a
+ // `TruncatePartition` is replicated through metadata (Option A). Each
+ // replica's reconciler trims to the committed watermark. `classify`
+ // names it ahead of the partition and metadata classes.
+ handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
+ }
+ RequestClass::Partition => {
+ // `bound` is Some here: `classify` sends unbound transports to
+ // `UnboundReplicated`.
+ let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
+ // `get_session` discards the acting user id the partition gate needs;
+ // resolve it from the same bound connection. A bound transport always
+ // has one, but the gate fails closed on `None` rather than trust that.
+ let acting_user_id = sessions.borrow().get_user_id(transport_client_id);
+ dispatch_partition_request(
+ shard,
+ request,
+ vsr_client_id,
+ bound_session,
+ transport_client_id,
+ acting_user_id,
+ )
+ .await;
+ }
+ RequestClass::ReplicatedMetadata => {
+ let request =
+ request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
+ *new_header = header;
+ // Metadata-plane ops route by operation: stamp the sentinel group.
+ new_header.group = server_common::sharding::METADATA_GROUP;
+ // `bound` is always Some here (`classify` sends unbound transports to
+ // `UnboundReplicated`); this sets the consensus client id + session
+ // for the replicated op.
+ if let Some((bound_client_id, bound_session)) = bound {
+ new_header.client = bound_client_id;
+ new_header.session = bound_session;
+ }
+ });
+ let (request, raw_pat_token) = match tcp_chain(
+ shard,
+ sessions,
+ transport_client_id,
+ max_tokens_per_user,
+ request,
+ ) {
+ Ok(rewritten) => rewritten,
+ Err(RewriteDeny { stage, error }) => {
+ send_pre_consensus_deny(shard, transport_client_id, &header, &error, stage)
+ .await;
+ return;
+ }
+ };
+ // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
+ // partition count for Join) before replication; see `crate::consumer_group`.
+ let request = match maybe_rewrite_consumer_group_request(shard, request).await {
+ Ok(rewritten) => rewritten,
Err(error) => {
- warn!(
+ // The rewrite only ever fails on an undecodable body, so a
+ // replay cannot help: deny typed instead of leaving the
+ // lockstep connection to its read timeout.
+ send_pre_consensus_deny(
+ shard,
transport_client_id,
- error = %error,
- "failed to build raw PAT reply"
- );
+ &header,
+ &error,
+ "consumer-group",
+ )
+ .await;
return;
}
};
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- operation = ?header.operation,
- "failed to deliver committed reply to client"
- );
+ let request_header = *request.header();
+ // Replicated request: run consensus on the metadata owner (shard 0) and
+ // bring the committed reply back here. This shard owns the connection,
+ // so it writes the reply to the socket via the transport client id --
+ // shard 0 can't route by the consensus client id (no home-shard bits).
+ match submit_client_request_on_owner(shard, request).await {
+ Some(reply) => {
+ // The raw PAT token never enters consensus (it is non-deterministic
+ // and secret), so the committed reply body is empty. Substitute the
+ // raw-token response here, on the minting client's home shard, using
+ // the confirmed commit position from the committed reply.
+ let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) {
+ Ok(reply) => reply,
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to build raw PAT reply"
Review Comment:
Deny sent. No row added, because with the deny in place this is no longer a silence: it rides `TypedDeny`, which the first row already covers.
The comment states the trade instead, that the op committed and only the reply could not be rendered, so the client is told the request failed and can read the minted token back. Silence on a committed op is the worse of the two.
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4036:
URL: https://github.com/apache/iggy/pull/4036#discussion_r3928027765
##########
core/server/src/dispatch/mod.rs:
##########
@@ -615,45 +457,34 @@ async fn handle_client_request<B, MJ, S, SB>(
sessions.borrow_mut().record_heartbeat(transport_client_id);
let header = *request.header();
- if header.operation == Operation::NonReplicated {
- // Auth bypass guard: `PING`, the liveness probe, is the only pre-auth
- // code, on every roster shape. `GET_CLUSTER_METADATA` describes the
- // private replica network and is not something an unauthenticated
- // caller gets to read; a client that dialed a backup no longer needs
- // it to find the leader, because the backup authenticates the login
- // locally and forwards only the consensus proposal
- // (`submit_register_local_or_forward`). Every other non-replicated
- // code MUST go through Register first, which binds the acting user
- // the per-op authz gates resolve.
- let nr_code = u32::from_le_bytes(request.header().reserved[..4].try_into().unwrap());
- // Legacy (pre-register) login codes. The server authenticates only via
- // the Register handshake (LOGIN_REGISTER / LOGIN_REGISTER_WITH_PAT,
- // Operation::Register); the vsr SDK funnels both logins there and never
- // emits these. Reject them uniformly with a typed MalformedLogin (the
- // SDK maps it to InvalidFormat) before the session gate, so a legacy or
- // foreign client fails fast instead of getting the generic
- // Unauthenticated deny the pre-auth guard would send unbound, or the
- // silent empty-ok Reply the bound non-replicated path would send.
- if matches!(
- nr_code,
- LOGIN_USER_CODE | LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE
- ) {
+ let bound = sessions.borrow().get_session(transport_client_id);
Review Comment:
Both deferred items are in this PR after all, so the trade lands the other way: the funnel prologue now costs fewer lookups than before the refactor, not one more.
`SessionManager::touch_connection` stamps the heartbeat and returns the bound session, the acting user and the peer address in ONE `connections` walk. That replaces `record_heartbeat`, `get_session`, the `Partition` arm's `get_user_id` and `reads.rs`'s `read_context`, and the `bus.client_meta` lookup now runs only on a transport's first frame, which is the only place the address has to come from the bus. `read_context` and the already-dead `connection_address` are gone.
`header` is bound as a reference; only the `ReplicatedMetadata` arm takes the by-value copy, which it needs for the pre-`transmute_header` echo on a deny. `queues`, `active` and `SessionManager`'s two maps are on `ahash` now: ids are server-minted and neither map is order-sensitive.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -299,8 +230,19 @@ fn enqueue_client_request<B, MJ, S, SB>(
return;
}
- let bus = shard.bus.clone();
+ let shard_handle = Rc::clone(shard_handle);
+ let sessions = Rc::clone(sessions);
+ let system_config = Arc::clone(system_config);
+ let queues = Rc::clone(queues);
+ let active = Rc::clone(active);
bus.spawn(async move {
+ // The handle is set once the shard is built. A frame that beats
+ // it stays queued and the client's next frame drains both, so the
+ // active slot must be released here or that next frame never spawns.
+ let Some(shard) = upgrade_shard_handle(&shard_handle) else {
+ active.borrow_mut().remove(&client_id);
Review Comment:
The connection-lost hook now frees both `queues` and `active` for the client, which covers the stranded queue entry and the panic-leaves-the-slot-taken case in one place. `pop_next_client_request` keeps the map entry when the queue drains empty, so a lockstep client no longer pays a `HashMap` insert plus a `VecDeque` alloc/free per request, and the hook is what removes it. Both maps moved to `ahash`.
The unbounded per-client `VecDeque` is deliberately NOT capped here. A cap needs an overflow policy, whether to shed the frame, deny typed, or evict the connection, plus an operator-facing knob to size it. Picking either silently inside a review round is worse than leaving the queue as it is. I can open a follow-up issue if you want it tracked.
##########
core/server/src/dispatch/session_ops.rs:
##########
@@ -552,35 +475,18 @@ async fn evict_stale_client<B, MJ, S, SB>(
if let Some((vsr_client_id, session)) = bound {
submit_disconnect_logout(Rc::clone(shard), vsr_client_id, session);
}
- let ctx = shard.plane.metadata().consensus.as_ref().map_or(
- consensus::EvictionContext {
- cluster: 0,
- view: 0,
- replica: 0,
- },
- consensus::EvictionContext::from_consensus,
- );
- let eviction = consensus::build_eviction_message(
- ctx,
+ warn!(
Review Comment:
Reworded rather than moved. `send_eviction` swallows the send result by design, since it is best-effort and logs its own failure, so there is no result left to branch on.
The line now claims only what it observed: the session was dropped and the `Logout` submitted, and the notice is being sent. A delivery failure still comes from `send_host_frame`, and it now carries the reason through the channel label (see the `send_host_frame` thread).
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4036:
URL: https://github.com/apache/iggy/pull/4036#discussion_r3926679128
##########
core/server/src/boot/mod.rs:
##########
@@ -627,6 +645,18 @@ async fn shard_main(
))
.await?;
+ // The shard above owns the metadata state machine's only write handle,
+ // and the peers read through it until their runtimes are gone. Declared
Review Comment:
**warning** — this comment claims more than the code delivers: "every exit from here on, clean or `?`, waits for the peers" is false for the exits that matter.
`build_shard_for_thread` takes `metadata: ServerMetadata` **by value** (`boot/recovery.rs:75`), so the `?` at `recovery.rs:112`, `:201` and `:237` drop shard 0's only metadata `WriteCell` *inside the callee*, and a failed `.build()` at `:265` drops it inside the builder. All four happen before `_peer_exit_wait` exists. Peers are reading concurrently — their own build's first statement is `metadata.mux_stm.streams().read(..)` (`recovery.rs:84`, again `:144`), then the reconciler's first pass (`partition_reconciler.rs:336` -> `:1115`) — and hit `.expect("read handle should be accessible")` at `metadata/src/stm/mod.rs:129`. Nothing serializes the two builds: `BootstrapBarrier` is signalled after both.
Same shape one level up: `broadcast_metadata_bundle(..).await?` (`boot/mod.rs:525-533`) can `Err` after serving *some* peers (`boot/handoff.rs:153`, `:157`), and `recovered.mux_stm` is a local of that match arm, so no outer binding can drop after it.
Drop-order cannot reach a callee-owned drop, so moving this declaration up does not fix it. Fix: arm the wait right after `broadcast_metadata_bundle`, and keep the writer out of a fallible call that drops it (`Rc` the mux state machine, clone held before the guard). Note returning `metadata` in the `Err` variant does **not** work — the receiving binding is declared after the guard and drops first.
The by-value signature predates this PR; the invariant claim is new, which is what makes it worth fixing here. Consequence today is misattribution, not data loss: panics unwind (no profile sets `panic = "abort"`), so `join_all` still reports shard 0's real `Err` first, but `first_panic` (`threads.rs:205-207`) records a peer's read-handle panic instead.
##########
core/server/src/boot/mod.rs:
##########
@@ -627,6 +645,18 @@ async fn shard_main(
))
.await?;
+ // The shard above owns the metadata state machine's only write handle,
+ // and the peers read through it until their runtimes are gone. Declared
+ // after the shard so it drops first: every exit from here on, clean or
+ // `?`, waits for the peers before the write side goes with the shard.
+ let _peer_exit_wait = (shard_id == 0).then(|| {
+ PeerExitWait::new(
+ peer_exit,
+ Arc::clone(&shutdown_flag_for_handoff),
Review Comment:
**warning** — the peer wait and the outer thread join share `shutdown_join_timeout` and **nest**, so shard 0 can be reported as the wedged shard while it is legitimately waiting.
`join_all` arms ONE shared `deadline = now + join_timeout` on the first 25ms poll that sees the shutdown flag (`boot/threads.rs:110-113`, `:236-238`). On graceful Ctrl-C the signal handler sets that flag (`threads.rs:71`) at T0, so the outer clock arms at T0+25ms independent of shard 0. Shard 0 cannot *start* this wait until after `await_pump_drain` (<= `shutdown_drain_timeout`) plus the watchdog's own `bus.shutdown(drain_timeout)` (`threads.rs:729`) — a second drain budget. So shard 0 returns at up to `2*drain + join` = 50s at defaults, against a 30s deadline.
Result: `"shard thread still running at the shutdown join deadline; abandoning it"`, a `Wedged` row, `JoinHandle` dropped while the OS thread still holds the metadata writer, and a non-zero exit on a correct shutdown. Because the deadline is shared and already blown, every shard after shard 0 in the loop is abandoned too.
Fix: bound this wait by the *remaining* join budget (thread a deadline through rather than passing the raw knob). Related: `validate_sharding_runtime_knobs` (`boot/threads.rs:496-536`) re-checks `poll <= drain` but not `join >= drain` (or `join <= MAX`), which `ShardingConfig::validate` does enforce (`configs/src/server_config/sharding.rs:251-268`) — that gap reaches the same abandon-mid-wait outcome by a second route, and `PeerExitWait` is a new consumer of `join`.
##########
core/server/src/boot/threads.rs:
##########
@@ -333,6 +333,132 @@ impl Drop for ShutdownOnDrop {
}
}
+/// Peer shards still running: each [`PeerExitGuard`] counts one out,
+/// [`PeerExitWait`] blocks shard 0 until the count is zero.
+///
+/// Shard 0 owns the metadata state machine's only write handle and every
+/// peer reads through handles that stop working the moment it drops, so
+/// the shard that owns the writer must outlive every reader.
Review Comment:
**nit (doc)** — stated unconditionally, but `PeerExitWait::drop` skips the wait entirely on `thread::panicking()` (`:447`), and `install_panic_hook` only sets the flag without aborting (`:196-210`). So on a shard-0 panic the writer drops with peers still reading and their pumps panic at `metadata/src/stm/mod.rs:129`.
The early return is the **right** behaviour and should stay — blocking an unwinding thread on a `Condvar` for up to 30s inside `runtime.block_on` would stall the io_uring driver during a panic, and if the panic came from shard 0's pump the peers may be waiting on shard 0, so the wait would burn its whole budget or deadlock. Only the comment needs narrowing: say the invariant holds on every exit **except** the panic path, and why.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
+//! | [`FrameChannel::Reply`] | status-0 success frame | host-built success replies: login/register, ping, logout, non-replicated read bodies, committed metadata replies |
+//! | silent drop | no frame | deliberate only where a reply would be wrong: an undecodable header (nothing to echo), a transient consensus submit failure (the SDK read-timeout replays) |
Review Comment:
**warning** — this row is not exhaustive, and its justification is false for one of the classes it covers.
"an undecodable header (nothing to echo)" — `try_into_typed::<RequestHeader>` (`dispatch/mod.rs:415`) runs `verify_frame()` then `validate()` (`server_common/src/consensus_message.rs:310-311`), and `RequestHeader::FRAME_SEALED = false` (`binary_protocol/src/consensus/header.rs:501`) so `verify_frame` is a no-op — `validate` is the whole check. `validate_request_fields` (`header.rs:442-490`) rejects `client == 0`, `Operation::Reserved`, `Register` with `session != 0`/`request != 0`, and non-register non-`NonReplicated` with `session == 0`/`request == 0`. Every one of those **decodes cleanly**, so `build_deny_reply` has all the fields it needs — there is plenty to echo. Those clients get no frame at all and wedge to their read timeout, and the victim of the `session == 0` case is exactly the lost-session client `RequestClass::UnboundReplicated` was built to fail fast. `dispatch/partition.rs:483`'s own `request.max(1)` comment ("older and internal callers may still send zero")
names callers this silently drops, and that normalization sits *after* `validate`.
Two more routes answer nothing and are not listed: `dispatch/mod.rs:663` (`build_raw_pat_reply` `Err` returns with no frame *after* the op committed — both `Err` arms look practically dead, but the row is still missing) and `shard/src/lib.rs:3400-3402` (`stage_transient_deny` sheds on a `try_send` failure and counts a frame drop).
Fix: enumerate the real set, or scope the table explicitly to `crate::dispatch`. Certifying an echoable drop as deliberate is the part that should not ship — it is what stops the next reader from checking.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
Review Comment:
**warning** — "the partition cannot answer yet ... and retries" does not describe two of the cases that ride this channel. An **undecodable client body** is a permanent client error, not a transient partition condition, and there is nothing to retry: `dispatch/partition.rs:514` (poll) and `:698` (consumer-offset) both answer a malformed request with a success-shaped status-0 frame.
The taxonomy has no row for "undecodable client body answered success-shaped". Add one, or give those two sites a distinct `context` — today `:698` and the legitimate no-stored-offset success at `:757` send byte-identical `Bytes::new()` on the same channel with the same `"get_consumer_offset"` label, so nothing in the log tells them apart.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
+//! | [`FrameChannel::Reply`] | status-0 success frame | host-built success replies: login/register, ping, logout, non-replicated read bodies, committed metadata replies |
+//! | silent drop | no frame | deliberate only where a reply would be wrong: an undecodable header (nothing to echo), a transient consensus submit failure (the SDK read-timeout replays) |
+//! | HTTP status | HTTP status code | the HTTP spine maps the same rejections in `crate::http::error`; it never rides these frames |
+//!
+//! The last two send nothing, so [`FrameChannel`] has no variant for them.
+//! This exit covers HOST-built frames only: the partitions engine builds and
Review Comment:
**warning** — the exemption is scoped to "produce/poll replies", but two `TypedDeny`-shaped client frames are built and sent from outside this module, so "the one send exit every host-built frame takes" holds only for frames built in this crate.
`IggyShard::deny_partition_request_transient` (`shard/src/lib.rs:3346-3358`) and `stage_transient_deny` (`:3382-3390`) both build `build_deny_reply_from_request_header(request_header, IggyError::TransientNotAccepted.as_code())` — a nonzero-status client Reply, exactly the `TypedDeny` shape at `:26` — and deliver it via `bus.send_to_client` and `LifecycleFrame::ForwardClientSend` respectively. A transient deny is neither a produce nor a poll reply.
Fix: widen the exemption to name the shard-crate partition denies, or route them through `send_host_frame`.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
+//! | [`FrameChannel::Reply`] | status-0 success frame | host-built success replies: login/register, ping, logout, non-replicated read bodies, committed metadata replies |
+//! | silent drop | no frame | deliberate only where a reply would be wrong: an undecodable header (nothing to echo), a transient consensus submit failure (the SDK read-timeout replays) |
+//! | HTTP status | HTTP status code | the HTTP spine maps the same rejections in `crate::http::error`; it never rides these frames |
+//!
+//! The last two send nothing, so [`FrameChannel`] has no variant for them.
+//! This exit covers HOST-built frames only: the partitions engine builds and
+//! sends produce/poll replies on its own path by design.
+
+use crate::responses::{
+ NonReplicatedResponse, build_deny_reply, build_empty_reply, current_metadata_commit,
+};
+use crate::shell::{ShellBus, ShellShard};
+use bytes::Bytes;
+use consensus::{
+ EvictionContext, MetadataHandle, build_eviction_message,
+ build_incompatible_protocol_eviction_message, build_result_rejection_reply,
+};
+use iggy_binary_protocol::{EvictionReason, PrepareHeader, RoutedRequestHeader};
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use message_bus::BusMessage;
+use server_common::Message;
+use std::rc::Rc;
+use tracing::warn;
+
+/// Labels the channel a host-built frame rides, for the send-failure log.
+/// The taxonomy, including the two channels that never construct a frame,
+/// is on the module doc.
+#[derive(Clone, Copy, Debug)]
+pub(in crate::dispatch) enum FrameChannel {
+ TypedDeny,
+ Eviction,
+ ResyncSentinel,
+ EmptyFrame,
+ Reply,
+}
+
+/// The one send exit for host-built client frames. Best-effort: a failed
+/// send means the connection is gone (or its queue is full), and there is
+/// nothing left to reply on, so the error is logged and dropped. `frame` is
+/// any bus message: a contiguous frozen frame or the vectored poll reply.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_host_frame<B: ShellBus>(
+ bus: &B,
+ transport_client_id: u128,
+ frame: impl Into<BusMessage>,
+ channel: FrameChannel,
+ context: &'static str,
+) {
+ if let Err(send_error) = bus.send_to_client(transport_client_id, frame).await {
+ warn!(
+ transport_client_id,
+ error = %send_error,
+ channel = ?channel,
+ context,
+ "failed to send host frame to client"
+ );
+ }
+}
+
+/// Reply to a request rejected before it reached its plane with the request's
+/// own frame: empty body + nonzero `status`. The nonzero status is the whole
+/// point: the SDK peeks it and surfaces the typed error, whereas a status-0
+/// frame reads as a committed ack for work that never happened. Silence is no
+/// better, the connection decodes replies in lockstep and would wedge on every
+/// later request.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_deny_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ status: u32,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(request_header, transport_client_id, 0, commit, status);
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ "request denial",
+ )
+ .await;
+}
+
+/// Deny a request from an unbound transport without disclosing the metadata
+/// commit frontier. The status is the only field a pre-authenticated caller
+/// needs, while the live commit would expose cluster write activity.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_unbound_deny_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ status: u32,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let reply = build_deny_reply(request_header, transport_client_id, 0, 0, status);
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ "unbound request denial",
+ )
+ .await;
+}
+
+/// Reply to a denied non-replicated read with the request's reply frame: empty
+/// body + nonzero `status`. The SDK peeks the status before body decode and
+/// surfaces the typed error, so a poll denial never reaches the empty-poll
+/// "0 messages" body path.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_non_replicated_deny<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ request: &Message<RoutedRequestHeader>,
+ transport_client_id: u128,
+ status: u32,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(
+ request.header(),
+ request.header().client,
+ request.header().session,
+ commit,
+ status,
+ );
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ "non-replicated denial",
+ )
+ .await;
+}
+
+/// Reject a request before it reaches consensus: warn, then send the typed
+/// deny reply. A silent drop would wedge every later request on the
+/// connection until the socket read timeout. `context` labels the rejection
+/// site in both log lines.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_pre_consensus_deny<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ error: &IggyError,
+ context: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ warn!(
+ transport_client_id,
+ error = %error,
+ operation = ?request_header.operation,
+ context,
+ "denying request pre-consensus"
+ );
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(
+ request_header,
+ transport_client_id,
+ 0,
+ commit,
+ error.as_code(),
+ );
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ context,
+ )
+ .await;
+}
+
+/// Result-framed rejection Reply: status 0, body `[count=1][index][code]`.
+/// The SDK decodes the nonzero result code, so a transient code makes it
+/// replay the same request at once instead of waiting out its read timeout.
+/// Replying empty instead would surface as a hard `InvalidFormat` decode
+/// failure and break the replay.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_result_rejection<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ error: &IggyError,
+ context: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = build_result_rejection_reply(request_header, commit, error.as_code());
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ context,
+ )
+ .await;
+}
+
+/// Best-effort session-terminal `Eviction` frame: the client's session is
+/// gone (or was never granted), so it must register again. Every frame
+/// transport decodes `Command::Eviction` and maps the typed reason
+/// (`NoSession` -> `Unauthenticated`, ...), so clients fail fast with the
+/// real cause instead of a body-decode failure or a timeout. Consensus
+/// context (cluster/view/replica) is stamped on the metadata shard and
+/// zeroed elsewhere; the SDK only reads the reason, plus the protocol
+/// window on `IncompatibleProtocol`.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_eviction<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ vsr_client_id: u128,
+ reason: EvictionReason,
+ context: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let ctx = shard.plane.metadata().consensus.as_ref().map_or(
+ EvictionContext {
+ cluster: 0,
+ view: 0,
+ replica: 0,
+ },
+ EvictionContext::from_consensus,
+ );
+ let eviction = match reason {
+ EvictionReason::IncompatibleProtocol => {
+ build_incompatible_protocol_eviction_message(ctx, vsr_client_id)
+ }
+ _ => build_eviction_message(ctx, vsr_client_id, reason),
+ };
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ eviction.into_generic().into_frozen(),
+ FrameChannel::Eviction,
+ context,
+ )
+ .await;
+}
+
+/// Send a non-replicated reply body to a client, stamping the current
+/// metadata commit. Shared by the non-replicated read arms; `channel`
+/// labels the body shape (a real answer, a fail-fast empty poll, or the
+/// re-sync sentinel) for the send-failure log.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_non_replicated_bytes<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ request: &Message<RoutedRequestHeader>,
+ transport_client_id: u128,
+ bytes: Bytes,
+ channel: FrameChannel,
+ context: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = NonReplicatedResponse::Bytes(bytes).into_reply(
+ request.header(),
+ request.header().client,
+ request.header().session,
+ commit,
+ );
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ channel,
+ context,
+ )
+ .await;
+}
+
+/// Ack a consumer-offset op whose body could not be rewritten for the
+/// partition plane with an empty Reply. The SDK connection processes replies
+/// in lockstep, so a silent drop wedges every subsequent request on that
+/// connection.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_empty_partition_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = build_empty_reply(request_header, transport_client_id, 0, commit);
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::EmptyFrame,
+ "empty partition reply",
+ )
+ .await;
+}
+
+// Byte snapshots pinning each channel's frame to the pre-refactor inline
+// construction. DELIBERATELY temporary: they freeze the refactor, not the
Review Comment:
**warning** — repo rule 8 says comments must never reference the current or a future task/PR, and "a later PR removes them" is an unenforceable removal trigger: it leaves ~250 lines of snapshot test with nothing a future reader can evaluate. State the invariant these snapshots pin instead, or file the removal and drop the note.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,652 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries |
+//! | [`FrameChannel::Reply`] | status-0 success frame | host-built success replies: login/register, ping, logout, non-replicated read bodies, committed metadata replies |
+//! | silent drop | no frame | deliberate only where a reply would be wrong: an undecodable header (nothing to echo), a transient consensus submit failure (the SDK read-timeout replays) |
+//! | HTTP status | HTTP status code | the HTTP spine maps the same rejections in `crate::http::error`; it never rides these frames |
+//!
+//! The last two send nothing, so [`FrameChannel`] has no variant for them.
+//! This exit covers HOST-built frames only: the partitions engine builds and
+//! sends produce/poll replies on its own path by design.
+
+use crate::responses::{
+ NonReplicatedResponse, build_deny_reply, build_empty_reply, current_metadata_commit,
+};
+use crate::shell::{ShellBus, ShellShard};
+use bytes::Bytes;
+use consensus::{
+ EvictionContext, MetadataHandle, build_eviction_message,
+ build_incompatible_protocol_eviction_message, build_result_rejection_reply,
+};
+use iggy_binary_protocol::{EvictionReason, PrepareHeader, RoutedRequestHeader};
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use message_bus::BusMessage;
+use server_common::Message;
+use std::rc::Rc;
+use tracing::warn;
+
+/// Labels the channel a host-built frame rides, for the send-failure log.
+/// The taxonomy, including the two channels that never construct a frame,
+/// is on the module doc.
+#[derive(Clone, Copy, Debug)]
+pub(in crate::dispatch) enum FrameChannel {
+ TypedDeny,
+ Eviction,
+ ResyncSentinel,
+ EmptyFrame,
+ Reply,
+}
+
+/// The one send exit for host-built client frames. Best-effort: a failed
+/// send means the connection is gone (or its queue is full), and there is
+/// nothing left to reply on, so the error is logged and dropped. `frame` is
+/// any bus message: a contiguous frozen frame or the vectored poll reply.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_host_frame<B: ShellBus>(
+ bus: &B,
+ transport_client_id: u128,
+ frame: impl Into<BusMessage>,
+ channel: FrameChannel,
+ context: &'static str,
+) {
+ if let Err(send_error) = bus.send_to_client(transport_client_id, frame).await {
+ warn!(
+ transport_client_id,
+ error = %send_error,
+ channel = ?channel,
+ context,
Review Comment:
**warning** — this PR deleted a diagnostic it had in hand. The `send_login_eviction` helper this replaced logged `reason = ?reason` on a send failure; `send_host_frame` logs only `transport_client_id`, `error`, `channel` and `context`. `send_eviction` (`:267-302`) receives `reason` and never logs it, and five call sites in `dispatch/session_ops.rs` (`:339`, `:1186`, `:1204`, `:1302`, `:1317`) all pass the same `"login rejection"` label across four distinct `EvictionReason`s — so `MalformedLogin`, `IncompatibleProtocol` and `InvalidCredentials` are indistinguishable when a send fails. Add `reason` to the eviction send-failure line.
##########
core/server/src/dispatch/authz.rs:
##########
@@ -328,46 +261,17 @@ where
|request| (&request.stream_id, &request.topic_id),
Permissioner::get_consumer_groups,
),
- _ => Ok(()),
- }
-}
-
-/// Reply to a denied non-replicated read with the request's reply frame: empty
-/// body + nonzero `status`. The SDK peeks the status before body decode and
-/// surfaces the typed error, so a poll denial never reaches the empty-poll
-/// "0 messages" body path.
-#[allow(clippy::future_not_send)]
-pub(in crate::dispatch) async fn send_non_replicated_deny<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: &Message<RoutedRequestHeader>,
- transport_client_id: u128,
- status: u32,
-) where
- B: ShellBus,
- MJ: JournalHandle + 'static,
- MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
- S: 'static,
- SB: SuperblockStore + 'static,
-{
- let commit = current_metadata_commit(shard);
- let reply = build_deny_reply(
- request.header(),
- request.header().client,
- request.header().session,
- commit,
- status,
- );
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- status,
- error = %error,
- "failed to surface non-replicated authz denial"
- );
+ // No on-demand flush primitive exists, so the code stays the builder's
+ // `FeatureUnavailable` (its arm still serves the HTTP caller).
Review Comment:
**warning** — "(its arm still serves the HTTP caller)" is false. `read_local` (`http/reads.rs:104`) is the only HTTP entry to `build_non_replicated_response`, and its eleven call sites pass fixed constants (`http/handlers.rs:290`, `314`, `348`, `382`, `419`, `447`, `485`, `526`, `566`, `594`, `1722`) — `DESCRIBE_OPTIONS`, `GET_STREAM(S)`, `GET_TOPIC(S)`, `GET_USER(S)`, `GET_CONSUMER_GROUP(S)`, `GET_STATS`, `GET_PERSONAL_ACCESS_TOKENS`. No `FLUSH_UNSAVED_BUFFER`, and there are zero flush references anywhere under `core/server/src/http/` — which is why `flush_vsr.rs` is declared `test_client_transport = [Tcp]`.
So this gate arm is now the only thing producing `FeatureUnavailable` for flush (keep it), and `responses.rs:590` is dead. Drop the false clause from the comment.
##########
core/server/src/dispatch/authz.rs:
##########
@@ -328,46 +261,17 @@ where
|request| (&request.stream_id, &request.topic_id),
Permissioner::get_consumer_groups,
),
- _ => Ok(()),
- }
-}
-
-/// Reply to a denied non-replicated read with the request's reply frame: empty
-/// body + nonzero `status`. The SDK peeks the status before body decode and
-/// surfaces the typed error, so a poll denial never reaches the empty-poll
-/// "0 messages" body path.
-#[allow(clippy::future_not_send)]
-pub(in crate::dispatch) async fn send_non_replicated_deny<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: &Message<RoutedRequestHeader>,
- transport_client_id: u128,
- status: u32,
-) where
- B: ShellBus,
- MJ: JournalHandle + 'static,
- MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
- S: 'static,
- SB: SuperblockStore + 'static,
-{
- let commit = current_metadata_commit(shard);
- let reply = build_deny_reply(
- request.header(),
- request.header().client,
- request.header().session,
- commit,
- status,
- );
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- status,
- error = %error,
- "failed to surface non-replicated authz denial"
- );
+ // No on-demand flush primitive exists, so the code stays the builder's
+ // `FeatureUnavailable` (its arm still serves the HTTP caller).
+ FLUSH_UNSAVED_BUFFER_CODE => Err(IggyError::FeatureUnavailable),
+ // A replicated code smuggled inside a `NonReplicated` header keeps the
+ // builder's `FeatureUnavailable`; a table-listed code with no arm above
+ // and an unknown code are both refused as `InvalidCommand`, never
+ // deferred to the builder's empty-ok catch-all.
Review Comment:
**nit** — with this tail refusing every armless code first, `build_non_replicated_response`'s `_ => Ok(NonReplicatedResponse::Empty)` (`responses.rs:598`) is unreachable from **both** callers: the gate refuses it on the TCP side, and HTTP passes only its eleven named codes. The empty-ok catch-all this PR set out to kill therefore survives as a fail-open trap for the next caller that lands.
`Err(IggyError::InvalidCommand)` there is provably safe: all eleven HTTP codes have named arms above the `_`, and the not-found -> 404 mapping comes from *inside* `GET_STREAM`/`GET_TOPIC`/`GET_USER`'s own arms via `response.map_or(Empty, ..)`, not from the catch-all. Sequence it with the flush arm above so flush's answer does not silently flip from `FeatureUnavailable` to `InvalidCommand`.
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4036:
URL: https://github.com/apache/iggy/pull/4036#discussion_r3932143500
##########
core/server/src/boot/threads.rs:
##########
@@ -333,6 +320,173 @@ impl Drop for ShutdownOnDrop {
}
}
+/// The single post-shutdown budget, shared by the main thread's shard
+/// joins and shard 0's peer wait.
+///
+/// Both waits are bounded by `system.sharding.shutdown_join_timeout` and
+/// they NEST: shard 0 cannot start waiting for its peers until its own
+/// drain returned, which is already inside the join budget. Arming one
+/// instant on first use, whichever wait gets there first, keeps the two
+/// inside one deadline instead of stacking two full budgets, so a
+/// correct shutdown cannot report shard 0 as wedged.
+pub(in crate::boot) struct ShutdownDeadline {
+ armed: OnceLock<Instant>,
+ budget: Duration,
+}
+
+impl ShutdownDeadline {
+ pub(in crate::boot) const fn new(budget: Duration) -> Self {
+ Self {
+ armed: OnceLock::new(),
+ budget,
+ }
+ }
+
+ /// Time left in the shared budget, arming it on the first call.
+ /// Callers must only reach this once the shutdown flag is set: a
+ /// running server would otherwise start the clock.
+ fn remaining(&self) -> Duration {
+ self.armed
+ .get_or_init(|| Instant::now() + self.budget)
+ .saturating_duration_since(Instant::now())
+ }
+}
+
+/// Peer shards still running: each [`PeerExitGuard`] counts one out,
+/// [`PeerExitWait`] blocks shard 0 until the count is zero.
+///
+/// Shard 0 owns the metadata state machine's only write handle and every
+/// peer reads through handles that stop working the moment it drops, so
+/// the shard that owns the writer must outlive every reader on every exit
+/// but one: [`PeerExitWait::drop`] skips the wait while shard 0 is
+/// panicking, because parking an unwinding thread on a `Condvar` inside
+/// `runtime.block_on` would stall the `io_uring` driver, and a panic in
+/// shard 0's own pump can be exactly what the peers are blocked on. A
+/// peer mid-read then panics too; the first panic is already recorded and
+/// the process is going down either way.
+pub(in crate::boot) struct PeerExitCountdown {
+ running: Mutex<usize>,
+ all_exited: Condvar,
+}
+
+impl PeerExitCountdown {
+ pub(in crate::boot) const fn new(peers: usize) -> Self {
+ Self {
+ running: Mutex::new(peers),
+ all_exited: Condvar::new(),
+ }
+ }
+
+ /// Block until every peer has counted itself out or `timeout` elapses.
+ /// `Err` carries the number of peers still running at the deadline.
+ fn wait(&self, timeout: Duration) -> Result<(), usize> {
+ let (guard, _) = self
+ .all_exited
+ .wait_timeout_while(
+ self.running.lock().unwrap_or_else(PoisonError::into_inner),
+ timeout,
+ |running| *running > 0,
+ )
+ .unwrap_or_else(PoisonError::into_inner);
+ let running = *guard;
+ drop(guard);
+ if running == 0 { Ok(()) } else { Err(running) }
+ }
+
+ fn peer_exited(&self) {
+ let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner);
+ *running = running.saturating_sub(1);
+ if *running == 0 {
+ self.all_exited.notify_all();
+ }
+ }
+
+ /// Count out peers that never spawned. The countdown is sized before the
+ /// spawn loop, so a failed `thread::Builder::spawn` leaves peers whose
+ /// [`PeerExitGuard`] will never exist: without this shard 0 waits out its
+ /// whole join budget for threads that were never there.
+ pub(in crate::boot) fn peers_never_spawned(&self, count: usize) {
+ for _ in 0..count {
+ self.peer_exited();
+ }
+ }
+}
+
+/// Counts one peer shard out of the [`PeerExitCountdown`] on drop.
+///
+/// Held by `run_shard_thread` from before the runtime exists, so it drops
+/// after the runtime and its tasks are gone: past that point nothing on
+/// the peer's thread can still read shard 0's metadata. Drop runs on every
+/// exit path, the error `?` returns and panic unwinds included.
+struct PeerExitGuard {
+ countdown: Arc<PeerExitCountdown>,
+}
+
+impl PeerExitGuard {
+ const fn new(countdown: Arc<PeerExitCountdown>) -> Self {
+ Self { countdown }
+ }
+}
+
+impl Drop for PeerExitGuard {
+ fn drop(&mut self) {
+ self.countdown.peer_exited();
+ }
+}
+
+/// Shard 0's side of the [`PeerExitCountdown`]: blocks on drop until every
+/// peer has exited, so whatever is declared before it outlives every
+/// peer's reads.
+///
+/// Flips the shutdown flag before waiting: a peer parked on its bus token
+/// only starts its drain once the flag is set, and the thread-level
+/// `ShutdownOnDrop` flips it only after `block_on` returns, which is after
+/// this wait. Bounded by what is left of the [`ShutdownDeadline`] that
+/// [`ShardHandles::join_all`] shares, with the same abandon-and-log
+/// policy, so a wedged peer cannot hold shard 0 past the deadline the
+/// main thread is itself counting down.
+pub(in crate::boot) struct PeerExitWait {
+ countdown: Arc<PeerExitCountdown>,
+ shutdown_flag: Arc<AtomicBool>,
+ deadline: Arc<ShutdownDeadline>,
+}
+
+impl PeerExitWait {
+ pub(in crate::boot) const fn new(
+ countdown: Arc<PeerExitCountdown>,
+ shutdown_flag: Arc<AtomicBool>,
+ deadline: Arc<ShutdownDeadline>,
+ ) -> Self {
+ Self {
+ countdown,
+ shutdown_flag,
+ deadline,
+ }
+ }
+}
+
+impl Drop for PeerExitWait {
+ fn drop(&mut self) {
+ // The panic is the fault to report and `ShutdownOnDrop` still
+ // flips the flag for the peers; blocking an unwinding thread here
+ // would only delay it.
+ if thread::panicking() {
+ return;
+ }
+ self.shutdown_flag.store(true, Ordering::Relaxed);
+ let remaining = self.deadline.remaining();
Review Comment:
**blocker** — this closes the nested-budget problem by making the peer-exit wait a debtor of the join budget, which reopens the writer-drop race the commit exists to close.
`remaining()` is `saturating_duration_since` (`:349-353`), so it floors at `Duration::ZERO`. `join_all` arms the shared instant at flag-flip (`:229`), and shard 0 cannot reach this drop until `bus.shutdown()` and `await_pump_drain` have finished. When the join leg has spent the budget, `countdown.wait(ZERO)` re-checks the predicate and returns `Err(running)` **without blocking** (`:381-395`), shard 0 releases, the last `Rc<ServerMuxStateMachine>` drops, and any peer inside `LeftRight::read` panics on `.expect("read handle should be accessible")` (`core/metadata/src/stm/mod.rs:129`).
Reachability: `join == drain` is legal and explicitly blessed by `core/configs/src/server_config/sharding.rs:378` (`join_equal_to_drain_is_accepted`), and at that setting one full-length drain exhausts the budget on its own. At the shipped default (join 30s / drain 10s, `core/server/config.toml:888`) zero is not reachable, but the margin still silently drops from a guaranteed 30s to a worst-case ~10s.
The distinction that makes this worth blocking on: the join budget bounds a LIVENESS concern (do not hang process exit on a wedged shard), while this wait enforces a SAFETY invariant (the writer must outlive every reader). Funding the second out of the first lets an exit-latency timeout cancel a correctness fence. The invariant doc at `:359-365` names exactly one exception (panic); deadline exhaustion is now a second one and is not named.
Note the original finding was about `join_all`'s **verdict** — shard 0 being reported `Wedged` on a correct shutdown — not about the peer wait's length. Fix: floor it (`self.deadline.remaining().max(drain_timeout)`), or give the wait its own budget and let `join_all` accept shard 0 returning within `budget + peer_budget` without a `Wedged` row.
##########
core/server/src/boot/threads.rs:
##########
@@ -649,6 +827,188 @@ mod tests {
);
}
+ #[test]
+ fn peer_exit_countdown_releases_the_waiter_once_every_peer_is_out() {
+ let countdown = Arc::new(PeerExitCountdown::new(2));
+ let guards: [PeerExitGuard; 2] =
+ std::array::from_fn(|_| PeerExitGuard::new(Arc::clone(&countdown)));
+ assert_eq!(
+ countdown.wait(Duration::from_millis(10)),
+ Err(2),
+ "two live peers must hold the waiter past a short deadline"
+ );
+ let peers: Vec<_> = guards
+ .into_iter()
+ .map(|guard| {
+ thread::spawn(move || {
+ thread::sleep(Duration::from_millis(20));
+ drop(guard);
+ })
+ })
+ .collect();
+ assert_eq!(
+ countdown.wait(Duration::from_secs(30)),
+ Ok(()),
+ "the last guard drop must release the waiter"
+ );
+ for peer in peers {
+ peer.join()
+ .expect("peer thread dropped its guard without panicking");
+ }
+ }
+
+ #[test]
+ fn peer_exit_guard_counts_out_during_unwind() {
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let guard = PeerExitGuard::new(Arc::clone(&countdown));
+ // `resume_unwind` skips the panic hook, so the unwind is silent.
+ let unwound = panic::catch_unwind(panic::AssertUnwindSafe(|| {
+ let _guard = guard;
+ panic::resume_unwind(Box::new("peer shard body panicked"));
+ }));
+ assert!(unwound.is_err());
+ assert_eq!(
+ countdown.wait(Duration::ZERO),
+ Ok(()),
+ "a guard dropped by a panic unwind must still count its peer out"
+ );
+ }
+
+ #[test]
+ fn peer_exit_wait_flips_the_flag_before_waiting() {
+ // Stands in for a peer parked on its bus token: it exits only once
+ // the shutdown flag is set, so a waiter that set the flag after
+ // waiting would sit out the whole budget.
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let shutdown_flag = Arc::new(AtomicBool::new(false));
+ let peer = thread::spawn({
+ let guard = PeerExitGuard::new(Arc::clone(&countdown));
+ let shutdown_flag = Arc::clone(&shutdown_flag);
+ move || {
+ while !shutdown_flag.load(Ordering::Relaxed) {
+ thread::sleep(Duration::from_millis(1));
+ }
+ drop(guard);
+ }
+ });
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ countdown,
+ Arc::clone(&shutdown_flag),
+ Arc::new(ShutdownDeadline::new(Duration::from_secs(30))),
+ ));
+ assert!(
+ started.elapsed() < Duration::from_secs(5),
+ "the waiter must not sit out its budget on a peer that waits for the flag"
+ );
+ assert!(shutdown_flag.load(Ordering::Relaxed));
+ peer.join()
+ .expect("peer thread dropped its guard without panicking");
+ }
+
+ #[test]
+ fn peer_exit_wait_gives_up_at_the_deadline() {
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let _wedged_peer = PeerExitGuard::new(Arc::clone(&countdown));
+ let timeout = Duration::from_millis(50);
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ countdown,
+ Arc::new(AtomicBool::new(false)),
+ Arc::new(ShutdownDeadline::new(timeout)),
+ ));
+ let waited = started.elapsed();
+ assert!(
+ waited >= timeout && waited < Duration::from_secs(5),
+ "a wedged peer must be abandoned at the deadline, waited {waited:?}"
+ );
+ }
+
+ #[test]
+ fn peer_wait_and_shard_join_share_one_shutdown_budget() {
+ // Regression: the two waits nest (shard 0 cannot start waiting for
+ // its peers until its own drain returned, already inside the join
+ // budget). With a budget each, a clean shutdown reported shard 0 as
+ // wedged and exited non-zero.
+ let deadline = Arc::new(ShutdownDeadline::new(Duration::from_millis(200)));
+ let shutdown_flag = AtomicBool::new(true);
+ // Never finishes: stands in for the slow drain that arms and then
+ // spends the shared budget. The thread leaks into the test process.
+ let wedged_shard = thread::spawn(|| -> Result<(), ServerError> {
+ loop {
+ thread::sleep(Duration::from_secs(1));
+ }
+ });
+ assert!(
+ join_until_shutdown_deadline(wedged_shard, &shutdown_flag, &deadline).is_none(),
+ "the join must spend the budget it armed"
+ );
+
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let _wedged_peer = PeerExitGuard::new(Arc::clone(&countdown));
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ countdown,
+ Arc::new(AtomicBool::new(false)),
+ Arc::clone(&deadline),
+ ));
+ assert!(
+ started.elapsed() < Duration::from_millis(100),
Review Comment:
**blocker (same defect, second half)** — this assertion pins the unsafe release as intended behaviour, which is why the item above cannot be resolved by a comment.
The test builds `PeerExitCountdown::new(1)` and holds `_wedged_peer`, so the peer is **never counted out**. It then spends the entire 200ms budget in `join_until_shutdown_deadline` and asserts that dropping `PeerExitWait` returns in under 100ms. That is an assertion that shard 0 drops the metadata writer while a peer is still running — a green test encoding exactly the race `PeerExitWait` was added to prevent.
The stated intent ("inherit what is left of the budget, not arm a second one") is reasonable; it is the live peer guard that turns it into a specification of the starvation. A future reader will treat this as settled and no test will catch a regression.
Fix: keep a case that pins budget inheritance with the peer already counted out, and add one that asserts the wait still HONOURS a live peer (blocks for the floor, or for its own budget).
##########
core/server/src/dispatch/mod.rs:
##########
@@ -351,208 +308,106 @@ fn pop_next_client_request(
active: &ActiveClientRequests,
client_id: u128,
) -> Option<Message<GenericHeader>> {
- let mut queues = queues.borrow_mut();
- let Some(queue) = queues.get_mut(&client_id) else {
- active.borrow_mut().remove(&client_id);
- return None;
- };
- let message = queue.pop_front();
- if queue.is_empty() {
- queues.remove(&client_id);
- }
+ // The entry survives draining to empty: removing it would cost a
Review Comment:
**warning** — moving the free to the connection-lost hook makes the hook the sole owner, but the hook is not the last toucher of this map, so an entry can now leak for process lifetime.
The hook fires from the TRANSPORT task's scopeguard (`core/message_bus/src/installer/tcp.rs:118-130`), while the separate DISPATCH task still holds buffered frames in `in_rx`. `ctx.in_tx` drops when `conn.run(ctx)` returns, and `async_channel` delivers every buffered item before `recv()` reports `Closed` — so `client_dispatch_loop` (`tcp.rs:259-289`) still calls `on_request` after the hook ran, and `enqueue_client_request` re-creates the entry via `.entry(client_id).or_default()` (`:239-243`). Nothing frees it again: the hook is once-only (`message_bus/src/lib.rs:983` guards on `remove(..).is_some()`) and `client_id` is monotonic `(shard << 112) | seq`, never reused. The two guards that could stop the loop — `aborted` (`tcp.rs:275`) and `conn_token` — are both insert-race-only, so this is not a rare interleaving; it happens whenever a frame was buffered at close.
Pre-PR there was no leak, because this branch freed the entry every drain.
Fix: restore the removal here and keep the hook's removal for the mid-flight case. That also resolves the sticky-capacity issue below, so no `shrink_to` is needed.
Separately, on the stated motive: `VecDeque::pop_front` never releases capacity, so as written a client's PEAK depth stays resident (no `shrink_to`/`shrink_to_fit` exists anywhere in `core/server` or `core/shard`). For the in-tree SDKs that is small — they are lockstep, so depth stays at the 4-slot `RawVec` minimum, ~96 B per connection — but it interacts badly with the uncapped-queue item you deferred: an unbounded transient spike becomes a permanent high-water mark. Restoring the removal makes both moot.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -725,13 +781,15 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
SB: SuperblockStore + 'static,
{
let Ok(wire) = GetConsumerOffsetRequest::decode_from(request_body(request)) else {
- // Undecodable: an empty body decodes as None (no offset) on the SDK.
- send_non_replicated_bytes(
+ // Same rule as the poll above: an empty body is the legitimate
+ // "no offset stored" answer, so a malformed request must not send
+ // one. Byte-identical frames on the same channel with the same
+ // context would also be indistinguishable in the send-failure log.
+ send_non_replicated_deny(
shard,
request,
transport_client_id,
- Bytes::new(),
- "get_consumer_offset",
+ IggyError::InvalidCommand.as_code(),
Review Comment:
**warning** — the undecodable-body fail-open is closed here, but the not-found one nine lines below is still open, and it is the only finding in this round reachable from a **well-formed request through a public SDK method**.
`:827` denies typed only on `IggyError::PartitionNotFound`; `StreamIdNotFound` and `TopicIdNotFound` from `resolve_partition_namespace` (`core/server/src/responses.rs:466-476`) fall through to the generic `Err` arm at `:836` and answer `Bytes::new()` — which every SDK decodes as "no offset stored":
- Rust `get_consumer_offset(..) -> Result<Option<ConsumerOffsetInfo>, IggyError>` returns `Ok(None)`, minted at `core/common/src/traits/binary_impls/consumer_offsets.rs:85`
- Go `GetConsumerOffset(..)` returns `(nil, nil)` (`binary_serialization/binary_response_deserializer.go:41-43`)
- C# `GetOffsetAsync(..)` returns `null` (`Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs:386-389`)
- Java `Optional.empty()`, Node `null`
So `get_consumer_offset` against a typo'd or deleted stream is indistinguishable from a fresh consumer: it resumes from its configured default and silently reprocesses, with no error anywhere. Two contrasts make it hard to defend: the SAME typo through `poll_messages` returns `Err(StreamIdNotFound)` (`:654-659` denies all three variants), and the SAME read over REST 404s (`core/server/src/http/handlers.rs:1312`). One lookup, three answers.
Fix: extend the `:827` match to `StreamIdNotFound | TopicIdNotFound`, mirroring `:654-659`. Error-path only, and cheaper than today since it drops the `Bytes::new()` + `send_non_replicated_bytes` for a header-only deny.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,677 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions. The reason rides the channel label, since one `context` covers four of them |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries. A permanent client error never rides this channel: an undecodable body denies typed, because there is nothing to retry |
Review Comment:
**warning** — "A permanent client error never rides this channel" is false as written, and it is a rule this commit authored.
A consumer-group poll that omits the partition id decodes cleanly (`core/binary_protocol/src/requests/messages/poll_messages.rs:95` maps `partition_flag != 1` to `None`), then `resolve_poll_request` returns `IggyError::InvalidIdentifier` (`partition.rs:946`). That variant is absent from the deny set at `partition.rs:654-659`, so it falls to `empty_poll_fallback(0)` (`:754`) and rides `EmptyFrame` as a status-0 16-byte empty poll — a permanent client error answered as a successful 0-message read, which is the exact symptom your commit message describes.
Reachability is foreign/hand-rolled clients only (every in-tree SDK resolves the partition client-side, e.g. `core/common/src/traits/binary_impls/messages.rs:317-320`) — but that is the same reachability class as the three fail-opens this commit closed, so it cannot be the reason to leave the sentence absolute.
Fix: add `InvalidIdentifier` to the `:654-659` match (one pattern, error path, strictly cheaper than the current `Bytes` allocation), or drop the absolute claim from this row.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -581,8 +583,10 @@ async fn relay_partition_reply<B, MJ, S, SB>(
/// the owning shard ([`shard::IggyShard::partition_read`]), and re-encode
/// the stored batches into the legacy wire `PolledMessages` body.
///
-/// Failures reply with an empty body so the SDK fails fast on decode
-/// instead of hanging until its read timeout.
+/// A partition that cannot answer yet replies with the 16-byte empty poll,
+/// which the SDK reads as 0 messages and retries. Permanent client errors
+/// (undecodable body, authz, unresolved target) deny with a nonzero status
Review Comment:
**warning** — this sentence, added by this commit, says permanent client errors including an "unresolved target" deny with a nonzero status. `IggyError::InvalidIdentifier` IS an unresolved target and does not: it is missing from the deny match at `:654-659` and falls to `empty_poll_fallback(0)`. Same defect as the `failure.rs:29` row. Fix the match or narrow the sentence.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -695,18 +677,92 @@ pub(in crate::dispatch) async fn handle_poll_messages<B, MJ, S, SB>(
"poll_messages request rejected; replying empty poll"
);
let partition_id = if matches!(error, IggyError::ConsumerGroupPartitionNotOwned(..)) {
- iggy_common::RESYNC_REQUIRED_PARTITION_SENTINEL
+ RESYNC_REQUIRED_PARTITION_SENTINEL
} else {
0
};
- empty_polled_messages_body(partition_id)
+ empty_poll_fallback(partition_id)
}
};
- send_non_replicated_bytes(shard, request, transport_client_id, body, "poll_messages").await;
+ send_non_replicated_bytes(
+ shard,
+ request,
+ transport_client_id,
+ body,
+ channel,
+ "poll_messages",
+ )
+ .await;
}
-/// Serve `get_consumer_offset`. An empty body decodes as `None` on the SDK
-/// side (no offset stored / partition unknown).
+/// Run the resolved poll on the owning shard and re-encode the stored
+/// batches into the wire `PolledMessages` reply. A failed read or re-encode
+/// hands back the fail-fast empty poll for the partition instead.
+#[allow(clippy::future_not_send)]
+async fn read_polled_messages<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request: &Message<RoutedRequestHeader>,
+ (namespace, partition_id, consumer, args): DecodedPollRequest,
+) -> Result<BusMessage, (Bytes, FrameChannel)>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ match shard
+ .partition_read(namespace, PartitionRead::Poll { consumer, args })
+ .await
+ {
+ Some(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ }) => build_polled_messages_reply(
+ request.header(),
+ current_metadata_commit(shard),
+ partition_id,
+ current_offset,
+ fragments,
+ shard.plane.partitions().config().encryptor.as_deref(),
+ )
+ .map_err(|error| {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to re-encode polled batches; replying empty poll"
+ );
+ empty_poll_fallback(partition_id)
+ }),
+ other => {
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ reply_was_none = other.is_none(),
+ "partition read failed; replying empty poll"
+ );
+ Err(empty_poll_fallback(partition_id))
+ }
+ }
+}
+
+/// The fail-fast poll reply for a partition that could not answer: the
+/// 16-byte empty poll for `partition_id`, riding the re-sync sentinel
+/// channel when the id is the sentinel and the empty-frame channel
+/// otherwise.
+fn empty_poll_fallback(partition_id: u32) -> (Bytes, FrameChannel) {
+ let channel = if partition_id == RESYNC_REQUIRED_PARTITION_SENTINEL {
+ FrameChannel::ResyncSentinel
+ } else {
+ FrameChannel::EmptyFrame
+ };
+ (empty_polled_messages_body(partition_id), channel)
+}
+
+/// Serve `get_consumer_offset`. An empty status-0 body decodes as `None` on
+/// the SDK side (no offset stored / partition unknown), so it is reserved for
Review Comment:
**warning** — this docstring, rewritten by this commit, is contradicted by its own function body. It reserves the empty status-0 body for "no offset stored / partition unknown", but `:827` denies `PartitionNotFound` typed, so partition-unknown never takes that path. Meanwhile the case that DOES take it — an unresolved stream or topic — is not mentioned.
Fix the `:827` match (see the comment there) and the sentence becomes true as written. If the match is deferred instead, this needs to say plainly that an unresolved stream or topic reads as "no offset stored", with an issue link — a documented gap is defensible, this sentence is not.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -188,50 +137,46 @@ where
{
let shard_handle = Rc::clone(shard_handle);
let sessions = Rc::clone(sessions);
- let queues: ClientRequestQueues = Rc::new(RefCell::new(HashMap::new()));
- let active: ActiveClientRequests = Rc::new(RefCell::new(HashSet::new()));
+ let queues: ClientRequestQueues = Rc::new(RefCell::new(AHashMap::new()));
+ let active: ActiveClientRequests = Rc::new(RefCell::new(AHashSet::new()));
+ let queues_for_disconnect = Rc::clone(&queues);
+ let active_for_disconnect = Rc::clone(&active);
let sessions_for_disconnect = Rc::clone(&sessions);
let shard_handle_for_disconnect = Rc::clone(&shard_handle);
let bus_for_spawn = (*bus).clone();
bus.set_client_connection_lost_fn(Rc::new(move |client_id| {
- if let Some((vsr_client_id, session)) = sessions_for_disconnect
- .borrow_mut()
- .remove_connection(client_id)
- && let Some(shard) = upgrade_shard_handle(&shard_handle_for_disconnect)
+ // The socket is gone: nothing will drain what is still queued and no
+ // later frame will release the active slot, so both entries go here.
+ // This is also the only recovery from a panic inside the drain task
+ // (compio catches it), which would otherwise leave the slot taken and
+ // every later frame for the client queued forever.
+ queues_for_disconnect.borrow_mut().remove(&client_id);
+ active_for_disconnect.borrow_mut().remove(&client_id);
Review Comment:
**nit** — clearing `active` from the hook breaks the one-drain-per-client invariant this handler's doc promises at `:118-123`. The hook can fire while drain #1 is suspended at an `.await` inside `handle_client_request`, so the post-hook `enqueue_client_request` (see the `:311` comment for why one can still arrive) wins `active.insert` at `:244` and spawns drain #2 concurrently. Two drains then pop the same `VecDeque`.
Impact is limited and I want to be accurate about it: because this closure is synchronous with no await between here and `remove_connection` at `:161`, the session is already stripped, and the hook fires FROM `remove_client_meta`, so `ensure_transport_connection` finds nothing and `classify(header, false)` routes every replicated op to `UnboundReplicated`. Drain #2 therefore commits nothing — it only emits `Eviction(NoSession)` frames at a dead socket. No op loss, no reordering.
Fix: release the slot from a scopeguard inside the spawned drain task instead. That still covers the panic case this line was written for, without racing the hook.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -681,353 +532,193 @@ async fn handle_client_request<B, MJ, S, SB>(
IggyError::Unauthenticated.as_code(),
)
.await;
- return;
}
- handle_non_replicated_request(shard, sessions, system_config, transport_client_id, request)
- .await;
- return;
- }
-
- if header.operation == Operation::Register && header.session == 0 && header.request == 0 {
- handle_login_register_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- if header.operation == Operation::Logout {
- handle_logout_request(shard, sessions, transport_client_id, request).await;
- return;
- }
-
- let bound = sessions.borrow().get_session(transport_client_id);
- if bound.is_none() {
- // Replicated request on an unbound transport. Without this short-
- // circuit, the rewrite below overwrites `header.client` with
- // `transport_client_id` and dispatches; the request_preflight then
- // rejects with `NoSession`/`Fenced` and the failure disappears
- // silently, wedging the SDK until the socket timeout. A typed
- // `Eviction(NoSession)` is right here, unlike the pre-auth read
- // guard above: a replicated request implies the client believes it
- // has a session, and that session is gone, so it must register
- // again. An empty status-0 Reply is not safe here, because
- // SendMessages is the one replicated operation without a result
- // section, and its decoder would read the empty body as a
- // successful send.
- warn!(
- transport_client_id,
- operation = ?header.operation,
- "rejecting replicated request from unbound transport with Eviction(NoSession)"
- );
- send_unauthenticated_eviction(shard, transport_client_id).await;
- return;
- }
-
- // DeleteSegments is neither a partition nor a metadata consensus op: the
- // owning shard resolves the requested count to a concrete offset, then a
- // `TruncatePartition` is replicated through metadata (Option A). Each
- // replica's reconciler trims to the committed watermark. Handle it here,
- // ahead of the partition/metadata routing below.
- if header.operation == Operation::DeleteSegments {
- handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
- return;
- }
-
- if header.operation.is_partition() {
- // `bound` is Some here (unbound transports returned above).
- let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
- // `get_session` discards the acting user id the partition gate needs;
- // resolve it from the same bound connection. A bound transport always
- // has one, but the gate fails closed on `None` rather than trust that.
- let acting_user_id = sessions.borrow().get_user_id(transport_client_id);
- dispatch_partition_request(
- shard,
- request,
- vsr_client_id,
- bound_session,
- transport_client_id,
- acting_user_id,
- )
- .await;
- return;
- }
-
- let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
- *new_header = header;
- // Metadata-plane ops route by operation: stamp the sentinel group.
- new_header.group = server_common::sharding::METADATA_GROUP;
- // `bound` is always Some here (unbound transports early-return above);
- // this sets the consensus client id + session for the replicated op.
- if let Some((bound_client_id, bound_session)) = bound {
- new_header.client = bound_client_id;
- new_header.session = bound_session;
- }
- });
- let (request, raw_pat_token) = match maybe_rewrite_pat_request(
- sessions,
- transport_client_id,
- max_tokens_per_user,
- |user_id| {
- shard
- .plane
- .metadata()
- .mux_stm
- .users()
- .read(|users| users.pat_count_of(user_id))
- },
- request,
- ) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Token cap reached, malformed body, or a lost session binding.
- send_pre_consensus_deny(
+ RequestClass::NonReplicatedRead => {
+ // The auth-bypass guard is `classify`'s `UnauthenticatedRead` class:
+ // `PING`, the liveness probe, is the only pre-auth code, on every
+ // roster shape. `GET_CLUSTER_METADATA` describes the private replica
+ // network and is not something an unauthenticated caller gets to
+ // read; a client that dialed a backup no longer needs it to find the
+ // leader, because the backup authenticates the login locally and
+ // forwards only the consensus proposal
+ // (`submit_register_local_or_forward`). Every other non-replicated
+ // code MUST go through Register first, which binds the acting user
+ // the per-op authz gates resolve.
+ handle_non_replicated_request(
shard,
- &header,
+ sessions,
+ system_config,
transport_client_id,
- &error,
- "personal-access-token",
+ request,
+ (user_id, client_address),
)
.await;
- return;
}
- };
- // Hash raw passwords and, for ChangePassword, verify the current password
- // on the primary before replication; see `crate::users`. Replicas store the
- // hash directly. A wrong current password is not denied here: it rides
- // consensus and applies as a committed InvalidCredentials no-op, so the only
- // Err returned is a malformed body.
- let request = match maybe_rewrite_user_password_request(shard, request) {
- Ok(rewritten) => rewritten,
- Err(error) => {
- // Malformed body: deny fast with InvalidCommand.
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "user-password")
- .await;
- return;
+ RequestClass::LoginRegister => {
+ handle_login_register_request(shard, sessions, transport_client_id, request).await;
}
- };
- // Static bounds run pre-consensus so a rejected request burns no
- // replicated log entry; HTTP covers the same bounds via
- // `command.validate()`. A body that fails to decode denies typed too
- // (`InvalidCommand`), instead of riding consensus just to fail there.
- let bounds = match header.operation {
- Operation::CreateTopic => CreateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_topic| {
- // `parse` doubles as the catalog gate: an unknown key or a
- // malformed value denies typed here, pre-consensus.
- let options = TopicCreateOptions::parse(&create_topic.options)?;
- if let Some(segment_size) = options.segment_size {
- validate_topic_segment_size(
- segment_size.as_bytes_u64(),
- iggy_common::MAX_TOPIC_SEGMENT_SIZE,
- )?;
- }
- let segment_size = options.segment_size.map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- if options
- .preallocate_segments
- .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS)
- {
- validate_preallocated_topic_bytes(segment_size, create_topic.partitions_count)?;
- }
- let max_topic_size = options
- .max_topic_size
- .unwrap_or(MaxTopicSize::ServerDefault);
- validate_topic_bounds(create_topic.partitions_count, max_topic_size, segment_size)?;
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- create_topic.partitions_count,
- );
- Ok(())
- }),
- Operation::CreatePartitions => CreatePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_partitions| {
- validate_partitions_change_count(create_partitions.partitions_count)?;
- let metadata = shard.plane.metadata();
- warn_unenforceable_topic_size_on_partition_add(
- metadata.mux_stm.streams(),
- &create_partitions.stream_id,
- &create_partitions.topic_id,
- shard.bus_max_message_size(),
- create_partitions.partitions_count,
- );
- Ok(())
- }),
- Operation::DeletePartitions => DeletePartitionsRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|delete_partitions| {
- validate_partitions_change_count(delete_partitions.partitions_count)
- }),
- // Only the updatable subset: the create-time knobs are pushed to
- // partitions when the topic is built and nothing re-pushes them, so
- // accepting one here would store a value no partition ever sees.
- Operation::UpdateTopic => UpdateTopicRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_topic| {
- validate_option_keys(&update_topic.options, UPDATABLE_TOPIC_OPTION_KEYS)?;
- let options = TopicCreateOptions::parse(&update_topic.options)?;
- let Some(max_topic_size) = options.max_topic_size else {
- return Ok(());
- };
- // An update can lower the cap below one segment just as a
- // create can, and the stored map would then report a size the
- // topic can never enforce. The floor is this topic's own
- // segment size, since that key is create-only.
- let metadata = shard.plane.metadata();
- let streams = metadata.mux_stm.streams();
- let segment_size = streams
- .topic_segment_size(&update_topic.stream_id, &update_topic.topic_id)
- .map_or_else(
- || iggy_common::DEFAULT_SEGMENT_SIZE,
- |segment_size| segment_size.as_bytes_u64(),
- );
- validate_topic_size_floor(max_topic_size, segment_size)?;
- let partitions_count = streams
- .topic_partitions_count(&update_topic.stream_id, &update_topic.topic_id)
- .unwrap_or(0);
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size,
- shard.bus_max_message_size(),
- u32::try_from(partitions_count).unwrap_or(u32::MAX),
- );
- Ok(())
- }),
- Operation::UpdateStream => UpdateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_stream| {
- validate_option_keys(&update_stream.options, UPDATABLE_STREAM_OPTION_KEYS)
- }),
- Operation::UpdateUser => UpdateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|update_user| {
- validate_option_keys(&update_user.options, UPDATABLE_USER_OPTION_KEYS)
- }),
- Operation::CreateStream => CreateStreamRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_stream| validate_option_keys(&create_stream.options, &[])),
- Operation::CreateUser => CreateUserRequest::decode_from(request_body(&request))
- .map_err(|_| IggyError::InvalidCommand)
- .and_then(|create_user| validate_option_keys(&create_user.options, &[])),
- _ => Ok(()),
- };
- if let Err(error) = bounds {
- send_pre_consensus_deny(shard, &header, transport_client_id, &error, "static-bounds").await;
- return;
- }
- // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
- // partition count for Join) before replication; see `crate::consumer_group`.
- let request = match maybe_rewrite_consumer_group_request(shard, request).await {
- Ok(rewritten) => rewritten,
- Err(error) => {
+ RequestClass::Logout => {
+ handle_logout_request(shard, sessions, transport_client_id, request).await;
+ }
+ RequestClass::UnboundReplicated => {
+ // Replicated request on an unbound transport. Without this short-
+ // circuit, the rewrite below overwrites `header.client` with
+ // `transport_client_id` and dispatches; the request_preflight then
+ // rejects with `NoSession`/`Fenced` and the failure disappears
+ // silently, wedging the SDK until the socket timeout. A typed
+ // `Eviction(NoSession)` is right here, unlike the plain deny of
+ // `UnauthenticatedRead`: a replicated request implies the client
+ // believes it has a session, and that session is gone, so it must
+ // register again. An empty status-0 Reply is not safe here, because
+ // SendMessages is the one replicated operation without a result
+ // section, and its decoder would read the empty body as a
+ // successful send.
warn!(
transport_client_id,
- error = %error,
operation = ?header.operation,
- "dropping consumer-group request with invalid payload"
+ "rejecting replicated request from unbound transport with Eviction(NoSession)"
);
- return;
+ // The eviction context is best-effort off the metadata consensus
+ // (peer shards have none; zeroes are cosmetic -- the SDK only
+ // reads the reason), and the evicted id is the transport id: no
+ // VSR session exists to name.
+ send_eviction(
+ shard,
+ transport_client_id,
+ transport_client_id,
+ EvictionReason::NoSession,
+ "unbound replicated request",
+ )
+ .await;
}
- };
- let request_header = *request.header();
- // Replicated request: run consensus on the metadata owner (shard 0) and
- // bring the committed reply back here. This shard owns the connection,
- // so it writes the reply to the socket via the transport client id --
- // shard 0 can't route by the consensus client id (no home-shard bits).
- match submit_client_request_on_owner(shard, request).await {
- Some(reply) => {
- // The raw PAT token never enters consensus (it is non-deterministic
- // and secret), so the committed reply body is empty. Substitute the
- // raw-token response here, on the minting client's home shard, using
- // the confirmed commit position from the committed reply.
- let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) {
- Ok(reply) => reply,
+ RequestClass::DeleteSegments => {
+ // DeleteSegments is neither a partition nor a metadata consensus op: the
+ // owning shard resolves the requested count to a concrete offset, then a
+ // `TruncatePartition` is replicated through metadata (Option A). Each
+ // replica's reconciler trims to the committed watermark. `classify`
+ // names it ahead of the partition and metadata classes.
+ handle_delete_segments_request(shard, transport_client_id, bound, &request).await;
+ }
+ RequestClass::Partition => {
+ // `bound` is Some here: `classify` sends unbound transports to
+ // `UnboundReplicated`.
+ let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0));
+ // The acting user comes from the prologue's lookup. A bound
+ // transport always has one, but the gate below fails closed on
+ // `None` rather than trust that.
+ dispatch_partition_request(
+ shard,
+ request,
+ vsr_client_id,
+ bound_session,
+ transport_client_id,
+ user_id,
+ )
+ .await;
+ }
+ RequestClass::ReplicatedMetadata => {
+ // The one arm that needs the by-value copy: the rewrite below
+ // stamps the consensus client / session / group over the header,
+ // and a pre-consensus deny still has to echo what the client sent.
+ let header = *header;
+ let request =
+ request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
+ *new_header = header;
+ // Metadata-plane ops route by operation: stamp the sentinel group.
+ new_header.group = server_common::sharding::METADATA_GROUP;
+ // `bound` is always Some here (`classify` sends unbound transports to
+ // `UnboundReplicated`); this sets the consensus client id + session
+ // for the replicated op.
+ if let Some((bound_client_id, bound_session)) = bound {
+ new_header.client = bound_client_id;
+ new_header.session = bound_session;
+ }
+ });
+ let (request, raw_pat_token) = match tcp_chain(
+ shard,
+ sessions,
+ transport_client_id,
+ max_tokens_per_user,
+ request,
+ ) {
+ Ok(rewritten) => rewritten,
+ Err(RewriteDeny { stage, error }) => {
+ send_pre_consensus_deny(shard, transport_client_id, &header, &error, stage)
+ .await;
+ return;
+ }
+ };
+ // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
+ // partition count for Join) before replication; see `crate::consumer_group`.
+ let request = match maybe_rewrite_consumer_group_request(shard, request).await {
+ Ok(rewritten) => rewritten,
Err(error) => {
- warn!(
+ // Both of the rewrite's own failures are `InvalidCommand`
+ // decode errors, so a replay cannot help: deny typed
+ // instead of leaving the lockstep connection to its read
+ // timeout. (Its third error path needs a body past
+ // `u32::MAX` against a 64 MiB message cap, so no client
+ // frame reaches it; the deny is correct there too.)
+ send_pre_consensus_deny(
+ shard,
transport_client_id,
- error = %error,
- "failed to build raw PAT reply"
- );
+ &header,
+ &error,
+ RewriteStage::ConsumerGroup,
+ )
+ .await;
return;
}
};
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- operation = ?header.operation,
- "failed to deliver committed reply to client"
- );
+ let request_header = *request.header();
+ // Replicated request: run consensus on the metadata owner (shard 0) and
+ // bring the committed reply back here. This shard owns the connection,
+ // so it writes the reply to the socket via the transport client id --
+ // shard 0 can't route by the consensus client id (no home-shard bits).
+ match submit_client_request_on_owner(shard, request).await {
+ Some(reply) => {
+ // The raw PAT token never enters consensus (it is non-deterministic
+ // and secret), so the committed reply body is empty. Substitute the
+ // raw-token response here, on the minting client's home shard, using
+ // the confirmed commit position from the committed reply.
+ let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) {
+ Ok(reply) => reply,
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to build raw PAT reply"
+ );
+ // The op COMMITTED; only the reply could not be
+ // rendered. A typed deny is still the right frame:
+ // silence wedges the lockstep connection on a
+ // request that succeeded server-side, and the
+ // client can read the minted token back.
Review Comment:
**nit** — "the client can read the minted token back" is false, so the comment justifies the right frame with a recovery path that does not exist. `PersonalAccessTokenResponse` carries `name` + `expiry_at` only (`core/binary_protocol/src/responses/personal_access_tokens/get_personal_access_tokens.rs:30-33`, `core/common/src/types/permissions/personal_access_token.rs:35-40`); the raw secret exists in that one reply and nowhere else. A retry hits `ClientTable` dedup and replays the empty committed body, so it is unrecoverable and the caller must delete and re-create.
The deny is still correct — silence on a committed op is worse. Just drop the last clause.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -1176,8 +1186,7 @@ pub(in crate::dispatch) async fn handle_delete_segments_request<B, MJ, S, SB>(
/// truncate commits under. A resolvable namespace with nothing sealed to delete
/// still yields a `TruncatePartition(up_to_offset = 0)` so the metadata request
/// sequence stays contiguous. `Err` on a malformed body or an unresolved
-/// namespace: the TCP caller drops it to a silent replay, the HTTP caller renders
-/// the error.
+/// namespace: the TCP caller denies typed, the HTTP caller renders the error.
Review Comment:
**nit** — two stale rustdoc sentences around the lines this commit edited.
Here: "`Err` on a malformed body or an unresolved namespace" — an unresolved namespace returns `Ok(build_truncate_partition_client_message_with_identifiers(..))` at `:1223-1230`. The only `Err` values are `InvalidCommand` and `TransientNotAccepted`.
And at `:1075-1076`: "Only a malformed / unresolvable request is acked empty without a commit" — now false on both halves, since malformed denies typed at `:1143` and unresolvable returns `Ok`.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -188,50 +137,46 @@ where
{
let shard_handle = Rc::clone(shard_handle);
let sessions = Rc::clone(sessions);
- let queues: ClientRequestQueues = Rc::new(RefCell::new(HashMap::new()));
- let active: ActiveClientRequests = Rc::new(RefCell::new(HashSet::new()));
+ let queues: ClientRequestQueues = Rc::new(RefCell::new(AHashMap::new()));
+ let active: ActiveClientRequests = Rc::new(RefCell::new(AHashSet::new()));
+ let queues_for_disconnect = Rc::clone(&queues);
+ let active_for_disconnect = Rc::clone(&active);
let sessions_for_disconnect = Rc::clone(&sessions);
let shard_handle_for_disconnect = Rc::clone(&shard_handle);
let bus_for_spawn = (*bus).clone();
bus.set_client_connection_lost_fn(Rc::new(move |client_id| {
- if let Some((vsr_client_id, session)) = sessions_for_disconnect
- .borrow_mut()
- .remove_connection(client_id)
- && let Some(shard) = upgrade_shard_handle(&shard_handle_for_disconnect)
+ // The socket is gone: nothing will drain what is still queued and no
+ // later frame will release the active slot, so both entries go here.
+ // This is also the only recovery from a panic inside the drain task
+ // (compio catches it), which would otherwise leave the slot taken and
+ // every later frame for the client queued forever.
+ queues_for_disconnect.borrow_mut().remove(&client_id);
+ active_for_disconnect.borrow_mut().remove(&client_id);
+ // Upgrade FIRST: `remove_connection` strips the `SessionManager`
+ // entry, so running it ahead of a failed upgrade would drop the
+ // binding without ever submitting the replicated `Logout`, leaking
+ // the `ClientTable` entry and its consumer-group memberships. The
+ // window is pre-build / post-runtime-drop only.
+ if let Some(shard) = upgrade_shard_handle(&shard_handle_for_disconnect)
+ && let Some((vsr_client_id, session)) = sessions_for_disconnect
Review Comment:
**nit** — upgrading before `remove_connection` is the right trade (a leaked local `SessionManager` entry beats a leaked replicated `ClientTable` entry), but the failure is now silent: if the upgrade fails, `connections` and `client_to_connection` keep their entries and nothing says so. Worth noting the heartbeat verifier does not always reap it — it is gated on `config.heartbeat.enabled` (`boot/mod.rs:806`) and `collect_stale` only considers `Bound`/`Authenticated` (`session_manager.rs:196-199`), so a `Connected` leak is never reaped. Window is pre-build / post-runtime-drop only, so this is about observability, not incidence. Fix: `error!` on the upgrade-failed branch.
##########
core/server/src/boot/threads.rs:
##########
@@ -406,6 +560,22 @@ pub(in crate::boot) fn validate_sharding_runtime_knobs(
drain: drain_timeout,
});
}
+ let join_timeout = sharding.shutdown_join_timeout.get_duration();
Review Comment:
**nit** — the restored `join >= drain` and `join <= MAX` checks match `ShardingConfig::validate` exactly (`core/configs/src/server_config/sharding.rs:251-267`), but the doc at `:519-522` claims this fn "Mirrors `ShardingConfig::validate`" and one clause is still missing: `reconcile_periodic_interval`, which `validate` rejects at zero and above `RECONCILE_PERIODIC_INTERVAL_MAX` (`sharding.rs:269-287`).
That one is not cosmetic for a caller that bypasses `validate`: zero feeds `run_reconciler`'s `ctx.shard.bus.sleep(periodic)` inside an unconditional loop (`core/server/src/partition_reconciler.rs:338-355`), so the sleep arm is ready every iteration and `reconcile_once` runs back-to-back, starving the pump on that core. Add the two checks, or narrow the doc to the knobs it actually mirrors.
##########
core/server/src/dispatch/authz.rs:
##########
@@ -328,46 +261,19 @@ where
|request| (&request.stream_id, &request.topic_id),
Permissioner::get_consumer_groups,
),
- _ => Ok(()),
- }
-}
-
-/// Reply to a denied non-replicated read with the request's reply frame: empty
-/// body + nonzero `status`. The SDK peeks the status before body decode and
-/// surfaces the typed error, so a poll denial never reaches the empty-poll
-/// "0 messages" body path.
-#[allow(clippy::future_not_send)]
-pub(in crate::dispatch) async fn send_non_replicated_deny<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: &Message<RoutedRequestHeader>,
- transport_client_id: u128,
- status: u32,
-) where
- B: ShellBus,
- MJ: JournalHandle + 'static,
- MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
- S: 'static,
- SB: SuperblockStore + 'static,
-{
- let commit = current_metadata_commit(shard);
- let reply = build_deny_reply(
- request.header(),
- request.header().client,
- request.header().session,
- commit,
- status,
- );
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- status,
- error = %error,
- "failed to surface non-replicated authz denial"
- );
+ // No on-demand flush primitive exists. This arm is the only thing
+ // answering `FeatureUnavailable` for flush now: the HTTP path reaches
+ // the builder through `read_local`, whose call sites pass eleven fixed
Review Comment:
**nit** — the corrected rationale is true today but rests on a cross-module count nothing enforces: "whose call sites pass eleven fixed read codes, none of them flush". I counted eleven `read_local` sites too (`core/server/src/http/handlers.rs:286,310,344,378,415,443,481,522,562,590,1718`), so it is accurate — but a twelfth site silently falsifies it, and the reader has no link to check. Prefer stating the local requirement (flush has no HTTP route, so this arm is the only source of `FeatureUnavailable` for it) without the number.
##########
core/server/src/dispatch/failure.rs:
##########
@@ -0,0 +1,677 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The wire failure channels and the one send exit for host-built frames.
+//!
+//! Every frame the dispatch host builds, success or rejection, leaves through
+//! [`send_host_frame`], so the send-failure log has one shape. Which channel a
+//! failure rides is a wire contract with the SDK:
+//!
+//! | channel | carrier | when |
+//! |---|---|---|
+//! | [`FrameChannel::TypedDeny`] | Reply, nonzero status + empty body, or a result-framed rejection body | rejections that must unblock the SDK's lockstep request slot: checksum, authz, pre-consensus rewrite, unknown or unsupported non-replicated code, unbound non-PING read, transient replay hints |
+//! | [`FrameChannel::Eviction`] | session-terminal Eviction frame with a typed reason | the client must register again: `NoSession`, `MalformedLogin`, heartbeat and login evictions. The reason rides the channel label, since one `context` covers four of them |
+//! | [`FrameChannel::ResyncSentinel`] | status-0 poll reply, body carries `RESYNC_REQUIRED_PARTITION_SENTINEL` | a fenced consumer-group poll: the consumer must re-sync its assignment; HTTP mirrors it as `resync_required_polled_messages` in `crate::http::wire` |
+//! | [`FrameChannel::EmptyFrame`] | status-0 fail-fast body, empty or the 16-byte empty poll | the partition cannot answer yet; the SDK fails fast (empty poll) and retries. A permanent client error never rides this channel: an undecodable body denies typed, because there is nothing to retry |
+//! | [`FrameChannel::Reply`] | status-0 success frame | host-built success replies: login/register, ping, logout, non-replicated read bodies, committed metadata replies |
+//! | silent drop | no frame | one deliberate case, a transient consensus submit failure: the SDK read-timeout replays the same request id, and a synthesized failure could contradict a write that commits moments later. A header `RequestHeader::validate` rejected also drops, but that one is a GAP, not a contract - the fields decode, so a deny could be echoed under the transport id, and the client instead waits out its read timeout |
+//! | HTTP status | HTTP status code | the HTTP spine maps the same rejections in `crate::http::error`; it never rides these frames |
+//!
+//! The last two send nothing, so [`FrameChannel`] has no variant for them.
+//!
+//! Scope: the table covers `crate::dispatch` only. Two neighbours answer on
+//! their own paths by design - the partitions engine builds and sends
+//! produce/poll replies, and the shard crate builds client-shaped denies of
+//! its own (`IggyShard::deny_partition_request_transient` and
+//! `stage_transient_deny`, both `TypedDeny`-shaped, the latter shedding the
+//! frame outright when its lifecycle queue is full).
+
+use crate::responses::{
+ NonReplicatedResponse, build_deny_reply, build_empty_reply, current_metadata_commit,
+};
+use crate::rewrite::RewriteStage;
+use crate::shell::{ShellBus, ShellShard};
+use bytes::Bytes;
+use consensus::{
+ EvictionContext, MetadataHandle, build_eviction_message,
+ build_incompatible_protocol_eviction_message, build_result_rejection_reply,
+};
+use iggy_binary_protocol::{EvictionReason, PrepareHeader, RoutedRequestHeader};
+use iggy_common::IggyError;
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use message_bus::BusMessage;
+use server_common::Message;
+use std::rc::Rc;
+use tracing::warn;
+
+/// Labels the channel a host-built frame rides, for the send-failure log.
+/// The taxonomy, including the two channels that never construct a frame,
+/// is on the module doc.
+#[derive(Clone, Copy, Debug)]
+pub(in crate::dispatch) enum FrameChannel {
+ TypedDeny,
+ /// The reason travels with the channel: five call sites share the
+ /// `"login_rejection"` context across four distinct reasons, so
+ /// `context` alone cannot tell a `MalformedLogin` send failure from an
+ /// `InvalidCredentials` one.
+ Eviction(EvictionReason),
+ ResyncSentinel,
+ EmptyFrame,
+ Reply,
+}
+
+impl std::fmt::Display for FrameChannel {
+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::TypedDeny => formatter.write_str("typed_deny"),
+ Self::Eviction(reason) => write!(formatter, "eviction({reason:?})"),
+ Self::ResyncSentinel => formatter.write_str("resync_sentinel"),
+ Self::EmptyFrame => formatter.write_str("empty_frame"),
+ Self::Reply => formatter.write_str("reply"),
+ }
+ }
+}
+
+/// The one send exit for host-built client frames. Best-effort: a failed
+/// send means the connection is gone (or its queue is full), and there is
+/// nothing left to reply on, so the error is logged and dropped. `frame` is
+/// any bus message: a contiguous frozen frame or the vectored poll reply.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_host_frame<B: ShellBus>(
+ bus: &B,
+ transport_client_id: u128,
+ frame: impl Into<BusMessage>,
+ channel: FrameChannel,
+ context: &'static str,
+) {
+ if let Err(send_error) = bus.send_to_client(transport_client_id, frame).await {
+ warn!(
+ transport_client_id,
+ error = %send_error,
+ channel = %channel,
+ context,
+ "failed to send host frame to client"
+ );
+ }
+}
+
+/// Reply to a request rejected before it reached its plane with the request's
+/// own frame: empty body + nonzero `status`. The nonzero status is the whole
+/// point: the SDK peeks it and surfaces the typed error, whereas a status-0
+/// frame reads as a committed ack for work that never happened. Silence is no
+/// better, the connection decodes replies in lockstep and would wedge on every
+/// later request.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_deny_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ status: u32,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(request_header, transport_client_id, 0, commit, status);
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ "request_denial",
+ )
+ .await;
+}
+
+/// Deny a request from an unbound transport without disclosing the metadata
+/// commit frontier. The status is the only field a pre-authenticated caller
+/// needs, while the live commit would expose cluster write activity.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_unbound_deny_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ status: u32,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let reply = build_deny_reply(request_header, transport_client_id, 0, 0, status);
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ "unbound_request_denial",
+ )
+ .await;
+}
+
+/// Reply to a denied non-replicated read with the request's reply frame: empty
+/// body + nonzero `status`. The SDK peeks the status before body decode and
+/// surfaces the typed error, so a poll denial never reaches the empty-poll
+/// "0 messages" body path.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_non_replicated_deny<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ request: &Message<RoutedRequestHeader>,
+ transport_client_id: u128,
+ status: u32,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(
+ request.header(),
+ request.header().client,
+ request.header().session,
+ commit,
+ status,
+ );
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ "non_replicated_denial",
+ )
+ .await;
+}
+
+/// Reject a request before it reaches consensus: warn, then send the typed
+/// deny reply. A silent drop would wedge every later request on the
+/// connection until the socket read timeout. `stage` names the chain step
+/// for both log lines, so the set stays enumerable.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_pre_consensus_deny<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ error: &IggyError,
+ stage: RewriteStage,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let context = stage.as_str();
+ warn!(
+ transport_client_id,
+ error = %error,
+ operation = ?request_header.operation,
+ context,
+ "denying request pre-consensus"
+ );
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(
+ request_header,
+ transport_client_id,
+ 0,
+ commit,
+ error.as_code(),
+ );
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ context,
+ )
+ .await;
+}
+
+/// Result-framed rejection Reply: status 0, body `[count=1][index][code]`.
+/// The SDK decodes the nonzero result code, so a transient code makes it
+/// replay the same request at once instead of waiting out its read timeout.
+/// Replying empty instead would surface as a hard `InvalidFormat` decode
+/// failure and break the replay.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_result_rejection<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ error: &IggyError,
+ context: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = build_result_rejection_reply(request_header, commit, error.as_code());
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ context,
+ )
+ .await;
+}
+
+/// Best-effort session-terminal `Eviction` frame: the client's session is
+/// gone (or was never granted), so it must register again. Every frame
+/// transport decodes `Command::Eviction` and maps the typed reason
+/// (`NoSession` -> `Unauthenticated`, ...), so clients fail fast with the
+/// real cause instead of a body-decode failure or a timeout. Consensus
+/// context (cluster/view/replica) is stamped on the metadata shard and
+/// zeroed elsewhere; the SDK only reads the reason, plus the protocol
+/// window on `IncompatibleProtocol`.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_eviction<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ vsr_client_id: u128,
+ reason: EvictionReason,
+ context: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let ctx = shard.plane.metadata().consensus.as_ref().map_or(
+ EvictionContext {
+ cluster: 0,
+ view: 0,
+ replica: 0,
+ },
+ EvictionContext::from_consensus,
+ );
+ let eviction = match reason {
+ EvictionReason::IncompatibleProtocol => {
+ build_incompatible_protocol_eviction_message(ctx, vsr_client_id)
+ }
+ _ => build_eviction_message(ctx, vsr_client_id, reason),
+ };
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ eviction.into_generic().into_frozen(),
+ FrameChannel::Eviction(reason),
+ context,
+ )
+ .await;
+}
+
+/// Send a non-replicated reply body to a client, stamping the current
+/// metadata commit. Shared by the non-replicated read arms; `channel`
+/// labels the body shape (a real answer, a fail-fast empty poll, or the
+/// re-sync sentinel) for the send-failure log.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_non_replicated_bytes<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ request: &Message<RoutedRequestHeader>,
+ transport_client_id: u128,
+ bytes: Bytes,
+ channel: FrameChannel,
+ context: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let commit = current_metadata_commit(shard);
+ let reply = NonReplicatedResponse::Bytes(bytes).into_reply(
+ request.header(),
+ request.header().client,
+ request.header().session,
+ commit,
+ );
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ channel,
+ context,
+ )
+ .await;
+}
+
+/// Ack a consumer-offset op whose body could not be rewritten for the
+/// partition plane with an empty Reply. The SDK connection processes replies
+/// in lockstep, so a silent drop wedges every subsequent request on that
+/// connection.
+#[allow(clippy::future_not_send)]
+pub(in crate::dispatch) async fn send_empty_partition_reply<B, MJ, S, SB>(
Review Comment:
**nit** — `send_empty_partition_reply` appears to have no reachable producer. Its only caller is the `Err` arm of `maybe_rewrite_consumer_offset_request` (`partition.rs:472-483`), but `resolve_partition_request_namespace` already decodes the same bytes with the same `decode_from` earlier at `partition.rs:377` and denies typed at `:396-405` (`core/server/src/responses.rs:361-384`), so an undecodable body never reaches the rewrite. The second decode is on unmutated bytes, and `rewrite_request_body`'s only other error needs a body past `u32::MAX` against a 64 MiB cap.
If that holds, the `EmptyFrame` row's "empty" carrier documents a shape nothing produces. Worth confirming and then either deleting this helper or noting why it is kept.
--
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] refactor(server): make dispatch routing and failure paths explicit [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4036:
URL: https://github.com/apache/iggy/pull/4036#discussion_r3932145254
##########
core/server/src/boot/threads.rs:
##########
@@ -649,6 +827,188 @@ mod tests {
);
}
+ #[test]
+ fn peer_exit_countdown_releases_the_waiter_once_every_peer_is_out() {
+ let countdown = Arc::new(PeerExitCountdown::new(2));
+ let guards: [PeerExitGuard; 2] =
+ std::array::from_fn(|_| PeerExitGuard::new(Arc::clone(&countdown)));
+ assert_eq!(
+ countdown.wait(Duration::from_millis(10)),
+ Err(2),
+ "two live peers must hold the waiter past a short deadline"
+ );
+ let peers: Vec<_> = guards
+ .into_iter()
+ .map(|guard| {
+ thread::spawn(move || {
+ thread::sleep(Duration::from_millis(20));
+ drop(guard);
+ })
+ })
+ .collect();
+ assert_eq!(
+ countdown.wait(Duration::from_secs(30)),
+ Ok(()),
+ "the last guard drop must release the waiter"
+ );
+ for peer in peers {
+ peer.join()
+ .expect("peer thread dropped its guard without panicking");
+ }
+ }
+
+ #[test]
+ fn peer_exit_guard_counts_out_during_unwind() {
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let guard = PeerExitGuard::new(Arc::clone(&countdown));
+ // `resume_unwind` skips the panic hook, so the unwind is silent.
+ let unwound = panic::catch_unwind(panic::AssertUnwindSafe(|| {
+ let _guard = guard;
+ panic::resume_unwind(Box::new("peer shard body panicked"));
+ }));
+ assert!(unwound.is_err());
+ assert_eq!(
+ countdown.wait(Duration::ZERO),
+ Ok(()),
+ "a guard dropped by a panic unwind must still count its peer out"
+ );
+ }
+
+ #[test]
+ fn peer_exit_wait_flips_the_flag_before_waiting() {
+ // Stands in for a peer parked on its bus token: it exits only once
+ // the shutdown flag is set, so a waiter that set the flag after
+ // waiting would sit out the whole budget.
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let shutdown_flag = Arc::new(AtomicBool::new(false));
+ let peer = thread::spawn({
+ let guard = PeerExitGuard::new(Arc::clone(&countdown));
+ let shutdown_flag = Arc::clone(&shutdown_flag);
+ move || {
+ while !shutdown_flag.load(Ordering::Relaxed) {
+ thread::sleep(Duration::from_millis(1));
+ }
+ drop(guard);
+ }
+ });
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ countdown,
+ Arc::clone(&shutdown_flag),
+ Arc::new(ShutdownDeadline::new(Duration::from_secs(30))),
+ ));
+ assert!(
+ started.elapsed() < Duration::from_secs(5),
+ "the waiter must not sit out its budget on a peer that waits for the flag"
+ );
+ assert!(shutdown_flag.load(Ordering::Relaxed));
+ peer.join()
+ .expect("peer thread dropped its guard without panicking");
+ }
+
+ #[test]
+ fn peer_exit_wait_gives_up_at_the_deadline() {
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let _wedged_peer = PeerExitGuard::new(Arc::clone(&countdown));
+ let timeout = Duration::from_millis(50);
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ countdown,
+ Arc::new(AtomicBool::new(false)),
+ Arc::new(ShutdownDeadline::new(timeout)),
+ ));
+ let waited = started.elapsed();
+ assert!(
+ waited >= timeout && waited < Duration::from_secs(5),
+ "a wedged peer must be abandoned at the deadline, waited {waited:?}"
+ );
+ }
+
+ #[test]
+ fn peer_wait_and_shard_join_share_one_shutdown_budget() {
+ // Regression: the two waits nest (shard 0 cannot start waiting for
+ // its peers until its own drain returned, already inside the join
+ // budget). With a budget each, a clean shutdown reported shard 0 as
+ // wedged and exited non-zero.
+ let deadline = Arc::new(ShutdownDeadline::new(Duration::from_millis(200)));
+ let shutdown_flag = AtomicBool::new(true);
+ // Never finishes: stands in for the slow drain that arms and then
+ // spends the shared budget. The thread leaks into the test process.
+ let wedged_shard = thread::spawn(|| -> Result<(), ServerError> {
+ loop {
+ thread::sleep(Duration::from_secs(1));
+ }
Review Comment:
**nit** — this test spawns `thread::spawn(|| loop { thread::sleep(1s) })` with no stop path, so the thread leaks for the life of the test binary (the comment acknowledges it). Parked in `sleep`, so it costs stack rather than CPU. Note `:1143`'s pre-existing `abandons_a_shard_still_running_past_the_deadline` does the same thing, so this copied an existing pattern — worth an `AtomicBool` stop flag in both or neither, not just the new one.
##########
core/server/src/session_manager.rs:
##########
@@ -26,13 +26,30 @@
//! transport connection and the consensus-level `(client_id, session)` pair.
use crate::cluster_meta::ClusterRoster;
+use ahash::AHashMap;
use message_bus::installer::conn_info::ClientTransportKind;
use shard::ConnectedClientInfo;
-use std::collections::HashMap;
use std::net::SocketAddr;
use std::rc::Rc;
use std::time::{Duration, Instant};
+/// What the request funnel resolves from one `connections` lookup per frame.
+///
+/// The bound consensus session, the acting user, and the transport peer
+/// address. `Default` (everything absent, no address) stands for a
+/// connection neither this map nor the bus knows.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct ConnectionContext {
Review Comment:
**nit** — `ConnectionContext` is bare `pub` in `pub mod session_manager` (`core/server/src/lib.rs:49`), in the same commit that tightened `RequestClass`/`classify` to `pub(in crate::dispatch)`. Harmless (`server` is `publish = false`) but inconsistent. `pub(crate)` fits.
(`RewriteStage` looks like the same issue but is not — its `pub` is forced by `pub struct RewriteDeny { pub stage: RewriteStage }` in a `pub(crate)` module, so narrowing it trips `private_interfaces`.)
##########
core/server/src/dispatch/mod.rs:
##########
@@ -299,8 +245,19 @@ fn enqueue_client_request<B, MJ, S, SB>(
return;
}
- let bus = shard.bus.clone();
+ let shard_handle = Rc::clone(shard_handle);
+ let sessions = Rc::clone(sessions);
+ let system_config = Arc::clone(system_config);
Review Comment:
**nit, pre-existing, follow-up material** — recording the measurement next to the queue-cap item you deferred, since they belong in the same change.
A lockstep client pays one `compio::runtime::spawn` task allocation plus 4 `Rc::clone` and 1 `Arc::clone` (atomic RMW) per request, because `drain_client_requests` returns whenever the queue empties (`:290-303`) so `active.insert` wins again on the next frame and respawns. The boxed future holds ~512 B of headers live across the consensus await in the `ReplicatedMetadata` arm, so it is ~1 KB+ — an order of magnitude larger than the `VecDeque` allocation the `:311` change was made to avoid.
Not introduced here and not worth acting on alone. But it is the reason restoring the `:311` removal is the right trade: the allocation it reintroduces is the smaller half of a pair this path already pays. Folding `active` into the queue entry would recover both, and drop 2 of the 4 per-frame hash lookups.
--
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