Make shutdown event-driven and address review findings

Replace the non-blocking accept + 50ms poll with a plain blocking
accept(). A signal-hook thread waits for SIGTERM/SIGINT, sets a flag,
and wakes accept() with a throwaway connection to the socket.

On shutdown the listener is closed and the socket file unlinked before
the grace period, so clients that connect late fail fast with ENOENT
instead of queueing and being reset. In-flight handlers are tracked by
a counter + Condvar rather than a Vec of JoinHandles.

Accepted streams now get a 5s read/write timeout so a silent client
cannot pin a handler thread or stretch every shutdown to the full
grace period. Transient accept() errors are logged and retried instead
of exiting the daemon. A second SIGTERM/SIGINT during the grace period
forces an immediate exit with status 1.

Tests: wrap the child in the kill-on-drop guard immediately after
spawn, drop the redundant wall-clock assertion, and add a case for a
client that connects but never sends.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Kmrgkpp9YBJ6AMaWYPPZuD

CLAUDE.md

@@ -56,11 +56,14 @@ rejected before touching the filesystem. `service` and `realm` are read but igno
 ### Server lifecycle (`server/src/server.rs`)
 
 Startup removes any stale file at the socket path, binds, and chmods the socket 0777.
-The accept loop is non-blocking and polls a flag set by `signal-hook` for SIGTERM/SIGINT.
-On a signal it stops accepting, waits up to 3 s for in-flight handler threads, then
-`Drop` removes the socket file. Exit status is 0. This matters because as container
-PID 1 the process gets no default signal handling; without it `docker stop` hangs for
-10 s and host reboots stall.
+The accept loop is a plain blocking `accept()`. A `signal-hook` thread waits for
+SIGTERM/SIGINT; on the first one it sets a shutdown flag and makes a throwaway
+connection to the socket to wake the accept loop, which then closes the listener,
+unlinks the socket file, and waits up to 3 s for in-flight handler threads before
+exiting 0. A second signal exits immediately with status 1. Accepted streams get a 5 s
+read/write timeout so a silent client cannot pin a thread. This matters because as
+container PID 1 the process gets no default signal handling; without it `docker stop`
+hangs for 10 s and host reboots stall.
 
 CLI options live in `options.rs` as a `LazyLock<Opt>` (clap derive) and are read from
 anywhere via `OPTIONS`.

server/src/main.rs

@@ -11,6 +11,6 @@ fn main() -> std::io::Result<()> {
     env_logger::init();
     
     let repository = PasswordDirectory::new(&OPTIONS.password_dir);
-    let mut server = Server::new(&OPTIONS.socket_name, repository)?;
+    let server = Server::new(&OPTIONS.socket_name, repository)?;
     server.run()
 }

server/src/server.rs

@@ -1,30 +1,42 @@
 use crate::handler::Handler;
 use crate::repository::PasswordDirectory;
-use log::{info, warn};
+use log::{error, info, warn};
 use signal_hook::consts::{SIGINT, SIGTERM};
 use signal_hook::flag;
+use signal_hook::iterator::Signals;
+use signal_hook::low_level::signal_name;
 use std::fs::{metadata, set_permissions};
 use std::io::{Error, ErrorKind};
 use std::os::unix::fs::PermissionsExt;
-use std::os::unix::net::UnixListener;
+use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::PathBuf;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::thread::{self, JoinHandle};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Condvar, Mutex};
+use std::thread;
 use std::time::{Duration, Instant};
 
-/// How long to sleep between polls of the non-blocking listener while idle.
-const ACCEPT_POLL_INTERVAL: Duration = Duration::from_millis(50);
+/// Signals that request a graceful shutdown.
+const SHUTDOWN_SIGNALS: [i32; 2] = [SIGTERM, SIGINT];
+
+/// Exit status used when a second shutdown signal forces an immediate exit.
+const FORCED_EXIT_STATUS: i32 = 1;
+
+/// Upper bound on how long a single client may take to send its request
+/// or receive its response.
+const CLIENT_IO_TIMEOUT: Duration = Duration::from_secs(5);
 
 /// How long to wait for in-flight client handlers to finish on shutdown.
 const SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(3);
 
