fix(journal): repair a torn WAL tail instead of panicking the shard

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);
+        });
+    }
+}

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

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<dyn Fn(u128, Message<GenericHeader>)>;
+
+#[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());
+        });
+    }
+}

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

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

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;

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);

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 {

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 {