diff --git a/core/journal/src/file_storage.rs b/core/journal/src/file_storage.rs index 1273d27b98..b8f07fb089 100644 --- a/core/journal/src/file_storage.rs +++ b/core/journal/src/file_storage.rs @@ -19,7 +19,9 @@ use crate::Storage; use compio::buf::IoBuf; use compio::io::{AsyncReadAtExt, AsyncWriteAtExt}; use std::cell::{Cell, UnsafeCell}; +use std::fs; use std::io; +use std::os::fd::AsFd; use std::path::{Path, PathBuf}; /// File-backed storage implementing the `Storage` trait. @@ -57,20 +59,30 @@ impl FileStorage { self.write_offset.get() } - /// Truncate the file to `len` bytes. + /// Truncate the file to `len` bytes and make the new length durable. + /// + /// Synchronous `std::fs` on a duplicate of the open descriptor, not compio: + /// compio's `set_len` submits `IORING_OP_FTRUNCATE`, which landed in + /// mainline Linux 6.9. When the opcode is unavailable, the driver falls + /// back to its blocking pool, and shard proactors run with + /// `thread_pool_limit(0)`, so the fallback panics the shard instead of + /// repairing the WAL. `std::fs` needs neither the opcode nor the pool. The + /// sole caller is boot-time torn-tail repair, so blocking the shard thread + /// here costs nothing. + /// + /// `sync_all` makes the durable-truncation contract explicit and matches + /// segment recovery. Its additional metadata synchronization is acceptable + /// because this runs only during boot-time repair. /// /// # Errors - /// Returns an I/O error if truncation fails. - // TODO(hubcio): compio `set_len` submits IORING_OP_FTRUNCATE, which kernels - // below 6.9 do not support; the driver then falls back to its blocking - // pool, and shard proactors run with `thread_pool_limit(0)`, so the torn - // WAL repair panics the shard on such kernels instead of repairing. Use a - // synchronous `std::fs` truncate here (boot-time path) or gate on a probe. - pub async fn truncate(&self, len: u64) -> io::Result<()> { + /// Returns an I/O error if the descriptor cannot be cloned, truncated, or synced. + pub(crate) fn truncate(&self, len: u64) -> io::Result<()> { + // SAFETY: single-threaded compio runtime, no concurrent access to the file. let file = unsafe { &*self.file.get() }; - file.set_len(len).await?; + let file = fs::File::from(file.as_fd().try_clone_to_owned()?); + file.set_len(len)?; self.write_offset.set(len); - Ok(()) + file.sync_all() } /// Fsync the file to disk. @@ -178,3 +190,49 @@ impl Storage for FileStorage { Ok(buffer) } } + +#[cfg(test)] +mod tests { + use super::FileStorage; + use server_common::executor::create_shard_executor; + use tempfile::tempdir; + + /// Pins the synchronous truncate signature and verifies it works inside a + /// shard executor with no blocking pool. A modern test kernel supports + /// `IORING_OP_FTRUNCATE`, so this does not reproduce compio's fallback. + #[test] + fn given_a_shard_executor_with_no_blocking_pool_when_truncating_should_repair_the_file() { + let runtime = create_shard_executor().unwrap(); + runtime.block_on(async { + let dir = tempdir().unwrap(); + let path = dir.path().join("journal.wal"); + let storage = FileStorage::open(&path).await.unwrap(); + storage.write_append(vec![0xAB_u8; 128]).await.unwrap(); + + storage.truncate(64).unwrap(); + + assert_eq!(storage.file_len(), 64); + assert_eq!(std::fs::metadata(&path).unwrap().len(), 64); + }); + } + + #[test] + fn given_a_replaced_path_when_truncating_should_truncate_the_open_file() { + let runtime = create_shard_executor().unwrap(); + runtime.block_on(async { + let dir = tempdir().unwrap(); + let path = dir.path().join("journal.wal"); + let renamed_path = dir.path().join("journal.renamed.wal"); + let storage = FileStorage::open(&path).await.unwrap(); + storage.write_append(vec![0xAB_u8; 128]).await.unwrap(); + std::fs::rename(&path, &renamed_path).unwrap(); + std::fs::write(&path, vec![0xCD_u8; 256]).unwrap(); + + storage.truncate(64).unwrap(); + + assert_eq!(storage.file_len(), 64); + assert_eq!(std::fs::metadata(&renamed_path).unwrap().len(), 64); + assert_eq!(std::fs::metadata(&path).unwrap().len(), 256); + }); + } +} diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs index b00f24b560..d7de238752 100644 --- a/core/journal/src/prepare_journal.rs +++ b/core/journal/src/prepare_journal.rs @@ -243,12 +243,7 @@ async fn truncate_or_fail( reason, "truncating torn WAL tail; no complete entry follows the damage" ); - storage.truncate(pos).await?; - // The repair must be crash-durable. `FileStorage::truncate` is a - // bare `set_len`; without this fsync a power loss right after the - // repair re-presents the torn tail on the next boot. Mirrors the - // write-then-fsync the `append` path already does. - storage.fsync().await?; + storage.truncate(pos)?; Ok(()) } @@ -1802,8 +1797,7 @@ mod tests { let storage = FileStorage::open(&path).await.unwrap(); let full_len = storage.file_len(); // Remove the last 10 bytes (partial second entry) - storage.truncate(full_len - 10).await.unwrap(); - storage.fsync().await.unwrap(); + storage.truncate(full_len - 10).unwrap(); } // Reopen, should recover only the first entry diff --git a/core/message_bus/src/client_listener/mod.rs b/core/message_bus/src/client_listener/mod.rs index 1ed966bd22..f0f739496b 100644 --- a/core/message_bus/src/client_listener/mod.rs +++ b/core/message_bus/src/client_listener/mod.rs @@ -85,12 +85,16 @@ //! violation). DoS-shaped abuse is bounded instead by handshake-grace //! 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; +use compio::net::TcpListener; +use iggy_common::IggyError; +use socket2::SockRef; + +use crate::socket_opts::bind_reusable_tcp_listener; +use crate::{GenericHeader, Message}; + pub mod quic; pub mod tcp; pub mod tcp_tls; @@ -100,40 +104,22 @@ 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_REUSEADDR` is set so a restarted server can rebind the port -/// while a previous client connection lingers in `TIME_WAIT` (matching the -/// replica and TLS listeners; QUIC sets no reuse flag since UDP has no -/// `TIME_WAIT`); `SO_REUSEPORT` is intentionally not set: only shard 0 -/// binds the client listeners (see each caller). +/// Binding stays synchronous through `bind_reusable_tcp_listener` so shard +/// startup does not depend on compio's `IORING_OP_BIND` and +/// `IORING_OP_LISTEN` fallback path. `TCP_NODELAY` remains set on the listener +/// to preserve the previous plain-TCP and WebSocket bind configuration. /// /// # 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 - .set_reuseaddr(true) - .map_err(|e| IggyError::IoError(e.to_string()))?; - socket - .bind(addr) - .await - .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?; - let listener = socket - .listen(libc::SOMAXCONN) - .await +/// Returns [`IggyError::CannotBindToSocket`] if the bind or listen fails, or +/// [`IggyError::IoError`] if configuring `TCP_NODELAY` or reading the bound +/// address fails. +pub fn bind_nodelay_listener(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> { + let listener = bind_reusable_tcp_listener(addr) .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?; + SockRef::from(&listener) + .set_tcp_nodelay(true) + .map_err(|e| IggyError::IoError(e.to_string()))?; let actual = listener .local_addr() .map_err(|e| IggyError::IoError(e.to_string()))?; @@ -147,3 +133,26 @@ pub async fn bind_nodelay_listener( /// `RequestHeader` message with the same handler signature, regardless /// of whether the wire is plain TCP, TLS, WS, WSS, or QUIC. pub type RequestHandler = Rc)>; + +#[cfg(test)] +mod tests { + use std::net::{Ipv4Addr, SocketAddr}; + + use server_common::executor::create_shard_executor; + use socket2::SockRef; + + use super::bind_nodelay_listener; + + #[test] + fn given_a_shard_executor_when_binding_a_client_listener_should_preserve_nodelay() { + let runtime = create_shard_executor().unwrap(); + runtime.block_on(async { + let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0); + + let (listener, bound_addr) = bind_nodelay_listener(addr).unwrap(); + + assert_ne!(bound_addr.port(), 0); + assert!(SockRef::from(&listener).tcp_nodelay().unwrap()); + }); + } +} diff --git a/core/message_bus/src/client_listener/tcp.rs b/core/message_bus/src/client_listener/tcp.rs index c900d390cf..a14673774a 100644 --- a/core/message_bus/src/client_listener/tcp.rs +++ b/core/message_bus/src/client_listener/tcp.rs @@ -40,13 +40,13 @@ use tracing::{debug, error, info}; /// /// # Errors /// -/// Returns [`IggyError::CannotBindToSocket`] if the bind fails. -#[allow(clippy::future_not_send)] -pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> { +/// Returns [`IggyError::CannotBindToSocket`] if the bind fails, or +/// [`IggyError::IoError`] if listener configuration fails. +pub fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> { // `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`. - bind_nodelay_listener(addr).await + bind_nodelay_listener(addr) } /// 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 9d2521209f..f61f8c7e40 100644 --- a/core/message_bus/src/client_listener/ws.rs +++ b/core/message_bus/src/client_listener/ws.rs @@ -54,13 +54,13 @@ use tracing::{debug, error, info}; /// /// # Errors /// -/// Returns [`IggyError::CannotBindToSocket`] if the bind fails. -#[allow(clippy::future_not_send)] -pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> { +/// Returns [`IggyError::CannotBindToSocket`] if the bind fails, or +/// [`IggyError::IoError`] if listener configuration fails. +pub fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> { // `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`. - bind_nodelay_listener(addr).await + bind_nodelay_listener(addr) } /// Run the WS pre-upgrade listener accept loop until the shutdown diff --git a/core/message_bus/src/replica/io.rs b/core/message_bus/src/replica/io.rs index 290db8e2d2..9368ef5a7e 100644 --- a/core/message_bus/src/replica/io.rs +++ b/core/message_bus/src/replica/io.rs @@ -212,7 +212,7 @@ pub async fn start_on_shard_zero( ); let (replica_listener, replica_bound) = bind_replica_listener(replica_listen_addr).await?; - let (clients_listener, client_bound) = client_listener::tcp::bind(client_listen_addr).await?; + let (clients_listener, client_bound) = client_listener::tcp::bind(client_listen_addr)?; let token_for_replica = bus.token(); let replica_handle = compio::runtime::spawn(async move { @@ -228,7 +228,7 @@ pub async fn start_on_shard_zero( let ws_bound = match (ws_listen_addr, on_accepted_ws_client) { (Some(addr), Some(on_accepted_ws)) => { - let (ws_listener, ws_bound) = client_listener::ws::bind(addr).await?; + let (ws_listener, ws_bound) = client_listener::ws::bind(addr)?; let token_for_ws = bus.token(); let ws_handle = compio::runtime::spawn(async move { client_listener::ws::run(ws_listener, token_for_ws, on_accepted_ws).await; diff --git a/core/message_bus/tests/graceful_shutdown.rs b/core/message_bus/tests/graceful_shutdown.rs index b37ac40488..cee1a8a4bd 100644 --- a/core/message_bus/tests/graceful_shutdown.rs +++ b/core/message_bus/tests/graceful_shutdown.rs @@ -32,7 +32,7 @@ use std::time::Duration; async fn drains_all_clients_within_timeout() { let bus = Rc::new(IggyMessageBus::new(0)); let on_request: RequestHandler = Rc::new(|_, _| {}); - let (listener, addr) = bind(loopback()).await.unwrap(); + let (listener, addr) = bind(loopback()).unwrap(); let token = bus.token(); let accept_delegate = install_clients_locally(bus.clone(), on_request); @@ -87,7 +87,7 @@ async fn drains_all_clients_within_timeout() { async fn connection_drain_precedes_slow_background() { let bus = Rc::new(IggyMessageBus::new(0)); let on_request: RequestHandler = Rc::new(|_, _| {}); - let (listener, addr) = bind(loopback()).await.unwrap(); + let (listener, addr) = bind(loopback()).unwrap(); let token = bus.token(); let accept_delegate = install_clients_locally(bus.clone(), on_request); diff --git a/core/message_bus/tests/tcp_client_roundtrip.rs b/core/message_bus/tests/tcp_client_roundtrip.rs index 9925026b23..b020ad8841 100644 --- a/core/message_bus/tests/tcp_client_roundtrip.rs +++ b/core/message_bus/tests/tcp_client_roundtrip.rs @@ -49,7 +49,7 @@ async fn request_reply_round_trip() { .detach(); }); - let (listener, addr) = bind(loopback()).await.expect("bind"); + let (listener, addr) = bind(loopback()).expect("bind"); let token = bus.token(); let accept_delegate = install_clients_locally(bus.clone(), on_request); let accept_handle = compio::runtime::spawn(async move { @@ -88,7 +88,7 @@ async fn unexpected_command_is_ignored() { let _ = tx.try_send(()); }); - let (listener, addr) = bind(loopback()).await.unwrap(); + let (listener, addr) = bind(loopback()).unwrap(); let token = bus.token(); let accept_delegate = install_clients_locally(bus.clone(), on_request); let accept_handle = compio::runtime::spawn(async move { diff --git a/core/message_bus/tests/ws_client_roundtrip.rs b/core/message_bus/tests/ws_client_roundtrip.rs index 3f10ff4e07..f14d9a88e6 100644 --- a/core/message_bus/tests/ws_client_roundtrip.rs +++ b/core/message_bus/tests/ws_client_roundtrip.rs @@ -88,7 +88,7 @@ async fn handshake_succeeds_and_round_trip_completes() { .detach(); }); - let (listener, server_addr) = bind(loopback()).await.expect("bind"); + let (listener, server_addr) = bind(loopback()).expect("bind"); let token = bus.token(); let on_accepted = install_ws_clients_locally(bus.clone(), on_request); let accept_handle = compio::runtime::spawn(async move { @@ -128,7 +128,7 @@ async fn handshake_succeeds_without_subprotocol_header() { let bus = Rc::new(IggyMessageBus::new(0)); let on_request: RequestHandler = Rc::new(|_, _| {}); - let (listener, server_addr) = bind(loopback()).await.expect("bind"); + let (listener, server_addr) = bind(loopback()).expect("bind"); let token = bus.token(); let on_accepted = install_ws_clients_locally(bus.clone(), on_request); let accept_handle = compio::runtime::spawn(async move { diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 9a58b8b78a..4320c3bc5e 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -3319,10 +3319,9 @@ async fn start_tcp_runtime( &config.cluster, Arc::clone(&config.system), &self_advertised, - self_ports, + &self_ports, shard_metrics_all, - ) - .await?; + )?; } Ok(()) @@ -3484,7 +3483,7 @@ async fn start_manual_runtime( None }; - let bound_clients = start_client_listeners(shard, config, topology, &accepted_clients).await?; + let bound_clients = start_client_listeners(shard, config, topology, &accepted_clients)?; write_current_config( config, Some(topology.self_replica_id), @@ -3849,7 +3848,7 @@ fn mint_client_meta( ClientConnMeta::new(coord.mint_shard_zero_client_id(), peer_addr, transport) } -async fn start_client_listeners( +fn start_client_listeners( shard: &Rc, config: &ServerConfig, topology: &TcpTopology, @@ -3859,7 +3858,6 @@ async fn start_client_listeners( if config.tcp.enabled && !config.tcp.tls.enabled { let (listener, bound_addr) = client_listener::tcp::bind(topology.client_listen_addr) - .await .map_err(|source| { error!( addr = %topology.client_listen_addr, @@ -3878,7 +3876,12 @@ async fn start_client_listeners( } if let Some(ws_addr) = topology.ws_listen_addr { - bound.ws = Some(start_websocket_listener(shard, config, ws_addr, accepted_clients).await?); + bound.ws = Some(start_websocket_listener( + shard, + config, + ws_addr, + accepted_clients, + )?); } if let Some(quic_addr) = topology.quic_listen_addr { @@ -4080,7 +4083,7 @@ fn load_tcp_tls_server_credentials( /// `websocket.tls.enabled` (the plain-WS accept loop must not also bind the /// port -- a plain upgrade parser fed a TLS `ClientHello` rejects every /// connection with an httparse error), plain WS otherwise. -async fn start_websocket_listener( +fn start_websocket_listener( shard: &Rc, config: &ServerConfig, ws_addr: SocketAddr, @@ -4101,11 +4104,10 @@ async fn start_websocket_listener( shard.bus.track_background(wss_handle); Ok(bound_addr) } else { - let (listener, bound_addr) = - client_listener::ws::bind(ws_addr).await.map_err(|source| { - error!(addr = %ws_addr, error = %source, "failed to bind websocket listener"); - source - })?; + let (listener, bound_addr) = client_listener::ws::bind(ws_addr).map_err(|source| { + error!(addr = %ws_addr, error = %source, "failed to bind websocket listener"); + source + })?; let token = shard.bus.token(); let accepted_ws = accepted_clients.ws.clone(); let ws_handle = compio::runtime::spawn(async move { diff --git a/core/server/src/http.rs b/core/server/src/http.rs index fc72acc4a7..ed670883d8 100644 --- a/core/server/src/http.rs +++ b/core/server/src/http.rs @@ -95,7 +95,7 @@ use crate::server_error::ServerError; /// `http_config.jwt`, the `[http.cors]` config is invalid, the `[http.tls]` /// credentials cannot be loaded, or the listener cannot bind to `addr`. #[allow(clippy::too_many_arguments)] -pub async fn start( +pub fn start( shard: &Rc, addr: SocketAddr, http_config: &HttpConfig, @@ -104,7 +104,7 @@ pub async fn start( cluster: &ClusterConfig, system_config: Arc, self_advertised: &str, - self_ports: TransportPorts, + self_ports: &TransportPorts, shard_metrics_all: &[shard::metrics::ShardMetrics], ) -> Result<(), ServerError> { // In cluster mode with no configured JWT secret the signing key derives @@ -139,7 +139,7 @@ pub async fn start( // Same early-fail rule for the scrape path: axum panics on a route // without a leading '/', so reject it as a config error instead. let metrics_endpoint = metrics::validated_endpoint(&http_config.metrics)?; - let (listener, bound_addr) = client_listener::tcp::bind(addr).await?; + let (listener, bound_addr) = client_listener::tcp::bind(addr)?; let state: HttpState = SendWrapper::new(Rc::new(HttpInner { shard: Rc::clone(shard), @@ -156,7 +156,7 @@ pub async fn start( // ports arrive resolved from the caller. self_ports: TransportPorts { http: Some(bound_addr.port()), - ..self_ports + ..self_ports.clone() }, // The HTTP listener is shard-0-only, where the live consensus // handle supplies the leader; the published-view fallback is diff --git a/helm/charts/iggy/README.md b/helm/charts/iggy/README.md index d9861ca13d..bfd38cb7ae 100644 --- a/helm/charts/iggy/README.md +++ b/helm/charts/iggy/README.md @@ -15,8 +15,15 @@ A Helm chart for Apache Iggy server and web-ui Iggy server uses `io_uring` for high-performance async I/O. This requires: -1. **IPC_LOCK capability** - For locking memory required by io_uring -2. **Unconfined seccomp profile** - To allow io_uring syscalls +1. **Linux kernel 5.19 or newer on the node** + + * Shard rings require `IORING_SETUP_COOP_TASKRUN` and `IORING_SETUP_TASKRUN_FLAG`. + * Compio's asynchronous socket creation requires `IORING_OP_SOCKET`. + * Mainline Linux provides these features starting in 5.19. + * Older kernels fail during shard startup. The node kernel matters, not the container image. + +2. **IPC_LOCK capability** - For locking memory required by io_uring +3. **Unconfined seccomp profile** - To allow io_uring syscalls These are configured by default for the Iggy server via the chart's root-level `securityContext` and `podSecurityContext`. The web UI uses `ui.securityContext` diff --git a/helm/charts/iggy/README.md.gotmpl b/helm/charts/iggy/README.md.gotmpl index 03a2700ea7..a36d30adcd 100644 --- a/helm/charts/iggy/README.md.gotmpl +++ b/helm/charts/iggy/README.md.gotmpl @@ -33,8 +33,15 @@ under the License. Iggy server uses `io_uring` for high-performance async I/O. This requires: -1. **IPC_LOCK capability** - For locking memory required by io_uring -2. **Unconfined seccomp profile** - To allow io_uring syscalls +1. **Linux kernel 5.19 or newer on the node** + + * Shard rings require `IORING_SETUP_COOP_TASKRUN` and `IORING_SETUP_TASKRUN_FLAG`. + * Compio's asynchronous socket creation requires `IORING_OP_SOCKET`. + * Mainline Linux provides these features starting in 5.19. + * Older kernels fail during shard startup. The node kernel matters, not the container image. + +2. **IPC_LOCK capability** - For locking memory required by io_uring +3. **Unconfined seccomp profile** - To allow io_uring syscalls These are configured by default for the Iggy server via the chart's root-level `securityContext` and `podSecurityContext`. The web UI uses `ui.securityContext`