+/// How long to pause after a transient accept() failure (e.g. EMFILE)
+/// before trying again.
+const ACCEPT_RETRY_DELAY: Duration = Duration::from_millis(100);
+
 pub struct Server {
     listener: UnixListener,
+    socket_file: SocketFile,
     repository: PasswordDirectory,
-    socket_name: PathBuf,
-    /// Set to the signal number (SIGTERM / SIGINT) once one has been received; 0 otherwise.
-    shutdown_signal: Arc<AtomicUsize>,
+    shutdown: Arc<AtomicBool>,
 }
 
 impl Server {
@@ -36,107 +48,163 @@ impl Server {
         }
 
         let listener = UnixListener::bind(&socket_name)?;
+        let socket_file = SocketFile(socket_name.clone());
 
         let mut perms = metadata(&socket_name)?.permissions();
         perms.set_mode(0o0777);
         set_permissions(&socket_name, perms)?;
 
-        // Non-blocking so the accept loop can notice a shutdown request.
-        listener.set_nonblocking(true)?;
+        let shutdown = Arc::new(AtomicBool::new(false));
 
-        let shutdown_signal = Arc::new(AtomicUsize::new(0));
-        flag::register_usize(SIGTERM, Arc::clone(&shutdown_signal), SIGTERM as usize)?;
-        flag::register_usize(SIGINT, Arc::clone(&shutdown_signal), SIGINT as usize)?;
+        // A second SIGTERM/SIGINT while already shutting down exits immediately.
+        // Registered first so it runs before the handler that sets the flag.
+        for signal in SHUTDOWN_SIGNALS {
+            flag::register_conditional_shutdown(signal, FORCED_EXIT_STATUS, Arc::clone(&shutdown))?;
+        }
+
+        // The first SIGTERM/SIGINT sets the flag, then wakes the blocking
+        // accept() with a throwaway connection to the socket.
+        let mut signals = Signals::new(SHUTDOWN_SIGNALS)?;
+        let wake = (Arc::clone(&shutdown), socket_name);
+        thread::spawn(move || {
+            if let Some(signal) = signals.forever().next() {
+                let name = signal_name(signal).unwrap_or("signal");
+                info!("received {name}, shutting down");
+                wake.0.store(true, Ordering::SeqCst);
+                if let Err(err) = UnixStream::connect(&wake.1) {
+                    warn!("could not wake the accept loop: {err}");
+                }
+            }
+        });
+
+        info!("listening on {}", socket_file.0.display());
 
         Ok(Server {
-            repository,
             listener,
-            socket_name,
-            shutdown_signal,
+            socket_file,
+            repository,
+            shutdown,
         })
     }
 
