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 11:28:53 UTC
(iggy) branch feat/message-bus-transports updated (125594897 -> 8d82f1673)
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
from 125594897 feat(message_bus): add install_client_quic_conn + install_client_ws_stream
new bc77c69a7 feat(message_bus): add listen-address overlap validator
new 3eb4a7ddf feat(message_bus): add shard-0 QUIC client listener
new b15f39667 feat(message_bus): add WS pre-upgrade listener with cross-shard fd-ship
new 8d82f1673 feat(message_bus): wire QUIC + WS binds into start_on_shard_zero
The 4 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.
Summary of changes:
core/message_bus/src/client_listener_quic.rs | 112 +++++++++++
.../{client_listener.rs => client_listener_ws.rs} | 66 ++++---
core/message_bus/src/installer.rs | 60 ++++++
core/message_bus/src/lib.rs | 29 +++
core/message_bus/src/replica_io.rs | 205 ++++++++++++++++++++-
core/message_bus/src/transports/quic.rs | 15 ++
core/message_bus/tests/common/mod.rs | 54 +++++-
core/message_bus/tests/quic_client_roundtrip.rs | 124 +++++++++++++
core/message_bus/tests/shard_zero_gating.rs | 88 ++++++++-
core/message_bus/tests/ws_client_roundtrip.rs | 168 +++++++++++++++++
core/shard/src/coordinator.rs | 32 ++++
core/shard/src/lib.rs | 29 ++-
core/shard/src/router.rs | 10 +
13 files changed, 941 insertions(+), 51 deletions(-)
create mode 100644 core/message_bus/src/client_listener_quic.rs
copy core/message_bus/src/{client_listener.rs => client_listener_ws.rs} (51%)
create mode 100644 core/message_bus/tests/quic_client_roundtrip.rs
create mode 100644 core/message_bus/tests/ws_client_roundtrip.rs
(iggy) 03/04: feat(message_bus): add WS pre-upgrade listener with cross-shard fd-ship
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 b15f3966797c9363f044fed71bcce32c3446a925
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 13:22:02 2026 +0200
feat(message_bus): add WS pre-upgrade listener with cross-shard fd-ship
Wire the shard-0 WebSocket client listener so accepted TCP sockets
ride the existing fd-ship plumbing pre-upgrade. The HTTP-Upgrade
handshake (and the iggy.consensus.v1 subprotocol enforcement) runs on
the owning shard inside the new install_client_ws_fd entry on the
ConnectionInstaller trait, which wraps the dup'd fd and drives
compio_ws::accept_hdr_async with a tungstenite Callback that returns
HTTP 400 on missing or wrong subprotocol.
Reverses the 2026-04-27 design correction in core/message_bus/CLAUDE.md
that claimed WS connections must terminate on shard 0: the !Send
constraint on compio_ws::WebSocketStream<TcpStream> is post-upgrade,
not pre-upgrade. The shipped fd is plain TCP at ship-time, so
invariant I5 (fd-delegation is TCP-only) holds.
What landed:
- core/message_bus/src/client_listener_ws.rs: bind + run for the
pre-upgrade TCP listener. No upgrade work runs on shard 0; the
callback (AcceptedWsClientFn) hands the raw stream off.
- core/message_bus/src/installer.rs: install_client_ws_fd on the
ConnectionInstaller trait + Rc<IggyMessageBus> impl. Wraps fd,
spawns a tracked task that runs compio_ws::accept_hdr_async with
the new ws_subprotocol_callback, then delegates to the existing
install_client_ws_stream on success.
- core/message_bus/src/lib.rs: AcceptedWsClientFn type alias.
- core/shard/src/lib.rs: new ShardFramePayload::ClientWsConnectionSetup
variant alongside ClientConnectionSetup. Doc on ClientConnectionSetup
refreshed to drop the obsolete "WS terminates on shard 0" claim.
- core/shard/src/coordinator.rs: new delegate_ws_client(stream) that
mirrors delegate_client. Same round-robin target, same mint_client_id,
same dup_fd; ships ClientWsConnectionSetup instead.
- core/shard/src/router.rs: match arm for the new variant calls
bus.install_client_ws_fd(fd, client_id, on_request.clone()).
- core/message_bus/tests/common/mod.rs: install_ws_clients_locally
helper for single-shard tests; dups locally and routes through
install_client_ws_fd, exercising the same accept_hdr_async path.
- core/message_bus/tests/ws_client_roundtrip.rs: 3 tests.
handshake_succeeds_and_request_reaches_handler verifies the install
+ Request arrives at the server-side handler via a side-channel
signal. missing_subprotocol_rejected_at_handshake and
wrong_subprotocol_rejected_at_handshake verify that the
ws_subprotocol_callback rejects with HTTP 400.
Known follow-up (out of P5 scope, Phase 7 work): the WS dispatcher in
transports/ws.rs cannot drive a full client -> server -> client
request/reply round trip. Phase B's stream.read().await blocks the
dispatcher and cannot react to outbound queue activity, and compio's
"no buffer drop with pending I/O" rule prevents canceling the read
via select! on out_rx.recv(). The unidirectional unit tests in
transports/ws.rs do not exercise this path; the integration test here
sends a Request and verifies arrival without round-tripping the Reply.
146/146 tests pass on cargo nextest run -p message_bus -p server-ng
-p shard. Workspace clippy clean. cargo doc clean modulo the
pre-existing core/shard/src/coordinator.rs:308 warning inherited from
master.
---
core/message_bus/src/client_listener_ws.rs | 103 ++++++++++++++++
core/message_bus/src/installer.rs | 60 +++++++++
core/message_bus/src/lib.rs | 1 +
core/message_bus/tests/common/mod.rs | 28 ++++-
core/message_bus/tests/ws_client_roundtrip.rs | 168 ++++++++++++++++++++++++++
core/shard/src/coordinator.rs | 32 +++++
core/shard/src/lib.rs | 29 +++--
core/shard/src/router.rs | 10 ++
8 files changed, 421 insertions(+), 10 deletions(-)
diff --git a/core/message_bus/src/client_listener_ws.rs b/core/message_bus/src/client_listener_ws.rs
new file mode 100644
index 000000000..50b859c17
--- /dev/null
+++ b/core/message_bus/src/client_listener_ws.rs
@@ -0,0 +1,103 @@
+// 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.
+
+//! Pre-upgrade TCP listener for consensus-protocol WebSocket clients.
+//!
+//! Runs only on shard 0. The accept loop performs no protocol work
+//! beyond `TcpListener::accept`: every accepted stream is handed
+//! verbatim to the supplied callback, which dups the fd and ships a
+//! [`ShardFramePayload::ClientWsConnectionSetup`] frame to the
+//! round-robin-selected target shard. The HTTP-Upgrade handshake (and
+//! the `iggy.consensus.v1` subprotocol enforcement) runs on the owning
+//! shard inside [`crate::installer::install_client_ws_fd`].
+//!
+//! The fd at ship-time is plain TCP; invariant I5 (fd-delegation is
+//! TCP-only) holds. The WS state machine only materialises after the
+//! upgrade, on the owning shard, where it can stay non-`Send`.
+//!
+//! `ShardFramePayload::ClientWsConnectionSetup` is defined in
+//! `core/shard/src/lib.rs`; the rustdoc cannot intra-link across crates
+//! without pulling `shard` in as a doc-only dep.
+
+use crate::AcceptedWsClientFn;
+use crate::lifecycle::ShutdownToken;
+use compio::net::{SocketOpts, TcpListener};
+use futures::FutureExt;
+use iggy_common::IggyError;
+use std::net::SocketAddr;
+use std::rc::Rc;
+use tracing::{debug, error, info};
+
+/// Bind the WS pre-upgrade TCP listener and return the bound address.
+///
+/// Mirrors [`crate::client_listener::bind`] in shape (`TCP_NODELAY` +
+/// `SO_KEEPALIVE` on by default). The receiving shard re-applies socket
+/// options on the dup'd fd via the existing client-install path, so
+/// kernel-level options propagate end-to-end.
+///
+/// # Errors
+///
+/// Returns [`IggyError::CannotBindToSocket`] if the bind fails.
+#[allow(clippy::future_not_send)]
+pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> {
+ // `SO_REUSEPORT` intentionally not set: only shard 0 binds the WS
+ // listener (per invariant I6). The shard-0 coordinator round-robins
+ // accepts to owning shards via `ShardFramePayload::ClientWsConnectionSetup`.
+ let opts = SocketOpts::new().nodelay(true).keepalive(true);
+ let listener = TcpListener::bind_with_options(addr, &opts)
+ .await
+ .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?;
+ let actual = listener
+ .local_addr()
+ .map_err(|e| IggyError::IoError(e.to_string()))?;
+ Ok((listener, actual))
+}
+
+/// Run the WS pre-upgrade listener accept loop until the shutdown
+/// token fires.
+///
+/// Each accepted [`compio::net::TcpStream`] is handed verbatim to
+/// `on_accepted` (no upgrade attempted here). The callback owns the
+/// stream from that point on.
+#[allow(clippy::future_not_send)]
+pub async fn run(listener: TcpListener, token: ShutdownToken, on_accepted: AcceptedWsClientFn) {
+ info!(
+ "Consensus WS client listener (pre-upgrade) accepting on {:?}",
+ listener.local_addr().ok()
+ );
+
+ let on_accepted: Rc<_> = on_accepted;
+ loop {
+ futures::select! {
+ () = token.wait().fuse() => {
+ debug!("Consensus WS client listener shutting down");
+ break;
+ }
+ result = listener.accept().fuse() => {
+ match result {
+ Ok((stream, peer_addr)) => {
+ debug!(%peer_addr, "WS client TCP accepted, delegating fd pre-upgrade");
+ on_accepted(stream);
+ }
+ Err(e) => {
+ error!("Consensus WS client listener accept failed: {e}");
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/core/message_bus/src/installer.rs b/core/message_bus/src/installer.rs
index 0b52304c7..f7fc7b861 100644
--- a/core/message_bus/src/installer.rs
+++ b/core/message_bus/src/installer.rs
@@ -62,6 +62,15 @@ pub trait ConnectionInstaller {
/// encoded in the top 16 bits of `client_id`.
fn install_client_fd(&self, fd: DupedFd, client_id: u128, on_request: RequestHandler);
+ /// Same for an SDK WebSocket client's pre-upgrade TCP fd. The
+ /// receiving shard wraps the fd, runs `compio_ws::accept_hdr_async`
+ /// with the iggy.consensus.v1 subprotocol callback to drive the
+ /// HTTP-Upgrade handshake, then installs WS reader / writer tasks
+ /// via [`install_client_ws_stream`] on success. On handshake
+ /// failure (e.g. wrong / missing subprotocol) the fd is closed by
+ /// dropping the wrapping `TcpStream`.
+ fn install_client_ws_fd(&self, fd: DupedFd, client_id: u128, on_request: RequestHandler);
+
/// Update the replica -> owning shard mapping used by the `send_to_replica`
/// slow path on non-owning shards.
fn set_shard_mapping(&self, replica: u8, owning_shard: u16);
@@ -82,6 +91,24 @@ impl ConnectionInstaller for Rc<IggyMessageBus> {
install_client_stream(self, client_id, stream, on_request);
}
+ fn install_client_ws_fd(&self, fd: DupedFd, client_id: u128, on_request: RequestHandler) {
+ let stream = fd_transfer::wrap_duped_fd(fd);
+ let bus = Self::clone(self);
+ let handle = compio::runtime::spawn(async move {
+ match compio_ws::accept_hdr_async(stream, ws_subprotocol_callback).await {
+ Ok(ws) => {
+ if !bus.is_shutting_down() {
+ install_client_ws_stream(&bus, client_id, ws, on_request);
+ }
+ }
+ Err(e) => {
+ warn!(client_id, "WS upgrade failed: {e}");
+ }
+ }
+ });
+ self.track_background(handle);
+ }
+
fn set_shard_mapping(&self, replica: u8, owning_shard: u16) {
IggyMessageBus::set_shard_mapping(self, replica, owning_shard);
}
@@ -91,6 +118,39 @@ impl ConnectionInstaller for Rc<IggyMessageBus> {
}
}
+/// HTTP-Upgrade callback for the WS client plane.
+///
+/// Inspects `Sec-WebSocket-Protocol`. Accepts only the exact value
+/// [`crate::transports::ws::WS_SUBPROTOCOL`] (`iggy.consensus.v1`); any
+/// other value (or absence) yields HTTP 400 with a body naming the
+/// expected subprotocol. The accepted value is mirrored back on the
+/// response so the negotiated subprotocol is unambiguous to the
+/// client.
+#[allow(clippy::result_large_err)] // tungstenite-defined Callback signature; not on hot path
+fn ws_subprotocol_callback(
+ req: &tungstenite::handshake::server::Request,
+ mut resp: tungstenite::handshake::server::Response,
+) -> Result<tungstenite::handshake::server::Response, tungstenite::handshake::server::ErrorResponse>
+{
+ let want = crate::transports::ws::WS_SUBPROTOCOL.as_bytes();
+ let proto = req
+ .headers()
+ .get(tungstenite::http::header::SEC_WEBSOCKET_PROTOCOL);
+ if proto.is_some_and(|hv| hv.as_bytes() == want) {
+ resp.headers_mut().insert(
+ tungstenite::http::header::SEC_WEBSOCKET_PROTOCOL,
+ tungstenite::http::HeaderValue::from_static(crate::transports::ws::WS_SUBPROTOCOL),
+ );
+ return Ok(resp);
+ }
+ let mut err = tungstenite::http::Response::new(Some(format!(
+ "missing or wrong subprotocol; expected {}",
+ crate::transports::ws::WS_SUBPROTOCOL
+ )));
+ *err.status_mut() = tungstenite::http::StatusCode::BAD_REQUEST;
+ Err(err)
+}
+
/// TCP entry point: apply socket options (keepalive, `TCP_NODELAY`) on
/// the raw stream and delegate to the transport-generic install path.
///
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index 628832cf3..271e6b000 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -82,6 +82,7 @@ pub mod auth_config;
pub mod cache;
pub mod client_listener;
pub mod client_listener_quic;
+pub mod client_listener_ws;
pub mod config;
pub mod connector;
mod error;
diff --git a/core/message_bus/tests/common/mod.rs b/core/message_bus/tests/common/mod.rs
index 4f7e88821..8ed91921a 100644
--- a/core/message_bus/tests/common/mod.rs
+++ b/core/message_bus/tests/common/mod.rs
@@ -27,11 +27,13 @@
#![allow(dead_code)] // each test binary uses a subset
use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message};
+use message_bus::ConnectionInstaller;
use message_bus::auth::{StaticSharedSecret, TokenSource};
use message_bus::client_listener::RequestHandler;
use message_bus::replica_listener::MessageHandler;
use message_bus::{
- AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, IggyMessageBus, installer,
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedWsClientFn, IggyMessageBus,
+ fd_transfer, installer,
};
use std::cell::Cell;
use std::net::SocketAddr;
@@ -120,3 +122,27 @@ pub fn install_quic_clients_locally(
);
})
}
+
+/// Build an [`AcceptedWsClientFn`] that mints a local client id, dups
+/// the accepted fd (mirroring the production cross-shard fd-ship
+/// path), and hands it to [`ConnectionInstaller::install_client_ws_fd`]
+/// on the given bus. Single-shard tests bypass the inter-shard
+/// channel by dup'ing locally; the install path runs `accept_hdr_async`,
+/// the subprotocol callback, and `install_client_ws_stream` exactly as
+/// the production owning-shard router would.
+#[must_use]
+pub fn install_ws_clients_locally(
+ bus: Rc<IggyMessageBus>,
+ on_request: RequestHandler,
+) -> AcceptedWsClientFn {
+ let counter: Rc<Cell<u128>> = Rc::new(Cell::new(1));
+ let shard_id = u128::from(bus.shard_id());
+ Rc::new(move |stream| {
+ let seq = counter.get();
+ counter.set(seq.wrapping_add(1));
+ let client_id = (shard_id << 112) | seq;
+ let fd = fd_transfer::dup_fd(&stream).expect("dup_fd");
+ drop(stream);
+ bus.install_client_ws_fd(fd, client_id, on_request.clone());
+ })
+}
diff --git a/core/message_bus/tests/ws_client_roundtrip.rs b/core/message_bus/tests/ws_client_roundtrip.rs
new file mode 100644
index 000000000..d8d8854f4
--- /dev/null
+++ b/core/message_bus/tests/ws_client_roundtrip.rs
@@ -0,0 +1,168 @@
+// 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.
+
+//! End-to-end: a real WebSocket client connects to the consensus WS
+//! pre-upgrade listener on shard 0; the listener's callback dups the
+//! TCP fd and (on the same shard, locally) hands it to
+//! `install_client_ws_fd`, which runs `compio_ws::accept_hdr_async`
+//! with the iggy.consensus.v1 subprotocol callback before installing
+//! the WS connection.
+//!
+//! Coverage:
+//!
+//! - **Positive**: client requests `iggy.consensus.v1`, handshake
+//! succeeds, a Request frame reaches the server-side handler.
+//! Caveat: a full Request -> Reply round trip would deadlock
+//! the per-connection WS dispatcher (`transports::ws.rs`); its
+//! phase-alternating drain-then-read loop blocks Phase B on
+//! `stream.read().await` and cannot react to outbound queue
+//! activity, so the server's Reply never reaches the wire while
+//! the client is parked in Phase B waiting for it. Tracked as a
+//! Phase 7 follow-up in the silverhand transport-plan; the design
+//! note in the dispatcher comment overstates its request/reply
+//! support.
+//! - **Negative (missing)**: client omits the `Sec-WebSocket-Protocol`
+//! header; server returns HTTP 400; handshake fails on the client.
+//! - **Negative (wrong)**: client requests an unrelated subprotocol;
+//! server returns HTTP 400.
+
+mod common;
+
+use common::{header_only, install_ws_clients_locally, loopback};
+use compio::net::TcpStream;
+use iggy_binary_protocol::Command2;
+use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
+use iggy_binary_protocol::consensus::iobuf::Frozen;
+use message_bus::IggyMessageBus;
+use message_bus::client_listener::RequestHandler;
+use message_bus::client_listener_ws::{bind, run};
+use message_bus::transports::ws::{WS_SUBPROTOCOL, WsTransportConn};
+use message_bus::transports::{TransportConn, TransportWriter};
+use std::rc::Rc;
+use std::time::Duration;
+
+#[compio::test]
+async fn handshake_succeeds_and_request_reaches_handler() {
+ let bus = Rc::new(IggyMessageBus::new(0));
+
+ // Side-channel signal: handler fires once when it sees the Request.
+ let (tx_seen, rx_seen) = async_channel::bounded::<u128>(1);
+ let on_request: RequestHandler = Rc::new(move |client_id, msg| {
+ assert_eq!(msg.header().command, Command2::Request);
+ let _ = tx_seen.try_send(client_id);
+ });
+
+ let (listener, server_addr) = bind(loopback()).await.expect("bind");
+ let token = bus.token();
+ let on_accepted = install_ws_clients_locally(bus.clone(), on_request);
+ let accept_handle = compio::runtime::spawn(async move {
+ run(listener, token, on_accepted).await;
+ });
+ bus.track_background(accept_handle);
+
+ // Dial as a real WS client with the iggy.consensus.v1 subprotocol.
+ let client_tcp = TcpStream::connect(server_addr).await.unwrap();
+ let url = format!("ws://{server_addr}/");
+ let req = tungstenite::ClientRequestBuilder::new(url.parse().unwrap())
+ .with_sub_protocol(WS_SUBPROTOCOL);
+ let (ws_client, _resp) = compio_ws::client_async(req, client_tcp)
+ .await
+ .expect("ws handshake");
+
+ // Reuse the existing WsTransportConn split for client-side framing.
+ // Drop the client's reader half; we only verify that the server saw
+ // the Request. A full round trip is blocked by the dispatcher
+ // limitation called out in the module-level comment.
+ let conn = WsTransportConn::new(ws_client);
+ let (_reader, mut writer) = conn.into_split();
+
+ let request = header_only(Command2::Request, 42, 0).into_frozen();
+ let mut batch: Vec<Frozen<MESSAGE_ALIGN>> = vec![request];
+ writer.send_batch(&mut batch).await.expect("client send");
+ assert!(batch.is_empty(), "send_batch must drain the Vec");
+
+ // Wait for the handler to observe the Request.
+ let observed = compio::time::timeout(Duration::from_secs(2), rx_seen.recv())
+ .await
+ .expect("handler must fire within 2 s")
+ .expect("handler must signal");
+ assert_eq!(
+ observed >> 112,
+ 0,
+ "client_id top 16 bits must encode shard 0"
+ );
+
+ bus.shutdown(Duration::from_secs(2)).await;
+}
+
+#[compio::test]
+async fn missing_subprotocol_rejected_at_handshake() {
+ let bus = Rc::new(IggyMessageBus::new(0));
+
+ // Handler should never run (handshake fails before install).
+ let on_request: RequestHandler = Rc::new(|_, _| {
+ panic!("handler must not be called when subprotocol is missing");
+ });
+
+ let (listener, server_addr) = bind(loopback()).await.expect("bind");
+ let token = bus.token();
+ let on_accepted = install_ws_clients_locally(bus.clone(), on_request);
+ let accept_handle = compio::runtime::spawn(async move {
+ run(listener, token, on_accepted).await;
+ });
+ bus.track_background(accept_handle);
+
+ // Dial without the subprotocol header.
+ let client_tcp = TcpStream::connect(server_addr).await.unwrap();
+ let url = format!("ws://{server_addr}/");
+ let result = compio_ws::client_async(url, client_tcp).await;
+ assert!(
+ result.is_err(),
+ "WS handshake should fail when subprotocol is absent"
+ );
+
+ bus.shutdown(Duration::from_secs(1)).await;
+}
+
+#[compio::test]
+async fn wrong_subprotocol_rejected_at_handshake() {
+ let bus = Rc::new(IggyMessageBus::new(0));
+
+ let on_request: RequestHandler = Rc::new(|_, _| {
+ panic!("handler must not be called when subprotocol is wrong");
+ });
+
+ let (listener, server_addr) = bind(loopback()).await.expect("bind");
+ let token = bus.token();
+ let on_accepted = install_ws_clients_locally(bus.clone(), on_request);
+ let accept_handle = compio::runtime::spawn(async move {
+ run(listener, token, on_accepted).await;
+ });
+ bus.track_background(accept_handle);
+
+ let client_tcp = TcpStream::connect(server_addr).await.unwrap();
+ let url = format!("ws://{server_addr}/");
+ let req = tungstenite::ClientRequestBuilder::new(url.parse().unwrap())
+ .with_sub_protocol("definitely.not.iggy");
+ let result = compio_ws::client_async(req, client_tcp).await;
+ assert!(
+ result.is_err(),
+ "WS handshake should fail when subprotocol is wrong"
+ );
+
+ bus.shutdown(Duration::from_secs(1)).await;
+}
diff --git a/core/shard/src/coordinator.rs b/core/shard/src/coordinator.rs
index edd88e07c..abe99b09b 100644
--- a/core/shard/src/coordinator.rs
+++ b/core/shard/src/coordinator.rs
@@ -229,6 +229,38 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
Ok(client_id)
}
+ /// Ship a WebSocket client's pre-upgrade TCP connection to the next
+ /// round-robin target shard.
+ ///
+ /// Identical wire path to [`Self::delegate_client`] but ships
+ /// [`ShardFramePayload::ClientWsConnectionSetup`] so the receiving
+ /// shard runs `compio_ws::accept_hdr_async` (HTTP-Upgrade +
+ /// `iggy.consensus.v1` subprotocol enforcement) before installing
+ /// the connection. The fd at ship-time is plain TCP; the WS state
+ /// machine only materialises post-upgrade on the owning shard.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error when `dup(2)` fails or the target shard's inbox
+ /// refuses the setup frame.
+ pub fn delegate_ws_client(&self, stream: TcpStream) -> Result<u128, SendError> {
+ let target = self.next_client_target();
+ let client_id = self.mint_client_id(target);
+
+ let fd = fd_transfer::dup_fd(&stream).map_err(SendError::DupFailed)?;
+ let setup = ShardFramePayload::ClientWsConnectionSetup { fd, client_id };
+ if let Err(e) = self.senders[target as usize].try_send(ShardFrame::lifecycle(setup)) {
+ warn!(
+ client_id,
+ target, "delegate_ws_client try_send failed: {e:?}"
+ );
+ return Err(SendError::RoutingFailed(target));
+ }
+
+ drop(stream);
+ Ok(client_id)
+ }
+
/// Broadcast a `ReplicaMappingClear` to every shard. Used by the
/// `ConnectionLost` handler before the next `delegate_replica` runs.
pub fn broadcast_mapping_clear(&self, replica_id: u8) {
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index e420e508c..4eb2a074a 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -199,16 +199,27 @@ 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 distributes an inbound SDK WebSocket client's pre-upgrade
+ /// TCP connection fd to the owning shard. The HTTP-Upgrade handshake
+ /// has NOT run yet at this point: the fd is plain TCP, the dup is
+ /// safe per invariant I5 (fd-delegation is TCP-only), and
+ /// `compio_ws::WebSocketStream<TcpStream>`'s `!Send` constraint
+ /// (compio `Rc<...>` driver state, post-upgrade) does not apply.
+ /// The receiving shard wraps the fd, runs `compio_ws::accept_hdr_async`
+ /// with the iggy.consensus.v1 subprotocol callback, then installs
+ /// client reader / writer tasks locally via
+ /// `message_bus::installer::install_client_ws_fd`. Owning shard is
+ /// encoded in the top 16 bits of `client_id`.
+ ///
+ /// QUIC clients deliberately do NOT get an analog variant: a
+ /// `compio_quic::Endpoint` binds one UDP socket and demuxes incoming
+ /// packets to per-connection `quinn-proto::Connection` objects by
+ /// Connection ID. Per-connection TLS / packet-number / congestion
+ /// state is non-serialisable and tied to the endpoint's reactor.
+ /// Shard 0 therefore terminates QUIC locally and uses the existing
+ /// `ForwardClientSend` variant for outbound traffic.
+ ClientWsConnectionSetup { 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.
ReplicaMappingUpdate { replica_id: u8, owning_shard: u16 },
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 95c1be665..6ea24225e 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -317,6 +317,16 @@ where
self.bus
.install_client_fd(fd, client_id, self.on_client_request.clone());
}
+ crate::ShardFramePayload::ClientWsConnectionSetup { fd, client_id } => {
+ tracing::info!(
+ shard = self.id,
+ client_id,
+ raw_fd = fd.as_raw_fd(),
+ "installing delegated WS client fd (pre-upgrade)"
+ );
+ self.bus
+ .install_client_ws_fd(fd, client_id, self.on_client_request.clone());
+ }
crate::ShardFramePayload::ReplicaMappingUpdate {
replica_id,
owning_shard,
(iggy) 04/04: feat(message_bus): wire QUIC + WS binds into start_on_shard_zero
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 8d82f1673a79bfdbf7216dab02c3fda5e06bdfd4
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 13:27:03 2026 +0200
feat(message_bus): wire QUIC + WS binds into start_on_shard_zero
Extends start_on_shard_zero with five new parameters: ws_listen_addr,
quic_listen_addr, quic_credentials (cert chain + key DER newtype),
on_accepted_ws_client, on_accepted_quic_client. Each new listener
binds only when its parameter trio is fully populated; partially
populated trios return IggyError::InvalidConfiguration so operator
misconfigurations surface at boot. Pre-bind validation calls
assert_listen_addrs_distinct on the populated TCP-family slots.
BoundPlanes grows two optional SocketAddr fields (ws, quic) populated
from each successful bind. start_on_shard_zero_default mirrors with
None defaults so existing TCP-only call sites compile unchanged.
Drive-by: assert_listen_addrs_distinct now skips the equality check
when port == 0. Two slots at port 0 ("OS-assigned") receive distinct
kernel ports at bind time and are not a real conflict; loopback
fixtures rely on this.
New shard_zero_gating::shard_zero_binds_all_four_planes_when_configured
test exercises the full bind path with a self-signed rcgen cert and
verifies all four BoundPlanes fields are populated with distinct
TCP-family ports. 147/147 tests pass on cargo nextest run -p
message_bus -p server-ng -p shard. Workspace clippy clean.
---
core/message_bus/src/replica_io.rs | 115 +++++++++++++++++++++++++---
core/message_bus/tests/shard_zero_gating.rs | 88 ++++++++++++++++++++-
2 files changed, 192 insertions(+), 11 deletions(-)
diff --git a/core/message_bus/src/replica_io.rs b/core/message_bus/src/replica_io.rs
index 2b3b3802a..44fbab75a 100644
--- a/core/message_bus/src/replica_io.rs
+++ b/core/message_bus/src/replica_io.rs
@@ -35,17 +35,39 @@ use std::rc::Rc;
use std::time::Duration;
use iggy_common::IggyError;
+use rustls::pki_types::{CertificateDer, PrivateKeyDer};
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};
+use crate::transports::quic::server_config_with_cert;
+use crate::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedWsClientFn, IggyMessageBus,
+ client_listener, client_listener_quic, client_listener_ws,
+};
+
+/// QUIC server credentials passed by the bootstrap layer.
+///
+/// The cert chain is the leaf-first sequence rustls expects; the key
+/// is the server's private key in DER form. Tests use rcgen to mint a
+/// throwaway pair; production callers load real PKI material via
+/// `core/server-ng`'s `[quic.certificate]` config section.
+pub struct QuicServerCredentials {
+ pub cert_chain: Vec<CertificateDer<'static>>,
+ pub key_der: PrivateKeyDer<'static>,
+}
/// Bound addresses returned to shard 0 after the listeners come up.
-#[derive(Debug, Clone, Copy)]
+///
+/// `ws` and `quic` are populated only when the corresponding
+/// `start_on_shard_zero` parameters are `Some`; an unconfigured plane
+/// stays `None`.
+#[derive(Debug, Clone)]
pub struct BoundPlanes {
pub replica: SocketAddr,
pub client: SocketAddr,
+ pub ws: Option<SocketAddr>,
+ pub quic: Option<SocketAddr>,
}
/// Boot-time check: every TCP listener address must occupy a distinct
@@ -80,8 +102,12 @@ pub fn assert_listen_addrs_distinct(
for j in (i + 1)..tcp_slots.len() {
let (a_name, a) = tcp_slots[i];
let (b_name, b) = tcp_slots[j];
+ // Port 0 means "OS-assigned"; two slots at port 0 receive
+ // distinct kernel ports at bind time, so they don't
+ // conflict. Tests rely on this for loopback fixtures.
if let (Some(a), Some(b)) = (a, b)
&& a == b
+ && a.port() != 0
{
panic!("listener address conflict: {a_name} and {b_name} both bind to {a}");
}
@@ -89,29 +115,47 @@ pub fn assert_listen_addrs_distinct(
}
}
-/// Bind the replica + client listeners on shard 0 and start the outbound
-/// replica connector. Non-zero shards early-return `Ok(None)`.
+/// Bind the replica + client listeners on shard 0 and start the
+/// outbound replica connector. Optionally bind WS and QUIC client
+/// listeners alongside. Non-zero shards early-return `Ok(None)`.
+///
+/// Each accepted / dialed connection is handed to the supplied
+/// delegate callback. The TCP callbacks (`on_accepted_replica`,
+/// `on_accepted_client`) and the WS callback (`on_accepted_ws_client`)
+/// are responsible for the dup-fd + inter-shard send. The QUIC
+/// callback (`on_accepted_quic_client`) installs locally on shard 0
+/// (QUIC has no cross-shard handover).
///
-/// Each accepted / dialed TCP stream is handed to the supplied delegate
-/// callback: `on_accepted_replica` for inbound replicas and for freshly
-/// dialed higher-id peers; `on_accepted_client` for SDK client accepts.
-/// The callbacks are responsible for the dup-fd + inter-shard send.
+/// `ws_listen_addr` / `on_accepted_ws_client` are paired: if either is
+/// `Some`, both must be. Same for the QUIC trio
+/// (`quic_listen_addr` / `quic_credentials` / `on_accepted_quic_client`).
+///
+/// # Panics
+///
+/// Panics on TCP-family `(ip, port)` overlap among the populated
+/// `replica`, `client`, `ws` slots — see
+/// [`assert_listen_addrs_distinct`].
///
/// # Errors
///
-/// Returns `IggyError::CannotBindToSocket` if either listener bind fails.
+/// Returns `IggyError::CannotBindToSocket` if any listener bind fails.
#[allow(clippy::future_not_send)]
#[allow(clippy::too_many_arguments)]
pub async fn start_on_shard_zero(
bus: &Rc<IggyMessageBus>,
replica_listen_addr: SocketAddr,
client_listen_addr: SocketAddr,
+ ws_listen_addr: Option<SocketAddr>,
+ quic_listen_addr: Option<SocketAddr>,
+ quic_credentials: Option<QuicServerCredentials>,
cluster_id: u128,
self_id: u8,
replica_count: u8,
peers: Vec<(u8, SocketAddr)>,
on_accepted_replica: AcceptedReplicaFn,
on_accepted_client: AcceptedClientFn,
+ on_accepted_ws_client: Option<AcceptedWsClientFn>,
+ on_accepted_quic_client: Option<AcceptedQuicClientFn>,
reconnect_period: Duration,
token_source: Rc<dyn TokenSource>,
) -> Result<Option<BoundPlanes>, IggyError> {
@@ -119,6 +163,13 @@ pub async fn start_on_shard_zero(
return Ok(None);
}
+ assert_listen_addrs_distinct(
+ replica_listen_addr,
+ client_listen_addr,
+ ws_listen_addr,
+ quic_listen_addr,
+ );
+
let (replica_listener, replica_bound) = bind_replica_listener(replica_listen_addr).await?;
let (clients_listener, client_bound) = client_listener::bind(client_listen_addr).await?;
@@ -149,6 +200,40 @@ pub async fn start_on_shard_zero(
});
bus.track_background(client_handle);
+ let ws_bound = match (ws_listen_addr, on_accepted_ws_client) {
+ (Some(addr), Some(on_accepted_ws)) => {
+ let (ws_listener, ws_bound) = client_listener_ws::bind(addr).await?;
+ let token_for_ws = bus.token();
+ let ws_handle = compio::runtime::spawn(async move {
+ client_listener_ws::run(ws_listener, token_for_ws, on_accepted_ws).await;
+ });
+ bus.track_background(ws_handle);
+ Some(ws_bound)
+ }
+ (None, None) => None,
+ _ => {
+ return Err(IggyError::InvalidConfiguration);
+ }
+ };
+
+ let quic_bound = match (quic_listen_addr, quic_credentials, on_accepted_quic_client) {
+ (Some(addr), Some(creds), Some(on_accepted_quic)) => {
+ let server_config = server_config_with_cert(creds.cert_chain, creds.key_der)
+ .map_err(|e| IggyError::IoError(format!("QUIC server config build failed: {e}")))?;
+ let (endpoint, quic_bound) = client_listener_quic::bind(addr, server_config).await?;
+ let token_for_quic = bus.token();
+ let quic_handle = compio::runtime::spawn(async move {
+ client_listener_quic::run(endpoint, token_for_quic, on_accepted_quic).await;
+ });
+ bus.track_background(quic_handle);
+ Some(quic_bound)
+ }
+ (None, None, None) => None,
+ _ => {
+ return Err(IggyError::InvalidConfiguration);
+ }
+ };
+
start_connector(
bus,
cluster_id,
@@ -163,12 +248,19 @@ pub async fn start_on_shard_zero(
Ok(Some(BoundPlanes {
replica: replica_bound,
client: client_bound,
+ ws: ws_bound,
+ quic: quic_bound,
}))
}
/// [`start_on_shard_zero`] defaulting `reconnect_period` to the bus's
/// [`crate::MessageBusConfig::reconnect_period`].
///
+/// Leaves the WS + QUIC listener slots unconfigured (`None`).
+/// Convenience entry for TCP-only deployments and existing tests;
+/// prefer the full [`start_on_shard_zero`] in production where WS /
+/// QUIC come from `core/server-ng`'s config.
+///
/// # Errors
///
/// Returns `IggyError::CannotBindToSocket` if either listener bind fails.
@@ -191,12 +283,17 @@ pub async fn start_on_shard_zero_default(
bus,
replica_listen_addr,
client_listen_addr,
+ None,
+ None,
+ None,
cluster_id,
self_id,
replica_count,
peers,
on_accepted_replica,
on_accepted_client,
+ None,
+ None,
reconnect_period,
token_source,
)
diff --git a/core/message_bus/tests/shard_zero_gating.rs b/core/message_bus/tests/shard_zero_gating.rs
index 1b99d0a5b..3a1b1f265 100644
--- a/core/message_bus/tests/shard_zero_gating.rs
+++ b/core/message_bus/tests/shard_zero_gating.rs
@@ -21,12 +21,16 @@
mod common;
-use common::{install_clients_locally, install_replicas_locally, loopback, test_token_source};
+use common::{
+ install_clients_locally, install_quic_clients_locally, install_replicas_locally,
+ install_ws_clients_locally, loopback, test_token_source,
+};
use message_bus::IggyMessageBus;
use message_bus::client_listener::RequestHandler;
use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
-use message_bus::replica_io::start_on_shard_zero;
+use message_bus::replica_io::{QuicServerCredentials, start_on_shard_zero};
use message_bus::replica_listener::{MessageHandler, bind, run};
+use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
use std::net::SocketAddr;
use std::rc::Rc;
use std::time::Duration;
@@ -72,12 +76,17 @@ async fn shard_zero_binds_listener_and_starts_connector() {
&bus_zero,
loopback(),
loopback(),
+ None,
+ None,
+ None,
CLUSTER,
0,
2,
vec![(1u8, peer_addr)],
accepted_replica,
accepted_client,
+ None,
+ None,
DEFAULT_RECONNECT_PERIOD,
test_token_source(),
)
@@ -124,12 +133,17 @@ async fn non_zero_shard_skips_io() {
&bus_one,
loopback(),
loopback(),
+ None,
+ None,
+ None,
CLUSTER,
1,
2,
vec![(0u8, dead_peer)],
accepted_replica,
accepted_client,
+ None,
+ None,
DEFAULT_RECONNECT_PERIOD,
test_token_source(),
)
@@ -155,3 +169,73 @@ async fn non_zero_shard_skips_io() {
"helper must not register any background tasks on non-zero shards"
);
}
+
+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)
+}
+
+#[compio::test]
+async fn shard_zero_binds_all_four_planes_when_configured() {
+ // Idempotent across same-process retries.
+ let _ = rustls::crypto::ring::default_provider().install_default();
+
+ let bus_zero = Rc::new(IggyMessageBus::new(0));
+ let on_message: MessageHandler = Rc::new(|_, _| {});
+ let on_request: RequestHandler = Rc::new(|_, _| {});
+ let accepted_replica = install_replicas_locally(bus_zero.clone(), on_message.clone());
+ let accepted_client = install_clients_locally(bus_zero.clone(), on_request.clone());
+ let accepted_ws = install_ws_clients_locally(bus_zero.clone(), on_request.clone());
+ let accepted_quic = install_quic_clients_locally(bus_zero.clone(), on_request);
+
+ let (cert, key) = self_signed();
+ let creds = QuicServerCredentials {
+ cert_chain: vec![cert],
+ key_der: key,
+ };
+
+ let bound = start_on_shard_zero(
+ &bus_zero,
+ loopback(),
+ loopback(),
+ Some(loopback()),
+ Some(loopback()),
+ Some(creds),
+ CLUSTER,
+ 0,
+ 1,
+ vec![],
+ accepted_replica,
+ accepted_client,
+ Some(accepted_ws),
+ Some(accepted_quic),
+ DEFAULT_RECONNECT_PERIOD,
+ test_token_source(),
+ )
+ .await
+ .expect("start_on_shard_zero must succeed");
+
+ let bound = bound.expect("shard 0 must return bound listeners");
+ assert_ne!(bound.replica.port(), 0, "replica must bind a port");
+ assert_ne!(bound.client.port(), 0, "client must bind a port");
+ let ws = bound.ws.expect("ws plane must be bound");
+ let quic = bound.quic.expect("quic plane must be bound");
+ assert_ne!(ws.port(), 0, "ws must bind a port");
+ assert_ne!(quic.port(), 0, "quic must bind a port");
+
+ // Ports must be pairwise distinct on the TCP-family slots.
+ let tcp_ports = [bound.replica.port(), bound.client.port(), ws.port()];
+ for i in 0..tcp_ports.len() {
+ for j in (i + 1)..tcp_ports.len() {
+ assert_ne!(
+ tcp_ports[i], tcp_ports[j],
+ "TCP-family listeners must use distinct ports",
+ );
+ }
+ }
+
+ bus_zero.shutdown(Duration::from_secs(2)).await;
+}
(iggy) 01/04: feat(message_bus): add listen-address overlap validator
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 bc77c69a706b37b8d177172844f3b498d07a8333
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 12:56:16 2026 +0200
feat(message_bus): add listen-address overlap validator
assert_listen_addrs_distinct(replica, client, ws, quic) helper in
replica_io.rs catches operator misconfiguration at boot. Pairwise-
compare populated TCP-family SocketAddrs; same (ip, port) panics with
operator-actionable message naming the conflicting pair. UDP (QUIC)
shares port namespace with TCP and is allowed to share a port with
any TCP listener (separate kernel namespaces).
Backs the upcoming Phase 5 listener wiring: start_on_shard_zero will
call this helper before binding QUIC + WS listeners alongside the
existing TCP replica + client listeners.
---
core/message_bus/src/replica_io.rs | 90 ++++++++++++++++++++++++++++++++++++++
1 file changed, 90 insertions(+)
diff --git a/core/message_bus/src/replica_io.rs b/core/message_bus/src/replica_io.rs
index cc2eca8c1..f0abddecb 100644
--- a/core/message_bus/src/replica_io.rs
+++ b/core/message_bus/src/replica_io.rs
@@ -48,6 +48,47 @@ pub struct BoundPlanes {
pub client: SocketAddr,
}
+/// Boot-time check: every TCP listener address must occupy a distinct
+/// `(ip, port)` slot.
+///
+/// TCP listeners (`replica`, `client`, `ws`) all share the TCP port
+/// space, so any pair sharing `(ip, port)` is a bind-time conflict.
+/// The QUIC listener binds a UDP socket: it occupies a separate port
+/// namespace and is allowed to share a port with any TCP listener (a
+/// common operator choice for "QUIC on 443 over UDP, HTTPS on 443 over
+/// TCP"). UDP-vs-UDP conflicts are not possible today because QUIC is
+/// the only UDP listener.
+///
+/// # Panics
+///
+/// Panics with a message naming the conflicting pair. Boot-time
+/// validation; surfaces operator misconfiguration loudly rather than
+/// letting one listener silently lose a `EADDRINUSE` race against
+/// another.
+pub fn assert_listen_addrs_distinct(
+ replica: SocketAddr,
+ client: SocketAddr,
+ ws: Option<SocketAddr>,
+ _quic: Option<SocketAddr>,
+) {
+ let tcp_slots: [(&str, Option<SocketAddr>); 3] = [
+ ("replica", Some(replica)),
+ ("client", Some(client)),
+ ("ws", ws),
+ ];
+ for i in 0..tcp_slots.len() {
+ for j in (i + 1)..tcp_slots.len() {
+ let (a_name, a) = tcp_slots[i];
+ let (b_name, b) = tcp_slots[j];
+ if let (Some(a), Some(b)) = (a, b)
+ && a == b
+ {
+ panic!("listener address conflict: {a_name} and {b_name} both bind to {a}",);
+ }
+ }
+ }
+}
+
/// Bind the replica + client listeners on shard 0 and start the outbound
/// replica connector. Non-zero shards early-return `Ok(None)`.
///
@@ -161,3 +202,52 @@ pub async fn start_on_shard_zero_default(
)
.await
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn addr(port: u16) -> SocketAddr {
+ format!("127.0.0.1:{port}").parse().unwrap()
+ }
+
+ #[test]
+ fn distinct_addrs_pass() {
+ assert_listen_addrs_distinct(addr(9090), addr(8090), Some(addr(8092)), Some(addr(8080)));
+ }
+
+ #[test]
+ fn ws_unset_passes() {
+ assert_listen_addrs_distinct(addr(9090), addr(8090), None, Some(addr(8080)));
+ }
+
+ #[test]
+ fn quic_shares_port_with_tcp_replica_ok() {
+ // QUIC on UDP and replica on TCP can share a port (separate
+ // kernel namespaces). Operator choice "443 on TCP and UDP".
+ assert_listen_addrs_distinct(addr(443), addr(8090), Some(addr(8092)), Some(addr(443)));
+ }
+
+ #[test]
+ fn quic_shares_port_with_tcp_client_ok() {
+ assert_listen_addrs_distinct(addr(9090), addr(443), Some(addr(8092)), Some(addr(443)));
+ }
+
+ #[test]
+ #[should_panic(expected = "listener address conflict: replica and client")]
+ fn replica_client_overlap_panics() {
+ assert_listen_addrs_distinct(addr(8090), addr(8090), Some(addr(8092)), None);
+ }
+
+ #[test]
+ #[should_panic(expected = "listener address conflict: replica and ws")]
+ fn replica_ws_overlap_panics() {
+ assert_listen_addrs_distinct(addr(9090), addr(8090), Some(addr(9090)), None);
+ }
+
+ #[test]
+ #[should_panic(expected = "listener address conflict: client and ws")]
+ fn client_ws_overlap_panics() {
+ assert_listen_addrs_distinct(addr(9090), addr(8090), Some(addr(8090)), None);
+ }
+}
(iggy) 02/04: feat(message_bus): add shard-0 QUIC client listener
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 3eb4a7ddf33ad9462535e6c96f2d45465737d7c8
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Apr 27 13:03:16 2026 +0200
feat(message_bus): add shard-0 QUIC client listener
client_listener_quic::{bind, run} drives the QUIC accept loop on
shard 0. bind constructs a compio_quic::Endpoint from a ServerConfig
(built by the bootstrap caller via transports::quic::server_config_with_cert
with operator-provided cert chain + key DER). run drives
wait_incoming -> Incoming::accept -> Connecting -> Connection::accept_bi
to completion (reusing the existing QuicTransportListener trait impl
which encodes the 0-RTT-refuse rule), then invokes
AcceptedQuicClientFn(connection, streams).
QUIC stays shard-0 terminal: a compio_quic::Endpoint binds one UDP
socket and demuxes by Connection ID, and per-conn TLS / packet-number
/ congestion state in quinn-proto::Connection is non-serialisable.
Cross-shard handover is fundamentally out of reach.
New AcceptedQuicClientFn type alias on lib.rs (mirrors
AcceptedClientFn shape but carries the (Connection, (SendStream,
RecvStream)) tuple). New into_parts accessor on QuicTransportConn
deconstructs back into the raw tuple for the install path. Tests:
new tests/quic_client_roundtrip.rs exercises a real rcgen-cert
loopback Request -> Reply round trip via bus.send_to_client. New
tests/common::install_quic_clients_locally helper mirrors the
existing install_clients_locally for unit-test minting + install.
143/143 tests pass on cargo nextest run -p message_bus -p server-ng
-p shard. Workspace clippy clean.
Drive-by: replica_io.rs panic! trailing comma removed (clippy lint
on the P5-T1 commit landed after the pre-commit fmt rewrap).
---
core/message_bus/src/client_listener_quic.rs | 112 +++++++++++++++++++++
core/message_bus/src/lib.rs | 28 ++++++
core/message_bus/src/replica_io.rs | 2 +-
core/message_bus/src/transports/quic.rs | 15 +++
core/message_bus/tests/common/mod.rs | 28 +++++-
core/message_bus/tests/quic_client_roundtrip.rs | 124 ++++++++++++++++++++++++
6 files changed, 307 insertions(+), 2 deletions(-)
diff --git a/core/message_bus/src/client_listener_quic.rs b/core/message_bus/src/client_listener_quic.rs
new file mode 100644
index 000000000..16079885a
--- /dev/null
+++ b/core/message_bus/src/client_listener_quic.rs
@@ -0,0 +1,112 @@
+// 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 listener for consensus-protocol SDK clients.
+//!
+//! Runs only on shard 0. The accept loop drives the QUIC handshake to
+//! completion AND accepts the first bidirectional stream pair before
+//! invoking the supplied callback, so the callback receives a
+//! ready-for-traffic [`compio_quic::Connection`] plus its
+//! `(SendStream, RecvStream)` pair. Subprotocol enforcement runs at
+//! the rustls ALPN layer (set to `iggy.consensus.v1` in
+//! [`crate::transports::quic::server_config_with_cert`]); no
+//! application-layer code runs for clients with the wrong ALPN.
+//!
+//! 0-RTT data is refused at accept time
+//! ([`crate::transports::quic::server_config_with_cert`] sets
+//! `max_early_data_size = 0` at the rustls layer; the accept path
+//! double-checks `RecvStream::is_0rtt()` for defense in depth and
+//! closes with `QUIC_PROTOCOL_VIOLATION` on mismatch).
+//!
+//! QUIC stays shard-0 terminal: a `compio_quic::Endpoint` binds a
+//! single UDP socket and demuxes incoming packets to in-flight
+//! connections by Connection ID, and per-connection TLS / packet-number
+//! / congestion state lives in `quinn-proto::Connection`, which is
+//! non-serialisable. Cross-shard handover is fundamentally out of
+//! reach for QUIC; the callback always installs locally on shard 0.
+
+use crate::lifecycle::ShutdownToken;
+use crate::transports::quic::QuicTransportListener;
+use crate::{AcceptedQuicClientFn, transports::TransportListener};
+use compio_quic::{Endpoint, ServerConfig};
+use futures::FutureExt;
+use iggy_common::IggyError;
+use std::net::SocketAddr;
+use tracing::{debug, error, info};
+
+/// Bind a QUIC [`Endpoint`] without starting the accept loop.
+///
+/// `server_config` is built by the bootstrap caller via
+/// [`crate::transports::quic::server_config_with_cert`] from the
+/// operator-provided cert chain + key DER.
+///
+/// # Errors
+///
+/// Returns [`IggyError::CannotBindToSocket`] if the bind fails.
+#[allow(clippy::future_not_send)]
+pub async fn bind(
+ addr: SocketAddr,
+ server_config: ServerConfig,
+) -> Result<(Endpoint, SocketAddr), IggyError> {
+ let endpoint = Endpoint::server(addr, server_config)
+ .await
+ .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?;
+ let actual = endpoint
+ .local_addr()
+ .map_err(|e| IggyError::IoError(e.to_string()))?;
+ Ok((endpoint, actual))
+}
+
+/// Run the QUIC listener accept loop until the shutdown token fires.
+///
+/// Each accepted connection is handed to `on_accepted` along with its
+/// first bidirectional stream pair. The callback owns the connection
+/// from that point on; it mints a `client_id` and calls
+/// [`crate::installer::install_client_quic_conn`].
+#[allow(clippy::future_not_send)]
+pub async fn run(endpoint: Endpoint, token: ShutdownToken, on_accepted: AcceptedQuicClientFn) {
+ info!(
+ "Consensus QUIC client listener accepting on {:?}",
+ endpoint.local_addr().ok()
+ );
+
+ let listener = QuicTransportListener::new(endpoint);
+ loop {
+ futures::select! {
+ () = token.wait().fuse() => {
+ debug!("Consensus QUIC client listener shutting down");
+ break;
+ }
+ result = listener.accept().fuse() => {
+ match result {
+ Ok((conn, peer_addr)) => {
+ debug!(%peer_addr, "QUIC client accepted, handing to installer");
+ let (connection, streams) = conn.into_parts();
+ on_accepted(connection, streams);
+ }
+ Err(e) => {
+ // Endpoint-fatal (e.g. endpoint closed). The
+ // shutdown token will normally fire just before
+ // this; surface the error and exit.
+ error!("Consensus QUIC client listener accept failed: {e}");
+ break;
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index 811102e51..628832cf3 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -81,6 +81,7 @@ pub mod auth;
pub mod auth_config;
pub mod cache;
pub mod client_listener;
+pub mod client_listener_quic;
pub mod config;
pub mod connector;
mod error;
@@ -157,6 +158,33 @@ pub type AcceptedReplicaFn = std::rc::Rc<dyn Fn(compio::net::TcpStream, u8)>;
/// assigning the client id as part of its delegation policy.
pub type AcceptedClientFn = std::rc::Rc<dyn Fn(compio::net::TcpStream)>;
+/// Callback invoked on every accepted SDK QUIC client connection.
+///
+/// Fires after shard 0's QUIC listener drives the handshake to
+/// completion AND accepts the first bidirectional stream pair, so the
+/// callback receives a ready-for-traffic [`compio_quic::Connection`]
+/// plus its `(SendStream, RecvStream)` pair. The callback mints a
+/// client id, then calls [`installer::install_client_quic_conn`] on
+/// the local bus.
+///
+/// QUIC stays shard-0 terminal (per `INVARIANTS.md` I6 and the
+/// non-serialisable nature of `quinn-proto::Connection` state); no
+/// cross-shard handover analog exists for this plane.
+pub type AcceptedQuicClientFn = std::rc::Rc<
+ dyn Fn(compio_quic::Connection, (compio_quic::SendStream, compio_quic::RecvStream)),
+>;
+
+/// Callback invoked on every accepted SDK WebSocket client connection.
+///
+/// Fires after shard 0's WS listener accepts a raw TCP socket. The
+/// HTTP-Upgrade handshake has NOT run yet; the callback hands the
+/// raw stream off to the owning shard via inter-shard fd-shipping
+/// (`ShardFramePayload::ClientWsConnectionSetup`). The owning shard
+/// runs the upgrade and the subprotocol check locally. This shape
+/// preserves invariant I5 (fd-delegation is TCP-only) because the
+/// shipped fd is plain TCP at ship-time.
+pub type AcceptedWsClientFn = std::rc::Rc<dyn Fn(compio::net::TcpStream)>;
+
/// Notifier fired when a delegated replica connection dies.
///
/// The delegated replica connection's writer / reader tasks invoke this on
diff --git a/core/message_bus/src/replica_io.rs b/core/message_bus/src/replica_io.rs
index f0abddecb..2b3b3802a 100644
--- a/core/message_bus/src/replica_io.rs
+++ b/core/message_bus/src/replica_io.rs
@@ -83,7 +83,7 @@ pub fn assert_listen_addrs_distinct(
if let (Some(a), Some(b)) = (a, b)
&& a == b
{
- panic!("listener address conflict: {a_name} and {b_name} both bind to {a}",);
+ panic!("listener address conflict: {a_name} and {b_name} both bind to {a}");
}
}
}
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
index a407b3954..b96b16408 100644
--- a/core/message_bus/src/transports/quic.rs
+++ b/core/message_bus/src/transports/quic.rs
@@ -238,6 +238,21 @@ impl QuicTransportConn {
streams,
}
}
+
+ /// Deconstruct into the raw `(Connection, (SendStream, RecvStream))`
+ /// tuple, mirror of [`Self::new`].
+ ///
+ /// `client_listener_quic::run` uses this to hand the
+ /// already-accepted connection + first bidi pair to
+ /// [`crate::installer::install_client_quic_conn`] (which then wraps
+ /// them in a fresh `QuicTransportConn` and dispatches via the
+ /// generic install path). [`TransportConn::into_split`] cannot be
+ /// used at that boundary because the install path needs the raw
+ /// tuple, not the split halves.
+ #[must_use]
+ pub fn into_parts(self) -> (Connection, (SendStream, RecvStream)) {
+ (self.connection, self.streams)
+ }
}
impl TransportConn for QuicTransportConn {
diff --git a/core/message_bus/tests/common/mod.rs b/core/message_bus/tests/common/mod.rs
index 81cdf1fa5..4f7e88821 100644
--- a/core/message_bus/tests/common/mod.rs
+++ b/core/message_bus/tests/common/mod.rs
@@ -30,7 +30,9 @@ 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};
+use message_bus::{
+ AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, IggyMessageBus, installer,
+};
use std::cell::Cell;
use std::net::SocketAddr;
use std::rc::Rc;
@@ -94,3 +96,27 @@ pub fn install_clients_locally(
installer::install_client_stream(&bus, client_id, stream, on_request.clone());
})
}
+
+/// Build an [`AcceptedQuicClientFn`] that mints a local client id and
+/// installs the QUIC connection directly on the given bus. Mirror of
+/// [`install_clients_locally`] for the QUIC plane.
+#[must_use]
+pub fn install_quic_clients_locally(
+ bus: Rc<IggyMessageBus>,
+ on_request: RequestHandler,
+) -> AcceptedQuicClientFn {
+ let counter: Rc<Cell<u128>> = Rc::new(Cell::new(1));
+ let shard_id = u128::from(bus.shard_id());
+ Rc::new(move |connection, streams| {
+ let seq = counter.get();
+ counter.set(seq.wrapping_add(1));
+ let client_id = (shard_id << 112) | seq;
+ installer::install_client_quic_conn(
+ &bus,
+ client_id,
+ connection,
+ streams,
+ on_request.clone(),
+ );
+ })
+}
diff --git a/core/message_bus/tests/quic_client_roundtrip.rs b/core/message_bus/tests/quic_client_roundtrip.rs
new file mode 100644
index 000000000..c3d96723a
--- /dev/null
+++ b/core/message_bus/tests/quic_client_roundtrip.rs
@@ -0,0 +1,124 @@
+// 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.
+
+//! End-to-end: a real QUIC client connects to the consensus QUIC client
+//! listener, sends a Request, the handler echoes a Reply back via
+//! `bus.send_to_client`, the client reads the Reply.
+
+mod common;
+
+use common::{header_only, install_quic_clients_locally, loopback};
+use compio_quic::{ClientBuilder, Endpoint};
+use iggy_binary_protocol::Command2;
+use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
+use iggy_binary_protocol::consensus::iobuf::Frozen;
+use message_bus::client_listener::RequestHandler;
+use message_bus::client_listener_quic::{bind, run};
+use message_bus::framing;
+use message_bus::transports::quic::{QuicTransportConn, server_config_with_cert};
+use message_bus::transports::{TransportConn, TransportReader, TransportWriter};
+use message_bus::{IggyMessageBus, MessageBus};
+use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
+use std::rc::Rc;
+use std::time::Duration;
+
+fn install_crypto_provider() {
+ // Idempotent across same-process retries.
+ let _ = rustls::crypto::ring::default_provider().install_default();
+}
+
+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 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]
+async fn request_reply_round_trip() {
+ install_crypto_provider();
+
+ let bus = Rc::new(IggyMessageBus::new(7));
+
+ let bus_for_handler = bus.clone();
+ let on_request: RequestHandler = Rc::new(move |client_id, msg| {
+ assert_eq!(msg.header().command, Command2::Request);
+ let bus = bus_for_handler.clone();
+ compio::runtime::spawn(async move {
+ let reply = header_only(Command2::Reply, 42, 0);
+ bus.send_to_client(client_id, reply.into_frozen())
+ .await
+ .expect("send_to_client should succeed");
+ })
+ .detach();
+ });
+
+ let (cert, key) = self_signed();
+ let server_cfg = server_config_with_cert(vec![cert.clone()], key).expect("server config");
+ let (endpoint, server_addr) = bind(loopback(), server_cfg).await.expect("bind");
+
+ let token = bus.token();
+ let on_accepted = install_quic_clients_locally(bus.clone(), on_request);
+ let accept_handle = compio::runtime::spawn(async move {
+ run(endpoint, token, on_accepted).await;
+ });
+ bus.track_background(accept_handle);
+
+ // Dial as a real QUIC client.
+ 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");
+
+ // Reuse the existing QuicTransportConn split for client-side framing.
+ let conn = QuicTransportConn::new(connection, (send, recv));
+ let (mut reader, mut writer) = conn.into_split();
+
+ // Send a Request as a single-frame batch via the writer's atomic
+ // send_batch (mirrors the production hot path).
+ let request = header_only(Command2::Request, 42, 0).into_frozen();
+ let mut batch: Vec<Frozen<MESSAGE_ALIGN>> = vec![request];
+ writer.send_batch(&mut batch).await.expect("client send");
+ assert!(batch.is_empty(), "send_batch must drain the Vec");
+
+ // Read the Reply on the RecvStream.
+ let reply = reader
+ .read_message(framing::MAX_MESSAGE_SIZE)
+ .await
+ .expect("client read");
+ assert_eq!(reply.header().command, Command2::Reply);
+ assert_eq!(reply.header().cluster, 42);
+
+ let outcome = bus.shutdown(Duration::from_secs(2)).await;
+ assert_eq!(
+ outcome.force, 0,
+ "graceful shutdown should not force-cancel"
+ );
+}