Use sd_notify to notify systemd when the server is ready

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.

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;
@@ -29,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(
@@ -92,10 +94,10 @@ async fn start_nfs_servers(
     fs: Arc<ZeroFS>,
     config: Option<&NfsConfig>,
     shutdown: CancellationToken,
-) -> Vec<JoinHandle<Result<(), std::io::Error>>> {
+) -> Result<Vec<JoinHandle<Result<(), std::io::Error>>>> {
     let config = match config {
         Some(c) => c,
-        None => return Vec::new(),
+        None => return Ok(Vec::new()),
     };
     let mut handles = Vec::new();
 
@@ -105,17 +107,24 @@ 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())),
                 }
             }));
         }
     }
 
-    handles
+    Ok(handles)
 }
 
 fn start_ninep_servers(
@@ -132,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
@@ -147,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
@@ -189,10 +198,10 @@ async fn start_nbd_servers(
     fs: Arc<ZeroFS>,
     config: Option<&NbdConfig>,
     shutdown: CancellationToken,
-) -> Vec<JoinHandle<Result<(), std::io::Error>>> {
+) -> Result<Vec<JoinHandle<Result<(), std::io::Error>>>> {
     let config = match config {
         Some(c) => c,
-        None => return Vec::new(),
+        None => return Ok(Vec::new()),
     };
     let mut handles = Vec::new();
 
@@ -202,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 {
@@ -219,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 {
@@ -230,7 +239,7 @@ async fn start_nbd_servers(
         }));
     }
 
-    handles
+    Ok(handles)
 }
 
 async fn start_rpc_servers(
@@ -940,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),
@@ -953,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.
@@ -1096,6 +1105,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

zerofs/src/mount.rs

@@ -2029,7 +2029,7 @@ mod client_tests {
     }
 
     fn start_server(fs: Arc<ZeroFS>, 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 {

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<ZeroFS>, socket: SocketAddr) -> Self {
-        Self {
+    pub async fn new_tcp(filesystem: Arc<ZeroFS>, socket: SocketAddr) -> std::io::Result<Self> {
+        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<ZeroFS>, socket_path: impl Into<std::path::PathBuf>) -> Self {
-        Self {
+    pub fn new_unix(
+        filesystem: Arc<ZeroFS>,
+        socket_path: impl Into<std::path::PathBuf>,
+    ) -> std::io::Result<Self> {
+        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<S>(&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() => {

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<ZeroFS>,
     socket: SocketAddr,
-    shutdown: CancellationToken,
-) -> anyhow::Result<()> {
+) -> anyhow::Result<NFSTcpListener<NFSAdapter>> {
     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)]