-    pub fn run(&mut self) -> Result<(), Error> {
-        let mut workers: Vec<JoinHandle<()>> = Vec::new();
+    pub fn run(self) -> Result<(), Error> {
+        let Server {
+            listener,
+            socket_file,
+            repository,
+            shutdown,
+        } = self;
+
+        let in_flight = Arc::new(InFlight::default());
 
         // accept connections and process them, spawning a new thread for each one,
         // until a shutdown signal arrives
         loop {
-            let signal = self.shutdown_signal.load(Ordering::Relaxed);
-            if signal != 0 {
-                info!("received {}, shutting down", signal_name(signal));
-                break;
-            }
-
-            match self.listener.accept() {
+            match listener.accept() {
                 Ok((stream, _addr)) => {
-                    /* connection succeeded */
-                    // On some platforms (e.g. macOS) the accepted stream inherits the
-                    // listener's non-blocking mode; the handler expects blocking I/O.
-                    stream.set_nonblocking(false)?;
-                    let repository = self.repository.clone();
-                    workers.push(thread::spawn(|| {
+                    if shutdown.load(Ordering::SeqCst) {
+                        // Either the wake-up connection from the signal thread or a
+                        // client that raced the shutdown; either way, stop here.
+                        break;
+                    }
+                    if let Err(err) = configure_stream(&stream) {
+                        error!("could not configure client connection: {err}");
+                        continue;
+                    }
+                    let repository = repository.clone();
+                    let guard = in_flight.start();
+                    thread::spawn(move || {
+                        let _guard = guard;
                         Handler::new(stream, repository).handle_client()
-                    }));
-                    workers.retain(|worker| !worker.is_finished());
-                }
-                Err(err) if err.kind() == ErrorKind::WouldBlock => {
-                    /* nothing pending */
-                    thread::sleep(ACCEPT_POLL_INTERVAL);
+                    });
                 }
+                Err(err) if err.kind() == ErrorKind::Interrupted => {}
                 Err(err) => {
-                    /* connection failed */
-                    return Err(err);
+                    // Transient failures (EMFILE, ECONNABORTED, ...) should not
+                    // take the daemon down; pause briefly and keep serving.
+                    error!("accept failed: {err}");
+                    if shutdown.load(Ordering::SeqCst) {
+                        break;
+                    }
+                    thread::sleep(ACCEPT_RETRY_DELAY);
                 }
             }
         }
 
-        // Stop accepting new connections; the listener is dropped with the Server,
-        // but from here on we simply never call accept() again.
-        self.wait_for_workers(workers);
+        // Stop accepting and remove the socket path right away, so clients that
+        // connect from here on fail fast instead of queueing behind the shutdown.
+        drop(listener);
+        drop(socket_file);
+
+        in_flight.wait_until_idle(SHUTDOWN_GRACE_PERIOD);
+        info!("shutdown complete");
 
         Ok(())
     }
+}
 
-    /// Give in-flight client handlers a bounded amount of time to finish.
-    fn wait_for_workers(&self, mut workers: Vec<JoinHandle<()>>) {
-        workers.retain(|worker| !worker.is_finished());
-        if workers.is_empty() {
+/// Put an accepted stream into blocking mode with bounded I/O, so a client that
+/// stalls cannot pin a handler thread forever.
+fn configure_stream(stream: &UnixStream) -> Result<(), Error> {
+    // On some platforms (e.g. macOS) an accepted stream can inherit the
+    // listener's flags; the handler expects blocking I/O.
+    stream.set_nonblocking(false)?;
+    stream.set_read_timeout(Some(CLIENT_IO_TIMEOUT))?;
+    stream.set_write_timeout(Some(CLIENT_IO_TIMEOUT))?;
+    Ok(())
+}
+
+/// Owns the socket path on disk; removes it when dropped.
+struct SocketFile(PathBuf);
+
+impl Drop for SocketFile {
+    fn drop(&mut self) {
+        let _result = std::fs::remove_file(&self.0);
+    }
+}
+
+/// Counts client handlers that are still running.
+#[derive(Default)]
+struct InFlight {
+    count: Mutex<usize>,
+    idle: Condvar,
+}
+
+impl InFlight {
+    fn start(self: &Arc<Self>) -> InFlightGuard {
+        *self.count.lock().unwrap() += 1;
+        InFlightGuard(Arc::clone(self))
+    }
+
+    fn wait_until_idle(&self, timeout: Duration) {
+        let mut count = self.count.lock().unwrap();
+        if *count == 0 {
             return;
         }
 
-        info!("waiting for {} in-flight request(s) to finish", workers.len());
-        let deadline = Instant::now() + SHUTDOWN_GRACE_PERIOD;
+        info!("waiting for {} in-flight request(s) to finish", *count);
+        let deadline = Instant::now() + timeout;
 
-        while Instant::now() < deadline {
-            workers.retain(|worker| !worker.is_finished());
-            if workers.is_empty() {
+        while *count > 0 {
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                warn!("abandoning {} request(s) still in flight after {timeout:?}", *count);
                 break;
             }
-            thread::sleep(Duration::from_millis(10));
-        }
-
-        for worker in workers {
-            if worker.is_finished() {
-                let _ = worker.join();
-            } else {
-                warn!("abandoning a request still in flight after {SHUTDOWN_GRACE_PERIOD:?}");
-            }
+            count = self.idle.wait_timeout(count, remaining).unwrap().0;
         }
     }
 }
 
-impl Drop for Server {
+/// Decrements the in-flight count when the handler thread finishes, even on panic.
+struct InFlightGuard(Arc<InFlight>);
+
+impl Drop for InFlightGuard {
     fn drop(&mut self) {
-        let _result = std::fs::remove_file(&self.socket_name);
-        info!("shutdown complete");
-    }
-}
-
-fn signal_name(signal: usize) -> &'static str {
-    match signal as i32 {
-        SIGTERM => "SIGTERM",
-        SIGINT => "SIGINT",
-        _ => "signal",
+        let mut count = self.0.count.lock().unwrap();
+        *count -= 1;
+        if *count == 0 {
+            self.0.idle.notify_all();
+        }
     }
 }

server/tests/shutdown_test.rs

@@ -6,6 +6,10 @@ use std::path::PathBuf;
 use std::process::{Child, Command};
 use std::time::{Duration, Instant};
 
+/// Idle shutdown takes milliseconds; anything approaching the daemon's 3s
+/// grace period means it is waiting on something it should not be.
+const EXIT_TIMEOUT: Duration = Duration::from_secs(2);
+
 const PASSWORD_DIR: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../test_passwords");
 
 fn temp_socket_path(tag: &str) -> PathBuf {
@@ -41,13 +45,15 @@ fn spawn_daemon(socket: &PathBuf) -> Daemon {
         .arg(PASSWORD_DIR)
         .spawn()
         .expect("failed to spawn saslauthd");
+    // Wrap immediately so a failed startup below still kills the process.
+    let daemon = Daemon(child);
 
     let deadline = Instant::now() + Duration::from_secs(5);
     while !is_socket(socket) {
         assert!(Instant::now() < deadline, "daemon never created {socket:?}");
         std::thread::sleep(Duration::from_millis(10));
     }
-    Daemon(child)
+    daemon
 }
 
 fn send_signal(daemon: &Daemon, signal: &str) {
@@ -68,8 +74,6 @@ fn wait_for_exit(daemon: &mut Daemon, timeout: Duration) -> std::process::ExitSt
             return status;
         }
         if Instant::now() >= deadline {
-            let _ = child.kill();
-            let _ = child.wait();
             panic!("daemon did not exit within {timeout:?}");
         }
         std::thread::sleep(Duration::from_millis(10));
@@ -92,16 +96,10 @@ fn assert_clean_shutdown(signal: &str) {
     assert_eq!(authenticate(&socket, "some-rad-password"), "OK");
     assert_eq!(authenticate(&socket, "wrong-password"), "NO");
 
-    let started = Instant::now();
     send_signal(&daemon, signal);
-    let status = wait_for_exit(&mut daemon, Duration::from_secs(1));
+    let status = wait_for_exit(&mut daemon, EXIT_TIMEOUT);
 
     assert!(status.success(), "expected exit status 0 on {signal}, got {status}");
-    assert!(
-        started.elapsed() < Duration::from_secs(1),
-        "shutdown took {:?}",
-        started.elapsed()
-    );
     assert!(!socket.exists(), "socket file {socket:?} was left behind");
 }
 
@@ -124,7 +122,22 @@ fn it_recreates_a_stale_socket_on_startup() {
     assert_eq!(authenticate(&socket, "some-rad-password"), "OK");
 
     send_signal(&daemon, "TERM");
-    let status = wait_for_exit(&mut daemon, Duration::from_secs(1));
+    let status = wait_for_exit(&mut daemon, EXIT_TIMEOUT);
+    assert!(status.success());
+    assert!(!socket.exists());
+}
+
+#[test]
+fn it_does_not_wait_forever_for_a_silent_client() {
+    let socket = temp_socket_path("silent");
+    let mut daemon = spawn_daemon(&socket);
+
+    // Connect but never send a request, leaving a handler blocked on read.
+    let _silent = UnixStream::connect(&socket).expect("connect to daemon");
+
+    send_signal(&daemon, "TERM");
+    // Must finish once the grace period (3s) expires, whatever the client does.
+    let status = wait_for_exit(&mut daemon, Duration::from_secs(5));
     assert!(status.success());
     assert!(!socket.exists());
 }