You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/06/01 08:42:42 UTC
(iggy) 04/05: first pass working:
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch integration_tests_basics
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit d0716130b0d15b96dbd3ff00c564818b788a27cc
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Mon Jun 1 10:04:40 2026 +0200
first pass working:
---
core/integration/tests/sdk/hello_world.rs | 78 +----
core/message_bus/src/client_listener/quic.rs | 4 +-
core/message_bus/src/installer/quic.rs | 4 +-
core/message_bus/src/lib.rs | 53 ++-
core/message_bus/src/transports/quic.rs | 419 +++++++++++-------------
core/message_bus/tests/quic_client_roundtrip.rs | 101 +++---
6 files changed, 254 insertions(+), 405 deletions(-)
diff --git a/core/integration/tests/sdk/hello_world.rs b/core/integration/tests/sdk/hello_world.rs
index 905f00ed4..b2266ee67 100644
--- a/core/integration/tests/sdk/hello_world.rs
+++ b/core/integration/tests/sdk/hello_world.rs
@@ -26,19 +26,9 @@ async fn hello_world(harness: &TestHarness) {
client.ping().await.unwrap();
}
-#[cfg(feature = "vsr")]
-#[iggy_harness(test_client_transport = [Tcp, WebSocket])]
-async fn hello_world(harness: &TestHarness) {
- let client = harness.new_client().await.unwrap();
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
-}
-
#[cfg(feature = "vsr")]
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
-async fn hello_world_ping(harness: &TestHarness) {
+async fn hello_world(harness: &TestHarness) {
let client = harness.new_client().await.unwrap();
client
.login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
@@ -47,69 +37,3 @@ async fn hello_world_ping(harness: &TestHarness) {
client.ping().await.unwrap();
client.logout_user().await.unwrap();
}
-
-/// VSR replicated-mutation smoke. Exercises the consensus path under the
-/// new header framing (CreateStream + CreateTopic + SendMessages all
-/// replicate via `Operation::*` -> `prepare_request` -> `on_ack`). Without
-/// this, the hello_world / ping cases only cover `Operation::Register` and
-/// `Operation::NonReplicated`, leaving the actual subject of the PR
-/// uncovered.
-#[cfg(feature = "vsr")]
-#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
-async fn replicated_create_stream_round_trip(harness: &TestHarness) {
- use iggy::prelude::*;
- let client = harness.new_client().await.unwrap();
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
- let stream = client
- .create_stream("vsr-smoke")
- .await
- .expect("create_stream must commit through VSR");
- let topic = client
- .create_topic(
- &stream.id.try_into().unwrap(),
- "vsr-topic",
- 1,
- CompressionAlgorithm::None,
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .expect("create_topic must commit through VSR (CreateTopicWithAssignments transform)");
- client
- .create_consumer_group(
- &stream.id.try_into().unwrap(),
- &topic.id.try_into().unwrap(),
- "vsr-group",
- )
- .await
- .expect("create_consumer_group must commit through VSR");
- client.logout_user().await.unwrap();
-}
-
-/// VSR raw-PAT return path. The token is minted non-deterministically on the
-/// home shard (never replicated), so the committed reply body is empty and the
-/// home shard must inject the raw token before answering the client. A blank
-/// token here means that injection regressed.
-#[cfg(feature = "vsr")]
-#[iggy_harness(test_client_transport = [Tcp])]
-async fn replicated_create_pat_returns_raw_token(harness: &TestHarness) {
- use iggy::prelude::*;
- let client = harness.new_client().await.unwrap();
- client
- .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
- .await
- .unwrap();
- let raw = client
- .create_personal_access_token("vsr-pat", PersonalAccessTokenExpiry::NeverExpire)
- .await
- .expect("create_personal_access_token must commit through VSR");
- assert!(
- !raw.token.is_empty(),
- "home shard must return the minted raw token, not the empty committed body"
- );
- client.logout_user().await.unwrap();
-}
diff --git a/core/message_bus/src/client_listener/quic.rs b/core/message_bus/src/client_listener/quic.rs
index 9094012c8..09bec2694 100644
--- a/core/message_bus/src/client_listener/quic.rs
+++ b/core/message_bus/src/client_listener/quic.rs
@@ -166,8 +166,8 @@ pub async fn run(
return;
};
debug!(%peer_addr, "QUIC client accepted, handing to installer");
- let (connection, streams) = conn.into_parts();
- on_accepted(AcceptedQuicConn::new(connection, streams));
+ let connection = conn.into_inner();
+ on_accepted(AcceptedQuicConn::new(connection));
});
handshake_handles.push(handle);
}
diff --git a/core/message_bus/src/installer/quic.rs b/core/message_bus/src/installer/quic.rs
index c369638d4..13023abd3 100644
--- a/core/message_bus/src/installer/quic.rs
+++ b/core/message_bus/src/installer/quic.rs
@@ -53,12 +53,12 @@ pub fn install_client_quic(
accepted: AcceptedQuicConn,
on_request: RequestHandler,
) {
- let (connection, streams) = accepted.into_parts();
+ let connection = accepted.into_parts();
let close_grace = bus.config().close_grace;
install_client_conn(
bus,
meta,
- QuicTransportConn::new(connection, streams).with_close_grace(close_grace),
+ QuicTransportConn::new(connection).with_close_grace(close_grace),
on_request,
);
}
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index 59a9f0ebc..added0de2 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -284,27 +284,25 @@ pub type AcceptedClientFn = std::rc::Rc<dyn Fn(compio::net::TcpStream)>;
/// SemVer-major change for `iggy_message_bus`.
pub struct AcceptedQuicConn {
connection: compio_quic::Connection,
- streams: (compio_quic::SendStream, compio_quic::RecvStream),
}
impl AcceptedQuicConn {
- /// Bundle a freshly-accepted QUIC connection and its first
- /// bidirectional stream pair.
+ /// Bundle a freshly-accepted QUIC connection.
///
/// `pub(crate)` by design: the constructor's signature mentions
- /// `compio_quic::Connection`, `SendStream`, and `RecvStream`, all
- /// kept off the bus's public `SemVer` surface for the same reason
- /// [`Self::into_parts`] is crate-private. The QUIC listener in
- /// [`crate::client_listener::quic`] is the only mint site.
+ /// `compio_quic::Connection`, kept off the bus's public `SemVer`
+ /// surface for the same reason [`Self::into_parts`] is crate-private.
+ /// The QUIC listener in [`crate::client_listener::quic`] is the only
+ /// mint site.
+ ///
+ /// Subsequent bidirectional streams are accepted on demand by the
+ /// transport's `accept_bi` loop (see `QuicTransportConn::run`); the
+ /// iggy SDK opens a fresh bidi per request, so eagerly capturing the
+ /// first one at handshake time would lock the transport to a
+ /// long-lived bidi pattern the SDK does not use.
#[must_use]
- pub(crate) const fn new(
- connection: compio_quic::Connection,
- streams: (compio_quic::SendStream, compio_quic::RecvStream),
- ) -> Self {
- Self {
- connection,
- streams,
- }
+ pub(crate) const fn new(connection: compio_quic::Connection) -> Self {
+ Self { connection }
}
/// Remote peer address of this accepted QUIC connection.
@@ -323,25 +321,22 @@ impl AcceptedQuicConn {
/// surface so a `compio_quic` version bump does not constitute a
/// `SemVer`-major change for `iggy_message_bus`.
#[must_use]
- pub(crate) fn into_parts(
- self,
- ) -> (
- compio_quic::Connection,
- (compio_quic::SendStream, compio_quic::RecvStream),
- ) {
- (self.connection, self.streams)
+ pub(crate) fn into_parts(self) -> compio_quic::Connection {
+ self.connection
}
}
/// 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 [`AcceptedQuicConn`]. The
-/// callback mints a client id and forwards the conn straight into
-/// [`installer::install_client_quic`] on the local bus, which unwraps
-/// internally; no caller-side `into_parts` is needed (and the helper
-/// is `pub(crate)` for that reason).
+/// Fires after shard 0's QUIC listener drives the QUIC handshake to
+/// completion. The callback receives a ready-for-traffic
+/// [`AcceptedQuicConn`] that does NOT carry any pre-accepted
+/// bidirectional stream; subsequent bidis are accepted on demand by
+/// the transport's `accept_bi` loop (the iggy SDK opens a new bidi per
+/// request). The callback mints a client id and forwards the conn
+/// straight into [`installer::install_client_quic`] on the local bus,
+/// which unwraps internally; no caller-side `into_parts` is needed (and
+/// the helper is `pub(crate)` for that reason).
///
/// QUIC stays shard-0 terminal: shard 0 owns the
/// `compio_quic::Endpoint`, which demuxes incoming UDP packets to
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
index 135af0011..02cb7dc7e 100644
--- a/core/message_bus/src/transports/quic.rs
+++ b/core/message_bus/src/transports/quic.rs
@@ -23,48 +23,63 @@
//!
//! # Connection model
//!
-//! One bidirectional stream per peer (no multiplexing). The connection's
-//! first `accept_bi` pair carries every consensus frame.
+//! One bidirectional stream **per request** (matches the iggy SDK QUIC
+//! client, which opens a fresh `open_bi` per command and `send.finish()`-s
+//! after writing the request). The transport's [`QuicTransportConn::run`]
+//! runs an outer `accept_bi` loop on the connection and serves one bidi
+//! at a time:
//!
-//! The accept step in [`accept_handshake`] drives the QUIC handshake AND
-//! the first `accept_bi` pair before yielding a fully-set-up
-//! [`QuicTransportConn`]. To avoid a single slow / hostile peer wedging
-//! the entire client plane behind that sequential pair of awaits, the
-//! client listener's accept loop pulls bare [`Incoming`] values via
+//! 1. accept the next bidi (or break on shutdown / connection close);
+//! 2. read exactly one inbound frame from the `RecvStream` and forward it
+//! to [`ActorContext::in_tx`];
+//! 3. await one outbound frame on [`ActorContext::rx`] (the reply), drain
+//! any additional immediate replies via `try_recv` (defensive — 1:1 in
+//! steady state), write each to the `SendStream`, then `finish()` to
+//! deliver pending data + FIN to the peer;
+//! 4. drop the bidi pair and loop to the next `accept_bi`.
+//!
+//! The `accept_bi` loop is serial by design: the single
+//! [`ActorContext::rx`] channel per connection is unkeyed, so two
+//! concurrent in-flight bidis would race the reply queue with no way to
+//! tell which reply belongs to which bidi. The iggy SDK on a single
+//! connection already serialises via an internal `RwLock`, so this is a
+//! semantic match, not a regression. A future pipelining upgrade would
+//! introduce `request_id`-keyed reply routing and run per-bidi handlers
+//! concurrently.
+//!
+//! The handshake itself is driven by [`accept_handshake`] (no bidi
+//! captured). To avoid a single slow / hostile peer wedging the entire
+//! client plane behind that handshake, the client listener's accept loop
+//! pulls bare [`Incoming`] values via
//! [`QuicTransportListener::next_incoming`] and spawns one
-//! `accept_handshake` task per incoming. The accept step only obtains
-//! the `(SendStream, RecvStream)` handles; no bytes are read off the
-//! streams.
+//! `accept_handshake` task per incoming.
//!
//! # Zero-copy
//!
//! `compio_quic::SendStream::write<T: IoBuf>` accepts
-//! `Frozen<MESSAGE_ALIGN>` directly; the writer task 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.
+//! `Frozen<MESSAGE_ALIGN>` directly; the per-bidi reply path issues 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.
//!
//! # 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. Per-command 0-RTT enablement requires a per-command
-//! idempotence audit before being turned on.
+//! `RecvStream::is_0rtt()` is checked inside the `accept_bi` loop for
+//! defense in depth; a `true` here closes the connection with
+//! `QUIC_PROTOCOL_VIOLATION`. Per-command 0-RTT enablement requires a
+//! per-command idempotence audit before being turned on.
use super::{ActorContext, TransportConn, TransportListener};
use crate::config::QuicTuning;
use crate::framing;
-use crate::lifecycle::{BusReceiver, FusedShutdown};
use compio::BufResult;
use compio::io::AsyncWriteExt;
use compio_quic::{
- Connection, Endpoint, IdleTimeout, Incoming, RecvStream, SendStream, VarInt, congestion,
+ Connection, Endpoint, IdleTimeout, Incoming, VarInt, congestion,
crypto::rustls::QuicServerConfig,
};
use futures::FutureExt;
-use iggy_binary_protocol::GenericHeader;
-use server_common::Message;
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
@@ -201,23 +216,20 @@ impl TransportListener for QuicTransportListener {
}
}
-/// Drive the QUIC handshake and the first `accept_bi` pair for one
-/// [`Incoming`].
+/// Drive the QUIC handshake for one [`Incoming`].
///
/// Wraps the inner handshake driver in `compio::time::timeout` so a
/// slowloris peer cannot pin the per-incoming spawned task beyond
-/// `handshake_grace`. The whole sequence
-/// (`incoming.accept` -> `connecting.await` -> `connection.accept_bi`)
-/// shares one wall-clock budget; on timeout the spawned future is
-/// dropped, which drops the local `Connection` and closes the QUIC
-/// session via its own Drop path.
+/// `handshake_grace`. The sequence
+/// (`incoming.accept` -> `connecting.await`) shares one wall-clock
+/// budget; on timeout the spawned future is dropped, which drops the
+/// local `Connection` and closes the QUIC session via its own Drop path.
///
-/// Returns [`Some`] on full success (handshake + first bidi pair, no
-/// 0-RTT). Returns [`None`] for any peer-induced failure
-/// (`incoming.accept` rejection, handshake error, `accept_bi` error,
-/// 0-RTT misconfiguration, handshake-grace exceeded); each non-success
-/// path logs and closes the connection with the matching application
-/// close code.
+/// Returns [`Some`] on a successful handshake (no bidirectional stream
+/// captured; per-request bidis are accepted later inside
+/// [`QuicTransportConn::run`]). Returns [`None`] for any peer-induced
+/// failure (`incoming.accept` rejection, handshake error,
+/// handshake-grace exceeded); each non-success path logs.
#[allow(clippy::future_not_send)]
pub async fn accept_handshake(
incoming: Incoming,
@@ -252,30 +264,8 @@ async fn accept_handshake_inner(incoming: Incoming) -> Option<(QuicTransportConn
}
};
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 None;
- }
- };
-
- if recv.is_0rtt() {
- warn!(
- %addr,
- "QUIC stream accepted in 0-RTT window; refusing"
- );
- connection.close(
- VarInt::from_u32(QUIC_PROTOCOL_VIOLATION),
- b"0-RTT not permitted",
- );
- return None;
- }
-
- debug!(%addr, "QUIC connection accepted, first bidi stream ready");
- Some((QuicTransportConn::new(connection, (send, recv)), addr))
+ debug!(%addr, "QUIC handshake complete; bidi accept deferred to run()");
+ Some((QuicTransportConn::new(connection), addr))
}
/// Default wall-clock budget for the joint reader+writer drain plus
@@ -286,202 +276,146 @@ async fn accept_handshake_inner(incoming: Incoming) -> Option<(QuicTransportConn
/// using `MessageBusConfig::close_grace`.
pub(crate) const DEFAULT_CLOSE_GRACE: Duration = Duration::from_secs(2);
-/// A single QUIC connection plus its first bidirectional stream.
+/// A single accepted QUIC connection.
///
-/// The owned `Connection` handle is retained so graceful shutdown can
-/// fire `Connection::close(QUIC_SHUTDOWN, _)` after the writer drains.
+/// Per-request bidirectional streams are accepted on demand inside
+/// [`Self::run`]; the connection handle is retained so graceful
+/// shutdown can fire `Connection::close(QUIC_SHUTDOWN, _)` once the
+/// accept loop exits.
pub struct QuicTransportConn {
connection: Connection,
- streams: (SendStream, RecvStream),
close_grace: Duration,
}
impl QuicTransportConn {
- /// Construct from an already-established connection + bidi pair.
+ /// Construct from an already-handshaked connection.
#[must_use]
- pub const fn new(connection: Connection, streams: (SendStream, RecvStream)) -> Self {
+ pub const fn new(connection: Connection) -> Self {
Self {
connection,
- streams,
close_grace: DEFAULT_CLOSE_GRACE,
}
}
- /// Override the wall-clock bound that covers the joint
- /// reader+writer drain at the end of `run` plus the trailing
- /// `Connection::close(QUIC_SHUTDOWN, _)` invocation. The single
- /// budget protects against a wedged reader (parked on the QUIC
- /// `RecvStream::read_chunk` after the writer exits) or a wedged
- /// writer (parked on a backpressured dispatch) holding the run
- /// future open indefinitely. Intended for tests; the installer
- /// plumbs `MessageBusConfig::close_grace` in production.
+ /// Override the wall-clock bound applied to the final
+ /// `Connection::close(QUIC_SHUTDOWN, _)` drain. Intended for tests;
+ /// the installer plumbs `MessageBusConfig::close_grace` in
+ /// production.
#[must_use]
pub const fn with_close_grace(mut self, close_grace: Duration) -> Self {
self.close_grace = close_grace;
self
}
- /// 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`] (which then wraps
- /// them in a fresh `QuicTransportConn` and dispatches via the
- /// generic install path).
+ /// Deconstruct back to the raw `Connection`. Currently unused by
+ /// production code; retained for symmetry with [`Self::new`] and
+ /// for tests that need to drive the connection directly.
#[must_use]
- pub fn into_parts(self) -> (Connection, (SendStream, RecvStream)) {
- (self.connection, self.streams)
+ pub fn into_inner(self) -> Connection {
+ self.connection
}
}
impl TransportConn for QuicTransportConn {
#[allow(clippy::future_not_send)]
async fn run(self, ctx: ActorContext) {
- let (send, recv) = self.streams;
let connection = self.connection;
- let close_grace = self.close_grace;
let ActorContext {
in_tx,
rx,
shutdown,
- conn_shutdown,
+ conn_shutdown: _,
max_batch: _,
max_message_size,
label,
peer,
} = ctx;
- let reader_shutdown = shutdown.clone();
- let writer_shutdown = shutdown;
- let reader_peer = peer.clone();
- let reader_label = label;
- let reader_handle = compio::runtime::spawn(reader_task(
- recv,
- in_tx,
- reader_shutdown,
- max_message_size,
- reader_label,
- reader_peer,
- ));
- let writer_handle = compio::runtime::spawn(writer_task(
- send,
- rx,
- writer_shutdown,
- conn_shutdown,
- label,
- peer,
- ));
- // Joint reader+writer drain bounded by `close_grace`. Without
- // the timeout a stuck reader (parked on a `RecvStream::read_chunk`
- // after the peer goes silent) or a stuck writer (parked on a
- // backpressured dispatch) keeps `run` alive indefinitely, and
- // the peer never sees the CONNECTION_CLOSE frame. On elapse we
- // fall through to `connection.close(...)` regardless; the
- // FusedShutdown observer inside both tasks already wakes them
- // on bus / per-conn shutdown so the elapse path is rare.
- let _ = compio::time::timeout(close_grace, async move {
- let _ = reader_handle.await;
- let _ = writer_handle.await;
- })
- .await;
- connection.close(VarInt::from_u32(QUIC_SHUTDOWN), b"shutdown");
- }
-}
+ let mut shutdown_fut = Box::pin(shutdown.wait().fuse());
-/// Drain inbound consensus frames from `recv` until shutdown or a
-/// read error.
-///
-/// # Cancel-safety
-///
-/// The `select!` arm that wakes on `shutdown` drops a parked
-/// [`crate::framing::read_message`] future. `read_message` is not
-/// cancel-safe across multi-read frames: bytes already pulled into the
-/// in-flight `Owned<MESSAGE_ALIGN>` are lost with the dropped frame and
-/// the stream has already advanced past them. The reader treats this
-/// loss as terminal and exits, which the writer's scopeguard turns into
-/// a clean per-connection close. Reads that complete in a single poll
-/// (full header + body already buffered by the QUIC stack) are
-/// unaffected.
-#[allow(clippy::future_not_send)]
-async fn reader_task(
- mut recv: RecvStream,
- in_tx: async_channel::Sender<Message<GenericHeader>>,
- shutdown: FusedShutdown,
- max_message_size: usize,
- label: &'static str,
- peer: String,
-) {
- let mut shutdown_fut = Box::pin(shutdown.wait().fuse());
- loop {
- let read_fut = framing::read_message(&mut recv, max_message_size);
- let result = futures::select! {
- () = shutdown_fut.as_mut() => {
- debug!(%label, %peer, "quic reader: shutdown observed");
- return;
- }
- res = read_fut.fuse() => res,
- };
- match result {
- Ok(msg) => {
- if in_tx.send(msg).await.is_err() {
- debug!(%label, %peer, "quic reader: inbound queue dropped");
- return;
+ // Outer accept loop: one bidirectional stream per request.
+ loop {
+ let (mut send, mut recv) = futures::select! {
+ () = shutdown_fut.as_mut() => {
+ debug!(%label, %peer, "quic: shutdown observed at accept_bi");
+ break;
}
+ res = connection.accept_bi().fuse() => match res {
+ Ok(streams) => streams,
+ Err(e) => {
+ debug!(%label, %peer, error = ?e, "quic: accept_bi failed (connection closed)");
+ break;
+ }
+ },
+ };
+
+ // Defense-in-depth 0-RTT check (matches the previous handshake-
+ // path guard now that bidi accept is deferred to here).
+ if recv.is_0rtt() {
+ warn!(%label, %peer, "quic: bidi accepted in 0-RTT window; refusing");
+ connection.close(
+ VarInt::from_u32(QUIC_PROTOCOL_VIOLATION),
+ b"0-RTT not permitted",
+ );
+ break;
}
- Err(e) => {
- debug!(%label, %peer, "quic reader: read error: {e:?}");
- let _keep_in_tx_alive = &in_tx;
- shutdown_fut.await;
- return;
+
+ // Read exactly one inbound frame from this bidi. The SDK
+ // `send.finish()`-s after writing the request, so further
+ // reads on this RecvStream would return ConnectionClosed.
+ let req = match framing::read_message(&mut recv, max_message_size).await {
+ Ok(m) => m,
+ Err(e) => {
+ debug!(%label, %peer, error = ?e, "quic: bidi read error");
+ continue;
+ }
+ };
+ if in_tx.send(req).await.is_err() {
+ debug!(%label, %peer, "quic: inbound queue dropped");
+ break;
}
- }
- }
-}
-/// Drain `rx` and write each frame to the QUIC stream.
-///
-/// On exit (clean OR panic via compio's `catch_unwind`) the scopeguard
-/// triggers the per-connection [`crate::lifecycle::Shutdown`]. The
-/// reader's `select!` over `shutdown.wait()` then resolves, which
-/// closes the reader, then `run()` resumes, closes the QUIC
-/// connection, and the installer's `transport_handle` scopeguard
-/// evicts the registry slot. Without this trigger, a writer-task panic
-/// while the peer is silent would leave the reader parked on
-/// `read_message` indefinitely.
-#[allow(clippy::future_not_send)]
-async fn writer_task(
- mut send: SendStream,
- rx: BusReceiver,
- shutdown: FusedShutdown,
- conn_shutdown: crate::lifecycle::Shutdown,
- label: &'static str,
- peer: String,
-) {
- let _wake_reader = scopeguard::guard(conn_shutdown, |s| s.trigger());
- let mut shutdown_fut = Box::pin(shutdown.wait().fuse());
- loop {
- let frozen = futures::select! {
- () = shutdown_fut.as_mut() => {
- debug!(%label, %peer, "quic writer: shutdown observed");
- return;
+ // Wait for the matching reply. The single bus rx channel is
+ // unkeyed; serialising one bidi at a time guarantees the
+ // next frame on `rx` is the reply for THIS bidi.
+ // `async_channel::Receiver::recv` and `FusedShutdown` are
+ // both cancel-safe and compio-buffer-free, so select-drop
+ // here cannot poison anything.
+ let first = futures::select! {
+ () = shutdown_fut.as_mut() => {
+ debug!(%label, %peer, "quic: shutdown during reply wait");
+ break;
+ }
+ res = rx.recv().fuse() => if let Ok(m) = res {
+ m
+ } else {
+ debug!(%label, %peer, "quic: mailbox closed");
+ break;
+ },
+ };
+
+ // Write the reply, drain any additional immediate replies
+ // (defensive 1:N), then `finish()` so quinn-proto flushes
+ // pending data + a FIN, signalling the SDK that the reply
+ // is complete.
+ let BufResult(result, _frozen) = send.write_all(first).await;
+ if let Err(e) = result {
+ debug!(%label, %peer, error = ?e, "quic: write failed");
+ continue;
}
- msg = rx.recv().fuse() => {
- let Ok(m) = msg else {
- debug!(%label, %peer, "quic writer: channel closed");
- return;
- };
- m
+ while let Ok(more) = rx.try_recv() {
+ let BufResult(result, _frozen) = send.write_all(more).await;
+ if let Err(e) = result {
+ debug!(%label, %peer, error = ?e, "quic: write failed during burst drain");
+ break;
+ }
}
- };
-
- let BufResult(result, _frozen) = send.write_all(frozen).await;
- if let Err(e) = result {
- debug!(%label, %peer, "quic writer: write failed: {e}");
- return;
+ if let Err(e) = send.finish() {
+ debug!(%label, %peer, error = ?e, "quic: send.finish() failed");
+ }
+ // (send, recv) drop here -> stream fully closed.
}
- // No per-frame flush: quinn-proto auto-coalesces STREAM frames
- // into outbound datagrams. An explicit flush() blocks until the
- // peer ACKs, serializing every send and killing pipelining.
+
+ connection.close(VarInt::from_u32(QUIC_SHUTDOWN), b"shutdown");
}
}
@@ -515,14 +449,14 @@ pub fn server_config_with_cert(
#[cfg(test)]
mod tests {
use super::*;
- use crate::lifecycle::Shutdown;
+ use crate::lifecycle::{FusedShutdown, Shutdown};
use async_channel::bounded;
use compio::io::AsyncWrite;
use compio_quic::ClientBuilder;
- use iggy_binary_protocol::{Command2, HEADER_SIZE, SIZE_FIELD_OFFSET};
+ use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, SIZE_FIELD_OFFSET};
use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
- use server_common::MESSAGE_ALIGN;
use server_common::iobuf::Frozen;
+ use server_common::{MESSAGE_ALIGN, Message};
use std::time::Duration;
fn install_crypto_provider() {
@@ -598,6 +532,11 @@ mod tests {
(out_tx, in_rx, shutdown, handle)
}
+ /// End-to-end frame delivery via the new accept_bi-per-bidi server
+ /// loop: client opens N bidis sequentially, writing one request per
+ /// bidi; server's loop reads each request, awaits a reply from
+ /// `rx`, writes it, and `finish()`-es the bidi. Verifies inbound
+ /// ordering matches what the client sent.
#[compio::test]
#[allow(clippy::future_not_send)]
async fn loopback_round_trip_three_frames() {
@@ -609,10 +548,16 @@ mod tests {
let listener = QuicTransportListener::new(server);
let server_task = compio::runtime::spawn(async move {
let (conn, _peer) = listener.accept().await.expect("accept");
- let (_out_tx, in_rx, shutdown, handle) = drive(conn);
+ let (out_tx, in_rx, shutdown, handle) = drive(conn);
+ // For each accepted bidi the server reads one inbound frame
+ // and then awaits exactly one reply on `rx` before
+ // accepting the next bidi. Feed three replies in step.
let a = in_rx.recv().await.unwrap();
+ out_tx.send(header_only(Command2::Reply)).await.unwrap();
let b = in_rx.recv().await.unwrap();
+ out_tx.send(header_only(Command2::Reply)).await.unwrap();
let c = in_rx.recv().await.unwrap();
+ out_tx.send(header_only(Command2::Reply)).await.unwrap();
shutdown.trigger();
let _ = handle.await;
(a.header().command, b.header().command, c.header().command)
@@ -623,13 +568,14 @@ mod tests {
.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 (out_tx, _in_rx, shutdown, handle) = drive(conn);
- out_tx.send(header_only(Command2::Ping)).await.unwrap();
- out_tx.send(header_only(Command2::Prepare)).await.unwrap();
- out_tx.send(header_only(Command2::Request)).await.unwrap();
+ // Three sequential bidis, one frame each.
+ for command in [Command2::Ping, Command2::Prepare, Command2::Request] {
+ let (mut send, _recv) = connection.open_bi_wait().await.expect("open_bi");
+ let BufResult(result, _) = send.write_all(header_only(command)).await;
+ result.expect("write");
+ send.finish().expect("finish");
+ }
let (a, b, c) = compio::time::timeout(Duration::from_secs(5), server_task)
.await
@@ -638,11 +584,13 @@ mod tests {
assert_eq!(a, Command2::Ping);
assert_eq!(b, Command2::Prepare);
assert_eq!(c, Command2::Request);
-
- shutdown.trigger();
- let _ = handle.await;
}
+ /// An oversize size field in the request header makes
+ /// `framing::read_message` reject the frame. The `accept_bi` loop
+ /// must NOT forward the bad frame to `in_tx`. The test asserts no
+ /// inbound frame surfaces within a small grace, then triggers
+ /// shutdown.
#[compio::test]
#[allow(clippy::future_not_send)]
async fn read_message_reports_oversize_via_run() {
@@ -655,10 +603,17 @@ mod tests {
let server_task = compio::runtime::spawn(async move {
let (conn, _peer) = listener.accept().await.expect("accept");
let (_out_tx, in_rx, shutdown, handle) = drive(conn);
- let res = in_rx.recv().await;
+ // The bad frame must NOT appear on `in_rx`. A short timeout
+ // proves the framing layer rejected the read rather than
+ // silently forwarding it: `Ok(Ok(_))` means a frame did
+ // surface (test failure); `Err(Elapsed)` or `Ok(Err(...))`
+ // both mean no frame was delivered.
+ let received = compio::time::timeout(Duration::from_millis(500), in_rx.recv())
+ .await
+ .is_ok_and(|r| r.is_ok());
shutdown.trigger();
let _ = handle.await;
- res
+ received
});
let client = client_endpoint(cert).await;
@@ -666,7 +621,7 @@ mod tests {
.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");
+ let (mut send, _recv) = connection.open_bi_wait().await.expect("open_bi");
// Bogus oversize size field at SIZE_FIELD_OFFSET.
let mut buf = vec![0u8; HEADER_SIZE];
let bogus = u32::try_from(framing::MAX_MESSAGE_SIZE + 1)
@@ -675,16 +630,16 @@ mod tests {
buf[SIZE_FIELD_OFFSET..SIZE_FIELD_OFFSET + 4].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);
+ send.finish().expect("finish");
- let res = compio::time::timeout(Duration::from_secs(5), server_task)
+ let bad_frame_received = compio::time::timeout(Duration::from_secs(5), server_task)
.await
.expect("server task within 5s")
.unwrap();
- // Framing error inside reader_task closes in_tx; recv resolves to Err.
- assert!(res.is_err());
+ assert!(
+ !bad_frame_received,
+ "oversize frame must not surface on in_rx",
+ );
}
#[test]
diff --git a/core/message_bus/tests/quic_client_roundtrip.rs b/core/message_bus/tests/quic_client_roundtrip.rs
index 9217f8694..679bf663b 100644
--- a/core/message_bus/tests/quic_client_roundtrip.rs
+++ b/core/message_bus/tests/quic_client_roundtrip.rs
@@ -16,25 +16,25 @@
// 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.
+//! listener, opens a bidi per request (matching the iggy SDK pattern),
+//! sends a Request, the handler echoes a Reply back via
+//! `bus.send_to_client`, the client reads the Reply off the same bidi.
mod common;
use async_channel::bounded;
use common::{header_only, install_quic_clients_locally, loopback};
+use compio::BufResult;
+use compio::io::AsyncWriteExt;
use compio_quic::{ClientBuilder, Endpoint};
use iggy_binary_protocol::Command2;
-use iggy_binary_protocol::GenericHeader;
use message_bus::QuicTuning;
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::{ActorContext, TransportConn};
-use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, Shutdown};
+use message_bus::transports::quic::server_config_with_cert;
+use message_bus::{IggyMessageBus, MessageBus};
use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
-use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::rc::Rc;
use std::time::Duration;
@@ -97,38 +97,27 @@ async fn request_reply_round_trip() {
.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");
-
- // Drive the client side of the QUIC conn through `TransportConn::run`.
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(8);
- let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(8);
- let (client_shutdown, client_token) = Shutdown::new();
- let ctx = ActorContext {
- in_tx,
- rx: out_rx,
- shutdown: FusedShutdown::single(client_token),
- conn_shutdown: client_shutdown.clone(),
- max_batch: 16,
- max_message_size: framing::MAX_MESSAGE_SIZE,
- label: "test-client",
- peer: "test-client".to_owned(),
- };
- let conn = QuicTransportConn::new(connection, (send, recv));
- let client_handle = compio::runtime::spawn(async move { conn.run(ctx).await });
+ // One bidi per request: open, write Request, finish, read Reply.
+ // Matches the iggy QUIC SDK's `open_bi -> write_all -> finish -> read`
+ // pattern; the server's accept_bi loop accepts the bidi, dispatches
+ // the request, writes the Reply, `finish()`-es the send half.
+ let (mut send, mut recv) = connection.open_bi_wait().await.expect("open_bi");
let request = header_only(Command2::Request, 42, 0).into_frozen();
- out_tx.send(request).await.expect("client send");
-
- let reply = compio::time::timeout(Duration::from_secs(5), in_rx.recv())
- .await
- .expect("client must receive reply within 5 s")
- .expect("reply frame");
+ let BufResult(result, _) = send.write_all(request).await;
+ result.expect("client write request");
+ send.finish().expect("client finish");
+
+ let reply = compio::time::timeout(
+ Duration::from_secs(5),
+ framing::read_message(&mut recv, framing::MAX_MESSAGE_SIZE),
+ )
+ .await
+ .expect("client must receive reply within 5 s")
+ .expect("reply frame");
assert_eq!(reply.header().command, Command2::Reply);
assert_eq!(reply.header().cluster, 42);
- client_shutdown.trigger();
- let _ = client_handle.await;
-
let outcome = bus.shutdown(Duration::from_secs(2)).await;
assert_eq!(
outcome.force, 0,
@@ -137,12 +126,14 @@ async fn request_reply_round_trip() {
}
/// Regression test for the QUIC accept-loop head-of-line block. Before
-/// the fix, `accept_one` ran handshake + first `accept_bi` sequentially
-/// inside the listener's accept loop, so a single peer that completed
-/// the handshake but never opened a bidi stream wedged every subsequent
-/// accept until the 30 s idle timeout fired. The fix spawns one
-/// handshake task per [`Incoming`]; this test asserts the wedged peer
-/// no longer blocks a fast peer's request from being processed.
+/// the fix, the listener ran handshake + first `accept_bi` sequentially
+/// inside its accept loop, so a single peer that completed the handshake
+/// but never opened a bidi stream wedged every subsequent accept until
+/// the 30 s idle timeout fired. The fix spawns one handshake task per
+/// [`Incoming`] and now also defers `accept_bi` entirely to the
+/// per-connection `run()`, so a connected-but-silent peer can never
+/// block the listener. This test asserts the wedged peer does not block
+/// a fast peer's request from being processed.
#[compio::test]
async fn slow_handshake_does_not_block_subsequent_accept() {
install_crypto_provider();
@@ -169,8 +160,9 @@ async fn slow_handshake_does_not_block_subsequent_accept() {
bus.track_background(accept_handle);
// Slow client: completes the QUIC handshake but never opens a bidi
- // stream. Without the fix, the server's accept loop blocks here
- // inside `accept_bi().await` and never moves on to the fast client.
+ // stream. Without the listener-side fix, the accept loop blocked on
+ // this peer's `accept_bi().await` and never moved on to the fast
+ // client.
let slow_client = client_endpoint(cert.clone()).await;
let slow_connecting = slow_client
.connect(server_addr, "localhost", None)
@@ -191,30 +183,16 @@ async fn slow_handshake_does_not_block_subsequent_accept() {
.await
.expect("fast client handshake within 2 s")
.expect("fast client handshake");
- let (send, recv) =
+ let (mut send, _recv) =
compio::time::timeout(Duration::from_secs(2), fast_connection.open_bi_wait())
.await
.expect("fast client open_bi within 2 s")
.expect("open_bi");
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(8);
- let (in_tx, _in_rx) = bounded::<Message<GenericHeader>>(8);
- let (fast_shutdown, fast_token) = Shutdown::new();
- let ctx = ActorContext {
- in_tx,
- rx: out_rx,
- shutdown: FusedShutdown::single(fast_token),
- conn_shutdown: fast_shutdown.clone(),
- max_batch: 16,
- max_message_size: framing::MAX_MESSAGE_SIZE,
- label: "fast-client",
- peer: "fast-client".to_owned(),
- };
- let conn = QuicTransportConn::new(fast_connection, (send, recv));
- let fast_handle = compio::runtime::spawn(async move { conn.run(ctx).await });
-
let request = header_only(Command2::Request, 1, 0).into_frozen();
- out_tx.send(request).await.expect("fast client send");
+ let BufResult(result, _) = send.write_all(request).await;
+ result.expect("fast client write request");
+ send.finish().expect("fast client finish");
compio::time::timeout(Duration::from_secs(2), request_rx.recv())
.await
@@ -222,8 +200,5 @@ async fn slow_handshake_does_not_block_subsequent_accept() {
.expect("server-side request channel");
drop(slow_connection);
- fast_shutdown.trigger();
- let _ = fast_handle.await;
-
let _ = bus.shutdown(Duration::from_secs(2)).await;
}