From 002194dc6e38bf40e6f619afb3005eca1454b5fc Mon Sep 17 00:00:00 2001 From: ostylk Date: Fri, 17 Jul 2026 10:45:30 +0200 Subject: [PATCH 1/2] Use sd_notify to notify systemd when the server is ready to accept connections --- zerofs/Cargo.lock | 34 +++++++++++++++++++++++++++++++--- zerofs/Cargo.toml | 1 + zerofs/src/cli/server.rs | 9 +++++++++ 3 files changed, 41 insertions(+), 3 deletions(-) diff --git a/zerofs/Cargo.lock b/zerofs/Cargo.lock index c135c597..f1c02920 100644 --- a/zerofs/Cargo.lock +++ b/zerofs/Cargo.lock @@ -1842,7 +1842,7 @@ dependencies = [ "base64 0.21.7", "byteorder", "flate2", - "nom", + "nom 7.1.3", "num-traits", ] @@ -2363,6 +2363,24 @@ dependencies = [ "libc", ] +[[package]] +name = "libsystemd" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19c97a761fc86953c5b885422b22c891dbf5bcb9dcc99d0110d6ce4c052759f0" +dependencies = [ + "hmac", + "libc", + "log", + "nix 0.29.0", + "nom 8.0.0", + "once_cell", + "serde", + "sha2", + "thiserror 2.0.18", + "uuid", +] + [[package]] name = "line-clipping" version = "0.3.7" @@ -2699,6 +2717,15 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "nom" +version = "8.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405" +dependencies = [ + "memchr", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -4586,7 +4613,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4ea810f0692f9f51b382fff5893887bb4580f5fa246fde546e0b13e7fcee662" dependencies = [ "fnv", - "nom", + "nom 7.1.3", "phf", "phf_codegen", ] @@ -5625,7 +5652,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "998d2c24ec099a87daf9467808859f9d82b61f1d9c9701251aea037f514eae0e" dependencies = [ - "nom", + "nom 7.1.3", ] [[package]] @@ -6124,6 +6151,7 @@ dependencies = [ "hyper", "hyper-util", "libc", + "libsystemd", "lz4_flex", "metrics", "metrics-exporter-prometheus", diff --git a/zerofs/Cargo.toml b/zerofs/Cargo.toml index b339ab5a..c4091495 100644 --- a/zerofs/Cargo.toml +++ b/zerofs/Cargo.toml @@ -147,6 +147,7 @@ axum = { version = "0.8", features = ["ws"], optional = true } tonic-web = { version = "0.13", optional = true } tower-http = { version = "0.6", features = ["cors"], optional = true } foyer = "0.22" +libsystemd = "0.7.2" # The `zerofs mount` FUSE client. default-features = false uses the pure-Rust # mount path via the fusermount3 helper, so no libfuse linkage is required. diff --git a/zerofs/src/cli/server.rs b/zerofs/src/cli/server.rs index 22ab7117..06649de7 100644 --- a/zerofs/src/cli/server.rs +++ b/zerofs/src/cli/server.rs @@ -16,6 +16,7 @@ use foyer::{ BlockEngineConfig, DeviceBuilder, FsDeviceBuilder, HybridCacheBuilder, PsyncIoEngineConfig, S3FifoConfig, Spawner, }; +use libsystemd::daemon::{NotifyState, booted as systemd_booted, notify as systemd_notify}; use slatedb::admin::AdminBuilder; use slatedb::config::GarbageCollectorDirectoryOptions; use slatedb::config::GarbageCollectorOptions; @@ -1096,6 +1097,14 @@ pub async fn run_server( ); true } + }; + + // If app starts in a systemd environment, notify the server is ready. + if systemd_booted() { + let _ = systemd_notify(false, &[NotifyState::Ready])?; + } + + tokio::select! { _ = tokio::signal::ctrl_c() => { info!("Received SIGINT, initiating graceful shutdown..."); false From 42834ce394400b6838598c899eb932fce3a759d6 Mon Sep 17 00:00:00 2001 From: ostylk Date: Wed, 15 Jul 2026 22:40:49 +0200 Subject: [PATCH 2/2] Guarantee tcp/unix socket are bound before notifying ready to systemd --- zerofs/src/cli/server.rs | 36 ++++++++++------- zerofs/src/mount.rs | 2 +- zerofs/src/nbd/server.rs | 65 ++++++++++++++++++------------- zerofs/src/nfs.rs | 16 +++----- zerofs/src/ninep/server.rs | 62 ++++++++++++++++------------- zerofs/src/zerofs_client_tests.rs | 2 +- 6 files changed, 101 insertions(+), 82 deletions(-) diff --git a/zerofs/src/cli/server.rs b/zerofs/src/cli/server.rs index 06649de7..1ea12f93 100644 --- a/zerofs/src/cli/server.rs +++ b/zerofs/src/cli/server.rs @@ -30,6 +30,7 @@ use std::time::Duration; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::{debug, info}; +use zerofs_nfsserve::tcp::NFSTcp; /// Parse a WAL config into an object store rooted at the full URL path. pub(crate) fn parse_wal_object_store( @@ -93,10 +94,10 @@ async fn start_nfs_servers( fs: Arc, config: Option<&NfsConfig>, shutdown: CancellationToken, -) -> Vec>> { +) -> Result>>> { let config = match config { Some(c) => c, - None => return Vec::new(), + None => return Ok(Vec::new()), }; let mut handles = Vec::new(); @@ -106,9 +107,16 @@ async fn start_nfs_servers( let fs_clone = Arc::clone(&fs); let addr = *addr; let shutdown_clone = shutdown.clone(); + + let listener = crate::nfs::bind_nfs_server_with_config(fs_clone, addr).await?; handles.push(spawn_named("nfs-server", async move { - match crate::nfs::start_nfs_server_with_config(fs_clone, addr, shutdown_clone).await - { + info!( + "NFS server listening on {}:{}", + listener.get_listen_ip(), + listener.get_listen_port() + ); + + match listener.handle_with_shutdown(shutdown_clone).await { Ok(()) => Ok(()), Err(e) => Err(std::io::Error::other(e.to_string())), } @@ -116,7 +124,7 @@ async fn start_nfs_servers( } } - handles + Ok(handles) } fn start_ninep_servers( @@ -133,7 +141,7 @@ fn start_ninep_servers( if let Some(addresses) = &config.addresses { for addr in addresses { info!("Starting 9P server on {}", addr); - let ninep_tcp_server = crate::ninep::NinePServer::new(Arc::clone(&fs), *addr); + let ninep_tcp_server = crate::ninep::NinePServer::new(Arc::clone(&fs), *addr).await?; let shutdown_clone = shutdown.clone(); handles.push(spawn_named("9p-server", async move { ninep_tcp_server.start(shutdown_clone).await @@ -148,7 +156,7 @@ fn start_ninep_servers( ); let ninep_unix_fs = Arc::clone(&fs); let ninep_unix_server = - crate::ninep::NinePServer::new_unix(ninep_unix_fs, socket_path.clone()); + crate::ninep::NinePServer::new_unix(ninep_unix_fs, socket_path.clone())?; let shutdown_clone = shutdown.clone(); handles.push(spawn_named("9p-unix-server", async move { ninep_unix_server.start(shutdown_clone).await @@ -190,10 +198,10 @@ async fn start_nbd_servers( fs: Arc, config: Option<&NbdConfig>, shutdown: CancellationToken, -) -> Vec>> { +) -> Result>>> { let config = match config { Some(c) => c, - None => return Vec::new(), + None => return Ok(Vec::new()), }; let mut handles = Vec::new(); @@ -203,7 +211,7 @@ async fn start_nbd_servers( "Starting NBD server on {} (devices dynamically discovered from .nbd/)", addr ); - let nbd_tcp_server = NBDServer::new_tcp(Arc::clone(&fs), *addr); + let nbd_tcp_server = NBDServer::new_tcp(Arc::clone(&fs), *addr).await?; let shutdown_clone = shutdown.clone(); handles.push(spawn_named("nbd-server", async move { if let Err(e) = nbd_tcp_server.start(shutdown_clone).await { @@ -220,7 +228,7 @@ async fn start_nbd_servers( "Starting NBD server on Unix socket {} (devices dynamically discovered from .nbd/)", socket_path.display() ); - let nbd_unix_server = NBDServer::new_unix(Arc::clone(&fs), socket_path); + let nbd_unix_server = NBDServer::new_unix(Arc::clone(&fs), socket_path)?; let shutdown_clone = shutdown.clone(); handles.push(spawn_named("nbd-unix-server", async move { if let Err(e) = nbd_unix_server.start(shutdown_clone).await { @@ -231,7 +239,7 @@ async fn start_nbd_servers( })); } - handles + Ok(handles) } async fn start_rpc_servers( @@ -941,7 +949,7 @@ pub async fn run_server( settings.servers.nfs.as_ref(), shutdown.clone(), ) - .await; + .await?; let ninep_handles = start_ninep_servers( Arc::clone(&fs), @@ -954,7 +962,7 @@ pub async fn run_server( settings.servers.nbd.as_ref(), shutdown.clone(), ) - .await; + .await?; // A read-only admin over the same store for the GC's checkpoint gate; built // before the store/path are moved into the checkpoint manager below. diff --git a/zerofs/src/mount.rs b/zerofs/src/mount.rs index a9c90dad..293e5ce1 100644 --- a/zerofs/src/mount.rs +++ b/zerofs/src/mount.rs @@ -2029,7 +2029,7 @@ mod client_tests { } fn start_server(fs: Arc, sock: std::path::PathBuf) -> CancellationToken { - let server = NinePServer::new_unix(fs, sock); + let server = NinePServer::new_unix(fs, sock).unwrap(); let shutdown = CancellationToken::new(); let server_shutdown = shutdown.clone(); tokio::spawn(async move { diff --git a/zerofs/src/nbd/server.rs b/zerofs/src/nbd/server.rs index 6bbd2cd9..80211f91 100644 --- a/zerofs/src/nbd/server.rs +++ b/zerofs/src/nbd/server.rs @@ -17,8 +17,8 @@ const MAX_REQUEST_LENGTH: u32 = 128 * 1024 * 1024; const DISCARD_CHUNK_SIZE: usize = 64 * 1024; pub enum Transport { - Tcp(SocketAddr), - Unix(std::path::PathBuf), + Tcp(TcpListener), + Unix(UnixListener), } pub struct NBDServer { @@ -27,18 +27,36 @@ pub struct NBDServer { } impl NBDServer { - pub fn new_tcp(filesystem: Arc, socket: SocketAddr) -> Self { - Self { + pub async fn new_tcp(filesystem: Arc, socket: SocketAddr) -> std::io::Result { + let listener = TcpListener::bind(socket) + .await + .map_err(|e| crate::net_util::tcp_bind_error("NBD", socket, &e))?; + + Ok(Self { filesystem, - transport: Transport::Tcp(socket), - } + transport: Transport::Tcp(listener), + }) } - pub fn new_unix(filesystem: Arc, socket_path: impl Into) -> Self { - Self { + pub fn new_unix( + filesystem: Arc, + socket_path: impl Into, + ) -> std::io::Result { + let path = socket_path.into(); + + // Remove existing socket file if it exists + let _ = std::fs::remove_file(&path); + let listener = UnixListener::bind(&path).map_err(|e| { + std::io::Error::new( + e.kind(), + format!("Failed to bind NBD Unix socket at {:?}: {}", path, e), + ) + })?; + + Ok(Self { filesystem, - transport: Transport::Unix(socket_path.into()), - } + transport: Transport::Unix(listener), + }) } fn spawn_client_handler(&self, stream: S, shutdown: &CancellationToken, client_name: String) @@ -57,16 +75,13 @@ impl NBDServer { pub async fn start(&self, shutdown: CancellationToken) -> std::io::Result<()> { match &self.transport { - Transport::Tcp(socket) => { - let listener = TcpListener::bind(socket) - .await - .map_err(|e| crate::net_util::tcp_bind_error("NBD", socket, &e))?; - info!("NBD server listening on {}", socket); + Transport::Tcp(listener) => { + info!("NBD server listening on {}", listener.local_addr()?); loop { tokio::select! { _ = shutdown.cancelled() => { - info!("NBD TCP server shutting down on {}", socket); + info!("NBD TCP server shutting down on {}", listener.local_addr()?); break; } result = listener.accept() => { @@ -78,22 +93,16 @@ impl NBDServer { } } } - Transport::Unix(path) => { - // Remove existing socket file if it exists - let _ = std::fs::remove_file(path); - - let listener = UnixListener::bind(path).map_err(|e| { - std::io::Error::new( - e.kind(), - format!("Failed to bind NBD Unix socket at {:?}: {}", path, e), - ) - })?; - info!("NBD server listening on Unix socket {:?}", path); + Transport::Unix(listener) => { + info!( + "NBD server listening on Unix socket {:?}", + listener.local_addr()? + ); loop { tokio::select! { _ = shutdown.cancelled() => { - info!("NBD Unix socket server shutting down at {:?}", path); + info!("NBD Unix socket server shutting down at {:?}", listener.local_addr()?); break; } result = listener.accept() => { diff --git a/zerofs/src/nfs.rs b/zerofs/src/nfs.rs index 3dd68613..cd9215f6 100644 --- a/zerofs/src/nfs.rs +++ b/zerofs/src/nfs.rs @@ -6,14 +6,13 @@ use crate::fs::types::{FileType, InodeWithId, SetAttributes}; use async_trait::async_trait; use std::net::SocketAddr; use std::sync::Arc; -use tokio_util::sync::CancellationToken; -use tracing::{debug, info}; +use tracing::debug; use zerofs::fs::EXTENT_SIZE; use zerofs_nfsserve::nfs::{ FSF_CANSETTIME, FSF_HOMOGENEOUS, FSF_LINK, FSF_SYMLINK, fattr3, fileid3, filename3, fsinfo3, fsstat3, ftype3, nfspath3, nfsstat3, nfstime3, post_op_attr, sattr3, specdata3, writeverf3, }; -use zerofs_nfsserve::tcp::{NFSTcp, NFSTcpListener}; +use zerofs_nfsserve::tcp::NFSTcpListener; use zerofs_nfsserve::vfs::{AuthContext as NfsAuthContext, NFSFileSystem, VFSCapabilities}; /// Adapter struct that implements the NFS trait for ZeroFS. @@ -436,20 +435,15 @@ impl NFSFileSystem for NFSAdapter { } } -pub async fn start_nfs_server_with_config( +pub async fn bind_nfs_server_with_config( filesystem: Arc, socket: SocketAddr, - shutdown: CancellationToken, -) -> anyhow::Result<()> { +) -> anyhow::Result> { let adapter = NFSAdapter::new(filesystem); let listener = NFSTcpListener::bind(socket, adapter) .await .map_err(|e| crate::net_util::tcp_bind_error("NFS", socket, &e))?; - - info!("NFS server listening on {}", socket); - - listener.handle_with_shutdown(shutdown).await?; - Ok(()) + Ok(listener) } #[cfg(test)] diff --git a/zerofs/src/ninep/server.rs b/zerofs/src/ninep/server.rs index 04ed60d5..44d473b1 100644 --- a/zerofs/src/ninep/server.rs +++ b/zerofs/src/ninep/server.rs @@ -105,8 +105,8 @@ async fn join_with_timeout( } pub enum Transport { - Tcp(SocketAddr), - Unix(PathBuf), + Tcp(TcpListener), + Unix(UnixListener), } pub struct NinePServer { @@ -116,20 +116,33 @@ pub struct NinePServer { } impl NinePServer { - pub fn new(filesystem: Arc, addr: SocketAddr) -> Self { - Self { + pub async fn new(filesystem: Arc, addr: SocketAddr) -> std::io::Result { + let listener = TcpListener::bind(addr) + .await + .map_err(|e| crate::net_util::tcp_bind_error("9P", addr, &e))?; + + Ok(Self { filesystem, - transport: Transport::Tcp(addr), + transport: Transport::Tcp(listener), lock_manager: Arc::new(FileLockManager::new()), - } + }) } - pub fn new_unix(filesystem: Arc, path: PathBuf) -> Self { - Self { + pub fn new_unix(filesystem: Arc, path: PathBuf) -> std::io::Result { + let _ = std::fs::remove_file(&path); + + let listener = UnixListener::bind(&path).map_err(|e| { + std::io::Error::new( + e.kind(), + format!("Failed to bind Unix socket at {:?}: {}", path, e), + ) + })?; + + Ok(Self { filesystem, - transport: Transport::Unix(path), + transport: Transport::Unix(listener), lock_manager: Arc::new(FileLockManager::new()), - } + }) } fn spawn_client_handler( @@ -166,23 +179,23 @@ impl NinePServer { let mut clients = FuturesUnordered::new(); let clients_shutdown = shutdown.child_token(); let serve_result = match &self.transport { - Transport::Tcp(addr) => { - let listener = TcpListener::bind(addr) - .await - .map_err(|e| crate::net_util::tcp_bind_error("9P", addr, &e))?; - info!("9P server listening on TCP {}", addr); + + match &self.transport { + Transport::Tcp(listener) => { + info!("9P server listening on TCP {}", listener.local_addr()?); loop { tokio::select! { biased; _ = shutdown.cancelled() => { - info!("9P TCP server shutting down on {}", addr); + info!("9P TCP server shutting down on {}", listener.local_addr()?); break Ok(()); } finished = clients.next(), if !clients.is_empty() => { if let Some(Err(e)) = finished { warn!("9P client task failed: {e}"); } + break; } result = listener.accept() => { let (stream, peer_addr) = match result { @@ -207,22 +220,17 @@ impl NinePServer { } } } - Transport::Unix(path) => { - let _ = std::fs::remove_file(path); - - let listener = UnixListener::bind(path).map_err(|e| { - std::io::Error::new( - e.kind(), - format!("Failed to bind Unix socket at {:?}: {}", path, e), - ) - })?; - info!("9P server listening on Unix socket {:?}", path); + Transport::Unix(listener) => { + info!( + "9P server listening on Unix socket {:?}", + listener.local_addr()? + ); loop { tokio::select! { biased; _ = shutdown.cancelled() => { - info!("9P Unix socket server shutting down at {:?}", path); + info!("9P Unix socket server shutting down at {:?}", listener.local_addr()?); break Ok(()); } finished = clients.next(), if !clients.is_empty() => { diff --git a/zerofs/src/zerofs_client_tests.rs b/zerofs/src/zerofs_client_tests.rs index 5a5d2e2a..d157c52e 100644 --- a/zerofs/src/zerofs_client_tests.rs +++ b/zerofs/src/zerofs_client_tests.rs @@ -12,7 +12,7 @@ use zerofs_client::{ }; fn start_server(fs: Arc, sock: std::path::PathBuf) -> CancellationToken { - let server = NinePServer::new_unix(fs, sock); + let server = NinePServer::new_unix(fs, sock).unwrap(); let shutdown = CancellationToken::new(); let server_shutdown = shutdown.clone(); tokio::spawn(async move {