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/06/02 11:46:00 UTC
(iggy) branch master updated: fix(server): fix websocket recv (#3405)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 9f698a596 fix(server): fix websocket recv (#3405)
9f698a596 is described below
commit 9f698a596fdbfeff7daddfcfb2b7b1de1946a380
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Tue Jun 2 13:45:55 2026 +0200
fix(server): fix websocket recv (#3405)
---
Cargo.lock | 223 ++++++++++---------
Cargo.toml | 14 +-
core/message_bus/src/client_listener/mod.rs | 40 ++++
core/message_bus/src/client_listener/quic.rs | 27 +--
core/message_bus/src/client_listener/tcp.rs | 12 +-
core/message_bus/src/client_listener/ws.rs | 12 +-
core/message_bus/src/installer/mod.rs | 4 +-
core/message_bus/src/lib.rs | 9 +-
core/message_bus/src/replica/listener.rs | 23 +-
core/message_bus/src/transports/tcp.rs | 26 ++-
core/message_bus/src/transports/ws.rs | 246 ++++++++-------------
core/message_bus/src/transports/wss.rs | 19 +-
core/server-ng/src/bootstrap.rs | 8 +-
core/server/Cargo.toml | 1 -
core/server/src/http/http_server.rs | 9 +-
core/server/src/http/jwt/jwks.rs | 18 +-
core/server/src/http/jwt/middleware.rs | 17 +-
core/server/src/sender/websocket_tls_sender.rs | 5 +-
core/server/src/tcp/mod.rs | 49 ++++
core/server/src/tcp/tcp_listener.rs | 28 +--
core/server/src/tcp/tcp_tls_listener.rs | 28 +--
core/server/src/websocket/websocket_listener.rs | 8 +-
.../server/src/websocket/websocket_tls_listener.rs | 14 +-
23 files changed, 433 insertions(+), 407 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 8500d924d..d6adc0beb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1203,6 +1203,22 @@ dependencies = [
"syn 2.0.117",
]
+[[package]]
+name = "async-tungstenite"
+version = "0.34.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8447f02eaa65412035e2d3eeaa3fc82bbb8d7137c84c5976b4af685136012ee9"
+dependencies = [
+ "atomic-waker",
+ "futures-core",
+ "futures-io",
+ "futures-task",
+ "futures-util",
+ "log",
+ "pin-project-lite",
+ "tungstenite",
+]
+
[[package]]
name = "async_zip"
version = "0.0.18"
@@ -2798,9 +2814,9 @@ dependencies = [
[[package]]
name = "compio"
-version = "0.18.0"
+version = "0.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b84ee96a86948d04388f3a0b8c36b9f0a6b40b3528ac0d65737e53632fb37fe"
+checksum = "2a25ca13ca7580b0cfe1982166432ee8f60a00bff6a3bf781800fce581bb97cd"
dependencies = [
"compio-buf",
"compio-driver",
@@ -2817,9 +2833,9 @@ dependencies = [
[[package]]
name = "compio-buf"
-version = "0.8.1"
+version = "0.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a00d719dbd8c602ab0d25d219cbc6b517008858de7a8d6c51b4dc95aefff4dce"
+checksum = "1b9478803aae4726ce02139b5510354034ae3afc99ce2496734784314664b91c"
dependencies = [
"arrayvec",
"bytes",
@@ -2828,25 +2844,26 @@ dependencies = [
[[package]]
name = "compio-driver"
-version = "0.11.4"
+version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "74d42d98dc890ee4db00c1e68a723391711aab6d67085880d716b72830f7c715"
+checksum = "2bb85cde3af21a2baaaa6eb40dada718d8ef2ac4a1f4acf671091dea13f3f43d"
dependencies = [
- "cfg-if",
+ "bitflags 2.11.1",
"cfg_aliases",
"compio-buf",
"compio-log",
+ "compio-send-wrapper",
"crossbeam-queue",
"flume",
"futures-util",
"io-uring",
- "io_uring_buf_ring",
"libc",
+ "linux-raw-sys 0.12.1",
+ "mod_use",
"once_cell",
- "paste",
- "pin-project-lite",
+ "pastey 0.2.3",
"polling",
- "slab",
+ "rustix 1.1.4",
"smallvec",
"socket2 0.6.3",
"synchrony",
@@ -2854,52 +2871,71 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "compio-executor"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43a8134ddbf3d1ae400ba29fbb19f26e7da58784d4b049b370cef8230ae015a9"
+dependencies = [
+ "compio-log",
+ "compio-send-wrapper",
+ "crossbeam-queue",
+ "loom",
+ "slotmap",
+]
+
[[package]]
name = "compio-fs"
-version = "0.11.0"
+version = "0.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65ee36e1acf2cec4835efe9a986c012b2462c5ef53580e4ee84ae6d5a3d8e3b3"
+checksum = "934c384c7c1dca1d68540bd0ef16186a8e7f33fab330e54652b45ee80f051ee4"
dependencies = [
- "cfg-if",
"cfg_aliases",
"compio-buf",
"compio-driver",
"compio-io",
"compio-runtime",
+ "futures-util",
"libc",
- "os_pipe",
"pin-project-lite",
+ "rustix 1.1.4",
"widestring",
"windows-sys 0.61.2",
]
[[package]]
name = "compio-io"
-version = "0.9.1"
+version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "637522f28a64fd5f7dcceaa4ddef13fa8d8020025e8c993f7a069e237835580e"
+checksum = "cd2f4a095767ab3144de394b868b17cf0e7cab706b5d6464effbc7bf7ee95aaa"
dependencies = [
+ "bytemuck",
"compio-buf",
"futures-util",
- "paste",
+ "libc",
+ "pastey 0.2.3",
+ "pin-project-lite",
+ "rustix 1.1.4",
"synchrony",
+ "windows-sys 0.61.2",
]
[[package]]
name = "compio-log"
-version = "0.1.0"
+version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc4e560213c1996b618da369b7c9109564b41af9033802ae534465c4ee4e132f"
+checksum = "ef39fff6341af7ab6c27fae9a3887e1ec618320d43f3b9c924c7a644f693a859"
dependencies = [
"tracing",
]
[[package]]
name = "compio-macros"
-version = "0.1.2"
+version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f05ed201484967dc70de77a8f7a02b29aaa8e6c81cbea2e75492ee0c8d97766b"
+checksum = "a9ac573b3bb60fffabf47c2ce01c61536fb5eb2168d66e0375e85603e7d51614"
dependencies = [
+ "darling 0.23.0",
"proc-macro-crate 3.3.0",
"proc-macro2",
"quote",
@@ -2908,28 +2944,30 @@ dependencies = [
[[package]]
name = "compio-net"
-version = "0.11.1"
+version = "0.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "becd7d40522c885113752a3640cba9f9d347f205b646bb3f8ff3967173a228f2"
+checksum = "9c83b4b129666d6411f0b0b797f21669c4d3f3ca007e7cce6099957faaa0dc1d"
dependencies = [
- "cfg-if",
"compio-buf",
"compio-driver",
"compio-io",
"compio-runtime",
"either",
+ "futures-util",
"libc",
"once_cell",
+ "pin-project-lite",
"socket2 0.6.3",
+ "synchrony",
"widestring",
"windows-sys 0.61.2",
]
[[package]]
name = "compio-quic"
-version = "0.7.2"
+version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ad9efdad81b920108b9de57148e1b9d73dc408b6d06a59ee64836dde651cf026"
+checksum = "edd50cb1b2c2c8276728871bc3bf3c12618c551cdd550123d564aa49b059aea5"
dependencies = [
"cfg_aliases",
"compio-buf",
@@ -2943,6 +2981,7 @@ dependencies = [
"quinn-proto",
"rustc-hash",
"rustls",
+ "rustls-platform-verifier 0.7.0",
"synchrony",
"thiserror 2.0.18",
"windows-sys 0.61.2",
@@ -2950,32 +2989,40 @@ dependencies = [
[[package]]
name = "compio-runtime"
-version = "0.11.0"
+version = "0.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6c1c71f011bdd9c8f30e97d877b606505ee6d241c7782cfaed172f66acbd9cd"
+checksum = "7788d06de0a16b81025ffb9241bd54a8bf4583d67bc332151b512ae7d8896d04"
dependencies = [
- "async-task",
- "cfg-if",
"compio-buf",
"compio-driver",
+ "compio-executor",
+ "compio-io",
"compio-log",
+ "compio-send-wrapper",
"core_affinity",
- "crossbeam-queue",
"futures-util",
"libc",
- "once_cell",
"pin-project-lite",
"scoped-tls",
- "slab",
- "socket2 0.6.3",
+ "synchrony",
"windows-sys 0.61.2",
]
+[[package]]
+name = "compio-send-wrapper"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f1ef3947d751725afe49ab18ce447d819cd8f4ea9b58ae94b8498eb8ad1f864"
+dependencies = [
+ "cfg-if",
+ "loom",
+]
+
[[package]]
name = "compio-tls"
-version = "0.9.1"
+version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3a7056da226af42cda4c83b00a021cce3e1ee5f4cffc8a0ff8801381e618cf1c"
+checksum = "0177c2684878d0712ccfcf22113a198f5ddf74f0bf323801488dbb47f801a471"
dependencies = [
"compio-buf",
"compio-io",
@@ -2986,16 +3033,20 @@ dependencies = [
[[package]]
name = "compio-ws"
-version = "0.3.1"
+version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "99d45f47c6e64babcaa6b8df1dffced56012e60e58401255e679f428ddbe9fb6"
+checksum = "c0a9e5cbacce3ff2594cfd7321fd946d5f7afcf2dbc7537f6de7b42d798efb3a"
dependencies = [
+ "async-tungstenite",
"compio-buf",
"compio-io",
"compio-log",
"compio-net",
"compio-tls",
- "tungstenite 0.28.0",
+ "futures-util",
+ "pin-project-lite",
+ "rustls-platform-verifier 0.7.0",
+ "tungstenite",
]
[[package]]
@@ -3045,7 +3096,7 @@ dependencies = [
"static-toml",
"strum 0.28.0",
"tracing",
- "tungstenite 0.29.0",
+ "tungstenite",
]
[[package]]
@@ -3553,13 +3604,14 @@ dependencies = [
[[package]]
name = "cyper"
-version = "0.8.3"
+version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9123b02f7942d1f9e7608a2243d4d1face2935facda972a3551ea7a087f722ab"
+checksum = "5f8c8abaf78cfe3cb7838d7cfcf75e4e3bf02fe3e54571e0a1f10bb485cf8fbf"
dependencies = [
"async-stream",
"base64",
"cfg-if",
+ "cfg_aliases",
"compio",
"cyper-core",
"encoding_rs",
@@ -3569,10 +3621,11 @@ dependencies = [
"hyper",
"hyper-util",
"mime",
- "rustls-platform-verifier 0.6.2",
+ "rustls-platform-verifier 0.7.0",
"send_wrapper",
"serde",
"serde_urlencoded",
+ "synchrony",
"thiserror 2.0.18",
"tower-service",
"url",
@@ -3580,9 +3633,9 @@ dependencies = [
[[package]]
name = "cyper-axum"
-version = "0.8.0"
+version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d2f08d85bafecc0195b09bc6a9520db14a89ee13df75fd500c49d366093a0021"
+checksum = "c89ec923d3dff3dbc20a27f7dbdaacb3d76a399ebf8ca5f065697133a02d3307"
dependencies = [
"axum",
"axum-core",
@@ -3601,9 +3654,9 @@ dependencies = [
[[package]]
name = "cyper-core"
-version = "0.8.0"
+version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f606aa5ddfee60d1cd86e350a1f7bf45ad5c4dc060d80edf84eef38a9a6b2efc"
+checksum = "03c8847069e286c64987119637d5f08cdb71e12e85be0294fed649dc8007d32e"
dependencies = [
"compio",
"futures-util",
@@ -6690,7 +6743,7 @@ dependencies = [
"thiserror 2.0.18",
"tokio",
"tracing",
- "tungstenite 0.29.0",
+ "tungstenite",
"twox-hash",
"ulid",
"url",
@@ -7278,17 +7331,6 @@ dependencies = [
"libc",
]
-[[package]]
-name = "io_uring_buf_ring"
-version = "0.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1838759bb8c2f24cf05a35429d83145c4aa6af43f8ad38477295e12a7320a80e"
-dependencies = [
- "bytes",
- "io-uring",
- "rustix 1.1.4",
-]
-
[[package]]
name = "ipconfig"
version = "0.3.4"
@@ -7856,6 +7898,12 @@ dependencies = [
"local-waker",
]
+[[package]]
+name = "local-event"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76dda8459b10a8960dfae91c1c77316fc15e7caf94f9f18794126512a8dc77e5"
+
[[package]]
name = "local-waker"
version = "0.1.4"
@@ -7953,6 +8001,8 @@ dependencies = [
"generator",
"pin-utils",
"scoped-tls",
+ "serde",
+ "serde_json",
"tracing",
"tracing-subscriber",
]
@@ -8274,6 +8324,12 @@ dependencies = [
"syn 2.0.117",
]
+[[package]]
+name = "mod_use"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "21909324aa58f5c284d91cac514c6d081210901dc372d7c8ea6a9d7e0406097a"
+
[[package]]
name = "moka"
version = "0.12.15"
@@ -9041,16 +9097,6 @@ dependencies = [
"pin-project-lite",
]
-[[package]]
-name = "os_pipe"
-version = "1.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d8fae84b431384b68627d0f9b3b1245fcf9f46f6c0e3dc902e9dce64edd1967"
-dependencies = [
- "libc",
- "windows-sys 0.61.2",
-]
-
[[package]]
name = "outref"
version = "0.5.2"
@@ -11483,7 +11529,6 @@ dependencies = [
"iggy_common",
"jsonwebtoken",
"left-right",
- "metadata",
"mimalloc",
"mime_guess",
"nix",
@@ -12370,7 +12415,9 @@ version = "0.1.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "416090a4d8f6358526df5f9f65dfe28750b8b7bfd1fd8a5620f483fc4a75722c"
dependencies = [
+ "event-listener",
"futures-util",
+ "local-event",
"loom",
]
@@ -12633,9 +12680,12 @@ dependencies = [
[[package]]
name = "thin-cell"
-version = "0.1.2"
+version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4164c6c316ba9733b0ab021e7f9852c788a4b991b49c25820f1be48e1d41345b"
+checksum = "a9d3c47fea6e344ef1ac01be266c27945ec30ae8049c3292c897f0878de5c784"
+dependencies = [
+ "synchrony",
+]
[[package]]
name = "thiserror"
@@ -12865,7 +12915,7 @@ dependencies = [
"rustls-pki-types",
"tokio",
"tokio-rustls",
- "tungstenite 0.29.0",
+ "tungstenite",
"webpki-roots 0.26.11",
]
@@ -13241,25 +13291,6 @@ dependencies = [
"core_maths",
]
-[[package]]
-name = "tungstenite"
-version = "0.28.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442"
-dependencies = [
- "bytes",
- "data-encoding",
- "http 1.4.1",
- "httparse",
- "log",
- "rand 0.9.4",
- "rustls",
- "rustls-pki-types",
- "sha1 0.10.6",
- "thiserror 2.0.18",
- "utf-8",
-]
-
[[package]]
name = "tungstenite"
version = "0.29.0"
@@ -13615,12 +13646,6 @@ dependencies = [
"xmlwriter",
]
-[[package]]
-name = "utf-8"
-version = "0.7.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
-
[[package]]
name = "utf8-width"
version = "0.1.8"
diff --git a/Cargo.toml b/Cargo.toml
index 0f9a5268e..6e74b0087 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -109,7 +109,7 @@ clap_complete = "4.6.5"
clock = { path = "core/clock" }
colored = "3.1.1"
comfy-table = { version = "7.2.2", default-features = false }
-compio = { version = "=0.18.0", features = [
+compio = { version = "=0.19.0", features = [
"runtime",
"macros",
"io-uring",
@@ -122,11 +122,11 @@ compio = { version = "=0.18.0", features = [
"ws",
"fs",
] }
-compio-buf = "0.8.1"
+compio-buf = "0.8.2"
# Pin compio-driver >= 0.11.2 to fix musl compilation (compio-rs/compio#668)
-compio-driver = "0.11.4"
-compio-quic = "=0.7.2"
-compio-ws = "=0.3.1"
+compio-driver = "0.12.1"
+compio-quic = "=0.8.0"
+compio-ws = "=0.4.0"
configs = { path = "core/configs", version = "0.1.0" }
configs_derive = { path = "core/configs_derive", version = "0.1.0" }
consensus = { path = "core/consensus" }
@@ -137,8 +137,8 @@ csv = "1.4.0"
ctor = "1.0.6"
ctrlc = { version = "3.5", features = ["termination"] }
cucumber = "0.23"
-cyper = { version = "0.8.3", features = ["rustls"], default-features = false }
-cyper-axum = { version = "0.8.0" }
+cyper = { version = "0.9.0", features = ["rustls"], default-features = false }
+cyper-axum = { version = "0.9.0" }
darling = "0.23"
dashmap = "6.2.1"
deltalake = { version = "0.32.3", features = ["azure", "gcs", "s3"] }
diff --git a/core/message_bus/src/client_listener/mod.rs b/core/message_bus/src/client_listener/mod.rs
index 2eea83bd9..be3d9877a 100644
--- a/core/message_bus/src/client_listener/mod.rs
+++ b/core/message_bus/src/client_listener/mod.rs
@@ -86,6 +86,9 @@
//! timeouts and the bus-wide [`crate::installer`] backpressure budget.
use crate::{GenericHeader, Message};
+use compio::net::{TcpListener, TcpSocket};
+use iggy_common::IggyError;
+use std::net::SocketAddr;
use std::rc::Rc;
pub mod quic;
@@ -94,6 +97,43 @@ pub mod tcp_tls;
pub mod ws;
pub mod wss;
+/// Bind a TCP listener with `TCP_NODELAY` set, the shared shape used by
+/// the plain-TCP and WS pre-upgrade client listeners.
+///
+/// compio 0.19 replaced `TcpListener::bind_with_options(addr, SocketOpts)`
+/// with the `TcpSocket` builder; this preserves the prior `nodelay(true)`
+/// bind. `SO_REUSEPORT` is intentionally not set: only shard 0 binds the
+/// client listeners (see each caller).
+///
+/// # Errors
+///
+/// Returns [`IggyError::CannotBindToSocket`] if the bind/listen fails.
+#[allow(clippy::future_not_send)]
+pub async fn bind_nodelay_listener(
+ addr: SocketAddr,
+) -> Result<(TcpListener, SocketAddr), IggyError> {
+ let socket = match addr {
+ SocketAddr::V4(_) => TcpSocket::new_v4().await,
+ SocketAddr::V6(_) => TcpSocket::new_v6().await,
+ }
+ .map_err(|e| IggyError::IoError(e.to_string()))?;
+ socket
+ .set_nodelay(true)
+ .map_err(|e| IggyError::IoError(e.to_string()))?;
+ socket
+ .bind(addr)
+ .await
+ .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?;
+ let listener = socket
+ .listen(128)
+ .await
+ .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?;
+ let actual = listener
+ .local_addr()
+ .map_err(|e| IggyError::IoError(e.to_string()))?;
+ Ok((listener, actual))
+}
+
/// Callback for the owning-shard install path in [`crate::installer`].
/// Dispatches `(client_id, request_message)` into the shard's pipeline.
///
diff --git a/core/message_bus/src/client_listener/quic.rs b/core/message_bus/src/client_listener/quic.rs
index 09bec2694..31a4fb32e 100644
--- a/core/message_bus/src/client_listener/quic.rs
+++ b/core/message_bus/src/client_listener/quic.rs
@@ -42,7 +42,6 @@ use crate::lifecycle::ShutdownToken;
use crate::transports::quic::{QuicTransportListener, accept_handshake};
use crate::{AcceptedQuicClientFn, AcceptedQuicConn};
use compio::net::UdpSocket;
-use compio::runtime::JoinHandle;
use compio_quic::{Endpoint, EndpointConfig, ServerConfig};
use futures::FutureExt;
use iggy_common::IggyError;
@@ -139,17 +138,7 @@ pub async fn run(
);
let listener = QuicTransportListener::new(endpoint);
- let mut handshake_handles: Vec<JoinHandle<()>> = Vec::new();
loop {
- // Drop completed handles before blocking on the next incoming so
- // the vector stays bounded under sustained QUIC connect attempts.
- // Without this sweep, every accepted connection's `JoinHandle`
- // would persist for the listener's lifetime, leaking task-table
- // slots + scopeguard state proportional to total accepted
- // connections since boot. Dropping a finished `Task` is a no-op
- // (no in-flight work to cancel).
- handshake_handles.retain(|h| !h.is_finished());
-
futures::select! {
() = token.wait().fuse() => {
debug!("Client listener (QUIC) shutting down");
@@ -159,7 +148,13 @@ pub async fn run(
match result {
Ok(incoming) => {
let on_accepted = on_accepted.clone();
- let handle = compio::runtime::spawn(async move {
+ // Fire-and-forget: compio 0.19's `JoinHandle` has no
+ // `is_finished`, so the prior bounded-handle sweep is
+ // gone. `detach()` lets each handshake task self-manage
+ // (no Vec to grow). On shutdown the listener drops the
+ // `Endpoint`, which fails any in-flight handshake so
+ // the detached task exits.
+ compio::runtime::spawn(async move {
let Some((conn, peer_addr)) =
accept_handshake(incoming, handshake_grace).await
else {
@@ -168,8 +163,8 @@ pub async fn run(
debug!(%peer_addr, "QUIC client accepted, handing to installer");
let connection = conn.into_inner();
on_accepted(AcceptedQuicConn::new(connection));
- });
- handshake_handles.push(handle);
+ })
+ .detach();
}
Err(e) => {
// Endpoint-fatal (e.g. endpoint closed). The
@@ -182,8 +177,4 @@ pub async fn run(
}
}
}
-
- for handle in handshake_handles {
- let _ = handle.await;
- }
}
diff --git a/core/message_bus/src/client_listener/tcp.rs b/core/message_bus/src/client_listener/tcp.rs
index eee380185..c900d390c 100644
--- a/core/message_bus/src/client_listener/tcp.rs
+++ b/core/message_bus/src/client_listener/tcp.rs
@@ -27,8 +27,9 @@
//! writer + reader tasks via [`crate::installer`].
use crate::AcceptedClientFn;
+use crate::client_listener::bind_nodelay_listener;
use crate::lifecycle::ShutdownToken;
-use compio::net::{SocketOpts, TcpListener};
+use compio::net::TcpListener;
use futures::FutureExt;
use iggy_common::IggyError;
use std::net::SocketAddr;
@@ -45,14 +46,7 @@ pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyErr
// `SO_REUSEPORT` intentionally not set: only shard 0 binds the client
// listener. The shard-0 coordinator round-robins accepts to owning
// shards via `shard::LifecycleFrame::ClientConnectionSetup`.
- let opts = SocketOpts::new().nodelay(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))
+ bind_nodelay_listener(addr).await
}
/// Run the client listener accept loop until the shutdown token fires. The
diff --git a/core/message_bus/src/client_listener/ws.rs b/core/message_bus/src/client_listener/ws.rs
index 20775cbb9..9d2521209 100644
--- a/core/message_bus/src/client_listener/ws.rs
+++ b/core/message_bus/src/client_listener/ws.rs
@@ -35,8 +35,9 @@
//! without pulling `shard` in as a doc-only dep.
use crate::AcceptedWsClientFn;
+use crate::client_listener::bind_nodelay_listener;
use crate::lifecycle::ShutdownToken;
-use compio::net::{SocketOpts, TcpListener};
+use compio::net::TcpListener;
use futures::FutureExt;
use iggy_common::IggyError;
use std::net::SocketAddr;
@@ -59,14 +60,7 @@ pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyErr
// `SO_REUSEPORT` intentionally not set: only shard 0 binds the WS
// listener. The shard-0 coordinator round-robins accepts to owning
// shards via `shard::LifecycleFrame::ClientWsConnectionSetup`.
- let opts = SocketOpts::new().nodelay(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))
+ bind_nodelay_listener(addr).await
}
/// Run the WS pre-upgrade listener accept loop until the shutdown
diff --git a/core/message_bus/src/installer/mod.rs b/core/message_bus/src/installer/mod.rs
index 6af733657..939770507 100644
--- a/core/message_bus/src/installer/mod.rs
+++ b/core/message_bus/src/installer/mod.rs
@@ -107,9 +107,11 @@ impl ConnectionInstaller for Rc<IggyMessageBus> {
let ws_config = cfg.ws_config;
let handshake_grace = cfg.handshake_grace;
let handle = compio::runtime::spawn(async move {
+ // compio-ws 0.4's `WebSocketStream` accept future is large
+ // (~17 KB); box it to keep this spawned task's frame small.
let outcome = compio::time::timeout(
handshake_grace,
- compio_ws::accept_async_with_config(stream, ws_config),
+ Box::pin(compio_ws::accept_async_with_config(stream, ws_config)),
)
.await;
match outcome {
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index added0de2..10c96ec90 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -906,8 +906,15 @@ impl IggyMessageBus {
/// in a loop until empty, so a task pushed mid-shutdown is still
/// awaited.
pub fn track_background(&self, handle: JoinHandle<()>) {
+ use futures::FutureExt;
let mut tasks = self.background_tasks.borrow_mut();
- tasks.retain(|h| !h.is_finished());
+ // compio 0.19's `JoinHandle` dropped `is_finished`; poll each
+ // handle once (it is `Unpin` + a `Future`) and reap the ones that
+ // have already resolved. `now_or_never` on `&mut h` polls without
+ // consuming; `Some` means the task finished (drop it), `None`
+ // means still running (keep). Single-threaded runtime, so no flip
+ // between this poll and the drop.
+ tasks.retain_mut(|h| h.now_or_never().is_none());
tasks.push(handle);
}
diff --git a/core/message_bus/src/replica/listener.rs b/core/message_bus/src/replica/listener.rs
index 6a125de74..4f8b5bdd7 100644
--- a/core/message_bus/src/replica/listener.rs
+++ b/core/message_bus/src/replica/listener.rs
@@ -86,7 +86,6 @@ use crate::lifecycle::ShutdownToken;
use crate::socket_opts::bind_reusable_tcp_listener;
use crate::{AcceptedReplicaFn, GenericHeader, Message};
use compio::net::{TcpListener, TcpStream};
-use compio::runtime::JoinHandle;
use futures::FutureExt;
use iggy_binary_protocol::Command2;
use iggy_common::IggyError;
@@ -156,14 +155,7 @@ pub async fn run(
"Replica listener accepting on {:?}",
listener.local_addr().ok()
);
- let mut handshake_handles: Vec<JoinHandle<()>> = Vec::new();
loop {
- // Drop completed handles before blocking on the next accept so the
- // vector stays bounded under sustained inbound traffic. Dropping
- // an already-finished `Task` is a no-op (no in-flight work to
- // cancel).
- handshake_handles.retain(|h| !h.is_finished());
-
futures::select! {
() = token.wait().fuse() => {
debug!("Replica listener shutting down");
@@ -173,7 +165,12 @@ pub async fn run(
match result {
Ok((stream, peer_addr)) => {
let on_accepted = on_accepted.clone();
- let handle = compio::runtime::spawn(async move {
+ // Fire-and-forget: compio 0.19's `JoinHandle` has no
+ // `is_finished`, so the prior bounded-handle sweep is
+ // gone. `detach()` lets each handshake task self-manage;
+ // on shutdown the listener drops, failing in-flight
+ // handshakes so detached tasks exit.
+ compio::runtime::spawn(async move {
let mut stream = stream;
let res = compio::time::timeout(
handshake_grace,
@@ -201,8 +198,8 @@ pub async fn run(
);
}
}
- });
- handshake_handles.push(handle);
+ })
+ .detach();
}
Err(e) => {
error!("Replica listener accept failed: {e}");
@@ -211,10 +208,6 @@ pub async fn run(
}
}
}
-
- for handle in handshake_handles {
- let _ = handle.await;
- }
}
/// Read the 256 B `Ping` frame, enforce command + cluster match and the
diff --git a/core/message_bus/src/transports/tcp.rs b/core/message_bus/src/transports/tcp.rs
index 882447365..340d587ea 100644
--- a/core/message_bus/src/transports/tcp.rs
+++ b/core/message_bus/src/transports/tcp.rs
@@ -31,9 +31,9 @@
use super::{ActorContext, TransportConn, TransportListener};
use crate::framing;
use crate::lifecycle::BusMessage;
-use compio::driver::{SharedFd, ToSharedFd};
use compio::io::AsyncWriteExt;
-use compio::net::{OwnedReadHalf, OwnedWriteHalf, TcpListener, TcpStream};
+use compio::net::{TcpListener, TcpStream};
+use compio::runtime::fd::PollFd;
use futures::FutureExt;
use std::io;
use std::mem;
@@ -98,13 +98,15 @@ impl TcpTransportConn {
impl TransportConn for TcpTransportConn {
#[allow(clippy::future_not_send)]
async fn run(self, ctx: ActorContext) {
- // Capture a refcounted fd handle BEFORE `into_split` so the
+ // Capture a refcounted poll fd BEFORE `into_split` so the
// shutdown watchdog can wake the reader's parked io_uring read
// SQE via `libc::shutdown(SHUT_RD)`. No `select!` over a TCP
// read on a still-alive connection: the in-flight read returns
// `Ok(0)` naturally rather than being cancelled (compio's
// io_uring read is not cancel-safe in the protocol sense).
- let shared_fd = self.stream.to_shared_fd();
+ // compio 0.19 dropped `TcpStream::to_shared_fd`; `to_poll_fd`
+ // gives an equivalently refcounted fd handle (PollFd: AsRawFd).
+ let poll_fd = self.stream.to_poll_fd().ok();
let (read_half, write_half) = self.stream.into_split();
let ActorContext {
in_tx,
@@ -117,7 +119,9 @@ impl TransportConn for TcpTransportConn {
peer,
} = ctx;
- spawn_shutdown_watchdog(shared_fd, shutdown.clone(), label, peer.clone());
+ if let Some(poll_fd) = poll_fd {
+ spawn_shutdown_watchdog(poll_fd, shutdown.clone(), label, peer.clone());
+ }
let writer_shutdown = shutdown;
let reader_peer = peer.clone();
@@ -156,19 +160,19 @@ impl TransportConn for TcpTransportConn {
/// shutdown token; the watchdog wakes the parked `io_uring` read instead.
#[allow(clippy::future_not_send)]
fn spawn_shutdown_watchdog(
- shared_fd: SharedFd<socket2::Socket>,
+ poll_fd: PollFd<socket2::Socket>,
shutdown: crate::lifecycle::FusedShutdown,
label: &'static str,
peer: String,
) {
compio::runtime::spawn(async move {
shutdown.wait().await;
- // SAFETY: `shared_fd` is a refcounted clone obtained from the
+ // SAFETY: `poll_fd` is a refcounted handle obtained from the
// still-live `TcpStream` before `into_split`; the matching
// clones in both owned halves keep the kernel fd open. The
- // watchdog's clone independently keeps it open across this
+ // watchdog's handle independently keeps it open across this
// syscall.
- let raw_fd = shared_fd.as_raw_fd();
+ let raw_fd = poll_fd.as_raw_fd();
let rc = unsafe { libc::shutdown(raw_fd, libc::SHUT_RD) };
if rc != 0 {
let err = io::Error::last_os_error();
@@ -192,7 +196,7 @@ fn spawn_shutdown_watchdog(
/// iteration.
#[allow(clippy::future_not_send)]
async fn reader_loop(
- mut read_half: OwnedReadHalf<TcpStream>,
+ mut read_half: TcpStream,
in_tx: async_channel::Sender<server_common::Message<iggy_binary_protocol::GenericHeader>>,
max_message_size: usize,
label: &'static str,
@@ -228,7 +232,7 @@ async fn reader_loop(
/// indefinitely on the live socket.
#[allow(clippy::future_not_send)]
async fn writer_loop(
- mut write_half: OwnedWriteHalf<TcpStream>,
+ mut write_half: TcpStream,
rx: crate::lifecycle::BusReceiver,
shutdown: crate::lifecycle::FusedShutdown,
conn_shutdown: crate::lifecycle::Shutdown,
diff --git a/core/message_bus/src/transports/ws.rs b/core/message_bus/src/transports/ws.rs
index cad81cbb3..e7b625358 100644
--- a/core/message_bus/src/transports/ws.rs
+++ b/core/message_bus/src/transports/ws.rs
@@ -34,34 +34,26 @@
//!
//! # Cancel safety
//!
-//! `compio_ws::WebSocketStream::read` (0.3.1) is **not cancel-safe**: it
-//! awaits the underlying `compio_io::compat::SyncStream::fill_read_buf`,
-//! which submits the read buffer to `io_uring` and only restores it on
-//! await completion. Dropping `read()` mid-fill (e.g. via a losing
-//! `select!` arm) leaves `compio-io 0.9.1`'s `Buffer(Option<Slice>)` at
-//! `None` permanently; the next read access panics
-//! `"buffer was submitted for io and never returned"` (a compio-io
-//! drop-safety bug fixed only in the 0.10 line).
+//! The pump is a concurrent duplex `select_biased!` over shutdown, the
+//! outbound mailbox (`rx`), and an inbound `ws.read()`; the losing arm's
+//! future is dropped each iteration. This is safe on `compio_ws` 0.4 /
+//! `compio-io >= 0.10`:
//!
-//! `compio_ws` 0.3.1 exposes no `split()` (no `Sink`/`Stream` impls), so
-//! the TCP-transport split-reader/writer pattern can't be reused here:
-//! one `&mut WebSocketStream` must serve both reads and writes. To avoid
-//! ever dropping the read, the pump runs **serialized** request → reply
-//! and never races the read against an outbound `recv` arm. Shutdown is
-//! delivered out-of-band via `libc::shutdown(SHUT_RD)` on the underlying
-//! `TcpStream` fd (see [`spawn_shutdown_watchdog`]); the parked
-//! `io_uring` read SQE completes with `Ok(0)` / EOF and the loop exits.
+//! * `WebSocketStream::read` lifts the decoded `Message` into the
+//! stream's `next_item` cancel-buffer before any flush await, so a
+//! dropped `read()` never loses an already-decoded frame.
+//! * `compio-io 0.10`'s `SyncStream` restores its read buffer on drop,
+//! fixing the 0.9.1 bug where a cancelled `fill_read_buf` left the
+//! buffer `None` and the next read panicked
+//! `"buffer was submitted for io and never returned"`.
//!
-//! The remaining narrow `select!` is the reply wait (`rx.recv()` vs
-//! `shutdown`). Both sides are cancel-safe (`async_channel::Receiver` +
-//! `FusedShutdown`), neither touches a compio buffer.
+//! (History: under the 0.3.1 / 0.9.1 stack the read was NOT cancel-safe,
+//! so this pump had to run serialized request->reply with an out-of-band
+//! `libc::shutdown(SHUT_RD)` watchdog to avoid ever dropping the read.
+//! The 0.4 bump removed both workarounds.)
//!
-//! Limitation: if `handle_client_request` silently drops a request (e.g.
-//! transient consensus failure, dedup-absorbed) no reply lands in `rx`
-//! and the connection stays parked on `rx.recv()` until the bus-wide
-//! shutdown token fires or the per-connection shutdown is triggered. The
-//! peer's FIN is not observed mid-wait. SDK clients reconnect after their
-//! read timeout; the orphaned server-side pump exits at next shutdown.
+//! `rx.recv()` and the shutdown future are likewise cancel-safe. Sends
+//! run to completion outside the `select!`.
//!
//! # Pings
//!
@@ -73,16 +65,12 @@
use super::{ActorContext, TransportConn};
use crate::lifecycle::BusMessage;
use bytes::Bytes;
-use compio::driver::{SharedFd, ToSharedFd};
use compio::net::TcpStream;
use compio::ws::WebSocketStream;
-use compio::ws::tungstenite::Message as WsMessage;
+use compio::ws::tungstenite::{self, Message as WsMessage};
use futures::FutureExt;
-use futures::select_biased;
use iggy_binary_protocol::{GenericHeader, read_size_field};
use server_common::{MESSAGE_ALIGN, Message};
-use std::io;
-use std::os::fd::AsRawFd;
use std::time::Duration;
use tracing::{debug, warn};
@@ -165,36 +153,35 @@ impl TransportConn for WsTransportConn {
let label = ctx.label;
let peer = ctx.peer.clone();
let mut ws = self.stream;
- // Capture a refcounted fd handle so the shutdown watchdog can
- // wake the reader's parked `io_uring` read via
- // `libc::shutdown(SHUT_RD)`. `read()` must never sit inside a
- // `select!` against an outbound or shutdown arm: compio-io
- // 0.9.1's `SyncStream::fill_read_buf` poisons the buffer on
- // drop and panics the next access. The watchdog wakes the
- // parked SQE with `Ok(0)` instead.
- let shared_fd = ws.get_ref().to_shared_fd();
- spawn_shutdown_watchdog(shared_fd, ctx.shutdown.clone(), label, peer.clone());
run_pump(&mut ws, ctx).await;
drive_close(&mut ws, self.close_grace, label, &peer).await;
}
}
+/// Per-iteration outcome of the single-task select.
+enum PumpAction {
+ Shutdown,
+ Send(BusMessage),
+ Recv(Result<WsMessage, tungstenite::Error>),
+ MailboxClosed,
+}
+
/// Drive the WS connection until shutdown, peer Close, or an
/// unrecoverable error.
///
-/// Serialized request → reply loop: read one inbound frame, forward to
-/// `in_tx`, await its reply on `rx`, drain any additional buffered
-/// replies up to `max_batch`, write all and flush once. `read()` is
-/// **never** raced against an outbound `recv` arm or the shutdown
-/// token, which is the only safe pattern under compio-io 0.9.1 (see
-/// module header). Shutdown is observed in two places:
+/// Concurrent duplex pump: a single `select_biased!` races the shutdown
+/// token, the outbound mailbox (`rx`), and an inbound `ws.read()`. The
+/// losing arm's future is dropped each iteration.
///
-/// * Mid-read-park: the watchdog calls `libc::shutdown(SHUT_RD)` on the
-/// underlying fd; the SQE completes with `Ok(0)` and `ws.read()`
-/// returns an error that exits the loop.
-/// * Mid-reply-wait: a narrow `select!` over `rx.recv()` vs the
-/// shutdown future. Both sides are cancel-safe and neither touches a
-/// compio buffer.
+/// Cancel safety: `compio_ws::WebSocketStream::read` (0.4) is cancel-safe
+/// -- it lifts the decoded `Message` into the stream's `next_item`
+/// buffer before any flush await, and the underlying `compio-io >= 0.10`
+/// `SyncStream` restores its read buffer on drop. Dropping the read
+/// future when the `recv`/`shutdown` arm wins therefore neither loses a
+/// decoded frame nor poisons the buffer (the 0.9.1 panic this transport
+/// previously had to serialize around). `rx.recv()` and the shutdown
+/// future are likewise cancel-safe. Sends run to completion outside the
+/// `select!`.
#[allow(clippy::future_not_send)]
async fn run_pump(ws: &mut WebSocketStream<TcpStream>, ctx: ActorContext) {
let ActorContext {
@@ -208,60 +195,36 @@ async fn run_pump(ws: &mut WebSocketStream<TcpStream>, ctx: ActorContext) {
..
} = ctx;
let mut shutdown_fut = Box::pin(shutdown.wait().fuse());
+ let mut batch: Vec<BusMessage> = Vec::with_capacity(max_batch);
- // TODO(compio): when the workspace bumps `compio` past 0.18 (i.e. picks
- // up `compio-io >= 0.10`, which fixes `SyncStream::fill_read_buf` so
- // dropping the read future no longer poisons the buffer), restore the
- // original 3-arm `select_biased!(shutdown, recv, read)` pump and delete
- // the watchdog -- the serialized req->reply loop below is only here to
- // sidestep the 0.9.1 drop-safety bug. compio_ws 0.4 also gains
- // `Sink + Stream`, so a split reader/writer becomes preferable; see the
- // `tcp.rs` pattern.
loop {
- // Read inbound. Never raced -> never dropped -> compio buffer safe.
- let msg = match ws.read().await {
- Ok(m) => m,
- Err(e) => {
- debug!(%label, %peer, error = ?e, "ws reader: read error");
- return;
+ let action = {
+ let read_fut = ws.read();
+ let recv_fut = rx.recv();
+ futures::pin_mut!(read_fut);
+ futures::pin_mut!(recv_fut);
+
+ futures::select_biased! {
+ () = shutdown_fut.as_mut() => PumpAction::Shutdown,
+ msg = recv_fut.fuse() => msg.map_or(PumpAction::MailboxClosed, PumpAction::Send),
+ res = read_fut.fuse() => PumpAction::Recv(res),
}
};
- match msg {
- WsMessage::Binary(bytes) => {
- let frame = match decode_consensus_frame(&bytes, max_message_size) {
- Ok(f) => f,
- Err(e) => {
- warn!(%label, %peer, error = ?e, "ws reader: bad consensus frame");
- return;
- }
- };
- if in_tx.send(frame).await.is_err() {
- debug!(%label, %peer, "ws reader: inbound queue dropped");
- return;
- }
-
- // Wait for the matching reply. Selecting `rx.recv()` against
- // `shutdown_fut` is safe: `async_channel::Receiver::recv` is
- // cancel-safe and the shutdown future holds no compio state.
- let first = select_biased! {
- () = shutdown_fut.as_mut() => {
- debug!(%label, %peer, "ws pump: shutdown during reply wait");
- return;
- }
- res = rx.recv().fuse() => if let Ok(m) = res {
- m
- } else {
- debug!(%label, %peer, "ws pump: mailbox closed");
- return;
- },
- };
-
- // Drain any additional pending replies up to `max_batch`,
- // send all, flush once. tungstenite buffers each `send`
- // into its outbound queue; a single trailing `flush`
- // collapses N writev syscalls to 1 per drain.
- let mut batch: Vec<BusMessage> = Vec::with_capacity(max_batch);
+ match action {
+ PumpAction::Shutdown => {
+ debug!(%label, %peer, "ws pump: shutdown observed");
+ return;
+ }
+ PumpAction::MailboxClosed => {
+ debug!(%label, %peer, "ws pump: mailbox closed");
+ return;
+ }
+ PumpAction::Send(first) => {
+ // Drain mailbox up to `max_batch` and flush once.
+ // tungstenite buffers each `send` into its outbound
+ // queue; a single trailing `flush` shrinks N writev
+ // syscalls to 1 per drain.
batch.push(first);
while batch.len() < max_batch {
match rx.try_recv() {
@@ -271,8 +234,8 @@ async fn run_pump(ws: &mut WebSocketStream<TcpStream>, ctx: ActorContext) {
}
let drained = batch.len();
#[allow(clippy::iter_with_drain)]
- for m in batch.drain(..) {
- if let Err(e) = ws.send(WsMessage::Binary(Bytes::from_owner(m))).await {
+ for msg in batch.drain(..) {
+ if let Err(e) = ws.send(WsMessage::Binary(Bytes::from_owner(msg))).await {
warn!(%label, %peer, error = ?e, batch_len = drained, "ws writer: send failed");
return;
}
@@ -282,58 +245,42 @@ async fn run_pump(ws: &mut WebSocketStream<TcpStream>, ctx: ActorContext) {
return;
}
}
- WsMessage::Ping(_) | WsMessage::Pong(_) => {
- // Tungstenite queues an auto-Pong for inbound Pings; the
- // `read()` flush before delivery drains it. No reply
- // expected from the bus, so loop straight back to the
- // next read instead of awaiting `rx`.
- }
- WsMessage::Close(_) => {
- debug!(%label, %peer, "ws reader: peer initiated close");
- return;
- }
- WsMessage::Text(_) | WsMessage::Frame(_) => {
- warn!(%label, %peer, "ws reader: unexpected text/raw frame, closing");
+ PumpAction::Recv(Ok(msg)) => match msg {
+ WsMessage::Binary(bytes) => {
+ match decode_consensus_frame(&bytes, max_message_size) {
+ Ok(frame) => {
+ if in_tx.send(frame).await.is_err() {
+ debug!(%label, %peer, "ws reader: inbound queue dropped");
+ return;
+ }
+ }
+ Err(e) => {
+ warn!(%label, %peer, error = ?e, "ws reader: bad consensus frame");
+ return;
+ }
+ }
+ }
+ WsMessage::Ping(_) | WsMessage::Pong(_) => {
+ // Tungstenite queues an auto-Pong for inbound Pings;
+ // `read()` flushes before delivery so no explicit reply.
+ }
+ WsMessage::Close(_) => {
+ debug!(%label, %peer, "ws reader: peer initiated close");
+ return;
+ }
+ WsMessage::Text(_) | WsMessage::Frame(_) => {
+ warn!(%label, %peer, "ws reader: unexpected text/raw frame, closing");
+ return;
+ }
+ },
+ PumpAction::Recv(Err(e)) => {
+ debug!(%label, %peer, error = ?e, "ws reader: read error");
return;
}
}
}
}
-/// Detached watchdog: wait for the bus / per-connection shutdown token,
-/// then call `libc::shutdown(fd, SHUT_RD)` on the underlying socket.
-/// The parked `io_uring` read SQE inside `ws.read()` completes with
-/// `Ok(0)`, the read returns an error (or `Ok(0)`-style EOF), and
-/// `run_pump` exits without ever dropping the read future. Mirrors the
-/// shutdown model used by `tcp.rs` (necessary here because the WS
-/// stream is not splittable and compio-io 0.9.1 reads are not
-/// drop-safe).
-#[allow(clippy::future_not_send)]
-fn spawn_shutdown_watchdog(
- shared_fd: SharedFd<socket2::Socket>,
- shutdown: crate::lifecycle::FusedShutdown,
- label: &'static str,
- peer: String,
-) {
- compio::runtime::spawn(async move {
- shutdown.wait().await;
- // SAFETY: `shared_fd` is a refcounted clone obtained from the
- // still-live `TcpStream` borrowed through `WebSocketStream::get_ref`;
- // the watchdog's clone keeps the kernel fd open across the syscall
- // independently of the pump's own ownership.
- let raw_fd = shared_fd.as_raw_fd();
- let rc = unsafe { libc::shutdown(raw_fd, libc::SHUT_RD) };
- if rc != 0 {
- let err = io::Error::last_os_error();
- // ENOTCONN is expected if the peer closed first.
- if err.raw_os_error() != Some(libc::ENOTCONN) {
- debug!(%label, %peer, error = ?err, "ws watchdog: SHUT_RD returned");
- }
- }
- })
- .detach();
-}
-
/// Best-effort cooperative close: send WS Close frame, flush, then drop
/// the stream. Bounded by `close_grace`; on timeout the OS sends RST.
#[allow(clippy::future_not_send)]
@@ -360,6 +307,9 @@ async fn drive_close(
}
#[cfg(test)]
+// compio-ws 0.4's `WebSocketStream` read/send futures are large (~17 KB);
+// the test driver awaits them directly. Not worth boxing in tests.
+#[allow(clippy::large_futures)]
mod tests {
use super::*;
use crate::framing;
diff --git a/core/message_bus/src/transports/wss.rs b/core/message_bus/src/transports/wss.rs
index 853e689af..a83c67d4c 100644
--- a/core/message_bus/src/transports/wss.rs
+++ b/core/message_bus/src/transports/wss.rs
@@ -67,7 +67,7 @@ use crate::lifecycle::BusMessage;
use bytes::Bytes;
use compio::io::AsyncWrite;
use compio::net::TcpStream;
-use compio::tls::{TlsAcceptor, TlsConnector, TlsStream};
+use compio::tls::{MaybeTlsStream, TlsAcceptor, TlsConnector, TlsStream};
use compio::ws::tungstenite::{self, Message as WsMessage, protocol::WebSocketConfig};
use compio::ws::{WebSocketStream, accept_async_with_config, client_async_with_config};
use futures::FutureExt;
@@ -278,12 +278,19 @@ async fn ws_handshake(
is_server: bool,
peer: &str,
ws_config: WebSocketConfig,
-) -> Result<WebSocketStream<TlsStream<TcpStream>>, tungstenite::Error> {
+) -> Result<WebSocketStream<TcpStream>, tungstenite::Error> {
+ // compio-ws 0.4 owns the TLS layer via `MaybeTlsStream<S: Splittable>`
+ // (`TlsStream` itself is not `Splittable`, so it cannot be the WS
+ // stream's `S`). Wrap the already-handshaked server/client `TlsStream`
+ // as the `Tls` variant and let the WS layer drive it; the resulting
+ // stream is `WebSocketStream<TcpStream>`, the same type the plaintext
+ // WS transport uses.
+ let stream = MaybeTlsStream::new_tls(tls_stream);
if is_server {
- accept_async_with_config(tls_stream, ws_config).await
+ accept_async_with_config(stream, ws_config).await
} else {
let request = client_request(peer)?;
- let (ws, _resp) = client_async_with_config(request, tls_stream, ws_config).await?;
+ let (ws, _resp) = client_async_with_config(request, stream, ws_config).await?;
Ok(ws)
}
}
@@ -320,7 +327,7 @@ enum PumpAction {
/// Drive the WSS connection until shutdown, peer Close, or an
/// unrecoverable error.
#[allow(clippy::future_not_send)]
-async fn run_pump(ws: &mut WebSocketStream<TlsStream<TcpStream>>, ctx: ActorContext) {
+async fn run_pump(ws: &mut WebSocketStream<TcpStream>, ctx: ActorContext) {
let ActorContext {
in_tx,
rx,
@@ -428,7 +435,7 @@ async fn run_pump(ws: &mut WebSocketStream<TlsStream<TcpStream>>, ctx: ActorCont
/// Shutdown is a transaction (no `select!` racing against the token).
#[allow(clippy::future_not_send)]
async fn drive_close(
- ws: &mut WebSocketStream<TlsStream<TcpStream>>,
+ ws: &mut WebSocketStream<TcpStream>,
close_grace: Duration,
label: &'static str,
peer: &str,
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 50dfeddbe..044e73ece 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -4193,6 +4193,11 @@ mod tests {
#[compio::test]
async fn await_metadata_bundle_aborts_on_shutdown_flag() {
+ // compio 0.19 `JoinHandle` yields `Result<T, JoinError>`; the
+ // `ResumeUnwind` impl re-raises a task panic and maps cancellation
+ // to `None`.
+ use compio::runtime::ResumeUnwind;
+
let (_bundle_tx, bundle_rx) = crossfire::mpmc::bounded_async::<ServerNgMetadataBundle>(1);
let flag = Arc::new(AtomicBool::new(false));
@@ -4208,7 +4213,8 @@ mod tests {
let err = waiter
.await
- .unwrap_or_else(|e| std::panic::resume_unwind(e))
+ .resume_unwind()
+ .expect("waiter task was cancelled")
.expect_err("shutdown flag must abort the bundle wait");
assert!(
matches!(err, ServerNgError::MetadataHandoffAborted { shard_id: 1 }),
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 9e1b2dd56..647f06375 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -69,7 +69,6 @@ iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
jsonwebtoken = { workspace = true }
left-right = { workspace = true }
-metadata = { workspace = true }
mimalloc = { workspace = true, optional = true }
mime_guess = { workspace = true, optional = true }
nix = { workspace = true }
diff --git a/core/server/src/http/http_server.rs b/core/server/src/http/http_server.rs
index b8e56d51a..3b2c5c478 100644
--- a/core/server/src/http/http_server.rs
+++ b/core/server/src/http/http_server.rs
@@ -36,7 +36,7 @@ use axum::extract::connect_info::Connected;
use axum::http::Method;
use axum::{Router, middleware};
use axum_server::tls_rustls::RustlsConfig;
-use compio::net::{SocketOpts, TcpListener};
+use compio::net::TcpListener;
use err_trail::ErrContext;
use iggy_common::IggyError;
use iggy_common::TransportProtocol;
@@ -137,8 +137,11 @@ pub async fn start_http_server(
}
if !config.tls.enabled {
- let opts = SocketOpts::new().reuse_port(true).reuse_address(true);
- let listener = TcpListener::bind_with_options(config.address.clone(), &opts)
+ let bind_addr: SocketAddr = config
+ .address
+ .parse()
+ .unwrap_or_else(|_| panic!("Failed to parse HTTP address {}", config.address));
+ let listener = crate::tcp::bind_reuseport_listener(bind_addr, true, None)
.await
.unwrap_or_else(|_| panic!("Failed to bind to HTTP address {}", config.address));
let address = listener
diff --git a/core/server/src/http/jwt/jwks.rs b/core/server/src/http/jwt/jwks.rs
index db7a9a46a..804a41c84 100644
--- a/core/server/src/http/jwt/jwks.rs
+++ b/core/server/src/http/jwt/jwks.rs
@@ -21,13 +21,19 @@ use iggy_common::IggyError;
use jsonwebtoken::DecodingKey;
use serde::Deserialize;
use std::hash::Hash;
-use std::sync::OnceLock;
use strum::{Display, EnumString};
-static HTTP_CLIENT: OnceLock<cyper::Client> = OnceLock::new();
+thread_local! {
+ // cyper 0.9's `Client` is `!Send`/`!Sync` (`Rc`-backed) and `new()`
+ // now returns a `Result`, so it can no longer be a global `OnceLock`.
+ // compio is thread-per-core; keep one client per thread. The `Rc`
+ // inner makes cloning cheap, so callers take an owned handle.
+ static HTTP_CLIENT: cyper::Client =
+ cyper::Client::new().expect("failed to build cyper HTTP client for JWKS");
+}
-fn get_http_client() -> &'static cyper::Client {
- HTTP_CLIENT.get_or_init(cyper::Client::new)
+fn get_http_client() -> cyper::Client {
+ HTTP_CLIENT.with(cyper::Client::clone)
}
/// JWK key type enumeration
@@ -139,6 +145,10 @@ impl JwksClient {
}
async fn refresh_keys(&self, issuer: &str, jwks_url: &str) -> Result<(), IggyError> {
+ // The cyper client is `!Send` since 0.9; callers reached from axum
+ // middleware wrap this future in `SendWrapper` (see
+ // `http::jwt::middleware::jwt_auth`), so it's free to await cyper
+ // directly here.
let client = get_http_client();
let request = client
.get(jwks_url)
diff --git a/core/server/src/http/jwt/middleware.rs b/core/server/src/http/jwt/middleware.rs
index dfef00b03..1f8f3181d 100644
--- a/core/server/src/http/jwt/middleware.rs
+++ b/core/server/src/http/jwt/middleware.rs
@@ -26,6 +26,7 @@ use axum::{
response::Response,
};
use err_trail::ErrContext;
+use send_wrapper::SendWrapper;
use std::sync::Arc;
const COMPONENT: &str = "JWT_MIDDLEWARE";
@@ -75,16 +76,16 @@ pub async fn jwt_auth(
format!("{COMPONENT} (error: {e}) - failed to decode JWT header")
})
.map_err(|_| UNAUTHORIZED)?;
- let jwt_claims = state
- .jwt_manager
- .decode(jwt_token, token_header.alg)
+ // `decode` may fetch JWKS via the cyper client, which is `!Send` since
+ // cyper 0.9 (`Rc`-backed). axum requires this middleware's future to be
+ // `Send`, so wrap the `!Send` sub-futures in `SendWrapper` -- the same
+ // pattern the rest of the HTTP layer uses for compio shard ops.
+ // Sound under compio's thread-per-core model: the future is never
+ // polled from another thread.
+ let jwt_claims = SendWrapper::new(state.jwt_manager.decode(jwt_token, token_header.alg))
.await
.map_err(|_| UNAUTHORIZED)?;
- if state
- .jwt_manager
- .is_token_revoked(&jwt_claims.claims.jti)
- .await
- {
+ if SendWrapper::new(state.jwt_manager.is_token_revoked(&jwt_claims.claims.jti)).await {
return Err(StatusCode::UNAUTHORIZED);
}
diff --git a/core/server/src/sender/websocket_tls_sender.rs b/core/server/src/sender/websocket_tls_sender.rs
index 82f94b3ae..3078a5025 100644
--- a/core/server/src/sender/websocket_tls_sender.rs
+++ b/core/server/src/sender/websocket_tls_sender.rs
@@ -20,7 +20,6 @@ use super::Sender;
use bytes::{BufMut, BytesMut};
use compio::buf::IoBufMut;
use compio::net::TcpStream;
-use compio::tls::TlsStream;
use compio::ws::WebSocketStream;
use compio::ws::tungstenite::{Error as TungsteniteError, Message};
use iggy_common::IggyError;
@@ -33,13 +32,13 @@ const WRITE_BUFFER_CAPACITY: usize = 8192;
const STATUS_OK: &[u8] = &[0; 4];
pub struct WebSocketTlsSender {
- pub(crate) stream: WebSocketStream<TlsStream<TcpStream>>,
+ pub(crate) stream: WebSocketStream<TcpStream>,
pub(crate) read_buffer: BytesMut,
pub(crate) write_buffer: BytesMut,
}
impl WebSocketTlsSender {
- pub fn new(stream: WebSocketStream<TlsStream<TcpStream>>) -> Self {
+ pub fn new(stream: WebSocketStream<TcpStream>) -> Self {
Self {
stream,
read_buffer: BytesMut::with_capacity(READ_BUFFER_CAPACITY),
diff --git a/core/server/src/tcp/mod.rs b/core/server/src/tcp/mod.rs
index 7d81d1490..a5449ab4f 100644
--- a/core/server/src/tcp/mod.rs
+++ b/core/server/src/tcp/mod.rs
@@ -23,3 +23,52 @@ pub mod tcp_socket;
pub mod tcp_tls_listener;
pub const COMPONENT: &str = "TCP";
+
+/// Bind a `TcpListener` via the compio 0.19 `TcpSocket` builder.
+///
+/// compio 0.19 removed `TcpListener::bind_with_options(addr, SocketOpts)`;
+/// this is the shared replacement for every legacy listener bind.
+/// `SO_REUSEPORT` is always set (thread-per-core: many sockets bind the
+/// same addr+port); `reuseaddr` is opt-in. When `tuning` is `Some` and
+/// `override_defaults` is set, the socket buffers / keepalive / nodelay
+/// are applied (a zero linger maps to `set_zero_linger`; a non-zero
+/// linger has no compio-0.19 successor and is dropped).
+pub(crate) async fn bind_reuseport_listener(
+ addr: std::net::SocketAddr,
+ reuseaddr: bool,
+ tuning: Option<&crate::configs::tcp::TcpSocketConfig>,
+) -> std::io::Result<compio::net::TcpListener> {
+ use compio::net::TcpSocket;
+
+ let socket = match addr {
+ std::net::SocketAddr::V4(_) => TcpSocket::new_v4().await?,
+ std::net::SocketAddr::V6(_) => TcpSocket::new_v6().await?,
+ };
+ socket.set_reuseport(true)?;
+ if reuseaddr {
+ socket.set_reuseaddr(true)?;
+ }
+ if let Some(config) = tuning
+ && config.override_defaults
+ {
+ let recv_buffer_size = config
+ .recv_buffer_size
+ .as_bytes_u64()
+ .try_into()
+ .expect("Failed to parse recv_buffer_size for TCP socket");
+ let send_buffer_size = config
+ .send_buffer_size
+ .as_bytes_u64()
+ .try_into()
+ .expect("Failed to parse send_buffer_size for TCP socket");
+ socket.set_recv_buffer_size(recv_buffer_size)?;
+ socket.set_send_buffer_size(send_buffer_size)?;
+ socket.set_keepalive(config.keepalive)?;
+ if config.linger.get_duration().is_zero() {
+ socket.set_zero_linger()?;
+ }
+ socket.set_nodelay(config.nodelay)?;
+ }
+ socket.bind(addr).await?;
+ socket.listen(1024).await
+}
diff --git a/core/server/src/tcp/tcp_listener.rs b/core/server/src/tcp/tcp_listener.rs
index 5c70c920f..5c83ec91c 100644
--- a/core/server/src/tcp/tcp_listener.rs
+++ b/core/server/src/tcp/tcp_listener.rs
@@ -23,7 +23,7 @@ use crate::shard::IggyShard;
use crate::shard::task_registry::{ShutdownToken, TaskRegistry};
use crate::shard::transmission::event::ShardEvent;
use crate::tcp::connection_handler::{ConnectionAction, handle_connection, handle_error};
-use compio::net::{SocketOpts, TcpListener};
+use compio::net::TcpListener;
use err_trail::ErrContext;
use futures::FutureExt;
use iggy_common::{IggyError, TransportProtocol};
@@ -36,31 +36,7 @@ async fn create_listener(
addr: SocketAddr,
config: &TcpSocketConfig,
) -> Result<TcpListener, std::io::Error> {
- // Required by the thread-per-core model...
- // We create bunch of sockets on different threads, that bind to exactly the same address and port.
- let opts = SocketOpts::new().reuse_port(true).reuse_port(true);
- let opts = if config.override_defaults {
- let recv_buffer_size = config
- .recv_buffer_size
- .as_bytes_u64()
- .try_into()
- .expect("Failed to parse recv_buffer_size for TCP socket");
-
- let send_buffer_size = config
- .send_buffer_size
- .as_bytes_u64()
- .try_into()
- .expect("Failed to parse send_buffer_size for TCP socket");
-
- opts.recv_buffer_size(recv_buffer_size)
- .send_buffer_size(send_buffer_size)
- .keepalive(config.keepalive)
- .linger(config.linger.get_duration())
- .nodelay(config.nodelay)
- } else {
- opts
- };
- TcpListener::bind_with_options(addr, &opts).await
+ crate::tcp::bind_reuseport_listener(addr, false, Some(config)).await
}
pub async fn start(
diff --git a/core/server/src/tcp/tcp_tls_listener.rs b/core/server/src/tcp/tcp_tls_listener.rs
index 41781bdbf..e11c26919 100644
--- a/core/server/src/tcp/tcp_tls_listener.rs
+++ b/core/server/src/tcp/tcp_tls_listener.rs
@@ -22,7 +22,7 @@ use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
use crate::shard::transmission::event::ShardEvent;
use crate::tcp::connection_handler::{handle_connection, handle_error};
-use compio::net::{SocketOpts, TcpListener};
+use compio::net::TcpListener;
use compio::tls::TlsAcceptor;
use err_trail::ErrContext;
use futures::FutureExt;
@@ -126,31 +126,7 @@ async fn create_listener(
addr: SocketAddr,
config: &TcpSocketConfig,
) -> Result<TcpListener, std::io::Error> {
- // Required by the thread-per-core model...
- // We create bunch of sockets on different threads, that bind to exactly the same address and port.
- let opts = SocketOpts::new().reuse_port(true);
- let opts = if config.override_defaults {
- let recv_buffer_size = config
- .recv_buffer_size
- .as_bytes_u64()
- .try_into()
- .expect("Failed to parse recv_buffer_size for TCP socket");
-
- let send_buffer_size = config
- .send_buffer_size
- .as_bytes_u64()
- .try_into()
- .expect("Failed to parse send_buffer_size for TCP socket");
-
- opts.recv_buffer_size(recv_buffer_size)
- .send_buffer_size(send_buffer_size)
- .keepalive(config.keepalive)
- .linger(config.linger.get_duration())
- .nodelay(config.nodelay)
- } else {
- opts
- };
- TcpListener::bind_with_options(addr, &opts).await
+ crate::tcp::bind_reuseport_listener(addr, false, Some(config)).await
}
async fn accept_loop(
diff --git a/core/server/src/websocket/websocket_listener.rs b/core/server/src/websocket/websocket_listener.rs
index 56e6b9aac..766438fe7 100644
--- a/core/server/src/websocket/websocket_listener.rs
+++ b/core/server/src/websocket/websocket_listener.rs
@@ -22,7 +22,7 @@ use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
use crate::shard::transmission::event::ShardEvent;
use crate::websocket::connection_handler::{handle_connection, handle_error};
-use compio::net::{SocketOpts, TcpListener};
+use compio::net::TcpListener;
use compio::ws::accept_async_with_config;
use err_trail::ErrContext;
use futures::FutureExt;
@@ -33,10 +33,8 @@ use std::rc::Rc;
use tracing::{debug, error, info};
async fn create_listener(addr: SocketAddr) -> Result<TcpListener, std::io::Error> {
- // Required by the thread-per-core model...
- // We create bunch of sockets on different threads, that bind to exactly the same address and port.
- let opts = SocketOpts::new().reuse_port(true).reuse_address(true);
- TcpListener::bind_with_options(addr, &opts).await
+ // Thread-per-core: many sockets bind the same addr+port (SO_REUSEPORT).
+ crate::tcp::bind_reuseport_listener(addr, true, None).await
}
pub async fn start(
diff --git a/core/server/src/websocket/websocket_tls_listener.rs b/core/server/src/websocket/websocket_tls_listener.rs
index 4c17dc237..cd33e3e81 100644
--- a/core/server/src/websocket/websocket_tls_listener.rs
+++ b/core/server/src/websocket/websocket_tls_listener.rs
@@ -22,8 +22,8 @@ use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
use crate::shard::transmission::event::ShardEvent;
use crate::websocket::connection_handler::{handle_connection, handle_error};
-use compio::net::{SocketOpts, TcpListener};
-use compio::tls::TlsAcceptor;
+use compio::net::TcpListener;
+use compio::tls::{MaybeTlsStream, TlsAcceptor};
use compio::ws::accept_async_with_config;
use err_trail::ErrContext;
use futures::FutureExt;
@@ -38,10 +38,8 @@ use std::sync::Arc;
use tracing::{debug, error, info, trace, warn};
async fn create_listener(addr: SocketAddr) -> Result<TcpListener, std::io::Error> {
- // Required by the thread-per-core model...
- // We create bunch of sockets on different threads, that bind to exactly the same address and port.
- let opts = SocketOpts::new().reuse_port(true).reuse_address(true);
- TcpListener::bind_with_options(addr, &opts).await
+ // Thread-per-core: many sockets bind the same addr+port (SO_REUSEPORT).
+ crate::tcp::bind_reuseport_listener(addr, true, None).await
}
pub async fn start(
@@ -195,6 +193,10 @@ async fn accept_loop(
Ok(tls_stream) => {
info!("TLS handshake successful for {}, performing WebSocket upgrade...", remote_addr);
+ // compio-ws 0.4 drives TLS via `MaybeTlsStream<TcpStream>`
+ // (`TlsStream` is not `Splittable`); wrap the handshaked
+ // stream so the WS layer owns it.
+ let tls_stream = MaybeTlsStream::new_tls(tls_stream);
match accept_async_with_config(tls_stream, Some(ws_config_clone)).await {
Ok(websocket) => {
info!("WebSocket TLS handshake successful from: {}", remote_addr);