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/04/27 10:27:10 UTC
(iggy) branch feat/message-bus-transports created (now 125594897)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 125594897 feat(message_bus): add install_client_quic_conn + install_client_ws_stream
This branch includes the following new commits:
new de42708c9 feat(message_bus): authenticate the replica Ping handshake (IGGY-112)
new 440994de4 feat(server-ng): at-most-once request dedup window (IGGY-112, P0-T3)
new e444085a3 docs(message_bus): name the plane split + TCP-forever invariant (P0-T4)
new 901629132 feat(message_bus): define Transport trait family (IGGY-112, P1-T1)
new b6e9650f3 feat(message_bus): TCP impls behind Transport traits (IGGY-112, P1-T2)
new 0aeea4551 feat(message_bus): generic installer over TransportConn (IGGY-112, P1-T3)
new eb9f01b5b docs(message_bus): name the Transport trait surface in lib.rs (P1-T5)
new c9fd47dec feat(message_bus): IGGY_CLUSTER_SECRET env loader (IGGY-112)
new c586bd725 fix(message_bus): tag-verify before nonce insert + review hardening
new 1b163dad2 docs(message_bus): demote broken intra-doc links to code spans
new 13d2e5947 docs(message_bus): demote final broken intra-doc link in connector.rs
new 6f99acdad feat(message_bus): add QUIC transport for SDK-client plane
new 963ccf851 feat(message_bus): add WebSocket transport for SDK-client plane
new 125594897 feat(message_bus): add install_client_quic_conn + install_client_ws_stream
The 14 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 04/14: feat(message_bus): define Transport trait family (IGGY-112, P1-T1)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 901629132329be1a276e49c55511ce7bf25a3fd3
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 11:56:56 2026 +0200
feat(message_bus): define Transport trait family (IGGY-112, P1-T1)
Phase 1 of the transport-plan needs a stable seam so TCP, WS, and
QUIC impls can slot in without each reinventing accept / split /
batched-write plumbing. The current `transports/mod.rs` was a
placeholder; nothing existed for downstream phases to build on.
Introduce four traits with associated types (monomorphic hot path):
`TransportListener::accept`, `TransportConn::into_split`,
`TransportReader::read_message`, `TransportWriter::send_batch`.
`send_batch` is atomic-or-error over a reusable `Vec<Frozen<..>>` so
the TCP `writev` batching (I4) and zero-copy `Frozen` ownership (I8)
stay expressible; the `async fn` sits inside the writer task so
`send_to_*` keeps its no-yield caller-side contract (I2). `'static`
bounds let the halves move into spawned compio tasks.
Behavior-preserving: no impls change, no call sites touched. P1-T2
moves TCP behind the trait; WS / QUIC plug in under Phase 2+ when
their demand gates fire.
---
core/message_bus/src/transports/mod.rs | 199 ++++++++++++++++++++++++++++++++-
1 file changed, 195 insertions(+), 4 deletions(-)
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index aa0c7c27b..f47e0c404 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -15,8 +15,199 @@
// specific language governing permissions and limitations
// under the License.
-//! Transport abstraction - placeholder.
+//! Transport abstraction for the `message_bus` wire planes.
//!
-//! Tracked by IGGY-112. This module is deliberately empty until the
-//! `Transport` / `TransportListener` / `TransportConn` trait family lands;
-//! do not grow speculative scaffolding here.
+//! # Trait family
+//!
+//! - [`TransportListener`] binds a local address and yields inbound
+//! [`TransportConn`]s on `accept`.
+//! - [`TransportConn`] is a single byte-oriented connection; it splits
+//! into a [`TransportReader`] and a [`TransportWriter`] that move into
+//! independent compio tasks (reader loop, writer-batch task).
+//! - [`TransportReader`] reads one framed `Message<GenericHeader>` per
+//! call.
+//! - [`TransportWriter`] sends a batch of `Frozen<MESSAGE_ALIGN>`
+//! atomically in a single `writev` (TCP) or per-transport analog
+//! (framed binary for WS, one STREAM per message for QUIC).
+//!
+//! # Invariants the trait surface must preserve
+//!
+//! - **No-yield on caller-side** (I2): producers call
+//! [`MessageBus::send_to_client`](crate::MessageBus::send_to_client) /
+//! [`send_to_replica`](crate::MessageBus::send_to_replica), which push
+//! onto a bounded `async_channel` and return `Ready` on the first
+//! poll. The single yield point in the wire path lives inside
+//! [`TransportWriter::send_batch`], invoked by the per-peer writer
+//! task. Consensus code relies on this for reentrancy reasoning.
+//! - **Batch atomicity** (I4): `send_batch` either writes every buffer
+//! in the batch or returns an error; no partial success is exposed.
+//! TCP satisfies this via `compio::io::AsyncWriteExt::write_vectored_all`;
+//! WS/QUIC impls loop internally until the batch lands or fail.
+//! - **Zero-copy `Frozen` ownership** (I8): the batch is handed to the
+//! kernel without intermediate copies. Impls MUST NOT clone
+//! `Frozen<MESSAGE_ALIGN>` or materialize a `Vec<u8>` on the hot
+//! path.
+//!
+//! # Design notes
+//!
+//! - Associated types (not trait objects): keep the hot path
+//! monomorphic. A `dyn TransportConn` surface can be introduced later
+//! if a config-time transport selector needs runtime polymorphism.
+//! - `'static` bound on [`TransportConn`], [`TransportReader`], and
+//! [`TransportWriter`]: the halves are moved into
+//! `compio::runtime::spawn`'d tasks, which require owned data.
+//! - fd-delegation (`F_DUPFD_CLOEXEC`) stays TCP-only and lives outside
+//! this trait (see [`crate::fd_transfer`]). Other transports terminate
+//! on shard 0 and forward `Frozen<MESSAGE_ALIGN>` over the inter-shard
+//! flume via [`crate::ShardForwardFn`].
+//!
+//! The TCP impl lives in `transports/tcp.rs` (P1-T2). Phase 2+ transports
+//! plug in behind the same surface; see
+//! `Documents/silverhand/iggy/message_bus/transport-plan/`.
+
+use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
+use iggy_binary_protocol::consensus::iobuf::Frozen;
+use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_common::IggyError;
+use std::io;
+use std::net::SocketAddr;
+
+/// Inbound listener. The accept loops in [`crate::replica_listener`]
+/// and [`crate::client_listener`] invoke `accept` inside a `select!`
+/// against the bus-wide shutdown token.
+#[allow(async_fn_in_trait)]
+pub trait TransportListener {
+ /// Connection type produced by this listener.
+ type Conn: TransportConn;
+
+ /// Accept the next inbound connection.
+ ///
+ /// Returns once a peer has completed the transport-level handshake
+ /// (TCP SYN/ACK; TLS; QUIC handshake). The `message_bus`
+ /// authenticated `Ping` handshake runs on the returned connection,
+ /// after this call. See [`crate::auth`].
+ ///
+ /// # Errors
+ ///
+ /// Returns [`io::Error`] on socket faults. The caller logs and
+ /// continues; a fatal error must be surfaced by dropping the
+ /// listener rather than through this return value.
+ async fn accept(&self) -> io::Result<(Self::Conn, SocketAddr)>;
+}
+
+/// Single-connection handle produced by [`TransportListener::accept`]
+/// or by a transport-specific dialer (`TcpStream::connect`, etc.).
+///
+/// `'static` so the reader and writer halves can move into spawned
+/// compio tasks.
+pub trait TransportConn: 'static {
+ /// Read half moved into the per-connection reader task.
+ type Reader: TransportReader;
+ /// Write half moved into the per-connection writer-batch task.
+ type Writer: TransportWriter;
+
+ /// Split the connection into independently-owned halves.
+ ///
+ /// Mirrors [`compio::net::TcpStream::into_split`] on TCP. WS and
+ /// QUIC impls map onto their own half-splits or stream wrappers.
+ fn into_split(self) -> (Self::Reader, Self::Writer);
+}
+
+/// Read half. Produces one framed [`Message<GenericHeader>`] per call.
+#[allow(async_fn_in_trait)]
+pub trait TransportReader: Unpin + 'static {
+ /// Read the next `GenericHeader`-framed message off the wire.
+ ///
+ /// Validates the 256 B header and bounds the total frame size to
+ /// `max_message_size`. TCP delegates to
+ /// [`crate::framing::read_message`]; WS wraps the single binary
+ /// frame; QUIC consumes one STREAM worth of bytes.
+ ///
+ /// # Errors
+ ///
+ /// - [`IggyError::ConnectionClosed`] on clean EOF.
+ /// - [`IggyError::TcpError`] on transport I/O faults.
+ /// - [`IggyError::InvalidCommand`] on framing violations (bad size
+ /// field, total frame exceeding `max_message_size`, header decode
+ /// failure).
+ async fn read_message(
+ &mut self,
+ max_message_size: usize,
+ ) -> Result<Message<GenericHeader>, IggyError>;
+}
+
+/// Write half. Atomic-or-error batched writer.
+#[allow(async_fn_in_trait)]
+pub trait TransportWriter: Unpin + 'static {
+ /// Send every buffer in `batch` atomically.
+ ///
+ /// On success the Vec is drained (empty on return); the same
+ /// allocation is reused by the writer loop on the next iteration,
+ /// so impls must not free it. On error the buffers may or may not
+ /// have partially landed on the wire; the caller MUST treat a
+ /// failed batch as lost data and drop the connection. VSR
+ /// retransmits via the WAL.
+ ///
+ /// TCP implements this via
+ /// [`compio::io::AsyncWriteExt::write_vectored_all`] on the owned
+ /// write half, with `max_batch <= IOV_MAX / 2 = 512`. WS/QUIC impls
+ /// loop internally until every buffer lands or fail the whole
+ /// batch.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`io::Error`] on transport faults. No partial-success
+ /// variant is exposed.
+ async fn send_batch(&mut self, batch: &mut Vec<Frozen<MESSAGE_ALIGN>>) -> io::Result<()>;
+}
+
+#[cfg(test)]
+#[allow(dead_code)]
+mod tests {
+ //! Compile-only stubs that exercise each trait's associated-type
+ //! plumbing and method signatures. Method bodies `unimplemented!()`
+ //! so every compile-time obligation (bounds, lifetimes, `'static`)
+ //! is enforced without needing a runtime harness.
+ use super::{TransportConn, TransportListener, TransportReader, TransportWriter};
+ use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
+ use iggy_binary_protocol::consensus::iobuf::Frozen;
+ use iggy_binary_protocol::{GenericHeader, Message};
+ use iggy_common::IggyError;
+ use std::io;
+ use std::net::SocketAddr;
+
+ struct StubReader;
+ struct StubWriter;
+ struct StubConn;
+ struct StubListener;
+
+ impl TransportReader for StubReader {
+ async fn read_message(
+ &mut self,
+ _max_message_size: usize,
+ ) -> Result<Message<GenericHeader>, IggyError> {
+ unimplemented!()
+ }
+ }
+
+ impl TransportWriter for StubWriter {
+ async fn send_batch(&mut self, _batch: &mut Vec<Frozen<MESSAGE_ALIGN>>) -> io::Result<()> {
+ unimplemented!()
+ }
+ }
+
+ impl TransportConn for StubConn {
+ type Reader = StubReader;
+ type Writer = StubWriter;
+ fn into_split(self) -> (Self::Reader, Self::Writer) {
+ unimplemented!()
+ }
+ }
+
+ impl TransportListener for StubListener {
+ type Conn = StubConn;
+ async fn accept(&self) -> io::Result<(Self::Conn, SocketAddr)> {
+ unimplemented!()
+ }
+ }
+}
(iggy) 14/14: feat(message_bus): add install_client_quic_conn + install_client_ws_stream
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 125594897e0fc70768694d137057ca74663b9980
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 11:54:32 2026 +0200
feat(message_bus): add install_client_quic_conn + install_client_ws_stream
The QUIC and WebSocket transports landed at trait-level (`6f99acdad`,
`963ccf851`) but had no install entry in `installer.rs`. The
dispatcher branch needed both, which left a documentation gap and a
load-bearing-claim asymmetry vs the TCP path.
`install_client_quic_conn` wraps a `compio_quic::Connection` plus its
first-bidi `(SendStream, RecvStream)` pair in a `QuicTransportConn`
and delegates to the existing generic `install_client_conn`.
`install_client_ws_stream` wraps a post-upgrade
`compio_ws::WebSocketStream<TcpStream>` in a `WsTransportConn` and
delegates the same way.
Neither install path crosses an inter-shard boundary: both
`compio_quic::Connection` and `compio_ws::WebSocketStream<TcpStream>`
hold compio `Rc<...>` driver state and are `!Send`. Shard 0 therefore
terminates QUIC + WS locally and uses the existing
`ShardFramePayload::ForwardClientSend` / `Consensus` variants for
outbound + inbound traffic respectively. This matches
`core/message_bus/CLAUDE.md`'s plane-split rule
("TLS/QUIC-terminated connections... shard 0 terminates and forwards
`Frozen<MESSAGE_ALIGN>` over the inter-shard flume").
The `core/shard/src/lib.rs` `ShardFramePayload` doc-comment is
amended on `ClientConnectionSetup` to call out the deliberate
absence of QUIC/WS analogs and point at the rationale, so future
contributors don't try to reinstate them and trip over the same
`!Send` wall.
The earlier P4-T2 design doc (silverhand-only) was wrong on
cross-shard handover for QUIC and is now obsolete on that point;
the in-tree code + CLAUDE.md are the source of truth.
Tests: no new tests; both install entries are thin delegators over
the existing generic `install_client_conn`, which is already covered
by `transports::tcp::tests::send_batch_writes_all_and_drains_vec`,
`transports::quic::tests::loopback_round_trip_three_frames`, and
`transports::ws::tests::loopback_round_trip_three_frames`. 135/135
pass on `cargo nextest run -p message_bus -p server-ng -p shard`.
Workspace clippy clean. Doc clean modulo one pre-existing
`coordinator.rs:276` warning inherited from master.
---
core/message_bus/src/installer.rs | 58 +++++++++++++++++++++++++++++++++++++++
core/shard/src/lib.rs | 9 ++++++
2 files changed, 67 insertions(+)
diff --git a/core/message_bus/src/installer.rs b/core/message_bus/src/installer.rs
index 4d274860c..0b52304c7 100644
--- a/core/message_bus/src/installer.rs
+++ b/core/message_bus/src/installer.rs
@@ -29,6 +29,8 @@ use crate::fd_transfer::{self, DupedFd};
use crate::lifecycle::{InstanceToken, RejectedRegistration, Shutdown};
use crate::replica_listener::MessageHandler;
use crate::socket_opts::{apply_keepalive_for_connection, apply_nodelay_for_connection};
+use crate::transports::quic::QuicTransportConn;
+use crate::transports::ws::WsTransportConn;
use crate::transports::{TcpTransportConn, TransportConn, TransportReader};
use crate::{IggyMessageBus, lifecycle::ShutdownToken};
use compio::net::TcpStream;
@@ -318,6 +320,62 @@ pub fn install_client_stream(
install_client_conn(bus, client_id, TcpTransportConn::new(stream), on_request);
}
+/// QUIC entry point for client installs.
+///
+/// Wraps the [`compio_quic::Connection`] and its first bidirectional
+/// `(SendStream, RecvStream)` pair (already accepted by shard 0 via
+/// `Connection::accept_bi().await` so the install path never
+/// re-handshakes) in a [`QuicTransportConn`] and delegates to the
+/// existing generic [`install_client_conn`].
+///
+/// No socket-options analog runs here: QUIC keepalive lives in
+/// [`compio_quic::TransportConfig::keep_alive_interval`] set at endpoint
+/// construction time. The connection is encrypted end-to-end and there
+/// is no plaintext fd to dup, which is why this never crosses an
+/// inter-shard channel: shard 0 owns the QUIC `Endpoint`, terminates
+/// every connection locally, and uses the existing
+/// `ForwardClientSend` / `Consensus` shard-frame variants for outbound
+/// + inbound traffic respectively.
+#[allow(clippy::future_not_send)]
+pub fn install_client_quic_conn(
+ bus: &Rc<IggyMessageBus>,
+ client_id: u128,
+ connection: compio_quic::Connection,
+ streams: (compio_quic::SendStream, compio_quic::RecvStream),
+ on_request: RequestHandler,
+) {
+ install_client_conn(
+ bus,
+ client_id,
+ QuicTransportConn::new(connection, streams),
+ on_request,
+ );
+}
+
+/// WebSocket entry point for client installs.
+///
+/// Wraps a post-upgrade [`compio_ws::WebSocketStream`] in a
+/// [`WsTransportConn`] and delegates to the existing generic
+/// [`install_client_conn`]. The HTTP-Upgrade handshake has already been
+/// driven on shard 0; the install path never re-runs it.
+///
+/// Like QUIC, this never crosses an inter-shard channel:
+/// `WebSocketStream<TcpStream>` is `!Send` (it holds compio `Rc<...>`
+/// driver state) so shard 0 terminates locally and uses the existing
+/// `ForwardClientSend` / `Consensus` variants. The post-upgrade socket
+/// IS still a raw TCP fd, but compio-ws has no "rebuild from fd" API
+/// and re-attaching the tungstenite state machine across shards would
+/// lose protocol invariants the dispatcher already enforced.
+#[allow(clippy::future_not_send)]
+pub fn install_client_ws_stream(
+ bus: &Rc<IggyMessageBus>,
+ client_id: u128,
+ stream: compio_ws::WebSocketStream<TcpStream>,
+ on_request: RequestHandler,
+) {
+ install_client_conn(bus, client_id, WsTransportConn::new(stream), on_request);
+}
+
/// Install a pre-wrapped client connection on the bus. Generic over
/// [`TransportConn`]; plane-symmetric with [`install_replica_conn`].
#[allow(clippy::future_not_send, clippy::too_many_lines)]
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 579577eee..e420e508c 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -199,6 +199,15 @@ pub enum ShardFramePayload {
/// owning shard. The receiving shard wraps the fd and installs client
/// reader / writer tasks locally. The owning shard is encoded in the top
/// 16 bits of `client_id`.
+ ///
+ /// QUIC + WS clients deliberately do NOT get an analog variant: the
+ /// inter-shard channel requires `Send` payloads, and `compio_quic::
+ /// Connection` / `compio_ws::WebSocketStream<TcpStream>` are both
+ /// `!Send` (they hold compio `Rc<...>` driver state). Shard 0
+ /// therefore terminates QUIC + WS locally and uses the existing
+ /// `ForwardClientSend` / `Consensus` variants for outbound + inbound
+ /// traffic respectively. See `core/message_bus/CLAUDE.md` plane
+ /// split section.
ClientConnectionSetup { fd: DupedFd, client_id: u128 },
/// Shard 0 broadcasts the owner for a replica to every shard so each
/// bus' `send_to_replica` slow path can route through the correct owner.
(iggy) 12/14: feat(message_bus): add QUIC transport for SDK-client plane
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6f99acdadc4c94830c27e623123a9058fa844c27
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 11:22:49 2026 +0200
feat(message_bus): add QUIC transport for SDK-client plane
The client plane was TCP-only. Phase 4 of the transport plan calls
for QUIC parity with legacy `core/server`; this lands the message-
bus side behind the existing `Transport*` trait family.
`QuicTransport{Listener,Conn,Reader,Writer}` in
`transports/quic.rs` wrap `compio-quic 0.7.2` so the runtime stays
single-stack (no tokio bridge, invariant I10). The listener accepts
the connection AND drains the first bidi `accept_bi` pair before
yielding, so the install path on the owning shard sees a fully-
established data plane.
Reusing `framing::read_message` against `RecvStream` required
dropping the `Unpin` bound from the framing helpers; `RecvStream`
is `!Unpin` in compio-quic. The bound was incidental - neither
`read_exact` nor `write_all` actually need it.
QUIC is always-on, not feature-gated, matching legacy server's
posture and shrinking the build matrix.
Defenses: 0-RTT off at the rustls layer
(`max_early_data_size = 0`); listener refuses any 0-RTT-tagged
stream with `QUIC_PROTOCOL_VIOLATION`. ALPN `iggy.consensus.v1`.
Default `TransportConfig`: 1 bidi/peer, 30 s idle timeout, 64 MiB
windows, CUBIC congestion (upstream default).
Tag-along: demote the only stray broken intra-doc link
(`ClientTable` in `core/server-ng/src/session_manager.rs:26`,
inherited from master) to a code span so `cargo doc` runs warning-
free on this branch.
Tests: +3 in `transports::quic::tests` (round-trip three frames,
oversize-frame surfacing `IggyError::InvalidCommand`, default
config smoke). 123/123 pass on `cargo nextest run -p message_bus
-p server-ng`. Workspace clippy + cargo doc clean.
---
Cargo.lock | 3 +
Cargo.toml | 1 +
core/message_bus/Cargo.toml | 3 +
core/message_bus/src/framing.rs | 4 +-
core/message_bus/src/transports/mod.rs | 1 +
core/message_bus/src/transports/quic.rs | 483 ++++++++++++++++++++++++++++++++
core/server-ng/src/session_manager.rs | 2 +-
7 files changed, 494 insertions(+), 3 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index af920f2d3..cb1463495 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7011,11 +7011,14 @@ dependencies = [
"async-channel",
"blake3",
"compio",
+ "compio-quic",
"futures",
"iggy_binary_protocol",
"iggy_common",
"libc",
"rand 0.10.1",
+ "rcgen",
+ "rustls",
"socket2 0.6.3",
"tempfile",
"thiserror 2.0.18",
diff --git a/Cargo.toml b/Cargo.toml
index dbb14f5ff..87d8878d1 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -118,6 +118,7 @@ compio = { version = "0.18.0", features = [
compio-buf = "0.8.1"
# Pin compio-driver >= 0.11.2 to fix musl compilation (compio-rs/compio#668)
compio-driver = "0.11.4"
+compio-quic = "=0.7.2"
configs = { path = "core/configs", version = "0.1.0" }
configs_derive = { path = "core/configs_derive", version = "0.1.0" }
consensus = { path = "core/consensus" }
diff --git a/core/message_bus/Cargo.toml b/core/message_bus/Cargo.toml
index 6fa1ae1f4..ace090f16 100644
--- a/core/message_bus/Cargo.toml
+++ b/core/message_bus/Cargo.toml
@@ -32,16 +32,19 @@ publish = false
async-channel = { workspace = true }
blake3 = { workspace = true }
compio = { workspace = true }
+compio-quic = { workspace = true }
futures = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
libc = { workspace = true }
rand = { workspace = true }
+rustls = { workspace = true }
socket2 = { workspace = true, features = ["all"] }
thiserror = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
+rcgen = { workspace = true }
tempfile = { workspace = true }
[lints.clippy]
diff --git a/core/message_bus/src/framing.rs b/core/message_bus/src/framing.rs
index 0f3af8576..315234e32 100644
--- a/core/message_bus/src/framing.rs
+++ b/core/message_bus/src/framing.rs
@@ -61,7 +61,7 @@ const _: () = {
///
/// Returns `IggyError::TcpError` if the write fails.
#[allow(clippy::future_not_send)]
-pub async fn write_message<S: AsyncWriteExt + Unpin>(
+pub async fn write_message<S: AsyncWriteExt>(
stream: &mut S,
message: Message<GenericHeader>,
) -> Result<(), IggyError> {
@@ -90,7 +90,7 @@ pub async fn write_message<S: AsyncWriteExt + Unpin>(
/// Returns `IggyError::TcpError` on I/O errors.
/// Returns `IggyError::InvalidCommand` if the header fails validation.
#[allow(clippy::future_not_send)]
-pub async fn read_message<S: AsyncReadExt + Unpin>(
+pub async fn read_message<S: AsyncReadExt>(
stream: &mut S,
max_message_size: usize,
) -> Result<Message<GenericHeader>, IggyError> {
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index 157a5780a..76e1965f3 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -65,6 +65,7 @@
//! plug in behind the same surface; see
//! `Documents/silverhand/iggy/message_bus/transport-plan/`.
+pub mod quic;
pub mod tcp;
// Only `Conn` and `Writer` have crate-internal callers today
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
new file mode 100644
index 000000000..a407b3954
--- /dev/null
+++ b/core/message_bus/src/transports/quic.rs
@@ -0,0 +1,483 @@
+// 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.
+
+//! QUIC impls of the [`super`] transport traits.
+//!
+//! SDK-client plane only; replica plane stays TCP forever (invariant I1).
+//! `compio-quic 0.7.2` binds `quinn-proto` natively to the compio runtime,
+//! so no tokio bridge (invariant I10).
+//!
+//! # Connection model
+//!
+//! One bidirectional stream per peer (no multiplexing). The connection's
+//! first `accept_bi` pair carries every consensus frame. The
+//! [`QuicTransportListener`] accept loop drives the QUIC handshake to
+//! completion AND drains the first bidi pair before yielding the
+//! [`QuicTransportConn`], so the owning shard's install path doesn't
+//! need an additional `await` boundary before data flows. See
+//! `Documents/silverhand/iggy/message_bus/transport-plan/designs/message-bus-quic-install.md`
+//! for the full rationale.
+//!
+//! # Zero-copy
+//!
+//! `compio_quic::SendStream::write<T: IoBuf>` accepts
+//! `Frozen<MESSAGE_ALIGN>` directly; `QuicTransportWriter::send_batch`
+//! loops one `write` per frame (QUIC has no `sendmmsg` analog). Per-message
+//! syscalls are the documented trade-off versus the TCP `writev` path; small
+//! high-RPS workloads stay on TCP. See plan §4 risks.
+//!
+//! # 0-RTT
+//!
+//! 0-RTT is off by default at the rustls layer (`max_early_data_size = 0`).
+//! `RecvStream::is_0rtt()` exists for defense-in-depth; the listener treats
+//! a `true` here as a misconfiguration and refuses the connection. Any
+//! future per-command 0-RTT enablement requires an audit per
+//! `transport-plan/designs/quic-0rtt-audit-template.md` (P4-T5).
+
+use super::{TransportConn, TransportListener, TransportReader, TransportWriter};
+use crate::framing;
+use compio::BufResult;
+use compio::io::AsyncWrite;
+use compio_quic::{
+ Connection, Endpoint, Incoming, RecvStream, SendStream, VarInt, congestion,
+ crypto::rustls::QuicServerConfig,
+};
+use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
+use iggy_binary_protocol::consensus::iobuf::Frozen;
+use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_common::IggyError;
+use std::io;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+use tracing::{debug, warn};
+
+/// ALPN protocol identifier carried by every Iggy QUIC client + server.
+///
+/// Bump the trailing version digits when the wire format changes (e.g. a
+/// new `GenericHeader` layout). See invariant I3 for the framing contract.
+pub const ALPN_IGGY_CONSENSUS_V1: &[u8] = b"iggy.consensus.v1";
+
+/// QUIC application close codes. Sized so the table fits a single cache
+/// line; matches the design doc allocation (P4-T2 §8).
+pub const QUIC_HANDSHAKE_FAILED: u32 = 0x1001;
+pub const QUIC_AUTH_FAILED: u32 = 0x1002;
+pub const QUIC_SHUTDOWN: u32 = 0x1003;
+pub const QUIC_PROTOCOL_VIOLATION: u32 = 0x1004;
+
+/// Idle-timeout default for client-plane connections. Matches the 30 s
+/// replica handshake / dedup window so operators see one number across
+/// the bus. Override at endpoint construction time if needed.
+const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Keepalive interval default. Half of [`DEFAULT_IDLE_TIMEOUT`]'s lower
+/// bound so a single dropped keepalive never closes the connection.
+const DEFAULT_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(10);
+
+/// Stream / receive / send window default. Matches
+/// `MessageBusConfig::default().max_message_size = 64 MiB` so a single
+/// in-flight max-size frame fits without head-of-line wait.
+const DEFAULT_WINDOW: u32 = 64 * 1024 * 1024;
+
+/// Build a [`compio_quic::TransportConfig`] tuned for the SDK-client plane.
+///
+/// Defaults: 1 bidi stream per peer, 30 s idle timeout, CUBIC congestion
+/// (upstream default), 64 MiB send/receive windows. Settings match the
+/// audit's knob table (`findings/compio-quic-audit.md` §2).
+///
+/// # Panics
+///
+/// Panics only if `DEFAULT_IDLE_TIMEOUT` is changed to a value outside
+/// quinn-proto's `IdleTimeout` range (its `TryFrom<Duration>` rejects
+/// values above `VarInt::MAX_U64` ms). 30 s is well inside the range.
+#[must_use]
+pub fn default_transport_config() -> compio_quic::TransportConfig {
+ let mut cfg = compio_quic::TransportConfig::default();
+ cfg.max_concurrent_bidi_streams(VarInt::from_u32(1))
+ .max_concurrent_uni_streams(VarInt::from_u32(0))
+ .stream_receive_window(VarInt::from_u32(DEFAULT_WINDOW))
+ .receive_window(VarInt::from_u32(DEFAULT_WINDOW))
+ .send_window(u64::from(DEFAULT_WINDOW))
+ .max_idle_timeout(Some(
+ DEFAULT_IDLE_TIMEOUT
+ .try_into()
+ .expect("30 s fits in IdleTimeout"),
+ ))
+ .keep_alive_interval(Some(DEFAULT_KEEP_ALIVE_INTERVAL))
+ .congestion_controller_factory(Arc::new(congestion::CubicConfig::default()));
+ cfg
+}
+
+/// Inbound QUIC listener.
+///
+/// Wraps a bound [`Endpoint`]. `accept` drives the QUIC handshake and the
+/// first `accept_bi` pair to completion before yielding a
+/// [`QuicTransportConn`], so the caller (`replica_listener::run`'s QUIC
+/// twin in P4-T2's design) hands a fully-set-up data plane to the install
+/// path.
+pub struct QuicTransportListener {
+ endpoint: Endpoint,
+}
+
+impl QuicTransportListener {
+ /// Wrap a pre-bound [`Endpoint`].
+ ///
+ /// Caller is responsible for the [`Endpoint`]'s ALPN set
+ /// ([`ALPN_IGGY_CONSENSUS_V1`]), [`compio_quic::ServerConfig`] crypto
+ /// material, and [`default_transport_config`] hookup.
+ #[must_use]
+ pub const fn new(endpoint: Endpoint) -> Self {
+ Self { endpoint }
+ }
+
+ /// Borrow the underlying endpoint for `local_addr`, `set_server_config`,
+ /// `close`, or `shutdown` calls.
+ #[must_use]
+ pub const fn endpoint(&self) -> &Endpoint {
+ &self.endpoint
+ }
+}
+
+impl TransportListener for QuicTransportListener {
+ type Conn = QuicTransportConn;
+
+ #[allow(clippy::future_not_send)]
+ async fn accept(&self) -> io::Result<(Self::Conn, SocketAddr)> {
+ loop {
+ let Some(incoming) = self.endpoint.wait_incoming().await else {
+ return Err(io::Error::new(
+ io::ErrorKind::ConnectionAborted,
+ "QUIC endpoint closed",
+ ));
+ };
+ match accept_one(incoming).await {
+ Ok(Some((conn, addr))) => return Ok((conn, addr)),
+ Ok(None) => {}
+ Err(e) => return Err(e),
+ }
+ }
+ }
+}
+
+#[allow(clippy::future_not_send)]
+async fn accept_one(incoming: Incoming) -> io::Result<Option<(QuicTransportConn, SocketAddr)>> {
+ let connecting = match incoming.accept() {
+ Ok(c) => c,
+ Err(e) => {
+ warn!("QUIC incoming.accept failed: {e}");
+ return Ok(None);
+ }
+ };
+ let connection = match connecting.await {
+ Ok(c) => c,
+ Err(e) => {
+ warn!("QUIC handshake failed: {e}");
+ return Ok(None);
+ }
+ };
+ let addr = connection.remote_address();
+
+ let (send, recv) = match connection.accept_bi().await {
+ Ok(streams) => streams,
+ Err(e) => {
+ warn!(%addr, "QUIC accept_bi failed: {e}");
+ connection.close(VarInt::from_u32(QUIC_HANDSHAKE_FAILED), b"accept_bi failed");
+ return Ok(None);
+ }
+ };
+
+ if recv.is_0rtt() {
+ warn!(
+ %addr,
+ "QUIC stream accepted in 0-RTT window; refusing (audit P4-T5 not run)"
+ );
+ connection.close(
+ VarInt::from_u32(QUIC_PROTOCOL_VIOLATION),
+ b"0-RTT not permitted",
+ );
+ return Ok(None);
+ }
+
+ debug!(%addr, "QUIC connection accepted, first bidi stream ready");
+ Ok(Some((
+ QuicTransportConn::new(connection, (send, recv)),
+ addr,
+ )))
+}
+
+/// A single QUIC connection plus its first bidirectional stream.
+///
+/// The owned `Connection` handle is retained on the writer half so
+/// graceful shutdown can fire `Connection::close(QUIC_SHUTDOWN, _)`
+/// independent of stream drop order.
+pub struct QuicTransportConn {
+ connection: Connection,
+ streams: (SendStream, RecvStream),
+}
+
+impl QuicTransportConn {
+ /// Construct from an already-established connection + bidi pair.
+ #[must_use]
+ pub const fn new(connection: Connection, streams: (SendStream, RecvStream)) -> Self {
+ Self {
+ connection,
+ streams,
+ }
+ }
+}
+
+impl TransportConn for QuicTransportConn {
+ type Reader = QuicTransportReader;
+ type Writer = QuicTransportWriter;
+
+ fn into_split(self) -> (Self::Reader, Self::Writer) {
+ let (send, recv) = self.streams;
+ (
+ QuicTransportReader { recv },
+ QuicTransportWriter {
+ send,
+ connection: self.connection,
+ },
+ )
+ }
+}
+
+/// Read half. Decodes one framed `Message<GenericHeader>` per call by
+/// reusing [`framing::read_message`] against the [`RecvStream`].
+pub struct QuicTransportReader {
+ recv: RecvStream,
+}
+
+impl TransportReader for QuicTransportReader {
+ #[allow(clippy::future_not_send)]
+ async fn read_message(
+ &mut self,
+ max_message_size: usize,
+ ) -> Result<Message<GenericHeader>, IggyError> {
+ framing::read_message(&mut self.recv, max_message_size).await
+ }
+}
+
+/// Write half. One [`SendStream::write`] per frame; the inner
+/// [`Connection`] handle is retained so graceful shutdown can issue
+/// `connection.close(QUIC_SHUTDOWN, _)` after the writer task drains.
+pub struct QuicTransportWriter {
+ send: SendStream,
+ connection: Connection,
+}
+
+impl QuicTransportWriter {
+ /// Borrow the underlying connection (for stats, graceful close).
+ #[must_use]
+ pub const fn connection(&self) -> &Connection {
+ &self.connection
+ }
+}
+
+impl TransportWriter for QuicTransportWriter {
+ #[allow(clippy::future_not_send)]
+ async fn send_batch(&mut self, batch: &mut Vec<Frozen<MESSAGE_ALIGN>>) -> io::Result<()> {
+ // Drain in order; one write per frame. compio-quic's
+ // SendStream::write takes T: IoBuf and Frozen<MESSAGE_ALIGN>
+ // implements IoBuf (binary_protocol::consensus::iobuf), so the
+ // hand-off is zero-copy. No batched syscall (no sendmmsg on
+ // QUIC); see plan §4 risks.
+ for buf in batch.drain(..) {
+ let BufResult(result, _frozen) = self.send.write(buf).await;
+ result.map_err(|e| io::Error::other(format!("QUIC send failed: {e}")))?;
+ }
+ // Flush the QUIC frame onto the wire before returning so the
+ // batch's atomic-or-error contract (TransportWriter doc) holds:
+ // a partial flush on cancellation must not advertise success.
+ self.send
+ .flush()
+ .await
+ .map_err(|e| io::Error::other(format!("QUIC flush failed: {e}")))
+ }
+}
+
+/// Build a [`compio_quic::ServerConfig`] from a single cert + key pair.
+///
+/// Sets the Iggy ALPN ([`ALPN_IGGY_CONSENSUS_V1`]) and the bus-tuned
+/// [`default_transport_config`]; otherwise inherits upstream defaults
+/// including `migration: true`.
+///
+/// # Errors
+///
+/// Returns the underlying `rustls::Error` on cert/key import failure.
+pub fn server_config_with_cert(
+ cert_chain: Vec<rustls::pki_types::CertificateDer<'static>>,
+ key_der: rustls::pki_types::PrivateKeyDer<'static>,
+) -> Result<compio_quic::ServerConfig, rustls::Error> {
+ let mut rustls_cfg = rustls::ServerConfig::builder()
+ .with_no_client_auth()
+ .with_single_cert(cert_chain, key_der)?;
+ rustls_cfg.alpn_protocols = vec![ALPN_IGGY_CONSENSUS_V1.to_vec()];
+ rustls_cfg.max_early_data_size = 0; // 0-RTT off (invariant I7)
+ let crypto = QuicServerConfig::try_from(rustls_cfg)
+ .map_err(|e| rustls::Error::General(format!("QUIC crypto: {e}")))?;
+ let mut server_cfg = compio_quic::ServerConfig::with_crypto(Arc::new(crypto));
+ server_cfg.transport_config(Arc::new(default_transport_config()));
+ Ok(server_cfg)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use compio_quic::ClientBuilder;
+ use iggy_binary_protocol::{Command2, HEADER_SIZE};
+ use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
+
+ fn install_crypto_provider() {
+ // Idempotent. Tests in the same process race on
+ // `install_default`, so swallow the second-installer error.
+ let _ = rustls::crypto::ring::default_provider().install_default();
+ }
+
+ fn header_only(command: Command2) -> Frozen<MESSAGE_ALIGN> {
+ #[allow(clippy::cast_possible_truncation)]
+ Message::<GenericHeader>::new(HEADER_SIZE)
+ .transmute_header(|_, h: &mut GenericHeader| {
+ h.command = command;
+ h.size = HEADER_SIZE as u32;
+ })
+ .into_frozen()
+ }
+
+ fn self_signed() -> (CertificateDer<'static>, PrivateKeyDer<'static>) {
+ let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).expect("rcgen");
+ let cert_der = CertificateDer::from(cert.cert);
+ let key_der: PrivateKeyDer<'static> =
+ PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()).into();
+ (cert_der, key_der)
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn server_endpoint(
+ cert: CertificateDer<'static>,
+ key: PrivateKeyDer<'static>,
+ ) -> Endpoint {
+ let server_cfg = server_config_with_cert(vec![cert], key).expect("server config");
+ Endpoint::server("127.0.0.1:0", server_cfg)
+ .await
+ .expect("bind")
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn client_endpoint(server_cert: CertificateDer<'static>) -> Endpoint {
+ let mut builder = ClientBuilder::new_with_empty_roots()
+ .with_custom_certificate(server_cert)
+ .expect("trust cert")
+ .with_no_crls();
+ builder = builder.with_alpn_protocols(&["iggy.consensus.v1"]);
+ builder.bind("127.0.0.1:0").await.expect("client bind")
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn loopback_round_trip_three_frames() {
+ install_crypto_provider();
+ let (cert, key) = self_signed();
+ let server = server_endpoint(cert.clone(), key).await;
+ let server_addr = server.local_addr().unwrap();
+
+ let listener = QuicTransportListener::new(server);
+ let server_task = compio::runtime::spawn(async move {
+ let (conn, _peer) = listener.accept().await.expect("accept");
+ let (mut reader, _writer) = conn.into_split();
+ let a = reader
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ let b = reader
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ let c = reader
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ (a.header().command, b.header().command, c.header().command)
+ });
+
+ let client = client_endpoint(cert).await;
+ let connecting = client
+ .connect(server_addr, "localhost", None)
+ .expect("connect");
+ let connection = connecting.await.expect("client handshake");
+ let (send, recv) = connection.open_bi_wait().await.expect("open_bi");
+ let conn = QuicTransportConn::new(connection, (send, recv));
+ let (_r, mut w) = conn.into_split();
+
+ let mut batch = vec![
+ header_only(Command2::Ping),
+ header_only(Command2::Prepare),
+ header_only(Command2::Request),
+ ];
+ w.send_batch(&mut batch).await.expect("send_batch");
+ assert!(batch.is_empty(), "Vec must be drained on success");
+
+ let (a, b, c) = server_task.await.unwrap();
+ assert_eq!(a, Command2::Ping);
+ assert_eq!(b, Command2::Prepare);
+ assert_eq!(c, Command2::Request);
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn read_message_reports_oversize_via_trait() {
+ install_crypto_provider();
+ let (cert, key) = self_signed();
+ let server = server_endpoint(cert.clone(), key).await;
+ let server_addr = server.local_addr().unwrap();
+
+ let listener = QuicTransportListener::new(server);
+ let server_task = compio::runtime::spawn(async move {
+ let (conn, _peer) = listener.accept().await.expect("accept");
+ let (mut reader, _writer) = conn.into_split();
+ reader.read_message(framing::MAX_MESSAGE_SIZE).await
+ });
+
+ let client = client_endpoint(cert).await;
+ let connecting = client
+ .connect(server_addr, "localhost", None)
+ .expect("connect");
+ let connection = connecting.await.expect("handshake");
+ let (mut send, recv) = connection.open_bi_wait().await.expect("open_bi");
+ // Bogus oversize size field at offset 48.
+ let mut buf = vec![0u8; HEADER_SIZE];
+ let bogus = u32::try_from(framing::MAX_MESSAGE_SIZE + 1)
+ .unwrap_or(u32::MAX)
+ .to_le_bytes();
+ buf[48..52].copy_from_slice(&bogus);
+ let BufResult(result, _) = send.write(buf).await;
+ result.expect("write");
+ send.flush().await.expect("flush");
+ drop(send); // signal end-of-stream so reader sees the framed bytes
+ drop(recv);
+
+ let res = server_task.await.unwrap();
+ assert!(matches!(res, Err(IggyError::InvalidCommand)));
+ }
+
+ #[test]
+ fn default_transport_config_caps_streams_at_one() {
+ // Lightweight smoke: confirm the helper builds without panicking
+ // and the underlying defaults the audit named are reachable.
+ let _cfg = default_transport_config();
+ }
+}
diff --git a/core/server-ng/src/session_manager.rs b/core/server-ng/src/session_manager.rs
index b89ae4a00..6fead47a4 100644
--- a/core/server-ng/src/session_manager.rs
+++ b/core/server-ng/src/session_manager.rs
@@ -23,7 +23,7 @@
//! Each connection goes through: `connect → login → register → bound`.
//!
//! The [`SessionManager`] is the server-side counterpart of the SDK's
-//! session lifecycle. It does **not** own the [`ClientTable`]. That lives
+//! session lifecycle. It does **not** own the `ClientTable`. That lives
//! in the consensus layer. This module tracks the binding between a
//! transport connection and the consensus-level `(client_id, session)` pair.
(iggy) 07/14: docs(message_bus): name the Transport trait surface in lib.rs (P1-T5)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit eb9f01b5bb20beb67ebb5fff5292fb79674d07c1
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 12:11:22 2026 +0200
docs(message_bus): name the Transport trait surface in lib.rs (P1-T5)
Phase 1 closes out by pointing the crate-level rustdoc at the
`transports` module so readers landing on the crate doc without
context learn that TCP, WS, and QUIC all live behind a common
trait family (`TransportListener`, `TransportConn`, `TransportReader`,
`TransportWriter`). Names the TCP impl at `transports::tcp`, the
generic install entries `install_replica_conn` /
`install_client_conn`, and the generic drain loop
`writer_task::run_transport`.
No behavior change; docs only. `cargo doc --no-deps` green (the
pre-existing `MessageBusConfig::default` intra-doc warning in
`connector.rs` is untouched).
---
core/message_bus/src/lib.rs | 11 +++++++++++
1 file changed, 11 insertions(+)
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index daf52fd83..9d95cc63e 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -56,6 +56,17 @@
//! - 0-RTT stays disabled by default on any future QUIC path. Per-
//! command opt-in requires a checked-in idempotence audit.
//!
+//! # Transport abstraction
+//!
+//! [`transports`] defines the trait surface every wire plane sits
+//! behind: [`transports::TransportListener`],
+//! [`transports::TransportConn`], [`transports::TransportReader`],
+//! [`transports::TransportWriter`]. TCP is the only production impl
+//! (see [`transports::tcp`]); [`installer::install_replica_conn`] /
+//! [`installer::install_client_conn`] and
+//! [`writer_task::run_transport`] are generic over it so WS / QUIC
+//! plug in behind the same registry, fencing, and batching logic.
+//!
//! The full invariant list and the transport-plan design notes live
//! under `Documents/silverhand/iggy/message_bus/transport-plan/`.
(iggy) 06/14: feat(message_bus): generic installer over TransportConn (IGGY-112, P1-T3)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 0aeea4551c0ce4574757403775abaea47b8944b3
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 12:08:26 2026 +0200
feat(message_bus): generic installer over TransportConn (IGGY-112, P1-T3)
Closes out the installer side of Phase 1: `install_replica_stream` and
`install_client_stream` were `TcpStream`-hardcoded, which forced every
future transport (WS, QUIC) to duplicate the registry insert, the
`InstanceToken` fencing, the `install_aborted` race handling, and the
`conn_shutdown` reader-wake. Those invariants are load-bearing and
must not fork.
Split each install path into two entries:
- TCP entry keeps the existing `TcpStream` signature and is where
the per-connection socket options (keepalive, `TCP_NODELAY`)
live. These are TCP-specific by construction; other transports
terminate above the raw fd layer.
- Generic entry `install_replica_conn<C: TransportConn>` /
`install_client_conn<C: TransportConn>` carries the meat of the
install: split into reader/writer halves via the trait, spawn
the writer task on `writer_task::run_transport`, and run the
reader loop through `TransportReader::read_message`.
The read loops become `<R: TransportReader>`; the inline
`framing::read_message` call turns into a trait method call without
any behavioural change.
`install_replica_fd` / `install_client_fd` continue to call the TCP
entry (fd-delegation stays TCP-only per I5); existing test callers
(`tests/common/mod.rs`, `tests/duplicate_client_id.rs`) keep the
`TcpStream` signature unchanged.
All 100 tests pass; `cargo clippy -p message_bus --all-features
--all-targets -- -D warnings` clean; workspace builds.
---
core/message_bus/src/installer.rs | 90 +++++++++++++++++++++++++++------------
1 file changed, 63 insertions(+), 27 deletions(-)
diff --git a/core/message_bus/src/installer.rs b/core/message_bus/src/installer.rs
index 3ab6d615a..21d381296 100644
--- a/core/message_bus/src/installer.rs
+++ b/core/message_bus/src/installer.rs
@@ -26,12 +26,12 @@
use crate::client_listener::RequestHandler;
use crate::fd_transfer::{self, DupedFd};
-use crate::framing;
use crate::lifecycle::{InstanceToken, RejectedRegistration, Shutdown};
use crate::replica_listener::MessageHandler;
use crate::socket_opts::{apply_keepalive_for_connection, apply_nodelay_for_connection};
+use crate::transports::{TcpTransportConn, TransportConn, TransportReader};
use crate::{IggyMessageBus, lifecycle::ShutdownToken};
-use compio::net::{OwnedReadHalf, TcpStream};
+use compio::net::TcpStream;
use futures::FutureExt;
use iggy_binary_protocol::Command2;
use std::cell::Cell;
@@ -89,25 +89,22 @@ impl ConnectionInstaller for Rc<IggyMessageBus> {
}
}
-/// Install a pre-wrapped replica TCP stream on the bus. Shared by
-/// `install_replica_fd` and, once shard 0 acts as the delegate target for
-/// itself, the accept path.
-#[allow(clippy::future_not_send, clippy::too_many_lines)]
+/// TCP entry point: apply socket options (keepalive, `TCP_NODELAY`) on
+/// the raw stream and delegate to the transport-generic install path.
+///
+/// Socket options live here (not on the generic path) because they are
+/// TCP-specific; other transports (WSS via TLS terminator, QUIC) lack a
+/// raw-fd level at this layer. Kept as a separate function so the
+/// `install_replica_fd` fd-delegation entry and the accept callbacks
+/// (`AcceptedReplicaFn` in tests and shard-0 coordinator) converge on
+/// one place for those options.
+#[allow(clippy::future_not_send)]
pub fn install_replica_stream(
bus: &Rc<IggyMessageBus>,
peer_id: u8,
stream: TcpStream,
on_message: MessageHandler,
) {
- if bus.replicas().contains(peer_id) {
- debug!(
- replica = peer_id,
- "replica already registered on this shard, dropping delegated fd"
- );
- drop(stream);
- return;
- }
-
let cfg = bus.config();
if let Err(e) = apply_keepalive_for_connection(
&stream,
@@ -123,8 +120,34 @@ pub fn install_replica_stream(
// A miss means we stay Nagle-on for this peer, not a failure.
warn!(replica = peer_id, "nodelay failed on delegated fd: {e}");
}
+ install_replica_conn(bus, peer_id, TcpTransportConn::new(stream), on_message);
+}
+
+/// Install a pre-wrapped replica connection on the bus.
+///
+/// Generic over [`TransportConn`] so alternate transports (WS via
+/// shard-0 TLS terminator, QUIC via `compio-quic`) plug in behind the
+/// same registry-insert + instance-token fencing + install-race
+/// handling. TCP-specific socket options live in
+/// [`install_replica_stream`]; transports with no equivalent layer call
+/// this entry directly with their already-configured connection.
+#[allow(clippy::future_not_send, clippy::too_many_lines)]
+pub fn install_replica_conn<C: TransportConn>(
+ bus: &Rc<IggyMessageBus>,
+ peer_id: u8,
+ conn: C,
+ on_message: MessageHandler,
+) {
+ if bus.replicas().contains(peer_id) {
+ debug!(
+ replica = peer_id,
+ "replica already registered on this shard, dropping delegated fd"
+ );
+ drop(conn);
+ return;
+ }
- let (read_half, write_half) = stream.into_split();
+ let (read_half, write_half) = conn.into_split();
let (tx, rx) = async_channel::bounded(bus.peer_queue_capacity());
// Writer and reader both observe abnormal close and used to fire
@@ -168,7 +191,7 @@ pub fn install_replica_stream(
let token_for_writer = Rc::clone(&install_token);
let max_batch = bus.config().max_batch;
let writer_handle = compio::runtime::spawn(async move {
- crate::writer_task::run(
+ crate::writer_task::run_transport(
rx,
write_half,
writer_token,
@@ -262,8 +285,10 @@ pub fn install_replica_stream(
}
}
-/// Install a pre-wrapped client TCP stream on the bus.
-#[allow(clippy::future_not_send, clippy::too_many_lines)]
+/// TCP entry point for client installs. Applies socket options and
+/// delegates to [`install_client_conn`]. See [`install_replica_stream`]
+/// for the plane-symmetric docs.
+#[allow(clippy::future_not_send)]
pub fn install_client_stream(
bus: &Rc<IggyMessageBus>,
client_id: u128,
@@ -288,8 +313,19 @@ pub fn install_client_stream(
"nodelay failed on delegated client fd: {e}"
);
}
+ install_client_conn(bus, client_id, TcpTransportConn::new(stream), on_request);
+}
- let (read_half, write_half) = stream.into_split();
+/// Install a pre-wrapped client connection on the bus. Generic over
+/// [`TransportConn`]; plane-symmetric with [`install_replica_conn`].
+#[allow(clippy::future_not_send, clippy::too_many_lines)]
+pub fn install_client_conn<C: TransportConn>(
+ bus: &Rc<IggyMessageBus>,
+ client_id: u128,
+ conn: C,
+ on_request: RequestHandler,
+) {
+ let (read_half, write_half) = conn.into_split();
let (tx, rx) = async_channel::bounded(bus.peer_queue_capacity());
// If the registry insert below loses a race for `client_id`, the
@@ -313,7 +349,7 @@ pub fn install_client_stream(
let token_for_writer = Rc::clone(&install_token);
let max_batch = bus.config().max_batch;
let writer_handle = compio::runtime::spawn(async move {
- crate::writer_task::run(
+ crate::writer_task::run_transport(
rx,
write_half,
writer_token,
@@ -445,9 +481,9 @@ async fn drain_rejected_registration(rejected: RejectedRegistration, timeout: Du
/// the insert-race path: it wakes the reader off its `io_uring` read SQE
/// immediately rather than waiting for peer EOF.
#[allow(clippy::future_not_send)]
-async fn replica_read_loop(
+async fn replica_read_loop<R: TransportReader>(
replica_id: u8,
- mut read_half: OwnedReadHalf<TcpStream>,
+ mut read_half: R,
on_message: &MessageHandler,
token: &ShutdownToken,
conn_token: &ShutdownToken,
@@ -464,7 +500,7 @@ async fn replica_read_loop(
debug!(replica = replica_id, "replica read loop aborted by per-connection shutdown");
return;
}
- result = framing::read_message(&mut read_half, max_message_size).fuse() => {
+ result = read_half.read_message(max_message_size).fuse() => {
match result {
Ok(msg) => {
if aborted.get() {
@@ -495,9 +531,9 @@ async fn replica_read_loop(
/// the insert-race path: it wakes the reader off its `io_uring` read SQE
/// immediately rather than waiting for peer EOF.
#[allow(clippy::future_not_send)]
-async fn client_read_loop(
+async fn client_read_loop<R: TransportReader>(
client_id: u128,
- mut read_half: OwnedReadHalf<TcpStream>,
+ mut read_half: R,
on_request: &RequestHandler,
token: &ShutdownToken,
conn_token: &ShutdownToken,
@@ -514,7 +550,7 @@ async fn client_read_loop(
debug!(client = client_id, "client read loop aborted by per-connection shutdown");
return;
}
- result = framing::read_message(&mut read_half, max_message_size).fuse() => {
+ result = read_half.read_message(max_message_size).fuse() => {
match result {
Ok(msg) => {
if aborted.get() {
(iggy) 13/14: feat(message_bus): add WebSocket transport for SDK-client plane
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 963ccf851a4adcf926676f012e45f4f1b2df88cf
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 11:37:20 2026 +0200
feat(message_bus): add WebSocket transport for SDK-client plane
The client plane lacked WS support; Phase 2 of the transport plan
delivers it for browser SDKs and operators that need WS behind a
load-balancer-terminated TLS terminator. Lands the message-bus
side behind the existing `Transport*` trait family.
`WsTransport{Listener,Conn,Reader,Writer}` in `transports/ws.rs`
wrap `compio-ws 0.3.1` so the runtime stays single-stack (no tokio
bridge, invariant I10).
`WebSocketStream` cannot be split: tungstenite's protocol state
machine spans both directions. The transport's `into_split` spawns
a per-connection dispatcher task that owns the stream and bridges
two `async_channel::bounded` queues to the Reader/Writer halves.
A `select!`-based dispatcher hit compio's "no buffer drop with
pending I/O" panic on cancelled reads. The dispatcher is therefore
phase-alternating: drain up to 16 outbound frames non-blocking via
`try_recv`, then await one `stream.read()`. For request/reply
workloads outbound latency is bounded by the next inbound RTT;
server-push-heavy traffic gets a follow-up keepalive sweeper.
Outbound stays zero-copy: `bytes::Bytes::from_owner` wraps a
`Frozen<MESSAGE_ALIGN>` newtype that impls `AsRef<[u8]>` (the
underlying Frozen has `Deref` only). Inbound copies once into a
`MESSAGE_ALIGN`-aligned `Owned` because `Bytes` carries no
alignment guarantee.
WS is always-on (no Cargo feature), matching the QUIC posture and
shrinking the build matrix. `compio-ws + tungstenite 0.28 + bytes`
move to message_bus's regular `[dependencies]`.
`tungstenite = "0.28"` is pinned directly (not via the workspace's
0.29 used by tokio-tungstenite elsewhere) because compio-ws does
not re-export `Message` and we need to use the same enum compio-ws
expects.
Out of scope on this branch: shard-0 HTTP-Upgrade routing in
`client_listener` and the `install_client_ws_stream` entry point.
Both ride the dispatcher branch alongside QUIC's
`ShardFramePayload::ClientQuicConnectionSetup` variant.
Tests: +2 in `transports::ws::tests` (round-trip three frames,
malformed-frame closes connection). 125/125 pass on `cargo nextest
run -p message_bus -p server-ng`. Workspace clippy + cargo doc
clean.
---
Cargo.lock | 3 +
Cargo.toml | 1 +
core/message_bus/Cargo.toml | 5 +
core/message_bus/src/transports/mod.rs | 1 +
core/message_bus/src/transports/ws.rs | 448 +++++++++++++++++++++++++++++++++
5 files changed, 458 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index cb1463495..36f299edd 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7010,8 +7010,10 @@ version = "0.1.0"
dependencies = [
"async-channel",
"blake3",
+ "bytes",
"compio",
"compio-quic",
+ "compio-ws",
"futures",
"iggy_binary_protocol",
"iggy_common",
@@ -7023,6 +7025,7 @@ dependencies = [
"tempfile",
"thiserror 2.0.18",
"tracing",
+ "tungstenite 0.28.0",
]
[[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 87d8878d1..aa367d9a5 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -119,6 +119,7 @@ compio-buf = "0.8.1"
# Pin compio-driver >= 0.11.2 to fix musl compilation (compio-rs/compio#668)
compio-driver = "0.11.4"
compio-quic = "=0.7.2"
+compio-ws = "=0.3.1"
configs = { path = "core/configs", version = "0.1.0" }
configs_derive = { path = "core/configs_derive", version = "0.1.0" }
consensus = { path = "core/consensus" }
diff --git a/core/message_bus/Cargo.toml b/core/message_bus/Cargo.toml
index ace090f16..2be7d01a9 100644
--- a/core/message_bus/Cargo.toml
+++ b/core/message_bus/Cargo.toml
@@ -31,8 +31,10 @@ publish = false
[dependencies]
async-channel = { workspace = true }
blake3 = { workspace = true }
+bytes = { workspace = true }
compio = { workspace = true }
compio-quic = { workspace = true }
+compio-ws = { workspace = true }
futures = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
@@ -42,6 +44,9 @@ rustls = { workspace = true }
socket2 = { workspace = true, features = ["all"] }
thiserror = { workspace = true }
tracing = { workspace = true }
+# Pinned to the major compio-ws 0.3.1 wraps (it does not re-export Message),
+# so our send/receive sites use the same Message enum compio-ws expects.
+tungstenite = "=0.28.0"
[dev-dependencies]
rcgen = { workspace = true }
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index 76e1965f3..2e1875919 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -67,6 +67,7 @@
pub mod quic;
pub mod tcp;
+pub mod ws;
// Only `Conn` and `Writer` have crate-internal callers today
// (`installer.rs` wraps the dialed/accepted stream in a `TcpTransportConn`;
diff --git a/core/message_bus/src/transports/ws.rs b/core/message_bus/src/transports/ws.rs
new file mode 100644
index 000000000..9a52a40de
--- /dev/null
+++ b/core/message_bus/src/transports/ws.rs
@@ -0,0 +1,448 @@
+// 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.
+
+//! WebSocket impls of the [`super`] transport traits.
+//!
+//! SDK-client plane only; replica plane stays TCP forever (invariant I1).
+//! `compio-ws 0.3.1` binds `tungstenite`'s protocol state machine to
+//! compio I/O directly, no tokio bridge (invariant I10).
+//!
+//! # Connection model
+//!
+//! One binary WS frame per consensus message. The 256 B `GenericHeader`
+//! plus the optional body is the frame payload; the frame format
+//! constraints (FIN=1, opcode=0x2 binary, no fragmentation, no
+//! extensions) come from the WS spec layered on top of our 256 B
+//! framing invariant (I3).
+//!
+//! # Stream split via actor pattern
+//!
+//! `tungstenite`'s state machine spans both directions: control frames
+//! (Ping/Pong/Close) and the masked-from-client/unmasked-from-server
+//! rules force the read and write halves to cooperate on one
+//! `WebSocketStream` instance. We cannot satisfy
+//! [`TransportConn::into_split`]'s ownership-split contract by handing
+//! out two halves of the underlying TCP connection.
+//!
+//! Instead [`WsTransportConn::into_split`] spawns a **dispatcher task**
+//! that owns the [`WebSocketStream`] and bridges two
+//! `async_channel::bounded` queues:
+//!
+//! - inbound: dispatcher reads frames, pushes parsed
+//! `Message<GenericHeader>` to the reader's queue.
+//! - outbound: dispatcher pulls `Frozen<MESSAGE_ALIGN>` from the
+//! writer's queue, wraps in a binary WS frame, sends.
+//!
+//! Per-frame channel hop costs one queue push + one queue pop. Single-
+//! threaded compio means the hop is a noop scheduling-wise; the cost is
+//! one bounded buffer slot per inflight frame, which the per-peer queue
+//! capacity already pays for on the bus side.
+//!
+//! # Zero-copy payload
+//!
+//! `tungstenite::Message::Binary` takes `bytes::Bytes`. Wrapping
+//! `Frozen<MESSAGE_ALIGN>` via [`bytes::Bytes::from_owner`] is
+//! O(1) Arc-allocation, no memcpy. The 256 B header + body slice is
+//! handed to the WS frame writer as-is.
+
+use super::{TransportConn, TransportListener, TransportReader, TransportWriter};
+use crate::framing;
+use async_channel::{Receiver, Sender, bounded};
+use bytes::Bytes;
+use compio::net::{TcpListener, TcpStream};
+use compio_ws::WebSocketStream;
+use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
+use iggy_binary_protocol::consensus::iobuf::Frozen;
+use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_common::IggyError;
+use std::io;
+use std::net::SocketAddr;
+use tracing::{debug, warn};
+use tungstenite::protocol::CloseFrame;
+use tungstenite::protocol::frame::coding::CloseCode;
+
+/// Newtype wrapper bridging [`Frozen<MESSAGE_ALIGN>`] into the
+/// [`bytes::Bytes::from_owner`] surface (which requires `AsRef<[u8]>`,
+/// not `Deref`). The underlying byte slice is shared with the
+/// `Bytes`'s atomic-refcount holder; no memcpy. Cost is one
+/// `Arc<dyn>` allocation per outbound frame.
+struct FrozenBytesOwner(Frozen<MESSAGE_ALIGN>);
+
+impl AsRef<[u8]> for FrozenBytesOwner {
+ fn as_ref(&self) -> &[u8] {
+ self.0.as_slice()
+ }
+}
+
+/// Subprotocol identifier carried in `Sec-WebSocket-Protocol`. Bump the
+/// trailing version digits when the wire format changes (matches the
+/// QUIC ALPN constant in `transports/quic.rs`).
+pub const WS_SUBPROTOCOL: &str = "iggy.consensus.v1";
+
+/// Per-connection inbound queue depth. Bounded so an idle reader cannot
+/// let memory grow unboundedly under a chatty peer; mirrors the bus's
+/// per-peer outbound queue convention.
+const INBOUND_QUEUE_CAPACITY: usize = 256;
+
+/// Per-connection outbound queue depth. Same shape as the inbound side;
+/// the per-peer writer task already enforces backpressure upstream.
+const OUTBOUND_QUEUE_CAPACITY: usize = 256;
+
+/// Inbound WS listener.
+///
+/// Wraps a [`TcpListener`]. `accept` performs the WS HTTP-Upgrade
+/// handshake via `compio_ws::accept_async` before yielding a
+/// [`WsTransportConn`], so the install path on the owning shard never
+/// sees the upgrade request.
+pub struct WsTransportListener {
+ inner: TcpListener,
+}
+
+impl WsTransportListener {
+ /// Wrap a pre-bound [`TcpListener`].
+ #[must_use]
+ pub const fn new(inner: TcpListener) -> Self {
+ Self { inner }
+ }
+}
+
+impl TransportListener for WsTransportListener {
+ type Conn = WsTransportConn;
+
+ #[allow(clippy::future_not_send)]
+ async fn accept(&self) -> io::Result<(Self::Conn, SocketAddr)> {
+ loop {
+ let (stream, addr) = self.inner.accept().await?;
+ match compio_ws::accept_async(stream).await {
+ Ok(ws) => return Ok((WsTransportConn::new(ws), addr)),
+ Err(e) => {
+ warn!(%addr, "WS handshake failed: {e}");
+ // Drop the failed stream and accept the next; do
+ // not surface as a listener-fatal error.
+ }
+ }
+ }
+ }
+}
+
+/// A single WS connection holding the bidirectional [`WebSocketStream`]
+/// until [`Self::into_split`] hands ownership to a per-connection
+/// dispatcher task.
+pub struct WsTransportConn {
+ stream: WebSocketStream<TcpStream>,
+}
+
+impl WsTransportConn {
+ /// Construct from an already-upgraded [`WebSocketStream`].
+ #[must_use]
+ pub const fn new(stream: WebSocketStream<TcpStream>) -> Self {
+ Self { stream }
+ }
+}
+
+impl TransportConn for WsTransportConn {
+ type Reader = WsTransportReader;
+ type Writer = WsTransportWriter;
+
+ fn into_split(self) -> (Self::Reader, Self::Writer) {
+ let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(INBOUND_QUEUE_CAPACITY);
+ let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(OUTBOUND_QUEUE_CAPACITY);
+ compio::runtime::spawn(dispatcher(self.stream, in_tx, out_rx)).detach();
+ (
+ WsTransportReader { rx: in_rx },
+ WsTransportWriter { tx: out_tx },
+ )
+ }
+}
+
+/// Per-connection dispatcher loop. Phase-alternating design.
+///
+/// `tungstenite`'s state machine and compio's "no buffer drop with
+/// pending I/O" rule together rule out a `select!`-cancelling read +
+/// write race on a shared `&mut WebSocketStream`. Instead the loop
+/// alternates phases:
+///
+/// 1. **Phase A** — non-blocking drain of up to
+/// [`OUTBOUND_DRAIN_BURST`] outbound frames. `try_recv` returns
+/// immediately when the queue is empty; no in-flight I/O is held
+/// across the await.
+/// 2. **Phase B** — one blocking `stream.read()`. Returns when the
+/// peer sends a frame, EOF, error, or peer-close.
+///
+/// Outbound latency under this design = at most one inbound RTT (the
+/// time to receive the next frame). For request/reply workloads
+/// (every client request triggers a server reply) this is identical
+/// to a `select!`-based design. For server-push-heavy workloads
+/// the upper bound is the keepalive interval; future work can add a
+/// keepalive Ping/Pong sweeper that nudges the read past idle.
+const OUTBOUND_DRAIN_BURST: usize = 16;
+
+#[allow(clippy::future_not_send)]
+async fn dispatcher(
+ mut stream: WebSocketStream<TcpStream>,
+ in_tx: Sender<Message<GenericHeader>>,
+ out_rx: Receiver<Frozen<MESSAGE_ALIGN>>,
+) {
+ loop {
+ // Phase A: drain outbound non-blocking, capped at the burst.
+ for _ in 0..OUTBOUND_DRAIN_BURST {
+ let Ok(frame) = out_rx.try_recv() else {
+ break;
+ };
+ let payload = Bytes::from_owner(FrozenBytesOwner(frame));
+ if let Err(e) = stream.send(tungstenite::Message::Binary(payload)).await {
+ warn!("WS write failed: {e}; dispatcher exiting");
+ let _ = stream.close(None).await;
+ return;
+ }
+ }
+
+ // Phase B: one read. The compio I/O submission is owned by
+ // the await and only dropped after completion - matches
+ // compio's "no buffer drop with pending I/O" rule.
+ let ws_msg = stream.read().await;
+ match ws_msg {
+ Ok(tungstenite::Message::Binary(body)) => match decode_consensus_frame(&body) {
+ Ok(msg) => {
+ if in_tx.send(msg).await.is_err() {
+ debug!("WS reader dropped; dispatcher exiting");
+ break;
+ }
+ }
+ Err(e) => {
+ warn!("WS frame decode failed: {e:?}; closing connection");
+ let _ = stream
+ .close(Some(CloseFrame {
+ code: CloseCode::Protocol,
+ reason: "iggy frame decode failed".into(),
+ }))
+ .await;
+ break;
+ }
+ },
+ Ok(tungstenite::Message::Ping(payload)) => {
+ if let Err(e) = stream.send(tungstenite::Message::Pong(payload)).await {
+ debug!("WS pong write failed: {e}; dispatcher exiting");
+ break;
+ }
+ }
+ Ok(tungstenite::Message::Pong(_)) => {
+ // Server-initiated keepalive replies; nothing to do.
+ }
+ Ok(tungstenite::Message::Close(_)) => {
+ debug!("WS peer initiated close");
+ break;
+ }
+ Ok(tungstenite::Message::Text(_)) => {
+ warn!("WS text frame received on a binary plane; closing");
+ let _ = stream
+ .close(Some(CloseFrame {
+ code: CloseCode::Unsupported,
+ reason: "binary frames only".into(),
+ }))
+ .await;
+ break;
+ }
+ Ok(tungstenite::Message::Frame(_)) => {
+ // Tungstenite's contract: `Message::Frame` is never
+ // produced on the read path. Defensive log + drop.
+ warn!("unexpected raw WS frame on read path; ignoring");
+ }
+ Err(e) => {
+ debug!("WS read error: {e}; dispatcher exiting");
+ break;
+ }
+ }
+ }
+ let _ = stream.close(None).await;
+}
+
+#[derive(Debug)]
+enum FrameDecodeError {
+ BadHeader,
+ BadSize,
+}
+
+/// Parse a binary WS frame's payload into a `Message<GenericHeader>`.
+///
+/// Reuses the framing invariant (I3): `[256 B header][optional body]`
+/// where `body_len = header.size - HEADER_SIZE`. No partial frames; the
+/// WS layer guarantees a single message per binary frame (FIN=1).
+fn decode_consensus_frame(body: &Bytes) -> Result<Message<GenericHeader>, FrameDecodeError> {
+ if body.len() < iggy_binary_protocol::HEADER_SIZE {
+ return Err(FrameDecodeError::BadHeader);
+ }
+ let total_size = u32::from_le_bytes(
+ body[48..52]
+ .try_into()
+ .map_err(|_| FrameDecodeError::BadHeader)?,
+ ) as usize;
+ if !(iggy_binary_protocol::HEADER_SIZE..=framing::MAX_MESSAGE_SIZE).contains(&total_size)
+ || total_size != body.len()
+ {
+ return Err(FrameDecodeError::BadSize);
+ }
+ // Aligned MESSAGE_ALIGN backing memory is required by the
+ // `Message<GenericHeader>` invariant; `Bytes` carries no alignment
+ // guarantee, so this is a one-time copy from the WS-framed payload
+ // into a freshly allocated `Owned<MESSAGE_ALIGN>`. The cost is
+ // unavoidable on inbound for the WS plane (TCP path also does a
+ // copy via the staged read-into-Owned dance in framing.rs); the
+ // outbound side stays zero-copy via `Bytes::from_owner`.
+ let owned =
+ iggy_binary_protocol::consensus::iobuf::Owned::<MESSAGE_ALIGN>::copy_from_slice(body);
+ Message::<GenericHeader>::try_from(owned).map_err(|_| FrameDecodeError::BadHeader)
+}
+
+/// Read half. Polls the dispatcher's inbound queue.
+pub struct WsTransportReader {
+ rx: Receiver<Message<GenericHeader>>,
+}
+
+impl TransportReader for WsTransportReader {
+ #[allow(clippy::future_not_send)]
+ async fn read_message(
+ &mut self,
+ _max_message_size: usize,
+ ) -> Result<Message<GenericHeader>, IggyError> {
+ // Bound check happens inside `decode_consensus_frame` against
+ // the build-time `framing::MAX_MESSAGE_SIZE`; the per-call
+ // override exists for parity with the trait surface and will
+ // be plumbed through when a runtime-configurable cap lands.
+ self.rx
+ .recv()
+ .await
+ .map_err(|_| IggyError::ConnectionClosed)
+ }
+}
+
+/// Write half. Pushes frames into the dispatcher's outbound queue.
+pub struct WsTransportWriter {
+ tx: Sender<Frozen<MESSAGE_ALIGN>>,
+}
+
+impl TransportWriter for WsTransportWriter {
+ #[allow(clippy::future_not_send)]
+ async fn send_batch(&mut self, batch: &mut Vec<Frozen<MESSAGE_ALIGN>>) -> io::Result<()> {
+ // Drain in order; one queue push per frame. Atomic-or-error
+ // contract holds because the dispatcher writes each frame
+ // before pulling the next from the queue.
+ for buf in batch.drain(..) {
+ self.tx
+ .send(buf)
+ .await
+ .map_err(|_| io::Error::other("WS dispatcher queue closed"))?;
+ }
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use compio::net::TcpStream;
+ use iggy_binary_protocol::{Command2, HEADER_SIZE};
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn header_only(command: Command2) -> Frozen<MESSAGE_ALIGN> {
+ Message::<GenericHeader>::new(HEADER_SIZE)
+ .transmute_header(|_, h: &mut GenericHeader| {
+ h.command = command;
+ h.size = HEADER_SIZE as u32;
+ })
+ .into_frozen()
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn loopback_round_trip_three_frames() {
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let server_addr = listener.local_addr().unwrap();
+ let ws_listener = WsTransportListener::new(listener);
+
+ let server_task = compio::runtime::spawn(async move {
+ let (conn, _peer) = ws_listener.accept().await.expect("accept");
+ let (mut reader, _writer) = conn.into_split();
+ let a = reader
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ let b = reader
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ let c = reader
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ (a.header().command, b.header().command, c.header().command)
+ });
+
+ let client_tcp = TcpStream::connect(server_addr).await.unwrap();
+ let url = format!("ws://{server_addr}/");
+ let (ws_client, _resp) = compio_ws::client_async(url, client_tcp)
+ .await
+ .expect("ws handshake");
+ let conn = WsTransportConn::new(ws_client);
+ let (_r, mut w) = conn.into_split();
+
+ let mut batch = vec![
+ header_only(Command2::Ping),
+ header_only(Command2::Prepare),
+ header_only(Command2::Request),
+ ];
+ w.send_batch(&mut batch).await.expect("send_batch");
+ assert!(batch.is_empty(), "Vec must be drained on success");
+
+ let (a, b, c) = server_task.await.unwrap();
+ assert_eq!(a, Command2::Ping);
+ assert_eq!(b, Command2::Prepare);
+ assert_eq!(c, Command2::Request);
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn malformed_frame_closes_connection() {
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let server_addr = listener.local_addr().unwrap();
+ let ws_listener = WsTransportListener::new(listener);
+
+ let server_task = compio::runtime::spawn(async move {
+ let (conn, _peer) = ws_listener.accept().await.expect("accept");
+ let (mut reader, _writer) = conn.into_split();
+ reader.read_message(framing::MAX_MESSAGE_SIZE).await
+ });
+
+ let client_tcp = TcpStream::connect(server_addr).await.unwrap();
+ let url = format!("ws://{server_addr}/");
+ let (mut ws_client, _) = compio_ws::client_async(url, client_tcp).await.unwrap();
+ // Deliberately wrong total_size at offset 48 of the 256 B header.
+ let mut buf = vec![0u8; HEADER_SIZE];
+ let bogus = u32::try_from(framing::MAX_MESSAGE_SIZE + 1)
+ .unwrap_or(u32::MAX)
+ .to_le_bytes();
+ buf[48..52].copy_from_slice(&bogus);
+ ws_client
+ .send(tungstenite::Message::Binary(Bytes::from(buf)))
+ .await
+ .unwrap();
+ // Server's dispatcher closes the WS; reader queue closes,
+ // surfacing as ConnectionClosed.
+ let res = server_task.await.unwrap();
+ assert!(matches!(res, Err(IggyError::ConnectionClosed)));
+ }
+}
(iggy) 01/14: feat(message_bus): authenticate the replica Ping handshake (IGGY-112)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit de42708c96dfaf24de4130b91093a43b68e25ac8
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 11:37:20 2026 +0200
feat(message_bus): authenticate the replica Ping handshake (IGGY-112)
Replace the cluster_id-only handshake with a BLAKE3-keyed MAC carried
in GenericHeader.reserved_command[0..57]. Encoding is versioned (1 =
Blake3V1) with 32 B tag, 8 B sender-ns timestamp, 16 B random nonce,
1 B kind, and 71 B zero-checked reserved padding.
The listener verifies the envelope before the directional tiebreak so
unauthenticated peers never drive the nonce ring's eviction cursor.
Replay protection is layered: a 30 s timestamp window uses
IggyTimestamp::now().as_nanos(), and a 256-entry NonceRing catches
in-window replays. Credential failures collapse to
IggyError::Unauthenticated (reusing the existing variant rather than
adding a new one and dragging another From<_, IggyError> candidate
into every downstream crate's orphan-rule set).
auth.rs exposes the TokenSource trait, a StaticSharedSecret impl for
all-replicas-share-one-secret deployments (and the integration-test
harness via StaticSharedSecret::zero()), and pure encode/decode/verify
helpers so transport-specific listeners can wire the envelope into
other planes (client-plane Phase 2+) without re-implementing the
framing. Compio is single-threaded, so TokenSource carries no
Send+Sync bounds and the bus holds it as Rc<dyn TokenSource>.
GenericHeader layout is unchanged: the envelope occupies previously
zero-filled bytes in reserved_command, so invariant I3 (256 B wire
frame, size offset == 48) is preserved.
Rebuilds on top of #3134's async fire-and-forget transport without
altering writer_task batching (I4), zero-copy Frozen ownership (I8),
or the replica-plane FIFO byte-stream model (I1).
BREAKING: a replica upgraded to this commit rejects any peer running
the pre-auth handshake: the zero-filled reserved_command fails BLAKE3
verification. All replicas in a cluster must be upgraded together and
configured with matching cluster secrets; operators without a secret
channel can bootstrap with StaticSharedSecret::zero() to preserve the
old trust-the-network posture, acknowledging that it is no better than
the previous state. Authentication policy (per-client secrets, SSO
callouts, revocation) layers on top via additional TokenSource impls;
this commit only lands the replica-plane static-secret baseline.
Refs: Documents/silverhand/iggy/message_bus/transport-plan/ (P0-T2,
designs/message-bus-auth.md).
---
Cargo.lock | 1 +
core/message_bus/Cargo.toml | 1 +
core/message_bus/src/auth.rs | 450 +++++++++++++++++++++
core/message_bus/src/connector.rs | 53 ++-
core/message_bus/src/lib.rs | 1 +
core/message_bus/src/replica_io.rs | 7 +
core/message_bus/src/replica_listener.rs | 111 +++--
core/message_bus/tests/backpressure.rs | 4 +-
core/message_bus/tests/common/mod.rs | 8 +
core/message_bus/tests/connection_lost_notify.rs | 4 +-
core/message_bus/tests/directional_connection.rs | 16 +-
core/message_bus/tests/handshake_auth.rs | 144 +++++++
core/message_bus/tests/head_of_line.rs | 4 +-
core/message_bus/tests/reconnect.rs | 14 +-
core/message_bus/tests/reconnect_skip_connected.rs | 14 +-
core/message_bus/tests/replica_roundtrip.rs | 4 +-
core/message_bus/tests/shard_zero_gating.rs | 5 +-
core/message_bus/tests/vectored_batch.rs | 4 +-
18 files changed, 794 insertions(+), 51 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index fe7d8cafc..2c66fb79c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7009,6 +7009,7 @@ name = "message_bus"
version = "0.1.0"
dependencies = [
"async-channel",
+ "blake3",
"compio",
"futures",
"iggy_binary_protocol",
diff --git a/core/message_bus/Cargo.toml b/core/message_bus/Cargo.toml
index 4d3569482..d0487db0f 100644
--- a/core/message_bus/Cargo.toml
+++ b/core/message_bus/Cargo.toml
@@ -30,6 +30,7 @@ publish = false
[dependencies]
async-channel = { workspace = true }
+blake3 = { workspace = true }
compio = { workspace = true }
futures = { workspace = true }
iggy_binary_protocol = { workspace = true }
diff --git a/core/message_bus/src/auth.rs b/core/message_bus/src/auth.rs
new file mode 100644
index 000000000..6c2353a6b
--- /dev/null
+++ b/core/message_bus/src/auth.rs
@@ -0,0 +1,450 @@
+// 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.
+
+//! Bus-layer authenticated handshake (IGGY-112).
+//!
+//! The replica-plane handshake (`replica_listener::handshake`) and the
+//! forthcoming SDK-client plane (Phase 2+) share one envelope, carried in
+//! `GenericHeader.reserved_command[0..57]`:
+//!
+//! ```text
+//! [ 0 .. 32) auth_tag BLAKE3 keyed_hash(secret, challenge) full output
+//! [ 32 .. 40) auth_timestamp sender ns since UNIX_EPOCH, little-endian u64
+//! [ 40 .. 56) auth_nonce random u128, little-endian
+//! [ 56 .. 57) auth_kind envelope version tag; 1 = Blake3V1
+//! [ 57 ..128) reserved zero on the wire; zero-checked on receive
+//! ```
+//!
+//! Kept out of `GenericHeader` itself so the 256 B wire layout is
+//! preserved (invariant **I3**). Encode / decode operate on a
+//! `&mut [u8; 128]` / `&[u8; 128]` view of `GenericHeader.reserved_command`.
+
+use iggy_common::IggyError;
+use thiserror::Error;
+
+/// BLAKE3 `keyed_hash` key size, in bytes.
+pub const SECRET_SIZE: usize = 32;
+
+/// Byte range of the envelope within `GenericHeader.reserved_command`.
+pub const ENVELOPE_LEN: usize = 57;
+
+/// Acceptance window around `auth_timestamp`, in nanoseconds. 30 s.
+pub const TIMESTAMP_WINDOW_NS: u128 = 30_000_000_000;
+
+/// Default nonce-dedup LRU capacity on the acceptor.
+pub const NONCE_LRU_CAPACITY: usize = 256;
+
+const LABEL: &[u8; 16] = b"iggy-bus-auth-v1";
+
+const KIND_OFFSET: usize = 56;
+const TAG_RANGE: std::ops::Range<usize> = 0..32;
+const TIMESTAMP_RANGE: std::ops::Range<usize> = 32..40;
+const NONCE_RANGE: std::ops::Range<usize> = 40..56;
+const RESERVED_RANGE: std::ops::Range<usize> = ENVELOPE_LEN..128;
+
+/// Envelope version tag.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[repr(u8)]
+pub enum AuthKind {
+ Blake3V1 = 1,
+}
+
+impl AuthKind {
+ const fn from_byte(b: u8) -> Result<Self, AuthError> {
+ match b {
+ 1 => Ok(Self::Blake3V1),
+ _ => Err(AuthError::UnsupportedKind),
+ }
+ }
+}
+
+/// Fully reconstructed challenge input. All fields come from the header
+/// (`GenericHeader.cluster`, `GenericHeader.replica`, `GenericHeader.release`)
+/// plus the envelope timestamp and nonce.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AuthChallenge {
+ pub cluster: u128,
+ pub peer_id: u128,
+ pub timestamp_ns: u64,
+ pub release: u32,
+ pub nonce: u128,
+}
+
+impl AuthChallenge {
+ /// Serialize the challenge into a fixed-size buffer. Domain-separated by
+ /// a constant label so the same secret cannot be misused to forge a
+ /// different protocol's MAC.
+ fn encode(&self) -> [u8; 16 + 16 + 16 + 8 + 4 + 16] {
+ let mut out = [0u8; 76];
+ out[0..16].copy_from_slice(LABEL);
+ out[16..32].copy_from_slice(&self.cluster.to_le_bytes());
+ out[32..48].copy_from_slice(&self.peer_id.to_le_bytes());
+ out[48..56].copy_from_slice(&self.timestamp_ns.to_le_bytes());
+ out[56..60].copy_from_slice(&self.release.to_le_bytes());
+ out[60..76].copy_from_slice(&self.nonce.to_le_bytes());
+ out
+ }
+}
+
+/// MAC over an `AuthChallenge`. Wraps `blake3::Hash` for its
+/// constant-time `PartialEq` implementation.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct AuthTag(pub blake3::Hash);
+
+impl AuthTag {
+ const fn as_bytes(&self) -> &[u8; 32] {
+ self.0.as_bytes()
+ }
+}
+
+/// Credential-layer errors surfaced by the verify path.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+pub enum AuthError {
+ #[error("timestamp outside acceptance window")]
+ TimestampOutOfWindow,
+ #[error("nonce replay")]
+ NonceReplay,
+ #[error("tag mismatch")]
+ TagMismatch,
+ #[error("unknown peer")]
+ UnknownPeer,
+ #[error("unsupported envelope version")]
+ UnsupportedKind,
+ #[error("reserved bytes nonzero")]
+ ReservedNonzero,
+}
+
+/// Map a credential-layer `AuthError` to the public `IggyError` surface.
+///
+/// Not implemented as `From<AuthError> for IggyError` because the impl
+/// would leak into every crate transitively linking `message_bus`, adding
+/// an extra `impl From<_> for IggyError` candidate to the orphan-rule set
+/// and tripping type inference (`?` ambiguity) in downstream crates.
+#[must_use]
+pub const fn to_iggy_error(e: AuthError) -> IggyError {
+ match e {
+ AuthError::UnsupportedKind | AuthError::ReservedNonzero => IggyError::InvalidCommand,
+ AuthError::TagMismatch
+ | AuthError::UnknownPeer
+ | AuthError::TimestampOutOfWindow
+ | AuthError::NonceReplay => IggyError::Unauthenticated,
+ }
+}
+
+/// Signer + verifier of auth envelopes. The bus runs on a single-threaded
+/// compio runtime, so no `Send + Sync` bounds.
+pub trait TokenSource {
+ fn sign(&self, challenge: &AuthChallenge) -> AuthTag;
+ /// # Errors
+ /// Returns `AuthError::TagMismatch` when `tag` does not match the
+ /// MAC produced for `challenge`; may return other variants in
+ /// token-source impls that perform additional checks.
+ fn verify(&self, challenge: &AuthChallenge, tag: &AuthTag) -> Result<(), AuthError>;
+}
+
+/// `TokenSource` that holds a single 32-byte cluster-wide secret. All
+/// replicas share one secret on the replica plane; the integration-test
+/// harness instantiates this with a fixed zero-byte secret.
+pub struct StaticSharedSecret {
+ secret: [u8; SECRET_SIZE],
+}
+
+impl StaticSharedSecret {
+ #[must_use]
+ pub const fn new(secret: [u8; SECRET_SIZE]) -> Self {
+ Self { secret }
+ }
+
+ #[must_use]
+ pub const fn zero() -> Self {
+ Self::new([0u8; SECRET_SIZE])
+ }
+}
+
+impl TokenSource for StaticSharedSecret {
+ fn sign(&self, challenge: &AuthChallenge) -> AuthTag {
+ let input = challenge.encode();
+ AuthTag(blake3::keyed_hash(&self.secret, &input))
+ }
+
+ fn verify(&self, challenge: &AuthChallenge, tag: &AuthTag) -> Result<(), AuthError> {
+ let expected = self.sign(challenge);
+ if expected.0 == tag.0 {
+ Ok(())
+ } else {
+ Err(AuthError::TagMismatch)
+ }
+ }
+}
+
+/// Fixed-capacity ring-buffer nonce dedup.
+///
+/// Eviction is O(1) via a round-robin cursor; lookup is O(N) over
+/// `NONCE_LRU_CAPACITY` entries, which is a 256-wide linear scan on the
+/// hot path. Memory footprint is deterministic and bounded regardless of
+/// peer behaviour.
+#[derive(Debug)]
+pub struct NonceRing {
+ slots: Vec<u128>,
+ cursor: usize,
+ capacity: usize,
+}
+
+impl NonceRing {
+ #[must_use]
+ pub const fn new(capacity: usize) -> Self {
+ Self {
+ slots: Vec::new(),
+ cursor: 0,
+ capacity,
+ }
+ }
+
+ /// Returns `true` if `nonce` was not present (and has now been
+ /// recorded); `false` if it was already seen.
+ pub fn insert(&mut self, nonce: u128) -> bool {
+ if self.slots.contains(&nonce) {
+ return false;
+ }
+ if self.slots.len() < self.capacity {
+ self.slots.push(nonce);
+ } else {
+ self.slots[self.cursor] = nonce;
+ self.cursor = (self.cursor + 1) % self.capacity;
+ }
+ true
+ }
+}
+
+impl Default for NonceRing {
+ fn default() -> Self {
+ Self::new(NONCE_LRU_CAPACITY)
+ }
+}
+
+/// Encode an auth envelope into `reserved_command`.
+///
+/// Overwrites the first `ENVELOPE_LEN` bytes and zero-fills the trailing
+/// reserved range. Source of the tag is `token_source`; `timestamp_ns`
+/// and `nonce` are caller-controlled so the same routine can be used for
+/// deterministic tests (fake clock, fixed nonce) and production (system
+/// clock, CSPRNG).
+pub fn encode_envelope(
+ reserved: &mut [u8; 128],
+ token_source: &dyn TokenSource,
+ challenge: &AuthChallenge,
+) {
+ let tag = token_source.sign(challenge);
+ reserved[TAG_RANGE].copy_from_slice(tag.as_bytes());
+ reserved[TIMESTAMP_RANGE].copy_from_slice(&challenge.timestamp_ns.to_le_bytes());
+ reserved[NONCE_RANGE].copy_from_slice(&challenge.nonce.to_le_bytes());
+ reserved[KIND_OFFSET] = AuthKind::Blake3V1 as u8;
+ for b in &mut reserved[RESERVED_RANGE] {
+ *b = 0;
+ }
+}
+
+/// Parsed view of an envelope's raw bytes. No cryptographic checks.
+#[derive(Debug, Clone, Copy)]
+pub struct DecodedEnvelope {
+ pub tag: AuthTag,
+ pub timestamp_ns: u64,
+ pub nonce: u128,
+}
+
+/// Decode an envelope and reject malformed layout.
+///
+/// Rejects unknown envelope kinds and nonzero reserved bytes. Does not
+/// check the tag; caller passes the returned `DecodedEnvelope` to
+/// [`verify_envelope`] with the full challenge context.
+///
+/// # Errors
+///
+/// Returns `AuthError::UnsupportedKind` for unknown envelope versions or
+/// `AuthError::ReservedNonzero` for non-zero reserved padding.
+pub fn decode_envelope(reserved: &[u8; 128]) -> Result<DecodedEnvelope, AuthError> {
+ AuthKind::from_byte(reserved[KIND_OFFSET])?;
+ if reserved[RESERVED_RANGE].iter().any(|&b| b != 0) {
+ return Err(AuthError::ReservedNonzero);
+ }
+ let mut tag_bytes = [0u8; 32];
+ tag_bytes.copy_from_slice(&reserved[TAG_RANGE]);
+ let mut ts_bytes = [0u8; 8];
+ ts_bytes.copy_from_slice(&reserved[TIMESTAMP_RANGE]);
+ let mut nonce_bytes = [0u8; 16];
+ nonce_bytes.copy_from_slice(&reserved[NONCE_RANGE]);
+ Ok(DecodedEnvelope {
+ tag: AuthTag(blake3::Hash::from_bytes(tag_bytes)),
+ timestamp_ns: u64::from_le_bytes(ts_bytes),
+ nonce: u128::from_le_bytes(nonce_bytes),
+ })
+}
+
+/// Full credential check: timestamp window -> nonce dedup -> tag.
+///
+/// # Errors
+///
+/// Returns the first `AuthError` triggered by the check sequence.
+pub fn verify_envelope(
+ token_source: &dyn TokenSource,
+ challenge: &AuthChallenge,
+ decoded: &DecodedEnvelope,
+ now_ns: u128,
+ nonces: &mut NonceRing,
+) -> Result<(), AuthError> {
+ let ts_u128 = u128::from(decoded.timestamp_ns);
+ let delta = now_ns.abs_diff(ts_u128);
+ if delta > TIMESTAMP_WINDOW_NS {
+ return Err(AuthError::TimestampOutOfWindow);
+ }
+ if !nonces.insert(decoded.nonce) {
+ return Err(AuthError::NonceReplay);
+ }
+ token_source.verify(challenge, &decoded.tag)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn sample_challenge() -> AuthChallenge {
+ AuthChallenge {
+ cluster: 0xdead_beef_1234_5678_9abc_def0_1122_3344,
+ peer_id: 7,
+ timestamp_ns: 1_700_000_000_000_000_000,
+ release: 1,
+ nonce: 0x11_2233_4455_6677_8899_aabb_ccdd_eeff,
+ }
+ }
+
+ #[test]
+ fn roundtrip_ok() {
+ let source = StaticSharedSecret::new([1u8; SECRET_SIZE]);
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &source, &challenge);
+
+ let decoded = decode_envelope(&reserved).expect("decode");
+ let mut ring = NonceRing::new(32);
+ verify_envelope(
+ &source,
+ &challenge,
+ &decoded,
+ u128::from(challenge.timestamp_ns),
+ &mut ring,
+ )
+ .expect("verify");
+ }
+
+ #[test]
+ fn wrong_secret_rejected() {
+ let signer = StaticSharedSecret::new([1u8; SECRET_SIZE]);
+ let verifier = StaticSharedSecret::new([2u8; SECRET_SIZE]);
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &signer, &challenge);
+
+ let decoded = decode_envelope(&reserved).unwrap();
+ let mut ring = NonceRing::new(32);
+ let err = verify_envelope(
+ &verifier,
+ &challenge,
+ &decoded,
+ u128::from(challenge.timestamp_ns),
+ &mut ring,
+ )
+ .unwrap_err();
+ assert_eq!(err, AuthError::TagMismatch);
+ }
+
+ #[test]
+ fn replay_rejected() {
+ let source = StaticSharedSecret::zero();
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &source, &challenge);
+
+ let decoded = decode_envelope(&reserved).unwrap();
+ let mut ring = NonceRing::new(32);
+ let now = u128::from(challenge.timestamp_ns);
+ verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap();
+ let err = verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap_err();
+ assert_eq!(err, AuthError::NonceReplay);
+ }
+
+ #[test]
+ fn stale_timestamp_rejected() {
+ let source = StaticSharedSecret::zero();
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &source, &challenge);
+
+ let decoded = decode_envelope(&reserved).unwrap();
+ let mut ring = NonceRing::new(32);
+ let now = u128::from(challenge.timestamp_ns) + TIMESTAMP_WINDOW_NS + 1;
+ let err = verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap_err();
+ assert_eq!(err, AuthError::TimestampOutOfWindow);
+ }
+
+ #[test]
+ fn future_timestamp_rejected() {
+ let source = StaticSharedSecret::zero();
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &source, &challenge);
+
+ let decoded = decode_envelope(&reserved).unwrap();
+ let mut ring = NonceRing::new(32);
+ let now = u128::from(challenge.timestamp_ns).saturating_sub(TIMESTAMP_WINDOW_NS + 1);
+ let err = verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap_err();
+ assert_eq!(err, AuthError::TimestampOutOfWindow);
+ }
+
+ #[test]
+ fn unsupported_kind_rejected() {
+ let source = StaticSharedSecret::zero();
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &source, &challenge);
+ reserved[KIND_OFFSET] = 99;
+ let err = decode_envelope(&reserved).unwrap_err();
+ assert_eq!(err, AuthError::UnsupportedKind);
+ }
+
+ #[test]
+ fn reserved_nonzero_rejected() {
+ let source = StaticSharedSecret::zero();
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &source, &challenge);
+ reserved[100] = 0xff;
+ let err = decode_envelope(&reserved).unwrap_err();
+ assert_eq!(err, AuthError::ReservedNonzero);
+ }
+
+ #[test]
+ fn nonce_ring_fills_and_evicts_fifo() {
+ let mut ring = NonceRing::new(3);
+ assert!(ring.insert(1));
+ assert!(ring.insert(2));
+ assert!(ring.insert(3));
+ assert!(!ring.insert(1));
+ assert!(ring.insert(4));
+ // Nonce 1 evicted; reinserting it should now succeed again.
+ assert!(ring.insert(1));
+ }
+}
diff --git a/core/message_bus/src/connector.rs b/core/message_bus/src/connector.rs
index 641d50683..01d18a255 100644
--- a/core/message_bus/src/connector.rs
+++ b/core/message_bus/src/connector.rs
@@ -25,12 +25,15 @@
//! (see `shard::coordinator::ShardZeroCoordinator`).
use crate::IggyMessageBus;
+use crate::auth::{self, TokenSource};
use crate::framing;
use crate::lifecycle::ShutdownToken;
use crate::socket_opts::apply_keepalive_for_connection;
use crate::{AcceptedReplicaFn, GenericHeader, Message};
use compio::net::TcpStream;
use iggy_binary_protocol::{Command2, HEADER_SIZE};
+use iggy_common::IggyTimestamp;
+use rand::Rng;
use std::mem::size_of;
use std::net::SocketAddr;
use std::rc::Rc;
@@ -57,12 +60,14 @@ pub async fn start(
peers: Vec<(u8, SocketAddr)>,
on_dialed: AcceptedReplicaFn,
reconnect_period: Duration,
+ token_source: Rc<dyn TokenSource>,
) {
- connect_all(bus, cluster_id, self_id, &peers, &on_dialed).await;
+ connect_all(bus, cluster_id, self_id, &peers, &on_dialed, &token_source).await;
let handler = on_dialed.clone();
let token = bus.token();
let bus_for_task = Rc::clone(bus);
+ let token_source_for_task = Rc::clone(&token_source);
let handle = compio::runtime::spawn(async move {
periodic_reconnect(
&bus_for_task,
@@ -72,6 +77,7 @@ pub async fn start(
handler,
reconnect_period,
token,
+ token_source_for_task,
)
.await;
});
@@ -85,6 +91,7 @@ async fn connect_all(
self_id: u8,
peers: &[(u8, SocketAddr)],
on_dialed: &AcceptedReplicaFn,
+ token_source: &Rc<dyn TokenSource>,
) {
for &(peer_id, addr) in peers {
if peer_id <= self_id {
@@ -105,11 +112,21 @@ async fn connect_all(
);
continue;
}
- connect_one(bus, cluster_id, self_id, peer_id, addr, on_dialed).await;
+ connect_one(
+ bus,
+ cluster_id,
+ self_id,
+ peer_id,
+ addr,
+ on_dialed,
+ token_source,
+ )
+ .await;
}
}
#[allow(clippy::future_not_send)]
+#[allow(clippy::too_many_arguments)]
async fn periodic_reconnect(
bus: &Rc<IggyMessageBus>,
cluster_id: u128,
@@ -118,9 +135,10 @@ async fn periodic_reconnect(
on_dialed: AcceptedReplicaFn,
period: Duration,
token: ShutdownToken,
+ token_source: Rc<dyn TokenSource>,
) {
while token.sleep_or_shutdown(period).await {
- connect_all(bus, cluster_id, self_id, &peers, &on_dialed).await;
+ connect_all(bus, cluster_id, self_id, &peers, &on_dialed, &token_source).await;
}
debug!("replica reconnect periodic task exiting");
}
@@ -129,6 +147,7 @@ async fn periodic_reconnect(
/// `on_dialed` on success. Dial / handshake failures are logged and
/// swallowed; VSR tolerates missing peers and the periodic sweep retries.
#[allow(clippy::future_not_send)]
+#[allow(clippy::too_many_arguments)]
async fn connect_one(
bus: &Rc<IggyMessageBus>,
cluster_id: u128,
@@ -136,6 +155,7 @@ async fn connect_one(
peer_id: u8,
addr: SocketAddr,
on_dialed: &AcceptedReplicaFn,
+ token_source: &Rc<dyn TokenSource>,
) {
let mut stream = match TcpStream::connect(addr).await {
Ok(s) => s,
@@ -160,7 +180,7 @@ async fn connect_one(
);
}
- let ping = build_ping_message(cluster_id, self_id);
+ let ping = build_ping_message(cluster_id, self_id, token_source.as_ref());
if let Err(e) = framing::write_message(&mut stream, ping).await {
warn!(replica = peer_id, %addr, "handshake write failed: {e}");
return;
@@ -170,8 +190,21 @@ async fn connect_one(
on_dialed(stream, peer_id);
}
-/// Build a Ping handshake message identifying our replica id.
-fn build_ping_message(cluster_id: u128, replica_id: u8) -> Message<GenericHeader> {
+/// Build a Ping handshake message identifying our replica id and carrying
+/// a fresh auth envelope stamped by `token_source`.
+fn build_ping_message(
+ cluster_id: u128,
+ replica_id: u8,
+ token_source: &dyn TokenSource,
+) -> Message<GenericHeader> {
+ #[allow(clippy::cast_possible_truncation)]
+ let timestamp_ns = IggyTimestamp::now().as_nanos() as u64;
+ let nonce = {
+ let mut buf = [0u8; 16];
+ rand::rng().fill_bytes(&mut buf);
+ u128::from_le_bytes(buf)
+ };
+
#[allow(clippy::cast_possible_truncation)]
Message::<GenericHeader>::new(size_of::<GenericHeader>()).transmute_header(
|_, h: &mut GenericHeader| {
@@ -179,6 +212,14 @@ fn build_ping_message(cluster_id: u128, replica_id: u8) -> Message<GenericHeader
h.cluster = cluster_id;
h.replica = replica_id;
h.size = HEADER_SIZE as u32;
+ let challenge = auth::AuthChallenge {
+ cluster: cluster_id,
+ peer_id: u128::from(replica_id),
+ timestamp_ns,
+ release: h.release,
+ nonce,
+ };
+ auth::encode_envelope(&mut h.reserved_command, token_source, &challenge);
},
)
}
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index bb993fc8c..99c76633a 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+pub mod auth;
pub mod cache;
pub mod client_listener;
pub mod config;
diff --git a/core/message_bus/src/replica_io.rs b/core/message_bus/src/replica_io.rs
index 5ef866ff3..3731cc6d8 100644
--- a/core/message_bus/src/replica_io.rs
+++ b/core/message_bus/src/replica_io.rs
@@ -36,6 +36,7 @@ use std::time::Duration;
use iggy_common::IggyError;
+use crate::auth::TokenSource;
use crate::connector::start as start_connector;
use crate::replica_listener::{bind as bind_replica_listener, run as run_replica_listener};
use crate::{AcceptedClientFn, AcceptedReplicaFn, IggyMessageBus, client_listener};
@@ -71,6 +72,7 @@ pub async fn start_on_shard_zero(
on_accepted_replica: AcceptedReplicaFn,
on_accepted_client: AcceptedClientFn,
reconnect_period: Duration,
+ token_source: Rc<dyn TokenSource>,
) -> Result<Option<BoundPlanes>, IggyError> {
if bus.shard_id() != 0 {
return Ok(None);
@@ -82,6 +84,7 @@ pub async fn start_on_shard_zero(
let token_for_replica = bus.token();
let on_accepted_replica_for_listener = on_accepted_replica.clone();
let listener_max_message_size = bus.config().max_message_size;
+ let token_source_for_listener = Rc::clone(&token_source);
let replica_handle = compio::runtime::spawn(async move {
run_replica_listener(
replica_listener,
@@ -91,6 +94,7 @@ pub async fn start_on_shard_zero(
replica_count,
on_accepted_replica_for_listener,
listener_max_message_size,
+ token_source_for_listener,
)
.await;
});
@@ -109,6 +113,7 @@ pub async fn start_on_shard_zero(
peers,
on_accepted_replica,
reconnect_period,
+ token_source,
)
.await;
@@ -136,6 +141,7 @@ pub async fn start_on_shard_zero_default(
peers: Vec<(u8, SocketAddr)>,
on_accepted_replica: AcceptedReplicaFn,
on_accepted_client: AcceptedClientFn,
+ token_source: Rc<dyn TokenSource>,
) -> Result<Option<BoundPlanes>, IggyError> {
let reconnect_period = bus.config().reconnect_period;
start_on_shard_zero(
@@ -149,6 +155,7 @@ pub async fn start_on_shard_zero_default(
on_accepted_replica,
on_accepted_client,
reconnect_period,
+ token_source,
)
.await
}
diff --git a/core/message_bus/src/replica_listener.rs b/core/message_bus/src/replica_listener.rs
index 9d9db9192..d2d82d586 100644
--- a/core/message_bus/src/replica_listener.rs
+++ b/core/message_bus/src/replica_listener.rs
@@ -29,20 +29,23 @@
//!
//! # Security
//!
-//! The `Ping` handshake validates `cluster_id`, the directional bound, and
-//! `replica_count`. There is NO shared secret, no mTLS, no version
-//! negotiation. Deploy the replica listener on a trusted network boundary
-//! (cluster-local VPC, private subnet, overlay, or equivalent) - NEVER
-//! expose the replica port directly to the public internet. Follow-up work
-//! on an authenticated handshake is tracked under IGGY-112.
+//! The `Ping` handshake validates `cluster_id`, the directional bound,
+//! `replica_count`, and a BLAKE3-keyed MAC carried in
+//! `GenericHeader.reserved_command[0..57]` (see [`crate::auth`]). Peers
+//! that cannot produce a valid tag are rejected. Even so, TLS/encryption is
+//! NOT provided here: operators still need to deploy the replica port on a
+//! trusted network boundary (cluster-local VPC, private subnet, overlay)
+//! if wire confidentiality is required.
+use crate::auth::{self, NonceRing, TokenSource};
use crate::framing;
use crate::lifecycle::ShutdownToken;
use crate::{AcceptedReplicaFn, GenericHeader, Message};
use compio::net::{SocketOpts, TcpListener, TcpStream};
use futures::FutureExt;
use iggy_binary_protocol::Command2;
-use iggy_common::IggyError;
+use iggy_common::{IggyError, IggyTimestamp};
+use std::cell::RefCell;
use std::net::SocketAddr;
use std::rc::Rc;
use tracing::{debug, error, info, warn};
@@ -87,11 +90,13 @@ pub async fn run(
replica_count: u8,
on_accepted: AcceptedReplicaFn,
max_message_size: usize,
+ token_source: Rc<dyn TokenSource>,
) {
info!(
"Replica listener accepting on {:?}",
listener.local_addr().ok()
);
+ let nonces = RefCell::new(NonceRing::default());
loop {
futures::select! {
() = token.wait().fuse() => {
@@ -101,14 +106,23 @@ pub async fn run(
result = listener.accept().fuse() => {
match result {
Ok((mut stream, peer_addr)) => {
- match handshake(
- &mut stream,
- cluster_id,
- self_id,
- replica_count,
- max_message_size,
- )
- .await {
+ let read = handshake_read(&mut stream, cluster_id, max_message_size).await;
+ let outcome = match read {
+ Ok(parsed) => {
+ let now_ns = IggyTimestamp::now().as_nanos();
+ let mut ring = nonces.borrow_mut();
+ handshake_verify(
+ &parsed,
+ token_source.as_ref(),
+ &mut ring,
+ now_ns,
+ self_id,
+ replica_count,
+ )
+ }
+ Err(e) => Err(e),
+ };
+ match outcome {
Ok(peer_id) => {
on_accepted(stream, peer_id);
}
@@ -126,26 +140,23 @@ pub async fn run(
}
}
-/// Read and validate the `Ping` handshake from an inbound peer connection.
-///
-/// Returns the peer replica id on success.
-//
-// TODO(IGGY-112): authenticate the replica handshake.
-//
-// Current validation covers `cluster_id` and the directional id bound
-// only: no shared secret, no mTLS, no protocol version negotiation.
-// Any peer that reaches the replica port and knows the `cluster_id`
-// can pose as a replica. Acceptable on the assumed trusted network
-// boundary; hardening is tracked under IGGY-112 (authenticated
-// handshake + version negotiation).
+/// Parsed-but-unverified handshake captured from the wire. Held across
+/// the await boundary between I/O and the sync verify step.
+struct ParsedHandshake {
+ replica: u8,
+ decoded: auth::DecodedEnvelope,
+ challenge: auth::AuthChallenge,
+}
+
+/// I/O-only portion of the handshake: read the 256 B `Ping` frame, enforce
+/// command + cluster match, and parse the auth envelope's bytes. Does NOT
+/// touch the nonce ring or verify the tag.
#[allow(clippy::future_not_send)]
-async fn handshake(
+async fn handshake_read(
stream: &mut TcpStream,
our_cluster: u128,
- self_id: u8,
- replica_count: u8,
max_message_size: usize,
-) -> Result<u8, IggyError> {
+) -> Result<ParsedHandshake, IggyError> {
let msg = framing::read_message(stream, max_message_size).await?;
let header = msg.header();
if header.command != Command2::Ping {
@@ -154,11 +165,45 @@ async fn handshake(
if header.cluster != our_cluster {
return Err(IggyError::InvalidCommand);
}
+ let decoded = auth::decode_envelope(&header.reserved_command).map_err(auth::to_iggy_error)?;
+ let challenge = auth::AuthChallenge {
+ cluster: header.cluster,
+ peer_id: u128::from(header.replica),
+ timestamp_ns: decoded.timestamp_ns,
+ release: header.release,
+ nonce: decoded.nonce,
+ };
+ Ok(ParsedHandshake {
+ replica: header.replica,
+ decoded,
+ challenge,
+ })
+}
+
+/// Synchronous verifier: tag + replay, then the directional tiebreak.
+/// Runs outside any await so the `&mut NonceRing` borrow cannot span I/O.
+fn handshake_verify(
+ parsed: &ParsedHandshake,
+ token_source: &dyn TokenSource,
+ nonces: &mut NonceRing,
+ now_ns: u128,
+ self_id: u8,
+ replica_count: u8,
+) -> Result<u8, IggyError> {
+ auth::verify_envelope(
+ token_source,
+ &parsed.challenge,
+ &parsed.decoded,
+ now_ns,
+ nonces,
+ )
+ .map_err(auth::to_iggy_error)?;
+
// Directional rule: a replica only accepts inbound from peers with
// strictly lower ids. The peer is responsible for not dialing us if
// it has the higher id; this is just defensive.
- if header.replica >= replica_count || header.replica >= self_id {
+ if parsed.replica >= replica_count || parsed.replica >= self_id {
return Err(IggyError::InvalidCommand);
}
- Ok(header.replica)
+ Ok(parsed.replica)
}
diff --git a/core/message_bus/tests/backpressure.rs b/core/message_bus/tests/backpressure.rs
index 1a399aa04..7e849a981 100644
--- a/core/message_bus/tests/backpressure.rs
+++ b/core/message_bus/tests/backpressure.rs
@@ -24,7 +24,7 @@
mod common;
-use common::{header_only, install_replicas_locally, loopback};
+use common::{header_only, install_replicas_locally, loopback, test_token_source};
use iggy_binary_protocol::Command2;
use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
use message_bus::replica_listener::{MessageHandler, bind, run};
@@ -62,6 +62,7 @@ async fn try_send_returns_backpressure_when_queue_full() {
2,
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -77,6 +78,7 @@ async fn try_send_returns_backpressure_when_queue_full() {
vec![(1, addr1)],
dial_0,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await;
diff --git a/core/message_bus/tests/common/mod.rs b/core/message_bus/tests/common/mod.rs
index b00fbb310..81cdf1fa5 100644
--- a/core/message_bus/tests/common/mod.rs
+++ b/core/message_bus/tests/common/mod.rs
@@ -27,6 +27,7 @@
#![allow(dead_code)] // each test binary uses a subset
use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message};
+use message_bus::auth::{StaticSharedSecret, TokenSource};
use message_bus::client_listener::RequestHandler;
use message_bus::replica_listener::MessageHandler;
use message_bus::{AcceptedClientFn, AcceptedReplicaFn, IggyMessageBus, installer};
@@ -34,6 +35,13 @@ use std::cell::Cell;
use std::net::SocketAddr;
use std::rc::Rc;
+/// Default zero-byte shared secret for integration tests. All replicas
+/// authenticate against the same dummy source.
+#[must_use]
+pub fn test_token_source() -> Rc<dyn TokenSource> {
+ Rc::new(StaticSharedSecret::zero())
+}
+
/// Loopback address with OS-chosen port.
#[must_use]
pub fn loopback() -> SocketAddr {
diff --git a/core/message_bus/tests/connection_lost_notify.rs b/core/message_bus/tests/connection_lost_notify.rs
index 8babea3ad..00a05b5f7 100644
--- a/core/message_bus/tests/connection_lost_notify.rs
+++ b/core/message_bus/tests/connection_lost_notify.rs
@@ -22,7 +22,7 @@
mod common;
-use common::{install_replicas_locally, loopback};
+use common::{install_replicas_locally, loopback, test_token_source};
use message_bus::IggyMessageBus;
use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
use message_bus::replica_listener::{MessageHandler, bind, run};
@@ -64,6 +64,7 @@ async fn connection_lost_fires_exactly_once_per_peer_disconnect() {
2,
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -77,6 +78,7 @@ async fn connection_lost_fires_exactly_once_per_peer_disconnect() {
vec![(1u8, addr1)],
dial_0,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await;
diff --git a/core/message_bus/tests/directional_connection.rs b/core/message_bus/tests/directional_connection.rs
index ac8a3a7d1..6251939ce 100644
--- a/core/message_bus/tests/directional_connection.rs
+++ b/core/message_bus/tests/directional_connection.rs
@@ -22,7 +22,7 @@
mod common;
-use common::{install_replicas_locally, loopback};
+use common::{install_replicas_locally, loopback, test_token_source};
use message_bus::IggyMessageBus;
use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
use message_bus::replica_listener::{MessageHandler, bind, run};
@@ -53,6 +53,7 @@ async fn lower_id_dials_higher_id_accepts() {
2,
accept_0,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -69,6 +70,7 @@ async fn lower_id_dials_higher_id_accepts() {
2,
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -85,10 +87,20 @@ async fn lower_id_dials_higher_id_accepts() {
peers.clone(),
dial_0,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await;
let dial_1 = install_replicas_locally(bus1.clone(), on_message.clone());
- start_connector(&bus1, CLUSTER, 1, peers, dial_1, DEFAULT_RECONNECT_PERIOD).await;
+ start_connector(
+ &bus1,
+ CLUSTER,
+ 1,
+ peers,
+ dial_1,
+ DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
+ )
+ .await;
// Wait for the directional connection to settle.
let deadline = std::time::Instant::now() + Duration::from_secs(2);
diff --git a/core/message_bus/tests/handshake_auth.rs b/core/message_bus/tests/handshake_auth.rs
new file mode 100644
index 000000000..b833354e7
--- /dev/null
+++ b/core/message_bus/tests/handshake_auth.rs
@@ -0,0 +1,144 @@
+// 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.
+
+//! Integration coverage for the authenticated replica handshake.
+//!
+//! Complements the in-crate unit tests in `core/message_bus/src/auth.rs`
+//! by exercising the full end-to-end path (dial -> Ping frame -> listener
+//! verify) through `replica_listener::run` + `connector::start`.
+
+mod common;
+
+use common::{install_replicas_locally, loopback};
+use message_bus::IggyMessageBus;
+use message_bus::auth::{StaticSharedSecret, TokenSource};
+use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::replica_listener::{MessageHandler, bind, run};
+use std::rc::Rc;
+use std::time::Duration;
+
+const CLUSTER: u128 = 0xCAFE;
+
+/// Matching secrets on both sides: handshake succeeds, peer registers.
+#[compio::test]
+async fn handshake_with_matching_secret_registers_peer() {
+ let bus_listener = Rc::new(IggyMessageBus::new(0));
+ let bus_dialer = Rc::new(IggyMessageBus::new(0));
+
+ let secret: Rc<dyn TokenSource> = Rc::new(StaticSharedSecret::new([7u8; 32]));
+
+ let on_msg: MessageHandler = Rc::new(|_, _| {});
+
+ let (listener, addr) = bind(loopback()).await.unwrap();
+ let token_for_listener = bus_listener.token();
+ let accept = install_replicas_locally(bus_listener.clone(), on_msg.clone());
+ let secret_for_listener = Rc::clone(&secret);
+ let listener_handle = compio::runtime::spawn(async move {
+ run(
+ listener,
+ token_for_listener,
+ CLUSTER,
+ 1,
+ 2,
+ accept,
+ message_bus::framing::MAX_MESSAGE_SIZE,
+ secret_for_listener,
+ )
+ .await;
+ });
+ bus_listener.track_background(listener_handle);
+
+ let dial = install_replicas_locally(bus_dialer.clone(), on_msg);
+ start_connector(
+ &bus_dialer,
+ CLUSTER,
+ 0,
+ vec![(1u8, addr)],
+ dial,
+ DEFAULT_RECONNECT_PERIOD,
+ Rc::clone(&secret),
+ )
+ .await;
+
+ let deadline = std::time::Instant::now() + Duration::from_secs(2);
+ while !bus_dialer.replicas().contains(1) {
+ assert!(
+ std::time::Instant::now() < deadline,
+ "matching-secret handshake did not register peer"
+ );
+ compio::time::sleep(Duration::from_millis(5)).await;
+ }
+
+ bus_dialer.shutdown(Duration::from_secs(2)).await;
+ bus_listener.shutdown(Duration::from_secs(2)).await;
+}
+
+/// Mismatched secrets: the listener rejects the dialed Ping, so neither
+/// side ends up with a registered peer within the test window.
+#[compio::test]
+async fn handshake_with_mismatched_secret_rejects_peer() {
+ let bus_listener = Rc::new(IggyMessageBus::new(0));
+ let bus_dialer = Rc::new(IggyMessageBus::new(0));
+
+ let listener_secret: Rc<dyn TokenSource> = Rc::new(StaticSharedSecret::new([1u8; 32]));
+ let dialer_secret: Rc<dyn TokenSource> = Rc::new(StaticSharedSecret::new([2u8; 32]));
+
+ let on_msg: MessageHandler = Rc::new(|_, _| {});
+
+ let (listener, addr) = bind(loopback()).await.unwrap();
+ let token_for_listener = bus_listener.token();
+ let accept = install_replicas_locally(bus_listener.clone(), on_msg.clone());
+ let listener_handle = compio::runtime::spawn(async move {
+ run(
+ listener,
+ token_for_listener,
+ CLUSTER,
+ 1,
+ 2,
+ accept,
+ message_bus::framing::MAX_MESSAGE_SIZE,
+ listener_secret,
+ )
+ .await;
+ });
+ bus_listener.track_background(listener_handle);
+
+ let dial = install_replicas_locally(bus_dialer.clone(), on_msg);
+ start_connector(
+ &bus_dialer,
+ CLUSTER,
+ 0,
+ vec![(1u8, addr)],
+ dial,
+ Duration::from_millis(50),
+ dialer_secret,
+ )
+ .await;
+
+ compio::time::sleep(Duration::from_millis(400)).await;
+ assert!(
+ !bus_dialer.replicas().contains(1),
+ "mismatched-secret handshake must not register the peer"
+ );
+ assert!(
+ !bus_listener.replicas().contains(0),
+ "mismatched-secret handshake must not register the peer on the listener either"
+ );
+
+ bus_dialer.shutdown(Duration::from_secs(2)).await;
+ bus_listener.shutdown(Duration::from_secs(2)).await;
+}
diff --git a/core/message_bus/tests/head_of_line.rs b/core/message_bus/tests/head_of_line.rs
index 57c644971..b756897ce 100644
--- a/core/message_bus/tests/head_of_line.rs
+++ b/core/message_bus/tests/head_of_line.rs
@@ -25,7 +25,7 @@
mod common;
-use common::{header_only, install_replicas_locally, loopback};
+use common::{header_only, install_replicas_locally, loopback, test_token_source};
use compio::net::TcpListener;
use iggy_binary_protocol::Command2;
use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
@@ -58,6 +58,7 @@ async fn slow_peer_does_not_block_other_peers() {
3,
accept_a,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -91,6 +92,7 @@ async fn slow_peer_does_not_block_other_peers() {
vec![(1, addr_a), (2, addr_b)],
dial_delegate,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await;
diff --git a/core/message_bus/tests/reconnect.rs b/core/message_bus/tests/reconnect.rs
index 0eef98e94..93ac64332 100644
--- a/core/message_bus/tests/reconnect.rs
+++ b/core/message_bus/tests/reconnect.rs
@@ -21,7 +21,7 @@
mod common;
-use common::{install_replicas_locally, loopback};
+use common::{install_replicas_locally, loopback, test_token_source};
use message_bus::IggyMessageBus;
use message_bus::connector::start as start_connector;
use message_bus::replica_listener::{MessageHandler, bind, run};
@@ -43,7 +43,16 @@ async fn periodic_retry_picks_up_late_listener() {
let on_message: MessageHandler = Rc::new(|_, _| {});
let period = Duration::from_millis(100);
let dial_delegate = install_replicas_locally(bus0.clone(), on_message.clone());
- start_connector(&bus0, CLUSTER, 0, vec![(1, addr)], dial_delegate, period).await;
+ start_connector(
+ &bus0,
+ CLUSTER,
+ 0,
+ vec![(1, addr)],
+ dial_delegate,
+ period,
+ test_token_source(),
+ )
+ .await;
assert!(!bus0.replicas().contains(1), "first connect should fail");
// Bring bus 1 online on the same address.
@@ -60,6 +69,7 @@ async fn periodic_retry_picks_up_late_listener() {
2,
accept_delegate,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
diff --git a/core/message_bus/tests/reconnect_skip_connected.rs b/core/message_bus/tests/reconnect_skip_connected.rs
index 5d9a77b3b..31159e1db 100644
--- a/core/message_bus/tests/reconnect_skip_connected.rs
+++ b/core/message_bus/tests/reconnect_skip_connected.rs
@@ -22,7 +22,7 @@
mod common;
-use common::{install_replicas_locally, loopback};
+use common::{install_replicas_locally, loopback, test_token_source};
use message_bus::IggyMessageBus;
use message_bus::connector::start as start_connector;
use message_bus::replica_listener::{MessageHandler, bind, run};
@@ -59,6 +59,7 @@ async fn periodic_reconnect_skips_already_connected_peer() {
2,
accept_delegate,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -69,7 +70,16 @@ async fn periodic_reconnect_skips_already_connected_peer() {
// missing, bus1's listener will see N extra accepts.
let period = Duration::from_millis(50);
let dial_delegate = install_replicas_locally(bus0.clone(), on_message.clone());
- start_connector(&bus0, CLUSTER, 0, vec![(1u8, addr1)], dial_delegate, period).await;
+ start_connector(
+ &bus0,
+ CLUSTER,
+ 0,
+ vec![(1u8, addr1)],
+ dial_delegate,
+ period,
+ test_token_source(),
+ )
+ .await;
// Wait for the initial connection to settle on both sides.
let deadline = std::time::Instant::now() + Duration::from_secs(2);
diff --git a/core/message_bus/tests/replica_roundtrip.rs b/core/message_bus/tests/replica_roundtrip.rs
index e557b501a..8d8fde80a 100644
--- a/core/message_bus/tests/replica_roundtrip.rs
+++ b/core/message_bus/tests/replica_roundtrip.rs
@@ -23,7 +23,7 @@
mod common;
use async_channel::Receiver;
-use common::{header_only, install_replicas_locally, loopback};
+use common::{header_only, install_replicas_locally, loopback, test_token_source};
use iggy_binary_protocol::Command2;
use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
use message_bus::replica_listener::{MessageHandler, bind, run};
@@ -55,6 +55,7 @@ async fn two_replicas_exchange_prepare_and_ack() {
2,
accept_delegate_1,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -76,6 +77,7 @@ async fn two_replicas_exchange_prepare_and_ack() {
vec![(1, addr1)],
dial_delegate_0,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await;
diff --git a/core/message_bus/tests/shard_zero_gating.rs b/core/message_bus/tests/shard_zero_gating.rs
index a65fd84d7..1dc0cc339 100644
--- a/core/message_bus/tests/shard_zero_gating.rs
+++ b/core/message_bus/tests/shard_zero_gating.rs
@@ -21,7 +21,7 @@
mod common;
-use common::{install_clients_locally, install_replicas_locally, loopback};
+use common::{install_clients_locally, install_replicas_locally, loopback, test_token_source};
use message_bus::IggyMessageBus;
use message_bus::client_listener::RequestHandler;
use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
@@ -51,6 +51,7 @@ async fn shard_zero_binds_listener_and_starts_connector() {
2,
peer_accept,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -76,6 +77,7 @@ async fn shard_zero_binds_listener_and_starts_connector() {
accepted_replica,
accepted_client,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await
.expect("start_on_shard_zero must succeed on shard 0");
@@ -127,6 +129,7 @@ async fn non_zero_shard_skips_io() {
accepted_replica,
accepted_client,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await
.expect("start_on_shard_zero must succeed on non-zero shard (no-op)");
diff --git a/core/message_bus/tests/vectored_batch.rs b/core/message_bus/tests/vectored_batch.rs
index c8fc10c6b..b079725a5 100644
--- a/core/message_bus/tests/vectored_batch.rs
+++ b/core/message_bus/tests/vectored_batch.rs
@@ -20,7 +20,7 @@
mod common;
-use common::{header_only, install_replicas_locally, loopback};
+use common::{header_only, install_replicas_locally, loopback, test_token_source};
use iggy_binary_protocol::Command2;
use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
use message_bus::replica_listener::{MessageHandler, bind, run};
@@ -57,6 +57,7 @@ async fn writer_batches_pipelined_sends_in_order() {
2,
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
+ test_token_source(),
)
.await;
});
@@ -72,6 +73,7 @@ async fn writer_batches_pipelined_sends_in_order() {
vec![(1, addr1)],
dial_0,
DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
)
.await;
(iggy) 03/14: docs(message_bus): name the plane split + TCP-forever invariant (P0-T4)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e444085a3e1c0838752832951e9ce8c704125b42
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 11:46:47 2026 +0200
docs(message_bus): name the plane split + TCP-forever invariant (P0-T4)
Add a crate-level doc comment to lib.rs covering the replica plane /
SDK-client plane distinction, the authenticated-handshake envelope,
and the four invariants a module editor has to preserve: fast-path
Ready return, writev batching, TCP-only fd delegation, and the
0-RTT-off default on any future QUIC path. Points at the
transport-plan notes under Documents/silverhand/iggy/message_bus for
full context.
Promote the "replica plane = TCP forever" rationale from a prose
paragraph at the top of replica_listener.rs to a first-class section
so a contributor touching the listener sees the VSR / fd-delegation /
writev / RTT reasoning immediately, not as implementation detail.
Refs: Documents/silverhand/iggy/message_bus/transport-plan/ (P0-T4,
INVARIANTS.md I1).
---
core/message_bus/src/lib.rs | 44 ++++++++++++++++++++++++++++++++
core/message_bus/src/replica_listener.rs | 14 ++++++++++
2 files changed, 58 insertions(+)
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index 99c76633a..daf52fd83 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -15,6 +15,50 @@
// specific language governing permissions and limitations
// under the License.
+//! Shard-local message bus with two wire planes.
+//!
+//! # Plane split
+//!
+//! - **Replica plane (TCP forever)**: VSR consensus traffic between
+//! replicas. Implemented in [`replica_listener`], [`connector`], and
+//! [`replica_io`]. Datagram or gateway-terminated transports are NOT
+//! supported here and never will be — see
+//! `replica_listener`'s module docs for the rationale.
+//! - **SDK-client plane**: ephemeral client connections. TCP today via
+//! [`client_listener`]; Phase 2+ adds WebSocket / QUIC on a
+//! demand-gated basis. At-most-once semantics come from the
+//! `(client: u128, request: u64)` pair in `RequestHeader`; the dedup
+//! store lives in `core/server-ng/src/dedup.rs` on each shard.
+//!
+//! # Authenticated handshake
+//!
+//! Both planes use [`auth::TokenSource`] for identity. The replica
+//! plane ships with [`auth::StaticSharedSecret`] (all replicas share
+//! one 32-byte key); the client plane will grow per-client token
+//! sources in Phase 2+. The envelope lives in
+//! `GenericHeader.reserved_command[0..57]` — 32 B BLAKE3 MAC, 8 B
+//! timestamp, 16 B nonce, 1 B kind — so the 256 B header layout is
+//! unchanged.
+//!
+//! # Invariants worth naming
+//!
+//! - [`send_to_client`](IggyMessageBus::send_to_client) and
+//! [`send_to_replica`](IggyMessageBus::send_to_replica) return
+//! `Ready` on first poll. Consensus code relies on this for
+//! reentrancy reasoning; any `.await` in the body breaks it.
+//! - [`writer_task`] coalesces up to `max_batch = 256`
+//! `Frozen<MESSAGE_ALIGN>` into one `write_vectored_all`. Don't
+//! introduce per-message syscalls or per-message encryption.
+//! - fd-delegation ([`fd_transfer`]) is TCP-only. TLS / QUIC
+//! connections have no dupable plaintext fd, so shard 0 terminates
+//! and forwards `Frozen<MESSAGE_ALIGN>` over the existing
+//! inter-shard flume.
+//! - 0-RTT stays disabled by default on any future QUIC path. Per-
+//! command opt-in requires a checked-in idempotence audit.
+//!
+//! The full invariant list and the transport-plan design notes live
+//! under `Documents/silverhand/iggy/message_bus/transport-plan/`.
+
pub mod auth;
pub mod cache;
pub mod client_listener;
diff --git a/core/message_bus/src/replica_listener.rs b/core/message_bus/src/replica_listener.rs
index d2d82d586..b056f30ea 100644
--- a/core/message_bus/src/replica_listener.rs
+++ b/core/message_bus/src/replica_listener.rs
@@ -17,6 +17,20 @@
//! Inbound TCP listener for replica-to-replica consensus traffic.
//!
+//! # Replica plane = TCP forever (load-bearing invariant)
+//!
+//! This module is TCP-only by design, not by omission. Do NOT add WS,
+//! QUIC, or HTTP paths here. The VSR chain-hash safety proof
+//! (`last_prepare_checksum` in `core/consensus/src/plane_helpers.rs`),
+//! the fd-delegation path (`F_DUPFD_CLOEXEC` needs a dupable plaintext
+//! fd), the `writev` batching in `writer_task.rs`, and the single-digit
+//! RTT assumptions used by the view-change timer all rest on a FIFO
+//! byte stream between a bounded set of replicas on a trusted LAN. A
+//! datagram-oriented or gateway-terminated transport violates one or
+//! more of those assumptions. See
+//! `Documents/silverhand/iggy/message_bus/transport-plan/INVARIANTS.md`
+//! (I1) for the debate context.
+//!
//! Runs only on shard 0. On every successful `Ping` handshake the listener
//! hands the accepted `TcpStream` to an `on_accepted` callback provided by
//! the shard bootstrap, which dup-and-ships the fd to the owning shard via
(iggy) 05/14: feat(message_bus): TCP impls behind Transport traits (IGGY-112, P1-T2)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b6e9650f3644133d4c49a5989fbd7901f25dcd1e
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 12:03:28 2026 +0200
feat(message_bus): TCP impls behind Transport traits (IGGY-112, P1-T2)
Phase 1 plugs TCP into the trait surface landed in P1-T1. Without
an impl, the traits are untestable scaffolding and no downstream
transport (WS, QUIC) has a reference to compare against. Wire and
socket behaviour must not change: existing `write_vectored_all`
batching, zero-copy `Frozen` ownership, and the per-peer drain
cadence are load-bearing (I4, I8).
Add `transports/tcp.rs` with `TcpTransportListener`,
`TcpTransportConn`, `TcpTransportReader`, `TcpTransportWriter`.
`send_batch` does `mem::take` + `write_vectored_all` + `clear`,
returning the drained Vec through the caller's `&mut` slot so the
allocation is reused across iterations. Migrate `writer_task::run`
to drive the trait via a new generic `run_transport<W: TransportWriter>`;
the TCP entry point keeps its exact public signature and just wraps
the owned write half in `TcpTransportWriter` before delegating. Fast
path stays monomorphized - no dyn dispatch, no vtable.
Behavior-preserving: 100 tests pass (96 existing + 4 new trait
exercises covering listener accept, batched writes, empty batch,
framing error). `installer::install_*_stream` still takes
`TcpStream` directly; generic migration there is P1-T3.
---
core/message_bus/src/transports/mod.rs | 4 +
core/message_bus/src/transports/tcp.rs | 275 +++++++++++++++++++++++++++++++++
core/message_bus/src/writer_task.rs | 48 ++++--
3 files changed, 314 insertions(+), 13 deletions(-)
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index f47e0c404..c98faf467 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -65,6 +65,10 @@
//! plug in behind the same surface; see
//! `Documents/silverhand/iggy/message_bus/transport-plan/`.
+pub mod tcp;
+
+pub use tcp::{TcpTransportConn, TcpTransportListener, TcpTransportReader, TcpTransportWriter};
+
use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_binary_protocol::{GenericHeader, Message};
diff --git a/core/message_bus/src/transports/tcp.rs b/core/message_bus/src/transports/tcp.rs
new file mode 100644
index 000000000..7478c99b4
--- /dev/null
+++ b/core/message_bus/src/transports/tcp.rs
@@ -0,0 +1,275 @@
+// 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.
+
+//! TCP impls of the [`super`] transport traits.
+//!
+//! Behavior-preserving wrappers around `compio::net::TcpListener`,
+//! `compio::net::TcpStream`, and the split halves. The hot path on
+//! [`TcpTransportWriter::send_batch`] is identical to
+//! [`crate::writer_task::run`]'s inner `write_vectored_all` call: one
+//! syscall per batch, zero intermediate copies of `Frozen`.
+//!
+//! Callers that need listener / dialer ergonomics (socket options,
+//! keepalive, directional accept rules) should continue to use the
+//! free functions in [`crate::replica_listener`] and
+//! [`crate::client_listener`]; this module is the minimal trait
+//! adapter, not a replacement for those call sites. Integration with
+//! `installer::install_*_stream` lands in P1-T3.
+
+use super::{TransportConn, TransportListener, TransportReader, TransportWriter};
+use crate::framing;
+use compio::io::AsyncWriteExt;
+use compio::net::{OwnedReadHalf, OwnedWriteHalf, TcpListener, TcpStream};
+use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
+use iggy_binary_protocol::consensus::iobuf::Frozen;
+use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_common::IggyError;
+use std::io;
+use std::mem;
+use std::net::SocketAddr;
+
+/// Inbound TCP listener wrapper.
+///
+/// Constructed from an already-bound [`TcpListener`] so the caller
+/// keeps control over socket options (`SO_REUSEPORT`, `nodelay`,
+/// `keepalive`) via `compio::net::SocketOpts`. See
+/// [`crate::replica_listener::bind`] and
+/// [`crate::client_listener::bind`] for the canonical construction.
+pub struct TcpTransportListener {
+ inner: TcpListener,
+}
+
+impl TcpTransportListener {
+ #[must_use]
+ pub const fn new(inner: TcpListener) -> Self {
+ Self { inner }
+ }
+}
+
+impl TransportListener for TcpTransportListener {
+ type Conn = TcpTransportConn;
+
+ #[allow(clippy::future_not_send)]
+ async fn accept(&self) -> io::Result<(Self::Conn, SocketAddr)> {
+ let (stream, addr) = self.inner.accept().await?;
+ Ok((TcpTransportConn::new(stream), addr))
+ }
+}
+
+/// Single TCP connection.
+///
+/// Produced by [`TcpTransportListener::accept`] or by wrapping the
+/// result of a `TcpStream::connect` on the dialer path. Takes ownership
+/// of the stream; [`Self::into_split`] transfers that ownership into
+/// the read and write halves bound to the per-connection tasks.
+pub struct TcpTransportConn {
+ stream: TcpStream,
+}
+
+impl TcpTransportConn {
+ #[must_use]
+ pub const fn new(stream: TcpStream) -> Self {
+ Self { stream }
+ }
+}
+
+impl TransportConn for TcpTransportConn {
+ type Reader = TcpTransportReader;
+ type Writer = TcpTransportWriter;
+
+ fn into_split(self) -> (Self::Reader, Self::Writer) {
+ let (read_half, write_half) = self.stream.into_split();
+ (
+ TcpTransportReader { inner: read_half },
+ TcpTransportWriter { inner: write_half },
+ )
+ }
+}
+
+/// Read half bound to the per-connection reader task.
+///
+/// [`TransportReader::read_message`] delegates to
+/// [`framing::read_message`]; the two paths share the same header
+/// decode, bounds check, and zero-copy `Owned<MESSAGE_ALIGN>`
+/// allocation strategy.
+pub struct TcpTransportReader {
+ inner: OwnedReadHalf<TcpStream>,
+}
+
+impl TcpTransportReader {
+ #[must_use]
+ pub const fn new(inner: OwnedReadHalf<TcpStream>) -> Self {
+ Self { inner }
+ }
+}
+
+impl TransportReader for TcpTransportReader {
+ #[allow(clippy::future_not_send)]
+ async fn read_message(
+ &mut self,
+ max_message_size: usize,
+ ) -> Result<Message<GenericHeader>, IggyError> {
+ framing::read_message(&mut self.inner, max_message_size).await
+ }
+}
+
+/// Write half bound to the per-connection writer-batch task.
+///
+/// [`TransportWriter::send_batch`] calls
+/// [`compio::io::AsyncWriteExt::write_vectored_all`] exactly once per
+/// invocation. The caller (e.g. [`crate::writer_task::run`]) is
+/// responsible for capping the batch size to
+/// `max_batch <= IOV_MAX / 2 = 512`; this impl does not enforce a cap
+/// because the Vec is already drained by the caller's admission
+/// control.
+pub struct TcpTransportWriter {
+ inner: OwnedWriteHalf<TcpStream>,
+}
+
+impl TcpTransportWriter {
+ #[must_use]
+ pub const fn new(inner: OwnedWriteHalf<TcpStream>) -> Self {
+ Self { inner }
+ }
+}
+
+impl TransportWriter for TcpTransportWriter {
+ #[allow(clippy::future_not_send)]
+ async fn send_batch(&mut self, batch: &mut Vec<Frozen<MESSAGE_ALIGN>>) -> io::Result<()> {
+ // `write_vectored_all` consumes the Vec via compio's `IoVectoredBuf`
+ // surface and returns it through `BufResult` so we can reuse the
+ // allocation. Take the inner Vec, hand it to the kernel, put the
+ // (now-empty) returned Vec back into the caller's slot.
+ let owned = mem::take(batch);
+ let compio::BufResult(result, mut returned) = self.inner.write_vectored_all(owned).await;
+ returned.clear();
+ *batch = returned;
+ result
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy_binary_protocol::consensus::iobuf::Frozen;
+ use iggy_binary_protocol::{Command2, HEADER_SIZE};
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn header_only(command: Command2) -> Frozen<MESSAGE_ALIGN> {
+ Message::<GenericHeader>::new(HEADER_SIZE)
+ .transmute_header(|_, h: &mut GenericHeader| {
+ h.command = command;
+ h.size = HEADER_SIZE as u32;
+ })
+ .into_frozen()
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn local_pair() -> (TcpStream, TcpStream) {
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ let connect = TcpStream::connect(addr);
+ let accept = listener.accept();
+ let (client_res, accept_res) = futures::join!(connect, accept);
+ let (server, _) = accept_res.unwrap();
+ (client_res.unwrap(), server)
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn listener_accept_yields_conn() {
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ let wrapped = TcpTransportListener::new(listener);
+
+ let connect = TcpStream::connect(addr);
+ let accept = wrapped.accept();
+ let (_client, accept_res) = futures::join!(connect, accept);
+ let (_conn, _peer_addr) = accept_res.expect("accept via trait");
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn send_batch_writes_all_and_drains_vec() {
+ let (client, server) = local_pair().await;
+ let client_conn = TcpTransportConn::new(client);
+ let (_client_read, mut client_write) = client_conn.into_split();
+
+ let server_conn = TcpTransportConn::new(server);
+ let (mut server_read, _server_write) = server_conn.into_split();
+
+ let mut batch = vec![
+ header_only(Command2::Ping),
+ header_only(Command2::Prepare),
+ header_only(Command2::Request),
+ ];
+ client_write
+ .send_batch(&mut batch)
+ .await
+ .expect("send_batch");
+ assert!(batch.is_empty(), "Vec must be drained on success");
+ assert!(batch.capacity() >= 3, "allocation must be reused");
+
+ // Verify all three frames land intact in order.
+ let a = server_read
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ let b = server_read
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ let c = server_read
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .unwrap();
+ assert_eq!(a.header().command, Command2::Ping);
+ assert_eq!(b.header().command, Command2::Prepare);
+ assert_eq!(c.header().command, Command2::Request);
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn send_batch_empty_is_noop() {
+ let (client, _server) = local_pair().await;
+ let (_r, mut w) = TcpTransportConn::new(client).into_split();
+ let mut batch: Vec<Frozen<MESSAGE_ALIGN>> = Vec::with_capacity(8);
+ w.send_batch(&mut batch).await.expect("empty batch ok");
+ assert!(batch.is_empty());
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn read_message_reports_oversize_via_trait() {
+ use compio::io::AsyncWriteExt;
+ let (mut client, server) = local_pair().await;
+ let (mut r, _w) = TcpTransportConn::new(server).into_split();
+
+ // Hand-craft a header with a bogus oversize `size` field; the
+ // trait surface must surface the same `InvalidCommand` error the
+ // framing path does.
+ let mut buf = vec![0u8; HEADER_SIZE];
+ let bogus = u32::try_from(framing::MAX_MESSAGE_SIZE + 1)
+ .unwrap_or(u32::MAX)
+ .to_le_bytes();
+ buf[48..52].copy_from_slice(&bogus);
+ client.write_all(buf).await.0.unwrap();
+
+ let res = r.read_message(framing::MAX_MESSAGE_SIZE).await;
+ assert!(matches!(res, Err(IggyError::InvalidCommand)));
+ }
+}
diff --git a/core/message_bus/src/writer_task.rs b/core/message_bus/src/writer_task.rs
index de3f3304e..75b09c827 100644
--- a/core/message_bus/src/writer_task.rs
+++ b/core/message_bus/src/writer_task.rs
@@ -31,7 +31,7 @@
//! - a write to the wire fails (broken connection).
use crate::lifecycle::{BusMessage, BusReceiver, ShutdownToken};
-use compio::io::AsyncWriteExt;
+use crate::transports::{TcpTransportWriter, TransportWriter};
use compio::net::{OwnedWriteHalf, TcpStream};
use futures::FutureExt;
use tracing::{debug, error, trace};
@@ -42,10 +42,38 @@ use tracing::{debug, error, trace};
/// `max_batch` caps how many messages a single `writev` syscall coalesces.
/// Larger batches reduce syscalls per N messages at the cost of memory
/// per batch and worst-case latency for the head-of-batch message.
+///
+/// TCP entry point. Wraps the owned write half in a
+/// [`TcpTransportWriter`] and delegates the drain loop to
+/// [`run_transport`]; every syscall still flows through
+/// [`TransportWriter::send_batch`] so future transports drop in behind
+/// the same drain logic.
#[allow(clippy::future_not_send)]
pub async fn run(
rx: BusReceiver,
- mut write_half: OwnedWriteHalf<TcpStream>,
+ write_half: OwnedWriteHalf<TcpStream>,
+ token: ShutdownToken,
+ label: &'static str,
+ peer: String,
+ max_batch: usize,
+) {
+ let writer = TcpTransportWriter::new(write_half);
+ run_transport(rx, writer, token, label, peer, max_batch).await;
+}
+
+/// Generic drain loop over any [`TransportWriter`].
+///
+/// Pulls `BusMessage`s off the per-peer mpsc, coalesces up to
+/// `max_batch` into a single [`TransportWriter::send_batch`] call, and
+/// exits cleanly on shutdown, channel close, or write error.
+///
+/// Public so alternate transports (WS, QUIC) can reuse the admission
+/// control and batch sizing identically to TCP; keep the body
+/// transport-agnostic.
+#[allow(clippy::future_not_send)]
+pub async fn run_transport<W: TransportWriter>(
+ rx: BusReceiver,
+ mut writer: W,
token: ShutdownToken,
label: &'static str,
peer: String,
@@ -89,12 +117,10 @@ pub async fn run(
let drained = batch.len();
trace!(%label, %peer, batch = drained, "writev batch");
- // Single writev for the whole batch. write_vectored_all loops
- // internally on partial writes until the full batch lands or the
- // socket errors.
- let to_write = std::mem::take(&mut batch);
- let compio::BufResult(result, mut returned) = write_half.write_vectored_all(to_write).await;
- if let Err(e) = result {
+ // Single batch send. The `TransportWriter` impl is atomic-or-error
+ // and drains `batch` in place on success so the allocation is
+ // reused across iterations.
+ if let Err(e) = writer.send_batch(&mut batch).await {
// Error (not warn) because the batch is now on the floor:
// VSR's prepare-timeout or view-change will recover, but the
// operator needs a loud diagnostic to correlate with the
@@ -105,13 +131,9 @@ pub async fn run(
%peer,
error = ?e,
batch_len = drained,
- "writer task: write_vectored_all failed, dropping batch"
+ "writer task: send_batch failed, dropping batch"
);
return;
}
-
- // Reuse the Vec allocation across iterations.
- returned.clear();
- batch = returned;
}
}
(iggy) 09/14: fix(message_bus): tag-verify before nonce insert + review hardening
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c586bd72505b94f24ccf6d699b10a47357794712
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 14:20:25 2026 +0200
fix(message_bus): tag-verify before nonce insert + review hardening
verify_envelope inserted the nonce ring slot BEFORE verifying
the BLAKE3 tag. An off-secret attacker with TCP reachability
to the listener could flush the ring with forged-tag Pings
and open a 30 s replay window for a captured legit Ping.
Reorder to timestamp -> tag -> nonce-insert; add regression
test. Triple-confirmed by independent multi-expert review.
Same pass landed adjacent hardening: NonceRing per-peer +
bus-resident with inline [u128; CAP] storage; LABEL_REPLICA /
LABEL_CLIENT plane separation in AuthChallenge; UnsupportedKind
-> Unauthenticated for mixed-version triage; max_batch
boot-asserted in 1..=IOV_MAX/2; auth + TcpTransport* + dedup
surfaces shrunk to pub(crate); Unpin dropped from transport
traits; InvalidHex carries byte; Y2554 debug_assert in
build_ping_message; notify_connection_lost reentrant-safe;
framing.rs reserve_exact diagnostic log; dedup docstring
clarified to "at-most-once per-shard-lifetime"; installer.rs
race comments tightened. 10 new regression tests; 120 pass.
---
core/message_bus/src/auth.rs | 274 +++++++++++++++++----
core/message_bus/src/auth_config.rs | 20 +-
core/message_bus/src/config.rs | 11 +
core/message_bus/src/connector.rs | 18 +-
core/message_bus/src/framing.rs | 13 +-
core/message_bus/src/installer.rs | 8 +-
core/message_bus/src/lib.rs | 164 +++++++++++-
core/message_bus/src/replica_io.rs | 2 +
core/message_bus/src/replica_listener.rs | 44 +++-
core/message_bus/src/transports/mod.rs | 23 +-
core/message_bus/src/transports/tcp.rs | 26 +-
core/message_bus/src/writer_task.rs | 4 +-
core/message_bus/tests/backpressure.rs | 2 +
core/message_bus/tests/connection_lost_notify.rs | 2 +
core/message_bus/tests/directional_connection.rs | 4 +
core/message_bus/tests/handshake_auth.rs | 4 +
core/message_bus/tests/head_of_line.rs | 2 +
core/message_bus/tests/reconnect.rs | 2 +
core/message_bus/tests/reconnect_skip_connected.rs | 2 +
core/message_bus/tests/replica_roundtrip.rs | 2 +
core/message_bus/tests/shard_zero_gating.rs | 2 +
core/message_bus/tests/vectored_batch.rs | 2 +
core/server-ng/src/dedup.rs | 52 +++-
core/server-ng/src/lib.rs | 6 +-
24 files changed, 586 insertions(+), 103 deletions(-)
diff --git a/core/message_bus/src/auth.rs b/core/message_bus/src/auth.rs
index 1205a4578..dd90b18d2 100644
--- a/core/message_bus/src/auth.rs
+++ b/core/message_bus/src/auth.rs
@@ -40,15 +40,39 @@ use thiserror::Error;
pub const SECRET_SIZE: usize = 32;
/// Byte range of the envelope within `GenericHeader.reserved_command`.
-pub const ENVELOPE_LEN: usize = 57;
+pub(crate) const ENVELOPE_LEN: usize = 57;
/// Acceptance window around `auth_timestamp`, in nanoseconds. 30 s.
-pub const TIMESTAMP_WINDOW_NS: u128 = 30_000_000_000;
+pub(crate) const TIMESTAMP_WINDOW_NS: u128 = 30_000_000_000;
-/// Default nonce-dedup LRU capacity on the acceptor.
-pub const NONCE_LRU_CAPACITY: usize = 256;
-
-const LABEL: &[u8; 16] = b"iggy-bus-auth-v1";
+/// Per-peer nonce-dedup ring capacity on the acceptor.
+///
+/// 8 slots × 5-replica clusters × 16 B = 640 B inline; survives the
+/// 30 s acceptance window even at the worst-case dial cadence
+/// (`reconnect_period = 5 s`). Per-peer (rather than a single global
+/// ring) means a flapping or hostile peer cannot evict honest peers'
+/// nonce slots, so the verify-order guard in [`verify_envelope`] holds
+/// up against an attacker who can reach the listener but lacks the
+/// secret. See plan F2.
+#[doc(hidden)]
+pub const PER_PEER_NONCE_CAPACITY: usize = 8;
+
+/// Plane-tagged 16-byte label prefixed onto every challenge before the MAC.
+///
+/// Domain separation: same cluster secret produces distinct MACs across
+/// the replica plane and the (Phase 2) SDK-client plane, so a captured
+/// replica handshake cannot be replayed onto the client plane and vice
+/// versa even when both planes share a [`StaticSharedSecret`]. Bump the
+/// trailing version digits when the challenge layout itself changes;
+/// bump [`AuthKind`] when the MAC algorithm changes.
+pub(crate) const LABEL_REPLICA: &[u8; 16] = b"iggy-bus-rep-v01";
+
+/// Plane label for the SDK-client plane (reserved for Phase 2; not
+/// exercised on the wire today). Defined now so the cross-plane MAC
+/// guard test in this module can exercise both labels and the test
+/// catches a regression before Phase 2 wiring lands.
+#[allow(dead_code)]
+pub(crate) const LABEL_CLIENT: &[u8; 16] = b"iggy-bus-cli-v01";
const KIND_OFFSET: usize = 56;
const TAG_RANGE: std::ops::Range<usize> = 0..32;
@@ -59,7 +83,7 @@ const RESERVED_RANGE: std::ops::Range<usize> = ENVELOPE_LEN..128;
/// Envelope version tag.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
-pub enum AuthKind {
+pub(crate) enum AuthKind {
Blake3V1 = 1,
}
@@ -72,9 +96,13 @@ impl AuthKind {
}
}
-/// Fully reconstructed challenge input. All fields come from the header
-/// (`GenericHeader.cluster`, `GenericHeader.replica`, `GenericHeader.release`)
-/// plus the envelope timestamp and nonce.
+/// Fully reconstructed challenge input.
+///
+/// All fields come from the header (`GenericHeader.cluster`,
+/// `GenericHeader.replica`, `GenericHeader.release`) plus the envelope
+/// timestamp and nonce. `label` is the plane-tagged domain-separation
+/// prefix; pick [`LABEL_REPLICA`] for the consensus plane and
+/// [`LABEL_CLIENT`] for the SDK-client plane (Phase 2).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AuthChallenge {
pub cluster: u128,
@@ -82,15 +110,17 @@ pub struct AuthChallenge {
pub timestamp_ns: u64,
pub release: u32,
pub nonce: u128,
+ pub label: &'static [u8; 16],
}
impl AuthChallenge {
- /// Serialize the challenge into a fixed-size buffer. Domain-separated by
- /// a constant label so the same secret cannot be misused to forge a
- /// different protocol's MAC.
+ /// Serialize the challenge into a fixed-size buffer. The leading 16 B
+ /// are the plane-tagged label, so the same secret cannot be misused
+ /// to forge a different plane's MAC even with otherwise-identical
+ /// inputs.
fn encode(&self) -> [u8; 16 + 16 + 16 + 8 + 4 + 16] {
let mut out = [0u8; 76];
- out[0..16].copy_from_slice(LABEL);
+ out[0..16].copy_from_slice(self.label);
out[16..32].copy_from_slice(&self.cluster.to_le_bytes());
out[32..48].copy_from_slice(&self.peer_id.to_le_bytes());
out[48..56].copy_from_slice(&self.timestamp_ns.to_le_bytes());
@@ -102,11 +132,15 @@ impl AuthChallenge {
/// MAC over an `AuthChallenge`. Wraps `blake3::Hash` for its
/// constant-time `PartialEq` implementation.
+///
+/// The inner `blake3::Hash` is `pub(crate)` so downstream callers cannot
+/// reach in for byte-level `==` comparisons that would bypass the
+/// constant-time `PartialEq` on the wrapper.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct AuthTag(pub blake3::Hash);
+pub struct AuthTag(pub(crate) blake3::Hash);
impl AuthTag {
- const fn as_bytes(&self) -> &[u8; 32] {
+ pub(crate) const fn as_bytes(&self) -> &[u8; 32] {
self.0.as_bytes()
}
}
@@ -130,15 +164,25 @@ pub enum AuthError {
/// Map a credential-layer `AuthError` to the public `IggyError` surface.
///
+/// `UnsupportedKind` is routed to `Unauthenticated` rather than
+/// `InvalidCommand` so that pre-auth peers in a mixed-version cluster
+/// (whose all-zero `reserved_command` decodes to `AuthKind::from_byte(0)
+/// -> UnsupportedKind`) surface in the operator's auth-bucket triage,
+/// matching the actual deployment failure mode. `ReservedNonzero` keeps
+/// the `InvalidCommand` mapping because the peer DID stamp a recognized
+/// kind byte and then violated the envelope's reserved-zero contract:
+/// that is a genuine protocol bug, not an auth gap.
+///
/// Not implemented as `From<AuthError> for IggyError` because the impl
/// would leak into every crate transitively linking `message_bus`, adding
/// an extra `impl From<_> for IggyError` candidate to the orphan-rule set
/// and tripping type inference (`?` ambiguity) in downstream crates.
#[must_use]
-pub const fn to_iggy_error(e: AuthError) -> IggyError {
+pub(crate) const fn to_iggy_error(e: AuthError) -> IggyError {
match e {
- AuthError::UnsupportedKind | AuthError::ReservedNonzero => IggyError::InvalidCommand,
- AuthError::TagMismatch
+ AuthError::ReservedNonzero => IggyError::InvalidCommand,
+ AuthError::UnsupportedKind
+ | AuthError::TagMismatch
| AuthError::UnknownPeer
| AuthError::TimestampOutOfWindow
| AuthError::NonceReplay => IggyError::Unauthenticated,
@@ -193,51 +237,65 @@ impl TokenSource for StaticSharedSecret {
}
}
-/// Fixed-capacity ring-buffer nonce dedup.
+/// Fixed-capacity ring-buffer nonce dedup. Inline `[u128; CAP]` storage —
+/// no heap allocation, deterministic memory footprint.
///
-/// Eviction is O(1) via a round-robin cursor; lookup is O(N) over
-/// `NONCE_LRU_CAPACITY` entries, which is a 256-wide linear scan on the
-/// hot path. Memory footprint is deterministic and bounded regardless of
-/// peer behaviour.
+/// Eviction is O(1) via a round-robin cursor; lookup is O(N) over `CAP`
+/// entries (linear scan, but `CAP * 16 B` fits a single cache line for
+/// `CAP <= 4` and stays L1-resident through `CAP = 256`). Memory bound
+/// is `CAP * 16 B + 2 * sizeof(usize)` per ring; per-peer instances cap
+/// the cluster-wide cost.
+#[doc(hidden)]
#[derive(Debug)]
-pub struct NonceRing {
- slots: Vec<u128>,
+pub struct NonceRing<const CAP: usize> {
+ slots: [u128; CAP],
+ len: usize,
cursor: usize,
- capacity: usize,
}
-impl NonceRing {
+impl<const CAP: usize> NonceRing<CAP> {
#[must_use]
- pub const fn new(capacity: usize) -> Self {
+ #[doc(hidden)]
+ pub const fn new() -> Self {
Self {
- slots: Vec::new(),
+ slots: [0u128; CAP],
+ len: 0,
cursor: 0,
- capacity,
}
}
/// Returns `true` if `nonce` was not present (and has now been
/// recorded); `false` if it was already seen.
+ ///
+ /// `slots[..self.len]` is the only range scanned for membership, so
+ /// the `[0; CAP]` initial state never collides with a real `0`
+ /// nonce: until the ring fills, untouched slots are out-of-range.
+ #[doc(hidden)]
pub fn insert(&mut self, nonce: u128) -> bool {
- if self.slots.contains(&nonce) {
+ if self.slots[..self.len].contains(&nonce) {
return false;
}
- if self.slots.len() < self.capacity {
- self.slots.push(nonce);
+ if self.len < CAP {
+ self.slots[self.len] = nonce;
+ self.len += 1;
} else {
self.slots[self.cursor] = nonce;
- self.cursor = (self.cursor + 1) % self.capacity;
+ self.cursor = (self.cursor + 1) % CAP;
}
true
}
}
-impl Default for NonceRing {
+impl<const CAP: usize> Default for NonceRing<CAP> {
fn default() -> Self {
- Self::new(NONCE_LRU_CAPACITY)
+ Self::new()
}
}
+/// Per-peer nonce ring with the production [`PER_PEER_NONCE_CAPACITY`].
+#[doc(hidden)]
+pub type ReplicaNonceRing = NonceRing<PER_PEER_NONCE_CAPACITY>;
+
/// Encode an auth envelope into `reserved_command`.
///
/// Overwrites the first `ENVELOPE_LEN` bytes and zero-fills the trailing
@@ -245,7 +303,7 @@ impl Default for NonceRing {
/// and `nonce` are caller-controlled so the same routine can be used for
/// deterministic tests (fake clock, fixed nonce) and production (system
/// clock, CSPRNG).
-pub fn encode_envelope(
+pub(crate) fn encode_envelope(
reserved: &mut [u8; 128],
token_source: &dyn TokenSource,
challenge: &AuthChallenge,
@@ -262,10 +320,10 @@ pub fn encode_envelope(
/// Parsed view of an envelope's raw bytes. No cryptographic checks.
#[derive(Debug, Clone, Copy)]
-pub struct DecodedEnvelope {
- pub tag: AuthTag,
- pub timestamp_ns: u64,
- pub nonce: u128,
+pub(crate) struct DecodedEnvelope {
+ pub(crate) tag: AuthTag,
+ pub(crate) timestamp_ns: u64,
+ pub(crate) nonce: u128,
}
/// Decode an envelope and reject malformed layout.
@@ -278,7 +336,7 @@ pub struct DecodedEnvelope {
///
/// Returns `AuthError::UnsupportedKind` for unknown envelope versions or
/// `AuthError::ReservedNonzero` for non-zero reserved padding.
-pub fn decode_envelope(reserved: &[u8; 128]) -> Result<DecodedEnvelope, AuthError> {
+pub(crate) fn decode_envelope(reserved: &[u8; 128]) -> Result<DecodedEnvelope, AuthError> {
AuthKind::from_byte(reserved[KIND_OFFSET])?;
if reserved[RESERVED_RANGE].iter().any(|&b| b != 0) {
return Err(AuthError::ReservedNonzero);
@@ -296,27 +354,34 @@ pub fn decode_envelope(reserved: &[u8; 128]) -> Result<DecodedEnvelope, AuthErro
})
}
-/// Full credential check: timestamp window -> nonce dedup -> tag.
+/// Full credential check: timestamp window -> tag verify -> nonce dedup.
+///
+/// **Order is load-bearing.** A tag-failed envelope must NOT consume a
+/// nonce-ring slot, otherwise an off-secret attacker with TCP reachability
+/// to the listener could flush the ring with forged-tag Pings (valid
+/// timestamps + random nonces) and open a replay window for any captured
+/// legit Ping inside the remaining timestamp slack.
///
/// # Errors
///
/// Returns the first `AuthError` triggered by the check sequence.
-pub fn verify_envelope(
+pub(crate) fn verify_envelope<const CAP: usize>(
token_source: &dyn TokenSource,
challenge: &AuthChallenge,
decoded: &DecodedEnvelope,
now_ns: u128,
- nonces: &mut NonceRing,
+ nonces: &mut NonceRing<CAP>,
) -> Result<(), AuthError> {
let ts_u128 = u128::from(decoded.timestamp_ns);
let delta = now_ns.abs_diff(ts_u128);
if delta > TIMESTAMP_WINDOW_NS {
return Err(AuthError::TimestampOutOfWindow);
}
+ token_source.verify(challenge, &decoded.tag)?;
if !nonces.insert(decoded.nonce) {
return Err(AuthError::NonceReplay);
}
- token_source.verify(challenge, &decoded.tag)
+ Ok(())
}
#[cfg(test)]
@@ -330,6 +395,7 @@ mod tests {
timestamp_ns: 1_700_000_000_000_000_000,
release: 1,
nonce: 0x11_2233_4455_6677_8899_aabb_ccdd_eeff,
+ label: LABEL_REPLICA,
}
}
@@ -341,7 +407,7 @@ mod tests {
encode_envelope(&mut reserved, &source, &challenge);
let decoded = decode_envelope(&reserved).expect("decode");
- let mut ring = NonceRing::new(32);
+ let mut ring = NonceRing::<32>::new();
verify_envelope(
&source,
&challenge,
@@ -361,7 +427,7 @@ mod tests {
encode_envelope(&mut reserved, &signer, &challenge);
let decoded = decode_envelope(&reserved).unwrap();
- let mut ring = NonceRing::new(32);
+ let mut ring = NonceRing::<32>::new();
let err = verify_envelope(
&verifier,
&challenge,
@@ -381,7 +447,7 @@ mod tests {
encode_envelope(&mut reserved, &source, &challenge);
let decoded = decode_envelope(&reserved).unwrap();
- let mut ring = NonceRing::new(32);
+ let mut ring = NonceRing::<32>::new();
let now = u128::from(challenge.timestamp_ns);
verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap();
let err = verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap_err();
@@ -396,7 +462,7 @@ mod tests {
encode_envelope(&mut reserved, &source, &challenge);
let decoded = decode_envelope(&reserved).unwrap();
- let mut ring = NonceRing::new(32);
+ let mut ring = NonceRing::<32>::new();
let now = u128::from(challenge.timestamp_ns) + TIMESTAMP_WINDOW_NS + 1;
let err = verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap_err();
assert_eq!(err, AuthError::TimestampOutOfWindow);
@@ -410,7 +476,7 @@ mod tests {
encode_envelope(&mut reserved, &source, &challenge);
let decoded = decode_envelope(&reserved).unwrap();
- let mut ring = NonceRing::new(32);
+ let mut ring = NonceRing::<32>::new();
let now = u128::from(challenge.timestamp_ns).saturating_sub(TIMESTAMP_WINDOW_NS + 1);
let err = verify_envelope(&source, &challenge, &decoded, now, &mut ring).unwrap_err();
assert_eq!(err, AuthError::TimestampOutOfWindow);
@@ -438,9 +504,93 @@ mod tests {
assert_eq!(err, AuthError::ReservedNonzero);
}
+ #[test]
+ fn forged_tag_does_not_consume_ring_slot() {
+ // Sender signs with secret A; verifier holds secret B. Verifier's
+ // tag check fails. The ring MUST NOT have recorded the nonce -
+ // otherwise an attacker without the secret could flood forged
+ // Pings to evict honest entries (auth.rs verify_envelope ordering
+ // bug; review F1 critical).
+ let signer = StaticSharedSecret::new([1u8; SECRET_SIZE]);
+ let verifier = StaticSharedSecret::new([2u8; SECRET_SIZE]);
+ let challenge = sample_challenge();
+ let mut reserved = [0u8; 128];
+ encode_envelope(&mut reserved, &signer, &challenge);
+
+ let decoded = decode_envelope(&reserved).unwrap();
+ let mut ring = NonceRing::<4>::new();
+ let err = verify_envelope(
+ &verifier,
+ &challenge,
+ &decoded,
+ u128::from(challenge.timestamp_ns),
+ &mut ring,
+ )
+ .unwrap_err();
+ assert_eq!(err, AuthError::TagMismatch);
+
+ // Same nonce now signed correctly with secret B must succeed -
+ // proves the failed-tag attempt did NOT burn the ring slot.
+ let mut reserved_legit = [0u8; 128];
+ encode_envelope(&mut reserved_legit, &verifier, &challenge);
+ let decoded_legit = decode_envelope(&reserved_legit).unwrap();
+ verify_envelope(
+ &verifier,
+ &challenge,
+ &decoded_legit,
+ u128::from(challenge.timestamp_ns),
+ &mut ring,
+ )
+ .expect("legit verify must succeed; nonce slot must have been free");
+ }
+
+ #[test]
+ fn cross_plane_mac_does_not_verify() {
+ // Same secret, same numeric inputs - only label differs. Replica-
+ // plane signer must NOT pass client-plane verification (and vice
+ // versa). Otherwise a captured replica handshake could be replayed
+ // onto the client plane once Phase 2 ships. Review F4.
+ let secret = StaticSharedSecret::new([3u8; SECRET_SIZE]);
+ let replica_challenge = sample_challenge();
+ let mut client_challenge = replica_challenge;
+ client_challenge.label = LABEL_CLIENT;
+
+ let mut replica_envelope = [0u8; 128];
+ encode_envelope(&mut replica_envelope, &secret, &replica_challenge);
+ let decoded = decode_envelope(&replica_envelope).unwrap();
+ let mut ring = NonceRing::<4>::new();
+ let err = verify_envelope(
+ &secret,
+ &client_challenge,
+ &decoded,
+ u128::from(replica_challenge.timestamp_ns),
+ &mut ring,
+ )
+ .unwrap_err();
+ assert_eq!(err, AuthError::TagMismatch);
+ }
+
+ #[test]
+ fn unsupported_kind_maps_to_unauthenticated() {
+ // Mixed-version cluster: pre-auth peer sends zeroed
+ // reserved_command -> AuthKind::from_byte(0) -> UnsupportedKind.
+ // Operator log triage must hit the auth bucket, not the
+ // protocol-violation bucket. Review F3.
+ assert_eq!(
+ to_iggy_error(AuthError::UnsupportedKind),
+ IggyError::Unauthenticated
+ );
+ // ReservedNonzero stays InvalidCommand: peer DID stamp a known
+ // kind, then violated the envelope's reserved-zero contract.
+ assert_eq!(
+ to_iggy_error(AuthError::ReservedNonzero),
+ IggyError::InvalidCommand
+ );
+ }
+
#[test]
fn nonce_ring_fills_and_evicts_fifo() {
- let mut ring = NonceRing::new(3);
+ let mut ring = NonceRing::<3>::new();
assert!(ring.insert(1));
assert!(ring.insert(2));
assert!(ring.insert(3));
@@ -449,4 +599,22 @@ mod tests {
// Nonce 1 evicted; reinserting it should now succeed again.
assert!(ring.insert(1));
}
+
+ #[test]
+ fn nonce_ring_zero_initial_does_not_false_positive() {
+ // [0; CAP] initial state must NOT match a real `0` nonce until
+ // the slot has been written. Edge case for the const-generic
+ // refactor (review F2).
+ let mut ring = NonceRing::<4>::new();
+ assert!(ring.insert(0));
+ assert!(!ring.insert(0));
+ assert!(ring.insert(1));
+ assert!(ring.insert(2));
+ }
+
+ #[test]
+ fn nonce_ring_default_is_empty() {
+ let mut ring = NonceRing::<4>::default();
+ assert!(ring.insert(42));
+ }
}
diff --git a/core/message_bus/src/auth_config.rs b/core/message_bus/src/auth_config.rs
index af715c6d3..dfba8bb89 100644
--- a/core/message_bus/src/auth_config.rs
+++ b/core/message_bus/src/auth_config.rs
@@ -66,8 +66,8 @@ pub enum SecretLoadError {
BothSet,
#[error("secret has {got} hex characters; expected {HEX_SECRET_LEN}")]
WrongLength { got: usize },
- #[error("secret contains non-hex character at byte offset {0}")]
- InvalidHex(usize),
+ #[error("secret contains non-hex character {byte:#04x} at byte offset {offset}")]
+ InvalidHex { offset: usize, byte: u8 },
#[error("failed to read secret file {path}")]
FileRead {
path: String,
@@ -135,8 +135,14 @@ fn decode_hex_secret(input: &str) -> Result<[u8; SECRET_SIZE], SecretLoadError>
let bytes = input.as_bytes();
let mut out = [0u8; SECRET_SIZE];
for (i, chunk) in bytes.chunks_exact(2).enumerate() {
- let hi = decode_hex_nibble(chunk[0]).ok_or(SecretLoadError::InvalidHex(i * 2))?;
- let lo = decode_hex_nibble(chunk[1]).ok_or(SecretLoadError::InvalidHex(i * 2 + 1))?;
+ let hi = decode_hex_nibble(chunk[0]).ok_or(SecretLoadError::InvalidHex {
+ offset: i * 2,
+ byte: chunk[0],
+ })?;
+ let lo = decode_hex_nibble(chunk[1]).ok_or(SecretLoadError::InvalidHex {
+ offset: i * 2 + 1,
+ byte: chunk[1],
+ })?;
out[i] = (hi << 4) | lo;
}
Ok(out)
@@ -211,7 +217,10 @@ mod tests {
bad.replace_range(4..5, "Z");
let err = resolve(Some(&bad), None).unwrap_err();
match err {
- SecretLoadError::InvalidHex(offset) => assert_eq!(offset, 4),
+ SecretLoadError::InvalidHex { offset, byte } => {
+ assert_eq!(offset, 4);
+ assert_eq!(byte, b'Z');
+ }
other => panic!("unexpected error: {other:?}"),
}
}
@@ -260,6 +269,7 @@ mod tests {
timestamp_ns: 3,
release: 4,
nonce: 5,
+ label: crate::auth::LABEL_REPLICA,
};
let from_loader = source.sign(&challenge);
let from_reference = reference.sign(&challenge);
diff --git a/core/message_bus/src/config.rs b/core/message_bus/src/config.rs
index f82114461..ea50a817d 100644
--- a/core/message_bus/src/config.rs
+++ b/core/message_bus/src/config.rs
@@ -35,6 +35,17 @@
use std::time::Duration;
+/// Hard upper bound on `max_batch`, in iovecs.
+///
+/// Linux's `IOV_MAX` is 1024 (`/usr/include/bits/uio_lim.h`). Future WS
+/// transports emit one iovec for the header and one for the body, so a
+/// batch of N messages costs `2 * N` iovecs; cap `max_batch` at
+/// `IOV_MAX / 2 = 512` to keep that worst case below the syscall limit.
+/// Bus construction asserts this in [`crate::IggyMessageBus::with_config`];
+/// breaching it at boot panics rather than silently delivering writev
+/// `EMSGSIZE` errors on every batch.
+pub const IOV_MAX_LIMIT: usize = 512;
+
/// Aggregated runtime configuration for an `IggyMessageBus` instance.
///
/// All fields map onto constants that previously lived inline across
diff --git a/core/message_bus/src/connector.rs b/core/message_bus/src/connector.rs
index 01d18a255..a3b5a1c54 100644
--- a/core/message_bus/src/connector.rs
+++ b/core/message_bus/src/connector.rs
@@ -25,7 +25,7 @@
//! (see `shard::coordinator::ShardZeroCoordinator`).
use crate::IggyMessageBus;
-use crate::auth::{self, TokenSource};
+use crate::auth::{self, LABEL_REPLICA, TokenSource};
use crate::framing;
use crate::lifecycle::ShutdownToken;
use crate::socket_opts::apply_keepalive_for_connection;
@@ -102,9 +102,11 @@ async fn connect_all(
// connection lives on shard 0; `owning_shard` covers multi-shard
// deployments where the fd was delegated to a peer shard but the
// mapping broadcast reached shard 0. Either hit means a previous
- // sweep (or the inbound listener) already installed this peer and
- // we must not dial again - redialing would tear down the live
- // socket via the `AlreadyRegistered` race and flap the mapping.
+ // sweep (or the inbound listener) already installed this peer.
+ // Redialing wastes a TCP round-trip; the live entry stays intact
+ // (the loser of the registry insert race in `install_replica_conn`
+ // gets `RejectedRegistration` back and tears DOWN ITS OWN orphan
+ // tasks via `drain_rejected_registration`, never the winner's).
if bus.replicas().contains(peer_id) || bus.owning_shard(peer_id).is_some() {
debug!(
replica = peer_id,
@@ -197,8 +199,13 @@ fn build_ping_message(
replica_id: u8,
token_source: &dyn TokenSource,
) -> Message<GenericHeader> {
+ let now_ns = IggyTimestamp::now().as_nanos();
+ debug_assert!(
+ now_ns <= u128::from(u64::MAX),
+ "auth_timestamp overflows u64 ns (year ~2554) - envelope TIMESTAMP_RANGE is 8 bytes by design"
+ );
#[allow(clippy::cast_possible_truncation)]
- let timestamp_ns = IggyTimestamp::now().as_nanos() as u64;
+ let timestamp_ns = now_ns as u64;
let nonce = {
let mut buf = [0u8; 16];
rand::rng().fill_bytes(&mut buf);
@@ -218,6 +225,7 @@ fn build_ping_message(
timestamp_ns,
release: h.release,
nonce,
+ label: LABEL_REPLICA,
};
auth::encode_envelope(&mut h.reserved_command, token_source, &challenge);
},
diff --git a/core/message_bus/src/framing.rs b/core/message_bus/src/framing.rs
index 391a86aa2..0f3af8576 100644
--- a/core/message_bus/src/framing.rs
+++ b/core/message_bus/src/framing.rs
@@ -28,6 +28,7 @@ use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
use iggy_binary_protocol::consensus::iobuf::Owned;
use iggy_binary_protocol::{GenericHeader, HEADER_SIZE, Message};
use iggy_common::IggyError;
+use tracing::error;
/// Default hard ceiling on a single wire frame. Frames above this are
/// almost certainly a protocol violation or a malicious peer; we drop
@@ -126,7 +127,17 @@ pub async fn read_message<S: AsyncReadExt + Unpin>(
// (e.g. capacity overflow, unsupported buffer kind) and silently
// ignoring it would leave `owned` at header-only capacity, causing the
// subsequent `read_exact` to read into an ungrown buffer.
- IoBufMut::reserve_exact(&mut owned, body_size).map_err(|_| IggyError::TcpError)?;
+ //
+ // No `IggyError::OutOfMemory` variant exists; `TcpError` is the closest
+ // bucket. The `error!` log gives operators the actionable diagnostic
+ // (size, OOM-vs-overflow signal) the error code itself loses.
+ IoBufMut::reserve_exact(&mut owned, body_size).map_err(|_| {
+ error!(
+ body_size,
+ "framing reserve_exact failed (OOM or capacity overflow); dropping connection"
+ );
+ IggyError::TcpError
+ })?;
let BufResult(result, slice) = stream
.read_exact(owned.slice(HEADER_SIZE..total_size))
.await;
diff --git a/core/message_bus/src/installer.rs b/core/message_bus/src/installer.rs
index 21d381296..4d274860c 100644
--- a/core/message_bus/src/installer.rs
+++ b/core/message_bus/src/installer.rs
@@ -157,9 +157,11 @@ pub fn install_replica_conn<C: TransportConn>(
let notified = Rc::new(Cell::new(false));
// If the registry insert below races with a concurrent install for
// the same peer id and loses, both spawned halves must skip their
- // post-loop cleanup: calling `replicas().remove` / `close_peer` or
- // `notify_connection_lost` would evict the winner's entry and
- // clobber its mapping broadcast. `compio::runtime::JoinHandle::drop`
+ // post-loop cleanup: the loser's `replicas().remove` /
+ // `close_peer_if_token_matches` calls would no-op against the winner's
+ // generation token (so they can't evict the live entry), but
+ // `notify_connection_lost` has no token guard and would still broadcast
+ // a spurious mapping-clear round. `compio::runtime::JoinHandle::drop`
// does not cancel the spawned task, so we have to tell the tasks to
// stand down in-band.
let install_aborted = Rc::new(Cell::new(false));
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index 96d87d961..811102e51 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -40,6 +40,13 @@
//! timestamp, 16 B nonce, 1 B kind — so the 256 B header layout is
//! unchanged.
//!
+//! Each plane prepends a distinct 16-byte LABEL into the MAC input
+//! (`auth::LABEL_REPLICA` for the consensus plane;
+//! `auth::LABEL_CLIENT` reserved for Phase 2). The label is part of
+//! the cryptographic challenge, so a captured handshake on one plane
+//! cannot be replayed onto the other even if both share one
+//! `StaticSharedSecret`.
+//!
//! # Invariants worth naming
//!
//! - [`send_to_client`](IggyMessageBus::send_to_client) and
@@ -87,7 +94,7 @@ pub(crate) mod socket_opts;
pub mod transports;
pub mod writer_task;
-pub use config::MessageBusConfig;
+pub use config::{IOV_MAX_LIMIT, MessageBusConfig};
pub use error::SendError;
pub use installer::ConnectionInstaller;
pub use lifecycle::{
@@ -95,14 +102,31 @@ pub use lifecycle::{
ReplicaRegistry, Shutdown, ShutdownToken,
};
+use auth::ReplicaNonceRing;
use compio::runtime::JoinHandle;
use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_binary_protocol::{GenericHeader, Message};
use std::cell::RefCell;
use std::collections::HashMap;
+use std::rc::Rc;
use std::time::Duration;
+/// Per-replica nonce dedup store, owned by [`IggyMessageBus`] and
+/// shared with the listener task via [`Rc`]. Keyed by `replica_id` so
+/// each peer's ring is isolated; a flapping or hostile peer cannot
+/// evict honest peers' nonce slots. Bus-resident (rather than
+/// task-local on the listener) so the nonce window survives a listener
+/// task respawn within one process. Process restart still drops the
+/// store; cross-process persistence is out of scope.
+///
+/// Type is `pub` (rather than `pub(crate)`) only because it appears in
+/// the public signature of [`replica_listener::run`]; downstream code
+/// should obtain a clone via [`IggyMessageBus::replica_nonces`] rather
+/// than constructing one directly.
+#[doc(hidden)]
+pub type ReplicaNonceStore = RefCell<HashMap<u8, ReplicaNonceRing>>;
+
/// Callback for forwarding a consensus message to a remote shard.
///
/// Provided by the shard layer at bus construction and installed via
@@ -194,7 +218,10 @@ pub trait MessageBus {
/// - a [`ConnectionRegistry<u128>`] for clients (`u128` id is shard-packed),
/// - a [`ReplicaRegistry`] for replicas (`u8` id from the Ping handshake,
/// backed by a fixed-size array to avoid hash lookup on every send),
-/// - the `JoinHandle`s of background tasks (accept loops, reconnect periodic).
+/// - the `JoinHandle`s of background tasks (accept loops, reconnect periodic),
+/// - a per-peer [`ReplicaNonceStore`] used by the replica-plane handshake
+/// to dedup nonces inside the 30 s timestamp window (see
+/// [`Self::replica_nonces`]).
///
/// Send semantics:
/// - `send_to_*` clones the per-peer `Sender` out of the registry, calls
@@ -229,6 +256,10 @@ pub struct IggyMessageBus {
/// when they exit abnormally. `None` when running without a shard-0
/// coordinator (single-shard deployments and tests).
connection_lost_fn: RefCell<Option<ConnectionLostFn>>,
+ /// Per-peer replica handshake nonce dedup store. Held in an `Rc` so
+ /// the listener task can take a clone across the spawn boundary
+ /// without stealing the bus's only handle. See [`ReplicaNonceStore`].
+ replica_nonces: Rc<ReplicaNonceStore>,
}
impl IggyMessageBus {
@@ -253,8 +284,20 @@ impl IggyMessageBus {
}
/// Construct a bus with the given [`MessageBusConfig`].
+ ///
+ /// # Panics
+ ///
+ /// Panics if `config.max_batch == 0` or
+ /// `config.max_batch > IOV_MAX_LIMIT`. Boot-time validation; surfaces
+ /// operator misconfiguration loudly rather than letting every writev
+ /// fail silently with `EMSGSIZE` once traffic starts.
#[must_use]
pub fn with_config(shard_id: u16, config: MessageBusConfig) -> Self {
+ assert!(
+ config.max_batch > 0 && config.max_batch <= IOV_MAX_LIMIT,
+ "MessageBusConfig::max_batch must be in 1..={IOV_MAX_LIMIT} (writev IOV_MAX/2 cap); got {}",
+ config.max_batch,
+ );
let (shutdown, token) = Shutdown::new();
Self {
shard_id,
@@ -268,9 +311,24 @@ impl IggyMessageBus {
replica_forward_fn: RefCell::new(None),
client_forward_fn: RefCell::new(None),
connection_lost_fn: RefCell::new(None),
+ replica_nonces: Rc::new(RefCell::new(HashMap::new())),
}
}
+ /// Cheap clone of the bus-resident replica-plane nonce store.
+ ///
+ /// The listener task takes one clone at boot and reaches into it
+ /// from `handshake_verify`; the `Rc` keeps the store alive across
+ /// the listener task's lifetime and survives a respawn within one
+ /// process. Tests that drive `replica_listener::run` directly use
+ /// this accessor to obtain the same store the production bootstrap
+ /// would.
+ #[doc(hidden)]
+ #[must_use]
+ pub fn replica_nonces(&self) -> Rc<ReplicaNonceStore> {
+ Rc::clone(&self.replica_nonces)
+ }
+
/// Install the notifier used by delegated replica connections to tell
/// shard 0 that a connection died. Single place to inject in tests too.
pub fn set_connection_lost_fn(&self, f: ConnectionLostFn) {
@@ -278,8 +336,18 @@ impl IggyMessageBus {
}
/// Invoke the registered connection-lost notifier, if any.
+ ///
+ /// Clones the `Rc` out of the `RefCell` borrow before invoking the
+ /// closure so the closure body is free to call
+ /// [`Self::set_connection_lost_fn`] (which takes a `borrow_mut`)
+ /// without tripping the runtime borrow check.
pub(crate) fn notify_connection_lost(&self, replica_id: u8) {
- if let Some(f) = self.connection_lost_fn.borrow().as_ref() {
+ let cb = self
+ .connection_lost_fn
+ .borrow()
+ .as_ref()
+ .map(std::rc::Rc::clone);
+ if let Some(f) = cb {
f(replica_id);
}
}
@@ -688,6 +756,96 @@ mod tests {
assert_eq!(bus.background_tasks.borrow().len(), 2);
}
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn replica_nonce_store_isolates_peers() {
+ // Per-peer rings: two peers replaying the same nonce must both
+ // be admitted (different scopes), and a same-peer replay must
+ // be rejected. Review F2.
+ let bus = IggyMessageBus::new(0);
+ let nonces = bus.replica_nonces();
+ let mut store = nonces.borrow_mut();
+ assert!(store.entry(1).or_default().insert(0xdead_beef));
+ assert!(
+ store.entry(2).or_default().insert(0xdead_beef),
+ "different peer with same nonce must not be a replay"
+ );
+ assert!(
+ !store.entry(1).or_default().insert(0xdead_beef),
+ "same peer + same nonce is a replay"
+ );
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn replica_nonce_store_survives_handle_drop() {
+ // Listener task drop must NOT clear the nonce window; the store
+ // is `Rc`-shared and the bus retains a clone. Proves the
+ // restart-survival property the per-task `RefCell<NonceRing>`
+ // could not deliver. Review F2.
+ let bus = IggyMessageBus::new(0);
+ let nonces1 = bus.replica_nonces();
+ {
+ let mut store = nonces1.borrow_mut();
+ assert!(store.entry(1).or_default().insert(0xfeed_face));
+ }
+ drop(nonces1); // simulates listener task exit
+
+ let nonces2 = bus.replica_nonces();
+ let mut store = nonces2.borrow_mut();
+ assert!(
+ !store.entry(1).or_default().insert(0xfeed_face),
+ "post-restart replay must be rejected"
+ );
+ }
+
+ #[test]
+ #[should_panic(expected = "MessageBusConfig::max_batch must be in")]
+ fn max_batch_oversize_rejected() {
+ let cfg = MessageBusConfig {
+ max_batch: 4096,
+ ..MessageBusConfig::default()
+ };
+ let _ = IggyMessageBus::with_config(0, cfg);
+ }
+
+ #[test]
+ #[should_panic(expected = "MessageBusConfig::max_batch must be in")]
+ fn max_batch_zero_rejected() {
+ let cfg = MessageBusConfig {
+ max_batch: 0,
+ ..MessageBusConfig::default()
+ };
+ let _ = IggyMessageBus::with_config(0, cfg);
+ }
+
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn notify_connection_lost_handles_reentrant_install() {
+ // Closure swaps itself out via `set_connection_lost_fn`, which
+ // calls `borrow_mut`. The pre-fix code held a `Ref` across the
+ // closure invocation and panicked here.
+ let bus = std::rc::Rc::new(IggyMessageBus::new(0));
+ let bus_for_closure = bus.clone();
+ let counter: std::rc::Rc<std::cell::Cell<u8>> = std::rc::Rc::new(std::cell::Cell::new(0));
+ let counter_inner = counter.clone();
+ let cb: ConnectionLostFn = std::rc::Rc::new(move |_replica: u8| {
+ counter_inner.set(counter_inner.get() + 1);
+ // Reentrant install: replace the closure mid-invoke.
+ let counter_replacement = counter_inner.clone();
+ bus_for_closure.set_connection_lost_fn(std::rc::Rc::new(move |_| {
+ counter_replacement.set(counter_replacement.get() + 10);
+ }));
+ });
+ bus.set_connection_lost_fn(cb);
+
+ bus.notify_connection_lost(1); // first closure runs (+1) and swaps
+ assert_eq!(counter.get(), 1);
+
+ bus.notify_connection_lost(1); // second closure runs (+10)
+ assert_eq!(counter.get(), 11);
+ }
+
#[compio::test]
#[allow(clippy::future_not_send)]
async fn shutdown_loop_drains_tasks_added_during_shutdown() {
diff --git a/core/message_bus/src/replica_io.rs b/core/message_bus/src/replica_io.rs
index 3731cc6d8..cc2eca8c1 100644
--- a/core/message_bus/src/replica_io.rs
+++ b/core/message_bus/src/replica_io.rs
@@ -85,6 +85,7 @@ pub async fn start_on_shard_zero(
let on_accepted_replica_for_listener = on_accepted_replica.clone();
let listener_max_message_size = bus.config().max_message_size;
let token_source_for_listener = Rc::clone(&token_source);
+ let nonces_for_listener = bus.replica_nonces();
let replica_handle = compio::runtime::spawn(async move {
run_replica_listener(
replica_listener,
@@ -95,6 +96,7 @@ pub async fn start_on_shard_zero(
on_accepted_replica_for_listener,
listener_max_message_size,
token_source_for_listener,
+ nonces_for_listener,
)
.await;
});
diff --git a/core/message_bus/src/replica_listener.rs b/core/message_bus/src/replica_listener.rs
index b056f30ea..6a0789369 100644
--- a/core/message_bus/src/replica_listener.rs
+++ b/core/message_bus/src/replica_listener.rs
@@ -50,16 +50,35 @@
//! NOT provided here: operators still need to deploy the replica port on a
//! trusted network boundary (cluster-local VPC, private subnet, overlay)
//! if wire confidentiality is required.
+//!
+//! # Trust model
+//!
+//! The cluster ships with one shared 32 B secret across all replicas
+//! (see [`crate::auth_config`]). Any holder of the secret can mint a
+//! valid MAC for ANY `peer_id` in either direction. The dialer-side
+//! directional rule (`peer_id > self_id` only dials) plus the
+//! acceptor-side check (`parsed.replica < self_id`) blocks UPWARD
+//! impersonation - a low-id peer cannot pretend to be a high-id peer
+//! during dial - but cross-impersonation in the OTHER direction is open
+//! by design under a single shared secret. A compromised replica B can
+//! authenticate as any replica A with `A < B`. Mitigation comes from
+//! the trusted-LAN deployment assumption + the fact that the
+//! authenticated identity is only used for VSR consensus framing on the
+//! same plane.
+//!
+//! Per-peer secrets (each replica gets its own key, the acceptor looks
+//! up the right one for the announced `peer_id`) is the documented
+//! Phase 2+ hardening. Until then, the cluster operator must treat
+//! "secret compromise" as equivalent to "any-replica compromise."
-use crate::auth::{self, NonceRing, TokenSource};
+use crate::auth::{self, LABEL_REPLICA, TokenSource};
use crate::framing;
use crate::lifecycle::ShutdownToken;
-use crate::{AcceptedReplicaFn, GenericHeader, Message};
+use crate::{AcceptedReplicaFn, GenericHeader, Message, ReplicaNonceStore};
use compio::net::{SocketOpts, TcpListener, TcpStream};
use futures::FutureExt;
use iggy_binary_protocol::Command2;
use iggy_common::{IggyError, IggyTimestamp};
-use std::cell::RefCell;
use std::net::SocketAddr;
use std::rc::Rc;
use tracing::{debug, error, info, warn};
@@ -94,6 +113,12 @@ pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyErr
/// Run the inbound replica listener accept loop until the shutdown token
/// fires. Every successful handshake fires the `on_accepted` callback; the
/// callback owns the accepted stream from that point on.
+///
+/// `nonces` is the bus-resident per-peer nonce dedup store. The listener
+/// reaches into it via `borrow_mut().entry(peer).or_default()` for each
+/// handshake; the borrow is held only across the synchronous
+/// [`handshake_verify`], never across an `await`, so the `RefCell` is
+/// safe under the single-threaded compio runtime.
#[allow(clippy::future_not_send)]
#[allow(clippy::too_many_arguments)]
pub async fn run(
@@ -105,12 +130,12 @@ pub async fn run(
on_accepted: AcceptedReplicaFn,
max_message_size: usize,
token_source: Rc<dyn TokenSource>,
+ nonces: Rc<ReplicaNonceStore>,
) {
info!(
"Replica listener accepting on {:?}",
listener.local_addr().ok()
);
- let nonces = RefCell::new(NonceRing::default());
loop {
futures::select! {
() = token.wait().fuse() => {
@@ -124,11 +149,12 @@ pub async fn run(
let outcome = match read {
Ok(parsed) => {
let now_ns = IggyTimestamp::now().as_nanos();
- let mut ring = nonces.borrow_mut();
+ let mut store = nonces.borrow_mut();
+ let ring = store.entry(parsed.replica).or_default();
handshake_verify(
&parsed,
token_source.as_ref(),
- &mut ring,
+ ring,
now_ns,
self_id,
replica_count,
@@ -186,6 +212,7 @@ async fn handshake_read(
timestamp_ns: decoded.timestamp_ns,
release: header.release,
nonce: decoded.nonce,
+ label: LABEL_REPLICA,
};
Ok(ParsedHandshake {
replica: header.replica,
@@ -195,11 +222,12 @@ async fn handshake_read(
}
/// Synchronous verifier: tag + replay, then the directional tiebreak.
-/// Runs outside any await so the `&mut NonceRing` borrow cannot span I/O.
+/// Runs outside any await so the `&mut ReplicaNonceRing` borrow cannot
+/// span I/O.
fn handshake_verify(
parsed: &ParsedHandshake,
token_source: &dyn TokenSource,
- nonces: &mut NonceRing,
+ nonces: &mut auth::ReplicaNonceRing,
now_ns: u128,
self_id: u8,
replica_count: u8,
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index c98faf467..157a5780a 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -67,7 +67,14 @@
pub mod tcp;
-pub use tcp::{TcpTransportConn, TcpTransportListener, TcpTransportReader, TcpTransportWriter};
+// Only `Conn` and `Writer` have crate-internal callers today
+// (`installer.rs` wraps the dialed/accepted stream in a `TcpTransportConn`;
+// `writer_task.rs` wraps the owned write half in a `TcpTransportWriter`).
+// `TcpTransportListener` and `TcpTransportReader` stay reachable via the
+// `tcp` submodule and exist primarily so the trait surface compiles
+// end-to-end against a real impl; future WSS / QUIC listeners drop in
+// alongside without touching call sites.
+pub(crate) use tcp::{TcpTransportConn, TcpTransportWriter};
use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
use iggy_binary_protocol::consensus::iobuf::Frozen;
@@ -118,8 +125,14 @@ pub trait TransportConn: 'static {
}
/// Read half. Produces one framed [`Message<GenericHeader>`] per call.
+///
+/// `'static` so the half can be moved into a `compio::runtime::spawn`'d
+/// task. `Unpin` is NOT required at the trait level: TCP impls satisfy it
+/// trivially via `OwnedReadHalf<TcpStream>`, and a future
+/// `compio-quic::RecvStream` impl that is not `Unpin` can pin internally
+/// without forcing every other impl to `Box::pin`.
#[allow(async_fn_in_trait)]
-pub trait TransportReader: Unpin + 'static {
+pub trait TransportReader: 'static {
/// Read the next `GenericHeader`-framed message off the wire.
///
/// Validates the 256 B header and bounds the total frame size to
@@ -141,8 +154,12 @@ pub trait TransportReader: Unpin + 'static {
}
/// Write half. Atomic-or-error batched writer.
+///
+/// `'static` so the half can be moved into a `compio::runtime::spawn`'d
+/// task. `Unpin` is NOT required at the trait level (see
+/// [`TransportReader`] for the rationale).
#[allow(async_fn_in_trait)]
-pub trait TransportWriter: Unpin + 'static {
+pub trait TransportWriter: 'static {
/// Send every buffer in `batch` atomically.
///
/// On success the Vec is drained (empty on return); the same
diff --git a/core/message_bus/src/transports/tcp.rs b/core/message_bus/src/transports/tcp.rs
index 7478c99b4..79694b06d 100644
--- a/core/message_bus/src/transports/tcp.rs
+++ b/core/message_bus/src/transports/tcp.rs
@@ -49,13 +49,22 @@ use std::net::SocketAddr;
/// `keepalive`) via `compio::net::SocketOpts`. See
/// [`crate::replica_listener::bind`] and
/// [`crate::client_listener::bind`] for the canonical construction.
-pub struct TcpTransportListener {
+///
+/// Currently only used by trait-conformance tests; production `accept`
+/// loops in [`crate::replica_listener`] / [`crate::client_listener`]
+/// drive a raw `TcpListener` directly. Kept `pub(crate)` so the
+/// `TransportListener` impl below is actually exercisable; the
+/// `dead_code` allow is intentional and will fall away when the listener
+/// loops migrate behind the trait (Phase 2+).
+#[allow(dead_code)]
+pub(crate) struct TcpTransportListener {
inner: TcpListener,
}
impl TcpTransportListener {
#[must_use]
- pub const fn new(inner: TcpListener) -> Self {
+ #[allow(dead_code)]
+ pub(crate) const fn new(inner: TcpListener) -> Self {
Self { inner }
}
}
@@ -76,13 +85,13 @@ impl TransportListener for TcpTransportListener {
/// result of a `TcpStream::connect` on the dialer path. Takes ownership
/// of the stream; [`Self::into_split`] transfers that ownership into
/// the read and write halves bound to the per-connection tasks.
-pub struct TcpTransportConn {
+pub(crate) struct TcpTransportConn {
stream: TcpStream,
}
impl TcpTransportConn {
#[must_use]
- pub const fn new(stream: TcpStream) -> Self {
+ pub(crate) const fn new(stream: TcpStream) -> Self {
Self { stream }
}
}
@@ -106,13 +115,14 @@ impl TransportConn for TcpTransportConn {
/// [`framing::read_message`]; the two paths share the same header
/// decode, bounds check, and zero-copy `Owned<MESSAGE_ALIGN>`
/// allocation strategy.
-pub struct TcpTransportReader {
+pub(crate) struct TcpTransportReader {
inner: OwnedReadHalf<TcpStream>,
}
impl TcpTransportReader {
#[must_use]
- pub const fn new(inner: OwnedReadHalf<TcpStream>) -> Self {
+ #[allow(dead_code)]
+ pub(crate) const fn new(inner: OwnedReadHalf<TcpStream>) -> Self {
Self { inner }
}
}
@@ -136,13 +146,13 @@ impl TransportReader for TcpTransportReader {
/// `max_batch <= IOV_MAX / 2 = 512`; this impl does not enforce a cap
/// because the Vec is already drained by the caller's admission
/// control.
-pub struct TcpTransportWriter {
+pub(crate) struct TcpTransportWriter {
inner: OwnedWriteHalf<TcpStream>,
}
impl TcpTransportWriter {
#[must_use]
- pub const fn new(inner: OwnedWriteHalf<TcpStream>) -> Self {
+ pub(crate) const fn new(inner: OwnedWriteHalf<TcpStream>) -> Self {
Self { inner }
}
}
diff --git a/core/message_bus/src/writer_task.rs b/core/message_bus/src/writer_task.rs
index 75b09c827..14d292bab 100644
--- a/core/message_bus/src/writer_task.rs
+++ b/core/message_bus/src/writer_task.rs
@@ -43,8 +43,8 @@ use tracing::{debug, error, trace};
/// Larger batches reduce syscalls per N messages at the cost of memory
/// per batch and worst-case latency for the head-of-batch message.
///
-/// TCP entry point. Wraps the owned write half in a
-/// [`TcpTransportWriter`] and delegates the drain loop to
+/// TCP entry point. Wraps the owned write half in the crate-internal
+/// `TcpTransportWriter` and delegates the drain loop to
/// [`run_transport`]; every syscall still flows through
/// [`TransportWriter::send_batch`] so future transports drop in behind
/// the same drain logic.
diff --git a/core/message_bus/tests/backpressure.rs b/core/message_bus/tests/backpressure.rs
index 7e849a981..a92e94090 100644
--- a/core/message_bus/tests/backpressure.rs
+++ b/core/message_bus/tests/backpressure.rs
@@ -52,6 +52,7 @@ async fn try_send_returns_backpressure_when_queue_full() {
});
let (l1, addr1) = bind(loopback()).await.unwrap();
let token_for_l1 = bus1.token();
+ let nonces_for_l1 = bus1.replica_nonces();
let accept_1 = install_replicas_locally(bus1.clone(), on_message1);
let l1_handle = compio::runtime::spawn(async move {
run(
@@ -63,6 +64,7 @@ async fn try_send_returns_backpressure_when_queue_full() {
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l1,
)
.await;
});
diff --git a/core/message_bus/tests/connection_lost_notify.rs b/core/message_bus/tests/connection_lost_notify.rs
index 00a05b5f7..fcb2cce65 100644
--- a/core/message_bus/tests/connection_lost_notify.rs
+++ b/core/message_bus/tests/connection_lost_notify.rs
@@ -54,6 +54,7 @@ async fn connection_lost_fires_exactly_once_per_peer_disconnect() {
// bus1 listens; bus0 dials.
let (l1, addr1) = bind(loopback()).await.unwrap();
let token_for_l1 = bus1.token();
+ let nonces_for_l1 = bus1.replica_nonces();
let accept_1 = install_replicas_locally(bus1.clone(), on_message.clone());
let l1_handle = compio::runtime::spawn(async move {
run(
@@ -65,6 +66,7 @@ async fn connection_lost_fires_exactly_once_per_peer_disconnect() {
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l1,
)
.await;
});
diff --git a/core/message_bus/tests/directional_connection.rs b/core/message_bus/tests/directional_connection.rs
index 6251939ce..9f80f352c 100644
--- a/core/message_bus/tests/directional_connection.rs
+++ b/core/message_bus/tests/directional_connection.rs
@@ -43,6 +43,7 @@ async fn lower_id_dials_higher_id_accepts() {
let (l1, addr1) = bind(loopback()).await.unwrap();
let token_for_l0 = bus0.token();
+ let nonces_for_l0 = bus0.replica_nonces();
let accept_0 = install_replicas_locally(bus0.clone(), on_message.clone());
let l0_handle = compio::runtime::spawn(async move {
run(
@@ -54,12 +55,14 @@ async fn lower_id_dials_higher_id_accepts() {
accept_0,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l0,
)
.await;
});
bus0.track_background(l0_handle);
let token_for_l1 = bus1.token();
+ let nonces_for_l1 = bus1.replica_nonces();
let accept_1 = install_replicas_locally(bus1.clone(), on_message.clone());
let l1_handle = compio::runtime::spawn(async move {
run(
@@ -71,6 +74,7 @@ async fn lower_id_dials_higher_id_accepts() {
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l1,
)
.await;
});
diff --git a/core/message_bus/tests/handshake_auth.rs b/core/message_bus/tests/handshake_auth.rs
index b833354e7..6ed94257b 100644
--- a/core/message_bus/tests/handshake_auth.rs
+++ b/core/message_bus/tests/handshake_auth.rs
@@ -47,6 +47,7 @@ async fn handshake_with_matching_secret_registers_peer() {
let token_for_listener = bus_listener.token();
let accept = install_replicas_locally(bus_listener.clone(), on_msg.clone());
let secret_for_listener = Rc::clone(&secret);
+ let nonces_for_listener = bus_listener.replica_nonces();
let listener_handle = compio::runtime::spawn(async move {
run(
listener,
@@ -57,6 +58,7 @@ async fn handshake_with_matching_secret_registers_peer() {
accept,
message_bus::framing::MAX_MESSAGE_SIZE,
secret_for_listener,
+ nonces_for_listener,
)
.await;
});
@@ -102,6 +104,7 @@ async fn handshake_with_mismatched_secret_rejects_peer() {
let (listener, addr) = bind(loopback()).await.unwrap();
let token_for_listener = bus_listener.token();
let accept = install_replicas_locally(bus_listener.clone(), on_msg.clone());
+ let nonces_for_listener = bus_listener.replica_nonces();
let listener_handle = compio::runtime::spawn(async move {
run(
listener,
@@ -112,6 +115,7 @@ async fn handshake_with_mismatched_secret_rejects_peer() {
accept,
message_bus::framing::MAX_MESSAGE_SIZE,
listener_secret,
+ nonces_for_listener,
)
.await;
});
diff --git a/core/message_bus/tests/head_of_line.rs b/core/message_bus/tests/head_of_line.rs
index b756897ce..d046715a3 100644
--- a/core/message_bus/tests/head_of_line.rs
+++ b/core/message_bus/tests/head_of_line.rs
@@ -48,6 +48,7 @@ async fn slow_peer_does_not_block_other_peers() {
});
let (la, addr_a) = bind(loopback()).await.unwrap();
let token_a = bus_a.token();
+ let nonces_a = bus_a.replica_nonces();
let accept_a = install_replicas_locally(bus_a.clone(), on_a.clone());
let la_handle = compio::runtime::spawn(async move {
run(
@@ -59,6 +60,7 @@ async fn slow_peer_does_not_block_other_peers() {
accept_a,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_a,
)
.await;
});
diff --git a/core/message_bus/tests/reconnect.rs b/core/message_bus/tests/reconnect.rs
index 93ac64332..20add2649 100644
--- a/core/message_bus/tests/reconnect.rs
+++ b/core/message_bus/tests/reconnect.rs
@@ -59,6 +59,7 @@ async fn periodic_retry_picks_up_late_listener() {
let bus1 = Rc::new(IggyMessageBus::new(0));
let (l1, _) = bind(addr).await.unwrap();
let token_for_l1 = bus1.token();
+ let nonces_for_l1 = bus1.replica_nonces();
let accept_delegate = install_replicas_locally(bus1.clone(), on_message.clone());
let l1_handle = compio::runtime::spawn(async move {
run(
@@ -70,6 +71,7 @@ async fn periodic_retry_picks_up_late_listener() {
accept_delegate,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l1,
)
.await;
});
diff --git a/core/message_bus/tests/reconnect_skip_connected.rs b/core/message_bus/tests/reconnect_skip_connected.rs
index 31159e1db..635250a28 100644
--- a/core/message_bus/tests/reconnect_skip_connected.rs
+++ b/core/message_bus/tests/reconnect_skip_connected.rs
@@ -50,6 +50,7 @@ async fn periodic_reconnect_skips_already_connected_peer() {
accept_inner(stream, peer_id);
});
let token_for_l1 = bus1.token();
+ let nonces_for_l1 = bus1.replica_nonces();
let l1_handle = compio::runtime::spawn(async move {
run(
l1,
@@ -60,6 +61,7 @@ async fn periodic_reconnect_skips_already_connected_peer() {
accept_delegate,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l1,
)
.await;
});
diff --git a/core/message_bus/tests/replica_roundtrip.rs b/core/message_bus/tests/replica_roundtrip.rs
index 8d8fde80a..8b609c0a4 100644
--- a/core/message_bus/tests/replica_roundtrip.rs
+++ b/core/message_bus/tests/replica_roundtrip.rs
@@ -45,6 +45,7 @@ async fn two_replicas_exchange_prepare_and_ack() {
let (l1, addr1) = bind(loopback()).await.expect("bind r1");
let token_for_l1 = bus1.token();
+ let nonces_for_l1 = bus1.replica_nonces();
let accept_delegate_1 = install_replicas_locally(bus1.clone(), on_message1.clone());
let l1_handle = compio::runtime::spawn(async move {
run(
@@ -56,6 +57,7 @@ async fn two_replicas_exchange_prepare_and_ack() {
accept_delegate_1,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l1,
)
.await;
});
diff --git a/core/message_bus/tests/shard_zero_gating.rs b/core/message_bus/tests/shard_zero_gating.rs
index 1dc0cc339..1b99d0a5b 100644
--- a/core/message_bus/tests/shard_zero_gating.rs
+++ b/core/message_bus/tests/shard_zero_gating.rs
@@ -41,6 +41,7 @@ async fn shard_zero_binds_listener_and_starts_connector() {
let peer_handler: MessageHandler = Rc::new(|_, _| {});
let (peer_listener, peer_addr) = bind(loopback()).await.expect("bind peer listener");
let peer_token = peer_bus.token();
+ let peer_nonces = peer_bus.replica_nonces();
let peer_accept = install_replicas_locally(peer_bus.clone(), peer_handler.clone());
let peer_listen_handle = compio::runtime::spawn(async move {
run(
@@ -52,6 +53,7 @@ async fn shard_zero_binds_listener_and_starts_connector() {
peer_accept,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ peer_nonces,
)
.await;
});
diff --git a/core/message_bus/tests/vectored_batch.rs b/core/message_bus/tests/vectored_batch.rs
index b079725a5..95deac326 100644
--- a/core/message_bus/tests/vectored_batch.rs
+++ b/core/message_bus/tests/vectored_batch.rs
@@ -47,6 +47,7 @@ async fn writer_batches_pipelined_sends_in_order() {
});
let (l1, addr1) = bind(loopback()).await.unwrap();
let token_for_l1 = bus1.token();
+ let nonces_for_l1 = bus1.replica_nonces();
let accept_1 = install_replicas_locally(bus1.clone(), on_message.clone());
let l1_handle = compio::runtime::spawn(async move {
run(
@@ -58,6 +59,7 @@ async fn writer_batches_pipelined_sends_in_order() {
accept_1,
message_bus::framing::MAX_MESSAGE_SIZE,
test_token_source(),
+ nonces_for_l1,
)
.await;
});
diff --git a/core/server-ng/src/dedup.rs b/core/server-ng/src/dedup.rs
index 0022e9d85..cbc235129 100644
--- a/core/server-ng/src/dedup.rs
+++ b/core/server-ng/src/dedup.rs
@@ -17,25 +17,47 @@
* under the License.
*/
-//! At-most-once request dedup window (IGGY-112, P0-T3).
+// `Dedup` is consumed only by the `#[cfg(test)]` tests below today; the
+// request dispatcher loop in server-ng is not yet wired up. Once the
+// dispatcher lands and calls `lookup` / `mark_in_flight` / `complete` /
+// `evict_client`, the `dead_code` warnings disappear without needing
+// this allow. See plan F6.
+#![allow(dead_code)]
+
+//! At-most-once **per-shard-lifetime** request dedup window (IGGY-112, P0-T3).
//!
//! Indexed by `(client_id, request)` pairs carried in the existing
//! `RequestHeader.client` (u128) and `RequestHeader.request` (u64) fields.
//! No wire-format change on the server-ng client plane was needed: the
//! P0-T1 design note found the two fields already in place.
//!
+//! # Lifetime scope (load-bearing)
+//!
+//! "At-most-once" applies for the lifetime of THIS shard's `Dedup`
+//! instance. Process restart drops every entry: a client that retries a
+//! request after a server restart re-applies side effects unless the
+//! request is idempotent at the handler level. Cross-process persistence
+//! would need a checkpoint file backed by the consensus WAL and is
+//! deferred. Operators sizing client-side retry policy must treat the
+//! dedup window as a *crash-window* mitigation only.
+//!
+//! # Per-client ring
+//!
//! Per-client state is a fixed-capacity ring holding the last N
-//! `(request, Entry)` pairs.
+//! `(request, Entry)` pairs. A ring avoids unbounded growth under a
+//! chatty client without requiring TTL-driven purge ticks. Done entries
+//! also carry an `inserted_at_ns` timestamp; `lookup` treats entries
+//! older than `ttl_ns` as absent so a replay from far in the past sees
+//! `Fresh` rather than a stale cache hit.
//!
-//! A ring avoids unbounded growth under a chatty client without
-//! requiring TTL-driven purge ticks. Done entries also carry an
-//! `inserted_at_ns` timestamp; `lookup` treats entries older than
-//! `ttl_ns` as absent so a replay from far in the past sees `Fresh`
-//! rather than a stale cache hit.
+//! # Threading
//!
-//! The store is NOT thread-safe. Each server-ng shard owns a `Dedup`
-//! on its single-threaded compio runtime. Cross-shard routing happens
-//! above this layer.
+//! The store is NOT thread-safe. Each server-ng shard owns a `Dedup` on
+//! its single-threaded compio runtime. Cross-shard routing happens above
+//! this layer. Caller MUST invoke [`Dedup::evict_client`] on session
+//! disconnect / bind eviction so the per-client `AHashMap` entry is
+//! released; without that hook the outer map grows unbounded across the
+//! shard's session history.
use ahash::AHashMap;
use bytes::Bytes;
@@ -69,6 +91,11 @@ pub enum LookupResult<'a> {
/// the cached reply once handler completes), or buffer.
InFlight,
/// Handler already ran; return the cached reply.
+ ///
+ /// The borrow ties to `&Dedup`. Callers MUST `clone()` the `Bytes`
+ /// (cheap, ref-counted) before invoking any `&mut Dedup` method on
+ /// the same instance; the borrow checker enforces this at compile
+ /// time but the contract is worth naming explicitly.
Cached(&'a Bytes),
}
@@ -93,7 +120,10 @@ struct PerClient {
impl PerClient {
fn with_capacity(capacity: usize) -> Self {
- assert!(capacity > 0, "dedup ring capacity must be > 0");
+ // Capacity > 0 is validated at the public entry
+ // (`Dedup::with_config`); every callsite here goes through that
+ // path, so no second assert is needed.
+ debug_assert!(capacity > 0, "dedup ring capacity must be > 0");
Self {
slots: Vec::with_capacity(capacity),
cursor: 0,
diff --git a/core/server-ng/src/lib.rs b/core/server-ng/src/lib.rs
index 0aa424bba..d3849ab02 100644
--- a/core/server-ng/src/lib.rs
+++ b/core/server-ng/src/lib.rs
@@ -17,6 +17,10 @@
* under the License.
*/
-pub mod dedup;
+// `dedup` is gated `pub(crate)` until the request dispatcher loop wires
+// up the lookup / mark / complete / evict_client API end-to-end. Marking
+// it crate-private now blocks downstream crates from coupling to a shape
+// the dispatcher PR will likely refine (see plan F6).
+pub(crate) mod dedup;
pub mod login_register;
pub mod session_manager;
(iggy) 02/14: feat(server-ng): at-most-once request dedup window (IGGY-112, P0-T3)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 440994de444ff7f79d296b2e56eca0dee50c12fd
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 11:42:59 2026 +0200
feat(server-ng): at-most-once request dedup window (IGGY-112, P0-T3)
Add a per-shard Dedup store indexed by (client_id, request) pairs that
the existing RequestHeader already carries. The P0-T1 design note
found that RequestHeader has client: u128, request: u64, and
request_checksum: u128 today, so the phase-0 plan's "add request_id:
u128" wire widening is not needed; this commit skips it.
Dedup entries are stored in a fixed-capacity per-client ring
(DEFAULT_PER_CLIENT_CAPACITY = 64). Memory caps at
clients * capacity * avg_reply_size without needing a purge tick.
Entries carry inserted_at_ns; lookup treats entries older than
DEFAULT_ENTRY_TTL_NS (30 s, matching the replica-plane handshake
window) as absent so a stale replay reports Fresh. evict_client drops
all state for a (client_id, *) on disconnect; higher layers call it
from SessionManager on bind-eviction.
The API is lookup / mark_in_flight / complete / evict_client. Callers
see LookupResult { Fresh, InFlight, Cached(&Bytes) }. The `&Bytes`
return lets the fast-path cache hit clone a reference-counted handle
without copying the reply payload.
Not wired into a dispatcher yet. server-ng currently has no request
dispatch loop; the module ships as a standalone primitive ready for
when the loop lands. Benchmark against the 100 ns/request target in
the phase-0 acceptance is deferred to a follow-up commit.
Tests: 11 dedup unit tests exercising state transitions, TTL
boundaries, ring eviction, cross-client isolation, and the
refresh-on-overwrite timestamp path.
Refs: Documents/silverhand/iggy/message_bus/transport-plan/ (P0-T3,
designs/message-bus-auth.md).
---
core/server-ng/src/dedup.rs | 363 ++++++++++++++++++++++++++++++++++++++++++++
core/server-ng/src/lib.rs | 1 +
2 files changed, 364 insertions(+)
diff --git a/core/server-ng/src/dedup.rs b/core/server-ng/src/dedup.rs
new file mode 100644
index 000000000..0022e9d85
--- /dev/null
+++ b/core/server-ng/src/dedup.rs
@@ -0,0 +1,363 @@
+/*
+ * 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.
+ */
+
+//! At-most-once request dedup window (IGGY-112, P0-T3).
+//!
+//! Indexed by `(client_id, request)` pairs carried in the existing
+//! `RequestHeader.client` (u128) and `RequestHeader.request` (u64) fields.
+//! No wire-format change on the server-ng client plane was needed: the
+//! P0-T1 design note found the two fields already in place.
+//!
+//! Per-client state is a fixed-capacity ring holding the last N
+//! `(request, Entry)` pairs.
+//!
+//! A ring avoids unbounded growth under a chatty client without
+//! requiring TTL-driven purge ticks. Done entries also carry an
+//! `inserted_at_ns` timestamp; `lookup` treats entries older than
+//! `ttl_ns` as absent so a replay from far in the past sees `Fresh`
+//! rather than a stale cache hit.
+//!
+//! The store is NOT thread-safe. Each server-ng shard owns a `Dedup`
+//! on its single-threaded compio runtime. Cross-shard routing happens
+//! above this layer.
+
+use ahash::AHashMap;
+use bytes::Bytes;
+
+/// Per-client ring capacity on the dedup window. Sized for a pipelining
+/// client issuing tens of outstanding requests; small enough to cap
+/// memory at `clients * 64 * avg_reply_size`.
+pub const DEFAULT_PER_CLIENT_CAPACITY: usize = 64;
+
+/// Entry TTL.
+///
+/// An entry older than this is invisible to `lookup`; a replay past
+/// the window is treated as `Fresh`. 30 s matches the replica-plane
+/// handshake window (`core/message_bus/src/auth.rs`) so operators see
+/// one number in their threat model, not two.
+pub const DEFAULT_ENTRY_TTL_NS: u128 = 30_000_000_000;
+
+/// Result of `Dedup::lookup`.
+///
+/// Returned as a borrowed view into the cache: `Cached` hands the caller
+/// a `&Bytes` it can `clone()` (cheap, reference-counted) before
+/// releasing the borrow. Avoids copying the reply payload on the
+/// fast-path cache hit.
+#[derive(Debug)]
+pub enum LookupResult<'a> {
+ /// No (live) entry for this `(client, request)`. Caller should call
+ /// [`Dedup::mark_in_flight`] and run the handler.
+ Fresh,
+ /// The same request is currently being processed by an earlier
+ /// invocation. Caller policy decides: drop (client retry will find
+ /// the cached reply once handler completes), or buffer.
+ InFlight,
+ /// Handler already ran; return the cached reply.
+ Cached(&'a Bytes),
+}
+
+#[derive(Debug)]
+enum Entry {
+ InFlight,
+ Done { reply: Bytes, inserted_at_ns: u128 },
+}
+
+#[derive(Debug)]
+struct Slot {
+ request: u64,
+ entry: Entry,
+}
+
+#[derive(Debug)]
+struct PerClient {
+ slots: Vec<Slot>,
+ cursor: usize,
+ capacity: usize,
+}
+
+impl PerClient {
+ fn with_capacity(capacity: usize) -> Self {
+ assert!(capacity > 0, "dedup ring capacity must be > 0");
+ Self {
+ slots: Vec::with_capacity(capacity),
+ cursor: 0,
+ capacity,
+ }
+ }
+
+ fn find(&self, request: u64) -> Option<&Slot> {
+ self.slots.iter().find(|s| s.request == request)
+ }
+
+ fn find_mut(&mut self, request: u64) -> Option<&mut Slot> {
+ self.slots.iter_mut().find(|s| s.request == request)
+ }
+
+ fn push(&mut self, slot: Slot) {
+ if self.slots.len() < self.capacity {
+ self.slots.push(slot);
+ } else {
+ self.slots[self.cursor] = slot;
+ self.cursor = (self.cursor + 1) % self.capacity;
+ }
+ }
+}
+
+/// At-most-once request dedup window.
+///
+/// Single-shard, single-threaded. Callers hold a `&mut Dedup` and drive
+/// the lookup / mark / complete flow synchronously on the hot path.
+#[derive(Debug)]
+pub struct Dedup {
+ per_client: AHashMap<u128, PerClient>,
+ per_client_capacity: usize,
+ ttl_ns: u128,
+}
+
+impl Dedup {
+ /// Build a window with defaults: 64-entry per-client ring, 30 s TTL.
+ #[must_use]
+ pub fn new() -> Self {
+ Self::with_config(DEFAULT_PER_CLIENT_CAPACITY, DEFAULT_ENTRY_TTL_NS)
+ }
+
+ /// # Panics
+ /// Panics if `per_client_capacity == 0`. Every producing path in
+ /// the crate uses [`DEFAULT_PER_CLIENT_CAPACITY`]; the assertion
+ /// guards misconfiguration in tests or future callers.
+ #[must_use]
+ pub fn with_config(per_client_capacity: usize, ttl_ns: u128) -> Self {
+ assert!(per_client_capacity > 0, "capacity must be > 0");
+ Self {
+ per_client: AHashMap::new(),
+ per_client_capacity,
+ ttl_ns,
+ }
+ }
+
+ /// Inspect the state for `(client, request)` at `now_ns`.
+ ///
+ /// Done entries older than the TTL are treated as absent; the window
+ /// still holds the slot but does not report it. On eviction by a
+ /// later `push`, the slot is overwritten silently.
+ #[must_use]
+ pub fn lookup(&self, client: u128, request: u64, now_ns: u128) -> LookupResult<'_> {
+ let Some(state) = self.per_client.get(&client) else {
+ return LookupResult::Fresh;
+ };
+ let Some(slot) = state.find(request) else {
+ return LookupResult::Fresh;
+ };
+ match &slot.entry {
+ Entry::InFlight => LookupResult::InFlight,
+ Entry::Done {
+ reply,
+ inserted_at_ns,
+ } => {
+ if now_ns.saturating_sub(*inserted_at_ns) > self.ttl_ns {
+ LookupResult::Fresh
+ } else {
+ LookupResult::Cached(reply)
+ }
+ }
+ }
+ }
+
+ /// Mark the handler as starting. Returns `true` if the slot was
+ /// inserted; `false` if an entry for `(client, request)` already
+ /// existed (caller must consult [`Self::lookup`] for the specific
+ /// state).
+ pub fn mark_in_flight(&mut self, client: u128, request: u64) -> bool {
+ let state = self
+ .per_client
+ .entry(client)
+ .or_insert_with(|| PerClient::with_capacity(self.per_client_capacity));
+ if state.find(request).is_some() {
+ return false;
+ }
+ state.push(Slot {
+ request,
+ entry: Entry::InFlight,
+ });
+ true
+ }
+
+ /// Record a completed reply. Transitions the slot from `InFlight`
+ /// to `Done`. If no slot exists (e.g. TTL-evicted between
+ /// `mark_in_flight` and `complete`), a new `Done` slot is appended.
+ pub fn complete(&mut self, client: u128, request: u64, reply: Bytes, now_ns: u128) {
+ let state = self
+ .per_client
+ .entry(client)
+ .or_insert_with(|| PerClient::with_capacity(self.per_client_capacity));
+ if let Some(slot) = state.find_mut(request) {
+ slot.entry = Entry::Done {
+ reply,
+ inserted_at_ns: now_ns,
+ };
+ return;
+ }
+ state.push(Slot {
+ request,
+ entry: Entry::Done {
+ reply,
+ inserted_at_ns: now_ns,
+ },
+ });
+ }
+
+ /// Drop all state for `client`. Call on disconnect (or on
+ /// `bind_session` eviction inside `SessionManager`) so the
+ /// per-client ring is released deterministically rather than
+ /// waiting for TTL expiry of individual entries.
+ pub fn evict_client(&mut self, client: u128) {
+ self.per_client.remove(&client);
+ }
+
+ /// Current number of tracked clients. Test / metrics helper.
+ #[must_use]
+ pub fn client_count(&self) -> usize {
+ self.per_client.len()
+ }
+}
+
+impl Default for Dedup {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn bytes(s: &str) -> Bytes {
+ Bytes::copy_from_slice(s.as_bytes())
+ }
+
+ #[test]
+ fn fresh_lookup_on_empty_store() {
+ let d = Dedup::new();
+ assert!(matches!(d.lookup(1, 1, 0), LookupResult::Fresh));
+ }
+
+ #[test]
+ fn mark_then_lookup_returns_in_flight() {
+ let mut d = Dedup::new();
+ assert!(d.mark_in_flight(1, 1));
+ assert!(matches!(d.lookup(1, 1, 0), LookupResult::InFlight));
+ }
+
+ #[test]
+ fn double_mark_returns_false() {
+ let mut d = Dedup::new();
+ assert!(d.mark_in_flight(1, 1));
+ assert!(!d.mark_in_flight(1, 1));
+ }
+
+ #[test]
+ fn complete_transitions_to_cached() {
+ let mut d = Dedup::new();
+ d.mark_in_flight(1, 7);
+ d.complete(1, 7, bytes("ok"), 1_000);
+ match d.lookup(1, 7, 1_000) {
+ LookupResult::Cached(reply) => assert_eq!(&reply[..], b"ok"),
+ other => panic!("expected Cached, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn complete_without_mark_inserts_directly() {
+ let mut d = Dedup::new();
+ d.complete(1, 7, bytes("ok"), 0);
+ assert!(matches!(d.lookup(1, 7, 0), LookupResult::Cached(_)));
+ }
+
+ #[test]
+ fn ttl_expired_entry_reports_fresh() {
+ let mut d = Dedup::with_config(8, 1_000);
+ d.complete(1, 7, bytes("ok"), 0);
+ assert!(matches!(d.lookup(1, 7, 2_000), LookupResult::Fresh));
+ }
+
+ #[test]
+ fn ttl_at_boundary_still_cached() {
+ let mut d = Dedup::with_config(8, 1_000);
+ d.complete(1, 7, bytes("ok"), 0);
+ assert!(matches!(d.lookup(1, 7, 1_000), LookupResult::Cached(_)));
+ }
+
+ #[test]
+ fn ring_evicts_oldest_entry() {
+ let mut d = Dedup::with_config(3, DEFAULT_ENTRY_TTL_NS);
+ d.complete(1, 1, bytes("a"), 10);
+ d.complete(1, 2, bytes("b"), 20);
+ d.complete(1, 3, bytes("c"), 30);
+ d.complete(1, 4, bytes("d"), 40);
+ // Slot for request 1 was overwritten by request 4.
+ assert!(matches!(d.lookup(1, 1, 50), LookupResult::Fresh));
+ assert!(matches!(d.lookup(1, 2, 50), LookupResult::Cached(_)));
+ assert!(matches!(d.lookup(1, 3, 50), LookupResult::Cached(_)));
+ assert!(matches!(d.lookup(1, 4, 50), LookupResult::Cached(_)));
+ }
+
+ #[test]
+ fn evict_client_drops_all_entries() {
+ let mut d = Dedup::new();
+ d.complete(1, 1, bytes("a"), 10);
+ d.complete(1, 2, bytes("b"), 20);
+ d.evict_client(1);
+ assert!(matches!(d.lookup(1, 1, 30), LookupResult::Fresh));
+ assert_eq!(d.client_count(), 0);
+ }
+
+ #[test]
+ fn isolation_between_clients() {
+ let mut d = Dedup::new();
+ d.complete(1, 5, bytes("a"), 10);
+ d.complete(2, 5, bytes("b"), 20);
+ match d.lookup(1, 5, 30) {
+ LookupResult::Cached(r) => assert_eq!(&r[..], b"a"),
+ other => panic!("expected Cached a, got {other:?}"),
+ }
+ match d.lookup(2, 5, 30) {
+ LookupResult::Cached(r) => assert_eq!(&r[..], b"b"),
+ other => panic!("expected Cached b, got {other:?}"),
+ }
+ }
+
+ #[test]
+ fn complete_updates_inserted_at_on_refresh() {
+ let mut d = Dedup::with_config(8, 100);
+ d.complete(1, 1, bytes("v1"), 0);
+ // Refresh with new timestamp and payload.
+ d.complete(1, 1, bytes("v2"), 50);
+ match d.lookup(1, 1, 140) {
+ LookupResult::Cached(r) => assert_eq!(&r[..], b"v2"),
+ other => panic!("expected Cached v2, got {other:?}"),
+ }
+ // Would have been TTL-expired if inserted_at had stayed at 0.
+ }
+
+ #[test]
+ #[should_panic(expected = "capacity must be > 0")]
+ fn zero_capacity_rejected() {
+ let _ = Dedup::with_config(0, DEFAULT_ENTRY_TTL_NS);
+ }
+}
diff --git a/core/server-ng/src/lib.rs b/core/server-ng/src/lib.rs
index caae94a3a..0aa424bba 100644
--- a/core/server-ng/src/lib.rs
+++ b/core/server-ng/src/lib.rs
@@ -17,5 +17,6 @@
* under the License.
*/
+pub mod dedup;
pub mod login_register;
pub mod session_manager;
(iggy) 08/14: feat(message_bus): IGGY_CLUSTER_SECRET env loader (IGGY-112)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c9fd47dec4e4522841d042dc408af571b3dc6ebc
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 12:35:37 2026 +0200
feat(message_bus): IGGY_CLUSTER_SECRET env loader (IGGY-112)
Add auth_config module that resolves the replica-plane 32-byte cluster
secret from either IGGY_CLUSTER_SECRET (64-char hex literal) or
IGGY_CLUSTER_SECRET_FILE (path to hex-only file), mutually exclusive.
- SecretLoadError distinguishes Missing / BothSet / WrongLength /
InvalidHex(offset) / FileRead so boot-time errors point operators at
the exact failure mode.
- Pure resolve() is factored out so unit tests exercise every branch
without touching process env state (which races under cargo test).
- cluster_token_source() returns Rc<dyn TokenSource> directly so call
sites such as replica_io::start_on_shard_zero avoid the unsize
coercion dance.
- Inline 0-dep hex nibble decoder; no new workspace crate. Whitespace
is trimmed so file contents with a trailing newline decode cleanly.
The secret intentionally does not live on MessageBusConfig: it has a
different lifecycle (boot-resolved, no Clone-into-debug-traces) and
Default is not meaningful. MessageBusConfig and StaticSharedSecret
docstrings now cross-reference auth_config. Config section in the
local CLAUDE.md is updated; zeroize-on-drop is a follow-up.
Tests: 10 new unit tests covering happy path (both literal and file),
whitespace trim, uppercase hex, each error variant, plus an
end-to-end round trip through cluster_token_source(). Adds tempfile
as a dev-dep for the filesystem round-trip test. 110 pass total on
cargo nextest run -p message_bus -p server-ng.
---
Cargo.lock | 1 +
core/message_bus/Cargo.toml | 3 +
core/message_bus/src/auth.rs | 8 +-
core/message_bus/src/auth_config.rs | 268 ++++++++++++++++++++++++++++++++++++
core/message_bus/src/config.rs | 6 +
core/message_bus/src/lib.rs | 1 +
6 files changed, 284 insertions(+), 3 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 2c66fb79c..af920f2d3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7017,6 +7017,7 @@ dependencies = [
"libc",
"rand 0.10.1",
"socket2 0.6.3",
+ "tempfile",
"thiserror 2.0.18",
"tracing",
]
diff --git a/core/message_bus/Cargo.toml b/core/message_bus/Cargo.toml
index d0487db0f..6fa1ae1f4 100644
--- a/core/message_bus/Cargo.toml
+++ b/core/message_bus/Cargo.toml
@@ -41,6 +41,9 @@ socket2 = { workspace = true, features = ["all"] }
thiserror = { workspace = true }
tracing = { workspace = true }
+[dev-dependencies]
+tempfile = { workspace = true }
+
[lints.clippy]
enum_glob_use = "deny"
pedantic = "deny"
diff --git a/core/message_bus/src/auth.rs b/core/message_bus/src/auth.rs
index 6c2353a6b..1205a4578 100644
--- a/core/message_bus/src/auth.rs
+++ b/core/message_bus/src/auth.rs
@@ -156,9 +156,11 @@ pub trait TokenSource {
fn verify(&self, challenge: &AuthChallenge, tag: &AuthTag) -> Result<(), AuthError>;
}
-/// `TokenSource` that holds a single 32-byte cluster-wide secret. All
-/// replicas share one secret on the replica plane; the integration-test
-/// harness instantiates this with a fixed zero-byte secret.
+/// `TokenSource` that holds a single 32-byte cluster-wide secret.
+///
+/// All replicas share one secret on the replica plane; the integration-test
+/// harness instantiates this with a fixed zero-byte secret, and production
+/// bootstraps resolve one via [`crate::auth_config`].
pub struct StaticSharedSecret {
secret: [u8; SECRET_SIZE],
}
diff --git a/core/message_bus/src/auth_config.rs b/core/message_bus/src/auth_config.rs
new file mode 100644
index 000000000..af715c6d3
--- /dev/null
+++ b/core/message_bus/src/auth_config.rs
@@ -0,0 +1,268 @@
+// 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.
+
+//! Boot-time loader for the replica-plane cluster secret.
+//!
+//! Two environment variables drive the loader; exactly one must be set:
+//!
+//! - `IGGY_CLUSTER_SECRET` — 64-character hex-encoded 32-byte secret,
+//! passed inline. Convenient for `docker-compose` / `systemd`
+//! `Environment=` but visible in `/proc/$pid/environ`.
+//! - `IGGY_CLUSTER_SECRET_FILE` — path to a file whose contents are the
+//! 64-character hex secret. Leading / trailing whitespace is trimmed.
+//!
+//! Both variants decode to the same `[u8; 32]` used by
+//! [`crate::auth::StaticSharedSecret`]. `StaticSharedSecret::zero()`
+//! remains the integration-test default; production deployments must
+//! choose one of the two env vars above.
+//!
+//! The loader is a one-shot startup event; synchronous `std::fs` is
+//! intentional. Do not call it on the data plane.
+
+use std::env;
+use std::fs;
+use std::rc::Rc;
+
+use thiserror::Error;
+
+use crate::auth::{SECRET_SIZE, StaticSharedSecret, TokenSource};
+
+/// Env var carrying the hex-encoded secret literal.
+pub const CLUSTER_SECRET_ENV: &str = "IGGY_CLUSTER_SECRET";
+
+/// Env var carrying the path to a file containing the hex-encoded secret.
+pub const CLUSTER_SECRET_FILE_ENV: &str = "IGGY_CLUSTER_SECRET_FILE";
+
+/// Length of the hex-encoded form of a [`SECRET_SIZE`]-byte secret.
+pub const HEX_SECRET_LEN: usize = SECRET_SIZE * 2;
+
+/// Failure modes for secret resolution. Surfaced distinctly so operators
+/// get a pointed error at boot rather than a generic "invalid config".
+#[derive(Debug, Error)]
+pub enum SecretLoadError {
+ #[error(
+ "neither {CLUSTER_SECRET_ENV} nor {CLUSTER_SECRET_FILE_ENV} is set; \
+ one is required to boot the replica plane"
+ )]
+ Missing,
+ #[error(
+ "both {CLUSTER_SECRET_ENV} and {CLUSTER_SECRET_FILE_ENV} are set; \
+ exactly one is required"
+ )]
+ BothSet,
+ #[error("secret has {got} hex characters; expected {HEX_SECRET_LEN}")]
+ WrongLength { got: usize },
+ #[error("secret contains non-hex character at byte offset {0}")]
+ InvalidHex(usize),
+ #[error("failed to read secret file {path}")]
+ FileRead {
+ path: String,
+ #[source]
+ source: std::io::Error,
+ },
+}
+
+/// Resolve the 32-byte secret from the process environment.
+///
+/// # Errors
+///
+/// See [`SecretLoadError`].
+pub fn load_cluster_secret() -> Result<[u8; SECRET_SIZE], SecretLoadError> {
+ let literal = env::var(CLUSTER_SECRET_ENV).ok();
+ let file = env::var(CLUSTER_SECRET_FILE_ENV).ok();
+ resolve(literal.as_deref(), file.as_deref())
+}
+
+/// Resolve the secret and wrap it in a reference-counted
+/// [`StaticSharedSecret`]. Call sites that expect
+/// `Rc<dyn TokenSource>` get it via unsize coercion:
+///
+/// ```ignore
+/// let ts: Rc<dyn TokenSource> = auth_config::cluster_token_source()?;
+/// ```
+///
+/// # Errors
+///
+/// See [`SecretLoadError`].
+pub fn cluster_token_source() -> Result<Rc<dyn TokenSource>, SecretLoadError> {
+ let secret = load_cluster_secret()?;
+ Ok(Rc::new(StaticSharedSecret::new(secret)))
+}
+
+/// Pure resolution rule, factored out of [`load_cluster_secret`] so tests
+/// can drive it without touching the shared process environment (env-var
+/// tests race under `cargo test`).
+///
+/// # Errors
+///
+/// See [`SecretLoadError`].
+pub fn resolve(
+ literal: Option<&str>,
+ file_path: Option<&str>,
+) -> Result<[u8; SECRET_SIZE], SecretLoadError> {
+ match (literal, file_path) {
+ (Some(hex), None) => decode_hex_secret(hex.trim()),
+ (None, Some(path)) => {
+ let raw = fs::read_to_string(path).map_err(|source| SecretLoadError::FileRead {
+ path: path.to_owned(),
+ source,
+ })?;
+ decode_hex_secret(raw.trim())
+ }
+ (Some(_), Some(_)) => Err(SecretLoadError::BothSet),
+ (None, None) => Err(SecretLoadError::Missing),
+ }
+}
+
+fn decode_hex_secret(input: &str) -> Result<[u8; SECRET_SIZE], SecretLoadError> {
+ if input.len() != HEX_SECRET_LEN {
+ return Err(SecretLoadError::WrongLength { got: input.len() });
+ }
+ let bytes = input.as_bytes();
+ let mut out = [0u8; SECRET_SIZE];
+ for (i, chunk) in bytes.chunks_exact(2).enumerate() {
+ let hi = decode_hex_nibble(chunk[0]).ok_or(SecretLoadError::InvalidHex(i * 2))?;
+ let lo = decode_hex_nibble(chunk[1]).ok_or(SecretLoadError::InvalidHex(i * 2 + 1))?;
+ out[i] = (hi << 4) | lo;
+ }
+ Ok(out)
+}
+
+const fn decode_hex_nibble(c: u8) -> Option<u8> {
+ match c {
+ b'0'..=b'9' => Some(c - b'0'),
+ b'a'..=b'f' => Some(10 + c - b'a'),
+ b'A'..=b'F' => Some(10 + c - b'A'),
+ _ => None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ const SAMPLE_HEX: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
+ const SAMPLE_BYTES: [u8; SECRET_SIZE] = [
+ 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e,
+ 0x0f, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d,
+ 0x1e, 0x1f,
+ ];
+
+ #[test]
+ fn resolve_literal_happy_path() {
+ let got = resolve(Some(SAMPLE_HEX), None).expect("resolve literal");
+ assert_eq!(got, SAMPLE_BYTES);
+ }
+
+ #[test]
+ fn resolve_literal_trims_whitespace() {
+ let padded = format!(" {SAMPLE_HEX}\n");
+ let got = resolve(Some(&padded), None).expect("resolve literal with whitespace");
+ assert_eq!(got, SAMPLE_BYTES);
+ }
+
+ #[test]
+ fn resolve_literal_uppercase_ok() {
+ let upper = SAMPLE_HEX.to_uppercase();
+ let got = resolve(Some(&upper), None).expect("resolve uppercase literal");
+ assert_eq!(got, SAMPLE_BYTES);
+ }
+
+ #[test]
+ fn neither_set_is_missing() {
+ let err = resolve(None, None).unwrap_err();
+ assert!(matches!(err, SecretLoadError::Missing), "got {err:?}");
+ }
+
+ #[test]
+ fn both_set_is_both_set() {
+ let err = resolve(Some(SAMPLE_HEX), Some("/dev/null")).unwrap_err();
+ assert!(matches!(err, SecretLoadError::BothSet), "got {err:?}");
+ }
+
+ #[test]
+ fn wrong_length_is_wrong_length() {
+ let short = &SAMPLE_HEX[..SAMPLE_HEX.len() - 2];
+ let err = resolve(Some(short), None).unwrap_err();
+ match err {
+ SecretLoadError::WrongLength { got } => assert_eq!(got, HEX_SECRET_LEN - 2),
+ other => panic!("unexpected error: {other:?}"),
+ }
+ }
+
+ #[test]
+ fn non_hex_is_invalid_hex() {
+ let mut bad = SAMPLE_HEX.to_owned();
+ // Replace the 5th hex char with a non-hex rune.
+ bad.replace_range(4..5, "Z");
+ let err = resolve(Some(&bad), None).unwrap_err();
+ match err {
+ SecretLoadError::InvalidHex(offset) => assert_eq!(offset, 4),
+ other => panic!("unexpected error: {other:?}"),
+ }
+ }
+
+ #[test]
+ fn missing_file_is_file_read() {
+ // A path that definitely does not exist; /proc/self/ is real but
+ // the basename is a random ULID-shaped string.
+ let bogus = "/proc/self/this-path-will-not-exist-01HZZZ";
+ let err = resolve(None, Some(bogus)).unwrap_err();
+ match err {
+ SecretLoadError::FileRead { path, .. } => assert_eq!(path, bogus),
+ other => panic!("unexpected error: {other:?}"),
+ }
+ }
+
+ #[test]
+ fn file_round_trip_with_tempfile() {
+ use std::io::Write;
+
+ let dir = tempfile::tempdir().expect("tempdir");
+ let path = dir.path().join("secret.hex");
+ let mut f = fs::File::create(&path).expect("create");
+ writeln!(f, " {SAMPLE_HEX} ").expect("write");
+ drop(f);
+
+ let got =
+ resolve(None, Some(path.to_str().expect("utf-8 path"))).expect("resolve from tempfile");
+ assert_eq!(got, SAMPLE_BYTES);
+ }
+
+ #[test]
+ fn cluster_token_source_produces_matching_secret() {
+ // Exercise the Rc<dyn TokenSource> return path end-to-end: resolve,
+ // sign a challenge with the loader's secret, verify with a
+ // hand-built StaticSharedSecret over the same bytes.
+ use crate::auth::{AuthChallenge, TokenSource};
+
+ let source = Rc::new(StaticSharedSecret::new(
+ resolve(Some(SAMPLE_HEX), None).expect("resolve"),
+ )) as Rc<dyn TokenSource>;
+ let reference = StaticSharedSecret::new(SAMPLE_BYTES);
+ let challenge = AuthChallenge {
+ cluster: 1,
+ peer_id: 2,
+ timestamp_ns: 3,
+ release: 4,
+ nonce: 5,
+ };
+ let from_loader = source.sign(&challenge);
+ let from_reference = reference.sign(&challenge);
+ assert_eq!(from_loader, from_reference);
+ }
+}
diff --git a/core/message_bus/src/config.rs b/core/message_bus/src/config.rs
index d99b3fb49..f82114461 100644
--- a/core/message_bus/src/config.rs
+++ b/core/message_bus/src/config.rs
@@ -26,6 +26,12 @@
//! constructs [`crate::IggyMessageBus`] from `ServerConfig` lands.
//! Keeping the type in-crate for now avoids churning the configs crate
//! ahead of that wiring.
+//!
+//! The replica-plane auth secret intentionally does NOT live on this
+//! struct: it has a different lifecycle (boot-time resolution, no
+//! `Clone` in debug traces) and is constructed via
+//! [`crate::auth_config`] at process start and threaded as an
+//! `Rc<dyn TokenSource>` into [`crate::replica_io::start_on_shard_zero`].
use std::time::Duration;
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index 9d95cc63e..96d87d961 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -71,6 +71,7 @@
//! under `Documents/silverhand/iggy/message_bus/transport-plan/`.
pub mod auth;
+pub mod auth_config;
pub mod cache;
pub mod client_listener;
pub mod config;
(iggy) 10/14: docs(message_bus): demote broken intra-doc links to code spans
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 1b163dad2d78cf9b92cd1f94300b51b1b24615cf
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 24 14:41:57 2026 +0200
docs(message_bus): demote broken intra-doc links to code spans
`LABEL_REPLICA`, `LABEL_CLIENT`, `handshake_verify`, and
`TcpTransportWriter::send_batch` were not in rustdoc scope at their
reference sites, triggering broken-link warnings. Downgrade to plain
code spans. No behavior change.
---
core/message_bus/src/auth.rs | 4 ++--
core/message_bus/src/replica_listener.rs | 2 +-
core/message_bus/src/transports/tcp.rs | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/core/message_bus/src/auth.rs b/core/message_bus/src/auth.rs
index dd90b18d2..2fc2819f9 100644
--- a/core/message_bus/src/auth.rs
+++ b/core/message_bus/src/auth.rs
@@ -101,8 +101,8 @@ impl AuthKind {
/// All fields come from the header (`GenericHeader.cluster`,
/// `GenericHeader.replica`, `GenericHeader.release`) plus the envelope
/// timestamp and nonce. `label` is the plane-tagged domain-separation
-/// prefix; pick [`LABEL_REPLICA`] for the consensus plane and
-/// [`LABEL_CLIENT`] for the SDK-client plane (Phase 2).
+/// prefix; pick `LABEL_REPLICA` for the consensus plane and
+/// `LABEL_CLIENT` for the SDK-client plane (Phase 2).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AuthChallenge {
pub cluster: u128,
diff --git a/core/message_bus/src/replica_listener.rs b/core/message_bus/src/replica_listener.rs
index 6a0789369..db35362be 100644
--- a/core/message_bus/src/replica_listener.rs
+++ b/core/message_bus/src/replica_listener.rs
@@ -117,7 +117,7 @@ pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyErr
/// `nonces` is the bus-resident per-peer nonce dedup store. The listener
/// reaches into it via `borrow_mut().entry(peer).or_default()` for each
/// handshake; the borrow is held only across the synchronous
-/// [`handshake_verify`], never across an `await`, so the `RefCell` is
+/// `handshake_verify`, never across an `await`, so the `RefCell` is
/// safe under the single-threaded compio runtime.
#[allow(clippy::future_not_send)]
#[allow(clippy::too_many_arguments)]
diff --git a/core/message_bus/src/transports/tcp.rs b/core/message_bus/src/transports/tcp.rs
index 79694b06d..5864c39e2 100644
--- a/core/message_bus/src/transports/tcp.rs
+++ b/core/message_bus/src/transports/tcp.rs
@@ -19,7 +19,7 @@
//!
//! Behavior-preserving wrappers around `compio::net::TcpListener`,
//! `compio::net::TcpStream`, and the split halves. The hot path on
-//! [`TcpTransportWriter::send_batch`] is identical to
+//! `TcpTransportWriter::send_batch` is identical to
//! [`crate::writer_task::run`]'s inner `write_vectored_all` call: one
//! syscall per batch, zero intermediate copies of `Frozen`.
//!
(iggy) 11/14: docs(message_bus): demote final broken intra-doc link in connector.rs
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/message-bus-transports
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 13d2e5947f3cf411887a12f9166267d4d617cd44
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 10:49:48 2026 +0200
docs(message_bus): demote final broken intra-doc link in connector.rs
`MessageBusConfig::default` is not in connector.rs's rustdoc scope.
Demote the stray bracketed link to a plain code span, matching the
pattern 1b163dad2 applied to LABEL_REPLICA, LABEL_CLIENT,
handshake_verify, and TcpTransportWriter::send_batch. Doc warning
count on this branch falls to one (the pre-existing session_manager
ClientTable link in core/server-ng, unchanged from master).
---
core/message_bus/src/connector.rs | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/core/message_bus/src/connector.rs b/core/message_bus/src/connector.rs
index a3b5a1c54..f8a786cc2 100644
--- a/core/message_bus/src/connector.rs
+++ b/core/message_bus/src/connector.rs
@@ -44,7 +44,7 @@ use tracing::{debug, info, warn};
///
/// Equivalent to `MessageBusConfig::default().reconnect_period`; exposed
/// as a named const for test / bench ergonomics. Kept in sync with the
-/// [`MessageBusConfig::default`] impl. Remove once the configs-crate
+/// `MessageBusConfig::default` impl. Remove once the configs-crate
/// migration lands and bootstrap always reads the period from
/// `ServerConfig`.
pub const DEFAULT_RECONNECT_PERIOD: Duration = Duration::from_secs(5);