diff --git a/CLAUDE.md b/CLAUDE.md index fec772b..916aa04 100644 --- a/CLAUDE.md +++ b/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` (clap derive) and are read from anywhere via `OPTIONS`. diff --git a/server/src/main.rs b/server/src/main.rs index 39f45be..539287c 100644 --- a/server/src/main.rs +++ b/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() } diff --git a/server/src/server.rs b/server/src/server.rs index 3ad1c00..6d588f2 100644 --- a/server/src/server.rs +++ b/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, + shutdown: Arc, } 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> = 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>) { - 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, + idle: Condvar, +} + +impl InFlight { + fn start(self: &Arc) -> 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); + +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(); + } } } diff --git a/server/tests/shutdown_test.rs b/server/tests/shutdown_test.rs index c85a873..1df74a0 100644 --- a/server/tests/shutdown_test.rs +++ b/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()); }