You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/04 16:08:48 UTC
(iggy) branch master updated: refactor(server): make dispatch routing and failure paths explicit (#4036)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new be2265c9e refactor(server): make dispatch routing and failure paths explicit (#4036)
be2265c9e is described below
commit be2265c9ea132887a0b3414d43e431ed66d51cfd
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Sep 4 18:08:43 2026 +0200
refactor(server): make dispatch routing and failure paths explicit (#4036)
---
.../tests/tcp_test/offset_feature_deserialize.go | 9 +-
core/integration/tests/server/legacy_login_vsr.rs | 53 +-
core/integration/tests/server/mod.rs | 7 +
.../integration/tests/server/poll_semantics_vsr.rs | 57 +-
core/integration/tests/server/raw_tcp.rs | 207 +++
core/integration/tests/server/unknown_code_vsr.rs | 92 ++
core/metadata/src/impls/metadata.rs | 18 +-
core/server/src/boot/mod.rs | 126 +-
core/server/src/boot/recovery.rs | 27 +-
core/server/src/boot/threads.rs | 601 ++++++-
core/server/src/dispatch/authz.rs | 309 ++--
core/server/src/dispatch/failure.rs | 629 ++++++++
core/server/src/dispatch/mod.rs | 1638 +++++++++++---------
core/server/src/dispatch/partition.rs | 569 ++++---
core/server/src/dispatch/reads.rs | 101 +-
core/server/src/dispatch/session_ops.rs | 282 +---
core/server/src/dispatch/test_support.rs | 13 +-
core/server/src/http/handlers.rs | 13 +-
core/server/src/http/submit.rs | 23 +-
core/server/src/lib.rs | 1 +
core/server/src/responses.rs | 11 +-
core/server/src/rewrite.rs | 570 +++++++
core/server/src/server_error.rs | 21 +
core/server/src/session_manager.rs | 106 +-
core/simulator/src/replica.rs | 2 +-
25 files changed, 3923 insertions(+), 1562 deletions(-)
diff --git a/bdd/go/tests/tcp_test/offset_feature_deserialize.go b/bdd/go/tests/tcp_test/offset_feature_deserialize.go
index 28bb0b89c..229d200dd 100644
--- a/bdd/go/tests/tcp_test/offset_feature_deserialize.go
+++ b/bdd/go/tests/tcp_test/offset_feature_deserialize.go
@@ -21,6 +21,7 @@ import (
"context"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ ierror "github.com/apache/iggy/foreign/go/errors"
"github.com/onsi/ginkgo/v2"
"github.com/onsi/gomega"
)
@@ -175,7 +176,10 @@ var _ = ginkgo.Describe("GET CONSUMER OFFSET:", func() {
consumer := iggcon.NewGroupConsumer(randomU32Identifier())
partitionId := uint32(1)
- offset, err := client.GetConsumerOffset(
+ // An unresolved stream is an addressing error, not a fresh
+ // consumer: a nil offset would let a typo'd or deleted target read
+ // back as "no offset stored" and silently reprocess.
+ _, err := client.GetConsumerOffset(
context.Background(),
consumer,
randomU32Identifier(),
@@ -183,8 +187,7 @@ var _ = ginkgo.Describe("GET CONSUMER OFFSET:", func() {
&partitionId,
)
- itShouldNotReturnError(err)
- itShouldReturnNilOffsetForNewConsumerGroup(offset)
+ itShouldReturnSpecificError(err, ierror.ErrStreamIdNotFound)
})
})
diff --git a/core/integration/tests/server/legacy_login_vsr.rs b/core/integration/tests/server/legacy_login_vsr.rs
index 98b685e1b..f0a88c134 100644
--- a/core/integration/tests/server/legacy_login_vsr.rs
+++ b/core/integration/tests/server/legacy_login_vsr.rs
@@ -27,19 +27,15 @@
//! TCP socket: a header-only non-replicated frame carrying the code in the
//! reserved command slot.
-use iggy_binary_protocol::HEADER_SIZE;
+use iggy_binary_protocol::EvictionReason;
use iggy_binary_protocol::codes::{LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE};
-use iggy_binary_protocol::consensus::{Command, Operation, RequestHeader};
+use iggy_binary_protocol::consensus::Command;
use integration::harness::TestHarness;
use integration::iggy_harness;
-use std::mem::offset_of;
-use std::time::Duration;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::TcpStream;
-use tokio::time::timeout;
-// Wire byte pinned to `EvictionReason::MalformedLogin` in consensus::header.
-const EVICTION_REASON_MALFORMED_LOGIN: u8 = 15;
+use crate::server::raw_tcp::{
+ connect, eviction_reason, frame_command, non_replicated_header, read_frame_header, write_frame,
+};
#[iggy_harness]
async fn given_legacy_login_user_code_when_sent_raw_should_evict_malformed_login(
@@ -60,45 +56,26 @@ 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),
Command::Eviction as u8,
"expected an Eviction frame for legacy login code {code}, not a Reply"
);
assert_eq!(
- reply[HEADER_SIZE - 1],
- EVICTION_REASON_MALFORMED_LOGIN,
+ eviction_reason(&reply),
+ EvictionReason::MalformedLogin as u8,
"legacy login code {code} must evict with MalformedLogin"
);
}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 1c6411a8a..e32f2c5ab 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -21,9 +21,16 @@ mod a2a_jwt;
mod cg;
// Flush (FLUSH_UNSAVED_BUFFER) has no the server primitive; it must deny typed.
mod flush_vsr;
+// Raw TCP framing (connect, hand-crafted frames, root register) for the
+// server suites that send what the SDK cannot.
+pub(crate) mod raw_tcp;
// Legacy login codes (LOGIN_USER / LOGIN_WITH_PAT) have no the server handler;
// they must evict typed (MalformedLogin), not stall or reply empty-ok.
mod legacy_login_vsr;
+// A non-replicated code no read serves (unknown, or table-listed without an
+// arm) must deny typed (InvalidCommand) at the read gate, not stall or reply
+// empty-ok.
+mod unknown_code_vsr;
// A failed credential login must report the credential failure, not the
// payload shape it fell through to.
mod login_credentials_vsr;
diff --git a/core/integration/tests/server/poll_semantics_vsr.rs b/core/integration/tests/server/poll_semantics_vsr.rs
index a3a90f06d..243039742 100644
--- a/core/integration/tests/server/poll_semantics_vsr.rs
+++ b/core/integration/tests/server/poll_semantics_vsr.rs
@@ -19,7 +19,7 @@
//! topic does not have must surface a typed `PartitionNotFound`, not an empty
//! poll a consumer would read as end-of-partition; a poll whose stream or
//! topic does not resolve must surface the legacy `StreamIdNotFound` /
-//! `TopicIdNotFound` the same way; the partition addressing error on
+//! `TopicIdNotFound` the same way; the same three addressing errors on
//! `get_consumer_offset` must not decode as "no offset stored"; and a
//! timestamp poll must be at-or-after, including the message stamped exactly at
//! the queried timestamp (the timestamp replies report per message).
@@ -203,6 +203,61 @@ async fn given_missing_partition_when_getting_consumer_offset_should_reject_part
);
}
+/// The same swallow one level up: an unresolved STREAM or TOPIC also answered
+/// the empty body, so a typo'd or deleted target read back as a fresh
+/// consumer. The consumer then resumes from its configured default and
+/// silently reprocesses, with no error anywhere. The poll path denies both
+/// codes, and the identical read over REST 404s.
+#[iggy_harness(
+ test_client_transport = [Tcp]
+)]
+async fn given_missing_stream_when_getting_consumer_offset_should_reject_stream_not_found(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ let stream_id = Identifier::from_str_value("no-such-stream").expect("stream identifier");
+ let topic_id = Identifier::from_str_value("no-such-topic").expect("topic identifier");
+
+ let result = client
+ .get_consumer_offset(&Consumer::default(), &stream_id, &topic_id, Some(0))
+ .await;
+
+ let expected = IggyError::StreamIdNotFound(Identifier::default()).as_code();
+ assert!(
+ matches!(&result, Err(error) if error.as_code() == expected),
+ "get_consumer_offset on a missing stream must surface Err(StreamIdNotFound), \
+ got {result:?}"
+ );
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp]
+)]
+async fn given_missing_topic_when_getting_consumer_offset_should_reject_topic_not_found(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ client
+ .create_stream("offset-topicless-stream")
+ .await
+ .expect("create stream");
+ let stream_id =
+ Identifier::from_str_value("offset-topicless-stream").expect("stream identifier");
+ let topic_id = Identifier::from_str_value("no-such-topic").expect("topic identifier");
+
+ let result = client
+ .get_consumer_offset(&Consumer::default(), &stream_id, &topic_id, Some(0))
+ .await;
+
+ let expected =
+ IggyError::TopicIdNotFound(Identifier::default(), Identifier::default()).as_code();
+ assert!(
+ matches!(&result, Err(error) if error.as_code() == expected),
+ "get_consumer_offset on a missing topic of an existing stream must surface \
+ Err(TopicIdNotFound), got {result:?}"
+ );
+}
+
#[iggy_harness(
test_client_transport = [Tcp]
)]
diff --git a/core/integration/tests/server/raw_tcp.rs b/core/integration/tests/server/raw_tcp.rs
new file mode 100644
index 000000000..cfe5d8c69
--- /dev/null
+++ b/core/integration/tests/server/raw_tcp.rs
@@ -0,0 +1,207 @@
+// 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 {})",
+ eviction_reason(&reply_header)
+ );
+
+ let total_size = read_size_field(&reply_header).expect("reply size field") as usize;
+ // `read_size_field` is a bare 4-byte LE read with no floor (the SDK adds
+ // its own guard, which this helper cannot inherit), so an under-sized
+ // field would either panic on subtract overflow or wrap into a ~1.8e19
+ // allocation and abort - either way hiding the regression under test.
+ assert!(
+ total_size >= HEADER_SIZE,
+ "reply size field {total_size} is below the {HEADER_SIZE}-byte header"
+ );
+ let mut reply_body = vec![0u8; total_size - HEADER_SIZE];
+ timeout(REPLY_WAIT, stream.read_exact(&mut reply_body))
+ .await
+ .expect("reply body timed out")
+ .expect("reply body read failed");
+ (reply_header, reply_body)
+}
+
+pub(crate) fn frame_command(header: &[u8; HEADER_SIZE]) -> u8 {
+ header[offset_of!(RequestHeader, command)]
+}
+
+/// The typed reason byte of an `Eviction` frame, read by field offset rather
+/// than by a hardcoded index: `EvictionReason` is renumberable, so a literal
+/// would keep passing while checking a different reason.
+pub(crate) fn eviction_reason(header: &[u8; HEADER_SIZE]) -> u8 {
+ header[offset_of!(EvictionHeader, reason)]
+}
+
+pub(crate) fn reply_status(reply_header: &[u8; HEADER_SIZE]) -> u32 {
+ let offset = offset_of!(ReplyHeader, status);
+ u32::from_le_bytes(reply_header[offset..offset + 4].try_into().unwrap())
+}
+
+/// Root login/register on the socket as `client`, replayed on transient
+/// rejections until it commits; returns the bound session id. The session
+/// binds to THIS transport connection server-side, so a frame that must ride
+/// it has to reuse the stream.
+pub(crate) async fn register_root(stream: &mut TcpStream, client: u128) -> u64 {
+ let body = LoginRegisterRequest {
+ version_info: ClientVersionInfo {
+ protocol_version: IGGY_PROTOCOL_VERSION,
+ sdk_name: WireName::new("raw-tcp").unwrap(),
+ sdk_version: WireName::new("0.0.1").unwrap(),
+ },
+ username: WireName::new(DEFAULT_ROOT_USERNAME).unwrap(),
+ password: SecretString::from(DEFAULT_ROOT_PASSWORD),
+ client_context: None,
+ }
+ .to_bytes();
+ let header = request_header(Operation::Register, client, 0, 0, body.len());
+ let deadline = Instant::now() + COMMIT_BUDGET;
+ loop {
+ let (reply_header, reply_body) = exchange(stream, &header, &body).await;
+ // A pre-commit deny rides the status word; a committed verdict rides
+ // the result section that leads a Register reply body.
+ let code = match reply_status(&reply_header) {
+ 0 => result_code(&reply_body).expect("register reply must carry a result section"),
+ status => status,
+ };
+ if code == 0 {
+ let payload_start = result_section_len(&reply_body).unwrap();
+ let response = LoginRegisterResponse::decode_from(&reply_body[payload_start..])
+ .expect("register payload must decode");
+ assert_ne!(response.session, 0, "server must bind a nonzero session");
+ return response.session;
+ }
+ assert!(
+ is_transient(code),
+ "register rejected with a terminal code {code}"
+ );
+ assert!(
+ Instant::now() < deadline,
+ "register did not commit within {COMMIT_BUDGET:?}"
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+}
+
+fn is_transient(code: u32) -> bool {
+ code == IggyError::TransientNotCommitted.as_code()
+ || code == IggyError::TransientNotAccepted.as_code()
+}
diff --git a/core/integration/tests/server/unknown_code_vsr.rs b/core/integration/tests/server/unknown_code_vsr.rs
new file mode 100644
index 000000000..622e5bac1
--- /dev/null
+++ b/core/integration/tests/server/unknown_code_vsr.rs
@@ -0,0 +1,92 @@
+// 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.
+
+//! Armless non-replicated command codes against the server (vsr). The read
+//! gate is total over the protocol command table, so a bound session sending
+//! a `NonReplicated` header whose reserved command slot carries a code the
+//! reads have no arm for must get a typed `InvalidCommand` deny Reply. Two
+//! codes pin the two halves. One no table entry claims (the SDK forwards
+//! unknown codes untouched, `COMMAND_TABLE` being a registry rather than a
+//! capability list): the shared response builder's catch-all already denied
+//! it `InvalidCommand`, so that test pins the pre-existing deny now that the
+//! gate owns it. One the table lists but no read serves (`LOGOUT_USER`, which
+//! the SDK only ever sends as `Operation::Logout`): the old gate fell open
+//! and let the builder acknowledge it empty-ok, as if a logout had happened,
+//! so that test pins the closed fail-open. The frames are hand-crafted on a
+//! raw TCP socket to pin the status word the SDK maps through
+//! `IggyError::from_code`.
+
+use iggy::prelude::*;
+use iggy_binary_protocol::codes::LOGOUT_USER_CODE;
+use iggy_binary_protocol::lookup_command;
+use integration::harness::TestHarness;
+use integration::iggy_harness;
+
+use crate::server::raw_tcp::{
+ connect, exchange, non_replicated_header, register_root, reply_status,
+};
+
+/// A code no `COMMAND_TABLE` entry claims.
+const UNKNOWN_CODE: u32 = 9999;
+
+/// The header validator requires a nonzero client id; the value is otherwise
+/// free since nothing here reconnects.
+const CLIENT_ID: u128 = 0xBAD_C0DE;
+
+#[iggy_harness]
+async fn given_bound_session_when_unknown_non_replicated_code_sent_should_deny_invalid_command(
+ harness: &TestHarness,
+) {
+ assert!(
+ lookup_command(UNKNOWN_CODE).is_none(),
+ "test needs a code absent from COMMAND_TABLE"
+ );
+ assert_non_replicated_code_denied_invalid_command(harness, UNKNOWN_CODE).await;
+}
+
+#[iggy_harness]
+async fn given_bound_session_when_table_listed_code_without_read_arm_sent_should_deny_invalid_command(
+ harness: &TestHarness,
+) {
+ assert!(
+ lookup_command(LOGOUT_USER_CODE).is_some_and(|meta| !meta.is_replicated()),
+ "test needs a non-replicated COMMAND_TABLE entry"
+ );
+ assert_non_replicated_code_denied_invalid_command(harness, LOGOUT_USER_CODE).await;
+}
+
+/// Register root on a raw socket, send a header-only `NonReplicated` frame
+/// carrying `code` in the reserved command slot on that bound connection, and
+/// assert the server answers with an `InvalidCommand` deny Reply.
+async fn assert_non_replicated_code_denied_invalid_command(harness: &TestHarness, code: u32) {
+ let mut stream = connect(harness).await;
+ let session = register_root(&mut stream, CLIENT_ID).await;
+
+ let header = non_replicated_header(CLIENT_ID, session, 1, code);
+ let (reply_header, reply_body) = exchange(&mut stream, &header, &[]).await;
+
+ assert_eq!(
+ reply_status(&reply_header),
+ IggyError::InvalidCommand.as_code(),
+ "a bound session sending non-replicated code {code} must be denied InvalidCommand"
+ );
+ assert!(
+ reply_body.is_empty(),
+ "a deny Reply carries an empty body, got {} bytes",
+ reply_body.len()
+ );
+}
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index dc2399ecf..726a013c1 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -736,8 +736,13 @@ pub struct IggyMetadata<C, J, S, M, SB = PingPongSuperblock> {
/// policy.
superblock_write_failures: Cell<u64>,
superblock_retry_after_micros: Cell<u64>,
- /// State machine - lives on all shards
- pub mux_stm: M,
+ /// State machine - lives on all shards.
+ ///
+ /// Shared so shard 0's bootstrap can keep a clone alive past every
+ /// fallible step that owns this struct: the peer shards read through
+ /// handles minted off this writer, and dropping it makes their
+ /// `LeftRight::read` panic. See `server/src/boot::shard_main`.
+ pub mux_stm: Rc<M>,
pub allocator: ConsensusGroupAllocator,
/// Snapshot coordinator - present when persistent checkpointing is configured.
pub coordinator: Option<SnapshotCoordinator<M>>,
@@ -809,9 +814,10 @@ where
journal: Option<J>,
snapshot: Option<S>,
superblock: Option<Rc<SB>>,
- mux_stm: M,
+ mux_stm: impl Into<Rc<M>>,
data_dir: Option<std::path::PathBuf>,
) -> Self {
+ let mux_stm = mux_stm.into();
let allocator =
ConsensusGroupAllocator::new(mux_stm.streams().highest_partition_consensus_group_id());
let coordinator = data_dir.map(|dir| SnapshotCoordinator::new(dir, IggySnapshot::create));
@@ -2860,7 +2866,7 @@ where
// Normal op: apply SM, commit_reply. `Err` is decode/corruption
// only; a business rejection commits as a deterministic no-op
// whose `code` rides the reply body, replayed on retry.
- let apply = gated_apply(&self.mux_stm, prepare).unwrap_or_else(|err| {
+ let apply = gated_apply(&*self.mux_stm, prepare).unwrap_or_else(|err| {
panic!(
"on_ack: committed metadata op={} failed to apply: {err}",
prepare_header.op
@@ -3271,7 +3277,7 @@ where
// (see the phantom-op comment at the call site).
let client_table = self.client_table.borrow().to_snapshot();
let checksum = match coordinator.persist_snapshot(
- &self.mux_stm,
+ &*self.mux_stm,
snap_op,
created_at,
Some(client_table),
@@ -3632,7 +3638,7 @@ where
// table, while their state-machine effects still have to replay
// (the snapshot sits at a lower op).
apply_committed_prepare(
- &self.mux_stm,
+ &*self.mux_stm,
&self.client_table,
self.client_table_mutation_allowed(header.op),
|operation| self.fire_commit_notifier(operation),
diff --git a/core/server/src/boot/mod.rs b/core/server/src/boot/mod.rs
index 5053c8f8d..fc2de4397 100644
--- a/core/server/src/boot/mod.rs
+++ b/core/server/src/boot/mod.rs
@@ -47,12 +47,12 @@ use crate::boot::listeners::{
make_replica_delegation_fns, make_shard_zero_client_accept_fns, start_tcp_runtime,
};
use crate::boot::recovery::{
- RecoveredOwnerState, build_shard_for_thread, restore_metadata_consensus,
+ RecoveredOwnerState, ShardBuild, build_shard_for_thread, restore_metadata_consensus,
};
use crate::boot::threads::{
- StopSignals, await_pump_drain, install_panic_hook, join_partial_shard_survivors,
- resolve_shard_assignments, run_shard_thread, spawn_shutdown_watchdog,
- validate_sharding_runtime_knobs,
+ PeerExitCountdown, PeerExitWait, ShutdownDeadline, StopSignals, await_pump_drain,
+ install_panic_hook, join_partial_shard_survivors, resolve_shard_assignments, run_shard_thread,
+ spawn_shutdown_watchdog, validate_sharding_runtime_knobs,
};
use crate::boot::topology::{RosterCells, resolve_tcp_topology};
use crate::dispatch::partition::make_partition_read_handler;
@@ -60,8 +60,8 @@ use crate::dispatch::reads::read_frontier_budget;
use crate::dispatch::session_ops::warm_dummy_password_hash;
use crate::dispatch::submit::make_metadata_submit_handler;
use crate::dispatch::{
- make_client_request_handler, make_deferred_client_request_handler,
- make_deferred_replica_message_handler, make_list_clients_handler,
+ make_deferred_client_request_handler, make_deferred_replica_message_handler,
+ make_list_clients_handler,
};
use crate::server_error::ServerError;
use crate::session_manager::SessionManager;
@@ -278,6 +278,23 @@ pub fn bootstrap(
// thread body or in a task compio's `spawn` would swallow, escapes it.
let first_panic = install_panic_hook(Arc::clone(&shutdown_flag));
let config = Arc::new(config);
+ // One post-shutdown budget for the whole process: the main thread's
+ // shard joins and shard 0's peer wait both count down this instant
+ // instead of each arming a full `shutdown_join_timeout`. The floor keeps a
+ // spent join budget from releasing the metadata writer while a peer still
+ // reads through it, so it covers a peer's whole exit: one poll interval to
+ // observe the shutdown flag, then one drain budget to drain. Flooring at
+ // the drain alone is short by a poll interval, and
+ // `shutdown_join_timeout == shutdown_drain_timeout` is a legal config.
+ let shutdown_deadline = Arc::new(ShutdownDeadline::new(
+ config.system.sharding.shutdown_join_timeout.get_duration(),
+ config
+ .system
+ .sharding
+ .shutdown_drain_timeout
+ .get_duration()
+ .saturating_add(config.system.sharding.shutdown_poll_interval.get_duration()),
+ ));
// One owner table per server process, Arc-cloned into every shard's bus so
// any shard's bus reads the same atomic slots that the owning
// shard's installer / disconnect path writes.
@@ -302,6 +319,11 @@ pub fn bootstrap(
// count so a sender never blocks (each peer sends exactly once).
let (ready_tx, ready_rx) = crossfire::mpmc::bounded_async::<u16>(metadata_peers);
+ // Shard 0 blocks on this at exit until every peer thread is done (see
+ // `PeerExitCountdown`). Sized to the real peer count, not the clamped
+ // channel capacity above: a single-shard server must not wait at all.
+ let peer_exit = Arc::new(PeerExitCountdown::new(shards_count.saturating_sub(1)));
+
let mut shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerError>>)> =
Vec::with_capacity(shards_count);
let roster_cells = RosterCells::default();
@@ -354,6 +376,8 @@ pub fn bootstrap(
let roster_cells_for_shard = roster_cells.clone();
let applied_frontier_for_shard = Arc::clone(&metadata_applied_frontier);
let shard_metrics_for_shard = shard_metrics_all.clone();
+ let peer_exit_for_shard = Arc::clone(&peer_exit);
+ let shutdown_deadline_for_shard = Arc::clone(&shutdown_deadline);
let handle = match thread::Builder::new()
.name(format!("shard-{shard_id}"))
.spawn(move || -> Result<(), ServerError> {
@@ -373,6 +397,8 @@ pub fn bootstrap(
roster_cells_for_shard,
applied_frontier_for_shard,
shard_metrics_for_shard,
+ peer_exit_for_shard,
+ shutdown_deadline_for_shard,
)
}) {
Ok(handle) => handle,
@@ -389,10 +415,14 @@ pub fn bootstrap(
drop(metadata_bundle_rx);
drop(ready_tx);
drop(ready_rx);
- join_partial_shard_survivors(
- shard_threads,
- config.system.sharding.shutdown_join_timeout.get_duration(),
+ // Shard 0 spawns first, so the vec holds it plus every peer
+ // that made it; the rest are peers shard 0's exit wait would
+ // otherwise sit out its whole budget for.
+ let spawned_peers = shard_threads.len().saturating_sub(1);
+ peer_exit.peers_never_spawned(
+ shards_count.saturating_sub(1).saturating_sub(spawned_peers),
);
+ join_partial_shard_survivors(shard_threads, &shutdown_deadline);
return Err(ServerError::ShardSpawnFailed { shard_id, source });
}
};
@@ -416,7 +446,7 @@ pub fn bootstrap(
Ok(ShardHandles {
shutdown_flag,
shard_threads,
- join_timeout: config.system.sharding.shutdown_join_timeout.get_duration(),
+ deadline: shutdown_deadline,
first_panic,
})
}
@@ -441,6 +471,8 @@ async fn shard_main(
roster_cells: RosterCells,
metadata_applied_frontier: Arc<AppliedFrontier>,
shard_metrics_all: Vec<ShardMetrics>,
+ peer_exit: Arc<PeerExitCountdown>,
+ shutdown_deadline: Arc<ShutdownDeadline>,
) -> Result<(), ServerError> {
let topology = resolve_tcp_topology(config, replica_id)?;
let bus = Rc::new(IggyMessageBus::with_config_and_owner_table(
@@ -478,7 +510,12 @@ async fn shard_main(
// metadata VSR; per-commit `publish()` (in `WriteCell::apply`)
// bounds reader staleness to one op.
let data_dir = Path::new(&config.system.path);
- let (mux_stm, owner_state) = match metadata_handoff {
+ // The bundle broadcast is deliberately NOT inside the owner arm: it is
+ // the first moment a peer can hold a read handle over shard 0's writer,
+ // so the writer must first be parked in a binding that outlives the peer
+ // wait armed below. A `recover()` failure inside the arm is safe for the
+ // same reason -- no peer holds a handle yet.
+ let (mux_stm, pending_bundle_tx, owner_state) = match metadata_handoff {
MetadataHandoff::Owner { bundle_tx } => {
// Root is created locally at boot (never journaled), so replay
// must start from the same baseline or every WAL-created user
@@ -520,17 +557,9 @@ async fn shard_main(
}
}
});
- broadcast_metadata_bundle(
- shard_id,
- &bundle_tx,
- recovered.mux_stm.factory_bundle(),
- total_shards.saturating_sub(1),
- &shutdown_flag_for_handoff,
- poll_interval,
- )
- .await?;
(
- recovered.mux_stm,
+ Rc::new(recovered.mux_stm),
+ Some(bundle_tx),
Some(RecoveredOwnerState {
journal: recovered.journal,
snapshot: recovered.snapshot,
@@ -551,10 +580,40 @@ async fn shard_main(
poll_interval,
)
.await?;
- (ServerMuxStateMachine::from_factory_bundle(bundle), None)
+ (
+ Rc::new(ServerMuxStateMachine::from_factory_bundle(bundle)),
+ None,
+ None,
+ )
}
};
+ // Shard 0 owns the metadata state machine's only write handle, and the
+ // peers read through it until their runtimes are gone. Declared after
+ // `mux_stm` so it drops first: every exit from here on, clean or `?`,
+ // waits for the peers before the write side goes. The `Rc` above is what
+ // makes that hold -- the handle no longer travels by value into the
+ // fallible shard build, which would drop it inside the callee.
+ let _peer_exit_wait = (shard_id == 0).then(|| {
+ PeerExitWait::new(
+ peer_exit,
+ Arc::clone(&shutdown_flag_for_handoff),
+ shutdown_deadline,
+ )
+ });
+
+ if let Some(bundle_tx) = pending_bundle_tx {
+ broadcast_metadata_bundle(
+ shard_id,
+ &bundle_tx,
+ mux_stm.factory_bundle(),
+ total_shards.saturating_sub(1),
+ &shutdown_flag_for_handoff,
+ poll_interval,
+ )
+ .await?;
+ }
+
// Metadata consensus + journal + snapshot live only on shard 0.
// `IggyShard::tick_metadata` short-circuits when `consensus.is_none()`,
// so peer shards have no caller that reads `journal` or `snapshot`.
@@ -589,7 +648,7 @@ async fn shard_main(
journal_for_metadata,
snapshot_for_metadata,
superblock_for_metadata,
- mux_stm,
+ Rc::clone(&mux_stm),
Some(PathBuf::from(&config.system.path)),
)
.with_applied_frontier(metadata_applied_frontier);
@@ -625,7 +684,12 @@ async fn shard_main(
// Heap-pin like `run_shard_thread` pins `shard_main`: the builder future
// carries the whole shard construction state machine and outgrew clippy's
// `large_futures` cap; one allocation per shard startup.
- let (shard, sessions) = Box::pin(build_shard_for_thread(
+ let ShardBuild {
+ shard,
+ sessions,
+ on_client_request,
+ shard_handle,
+ } = Box::pin(build_shard_for_thread(
shard_id,
total_shards,
config,
@@ -892,11 +956,15 @@ async fn shard_main(
boot_view,
shard.plane.metadata().client_table.borrow().client_ids(),
);
- let on_client_request = make_client_request_handler(
- &shard,
- &sessions,
- Arc::clone(&config.system),
- config.personal_access_token.max_tokens_per_user,
+ // The request handler strands every frame until the weak
+ // self-reference is backfilled, so the build must have done that
+ // before the first listener binds.
+ debug_assert!(
+ shard_handle
+ .borrow()
+ .as_ref()
+ .is_some_and(|weak| weak.upgrade().is_some()),
+ "shard self-reference must be backfilled before listeners bind"
);
let (accepted_replica, dialed_replica) =
make_replica_delegation_fns(Rc::clone(&coord), &bus);
diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs
index eb99734d6..5d21e1f7c 100644
--- a/core/server/src/boot/recovery.rs
+++ b/core/server/src/boot/recovery.rs
@@ -23,7 +23,8 @@ use crate::partition_helpers::load_partition_or_fence;
use crate::server_error::ServerError;
use crate::session_manager::SessionManager;
use crate::shell::{
- ServerMetadata, ServerShard, ShellHandlers, consensus_timers, repair_retry_ticks,
+ ServerMetadata, ServerShard, ShellHandlers, ShellShardHandle, consensus_timers,
+ repair_retry_ticks,
};
use configs::server::ServerConfig;
use consensus::{
@@ -35,6 +36,7 @@ use journal::Journal;
use journal::prepare_journal::PrepareJournal;
use journal::superblock::PingPongSuperblock;
use message_bus::IggyMessageBus;
+use message_bus::client_listener::RequestHandler;
use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
use metadata::stm::snapshot::Snapshot;
use partitions::{IggyPartitions, PartitionsConfig};
@@ -52,6 +54,18 @@ use std::sync::Arc;
use std::time::Duration;
use tracing::{error, info, warn};
+/// A shard built for its thread, with what `shard_main` wires after the
+/// build: the session manager its request plane shares, the shard's one
+/// client-request handler (shard 0 hands the same instance to its local
+/// transports), and the weak self-reference the deferred handlers
+/// upgrade per frame, already backfilled.
+pub(in crate::boot) struct ShardBuild {
+ pub shard: Rc<ServerShard>,
+ pub sessions: Rc<RefCell<SessionManager>>,
+ pub on_client_request: RequestHandler,
+ pub shard_handle: ShellShardHandle<Rc<IggyMessageBus>, PrepareJournal, IggySnapshot>,
+}
+
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub(in crate::boot) async fn build_shard_for_thread(
shard_id: u16,
@@ -65,7 +79,7 @@ pub(in crate::boot) async fn build_shard_for_thread(
reply_inbox: ShardReceiver<ShardFrame>,
metrics: ShardMetrics,
roster_cells: &RosterCells,
-) -> Result<(Rc<ServerShard>, Rc<RefCell<SessionManager>>), ServerError> {
+) -> Result<ShardBuild, ServerError> {
let shard_local_id = ShardId::new(shard_id);
let total_partitions = metadata.mux_stm.streams().read(|inner| {
inner
@@ -243,7 +257,7 @@ pub(in crate::boot) async fn build_shard_for_thread(
ShardIdentity::new(shard_id, shard_name),
Rc::clone(&bus),
on_replica_message,
- on_client_request,
+ Rc::clone(&on_client_request),
on_metadata_submit,
on_list_clients,
on_partition_read,
@@ -289,7 +303,12 @@ pub(in crate::boot) async fn build_shard_for_thread(
usize::try_from(config.message_bus.max_message_size.as_bytes_u64()).unwrap_or(usize::MAX),
);
*shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
- Ok((shard, sessions))
+ Ok(ShardBuild {
+ shard,
+ sessions,
+ on_client_request,
+ shard_handle,
+ })
}
// Pin the configs-crate default literals (duplicated there to avoid a
diff --git a/core/server/src/boot/threads.rs b/core/server/src/boot/threads.rs
index 2a68d89b9..83c4e408c 100644
--- a/core/server/src/boot/threads.rs
+++ b/core/server/src/boot/threads.rs
@@ -25,7 +25,8 @@ use crate::shard_allocator::{ShardAllocator, ShardInfo};
use compio::runtime::ResumeUnwind;
use configs::server::ServerConfig;
use configs::sharding::{
- INBOX_CAPACITY_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX,
+ INBOX_CAPACITY_MAX, RECONCILE_PERIODIC_INTERVAL_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX,
+ SHUTDOWN_JOIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX,
};
use message_bus::{IggyMessageBus, ReplicaOwnerTable};
use metadata::AppliedFrontier;
@@ -36,7 +37,7 @@ use shard::{Receiver as ShardReceiver, Sender, ShardFrame, TaggedSender};
use std::backtrace::Backtrace;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, OnceLock};
+use std::sync::{Arc, Condvar, Mutex, OnceLock, PoisonError};
use std::time::{Duration, Instant};
use std::{panic, thread};
use tracing::{error, info, warn};
@@ -46,12 +47,12 @@ use tracing::{error, info, warn};
/// Carries the cross-thread shutdown flag, one OS-thread `JoinHandle`
/// per shard, and the first panic `install_panic_hook` recorded. The
/// caller flips the flag via [`Self::install_ctrlc_handler`] and then
-/// drains every shard via [`Self::join_all`], bounded by `join_timeout`
-/// (`system.sharding.shutdown_join_timeout`).
+/// drains every shard via [`Self::join_all`], bounded by the shared
+/// `ShutdownDeadline` (`system.sharding.shutdown_join_timeout`).
pub struct ShardHandles {
pub(in crate::boot) shutdown_flag: Arc<AtomicBool>,
pub(in crate::boot) shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerError>>)>,
- pub(in crate::boot) join_timeout: Duration,
+ pub(in crate::boot) deadline: Arc<ShutdownDeadline>,
pub(in crate::boot) first_panic: Arc<OnceLock<String>>,
}
@@ -94,7 +95,9 @@ impl ShardHandles {
/// deadline passes is abandoned (its `JoinHandle` dropped, the OS
/// thread left to die with the process) and reported as
/// [`ShardJoinFailureKind::Wedged`]: a wedged pump or listener must
- /// not block process exit forever.
+ /// not block process exit forever. Shard 0 gets the peer-wait floor
+ /// as grace on top, because its `PeerExitWait` is allowed to hold the
+ /// metadata writer that long past a spent budget.
///
/// # Errors
///
@@ -107,31 +110,32 @@ impl ShardHandles {
/// compio's `spawn` caught, which no thread result can carry.
pub fn join_all(self) -> Result<(), ServerError> {
let mut failures: Vec<ShardJoinFailure> = Vec::new();
- // Armed on the first poll that observes the shutdown flag, shared
- // across all shards: one budget covers the whole drain, not one
- // budget per shard.
- let mut deadline: Option<Instant> = None;
// Shards run thread-per-core with compio's blocking fallback pool
// disabled, so an io_uring opcode the kernel lacks aborts every shard
// with the same panic. Surface the actionable diagnostic once.
let mut io_uring_diagnostic_shown = false;
for (shard_id, handle) in self.shard_threads {
- let Some(joined) = join_until_shutdown_deadline(
- handle,
- &self.shutdown_flag,
- self.join_timeout,
- &mut deadline,
- ) else {
+ // Shard 0's `PeerExitWait` is allowed to overrun the budget by
+ // its floor, so its join has to tolerate the same overrun or a
+ // shutdown that correctly held the writer for a slow peer reports
+ // the shard that honoured the fence as wedged.
+ let grace = if shard_id == 0 {
+ self.deadline.peer_wait_floor
+ } else {
+ Duration::ZERO
+ };
+ let waited = self.deadline.budget.saturating_add(grace);
+ let Some(joined) =
+ join_until_shutdown_deadline(handle, &self.shutdown_flag, &self.deadline, grace)
+ else {
error!(
shard_id,
- waited = ?self.join_timeout,
+ ?waited,
"shard thread still running at the shutdown join deadline; abandoning it"
);
failures.push(ShardJoinFailure {
shard_id,
- kind: ShardJoinFailureKind::Wedged {
- waited: self.join_timeout,
- },
+ kind: ShardJoinFailureKind::Wedged { waited },
});
continue;
};
@@ -218,28 +222,24 @@ pub(in crate::boot) fn install_panic_hook(shutdown_flag: Arc<AtomicBool>) -> Arc
const JOIN_POLL_INTERVAL: Duration = Duration::from_millis(25);
/// Join `handle`, waiting indefinitely while the server runs. The
-/// `join_timeout` clock starts only when `shutdown_flag` is observed set
-/// (arming the caller-shared `deadline` once, so all shards drain under
-/// ONE budget); a running server parked here for hours must never be
-/// mistaken for a wedged shard. `None` means the thread was still
-/// running at the post-shutdown deadline and the handle was dropped
-/// (the OS thread keeps running detached; process exit reaps it).
-/// `JoinHandle` has no timed join, so this polls `is_finished` at
-/// [`JOIN_POLL_INTERVAL`]; the closing `join()` on a finished thread
-/// returns immediately.
+/// budget clock starts only when `shutdown_flag` is observed set (arming
+/// the shared [`ShutdownDeadline`], so every shard join and shard 0's
+/// peer wait drain under ONE budget); a running server parked here for
+/// hours must never be mistaken for a wedged shard. `grace` extends this
+/// caller's share of the budget by an overrun the joined thread is allowed
+/// (shard 0's peer-wait floor). `None` means the thread was still running
+/// at the post-shutdown deadline and the handle was dropped (the OS thread
+/// keeps running detached; process exit reaps it). `JoinHandle` has no
+/// timed join, so this polls `is_finished` at [`JOIN_POLL_INTERVAL`]; the
+/// closing `join()` on a finished thread returns immediately.
fn join_until_shutdown_deadline(
handle: thread::JoinHandle<Result<(), ServerError>>,
shutdown_flag: &AtomicBool,
- join_timeout: Duration,
- deadline: &mut Option<Instant>,
+ deadline: &ShutdownDeadline,
+ grace: Duration,
) -> Option<thread::Result<Result<(), ServerError>>> {
while !handle.is_finished() {
- if deadline.is_none() && shutdown_flag.load(Ordering::Relaxed) {
- *deadline = Some(Instant::now() + join_timeout);
- }
- if let Some(deadline) = deadline
- && Instant::now() >= *deadline
- {
+ if shutdown_flag.load(Ordering::Relaxed) && deadline.remaining_with_grace(grace).is_zero() {
return None;
}
thread::sleep(JOIN_POLL_INTERVAL);
@@ -274,9 +274,8 @@ fn panic_payload_to_string(payload: &(dyn std::any::Any + Send)) -> String {
/// spawn error instead of hanging on a wedged shard.
pub(in crate::boot) fn join_partial_shard_survivors(
shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerError>>)>,
- join_timeout: Duration,
+ deadline: &ShutdownDeadline,
) {
- let deadline = Instant::now() + join_timeout;
let mut remaining = shard_threads;
loop {
let mut still_running = Vec::with_capacity(remaining.len());
@@ -289,7 +288,7 @@ pub(in crate::boot) fn join_partial_shard_survivors(
}
}
remaining = still_running;
- if remaining.is_empty() || Instant::now() >= deadline {
+ if remaining.is_empty() || deadline.remaining().is_zero() {
break;
}
thread::sleep(JOIN_POLL_INTERVAL);
@@ -297,7 +296,7 @@ pub(in crate::boot) fn join_partial_shard_survivors(
for (shard_id, _survivor) in remaining {
error!(
shard_id,
- waited = ?join_timeout,
+ waited = ?deadline.budget,
"survivor shard thread still running at the shutdown join deadline; abandoning it"
);
}
@@ -334,6 +333,199 @@ 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.
+///
+/// The two waits enforce different things, so the budget is shared but not
+/// fungible: the join bounds LIVENESS (process exit must not hang on a
+/// wedged shard) while the peer wait enforces SAFETY (the metadata writer
+/// must outlive every reader). `peer_wait_floor` is what keeps an exhausted
+/// join budget from cancelling the safety fence, and the joins in turn
+/// tolerate that floor as grace (see [`ShardHandles::join_all`]).
+pub(in crate::boot) struct ShutdownDeadline {
+ armed: OnceLock<Instant>,
+ budget: Duration,
+ peer_wait_floor: Duration,
+}
+
+impl ShutdownDeadline {
+ pub(in crate::boot) const fn new(budget: Duration, peer_wait_floor: Duration) -> Self {
+ Self {
+ armed: OnceLock::new(),
+ budget,
+ peer_wait_floor,
+ }
+ }
+
+ /// Time left in the shared budget plus `grace`, 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_with_grace(&self, grace: Duration) -> Duration {
+ self.armed
+ .get_or_init(|| Instant::now() + self.budget)
+ .checked_add(grace)
+ .map_or(Duration::MAX, |limit| {
+ limit.saturating_duration_since(Instant::now())
+ })
+ }
+
+ fn remaining(&self) -> Duration {
+ self.remaining_with_grace(Duration::ZERO)
+ }
+
+ /// How long shard 0 may block on its peers: what is left of the shared
+ /// budget, floored at a peer's whole exit (one poll interval to observe the
+ /// shutdown flag plus one drain budget) so a join that already spent the
+ /// budget cannot release the metadata writer while a peer still reads
+ /// through it. Only [`PeerExitWait::drop`]'s panic arm skips the fence.
+ fn remaining_for_peer_wait(&self) -> Duration {
+ self.remaining().max(self.peer_wait_floor)
+ }
+}
+
+/// 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 [`ShutdownDeadline::remaining_for_peer_wait`] --
+/// what is left of the budget [`ShardHandles::join_all`] shares, but never
+/// less than one drain -- with the same abandon-and-log policy, so a wedged
+/// peer cannot hold shard 0 indefinitely while a spent join budget still
+/// cannot cut the wait to nothing.
+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_for_peer_wait();
+ if let Err(peers_running) = self.countdown.wait(remaining) {
+ warn!(
+ peers_running,
+ waited = ?remaining,
+ budget = ?self.deadline.budget,
+ "peer shards still running at the shutdown join deadline; \
+ releasing shard 0 anyway"
+ );
+ }
+ }
+}
+
/// Resolve the operator's `cpu_allocation` into concrete shard
/// assignments plus the checked `u16` shard count.
///
@@ -364,10 +556,13 @@ pub(in crate::boot) fn resolve_shard_assignments(
}
/// Re-validate the runtime sharding knobs that the per-shard runtime
-/// consumes directly. Mirrors `ShardingConfig::validate` so a caller
-/// that built the config without running it (e.g. tests, embedded
-/// usage) cannot OOM at boot or wedge process exit with an out-of-range
-/// value.
+/// consumes directly: the two inbox capacities, the three shutdown
+/// durations and their ordering, and the reconcile tick. Mirrors
+/// `ShardingConfig::validate` for exactly those, so a caller that built the
+/// config without running it (e.g. tests, embedded usage) cannot OOM at
+/// boot, starve a core, or wedge process exit with an out-of-range value.
+/// `cpu_allocation` is not re-checked here - [`resolve_shard_assignments`]
+/// resolves it through the allocator and rejects it there.
pub(in crate::boot) fn validate_sharding_runtime_knobs(
sharding: &configs::sharding::ShardingConfig,
) -> Result<(), ServerError> {
@@ -407,6 +602,32 @@ pub(in crate::boot) fn validate_sharding_runtime_knobs(
drain: drain_timeout,
});
}
+ let join_timeout = sharding.shutdown_join_timeout.get_duration();
+ if join_timeout > SHUTDOWN_JOIN_TIMEOUT_MAX {
+ return Err(ServerError::InvalidShutdownJoinTimeout {
+ value: join_timeout,
+ max: SHUTDOWN_JOIN_TIMEOUT_MAX,
+ });
+ }
+ // A join budget under the drain budget abandons shards mid-drain,
+ // interrupting the WAL fsync / replica drain, and now also cuts shard
+ // 0's peer wait short of the writer's last reader.
+ if join_timeout < drain_timeout {
+ return Err(ServerError::ShutdownJoinBelowDrain {
+ join: join_timeout,
+ drain: drain_timeout,
+ });
+ }
+ // Zero feeds `run_reconciler`'s sleep inside an unconditional loop, so the
+ // tick arm is ready every iteration and `reconcile_once` runs back to back,
+ // starving the pump on that core.
+ let reconcile_interval = sharding.reconcile_periodic_interval.get_duration();
+ if reconcile_interval.is_zero() || reconcile_interval > RECONCILE_PERIODIC_INTERVAL_MAX {
+ return Err(ServerError::InvalidReconcilePeriodicInterval {
+ value: reconcile_interval,
+ max: RECONCILE_PERIODIC_INTERVAL_MAX,
+ });
+ }
Ok(())
}
@@ -429,11 +650,17 @@ pub(in crate::boot) fn run_shard_thread(
roster_cells: RosterCells,
metadata_applied_frontier: Arc<AppliedFrontier>,
shard_metrics_all: Vec<ShardMetrics>,
+ peer_exit: Arc<PeerExitCountdown>,
+ shutdown_deadline: Arc<ShutdownDeadline>,
) -> Result<(), ServerError> {
// Armed for the whole thread body: a post-spawn error `?` or a panic
// unwind here must flip `shutdown_flag` so sibling watchdogs drive
// their bus shutdown instead of parking forever on `bus.token().wait()`.
let mut shutdown_guard = ShutdownOnDrop::new(Arc::clone(&shutdown_flag));
+ // Declared before the runtime so it drops after it: a peer counts
+ // itself out only once no task of its runtime can read shard 0's
+ // metadata any more.
+ let _peer_exit_guard = (shard_id != 0).then(|| PeerExitGuard::new(Arc::clone(&peer_exit)));
assignment
.bind_cpu()
@@ -473,6 +700,8 @@ pub(in crate::boot) fn run_shard_thread(
roster_cells,
metadata_applied_frontier,
shard_metrics_all,
+ peer_exit,
+ shutdown_deadline,
))
.await
});
@@ -652,6 +881,248 @@ 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"
+ );
+ }
+
+ /// Stops a parked stand-in thread once the test's binding goes out of
+ /// scope. The joiner under test drops the `JoinHandle`, so the thread
+ /// cannot be joined back; flagging it keeps it from parking for the life
+ /// of the test binary.
+ struct ThreadStopper(Arc<AtomicBool>);
+
+ impl Drop for ThreadStopper {
+ fn drop(&mut self) {
+ self.0.store(true, Ordering::Relaxed);
+ }
+ }
+
+ /// A shard thread that never returns on its own: the stand-in for a
+ /// wedged pump that the deadline tests need.
+ fn spawn_wedged_shard() -> (thread::JoinHandle<Result<(), ServerError>>, ThreadStopper) {
+ let stop = Arc::new(AtomicBool::new(false));
+ let handle = thread::spawn({
+ let stop = Arc::clone(&stop);
+ move || -> Result<(), ServerError> {
+ while !stop.load(Ordering::Relaxed) {
+ thread::sleep(JOIN_POLL_INTERVAL);
+ }
+ Ok(())
+ }
+ });
+ (handle, ThreadStopper(stop))
+ }
+
+ #[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),
+ Duration::from_secs(10),
+ )),
+ ));
+ 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));
+ // Degenerate budget == floor: the floor cannot extend the wait, so
+ // the abandon is bounded by the budget alone.
+ let timeout = Duration::from_millis(50);
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ countdown,
+ Arc::new(AtomicBool::new(false)),
+ Arc::new(ShutdownDeadline::new(timeout, timeout)),
+ ));
+ let waited = started.elapsed();
+ assert!(
+ waited >= timeout && waited < Duration::from_secs(5),
+ "a wedged peer must be abandoned at the deadline, waited {waited:?}"
+ );
+ }
+
+ /// 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. What is left of the shared budget is what the peer
+ /// wait gets, whenever that clears the floor.
+ #[test]
+ fn peer_wait_inherits_the_shared_budget_above_its_floor() {
+ const BUDGET: Duration = Duration::from_secs(30);
+ const FLOOR: Duration = Duration::from_secs(10);
+ let deadline = ShutdownDeadline::new(BUDGET, FLOOR);
+ let inherited = deadline.remaining_for_peer_wait();
+ assert!(
+ inherited > FLOOR && inherited <= BUDGET,
+ "an unspent budget must be inherited, not replaced by the floor: {inherited:?}"
+ );
+ }
+
+ /// The other half of the same nesting: the join budget bounds LIVENESS
+ /// (process exit must not hang on a wedged shard) while this wait enforces
+ /// SAFETY (shard 0 owns the metadata write handle every peer reads
+ /// through). Funding the fence out of the join budget let an exit-latency
+ /// timeout cancel it: the drain spent the budget, `wait(ZERO)` returned
+ /// without blocking, and shard 0 dropped the writer under a live reader.
+ #[test]
+ fn peer_wait_honours_a_live_peer_past_a_spent_join_budget() {
+ const BUDGET: Duration = Duration::from_millis(20);
+ const FLOOR: Duration = Duration::from_millis(200);
+ let deadline = Arc::new(ShutdownDeadline::new(BUDGET, FLOOR));
+ let shutdown_flag = AtomicBool::new(true);
+ // Stands in for the slow drain that arms and then spends the shared
+ // budget before shard 0 can reach its peer wait.
+ let (wedged_shard, _stopper) = spawn_wedged_shard();
+ assert!(
+ join_until_shutdown_deadline(wedged_shard, &shutdown_flag, &deadline, Duration::ZERO)
+ .is_none(),
+ "the join must spend the budget it armed"
+ );
+ assert!(deadline.remaining().is_zero(), "the budget must be spent");
+
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let _live_peer = PeerExitGuard::new(Arc::clone(&countdown));
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ countdown,
+ Arc::new(AtomicBool::new(false)),
+ Arc::clone(&deadline),
+ ));
+ let waited = started.elapsed();
+ assert!(
+ waited >= FLOOR,
+ "a spent join budget must not release the metadata writer while a peer still reads \
+ through it, waited {waited:?}"
+ );
+ }
+
+ #[test]
+ fn peer_exit_wait_ignores_peers_that_never_spawned() {
+ // A failed spawn leaves peers with no guard to count them out; the
+ // one that did spawn must still be the only thing the waiter waits on.
+ let countdown = Arc::new(PeerExitCountdown::new(3));
+ countdown.peers_never_spawned(2);
+ let spawned = PeerExitGuard::new(Arc::clone(&countdown));
+ assert_eq!(
+ countdown.wait(Duration::from_millis(10)),
+ Err(1),
+ "the peer that did spawn must still hold the waiter"
+ );
+ drop(spawned);
+ assert_eq!(countdown.wait(Duration::ZERO), Ok(()));
+ }
+
+ #[test]
+ fn peer_exit_wait_with_no_peers_returns_at_once() {
+ let started = Instant::now();
+ drop(PeerExitWait::new(
+ Arc::new(PeerExitCountdown::new(0)),
+ Arc::new(AtomicBool::new(false)),
+ Arc::new(ShutdownDeadline::new(
+ Duration::from_secs(30),
+ Duration::from_secs(10),
+ )),
+ ));
+ assert!(
+ started.elapsed() < Duration::from_secs(1),
+ "a single-shard server has nobody to wait for"
+ );
+ }
+
+ #[test]
+ fn peer_exit_wait_never_blocks_an_unwinding_thread() {
+ let countdown = Arc::new(PeerExitCountdown::new(1));
+ let _still_running = PeerExitGuard::new(Arc::clone(&countdown));
+ let wait = PeerExitWait::new(
+ countdown,
+ Arc::new(AtomicBool::new(false)),
+ Arc::new(ShutdownDeadline::new(
+ Duration::from_secs(30),
+ Duration::from_secs(10),
+ )),
+ );
+ let started = Instant::now();
+ let unwound = panic::catch_unwind(panic::AssertUnwindSafe(|| {
+ let _wait = wait;
+ panic::resume_unwind(Box::new("shard 0 body panicked"));
+ }));
+ assert!(unwound.is_err());
+ assert!(
+ started.elapsed() < Duration::from_secs(1),
+ "a waiter dropped during unwind must return without waiting"
+ );
+ }
+
#[compio::test]
async fn pump_drain_timeout_is_not_reported_as_clean() {
let mut config = ServerConfig::default();
@@ -714,19 +1185,15 @@ mod tests {
thread::sleep(Duration::from_millis(300));
Ok(())
});
- let mut deadline = None;
- let joined = join_until_shutdown_deadline(
- handle,
- &shutdown_flag,
- Duration::from_millis(20),
- &mut deadline,
- );
+ let deadline = ShutdownDeadline::new(Duration::from_millis(20), Duration::from_millis(20));
+ let joined =
+ join_until_shutdown_deadline(handle, &shutdown_flag, &deadline, Duration::ZERO);
assert!(
matches!(joined, Some(Ok(Ok(())))),
"a running server must be awaited indefinitely, not abandoned as wedged"
);
assert!(
- deadline.is_none(),
+ deadline.armed.get().is_none(),
"the join deadline must not arm before the shutdown flag flips"
);
}
@@ -744,7 +1211,10 @@ mod tests {
let handles = ShardHandles {
shutdown_flag: Arc::new(AtomicBool::new(true)),
shard_threads: vec![(0, handle)],
- join_timeout: Duration::from_secs(1),
+ deadline: Arc::new(ShutdownDeadline::new(
+ Duration::from_secs(1),
+ Duration::from_millis(500),
+ )),
first_panic,
};
let error = handles
@@ -781,24 +1251,17 @@ mod tests {
#[test]
fn join_abandons_a_wedged_shard_after_the_shutdown_deadline() {
let shutdown_flag = AtomicBool::new(true);
- // Never finishes: stands in for a wedged pump. The thread leaks
- // into the test process, which exits right after.
- let handle = thread::spawn(|| -> Result<(), ServerError> {
- loop {
- thread::sleep(Duration::from_secs(1));
- }
- });
- let mut deadline = None;
- let joined = join_until_shutdown_deadline(
- handle,
- &shutdown_flag,
- Duration::from_millis(100),
- &mut deadline,
- );
+ let (handle, _stopper) = spawn_wedged_shard();
+ let deadline = ShutdownDeadline::new(Duration::from_millis(100), Duration::from_millis(50));
+ let joined =
+ join_until_shutdown_deadline(handle, &shutdown_flag, &deadline, Duration::ZERO);
assert!(
joined.is_none(),
"a shard still running past the post-shutdown budget must be abandoned"
);
- assert!(deadline.is_some(), "the deadline arms once the flag is set");
+ assert!(
+ deadline.armed.get().is_some(),
+ "the deadline arms once the flag is set"
+ );
}
}
diff --git a/core/server/src/dispatch/authz.rs b/core/server/src/dispatch/authz.rs
index 436aa2222..c6fb2c6e4 100644
--- a/core/server/src/dispatch/authz.rs
+++ b/core/server/src/dispatch/authz.rs
@@ -29,9 +29,10 @@ use std::rc::Rc;
use consensus::MetadataHandle;
use iggy_binary_protocol::codes::{
- DESCRIBE_OPTIONS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE,
- GET_CONSUMER_GROUPS_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, GET_STATS_CODE, GET_STREAM_CODE,
- GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE,
+ DESCRIBE_OPTIONS_CODE, FLUSH_UNSAVED_BUFFER_CODE, GET_CLUSTER_METADATA_CODE,
+ GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE,
+ GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE,
+ GET_USER_CODE, GET_USERS_CODE,
};
use iggy_binary_protocol::requests::consumer_groups::{
GetConsumerGroupRequest, GetConsumerGroupsRequest,
@@ -39,20 +40,15 @@ use iggy_binary_protocol::requests::consumer_groups::{
use iggy_binary_protocol::requests::streams::GetStreamRequest;
use iggy_binary_protocol::requests::topics::{GetTopicRequest, GetTopicsRequest};
use iggy_binary_protocol::requests::users::GetUserRequest;
-use iggy_binary_protocol::{
- Operation, PrepareHeader, RoutedRequestHeader, WireDecode, WireIdentifier,
-};
+use iggy_binary_protocol::{Operation, PrepareHeader, WireDecode, WireIdentifier, lookup_command};
use iggy_common::IggyError;
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
use metadata::impls::metadata::StreamsFrontend;
use metadata::permissioner::Permissioner;
use server_common::Message;
-use tracing::warn;
-use crate::responses::{
- build_deny_reply, current_metadata_commit, resolve_stream_id, resolve_topic_id,
-};
+use crate::responses::{resolve_stream_id, resolve_topic_id};
use crate::shell::{ShellBus, ShellShard};
/// Authorize a partition-plane op on its resolved (stream, topic) for the
@@ -132,74 +128,6 @@ where
decision.err().map(|error| error.as_code())
}
-/// 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);
- 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,
- operation = ?request_header.operation,
- "failed to surface request denial"
- );
- }
-}
-
-/// 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);
- 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,
- operation = ?request_header.operation,
- "failed to surface unbound request denial"
- );
- }
-}
-
/// Run an unscoped non-replicated-read rule for the acting user. A `None` user
/// id (only the pre-auth path, which serves ungated codes) fails closed.
pub(in crate::dispatch) fn authorize_uid<B, MJ, S, SB>(
@@ -263,6 +191,11 @@ where
/// topic]) against committed state first. The PAT list is self-scoped, so
/// authentication is its whole rule, and `GET_CLUSTER_METADATA` -- which
/// describes the private replica network -- is gated the same way.
+///
+/// The gate is total over the protocol command table: every code the builder
+/// serves has a named arm, and the tail refuses everything else (a replicated
+/// code inside a `NonReplicated` header, a table-listed code with no arm, an
+/// unknown code) instead of deferring it to the builder's catch-all.
pub(in crate::dispatch) fn authorize_default_read<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
code: u32,
@@ -276,8 +209,8 @@ where
S: 'static,
SB: SuperblockStore + 'static,
{
- // A `u32` match cannot be exhaustive: every gated code is named explicitly,
- // and the final arm is the ungated set the builder serves without a rule.
+ // A `u32` match cannot be exhaustive, so totality is by construction:
+ // every code with a decision is named, and the tail refuses the rest.
match code {
GET_STATS_CODE => authorize_uid(shard, user_id, Permissioner::get_stats),
GET_USERS_CODE => authorize_uid(shard, user_id, Permissioner::get_users),
@@ -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, and flush has no HTTP route, so
+ // this arm is the only thing answering `FeatureUnavailable` for it.
+ 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`. The builder
+ // refuses the same set, so a new caller cannot land on a fail-open.
+ _ => match lookup_command(code) {
+ Some(meta) if meta.is_replicated() => Err(IggyError::FeatureUnavailable),
+ _ => Err(IggyError::InvalidCommand),
+ },
}
}
@@ -508,3 +412,168 @@ where
Some((stream_id, topic_id))
})
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::dispatch::test_support::{FIRST_BOOT, SpyBus, TestShard, test_shard};
+ use iggy_binary_protocol::COMMAND_TABLE;
+ use iggy_binary_protocol::codes::{
+ CREATE_STREAM_CODE, GET_CLIENT_CODE, GET_CLIENTS_CODE, GET_CONSUMER_OFFSET_CODE,
+ GET_ME_CODE, GET_SNAPSHOT_FILE_CODE, LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE,
+ LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, LOGOUT_USER_CODE, PING_CODE,
+ POLL_MESSAGES_CODE, SYNC_CONSUMER_GROUP_CODE,
+ };
+ use iggy_common::defaults::DEFAULT_ROOT_USER_ID;
+
+ /// A code no `COMMAND_TABLE` entry claims.
+ const UNKNOWN_CODE: u32 = 9999;
+
+ /// The gate's answer as the wire sees it: status 0 allows, anything else
+ /// is the deny code stamped into `ReplyHeader.status`.
+ #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+ enum Verdict {
+ Allow,
+ Deny(u32),
+ }
+
+ /// Shard with root seeded, so the root column below exercises the
+ /// permissioner rules rather than a missing-user deny.
+ fn gate_shard() -> Rc<TestShard> {
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .users()
+ .ensure_root_user("iggy", "hash");
+ shard
+ }
+
+ /// Gate a header-only frame (`body = &[]`) for `user_id`.
+ fn gate(shard: &Rc<TestShard>, code: u32, user_id: Option<u32>) -> Verdict {
+ match authorize_default_read(shard, code, &[], user_id) {
+ Ok(()) => Verdict::Allow,
+ Err(error) => Verdict::Deny(error.as_code()),
+ }
+ }
+
+ /// Expected verdicts per non-replicated code for a header-only frame, as
+ /// `(code, unbound caller, root)`. Header-only means the identifier-scoped
+ /// arms never decode a body and defer to the builder's own error for both
+ /// callers. Codes the reads router serves on its own arms (`PING`, `GET_ME`,
+ /// `GET_CLIENTS`, ...) and the codes the classifier settles (the legacy
+ /// logins) never reach the gate, which refuses them like any other armless
+ /// code.
+ fn expected_header_only_verdicts() -> Vec<(u32, Verdict, Verdict)> {
+ let allow = Verdict::Allow;
+ let unauthenticated = Verdict::Deny(IggyError::Unauthenticated.as_code());
+ let invalid_command = Verdict::Deny(IggyError::InvalidCommand.as_code());
+ let feature_unavailable = Verdict::Deny(IggyError::FeatureUnavailable.as_code());
+ vec![
+ (PING_CODE, invalid_command, invalid_command),
+ (GET_STATS_CODE, unauthenticated, allow),
+ (GET_SNAPSHOT_FILE_CODE, invalid_command, invalid_command),
+ (GET_CLUSTER_METADATA_CODE, unauthenticated, allow),
+ (GET_ME_CODE, invalid_command, invalid_command),
+ (GET_CLIENT_CODE, invalid_command, invalid_command),
+ (GET_CLIENTS_CODE, invalid_command, invalid_command),
+ (GET_USER_CODE, allow, allow),
+ (GET_USERS_CODE, unauthenticated, allow),
+ (LOGIN_USER_CODE, invalid_command, invalid_command),
+ (LOGOUT_USER_CODE, invalid_command, invalid_command),
+ (LOGIN_REGISTER_CODE, invalid_command, invalid_command),
+ (GET_PERSONAL_ACCESS_TOKENS_CODE, unauthenticated, allow),
+ (
+ LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE,
+ invalid_command,
+ invalid_command,
+ ),
+ (POLL_MESSAGES_CODE, invalid_command, invalid_command),
+ (
+ FLUSH_UNSAVED_BUFFER_CODE,
+ feature_unavailable,
+ feature_unavailable,
+ ),
+ (GET_CONSUMER_OFFSET_CODE, invalid_command, invalid_command),
+ (GET_STREAM_CODE, allow, allow),
+ (GET_STREAMS_CODE, unauthenticated, allow),
+ (GET_TOPIC_CODE, allow, allow),
+ (GET_TOPICS_CODE, allow, allow),
+ (GET_CONSUMER_GROUP_CODE, allow, allow),
+ (GET_CONSUMER_GROUPS_CODE, allow, allow),
+ (SYNC_CONSUMER_GROUP_CODE, invalid_command, invalid_command),
+ (
+ LOGIN_REGISTER_WITH_PAT_CODE,
+ invalid_command,
+ invalid_command,
+ ),
+ (DESCRIBE_OPTIONS_CODE, unauthenticated, allow),
+ ]
+ }
+
+ /// Every non-replicated table entry has a named decision above. A new
+ /// entry without a row fails by name, so it cannot slip past the gate
+ /// unnoticed; a row without a table entry is stale and fails too.
+ #[test]
+ fn gate_is_total_over_the_command_table() {
+ let shard = gate_shard();
+ let expected = expected_header_only_verdicts();
+ for meta in COMMAND_TABLE.iter().filter(|meta| !meta.is_replicated()) {
+ let Some((_, unbound, root)) = expected.iter().find(|row| row.0 == meta.code) else {
+ panic!(
+ "non-replicated command {} ({}) has no ratchet row: decide it in \
+ authorize_default_read and add the row",
+ meta.name, meta.code
+ );
+ };
+ assert_eq!(
+ gate(&shard, meta.code, None),
+ *unbound,
+ "{} ({}) from an unbound caller",
+ meta.name,
+ meta.code
+ );
+ assert_eq!(
+ gate(&shard, meta.code, Some(DEFAULT_ROOT_USER_ID)),
+ *root,
+ "{} ({}) from root",
+ meta.name,
+ meta.code
+ );
+ }
+ for (code, _, _) in &expected {
+ assert!(
+ lookup_command(*code).is_some_and(|meta| !meta.is_replicated()),
+ "ratchet row {code} names no non-replicated table entry"
+ );
+ }
+ }
+
+ #[test]
+ fn unknown_code_is_refused_as_invalid_command() {
+ assert!(
+ lookup_command(UNKNOWN_CODE).is_none(),
+ "test needs a code absent from COMMAND_TABLE"
+ );
+ let shard = gate_shard();
+ let invalid_command = Verdict::Deny(IggyError::InvalidCommand.as_code());
+ assert_eq!(gate(&shard, UNKNOWN_CODE, None), invalid_command);
+ assert_eq!(
+ gate(&shard, UNKNOWN_CODE, Some(DEFAULT_ROOT_USER_ID)),
+ invalid_command
+ );
+ }
+
+ #[test]
+ fn replicated_code_in_a_non_replicated_frame_is_refused_as_feature_unavailable() {
+ let shard = gate_shard();
+ let feature_unavailable = Verdict::Deny(IggyError::FeatureUnavailable.as_code());
+ assert_eq!(gate(&shard, CREATE_STREAM_CODE, None), feature_unavailable);
+ assert_eq!(
+ gate(&shard, CREATE_STREAM_CODE, Some(DEFAULT_ROOT_USER_ID)),
+ feature_unavailable
+ );
+ }
+}
diff --git a/core/server/src/dispatch/failure.rs b/core/server/src/dispatch/failure.rs
new file mode 100644
index 000000000..37bb5a5c6
--- /dev/null
+++ b/core/server/src/dispatch/failure.rs
@@ -0,0 +1,629 @@
+// 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, 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, an unresolved target, and a request the resolve rejects all deny 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, 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;
+}
+
+// Byte snapshots pinning each channel's frame to the pre-refactor inline
+// construction. What they hold is that routing a rejection through this
+// module changed no byte a client sees: same command, same status, same
+// header echo, same body length per channel. They are NOT the wire
+// contract - a deliberate protocol change updates them.
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::dispatch::handle_client_request;
+ use crate::dispatch::test_support::{
+ FIRST_BOOT, SpyBus, TestShard, request_message, test_shard,
+ };
+ use crate::responses::build_empty_reply;
+ use crate::session_manager::SessionManager;
+ use configs::server::ServerSystemConfig;
+ use iggy_binary_protocol::Operation;
+ use iggy_binary_protocol::codes::PING_CODE;
+ use iggy_common::RESYNC_REQUIRED_PARTITION_SENTINEL;
+ use std::cell::RefCell;
+ use std::sync::Arc;
+
+ const TRANSPORT: u128 = 42;
+ const VSR_CLIENT: u128 = 7;
+ const SESSION: u64 = 3;
+ const REQUEST: u64 = 5;
+
+ fn snapshot_shard() -> (SpyBus, Rc<TestShard>) {
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ (bus, shard)
+ }
+
+ fn sole_client_frame(bus: &SpyBus) -> (u128, Vec<u8>) {
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1, "expected exactly one client-bound frame");
+ replies[0].clone()
+ }
+
+ fn poll_request() -> Message<RoutedRequestHeader> {
+ request_message(Operation::NonReplicated, VSR_CLIENT, SESSION, REQUEST, &[])
+ }
+
+ /// The 16-byte empty `PolledMessages` body exactly as the pre-refactor
+ /// `partition::empty_polled_messages_body` built it.
+ fn old_empty_polled_messages_body(partition_id: u32) -> Bytes {
+ let mut body = Vec::with_capacity(16);
+ body.extend_from_slice(&partition_id.to_le_bytes());
+ body.extend_from_slice(&0u64.to_le_bytes());
+ body.extend_from_slice(&0u32.to_le_bytes());
+ Bytes::from(body)
+ }
+
+ fn old_eviction_context(shard: &Rc<TestShard>) -> EvictionContext {
+ shard.plane.metadata().consensus.as_ref().map_or(
+ EvictionContext {
+ cluster: 0,
+ view: 0,
+ replica: 0,
+ },
+ EvictionContext::from_consensus,
+ )
+ }
+
+ #[compio::test]
+ async fn snapshot_typed_deny_commit_stamped_frame_unchanged() {
+ let (bus, shard) = snapshot_shard();
+ let request = poll_request();
+ let status = IggyError::Unauthenticated.as_code();
+ let old = build_deny_reply(
+ request.header(),
+ TRANSPORT,
+ 0,
+ current_metadata_commit(&shard),
+ status,
+ )
+ .into_generic();
+
+ send_deny_reply(&shard, TRANSPORT, request.header(), status).await;
+
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ }
+
+ #[compio::test]
+ async fn snapshot_typed_deny_unbound_frame_unchanged() {
+ let (bus, shard) = snapshot_shard();
+ let request = poll_request();
+ let status = IggyError::Unauthenticated.as_code();
+ let old = build_deny_reply(request.header(), TRANSPORT, 0, 0, status).into_generic();
+
+ send_unbound_deny_reply(&shard, TRANSPORT, request.header(), status).await;
+
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ }
+
+ #[compio::test]
+ async fn snapshot_typed_deny_non_replicated_frame_unchanged() {
+ let (bus, shard) = snapshot_shard();
+ let request = poll_request();
+ let status = IggyError::Unauthenticated.as_code();
+ let old = build_deny_reply(
+ request.header(),
+ request.header().client,
+ request.header().session,
+ current_metadata_commit(&shard),
+ status,
+ )
+ .into_generic();
+
+ send_non_replicated_deny(&shard, &request, TRANSPORT, status).await;
+
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ }
+
+ #[compio::test]
+ async fn snapshot_typed_deny_result_framed_frame_unchanged() {
+ let (bus, shard) = snapshot_shard();
+ let request = poll_request();
+ let code = IggyError::TransientNotAccepted;
+ let old = build_result_rejection_reply(
+ request.header(),
+ current_metadata_commit(&shard),
+ code.as_code(),
+ )
+ .into_generic();
+
+ send_result_rejection(&shard, TRANSPORT, request.header(), &code, "snapshot").await;
+
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ }
+
+ #[compio::test]
+ async fn snapshot_eviction_frame_unchanged() {
+ let (bus, shard) = snapshot_shard();
+ let ctx = old_eviction_context(&shard);
+
+ // Old `send_unauthenticated_eviction`: reason NoSession, client id =
+ // the transport id.
+ let old = build_eviction_message(ctx, TRANSPORT, EvictionReason::NoSession).into_generic();
+ send_eviction(
+ &shard,
+ TRANSPORT,
+ TRANSPORT,
+ EvictionReason::NoSession,
+ "snapshot",
+ )
+ .await;
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ bus.client_replies.borrow_mut().clear();
+
+ // Old `send_login_eviction`: reason MalformedLogin, client id = the
+ // request's VSR client id.
+ let old =
+ build_eviction_message(ctx, VSR_CLIENT, EvictionReason::MalformedLogin).into_generic();
+ send_eviction(
+ &shard,
+ TRANSPORT,
+ VSR_CLIENT,
+ EvictionReason::MalformedLogin,
+ "snapshot",
+ )
+ .await;
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ bus.client_replies.borrow_mut().clear();
+
+ // Old `send_login_eviction` on `IncompatibleProtocol`: the protocol
+ // window rides the frame, client id = the request's VSR client id.
+ let old = build_incompatible_protocol_eviction_message(ctx, VSR_CLIENT).into_generic();
+ send_eviction(
+ &shard,
+ TRANSPORT,
+ VSR_CLIENT,
+ EvictionReason::IncompatibleProtocol,
+ "snapshot",
+ )
+ .await;
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ }
+
+ /// The HTTP mirror (`resync_required_polled_messages` in
+ /// `crate::http::wire`) is a JSON DTO, not a wire frame, so the shared
+ /// contract asserted here is the sentinel constant itself; the DTO's own
+ /// test pins its `partition_id` to the same constant.
+ #[compio::test]
+ async fn snapshot_resync_sentinel_frame_unchanged() {
+ let (bus, shard) = snapshot_shard();
+ let request = poll_request();
+ let body = old_empty_polled_messages_body(RESYNC_REQUIRED_PARTITION_SENTINEL);
+ assert_eq!(
+ body[..4],
+ RESYNC_REQUIRED_PARTITION_SENTINEL.to_le_bytes(),
+ "sentinel poll body must lead with the re-sync sentinel partition id"
+ );
+ let old = NonReplicatedResponse::Bytes(body.clone())
+ .into_reply(
+ request.header(),
+ request.header().client,
+ request.header().session,
+ current_metadata_commit(&shard),
+ )
+ .into_generic();
+
+ send_non_replicated_bytes(
+ &shard,
+ &request,
+ TRANSPORT,
+ body,
+ FrameChannel::ResyncSentinel,
+ "poll_messages",
+ )
+ .await;
+
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ }
+
+ /// The PING reply is the one host-built success Reply the funnel serves
+ /// without a consensus round; the old side is the frame the reads
+ /// router's PING arm built inline before the exit existed.
+ #[compio::test]
+ async fn snapshot_reply_frame_unchanged() {
+ let (bus, shard) = snapshot_shard();
+ let sessions = Rc::new(RefCell::new(SessionManager::new()));
+ let system_config = Arc::new(ServerSystemConfig::default());
+ let request = request_message(Operation::NonReplicated, VSR_CLIENT, SESSION, REQUEST, &[])
+ .transmute_header(|header, ping: &mut RoutedRequestHeader| {
+ *ping = header;
+ ping.reserved[..4].copy_from_slice(&PING_CODE.to_le_bytes());
+ // The funnel promotes the client wire header with `group`
+ // unset, so the old side must build from the same bytes.
+ ping.group = 0;
+ });
+ let old = build_empty_reply(
+ request.header(),
+ request.header().client,
+ request.header().session,
+ current_metadata_commit(&shard),
+ )
+ .into_generic();
+
+ handle_client_request(
+ &shard,
+ &sessions,
+ &system_config,
+ 1,
+ TRANSPORT,
+ request.into_generic(),
+ )
+ .await;
+
+ let (target, frame) = sole_client_frame(&bus);
+ assert_eq!(target, TRANSPORT);
+ assert_eq!(frame, old.as_slice().to_vec());
+ }
+}
diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs
index 68053050f..50b40dcd2 100644
--- a/core/server/src/dispatch/mod.rs
+++ b/core/server/src/dispatch/mod.rs
@@ -20,7 +20,9 @@
//! The tree: [`session_ops`] (login/register/logout and their replica
//! forwards), [`partition`] (the partition data plane, both mesh ends),
//! [`reads`] (the non-replicated read router), [`submit`] (the shard-0
-//! metadata-submit RPC), `authz` (the wire-path authorization gates).
+//! metadata-submit RPC), `authz` (the wire-path authorization gates),
+//! `failure` (the wire failure channels and the one send exit for
+//! host-built frames).
//!
//! Deliberate asymmetry (the two authz gates): replicated metadata ops are
//! authorized in-apply by the STM, in committed order on every replica;
@@ -30,6 +32,7 @@
//! error contract (404-before-403) is pinned client-visible behavior.
mod authz;
+mod failure;
pub mod login_error;
pub mod partition;
pub mod reads;
@@ -39,61 +42,44 @@ pub mod submit;
mod test_support;
use crate::consumer_group::maybe_rewrite_consumer_group_request;
-use crate::dispatch::authz::{send_deny_reply, send_unbound_deny_reply};
+use crate::dispatch::failure::{
+ FrameChannel, send_deny_reply, send_eviction, send_host_frame, send_pre_consensus_deny,
+ send_unbound_deny_reply,
+};
use crate::dispatch::partition::{dispatch_partition_request, handle_delete_segments_request};
use crate::dispatch::reads::handle_non_replicated_request;
use crate::dispatch::session_ops::{
- handle_login_register_request, handle_logout_request, send_login_eviction,
- send_unauthenticated_eviction, submit_disconnect_logout,
+ handle_login_register_request, handle_logout_request, submit_disconnect_logout,
};
use crate::dispatch::submit::{committed_reply_commit, submit_client_request_on_owner};
-use crate::pat::maybe_rewrite_pat_request;
-use crate::responses::{
- NonReplicatedResponse, build_deny_reply, build_raw_pat_reply, current_metadata_commit,
-};
-use crate::segment_cleaner::UNENFORCEABLE_TOPIC_SIZE_WARN;
-use crate::session_manager::SessionManager;
+use crate::responses::build_raw_pat_reply;
+use crate::rewrite::{RewriteDeny, RewriteStage, tcp_chain};
+use crate::session_manager::{ConnectionContext, SessionManager};
use crate::shell::{ShellBus, ShellShard, ShellShardHandle};
-use crate::users::maybe_rewrite_user_password_request;
-use crate::wire::{request_body, verify_request_checksum};
-use bytes::Bytes;
+use crate::wire::verify_request_checksum;
+use ahash::{AHashMap, AHashSet};
use configs::server::ServerSystemConfig;
-use consensus::MetadataHandle;
use iggy_binary_protocol::PrepareHeader;
use iggy_binary_protocol::codes::{
- GET_CLUSTER_METADATA_CODE, LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE,
+ LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE,
};
-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::{
- EvictionReason, GenericHeader, MAX_PARTITIONS_PER_REQUEST, Operation, RequestHeader,
- 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,
+ EvictionReason, GenericHeader, Operation, RequestHeader, RoutedRequestHeader,
};
+use iggy_common::IggyError;
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
-use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
use message_bus::replica::listener::MessageHandler;
-use metadata::impls::metadata::StreamsFrontend;
-use metadata::stm::stream::Streams;
use server_common::Message;
use shard::{ConnectedClientInfo, ListClientsHandler};
use std::cell::RefCell;
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::collections::VecDeque;
use std::rc::Rc;
use std::sync::Arc;
-use tracing::{debug, warn};
+use tracing::{debug, error, warn};
-type ClientRequestQueues = Rc<RefCell<HashMap<u128, VecDeque<Message<GenericHeader>>>>>;
+type ClientRequestQueues = Rc<RefCell<AHashMap<u128, VecDeque<Message<GenericHeader>>>>>;
/// Requests one client may have queued behind a request this shard has not
/// answered yet.
@@ -109,50 +95,7 @@ type ClientRequestQueues = Rc<RefCell<HashMap<u128, VecDeque<Message<GenericHead
/// re-issue it anywhere, including here once the queue drains. Sized far above
/// any SDK's in-flight window, so it only ever fires under a genuine stall.
const MAX_QUEUED_CLIENT_REQUESTS: usize = 1024;
-type ActiveClientRequests = Rc<RefCell<HashSet<u128>>>;
-
-pub fn make_client_request_handler<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- sessions: &Rc<RefCell<SessionManager>>,
- system_config: Arc<ServerSystemConfig>,
- max_tokens_per_user: u32,
-) -> RequestHandler
-where
- B: ShellBus,
- MJ: JournalHandle + 'static,
- MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
- S: 'static,
- SB: SuperblockStore + 'static,
-{
- let shard = Rc::clone(shard);
- let sessions = Rc::clone(sessions);
- let queues: ClientRequestQueues = Rc::new(RefCell::new(HashMap::new()));
- let active: ActiveClientRequests = Rc::new(RefCell::new(HashSet::new()));
- let sessions_for_disconnect = Rc::clone(&sessions);
- let shard_for_disconnect = Rc::clone(&shard);
- shard
- .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)
- {
- submit_disconnect_logout(Rc::clone(&shard_for_disconnect), vsr_client_id, session);
- }
- }));
- Rc::new(move |client_id, message| {
- enqueue_client_request(
- Rc::clone(&shard),
- Rc::clone(&sessions),
- Arc::clone(&system_config),
- max_tokens_per_user,
- Rc::clone(&queues),
- Rc::clone(&active),
- client_id,
- message,
- );
- })
-}
+type ActiveClientRequests = Rc<RefCell<AHashSet<u128>>>;
/// Build the per-shard [`ListClientsHandler`]: on a `ListClients`
/// broadcast, serialize this shard's locally-homed connected clients from
@@ -187,6 +130,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.
pub fn make_deferred_client_request_handler<B, MJ, S, SB>(
bus: &B,
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
@@ -203,50 +152,55 @@ 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 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| {
+ // The socket is gone, so nothing will drain what a live drain task
+ // left queued. The active slot is NOT released here: the transport
+ // task runs this hook while a drain may be suspended at an `.await`,
+ // and clearing the slot would let a frame the dispatch task still has
+ // buffered spawn a second drain over the same queue. The drain task's
+ // own guard covers every exit, the panic compio catches included.
+ queues_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.
+ let Some(shard) = upgrade_shard_handle(&shard_handle_for_disconnect) else {
+ // Nothing reaps what stays behind: the heartbeat verifier is
+ // optional and only collects `Bound` / `Authenticated` sessions,
+ // so a `Connected` row survives to process exit.
+ error!(
+ client_id,
+ "client connection lost with no live shard; session and client-table entries \
+ leak until process exit"
+ );
+ return;
+ };
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)
{
submit_disconnect_logout(shard, vsr_client_id, session);
}
}));
Rc::new(move |client_id, message| {
- 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);
- queues
- .borrow_mut()
- .entry(client_id)
- .or_default()
- .push_back(message);
- if !active.borrow_mut().insert(client_id) {
- return;
- }
- bus_for_spawn.spawn(async move {
- let Some(shard) = upgrade_shard_handle(&shard_handle) else {
- active.borrow_mut().remove(&client_id);
- return;
- };
- drain_client_requests(
- shard,
- sessions,
- system_config,
- max_tokens_per_user,
- queues,
- active,
- client_id,
- )
- .await;
- });
+ enqueue_client_request(
+ &bus_for_spawn,
+ &shard_handle,
+ &sessions,
+ &system_config,
+ max_tokens_per_user,
+ &queues,
+ &active,
+ client_id,
+ message,
+ );
})
}
@@ -290,12 +244,13 @@ where
#[allow(clippy::too_many_arguments)]
fn enqueue_client_request<B, MJ, S, SB>(
- shard: Rc<ShellShard<B, MJ, S, SB>>,
- sessions: Rc<RefCell<SessionManager>>,
- system_config: Arc<ServerSystemConfig>,
+ bus: &B,
+ shard_handle: &ShellShardHandle<B, MJ, S, SB>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ system_config: &Arc<ServerSystemConfig>,
max_tokens_per_user: u32,
- queues: ClientRequestQueues,
- active: ActiveClientRequests,
+ queues: &ClientRequestQueues,
+ active: &ActiveClientRequests,
client_id: u128,
message: Message<GenericHeader>,
) where
@@ -312,7 +267,7 @@ fn enqueue_client_request<B, MJ, S, SB>(
// Borrow released before the deny, which spawns onto this same
// task and would otherwise re-enter the table.
drop(queues);
- deny_overflowing_client_request(&shard, client_id, message);
+ deny_overflowing_client_request(shard_handle, client_id, message);
return;
}
queue.push_back(message);
@@ -321,15 +276,25 @@ 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 {
+ let _slot = ActiveDrainSlot { active, client_id };
+ // The handle is set once the shard is built. A frame that beats it
+ // stays queued and the client's next frame drains both, which needs
+ // the slot released on this path too - the guard above does it.
+ let Some(shard) = upgrade_shard_handle(&shard_handle) else {
+ return;
+ };
drain_client_requests(
shard,
sessions,
system_config,
max_tokens_per_user,
queues,
- active,
client_id,
)
.await;
@@ -340,10 +305,12 @@ fn enqueue_client_request<B, MJ, S, SB>(
/// [`MAX_QUEUED_CLIENT_REQUESTS`] with the retryable transient denial.
///
/// Spawned rather than awaited: the enqueue path is sync (it runs straight off
-/// frame arrival) and the reply goes out on the bus. A frame whose header will
-/// not even cast is dropped instead, exactly as the drain loop drops it.
+/// frame arrival) and the reply goes out on the bus. Two shapes are dropped
+/// rather than answered: a frame whose header will not even cast, exactly as
+/// the drain loop drops it, and a frame that arrives before the shard is
+/// built, which leaves nothing to render a reply from.
fn deny_overflowing_client_request<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ shard_handle: &ShellShardHandle<B, MJ, S, SB>,
transport_client_id: u128,
message: Message<GenericHeader>,
) where
@@ -353,6 +320,17 @@ fn deny_overflowing_client_request<B, MJ, S, SB>(
S: 'static,
SB: SuperblockStore + 'static,
{
+ // Unreachable in practice: a listener binds only after the build backfills
+ // the weak self-reference, which `crate::boot` pins with a `debug_assert`.
+ // Dropped rather than admitted past the cap -- the cap exists to refuse
+ // this frame, and the deny itself needs the shard for its metric and reply.
+ let Some(shard) = upgrade_shard_handle(shard_handle) else {
+ warn!(
+ transport_client_id,
+ "dropping over-queue client request received before the shard was built"
+ );
+ return;
+ };
shard.metrics().record_client_request_denied_queue_full();
let Ok(request) = message.try_into_typed::<RequestHeader>() else {
warn!(
@@ -368,7 +346,6 @@ fn deny_overflowing_client_request<B, MJ, S, SB>(
queued = MAX_QUEUED_CLIENT_REQUESTS,
"denying client request retryable: this connection's request queue is full"
);
- let shard = Rc::clone(shard);
let bus = shard.bus.clone();
bus.spawn(async move {
send_deny_reply(
@@ -381,6 +358,24 @@ fn deny_overflowing_client_request<B, MJ, S, SB>(
});
}
+/// Holds a client's one drain slot for as long as its drain task lives.
+///
+/// Releasing from the task's own `Drop` covers every exit, the panic compio
+/// catches included: a slot left taken with no live drain queues every later
+/// frame for that client forever. It also keeps the release out of the
+/// connection-lost hook, which fires from the transport task and could
+/// otherwise clear the slot under a drain suspended at an `.await`.
+struct ActiveDrainSlot {
+ active: ActiveClientRequests,
+ client_id: u128,
+}
+
+impl Drop for ActiveDrainSlot {
+ fn drop(&mut self) {
+ self.active.borrow_mut().remove(&self.client_id);
+ }
+}
+
#[allow(clippy::future_not_send)]
async fn drain_client_requests<B, MJ, S, SB>(
shard: Rc<ShellShard<B, MJ, S, SB>>,
@@ -388,7 +383,6 @@ async fn drain_client_requests<B, MJ, S, SB>(
system_config: Arc<ServerSystemConfig>,
max_tokens_per_user: u32,
queues: ClientRequestQueues,
- active: ActiveClientRequests,
client_id: u128,
) where
B: ShellBus,
@@ -398,7 +392,7 @@ async fn drain_client_requests<B, MJ, S, SB>(
SB: SuperblockStore + 'static,
{
loop {
- let Some(message) = pop_next_client_request(&queues, &active, client_id) else {
+ let Some(message) = pop_next_client_request(&queues, client_id) else {
return;
};
handle_client_request(
@@ -415,211 +409,110 @@ async fn drain_client_requests<B, MJ, S, SB>(
fn pop_next_client_request(
queues: &ClientRequestQueues,
- active: &ActiveClientRequests,
client_id: u128,
) -> Option<Message<GenericHeader>> {
+ // Freed as soon as it drains empty. Keeping it would cost the connection
+ // one `VecDeque` (at its peak depth, since `pop_front` never gives
+ // capacity back), and the connection-lost hook cannot be the sole owner:
+ // the dispatch task can still deliver a buffered frame after the hook
+ // ran, and `client_id` is never reused, so that entry would live until
+ // process exit.
let mut queues = queues.borrow_mut();
- let Some(queue) = queues.get_mut(&client_id) else {
- active.borrow_mut().remove(&client_id);
- return None;
- };
+ let queue = queues.get_mut(&client_id)?;
let message = queue.pop_front();
if queue.is_empty() {
queues.remove(&client_id);
}
- if message.is_none() {
- active.borrow_mut().remove(&client_id);
- }
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(in crate::dispatch) enum RequestClass {
+ /// Legacy pre-register login code: rejected with a typed
+ /// `MalformedLogin` eviction before the session gate.
+ LegacyLogin,
+ /// Non-replicated code other than PING on an unbound transport:
+ /// denied `Unauthenticated` (plain deny reply, never an eviction).
+ UnauthenticatedRead,
+ /// Non-replicated read for the reads router.
+ NonReplicatedRead,
+ /// The register handshake (`session == 0 && request == 0`).
+ LoginRegister,
+ Logout,
+ /// Replicated operation on an unbound transport: `Eviction(NoSession)`.
+ UnboundReplicated,
+ /// Neither a partition nor a metadata consensus op: resolved to a
+ /// replicated `TruncatePartition` by the owning shard.
+ DeleteSegments,
+ /// Partition-plane operation.
+ Partition,
+ /// Everything else: replicated metadata consensus.
+ ReplicatedMetadata,
}
-/// [`validate_partitions_count`] plus the zero rejection that create-partitions
-/// and delete-partitions carry: adding or removing zero partitions is a no-op
-/// that would still burn a replicated log entry, bump `Streams::revision` and
-/// force every shard through a rebalance pass. Legacy rejects it with
-/// `TooManyPartitions` in both handlers (`1..=MAX` on create, `== 0` on
-/// delete), so the code matches rather than inventing a new one.
-const fn validate_partitions_change_count(partitions_count: u32) -> Result<(), IggyError> {
- if partitions_count == 0 {
- return Err(IggyError::TooManyPartitions);
+/// Route a client request: [`handle_client_request`] matches on the result,
+/// so this function IS the routing and the order of the checks below is the
+/// semantics. Pure: `bound` stands for
+/// `sessions.get_session(transport_client_id).is_some()`, the only session
+/// fact the checks consult.
+///
+/// The pins, in check order:
+/// - the legacy-login rejection precedes the session gate: a legacy code
+/// must get the typed `MalformedLogin` eviction, not the generic
+/// unauthenticated deny;
+/// - the pre-auth allowlist is PING only; `GET_CLUSTER_METADATA` is
+/// deliberately NOT pre-auth (see the funnel's auth-bypass guard);
+/// - poll-messages and consumer-offset reads are non-replicated CODES, not
+/// partition operations: they classify
+/// [`RequestClass::NonReplicatedRead`] and route inside the reads router;
+/// - a `Register` with `session != 0` or `request != 0` falls through to
+/// the default. Those rows are classify-only: `RequestHeader::validate`
+/// rejects such a header before the funnel sees it, so no mid-session
+/// register ever reaches consensus;
+/// - `DeleteSegments` is neither a partition nor a metadata op, so its
+/// check sits before `is_partition`;
+/// - the checksum and heartbeat pre-gates run BEFORE classification in the
+/// funnel.
+pub(in crate::dispatch) fn classify(header: &RoutedRequestHeader, bound: bool) -> RequestClass {
+ if header.operation == Operation::NonReplicated {
+ let nr_code = non_replicated_code(header);
+ if matches!(
+ nr_code,
+ LOGIN_USER_CODE | LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE
+ ) {
+ return RequestClass::LegacyLogin;
+ }
+ if nr_code != PING_CODE && !bound {
+ return RequestClass::UnauthenticatedRead;
+ }
+ return RequestClass::NonReplicatedRead;
}
- validate_partitions_count(partitions_count)
-}
-
-/// Static create-topic bounds shared by the TCP and HTTP ingresses. Runs
-/// pre-consensus: a rejected request must not burn a replicated log entry,
-/// and `prepare_request` errors evict the session instead of denying typed.
-/// `ServerDefault` is exempt from the size floor (it resolves against server
-/// config at admission, matching legacy); `Unlimited` passes numerically.
-/// `segment_size_bytes` is the topic's RESOLVED segment size (explicit
-/// option, else this node's default), so a per-topic segment above the
-/// global default still floors the topic cap.
-pub fn validate_topic_bounds(
- partitions_count: u32,
- max_topic_size: MaxTopicSize,
- segment_size_bytes: u64,
-) -> Result<(), IggyError> {
- validate_partitions_count(partitions_count)?;
- validate_topic_size_floor(max_topic_size, segment_size_bytes)
-}
-
-/// A topic cap below one segment can never be enforced: the first segment
-/// already exceeds it. Split out of [`validate_topic_bounds`] because update
-/// admission checks the cap without a partitions count to check.
-pub fn validate_topic_size_floor(
- max_topic_size: MaxTopicSize,
- segment_size_bytes: u64,
-) -> Result<(), IggyError> {
- if !matches!(max_topic_size, MaxTopicSize::ServerDefault)
- && max_topic_size.as_bytes_u64() < segment_size_bytes
- {
- return Err(IggyError::InvalidTopicSize(
- max_topic_size,
- IggyByteSize::from(segment_size_bytes),
- ));
+ if header.operation == Operation::Register && header.session == 0 && header.request == 0 {
+ return RequestClass::LoginRegister;
}
- Ok(())
-}
-
-/// Announce an accepted `max_topic_size` the server cannot enforce as written.
-///
-/// [`validate_topic_size_floor`] admits any cap of one segment or more, but
-/// retention runs PER PARTITION and floors each partition's share at one SEALED
-/// segment, which reaches up to one maximum bus frame past `segment_size`. A cap
-/// between the two is stored and echoed back verbatim while the server actually
-/// keeps `(segment_size + max_message_size) * partitions_count`, so the only
-/// moment an operator can be told is the one where they set it.
-///
-/// Warns rather than rejects: which caps are accepted is client-visible wire
-/// behavior, and tightening it would break topics that already exist.
-pub fn warn_unenforceable_topic_size(
- max_topic_size: MaxTopicSize,
- segment_size_bytes: u64,
- max_message_size_bytes: usize,
- partitions_count: u32,
-) {
- let MaxTopicSize::Custom(configured) = max_topic_size else {
- return;
- };
- let max_message_size_bytes = u64::try_from(max_message_size_bytes).unwrap_or(u64::MAX);
- let per_partition_floor = segment_size_bytes.saturating_add(max_message_size_bytes);
- let topic_floor = per_partition_floor.saturating_mul(u64::from(partitions_count));
- if configured.as_bytes_u64() >= topic_floor {
- return;
+ if header.operation == Operation::Logout {
+ return RequestClass::Logout;
}
- warn!(
- max_topic_size = configured.as_bytes_u64(),
- partitions_count,
- segment_size = segment_size_bytes,
- enforced_per_partition = per_partition_floor,
- "{UNENFORCEABLE_TOPIC_SIZE_WARN}"
- );
-}
-
-/// Announce the same unenforceable cap when partitions are ADDED to a topic.
-///
-/// The cap is topic-wide but enforcement is per partition, so every added
-/// partition shrinks the share: a cap that cleared the floor when the topic was
-/// created can stop clearing it here. The request carries only the delta, so
-/// the stored cap, segment size and current partition count come from metadata.
-pub fn warn_unenforceable_topic_size_on_partition_add(
- streams: &Streams,
- stream_id: &WireIdentifier,
- topic_id: &WireIdentifier,
- max_message_size_bytes: usize,
- added_partitions_count: u32,
-) {
- let Some(((stream_slab, topic_slab), _)) = streams.partition_count_context(stream_id, topic_id)
- else {
- return;
- };
- let Some((_, max_topic_size, partitions_count, segment_size)) =
- streams.topic_retention_config(stream_slab, topic_slab)
- else {
- return;
- };
- warn_unenforceable_topic_size(
- max_topic_size,
- segment_size.map_or(iggy_common::DEFAULT_SEGMENT_SIZE, |segment_size| {
- segment_size.as_bytes_u64()
- }),
- max_message_size_bytes,
- u32::try_from(partitions_count)
- .unwrap_or(u32::MAX)
- .saturating_add(added_partitions_count),
- );
-}
-
-/// Reject option keys outside the resource's catalog, pre-consensus. Unknown
-/// keys are rejected rather than skipped: a silently ignored knob would hand
-/// the client server defaults without it ever learning. Streams and users
-/// have no catalog keys yet, so `known` is empty for both until one lands.
-pub fn validate_option_keys(options: &WireOptions, known: &[&str]) -> Result<(), IggyError> {
- for entry in options {
- // Wire validation already enforced UTF-8 string keys.
- let key = String::from_utf8_lossy(entry.key);
- if !known.contains(&key.as_ref()) {
- return Err(IggyError::UnsupportedOptionKey(key.into_owned()));
- }
+ if !bound {
+ return RequestClass::UnboundReplicated;
+ }
+ if header.operation == Operation::DeleteSegments {
+ return RequestClass::DeleteSegments;
}
- Ok(())
+ if header.operation.is_partition() {
+ return RequestClass::Partition;
+ }
+ RequestClass::ReplicatedMetadata
}
-/// 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)]
-async fn send_pre_consensus_deny<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- header: &RoutedRequestHeader,
- transport_client_id: u128,
- 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 = ?header.operation,
- context,
- "denying request pre-consensus"
- );
- let commit = current_metadata_commit(shard);
- let reply = build_deny_reply(header, transport_client_id, 0, commit, error.as_code());
- if let Err(send_error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %send_error,
- context,
- "failed to send pre-consensus deny reply"
- );
- }
+/// The command code a `NonReplicated` header carries in its first four
+/// reserved bytes.
+fn non_replicated_code(header: &RoutedRequestHeader) -> u32 {
+ u32::from_le_bytes(header.reserved[..4].try_into().unwrap())
}
#[allow(clippy::future_not_send, clippy::too_many_lines)]
@@ -673,69 +566,66 @@ async fn handle_client_request<B, MJ, S, SB>(
return;
}
- ensure_transport_connection(shard, sessions, transport_client_id);
-
- // Any request is liveness proof, not just PING: an idle-but-active client
- // (e.g. an admin issuing reads between long sleeps) must not be evicted by
- // the heartbeat verifier. A genuinely dead connection sends nothing, so the
- // intended stale-client eviction still fires. No-ops for an unbound client.
- sessions.borrow_mut().record_heartbeat(transport_client_id);
+ // ONE `connections` walk for the whole prologue. Any request is liveness
+ // proof, not just PING: an idle-but-active client (e.g. an admin issuing
+ // reads between long sleeps) must not be evicted by the heartbeat
+ // verifier. A genuinely dead connection sends nothing, so the intended
+ // stale-client eviction still fires.
+ // Bound, not matched: a `match` on `borrow_mut()` would hold the guard
+ // across the arm that borrows again.
+ let mut touched = sessions.borrow_mut().touch_connection(transport_client_id);
+ if touched.is_none() {
+ // A transport's first frame: the peer address and transport kind live
+ // on the bus, so only this path pays that lookup.
+ ensure_transport_connection(shard, sessions, transport_client_id);
+ touched = sessions.borrow_mut().touch_connection(transport_client_id);
+ }
+ let ConnectionContext {
+ bound,
+ user_id,
+ address: client_address,
+ metadata_watermark,
+ } = touched.unwrap_or_default();
- 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
- ) {
+ // Borrowed, not copied: the 256-byte header is only worth a by-value
+ // snapshot where an arm rewrites it (`ReplicatedMetadata`) and still has
+ // to echo the client's original fields on a deny.
+ let header = request.header();
+ 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() {
- // Foreign SDKs still probe `GET_CLUSTER_METADATA` before login
- // until they are fixed, so that rejection is routine traffic and
- // logs at debug rather than warn.
- if nr_code == GET_CLUSTER_METADATA_CODE {
- debug!(
- transport_client_id,
- "denying pre-auth cluster-metadata read with Unauthenticated"
- );
- } else {
- warn!(
- transport_client_id,
- code = nr_code,
- "denying pre-auth non-replicated read with Unauthenticated"
- );
- }
+ RequestClass::UnauthenticatedRead => {
+ let nr_code = non_replicated_code(header);
+ // No per-code exemption: every in-tree SDK reads the roster only
+ // after login, so an unauthenticated roster read is a real event
+ // and not something to hide at debug.
+ warn!(
+ transport_client_id,
+ code = nr_code,
+ "denying pre-auth non-replicated read with Unauthenticated"
+ );
// A plain deny Reply, not an Eviction: there is no session to
// evict, and an Eviction is session-terminal by wire contract,
// so SDKs would tear down the very connection their login is
@@ -748,360 +638,203 @@ 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, metadata_watermark),
)
.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) => {
- // Recorded before the reply reaches the socket, so a read the client
- // sends the instant it decodes this frame already sees the mark.
- if let Some(commit) = committed_reply_commit(&reply) {
- sessions
- .borrow_mut()
- .record_metadata_watermark(transport_client_id, commit);
- }
- // 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) => {
+ // Recorded before the reply reaches the socket, so a read the client
+ // sends the instant it decodes this frame already sees the mark.
+ if let Some(commit) = committed_reply_commit(&reply) {
+ sessions
+ .borrow_mut()
+ .record_metadata_watermark(transport_client_id, commit);
+ }
+ // 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. The raw
+ // token is unrecoverable either way -- it lives in
+ // that one reply and a retry dedups to the empty
+ // committed body -- so the caller must delete the
+ // token and mint a new one.
+ send_deny_reply(shard, transport_client_id, &header, error.as_code())
+ .await;
+ return;
+ }
+ };
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_frozen(),
+ FrameChannel::Reply,
+ "committed_reply",
+ )
+ .await;
+ }
+ None => {
+ // Transient submit failure (not primary / not caught up / dedup
+ // absorbed). Stay silent; the SDK read-timeout replays.
+ warn!(
+ transport_client_id,
+ operation = ?header.operation,
+ "replicated request not committed (transient); client will replay"
+ );
+ }
}
}
- None => {
- // Transient submit failure (not primary / not caught up / dedup
- // absorbed). Stay silent; the SDK read-timeout replays.
- warn!(
- transport_client_id,
- operation = ?header.operation,
- "replicated request not committed (transient); client will replay"
- );
- }
- }
-}
-
-/// Send a non-replicated reply body to a client, stamping the current
-/// metadata commit. Shared by the `get_me` / `get_clients` / `get_client`
-/// arms.
-#[allow(clippy::future_not_send)]
-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,
- label: &'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_reply_frame(
- shard,
- transport_client_id,
- reply.into_generic().into_frozen(),
- label,
- )
- .await;
-}
-
-/// Hand a built reply frame to the bus for `transport_client_id`.
-#[allow(clippy::future_not_send)]
-async fn send_reply_frame<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- transport_client_id: u128,
- frame: impl Into<BusMessage>,
- label: &'static str,
-) where
- B: ShellBus,
- MJ: JournalHandle + 'static,
- MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
- S: 'static,
- SB: SuperblockStore + 'static,
-{
- if let Err(error) = shard.bus.send_to_client(transport_client_id, frame).await {
- warn!(transport_client_id, label, error = %error, "failed to send non-replicated reply");
}
}
@@ -1146,7 +879,13 @@ mod tests {
use crate::cluster_meta::ClusterRoster;
use crate::dispatch::test_support::{FIRST_BOOT, SpyBus, TestMux, TestShard, test_shard};
use iggy_binary_protocol::Command;
+ use iggy_binary_protocol::codes::{
+ GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE, POLL_MESSAGES_CODE,
+ };
+ use iggy_binary_protocol::{EvictionHeader, ReplyHeader};
+ use journal::prepare_journal::PrepareJournal;
use metadata::IggyMetadata;
+ use metadata::impls::metadata::IggySnapshot;
use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig};
use server_common::MESSAGE_ALIGN;
use server_common::sharding::ShardId;
@@ -1299,7 +1038,6 @@ mod tests {
#[compio::test]
async fn pre_auth_cluster_metadata_denied_on_every_roster() {
use configs::cluster::{ClusterNodeConfig, TransportPorts};
- use iggy_binary_protocol::codes::GET_CLUSTER_METADATA_CODE;
use iggy_binary_protocol::{GenericHeader, ReplyHeader};
const TRANSPORT: u128 = 91;
@@ -1400,89 +1138,467 @@ mod tests {
}
}
+ /// Client-wire frame for the funnel tests, shaped like the SDK sends it
+ /// (the funnel promotes it to the routed shape itself).
+ fn wire_request(
+ operation: Operation,
+ client: u128,
+ session: u64,
+ request: u64,
+ body: &[u8],
+ ) -> Message<RequestHeader> {
+ let header_size = size_of::<RequestHeader>();
+ let total = header_size + body.len();
+ let mut message = Message::<RequestHeader>::new(total);
+ {
+ let slice = message.as_mut_slice();
+ slice[header_size..total].copy_from_slice(body);
+ let header =
+ bytemuck::checked::from_bytes_mut::<RequestHeader>(&mut slice[..header_size]);
+ *header = RequestHeader {
+ command: Command::Request,
+ operation,
+ size: u32::try_from(total).expect("test request fits u32"),
+ client,
+ session,
+ request,
+ ..Default::default()
+ };
+ }
+ message
+ }
+
+ fn non_replicated_request(client: u128, nr_code: u32) -> Message<GenericHeader> {
+ let mut message = wire_request(Operation::NonReplicated, client, 0, 0, &[]);
+ {
+ let header_size = size_of::<RequestHeader>();
+ let header = bytemuck::checked::from_bytes_mut::<RequestHeader>(
+ &mut message.as_mut_slice()[..header_size],
+ );
+ header.reserved[..4].copy_from_slice(&nr_code.to_le_bytes());
+ }
+ message.into_generic()
+ }
+
+ const fn frame_command(frame: &[u8]) -> u8 {
+ frame[std::mem::offset_of!(GenericHeader, command)]
+ }
+
+ const fn eviction_reason_byte(frame: &[u8]) -> u8 {
+ frame[std::mem::offset_of!(EvictionHeader, reason)]
+ }
+
+ fn reply_status(frame: &[u8]) -> u32 {
+ const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
+ u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap())
+ }
+
+ /// The classifier is the funnel's probe chain as data: every row pins
+ /// one routing decision, and the labeled rows pin the probe ORDERINGS
+ /// the funnel relies on.
+ #[allow(clippy::too_many_lines)]
#[test]
- fn create_topic_bounds_deny_pre_consensus() {
- let segment_size = iggy_common::DEFAULT_SEGMENT_SIZE;
- assert!(segment_size > 0, "default segment size must be nonzero");
+ fn classify_pins_probe_order() {
+ fn routed(operation: Operation, session: u64, request: u64) -> RoutedRequestHeader {
+ RoutedRequestHeader {
+ command: Command::Request,
+ operation,
+ client: 7,
+ session,
+ request,
+ ..Default::default()
+ }
+ }
+ fn nr(nr_code: u32) -> RoutedRequestHeader {
+ let mut header = routed(Operation::NonReplicated, 0, 0);
+ header.reserved[..4].copy_from_slice(&nr_code.to_le_bytes());
+ header
+ }
+ let table = [
+ (
+ "legacy login beats the session gate (unbound)",
+ nr(LOGIN_USER_CODE),
+ false,
+ RequestClass::LegacyLogin,
+ ),
+ (
+ "legacy login beats the bound reads route",
+ nr(LOGIN_USER_CODE),
+ true,
+ RequestClass::LegacyLogin,
+ ),
+ (
+ "legacy PAT login beats the session gate (unbound)",
+ nr(LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE),
+ false,
+ RequestClass::LegacyLogin,
+ ),
+ (
+ "legacy PAT login beats the bound reads route",
+ nr(LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE),
+ true,
+ RequestClass::LegacyLogin,
+ ),
+ (
+ "ping is the only pre-auth code",
+ nr(PING_CODE),
+ false,
+ RequestClass::NonReplicatedRead,
+ ),
+ (
+ "cluster metadata is not pre-auth",
+ nr(GET_CLUSTER_METADATA_CODE),
+ false,
+ RequestClass::UnauthenticatedRead,
+ ),
+ (
+ "poll-messages is a non-replicated code, not a partition op",
+ nr(POLL_MESSAGES_CODE),
+ true,
+ RequestClass::NonReplicatedRead,
+ ),
+ (
+ "consumer-offset read is a non-replicated code, not a partition op",
+ nr(GET_CONSUMER_OFFSET_CODE),
+ true,
+ RequestClass::NonReplicatedRead,
+ ),
+ (
+ "the register handshake",
+ routed(Operation::Register, 0, 0),
+ false,
+ RequestClass::LoginRegister,
+ ),
+ (
+ "register with a session falls to the default",
+ routed(Operation::Register, 1, 0),
+ true,
+ RequestClass::ReplicatedMetadata,
+ ),
+ (
+ "register with a request number falls to the default",
+ routed(Operation::Register, 0, 1),
+ true,
+ RequestClass::ReplicatedMetadata,
+ ),
+ (
+ "logout, bound",
+ routed(Operation::Logout, 1, 1),
+ true,
+ RequestClass::Logout,
+ ),
+ (
+ "logout beats the session gate",
+ routed(Operation::Logout, 1, 1),
+ false,
+ RequestClass::Logout,
+ ),
+ (
+ "unbound metadata op",
+ routed(Operation::CreateStream, 1, 1),
+ false,
+ RequestClass::UnboundReplicated,
+ ),
+ (
+ "the session gate beats the delete-segments probe",
+ routed(Operation::DeleteSegments, 1, 1),
+ false,
+ RequestClass::UnboundReplicated,
+ ),
+ (
+ "the session gate beats the partition probe",
+ routed(Operation::SendMessages, 1, 1),
+ false,
+ RequestClass::UnboundReplicated,
+ ),
+ (
+ "delete-segments is neither partition nor metadata",
+ routed(Operation::DeleteSegments, 1, 1),
+ true,
+ RequestClass::DeleteSegments,
+ ),
+ (
+ "send-messages is partition-plane",
+ routed(Operation::SendMessages, 1, 1),
+ true,
+ RequestClass::Partition,
+ ),
+ (
+ "store-consumer-offset is partition-plane",
+ routed(Operation::StoreConsumerOffset, 1, 1),
+ true,
+ RequestClass::Partition,
+ ),
+ (
+ "create-topic is replicated metadata",
+ routed(Operation::CreateTopic, 1, 1),
+ true,
+ RequestClass::ReplicatedMetadata,
+ ),
+ ];
+ for (label, header, bound, expected) in table {
+ assert_eq!(classify(&header, bound), expected, "{label}");
+ }
+ }
- assert!(
- validate_topic_bounds(
- MAX_PARTITIONS_PER_REQUEST,
- MaxTopicSize::ServerDefault,
- segment_size
+ /// A legacy login code must get the typed `MalformedLogin` eviction
+ /// BEFORE the session gate runs: an unbound sender must not fall into
+ /// the generic Unauthenticated deny the pre-auth guard sends for other
+ /// codes.
+ #[compio::test]
+ async fn legacy_login_codes_evicted_before_session_gate() {
+ const TRANSPORT: u128 = 94;
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ let sessions = Rc::new(RefCell::new(SessionManager::new()));
+ let system_config = Arc::new(ServerSystemConfig::default());
+
+ for code in [LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE] {
+ handle_client_request(
+ &shard,
+ &sessions,
+ &system_config,
+ 1,
+ TRANSPORT,
+ non_replicated_request(TRANSPORT, code),
)
- .is_ok(),
- "the partition cap itself is admissible"
+ .await;
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1, "code {code} must produce one frame");
+ let (client, frame) = &replies[0];
+ assert_eq!(*client, TRANSPORT);
+ assert_eq!(
+ frame_command(frame),
+ Command::Eviction as u8,
+ "legacy code {code} must be evicted, not denied"
+ );
+ assert_eq!(
+ eviction_reason_byte(frame),
+ EvictionReason::MalformedLogin as u8,
+ "legacy code {code} must carry MalformedLogin"
+ );
+ drop(replies);
+ bus.client_replies.borrow_mut().clear();
+ }
+ }
+
+ /// A replicated op from an unbound transport must get the typed
+ /// `Eviction(NoSession)`: the client believes it has a session and that
+ /// session is gone, so a silent drop or an empty Reply would wedge or
+ /// mislead it.
+ #[compio::test]
+ async fn unbound_replicated_request_gets_no_session_eviction() {
+ const TRANSPORT: u128 = 95;
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ let sessions = Rc::new(RefCell::new(SessionManager::new()));
+ let system_config = Arc::new(ServerSystemConfig::default());
+
+ handle_client_request(
+ &shard,
+ &sessions,
+ &system_config,
+ 1,
+ TRANSPORT,
+ wire_request(Operation::CreateStream, TRANSPORT, 1, 1, &[]).into_generic(),
+ )
+ .await;
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1, "unbound replicated op must be answered");
+ let (client, frame) = &replies[0];
+ assert_eq!(*client, TRANSPORT);
+ assert_eq!(
+ frame_command(frame),
+ Command::Eviction as u8,
+ "unbound replicated op must be evicted, not denied or dropped"
);
- assert!(
- matches!(
- validate_topic_bounds(
- MAX_PARTITIONS_PER_REQUEST + 1,
- MaxTopicSize::ServerDefault,
- segment_size
- ),
- Err(IggyError::TooManyPartitions)
- ),
- "one past the partition cap must deny"
+ assert_eq!(
+ eviction_reason_byte(frame),
+ EvictionReason::NoSession as u8,
+ "the eviction must carry NoSession"
);
- // ServerDefault is numerically 0 yet exempt from the segment-size
- // floor: it resolves against server config, matching legacy.
- assert!(validate_topic_bounds(1, MaxTopicSize::ServerDefault, segment_size).is_ok());
- assert!(validate_topic_bounds(1, MaxTopicSize::Unlimited, segment_size).is_ok());
- let below_floor = MaxTopicSize::Custom((segment_size - 1).into());
- assert!(
- matches!(
- validate_topic_bounds(1, below_floor, segment_size),
- Err(IggyError::InvalidTopicSize(size, floor))
- if size == below_floor && floor == IggyByteSize::from(segment_size)
- ),
- "custom size below the segment size must deny with the bounds"
+ }
+
+ /// PING is the one pre-auth code: an unbound transport's ping must get a
+ /// normal status-0 Reply, not an eviction and not a deny.
+ #[compio::test]
+ async fn pre_auth_ping_allowed() {
+ const TRANSPORT: u128 = 96;
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ let sessions = Rc::new(RefCell::new(SessionManager::new()));
+ let system_config = Arc::new(ServerSystemConfig::default());
+
+ handle_client_request(
+ &shard,
+ &sessions,
+ &system_config,
+ 1,
+ TRANSPORT,
+ non_replicated_request(TRANSPORT, PING_CODE),
+ )
+ .await;
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1, "pre-auth ping must be answered");
+ let (client, frame) = &replies[0];
+ assert_eq!(*client, TRANSPORT);
+ assert_eq!(
+ frame_command(frame),
+ Command::Reply as u8,
+ "pre-auth ping must get a Reply, not an eviction"
);
- let at_floor = MaxTopicSize::Custom(IggyByteSize::from(segment_size));
- assert!(
- validate_topic_bounds(1, at_floor, segment_size).is_ok(),
- "a topic exactly one segment large is admissible"
+ assert_eq!(reply_status(frame), 0, "pre-auth ping must succeed");
+ }
+
+ /// The request-checksum gate runs before every probe: a replicated op
+ /// from an UNBOUND transport with a bad stamp must get the checksum deny
+ /// Reply, not the `Eviction(NoSession)` the session gate would send.
+ #[compio::test]
+ async fn checksum_mismatch_denies_before_everything() {
+ const TRANSPORT: u128 = 97;
+ const BODY: &[u8] = b"stream-body";
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ let sessions = Rc::new(RefCell::new(SessionManager::new()));
+ let system_config = Arc::new(ServerSystemConfig::default());
+
+ let mut message = wire_request(Operation::CreateStream, TRANSPORT, 1, 1, BODY);
+ {
+ let header_size = size_of::<RequestHeader>();
+ let header = bytemuck::checked::from_bytes_mut::<RequestHeader>(
+ &mut message.as_mut_slice()[..header_size],
+ );
+ // Nonzero (zero means unstamped and skips the check) and never
+ // the body's real checksum.
+ header.request_checksum = u128::from(iggy_common::calculate_checksum(BODY)) + 1;
+ }
+ handle_client_request(
+ &shard,
+ &sessions,
+ &system_config,
+ 1,
+ TRANSPORT,
+ message.into_generic(),
+ )
+ .await;
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1, "bad stamp must be answered");
+ let (client, frame) = &replies[0];
+ assert_eq!(*client, TRANSPORT);
+ assert_eq!(
+ frame_command(frame),
+ Command::Reply as u8,
+ "a bad stamp must get the checksum deny, not the NoSession eviction"
);
+ assert_eq!(
+ reply_status(frame),
+ IggyError::InvalidFormat.as_code(),
+ "the deny status must carry the checksum error"
+ );
+ }
+
+ /// The late-bound self-reference as the shard build creates it: unset
+ /// until the shard exists.
+ fn unset_shard_handle() -> ShellShardHandle<SpyBus, PrepareJournal, IggySnapshot> {
+ Rc::new(RefCell::new(None))
}
+ /// Yield so the drain task the enqueue spawned on the bus can run.
+ async fn run_spawned_tasks() {
+ compio::time::sleep(std::time::Duration::from_millis(1)).await;
+ }
+
+ /// One handler per shard means one connection-lost hook on its bus: a
+ /// second install would overwrite the first and orphan the sessions
+ /// the first one bound.
#[test]
- fn partitions_count_cap_denies_pre_consensus() {
- assert!(
- validate_partitions_count(MAX_PARTITIONS_PER_REQUEST).is_ok(),
- "the cap itself is admissible"
+ fn deferred_handler_installs_one_connection_lost_hook() {
+ let bus = SpyBus::default();
+ let _handler = make_deferred_client_request_handler(
+ &bus,
+ &unset_shard_handle(),
+ &Rc::new(RefCell::new(SessionManager::new())),
+ Arc::new(ServerSystemConfig::default()),
+ 1,
+ );
+ assert_eq!(
+ bus.connection_lost_hooks.get(),
+ 1,
+ "one factory call must install exactly one connection-lost hook"
+ );
+ }
+
+ /// The handler is built before the shard it serves. A frame that
+ /// arrives while the self-reference is still unset stays queued, and
+ /// the enqueue must release the client's active slot: otherwise every
+ /// later frame for that client finds the slot taken and nothing is
+ /// ever drained, the stranded frame included.
+ #[compio::test]
+ async fn deferred_handler_drains_after_the_shard_handle_is_set() {
+ const TRANSPORT: u128 = 98;
+ let bus = SpyBus::default();
+ let shard_handle = unset_shard_handle();
+ let handler = make_deferred_client_request_handler(
+ &bus,
+ &shard_handle,
+ &Rc::new(RefCell::new(SessionManager::new())),
+ Arc::new(ServerSystemConfig::default()),
+ 1,
);
+
+ handler(TRANSPORT, non_replicated_request(TRANSPORT, PING_CODE));
+ run_spawned_tasks().await;
assert!(
- matches!(
- validate_partitions_count(MAX_PARTITIONS_PER_REQUEST + 1),
- Err(IggyError::TooManyPartitions)
- ),
- "one past the cap must deny"
+ bus.client_replies.borrow().is_empty(),
+ "nothing can be served before the shard exists"
+ );
+
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
+ handler(TRANSPORT, non_replicated_request(TRANSPORT, PING_CODE));
+ for _ in 0..500 {
+ if bus.client_replies.borrow().len() == 2 {
+ break;
+ }
+ run_spawned_tasks().await;
+ }
+
+ let replies = bus.client_replies.borrow();
+ assert_eq!(
+ replies.len(),
+ 2,
+ "the stranded ping and the new one must both be served once the shard exists"
);
- // Zero passes the shared cap because a zero-partition TOPIC is legal
- // (legacy `create_topic` admits `0..=MAX`).
- assert!(validate_partitions_count(0).is_ok());
+ for (client, frame) in replies.iter() {
+ assert_eq!(*client, TRANSPORT);
+ assert_eq!(frame_command(frame), Command::Reply as u8);
+ assert_eq!(reply_status(frame), 0, "a served ping succeeds");
+ }
}
+ /// The per-client queue entry goes with its last frame. The
+ /// connection-lost hook cannot be its only owner: the dispatch task still
+ /// delivers frames it had buffered when the socket closed, which re-create
+ /// the entry after the hook ran, and `client_id` is monotonic - so an
+ /// entry nothing frees again lives until process exit.
#[test]
- fn zero_partitions_change_denies_pre_consensus() {
- // Adding or removing zero partitions is a no-op that would still burn
- // a replicated log entry and force a rebalance. Legacy rejects it with
- // `TooManyPartitions` in both handlers, so the code matches.
+ fn draining_a_client_queue_frees_its_entry() {
+ const CLIENT: u128 = 7;
+ let queues: ClientRequestQueues = Rc::new(RefCell::new(AHashMap::new()));
+ queues
+ .borrow_mut()
+ .entry(CLIENT)
+ .or_default()
+ .push_back(non_replicated_request(CLIENT, PING_CODE));
+
+ assert!(pop_next_client_request(&queues, CLIENT).is_some());
assert!(
- matches!(
- validate_partitions_change_count(0),
- Err(IggyError::TooManyPartitions)
- ),
- "adding or removing zero partitions must deny"
+ queues.borrow().is_empty(),
+ "the entry must be freed as soon as it drains empty"
);
- assert!(validate_partitions_change_count(1).is_ok());
- assert!(validate_partitions_change_count(MAX_PARTITIONS_PER_REQUEST).is_ok());
assert!(
- matches!(
- validate_partitions_change_count(MAX_PARTITIONS_PER_REQUEST + 1),
- Err(IggyError::TooManyPartitions)
- ),
- "the cap still applies"
+ pop_next_client_request(&queues, CLIENT).is_none(),
+ "a drained client has nothing left to pop"
);
}
}
diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs
index b2e0789bf..ac20090ab 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -28,23 +28,25 @@
//! replies to committed writes itself, straight from the owning shard, while
//! everything the host builds here -- read bodies, denies, empty-poll shapes
//! -- goes out on the connection's home shard. The reply path is therefore
-//! split by plane, not unified, and the deny helpers in `authz` are the
+//! split by plane, not unified, and the deny helpers in `failure` are the
//! third leg (typed status replies for requests that never reach a plane).
use crate::consumer_group::maybe_rewrite_consumer_offset_request;
-use crate::dispatch::authz::{
- authorize_partition_op, authorize_partition_read, send_deny_reply, send_non_replicated_deny,
+use crate::dispatch::authz::{authorize_partition_op, authorize_partition_read};
+use crate::dispatch::failure::{
+ FrameChannel, send_deny_reply, send_host_frame, send_non_replicated_bytes,
+ send_non_replicated_deny, send_result_rejection,
};
use crate::dispatch::submit::submit_client_request_on_owner;
-use crate::dispatch::{send_non_replicated_bytes, send_reply_frame, upgrade_shard_handle};
+use crate::dispatch::upgrade_shard_handle;
use crate::responses::{
- build_consumer_offset_body, build_empty_reply, build_polled_messages_reply,
- current_metadata_commit, resolve_partition_namespace, resolve_partition_request_namespace,
+ build_consumer_offset_body, build_polled_messages_reply, current_metadata_commit,
+ resolve_partition_namespace, resolve_partition_request_namespace,
};
use crate::shell::{ShellBus, ShellShard, ShellShardHandle};
use crate::wire::{request_body, usize_to_u32};
use bytes::Bytes;
-use consensus::{Consensus, MetadataHandle, PartitionsHandle, build_result_rejection_reply};
+use consensus::{Consensus, MetadataHandle, PartitionsHandle};
use iggy_binary_protocol::PrepareHeader;
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy;
@@ -57,10 +59,10 @@ use iggy_binary_protocol::{
AckLevel, Command, KIND_CONSUMER_GROUP, Operation, RoutedRequestHeader, WireDecode, WireEncode,
WireIdentifier,
};
-use iggy_common::{IggyError, PollingStrategy};
+use iggy_common::{IggyError, PollingStrategy, RESYNC_REQUIRED_PARTITION_SENTINEL};
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
-use message_bus::AUTO_COMMIT_CLIENT_ID;
+use message_bus::{AUTO_COMMIT_CLIENT_ID, BusMessage};
use metadata::impls::metadata::{
StreamsFrontend, build_truncate_partition_client_message,
build_truncate_partition_client_message_with_identifiers,
@@ -469,14 +471,19 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>(
// metadata access to resolve it.
let request = match maybe_rewrite_consumer_offset_request(shard, request) {
Ok(rewritten) => rewritten,
+ // Not reachable through the wire path: the same body already decoded in
+ // `resolve_partition_request_namespace` above, and re-encoding it can
+ // only fail past `u32::MAX` bytes against a 64 MiB request cap. Denying
+ // typed keeps a future re-encode failure from acking work the partition
+ // plane never saw.
Err(error) => {
warn!(
transport_client_id,
error = %error,
operation = ?header.operation,
- "failed to rewrite consumer-offset request; replying empty"
+ "failed to rewrite consumer-offset request; replying denied"
);
- send_empty_partition_reply(shard, transport_client_id, &header).await;
+ send_deny_reply(shard, transport_client_id, &header, error.as_code()).await;
return;
}
};
@@ -581,8 +588,12 @@ 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; a consumer-group poll the
+/// coordinator fenced replies with the same shape carrying the re-sync
+/// sentinel. Every permanent client error (undecodable body, authz, and every
+/// rejection the resolve raises) denies with a nonzero status instead, so none
+/// of them can be mistaken for an empty partition.
#[allow(clippy::future_not_send)]
pub(in crate::dispatch) async fn handle_poll_messages<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
@@ -597,21 +608,23 @@ pub(in crate::dispatch) async fn handle_poll_messages<B, MJ, S, SB>(
SB: SuperblockStore + 'static,
{
let Ok(wire) = PollMessagesRequest::decode_from(request_body(request)) else {
- // Undecodable poll: keep the fail-fast empty-poll shape.
- send_non_replicated_bytes(
+ // A permanent client error, so it must not borrow the empty-poll
+ // shape: that body decodes as a successful 0-message poll, and a
+ // consumer looping on it never learns why. The server already sends
+ // nonzero-status poll replies (authz, unresolved target).
+ send_non_replicated_deny(
shard,
request,
transport_client_id,
- empty_polled_messages_body(0),
- "poll_messages",
+ IggyError::InvalidCommand.as_code(),
)
.await;
return;
};
// Gate on (stream, topic) before touching the partition plane. A resolution
- // miss falls through to the resolve path below (empty-poll / not-found); a
- // denial replies status!=0 with an empty body, distinct from the empty-poll
- // "0 messages" shape.
+ // miss denies typed on the resolve path below; a denial here replies
+ // status!=0 with an empty body, distinct from the empty-poll "0 messages"
+ // shape.
if let Some(status) = authorize_partition_read(
shard,
&wire.stream_id,
@@ -624,89 +637,132 @@ pub(in crate::dispatch) async fn handle_poll_messages<B, MJ, S, SB>(
send_non_replicated_deny(shard, request, transport_client_id, status).await;
return;
}
- let body = match resolve_poll_request(shard, &wire, request.header().client) {
- Ok((namespace, partition_id, consumer, args)) => {
- match shard
- .partition_read(namespace, PartitionRead::Poll { consumer, args })
- .await
- {
- Some(PartitionReadReply::Poll {
- fragments,
- current_offset,
- }) => match build_polled_messages_reply(
- request.header(),
- current_metadata_commit(shard),
- partition_id,
- current_offset,
- fragments,
- shard.plane.partitions().config().encryptor.as_deref(),
- ) {
- Ok(reply) => {
- send_reply_frame(shard, transport_client_id, reply, "poll_messages").await;
- return;
- }
- Err(error) => {
- warn!(
- transport_client_id,
- error = %error,
- "failed to re-encode polled batches; replying empty poll"
- );
- empty_polled_messages_body(partition_id)
- }
- },
- other => {
- warn!(
+ let (body, channel) = match resolve_poll_request(shard, &wire, request.header().client) {
+ Ok(resolved) => {
+ match read_polled_messages(shard, transport_client_id, request, resolved).await {
+ Ok(reply) => {
+ send_host_frame(
+ &shard.bus,
transport_client_id,
- namespace = namespace.inner(),
- reply_was_none = other.is_none(),
- "partition read failed; replying empty poll"
- );
- empty_polled_messages_body(partition_id)
+ reply,
+ FrameChannel::Reply,
+ "poll_messages",
+ )
+ .await;
+ return;
}
+ Err(fallback) => fallback,
}
}
+ // A generation fence: the client's cached assignment went stale after a
+ // rebalance. The empty poll carries the re-sync sentinel so the SDK
+ // re-syncs and retries instead of reading end-of-partition.
+ Err(error @ IggyError::ConsumerGroupPartitionNotOwned(..)) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "poll_messages fenced; replying re-sync sentinel"
+ );
+ empty_poll_fallback(RESYNC_REQUIRED_PARTITION_SENTINEL)
+ }
+ // Everything else the resolve rejects is a permanent client error: an
+ // unresolved stream, topic, or partition, a polling strategy or
+ // consumer kind outside the wire enum, or a group poll that omitted
+ // its partition id. All must surface typed -- the empty poll is a
+ // successful 0-message read, and a consumer looping on it never learns
+ // why.
Err(error) => {
- // A stream, topic, or partition id that does not resolve is a
- // client addressing error and must surface as a typed rejection,
- // not an empty poll a consumer would read as end-of-partition.
- if matches!(
- error,
- IggyError::PartitionNotFound(..)
- | IggyError::StreamIdNotFound(_)
- | IggyError::TopicIdNotFound(..)
- ) {
- warn!(
- transport_client_id,
- error = %error,
- "poll_messages rejected: target not found"
- );
- send_non_replicated_deny(shard, request, transport_client_id, error.as_code())
- .await;
- return;
- }
- // A zero-byte body would panic the SDK's `PolledMessages`
- // decoder; reply the 16-byte empty-poll shape instead. A generation
- // fence (the client's cached assignment is stale after a rebalance)
- // carries the re-sync sentinel so the SDK re-syncs and retries
- // rather than treating the empty poll as end-of-partition.
warn!(
transport_client_id,
error = %error,
- "poll_messages request rejected; replying empty poll"
+ "poll_messages rejected; replying denied"
);
- let partition_id = if matches!(error, IggyError::ConsumerGroupPartitionNotOwned(..)) {
- iggy_common::RESYNC_REQUIRED_PARTITION_SENTINEL
- } else {
- 0
- };
- empty_polled_messages_body(partition_id)
+ send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await;
+ return;
}
};
- 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, so it is reserved for "this consumer has no stored offset" --
+/// an unresolved consumer group included, since a deleted group has no offset
+/// to report. A malformed request or an unresolved stream, topic, or partition
+/// denies with a nonzero status instead, so an addressing typo cannot read
+/// back as a fresh consumer.
// TODO(hubcio): plain local partition_read with no primary gate, so a
// follower answers from its own (possibly lagging) offset state. Needs the
// same is-caught-up-primary gate the auto-commit path has, or an explicit
@@ -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(),
)
.await;
return;
@@ -749,7 +807,7 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
return;
}
let body = match resolve_consumer_offset_request(shard, &wire) {
- Ok((namespace, partition_id, consumer)) => {
+ Ok(Some((namespace, partition_id, consumer))) => {
match shard
.partition_read(namespace, PartitionRead::ConsumerOffset { consumer })
.await
@@ -761,71 +819,37 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
_ => Bytes::new(),
}
}
- // A partition id that does not exist in a resolvable topic is a client
- // addressing error, the same one the poll path denies typed. An empty
- // body decodes as `None` -- indistinguishable from "this consumer has
- // no stored offset yet" -- so the caller cannot tell a typo from a
- // fresh consumer.
- Err(error @ IggyError::PartitionNotFound(..)) => {
+ // An unresolved group has no offset to report, the one thing the
+ // empty body may mean.
+ Ok(None) => Bytes::new(),
+ // Everything the resolve rejects is a client error: an unresolved
+ // stream, topic, or partition, or `resolve_partition_namespace`'s
+ // generic rejection. All must surface typed -- an empty body decodes as
+ // `None`, indistinguishable from "no stored offset yet", so a typo'd or
+ // deleted target would read back as a fresh consumer and resume from
+ // its configured default with no error anywhere. The same read over
+ // REST 404s.
+ Err(error) => {
warn!(
transport_client_id,
error = %error,
- "get_consumer_offset rejected: partition not found"
+ "get_consumer_offset rejected; replying denied"
);
send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await;
return;
}
- Err(error) => {
- warn!(
- transport_client_id,
- error = %error,
- "get_consumer_offset request rejected; replying empty"
- );
- Bytes::new()
- }
};
send_non_replicated_bytes(
shard,
request,
transport_client_id,
body,
+ FrameChannel::Reply,
"get_consumer_offset",
)
.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)]
-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);
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- operation = ?request_header.operation,
- "failed to surface empty partition reply"
- );
- }
-}
-
/// Wait (bounded) until this shard holds a routing row for `namespace`. Fast
/// path: row already present -> no wait.
///
@@ -961,10 +985,15 @@ where
/// namespace, partition, and polling consumer. Shared by the TCP dispatch and
/// the HTTP route; needs no client id because offset reads are not fenced
/// (any client may read a group's offset, member or not).
+///
+/// `Ok(None)` is the one outcome that means "no offset exists to report" (an
+/// unresolved group). It is a separate outcome rather than an error code
+/// because [`resolve_partition_namespace`] also rejects an addressing miss
+/// with `InvalidIdentifier`, and the caller must deny that one typed.
pub fn resolve_consumer_offset_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
wire: &GetConsumerOffsetRequest,
-) -> Result<(IggyNamespace, u32, PollingConsumer), IggyError>
+) -> Result<Option<(IggyNamespace, u32, PollingConsumer)>, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -981,19 +1010,21 @@ where
// it, member or not), the same key the write path is rewritten to. An
// unresolved group (e.g. deleted) has no offset, so the read reports None.
let consumer = if wire.consumer.kind == KIND_CONSUMER_GROUP {
- let group_id = shard
+ let Some(group_id) = shard
.plane
.metadata()
.mux_stm
.streams()
.resolve_consumer_group_id(&wire.stream_id, &wire.topic_id, &wire.consumer.id)
- .ok_or(IggyError::InvalidIdentifier)?;
+ else {
+ return Ok(None);
+ };
#[allow(clippy::cast_possible_truncation)]
PollingConsumer::ConsumerGroup(group_id as usize, partition_id as usize)
} else {
polling_consumer_from_wire(&wire.consumer, partition_id)?
};
- Ok((namespace, partition_id, consumer))
+ Ok(Some((namespace, partition_id, consumer)))
}
fn polling_consumer_from_wire(
@@ -1045,8 +1076,9 @@ fn polling_strategy_from_wire(
/// The consensus reply is forwarded verbatim: nothing-to-delete commits a
/// no-op `TruncatePartition(0)` and acks, while a not-primary rejection
/// reaches the client as `TransientNotCommitted` so the SDK replays instead
-/// of mistaking a dropped delete for success. Only a malformed / unresolvable
-/// request is acked empty without a commit.
+/// of mistaking a dropped delete for success. Nothing is acked empty: a
+/// malformed body denies typed, and an unresolvable namespace commits a typed
+/// rejection against the client's raw identifiers so its retry dedups.
#[allow(clippy::future_not_send)]
pub(in crate::dispatch) async fn handle_delete_segments_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
@@ -1087,7 +1119,7 @@ pub(in crate::dispatch) async fn handle_delete_segments_request<B, MJ, S, SB>(
)
.await
{
- Ok(truncate) => Some(truncate),
+ Ok(truncate) => truncate,
// The owning partition has not converged on the committed log yet, so
// the delete cannot be resolved to a watermark. Reply with the
// result-framed transient rejection (under the TruncatePartition
@@ -1104,67 +1136,50 @@ pub(in crate::dispatch) async fn handle_delete_segments_request<B, MJ, S, SB>(
0,
0,
);
- let reply = build_result_rejection_reply(
+ send_result_rejection(
+ shard,
+ transport_client_id,
template.header(),
- current_metadata_commit(shard),
- IggyError::TransientNotAccepted.as_code(),
- );
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- "delete_segments: failed to send transient rejection"
- );
- }
+ &IggyError::TransientNotAccepted,
+ "delete_segments_transient_rejection",
+ )
+ .await;
return;
}
- Err(_) => None,
- };
-
- let reply = if let Some(truncate) = truncate {
- // Forward the consensus reply verbatim, exactly like the generic
- // metadata path: a committed success acks the delete, and a
- // result-framed `TransientNotCommitted` rejection makes the SDK
- // replay the request. Acking unconditionally here would swallow a
- // not-primary rejection and drop the delete on the floor while the
- // client believes it succeeded.
- let Some(reply) = submit_client_request_on_owner(shard, truncate).await else {
- // Transient submit failure (not primary / view change). Stay
- // silent; the SDK read-timeout replays the same request id,
- // which re-resolves and commits. Acking here would advance the
- // client past an unrecorded request and gap the next metadata
- // op.
- warn!(
- transport_client_id,
- "delete_segments: transient submit; client will replay"
- );
+ // Undecodable body (never produced by the SDK): deny typed rather
+ // than ack empty. Both keep the lockstep stream framed, but a
+ // status-0 ack reads as a completed trim. Unresolvable-but-well-formed
+ // targets commit a typed rejection instead (see the resolve).
+ Err(error) => {
+ send_deny_reply(shard, transport_client_id, &header, error.as_code()).await;
return;
- };
- reply
- } else {
- // Undecodable body (never produced by the SDK): ack empty so the
- // lockstep stream stays framed; the typed decoder surfaces the
- // failure client-side. Unresolvable-but-well-formed targets commit a
- // typed rejection instead (see the resolve), so only a wire-corrupt
- // request can gap the sequence here.
- let commit = current_metadata_commit(shard);
- build_empty_reply(&header, transport_client_id, session, commit).into_generic()
+ }
};
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_frozen())
- .await
- {
+
+ // Forward the consensus reply verbatim, exactly like the generic metadata
+ // path: a committed success acks the delete, and a result-framed
+ // `TransientNotCommitted` rejection makes the SDK replay the request.
+ // Acking unconditionally here would swallow a not-primary rejection and
+ // drop the delete on the floor while the client believes it succeeded.
+ let Some(reply) = submit_client_request_on_owner(shard, truncate).await else {
+ // Transient submit failure (not primary / view change). Stay silent;
+ // the SDK read-timeout replays the same request id, which re-resolves
+ // and commits. Acking here would advance the client past an
+ // unrecorded request and gap the next metadata op.
warn!(
transport_client_id,
- error = %error,
- "delete_segments: failed to send reply"
+ "delete_segments: transient submit; client will replay"
);
- }
+ return;
+ };
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_frozen(),
+ FrameChannel::Reply,
+ "delete_segments_reply",
+ )
+ .await;
}
/// Resolve a client `DeleteSegments` to the `TruncatePartition` that commits the
@@ -1175,9 +1190,11 @@ pub(in crate::dispatch) async fn handle_delete_segments_request<B, MJ, S, SB>(
/// `request` number; `client_id` / `session` are the bound VSR identity the
/// 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.
+/// sequence stays contiguous, and an UNRESOLVABLE one yields the truncate
+/// against the client's raw identifiers (the apply rejects it as a committed
+/// result). `Err` is only `InvalidCommand` for a malformed body and
+/// `TransientNotAccepted` for a partition behind the commit frontier: the TCP
+/// caller denies typed, the HTTP caller renders the error.
#[allow(clippy::future_not_send)]
#[allow(clippy::cast_possible_truncation)]
pub async fn resolve_delete_segments_truncate<B, MJ, S, SB>(
@@ -1279,7 +1296,7 @@ where
mod tests {
use super::*;
use crate::dispatch::test_support::{
- SpyBus, TestMux, TestShard, prepare_message, request_message,
+ SpyBus, TestMux, TestShard, prepare_message, request_message, test_shard,
};
use iggy_binary_protocol::ReplyHeader;
use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment;
@@ -1289,6 +1306,7 @@ mod tests {
CreateTopicRequest, CreateTopicWithAssignmentsRequest,
};
use iggy_binary_protocol::{WireName, WireOptions, WirePartitioning};
+ use iggy_common::Identifier;
use iggy_common::defaults::DEFAULT_ROOT_USER_ID;
use metadata::IggyMetadata;
use metadata::stm::StateMachine as _;
@@ -1302,6 +1320,145 @@ mod tests {
ShardIdentity, shard_channel,
};
+ /// An undecodable request body is a PERMANENT client error, so every read
+ /// on this path must answer a nonzero status. The fail-fast shapes these
+ /// used to borrow all decode as success: the 16-byte empty poll reads as a
+ /// 0-message poll, an empty consumer-offset body as "no offset stored",
+ /// and a status-0 `delete_segments` ack as a completed trim. A client on a
+ /// skewed protocol would loop on those forever with no error to surface.
+ #[compio::test]
+ async fn undecodable_read_bodies_must_deny_typed_not_fabricate_success() {
+ const TRANSPORT: u128 = 91;
+ const VSR_CLIENT: u128 = 1;
+ const SESSION: u64 = 1;
+ const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
+ // Shorter than any of the three request encodings, so each decoder
+ // fails on length before it can interpret a field.
+ const TRUNCATED_BODY: &[u8] = &[0x01];
+
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, 1));
+
+ let poll = request_message(
+ Operation::NonReplicated,
+ VSR_CLIENT,
+ SESSION,
+ 1,
+ TRUNCATED_BODY,
+ );
+ handle_poll_messages(&shard, TRANSPORT, &poll, Some(DEFAULT_ROOT_USER_ID)).await;
+
+ let offset = request_message(
+ Operation::NonReplicated,
+ VSR_CLIENT,
+ SESSION,
+ 2,
+ TRUNCATED_BODY,
+ );
+ handle_get_consumer_offset(&shard, TRANSPORT, &offset, Some(DEFAULT_ROOT_USER_ID)).await;
+
+ let delete = request_message(
+ Operation::DeleteSegments,
+ VSR_CLIENT,
+ SESSION,
+ 3,
+ TRUNCATED_BODY,
+ );
+ handle_delete_segments_request(&shard, TRANSPORT, Some((VSR_CLIENT, SESSION)), &delete)
+ .await;
+
+ let replies = bus.client_replies.borrow();
+ assert_eq!(
+ replies.len(),
+ 3,
+ "one deny frame per undecodable request, none of them silent"
+ );
+ for (label, (client, frame)) in ["poll_messages", "get_consumer_offset", "delete_segments"]
+ .into_iter()
+ .zip(replies.iter())
+ {
+ assert_eq!(*client, TRANSPORT, "{label} deny must target the transport");
+ let status =
+ u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
+ assert_eq!(
+ status,
+ IggyError::InvalidCommand.as_code(),
+ "{label} with an undecodable body must carry a nonzero status"
+ );
+ }
+ }
+
+ /// A WELL-FORMED read against a stream that does not exist is a permanent
+ /// addressing error, and it reaches the server through a public SDK method
+ /// (a typo'd or deleted stream). Both fail-fast shapes decode as success:
+ /// the 16-byte empty poll as a 0-message read, and an empty
+ /// consumer-offset body as `None`, which is indistinguishable from a fresh
+ /// consumer -- so the caller resumes from its configured default and
+ /// silently reprocesses. The same lookup 404s over REST.
+ #[compio::test]
+ async fn unresolved_read_targets_must_deny_typed_not_read_as_empty() {
+ const TRANSPORT: u128 = 91;
+ const VSR_CLIENT: u128 = 1;
+ const SESSION: u64 = 1;
+ const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
+ const MISSING_STREAM: u32 = 404;
+
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, 1));
+
+ let poll_body = PollMessagesRequest {
+ consumer: WireConsumer::consumer(WireIdentifier::Numeric(1)),
+ stream_id: WireIdentifier::Numeric(MISSING_STREAM),
+ topic_id: WireIdentifier::Numeric(1),
+ partition_id: Some(1),
+ strategy: WirePollingStrategy::offset(0),
+ count: 10,
+ auto_commit: false,
+ }
+ .to_bytes();
+ let poll = request_message(Operation::NonReplicated, VSR_CLIENT, SESSION, 1, &poll_body);
+ handle_poll_messages(&shard, TRANSPORT, &poll, Some(DEFAULT_ROOT_USER_ID)).await;
+
+ let offset_body = GetConsumerOffsetRequest {
+ consumer: WireConsumer::consumer(WireIdentifier::Numeric(1)),
+ stream_id: WireIdentifier::Numeric(MISSING_STREAM),
+ topic_id: WireIdentifier::Numeric(1),
+ partition_id: Some(1),
+ }
+ .to_bytes();
+ let offset = request_message(
+ Operation::NonReplicated,
+ VSR_CLIENT,
+ SESSION,
+ 2,
+ &offset_body,
+ );
+ handle_get_consumer_offset(&shard, TRANSPORT, &offset, Some(DEFAULT_ROOT_USER_ID)).await;
+
+ let replies = bus.client_replies.borrow();
+ assert_eq!(
+ replies.len(),
+ 2,
+ "one deny frame per unresolved read, none of them silent"
+ );
+ for (label, (client, frame)) in ["poll_messages", "get_consumer_offset"]
+ .into_iter()
+ .zip(replies.iter())
+ {
+ assert_eq!(*client, TRANSPORT, "{label} deny must target the transport");
+ let status =
+ u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
+ assert_eq!(
+ status,
+ IggyError::StreamIdNotFound(
+ Identifier::numeric(MISSING_STREAM).expect("a nonzero numeric identifier")
+ )
+ .as_code(),
+ "{label} against a missing stream must carry the not-found status"
+ );
+ }
+ }
+
/// A partition write whose routable wait exhausts (namespace committed,
/// but no reconciler ever seeds this shard's routing row -- the state a
/// teardown/rematerialise churn leaves behind) must answer a nonzero
diff --git a/core/server/src/dispatch/reads.rs b/core/server/src/dispatch/reads.rs
index 807b342b0..bcbb7de69 100644
--- a/core/server/src/dispatch/reads.rs
+++ b/core/server/src/dispatch/reads.rs
@@ -26,9 +26,11 @@
//! the HTTP layer), never in the builder.
use crate::cluster_meta::ClusterRoster;
-use crate::dispatch::authz::{authorize_default_read, authorize_uid, send_non_replicated_deny};
+use crate::dispatch::authz::{authorize_default_read, authorize_uid};
+use crate::dispatch::failure::{
+ FrameChannel, send_host_frame, send_non_replicated_bytes, send_non_replicated_deny,
+};
use crate::dispatch::partition::{handle_get_consumer_offset, handle_poll_messages};
-use crate::dispatch::send_non_replicated_bytes;
use crate::responses::{
build_empty_reply, build_get_me_response, build_get_personal_access_tokens_response,
build_non_replicated_response, connected_client_to_response, current_metadata_commit,
@@ -67,7 +69,7 @@ use metadata::permissioner::Permissioner;
use server_common::Message;
use std::cell::RefCell;
use std::future::Future;
-use std::net::IpAddr;
+use std::net::{IpAddr, SocketAddr};
use std::pin::pin;
use std::rc::Rc;
use std::sync::Arc;
@@ -96,6 +98,7 @@ async fn handle_get_personal_access_tokens<B, MJ, S, SB>(
request,
transport_client_id,
response.to_bytes(),
+ FrameChannel::Reply,
"get_personal_access_tokens",
)
.await;
@@ -123,6 +126,7 @@ async fn handle_get_me<B, MJ, S, SB>(
request,
transport_client_id,
response.to_bytes(),
+ FrameChannel::Reply,
"get_me",
)
.await;
@@ -340,6 +344,13 @@ pub(in crate::dispatch) async fn handle_non_replicated_request<B, MJ, S, SB>(
system_config: &Arc<ServerSystemConfig>,
transport_client_id: u128,
request: Message<RoutedRequestHeader>,
+ // Acting user, peer address and read-your-writes floor for the read gates
+ // below, resolved by the funnel in the same connection lookup as the
+ // heartbeat. `user_id` is `None` only on the pre-auth path (PING), which
+ // serves ungated codes; the gated arms fail closed on it. An unknown
+ // connection arrives with a floor of `0`: it was promised nothing, so its
+ // reads wait for nothing.
+ (user_id, client_address, watermark): (Option<u32>, Option<SocketAddr>, u64),
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -349,16 +360,10 @@ pub(in crate::dispatch) async fn handle_non_replicated_request<B, MJ, S, SB>(
{
const CODE_RANGE: std::ops::Range<usize> = 0..4;
let code = u32::from_le_bytes(request.header().reserved[CODE_RANGE].try_into().unwrap());
- // Acting user, peer address and read-your-writes floor for the gates
- // below, resolved in one connection lookup. `user_id` is `None` only on the
- // pre-auth path (PING), which serves ungated codes; the gated arms fail
- // closed on it.
- let (user_id, client_address, watermark) = sessions.borrow().read_context(transport_client_id);
match code {
PING_CODE => {
- // A ping is the client's liveness proof; reset its staleness clock
- // so the heartbeat verifier doesn't evict an active connection.
- sessions.borrow_mut().record_heartbeat(transport_client_id);
+ // No `record_heartbeat` here: the funnel records one for EVERY
+ // frame before classification, so a ping is already covered.
let commit = current_metadata_commit(shard);
let reply = build_empty_reply(
request.header(),
@@ -366,17 +371,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(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::Reply,
+ "ping_reply",
+ )
+ .await;
}
GET_ME_CODE => {
// Self-scoped, so no permissioner rule -- but the consumer-group
@@ -421,6 +423,7 @@ pub(in crate::dispatch) async fn handle_non_replicated_request<B, MJ, S, SB>(
&request,
transport_client_id,
response.to_bytes(),
+ FrameChannel::Reply,
"get_clients",
)
.await;
@@ -469,8 +472,15 @@ pub(in crate::dispatch) async fn handle_non_replicated_request<B, MJ, S, SB>(
}
.to_bytes()
});
- send_non_replicated_bytes(shard, &request, transport_client_id, bytes, "get_client")
- .await;
+ send_non_replicated_bytes(
+ shard,
+ &request,
+ transport_client_id,
+ bytes,
+ FrameChannel::Reply,
+ "get_client",
+ )
+ .await;
}
GET_SNAPSHOT_FILE_CODE => {
handle_get_snapshot(shard, system_config, transport_client_id, &request, user_id).await;
@@ -544,6 +554,27 @@ async fn handle_default_non_replicated<B, MJ, S, SB>(
})
.await
{
+ // Same line as the builder-`Err` branch below: `send_non_replicated_deny`
+ // logs only on send FAILURE, so a refusal that reaches the client would
+ // otherwise leave nothing server-side - including the refusal this gate
+ // now issues for every armless or unknown non-replicated code.
+ // The frontier wait reaches here through the same `Err` and has
+ // already counted and logged itself, so it stays at `debug!`: a durably
+ // lagging node refuses every held read of every client for as long as
+ // it lags, and warning here would be a line per refusal.
+ if matches!(error, IggyError::TransientNotAccepted) {
+ debug!(
+ transport_client_id,
+ code, "denying non-replicated VSR request; read frontier unreached"
+ );
+ } else {
+ warn!(
+ transport_client_id,
+ code,
+ error = %error,
+ "denying non-replicated VSR request"
+ );
+ }
send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await;
return;
}
@@ -571,18 +602,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(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::Reply,
+ "non_replicated_reply",
+ )
+ .await;
}
Err(error) => {
// Surface the builder's typed error (unsupported op, undecodable
@@ -659,6 +686,7 @@ async fn handle_get_snapshot<B, MJ, S, SB>(
request,
transport_client_id,
GetSnapshotResponse { data: archive }.to_bytes(),
+ FrameChannel::Reply,
"get_snapshot",
)
.await;
@@ -732,6 +760,7 @@ async fn handle_sync_consumer_group<B, MJ, S, SB>(
request,
transport_client_id,
body,
+ FrameChannel::Reply,
"sync_consumer_group",
)
.await;
diff --git a/core/server/src/dispatch/session_ops.rs b/core/server/src/dispatch/session_ops.rs
index d43bcc17b..a8d2e62db 100644
--- a/core/server/src/dispatch/session_ops.rs
+++ b/core/server/src/dispatch/session_ops.rs
@@ -32,6 +32,9 @@
//! two together, so logout and eviction must release BOTH -- every teardown
//! path below pairs `remove_connection` with a replicated `Logout`.
+use crate::dispatch::failure::{
+ FrameChannel, send_eviction, send_host_frame, send_result_rejection,
+};
use crate::dispatch::login_error::LoginRegisterError;
use crate::responses::{
build_deny_reply, build_empty_reply, build_login_register_reply, current_metadata_commit,
@@ -39,11 +42,7 @@ use crate::responses::{
use crate::session_manager::{ClientSdkInfo, SessionManager};
use crate::shell::{ShellBus, ShellShard};
use crate::wire::request_body;
-use consensus::{
- Consensus, DISCONNECT_LOGOUT_REQUEST_ID, EvictionContext, MetadataHandle,
- build_eviction_message, build_incompatible_protocol_eviction_message,
- build_result_rejection_reply,
-};
+use consensus::{Consensus, DISCONNECT_LOGOUT_REQUEST_ID, MetadataHandle};
use iggy_binary_protocol::PrepareHeader;
use iggy_binary_protocol::requests::users::{LoginRegisterRequest, LoginRegisterWithPatRequest};
use iggy_binary_protocol::{
@@ -241,10 +240,14 @@ where
let commit = current_metadata_commit(shard).max(session);
let reply =
build_login_register_reply(request_header, vsr_client_id, session, commit, user_id);
- let _ = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await;
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::Reply,
+ "login_replay_reply",
+ )
+ .await;
return Ok(());
}
@@ -288,17 +291,14 @@ where
// lower number would make one frame contradict itself.
let commit = current_metadata_commit(shard).max(session);
let reply = build_login_register_reply(request_header, vsr_client_id, session, commit, user_id);
- let send_result = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await;
- if let Err(error) = send_result {
- warn!(
- transport_client_id,
- error = %error,
- "failed to send login/register reply"
- );
- }
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::Reply,
+ "login_register_reply",
+ )
+ .await;
Ok(())
}
@@ -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",
)
.await;
} else {
// Which code the hint carries is what tells the client whether the
// replay may move to another node: see `transient_login_code`.
- send_login_transient_reply(
+ send_result_rejection(
shard,
transport_client_id,
request_header,
- transient_login_code(error),
+ &transient_login_code(error),
+ "login_transient_replay_hint",
)
.await;
}
}
-/// Result-framed transient Reply on a non-terminal failed Register. The SDK
-/// decodes the nonzero result code and replays the same login on the same
-/// connection. Only call for transient errors -- see
-/// [`surface_login_failure`].
-#[allow(clippy::future_not_send)]
-async fn send_login_transient_reply<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- transport_client_id: u128,
- request_header: &RoutedRequestHeader,
- code: IggyError,
-) 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, code.as_code());
- 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 login transient reply"
- );
- }
-}
-
/// Wire code for a transient (non-terminal) login/register failure.
///
/// `TransientNotAccepted` asserts nothing was committed: the register never
@@ -420,53 +390,6 @@ const fn eviction_reason_for(error: &LoginRegisterError) -> EvictionReason {
}
}
-/// Reject a replicated request from an unbound transport with a typed
-/// `Eviction(NoSession)` frame: the session the client believes it has is
-/// gone, so it must register again. Pre-auth non-replicated reads get a
-/// deny Reply instead (no session exists, so nothing is evicted).
-///
-/// The SDK's reply decoder maps eviction reasons to typed errors
-/// (`NoSession` -> `Unauthenticated`), so clients fail fast with the same
-/// error the legacy server returns instead of a body-decode failure. The
-/// eviction context is best-effort off the metadata consensus (peer shards
-/// have none; zeroes are cosmetic -- the SDK only reads the reason).
-#[allow(clippy::future_not_send)]
-pub(in crate::dispatch) async fn send_unauthenticated_eviction<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- transport_client_id: u128,
-) 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(
- consensus::EvictionContext {
- cluster: 0,
- view: 0,
- replica: 0,
- },
- consensus::EvictionContext::from_consensus,
- );
- let eviction = consensus::build_eviction_message(
- ctx,
- transport_client_id,
- iggy_binary_protocol::EvictionReason::NoSession,
- );
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, eviction.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- "failed to send unauthenticated eviction"
- );
- }
-}
-
/// Per-shard heartbeat verifier: evict connections that have not pinged within
/// `1.2 x interval`. Mirrors the legacy `verify_heartbeats` periodic task.
/// Eviction reuses the disconnect path (drops the client from its consumer
@@ -552,35 +475,21 @@ 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,
+ // The eviction itself is done and observed at this point (session
+ // dropped, `Logout` submitted). The client notice below is best-effort
+ // and logs its own send failure, so this line must not claim it.
+ warn!(
transport_client_id,
- iggy_binary_protocol::EvictionReason::StaleClient,
+ "evicted stale client (missed heartbeat); sending the eviction notice"
);
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, eviction.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- "failed to send stale-client eviction"
- );
- } else {
- warn!(
- transport_client_id,
- "evicted stale client (missed heartbeat)"
- );
- }
+ send_eviction(
+ shard,
+ transport_client_id,
+ transport_client_id,
+ EvictionReason::StaleClient,
+ "stale_client_eviction",
+ )
+ .await;
}
/// Answer a backup's forwarded `Register` from the node it named primary.
@@ -1180,17 +1089,14 @@ pub(in crate::dispatch) async fn handle_logout_request<B, MJ, S, SB>(
);
let commit = current_metadata_commit(shard);
let reply = build_empty_reply(request.header(), transport_client_id, 0, 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 unbound logout reply"
- );
- }
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::Reply,
+ "unbound_logout_reply",
+ )
+ .await;
return;
};
@@ -1210,17 +1116,14 @@ pub(in crate::dispatch) async fn handle_logout_request<B, MJ, S, SB>(
commit,
transient_logout_code(&error).as_code(),
);
- if let Err(send_error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %send_error,
- "failed to send logout deny reply"
- );
- }
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::TypedDeny,
+ "logout_transient_deny",
+ )
+ .await;
return;
}
};
@@ -1228,17 +1131,14 @@ pub(in crate::dispatch) async fn handle_logout_request<B, MJ, S, SB>(
sessions.borrow_mut().remove_connection(transport_client_id);
let reply = build_empty_reply(request.header(), vsr_client_id, 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 logout reply"
- );
- }
+ send_host_frame(
+ &shard.bus,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ FrameChannel::Reply,
+ "logout_reply",
+ )
+ .await;
}
/// Preserve the client identity when a Logout may already have entered the
@@ -1281,11 +1181,12 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>(
transport_client_id,
"rejecting login: body has no decodable version prefix"
);
- send_login_eviction(
+ send_eviction(
shard,
transport_client_id,
vsr_client_id,
EvictionReason::MalformedLogin,
+ "login_rejection",
)
.await;
return;
@@ -1298,11 +1199,12 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>(
sdk_version = %version_info.sdk_version,
"rejecting login: incompatible protocol version"
);
- send_login_eviction(
+ send_eviction(
shard,
transport_client_id,
vsr_client_id,
EvictionReason::IncompatibleProtocol,
+ "login_rejection",
)
.await;
return;
@@ -1395,11 +1297,12 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>(
transport_client_id,
"rejecting register request: invalid credentials"
);
- send_login_eviction(
+ send_eviction(
shard,
transport_client_id,
request.header().client,
EvictionReason::InvalidCredentials,
+ "login_rejection",
)
.await;
return;
@@ -1409,61 +1312,16 @@ pub(in crate::dispatch) async fn handle_login_register_request<B, MJ, S, SB>(
transport_client_id,
"rejecting register request with unsupported payload shape"
);
- send_login_eviction(
+ send_eviction(
shard,
transport_client_id,
request.header().client,
EvictionReason::MalformedLogin,
+ "login_rejection",
)
.await;
}
-/// Best-effort login-rejection eviction. Terminal one-way frame; a gone
-/// connection has nothing to recover, so the send error is logged and
-/// dropped. 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_login_eviction<B, MJ, S, SB>(
- shard: &Rc<ShellShard<B, MJ, S, SB>>,
- transport_client_id: u128,
- vsr_client_id: u128,
- reason: EvictionReason,
-) 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),
- };
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, eviction.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- reason = ?reason,
- "failed to send login eviction"
- );
- }
-}
-
#[cfg(test)]
mod tests {
use super::*;
diff --git a/core/server/src/dispatch/test_support.rs b/core/server/src/dispatch/test_support.rs
index e52578c64..c0a005734 100644
--- a/core/server/src/dispatch/test_support.rs
+++ b/core/server/src/dispatch/test_support.rs
@@ -54,12 +54,16 @@ pub type RecordedReplies = Rc<RefCell<Vec<(u128, Vec<u8>)>>>;
pub type RecordedReplicaSends = Rc<RefCell<Vec<(u8, Vec<u8>)>>>;
/// Records every client-bound reply and replica-bound frame (target +
-/// bytes) instead of writing to a socket; everything else is a no-op. The
-/// two `ShellBus` halves are stubbed.
+/// bytes) instead of writing to a socket, and counts connection-lost hook
+/// installs; everything else is a no-op. The two `ShellBus` halves are
+/// stubbed.
#[derive(Debug, Clone, Default)]
pub struct SpyBus {
pub client_replies: RecordedReplies,
pub replica_sends: RecordedReplicaSends,
+ /// Installs of the client connection-lost hook, one per handler built
+ /// on this bus.
+ pub connection_lost_hooks: Rc<Cell<usize>>,
/// Resolve [`MessageBus::sleep`] immediately instead of arming a real
/// timer. The register forward is the only path here that races a
/// timer, and its budget is five seconds -- too long to wait for in a
@@ -143,7 +147,10 @@ impl ConnectionInstaller for SpyBus {
fn client_meta(&self, _client_id: u128) -> Option<Rc<ClientConnMeta>> {
None
}
- fn set_client_connection_lost_fn(&self, _f: ClientConnectionLostFn) {}
+ fn set_client_connection_lost_fn(&self, _f: ClientConnectionLostFn) {
+ self.connection_lost_hooks
+ .set(self.connection_lost_hooks.get() + 1);
+ }
}
/// Consensus incarnations standing for two successive boots of one node, as
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index 2a082405b..fc877054b 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -121,10 +121,6 @@ use shard::{PartitionRead, PartitionReadReply};
use crate::dispatch::partition::{resolve_consumer_offset_request, resolve_poll_request};
use crate::dispatch::session_ops::{verify_login_credentials, verify_pat_credentials};
-use crate::dispatch::{
- validate_option_keys, validate_topic_bounds, validate_topic_size_floor,
- warn_unenforceable_topic_size, warn_unenforceable_topic_size_on_partition_add,
-};
use crate::http::error::{
Consistency, ConsistencyQuery, CustomError, PartitionWriteError, ProduceAck, ProduceQuery,
ReadError, WriteError,
@@ -150,6 +146,10 @@ use crate::http::wire::{
use crate::responses::{
build_polled_messages_body, build_raw_pat_reply, connected_client_to_response,
};
+use crate::rewrite::{
+ validate_option_keys, validate_topic_bounds, validate_topic_size_floor,
+ warn_unenforceable_topic_size, warn_unenforceable_topic_size_on_partition_add,
+};
use crate::snapshot;
/// `GET /ping` response body, matching the legacy HTTP server's health probe.
@@ -1335,8 +1335,9 @@ pub(in crate::http) async fn get_consumer_offset(
.await?;
let wire =
consumer_offset_wire_request(&stream_id, &topic_id, &query).map_err(ReadError::Rejected)?;
- let (namespace, partition_id, consumer) =
- resolve_consumer_offset_request(&state.shard, &wire).map_err(|_| ReadError::NotFound)?;
+ let (namespace, partition_id, consumer) = resolve_consumer_offset_request(&state.shard, &wire)
+ .map_err(|_| ReadError::NotFound)?
+ .ok_or(ReadError::NotFound)?;
let reply = SendWrapper::new(
state
.shard
diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs
index a4864086e..08c23aa63 100644
--- a/core/server/src/http/submit.rs
+++ b/core/server/src/http/submit.rs
@@ -23,12 +23,10 @@ use std::rc::Rc;
use std::time::{Duration, Instant};
use bytes::Bytes;
-use consensus::MetadataHandle;
use futures::channel::oneshot;
use iggy_binary_protocol::consensus::Command;
use iggy_binary_protocol::{GenericHeader, Operation, ReplyHeader, RoutedRequestHeader};
use iggy_common::IggyError;
-use metadata::impls::metadata::StreamsFrontend;
use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use tracing::warn;
@@ -41,10 +39,9 @@ use crate::http::reply::{classify_partition_reply, committed_payload, eviction_e
use crate::http::session::HttpSession;
use crate::http::state::HttpInner;
use crate::http::wire::build_request_message;
-use crate::pat::rewrite_pat_request_for_user;
use crate::responses::transient_code;
+use crate::rewrite::http_chain;
use crate::shell::ServerShard;
-use crate::users::maybe_rewrite_user_password_request;
use crate::wire::request_body;
/// Bound on a partition write's (produce / consumer-offset write) wait for its
@@ -217,22 +214,8 @@ async fn submit_gated(
request_id,
body,
);
- let (message, raw_token) = rewrite_pat_request_for_user(
- session.user_id,
- max_tokens_per_user,
- |user_id| {
- shard
- .plane
- .metadata()
- .mux_stm
- .users()
- .read(|users| users.pat_count_of(user_id))
- },
- message,
- )
- .map_err(WriteError::Rejected)?;
- let message =
- maybe_rewrite_user_password_request(shard, message).map_err(WriteError::Rejected)?;
+ let (message, raw_token) = http_chain(shard, session.user_id, max_tokens_per_user, message)
+ .map_err(WriteError::Rejected)?;
// `DeleteSegments` is not itself a consensus op: resolve it to the metadata
// `TruncatePartition` that commits the trim before it reaches consensus,
// mirroring the TCP dispatch. The truncate rides this session's burned
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index b8131e558..7eb70057f 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -45,6 +45,7 @@ pub(crate) mod consumer_group;
pub(crate) mod dispatch;
pub(crate) mod pat;
pub(crate) mod responses;
+pub(crate) mod rewrite;
pub mod session_manager;
pub mod shell;
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index 75a8d2511..6cea58642 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -595,10 +595,15 @@ where
// than let the catch-all's empty-ok attest an artifact that was never
// produced.
GET_SNAPSHOT_FILE_CODE => Err(IggyError::InvalidCommand),
+ // Sequenced AFTER the named arms above, so flush keeps answering
+ // `FeatureUnavailable`. A table-listed non-replicated code with no arm
+ // is a routing bug and an unknown code is a client bug; the empty-ok
+ // that used to cover both attested a read that never ran. Only the
+ // named arms return `Empty`, and there it means "resolved to nothing"
+ // (the 404 the HTTP path maps).
_ => match iggy_binary_protocol::dispatch::lookup_command(code) {
- Some(meta) if !meta.is_replicated() => Ok(NonReplicatedResponse::Empty),
- Some(_) => Err(IggyError::FeatureUnavailable),
- None => Err(IggyError::InvalidCommand),
+ Some(meta) if meta.is_replicated() => Err(IggyError::FeatureUnavailable),
+ _ => Err(IggyError::InvalidCommand),
},
}
}
diff --git a/core/server/src/rewrite.rs b/core/server/src/rewrite.rs
new file mode 100644
index 000000000..5f00c588c
--- /dev/null
+++ b/core/server/src/rewrite.rs
@@ -0,0 +1,570 @@
+// 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;
+
+/// The pre-consensus rewrite stages, in chain order.
+///
+/// One owner for the set: the funnel's consumer-group rewrite runs outside
+/// [`tcp_chain`] but denies through the same path, so it names a variant
+/// here instead of passing a bare literal.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RewriteStage {
+ PersonalAccessToken,
+ UserPassword,
+ StaticBounds,
+ ConsumerGroup,
+}
+
+impl RewriteStage {
+ /// The `context` log label, in the same `snake_case` shape as every other
+ /// frame context.
+ #[must_use]
+ pub const fn as_str(self) -> &'static str {
+ match self {
+ Self::PersonalAccessToken => "personal_access_token",
+ Self::UserPassword => "user_password",
+ Self::StaticBounds => "static_bounds",
+ Self::ConsumerGroup => "consumer_group",
+ }
+ }
+}
+
+/// 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: RewriteStage,
+ 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: RewriteStage::PersonalAccessToken,
+ 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: RewriteStage::UserPassword,
+ error,
+ })?;
+ static_bounds(shard, &request).map_err(|error| RewriteDeny {
+ stage: RewriteStage::StaticBounds,
+ 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
+/// handlers via `command.validate()` plus the validators below. Returns the
+/// rewritten request and the raw PAT token the caller substitutes into the
+/// committed reply.
+pub fn http_chain<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ user_id: u32,
+ max_tokens_per_user: u32,
+ request: Message<RoutedRequestHeader>,
+) -> Result<(Message<RoutedRequestHeader>, Option<String>), IggyError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let (request, raw_token) = rewrite_pat_request_for_user(
+ user_id,
+ max_tokens_per_user,
+ |user_id| {
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .users()
+ .read(|users| users.pat_count_of(user_id))
+ },
+ request,
+ )?;
+ let request = maybe_rewrite_user_password_request(shard, request)?;
+ Ok((request, raw_token))
+}
+
+/// 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> {
+ // The two transports carry the cap under different names: this one on the
+ // binary path, `MAX_PARTITIONS_COUNT` in the HTTP DTO validators. The
+ // parity is a documented contract ([`http_chain`]), so it fails the build
+ // rather than the next cross-transport test.
+ const _: () = assert!(
+ MAX_PARTITIONS_PER_REQUEST == iggy_common::MAX_PARTITIONS_COUNT,
+ "the binary and HTTP partitions-count caps must stay equal"
+ );
+
+ if partitions_count > MAX_PARTITIONS_PER_REQUEST {
+ return Err(IggyError::TooManyPartitions);
+ }
+ Ok(())
+}
+
+/// [`validate_partitions_count`] plus the zero rejection that create-partitions
+/// and delete-partitions carry: adding or removing zero partitions is a no-op
+/// that would still burn a replicated log entry, bump `Streams::revision` and
+/// force every shard through a rebalance pass. Legacy rejects it with
+/// `TooManyPartitions` in both handlers (`1..=MAX` on create, `== 0` on
+/// delete), so the code matches rather than inventing a new one.
+const fn validate_partitions_change_count(partitions_count: u32) -> Result<(), IggyError> {
+ if partitions_count == 0 {
+ return Err(IggyError::TooManyPartitions);
+ }
+ validate_partitions_count(partitions_count)
+}
+
+/// Static create-topic bounds shared by the TCP and HTTP ingresses. Runs
+/// pre-consensus: a rejected request must not burn a replicated log entry,
+/// and `prepare_request` errors evict the session instead of denying typed.
+/// `ServerDefault` is exempt from the size floor (it resolves against server
+/// config at admission, matching legacy); `Unlimited` passes numerically.
+/// `segment_size_bytes` is the topic's RESOLVED segment size (explicit
+/// option, else this node's default), so a per-topic segment above the
+/// global default still floors the topic cap.
+pub fn validate_topic_bounds(
+ partitions_count: u32,
+ max_topic_size: MaxTopicSize,
+ segment_size_bytes: u64,
+) -> Result<(), IggyError> {
+ validate_partitions_count(partitions_count)?;
+ validate_topic_size_floor(max_topic_size, segment_size_bytes)
+}
+
+/// A topic cap below one segment can never be enforced: the first segment
+/// already exceeds it. Split out of [`validate_topic_bounds`] because update
+/// admission checks the cap without a partitions count to check.
+pub fn validate_topic_size_floor(
+ max_topic_size: MaxTopicSize,
+ segment_size_bytes: u64,
+) -> Result<(), IggyError> {
+ if !matches!(max_topic_size, MaxTopicSize::ServerDefault)
+ && max_topic_size.as_bytes_u64() < segment_size_bytes
+ {
+ return Err(IggyError::InvalidTopicSize(
+ max_topic_size,
+ IggyByteSize::from(segment_size_bytes),
+ ));
+ }
+ Ok(())
+}
+
+/// Announce an accepted `max_topic_size` the server cannot enforce as written.
+///
+/// [`validate_topic_size_floor`] admits any cap of one segment or more, but
+/// retention runs PER PARTITION and floors each partition's share at one SEALED
+/// segment, which reaches up to one maximum bus frame past `segment_size`. A cap
+/// between the two is stored and echoed back verbatim while the server actually
+/// keeps `(segment_size + max_message_size) * partitions_count`, so the only
+/// moment an operator can be told is the one where they set it.
+///
+/// Warns rather than rejects: which caps are accepted is client-visible wire
+/// behavior, and tightening it would break topics that already exist.
+pub fn warn_unenforceable_topic_size(
+ max_topic_size: MaxTopicSize,
+ segment_size_bytes: u64,
+ max_message_size_bytes: usize,
+ partitions_count: u32,
+) {
+ let MaxTopicSize::Custom(configured) = max_topic_size else {
+ return;
+ };
+ let max_message_size_bytes = u64::try_from(max_message_size_bytes).unwrap_or(u64::MAX);
+ let per_partition_floor = segment_size_bytes.saturating_add(max_message_size_bytes);
+ let topic_floor = per_partition_floor.saturating_mul(u64::from(partitions_count));
+ if configured.as_bytes_u64() >= topic_floor {
+ return;
+ }
+ warn!(
+ max_topic_size = configured.as_bytes_u64(),
+ partitions_count,
+ segment_size = segment_size_bytes,
+ enforced_per_partition = per_partition_floor,
+ "{UNENFORCEABLE_TOPIC_SIZE_WARN}"
+ );
+}
+
+/// Announce the same unenforceable cap when partitions are ADDED to a topic.
+///
+/// The cap is topic-wide but enforcement is per partition, so every added
+/// partition shrinks the share: a cap that cleared the floor when the topic was
+/// created can stop clearing it here. The request carries only the delta, so
+/// the stored cap, segment size and current partition count come from metadata.
+pub fn warn_unenforceable_topic_size_on_partition_add(
+ streams: &Streams,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ max_message_size_bytes: usize,
+ added_partitions_count: u32,
+) {
+ let Some(((stream_slab, topic_slab), _)) = streams.partition_count_context(stream_id, topic_id)
+ else {
+ return;
+ };
+ let Some((_, max_topic_size, partitions_count, segment_size)) =
+ streams.topic_retention_config(stream_slab, topic_slab)
+ else {
+ return;
+ };
+ warn_unenforceable_topic_size(
+ max_topic_size,
+ segment_size.map_or(iggy_common::DEFAULT_SEGMENT_SIZE, |segment_size| {
+ segment_size.as_bytes_u64()
+ }),
+ max_message_size_bytes,
+ u32::try_from(partitions_count)
+ .unwrap_or(u32::MAX)
+ .saturating_add(added_partitions_count),
+ );
+}
+
+/// Reject option keys outside the resource's catalog, pre-consensus. Unknown
+/// keys are rejected rather than skipped: a silently ignored knob would hand
+/// the client server defaults without it ever learning. Streams and users
+/// have no catalog keys yet, so `known` is empty for both until one lands.
+pub fn validate_option_keys(options: &WireOptions, known: &[&str]) -> Result<(), IggyError> {
+ for entry in options {
+ // Wire validation already enforced UTF-8 string keys.
+ let key = String::from_utf8_lossy(entry.key);
+ if !known.contains(&key.as_ref()) {
+ return Err(IggyError::UnsupportedOptionKey(key.into_owned()));
+ }
+ }
+ Ok(())
+}
+
+/// 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.
+#[allow(clippy::too_many_lines)]
+fn static_bounds<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ request: &Message<RoutedRequestHeader>,
+) -> Result<(), IggyError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ match request.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(()),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn create_topic_bounds_deny_pre_consensus() {
+ let segment_size = iggy_common::DEFAULT_SEGMENT_SIZE;
+ assert!(segment_size > 0, "default segment size must be nonzero");
+
+ assert!(
+ validate_topic_bounds(
+ MAX_PARTITIONS_PER_REQUEST,
+ MaxTopicSize::ServerDefault,
+ segment_size
+ )
+ .is_ok(),
+ "the partition cap itself is admissible"
+ );
+ assert!(
+ matches!(
+ validate_topic_bounds(
+ MAX_PARTITIONS_PER_REQUEST + 1,
+ MaxTopicSize::ServerDefault,
+ segment_size
+ ),
+ Err(IggyError::TooManyPartitions)
+ ),
+ "one past the partition cap must deny"
+ );
+ // ServerDefault is numerically 0 yet exempt from the segment-size
+ // floor: it resolves against server config, matching legacy.
+ assert!(validate_topic_bounds(1, MaxTopicSize::ServerDefault, segment_size).is_ok());
+ assert!(validate_topic_bounds(1, MaxTopicSize::Unlimited, segment_size).is_ok());
+ let below_floor = MaxTopicSize::Custom((segment_size - 1).into());
+ assert!(
+ matches!(
+ validate_topic_bounds(1, below_floor, segment_size),
+ Err(IggyError::InvalidTopicSize(size, floor))
+ if size == below_floor && floor == IggyByteSize::from(segment_size)
+ ),
+ "custom size below the segment size must deny with the bounds"
+ );
+ let at_floor = MaxTopicSize::Custom(IggyByteSize::from(segment_size));
+ assert!(
+ validate_topic_bounds(1, at_floor, segment_size).is_ok(),
+ "a topic exactly one segment large is admissible"
+ );
+ }
+
+ #[test]
+ fn partitions_count_cap_denies_pre_consensus() {
+ assert!(
+ validate_partitions_count(MAX_PARTITIONS_PER_REQUEST).is_ok(),
+ "the cap itself is admissible"
+ );
+ assert!(
+ matches!(
+ validate_partitions_count(MAX_PARTITIONS_PER_REQUEST + 1),
+ Err(IggyError::TooManyPartitions)
+ ),
+ "one past the cap must deny"
+ );
+ // Zero passes the shared cap because a zero-partition TOPIC is legal
+ // (legacy `create_topic` admits `0..=MAX`).
+ assert!(validate_partitions_count(0).is_ok());
+ }
+
+ #[test]
+ fn zero_partitions_change_denies_pre_consensus() {
+ // Adding or removing zero partitions is a no-op that would still burn
+ // a replicated log entry and force a rebalance. Legacy rejects it with
+ // `TooManyPartitions` in both handlers, so the code matches.
+ assert!(
+ matches!(
+ validate_partitions_change_count(0),
+ Err(IggyError::TooManyPartitions)
+ ),
+ "adding or removing zero partitions must deny"
+ );
+ assert!(validate_partitions_change_count(1).is_ok());
+ assert!(validate_partitions_change_count(MAX_PARTITIONS_PER_REQUEST).is_ok());
+ assert!(
+ matches!(
+ validate_partitions_change_count(MAX_PARTITIONS_PER_REQUEST + 1),
+ Err(IggyError::TooManyPartitions)
+ ),
+ "the cap still applies"
+ );
+ }
+}
diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs
index b9e744755..517d3be63 100644
--- a/core/server/src/server_error.rs
+++ b/core/server/src/server_error.rs
@@ -122,6 +122,27 @@ pub enum ServerError {
poll: std::time::Duration,
drain: std::time::Duration,
},
+ #[error("system.sharding.shutdown_join_timeout must be <= {max:?}; got {value:?}")]
+ InvalidShutdownJoinTimeout {
+ value: std::time::Duration,
+ max: std::time::Duration,
+ },
+ #[error(
+ "system.sharding.shutdown_join_timeout ({join:?}) must be >= \
+ shutdown_drain_timeout ({drain:?})"
+ )]
+ ShutdownJoinBelowDrain {
+ join: std::time::Duration,
+ drain: std::time::Duration,
+ },
+ #[error(
+ "system.sharding.reconcile_periodic_interval must be in (0, {max:?}]; got {value:?}. \
+ Note that \"0\", \"none\", \"unlimited\", and \"disabled\" all parse to zero"
+ )]
+ InvalidReconcilePeriodicInterval {
+ value: std::time::Duration,
+ max: std::time::Duration,
+ },
#[error("failed to serialize current server config")]
CurrentConfigSerialize(#[source] toml::ser::Error),
#[error("failed to write current server config at {path}")]
diff --git a/core/server/src/session_manager.rs b/core/server/src/session_manager.rs
index d09bd4c07..de7bfead3 100644
--- a/core/server/src/session_manager.rs
+++ b/core/server/src/session_manager.rs
@@ -26,13 +26,36 @@
//! 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, the transport peer address
+/// and the read-your-writes floor. `Default` (everything absent, no address,
+/// floor `0`) stands for a connection neither this map nor the bus knows.
+#[derive(Debug, Clone, Copy, Default)]
+pub(crate) struct ConnectionContext {
+ /// `(client_id, session)` once register committed, `None` before.
+ pub bound: Option<(u128, u64)>,
+ /// Acting user from `login`, `None` while still `Connected`.
+ pub user_id: Option<u32>,
+ /// Peer address recorded by [`SessionManager::ensure_connection`]; the
+ /// non-replicated reads pick the advertised address from it, and
+ /// `None` degrades to the catch-all address.
+ pub address: Option<SocketAddr>,
+ /// Read-your-writes floor: the metadata commit this connection's own
+ /// writes have reached, which its reads must not be served below.
+ ///
+ /// A connection this map does not know reads as `0` -- it was promised
+ /// nothing, so its reads wait for nothing.
+ pub metadata_watermark: u64,
+}
+
/// Connection lifecycle states.
///
/// ```text
@@ -107,10 +130,14 @@ pub struct Connection {
/// per consensus session). If a client reconnects with the same `client_id`,
/// the old connection must be evicted first.
pub struct SessionManager {
- connections: HashMap<u128, Connection>,
+ /// `ahash` over `std`: connection and client ids are server-minted, so
+ /// there is no `HashDoS` surface, and both maps sit on the per-frame path.
+ /// Neither is order-sensitive (`iter_clients` and `collect_stale` both
+ /// consume the whole map).
+ connections: AHashMap<u128, Connection>,
/// Reverse index: `client_id` → `connection_id` for fast lookup when
/// a consensus reply arrives and needs routing to the right connection.
- client_to_connection: HashMap<u128, u128>,
+ client_to_connection: AHashMap<u128, u128>,
/// This shard's copy of the configured cluster roster, served by the
/// `GetClusterMetadata` read. Lives here because it is the
/// per-shard context already threaded to the non-replicated read path;
@@ -122,8 +149,8 @@ impl SessionManager {
#[must_use]
pub fn new() -> Self {
Self {
- connections: HashMap::new(),
- client_to_connection: HashMap::new(),
+ connections: AHashMap::new(),
+ client_to_connection: AHashMap::new(),
cluster_roster: Rc::new(ClusterRoster::disabled()),
}
}
@@ -157,12 +184,31 @@ impl SessionManager {
});
}
- /// Record a liveness heartbeat (`ping`) for a connection, resetting its
- /// staleness clock. No-op for an unknown connection.
- pub fn record_heartbeat(&mut self, connection_id: u128) {
- if let Some(conn) = self.connections.get_mut(&connection_id) {
- conn.last_heartbeat = Instant::now();
- }
+ /// The request funnel's per-frame view of a connection: stamp the
+ /// liveness clock and read back everything the dispatch arms resolve
+ /// from it, in ONE map lookup.
+ ///
+ /// `None` means the connection is not registered yet, which happens
+ /// only on a transport's first frame; the caller installs it from the
+ /// bus metadata ([`Self::ensure_connection`]) and asks again.
+ pub(crate) fn touch_connection(&mut self, connection_id: u128) -> Option<ConnectionContext> {
+ let conn = self.connections.get_mut(&connection_id)?;
+ conn.last_heartbeat = Instant::now();
+ let (bound, user_id) = match conn.state {
+ ConnectionState::Bound {
+ user_id,
+ client_id,
+ session,
+ } => (Some((client_id, session)), Some(user_id)),
+ ConnectionState::Authenticated { user_id } => (None, Some(user_id)),
+ ConnectionState::Connected => (None, None),
+ };
+ Some(ConnectionContext {
+ bound,
+ user_id,
+ address: Some(conn.address),
+ metadata_watermark: conn.metadata_watermark,
+ })
}
/// Connection ids whose last heartbeat is older than `max_age` -- the
@@ -331,7 +377,11 @@ impl SessionManager {
/// Look up the consensus session for a connection.
///
- /// Returns `(client_id, session)` if the connection is `Bound`, `None` otherwise.
+ /// Returns `(client_id, session)` if the connection is `Bound`, `None`
+ /// otherwise. The request funnel reads it through the crate-private
+ /// `touch_connection` instead, which resolves it in the same lookup as
+ /// the heartbeat; this is for the session-op paths that only need the
+ /// binding.
#[must_use]
pub fn get_session(&self, connection_id: u128) -> Option<(u128, u64)> {
let conn = self.connections.get(&connection_id)?;
@@ -343,38 +393,6 @@ impl SessionManager {
}
}
- /// The transport-level peer address a connection arrived from, recorded by
- /// [`Self::ensure_connection`] for every transport. The non-replicated
- /// read path uses it to pick the advertised address a client is told
- /// about; `None` (unknown connection) degrades to the catch-all address.
- #[must_use]
- pub fn connection_address(&self, connection_id: u128) -> Option<SocketAddr> {
- self.connections
- .get(&connection_id)
- .map(|conn| conn.address)
- }
-
- /// Acting user, transport peer address and metadata watermark for a
- /// connection, in one map lookup: the non-replicated dispatch path needs
- /// all three per request, and the separate accessors would walk the
- /// connection map (and take the shared borrow) once each.
- ///
- /// An unknown connection reads as a watermark of `0`: it was promised
- /// nothing, so its reads wait for nothing.
- #[must_use]
- pub fn read_context(&self, connection_id: u128) -> (Option<u32>, Option<SocketAddr>, u64) {
- let Some(conn) = self.connections.get(&connection_id) else {
- return (None, None, 0);
- };
- let user_id = match conn.state {
- ConnectionState::Authenticated { user_id } | ConnectionState::Bound { user_id, .. } => {
- Some(user_id)
- }
- ConnectionState::Connected => None,
- };
- (user_id, Some(conn.address), conn.metadata_watermark)
- }
-
/// Look up the authenticated user id for a connection.
#[must_use]
pub fn get_user_id(&self, connection_id: u128) -> Option<u32> {
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 5132085fd..28816aa46 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -362,7 +362,7 @@ pub fn new_shard(
// every op above the floor must mutate the table. A frontier only fences a
// LIVE table a state transfer just replaced.
apply_committed_prepare(
- &metadata.mux_stm,
+ &*metadata.mux_stm,
&metadata.client_table,
true,
|_| {},