crates/nunciod/src/grpc.rs

@@ -2283,6 +2283,73 @@ pub async fn serve_on_listener(
     .await
 }
 
+/// Binds a loopback TCP listener at `addr` exactly like [`serve`], but stops
+/// accepting new connections and begins draining in-flight requests as soon
+/// as `shutdown` resolves, instead of running until the transport errors.
+///
+/// `addr` MUST be a loopback address; see [`serve`].
+pub async fn serve_with_shutdown(
+    addr: &str,
+    event_bus: Arc<EventBus>,
+    db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
+    secrets: Arc<SecretManager>,
+    token: impl Into<Arc<str>>,
+    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
+) -> Result<(), GrpcServeError> {
+    let socket_addr = addr
+        .parse::<std::net::SocketAddr>()
+        .map_err(|_| GrpcServeError::NonLoopbackAddress(addr.to_string()))?;
+    if !socket_addr.ip().is_loopback() {
+        return Err(GrpcServeError::NonLoopbackAddress(addr.to_string()));
+    }
+
+    let listener = TcpListener::bind(addr)
+        .await
+        .map_err(|source| GrpcServeError::Bind {
+            addr: addr.to_string(),
+            source,
+        })?;
+    serve_on_listener_with_shutdown(
+        listener,
+        event_bus,
+        db,
+        filter_engine,
+        secrets,
+        token,
+        shutdown,
+    )
+    .await
+}
+
+/// Identical to [`serve_on_listener`], except the server stops accepting new
+/// connections and begins draining in-flight requests as soon as `shutdown`
+/// resolves.
+pub async fn serve_on_listener_with_shutdown(
+    listener: TcpListener,
+    event_bus: Arc<EventBus>,
+    db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
+    secrets: Arc<SecretManager>,
+    token: impl Into<Arc<str>>,
+    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
+) -> Result<(), GrpcServeError> {
+    serve_on_listener_with_overrides_and_shutdown(
+        listener,
+        event_bus,
+        db,
+        filter_engine,
+        secrets,
+        token,
+        MailEngineOverrides::default(),
+        CalendarEngineOverrides::default(),
+        ContactsEngineOverrides::default(),
+        AccountsEngineOverrides::default(),
+        shutdown,
+    )
+    .await
+}
+
 /// Identical to [`serve_on_listener`], except the `nuncio.v1.Mail` service's
 /// inbound sync and outbound send engines, the `nuncio.v1.Calendar`
 /// service's sync backend, and the `nuncio.v1.Contacts` service's sync
@@ -2306,6 +2373,45 @@ pub async fn serve_on_listener_with_overrides(
     calendar_overrides: CalendarEngineOverrides,
     contacts_overrides: ContactsEngineOverrides,
     accounts_overrides: AccountsEngineOverrides,
+) -> Result<(), GrpcServeError> {
+    // Delegates to the shutdown-aware variant with a shutdown future that
+    // never resolves, preserving this function's exact prior behavior (run
+    // until the transport itself errors) for every existing caller.
+    serve_on_listener_with_overrides_and_shutdown(
+        listener,
+        event_bus,
+        db,
+        filter_engine,
+        secrets,
+        token,
+        overrides,
+        calendar_overrides,
+        contacts_overrides,
+        accounts_overrides,
+        std::future::pending(),
+    )
+    .await
+}
+
+/// Identical to [`serve_on_listener_with_overrides`], except the server
+/// stops accepting new connections and begins draining in-flight requests
+/// as soon as `shutdown` resolves, instead of running until the transport
+/// errors. This is the function production actually calls (via
+/// [`serve_with_shutdown`]); [`serve_on_listener_with_overrides`] delegates
+/// here with a `shutdown` future that never resolves.
+#[allow(clippy::too_many_arguments)]
+pub async fn serve_on_listener_with_overrides_and_shutdown(
+    listener: TcpListener,
+    event_bus: Arc<EventBus>,
+    db: Arc<DatabaseEngine>,
+    filter_engine: Arc<FilterEngine>,
+    secrets: Arc<SecretManager>,
+    token: impl Into<Arc<str>>,
+    overrides: MailEngineOverrides,
+    calendar_overrides: CalendarEngineOverrides,
+    contacts_overrides: ContactsEngineOverrides,
+    accounts_overrides: AccountsEngineOverrides,
+    shutdown: impl std::future::Future<Output = ()> + Send + 'static,
 ) -> Result<(), GrpcServeError> {
     let token: Arc<str> = token.into();
 
@@ -2392,7 +2498,7 @@ pub async fn serve_on_listener_with_overrides(
         .add_service(audit_svc)
         .add_service(calendar_svc)
         .add_service(contacts_svc)
-        .serve_with_incoming(TcpListenerStream::new(listener))
+        .serve_with_incoming_shutdown(TcpListenerStream::new(listener), shutdown)
         .await
         .map_err(GrpcServeError::Transport)
 }

crates/nunciod/src/lib.rs

@@ -5,6 +5,7 @@
 pub mod calendar_sync;
 pub mod contacts_sync;
 pub mod grpc;
+pub mod lifecycle;
 pub mod orchestrator;
 pub mod outbox;
 pub mod send;

crates/nunciod/src/lifecycle.rs

@@ -0,0 +1,157 @@
+//! Coordinates graceful shutdown across every long-running task `nunciod`
+//! spawns: the gRPC server, the outbox worker, the auto-update-check loop,
+//! and the real-sync `CoreCommand` consumer.
+//!
+//! Previously the daemon installed no signal handler at all: every real
+//! exit was an OS kill (SIGKILL/SIGTERM without a handler, a terminated
+//! console), which races [`nuncio_store::DatabaseEngine::close`]'s WAL
+//! checkpoint and can corrupt the database. A single [`ShutdownSignal`],
+//! shared by every task, lets each one stop accepting new work and finish
+//! cleanly instead.
+
+use nuncio_core::{CoreCommand, EventBus};
+use std::sync::Arc;
+use tokio::sync::watch;
+
+/// Read-only handle to the shared shutdown flag. Cloned into every
+/// long-running task; `wait` resolves once shutdown has been requested,
+/// including if it was already requested before this clone started
+/// waiting.
+#[derive(Clone)]
+pub struct ShutdownSignal {
+    rx: watch::Receiver<bool>,
+}
+
+impl ShutdownSignal {
+    /// Resolves once shutdown has been requested. Safe to call from
+    /// `tokio::select!` on every loop iteration: a task that observes
+    /// shutdown mid-iteration finishes its current unit of work (the other
+    /// `select!` branch) before this is polled again, so no in-flight
+    /// database write is abandoned half-done.
+    pub async fn wait(&mut self) {
+        // `watch::Receiver::wait_for` returns immediately if the current
+        // value already satisfies the predicate, so a clone created after
+        // shutdown was already requested does not hang waiting for a
+        // change that already happened.
+        let _ = self.rx.wait_for(|requested| *requested).await;
+    }
+
+    /// Non-blocking check for loops that need to poll rather than await.
+    pub fn is_requested(&self) -> bool {
+        *self.rx.borrow()
+    }
+}
+
+/// Owns the shutdown trigger. Production wires [`ShutdownController::trigger`]
+/// to OS signal delivery via [`install_signal_handlers`]; tests call it
+/// directly to drive the shutdown path deterministically without sending a
+/// real OS signal.
+pub struct ShutdownController {
+    tx: watch::Sender<bool>,
+    event_bus: Arc<EventBus>,
+}
+
+impl ShutdownController {
+    /// Creates a controller/signal pair. `event_bus` is the SAME bus every
+    /// other subsystem shares, so [`trigger`](Self::trigger) drives the
+    /// existing `CoreCommand::Shutdown` / `CoreEvent::ShuttingDown` path
+    /// visibly (state flips to `ShuttingDown`, subscribers are notified)
+    /// instead of that path staying reachable only from tests.
+    pub fn new(event_bus: Arc<EventBus>) -> (Self, ShutdownSignal) {
+        let (tx, rx) = watch::channel(false);
+        (Self { tx, event_bus }, ShutdownSignal { rx })
+    }
+
+    /// Initiates shutdown. Idempotent: calling it more than once (e.g. a
+    /// second Ctrl+C while already draining) is a no-op beyond the first.
+    pub fn trigger(&self) {
+        self.event_bus.process_command(CoreCommand::Shutdown);
+        let _ = self.tx.send(true);
+    }
+}
+
+/// Waits for Ctrl+C, or (on Unix) SIGTERM, then triggers shutdown. Intended
+/// to be spawned once at startup and left running for the process lifetime.
+pub async fn install_signal_handlers(controller: Arc<ShutdownController>) {
+    wait_for_os_signal().await;
+    controller.trigger();
+}
+
+#[cfg(unix)]
+async fn wait_for_os_signal() {
+    use tokio::signal::unix::{signal, SignalKind};
+
+    match signal(SignalKind::terminate()) {
+        Ok(mut sigterm) => {
+            tokio::select! {
+                _ = tokio::signal::ctrl_c() => {
+                    tracing::info!("received Ctrl+C; starting graceful shutdown");
+                }
+                _ = sigterm.recv() => {
+                    tracing::info!("received SIGTERM; starting graceful shutdown");
+                }
+            }
+        }
+        Err(err) => {
+            // Ctrl+C still works even if SIGTERM registration failed (e.g.
+            // an exotic sandboxed environment); log so the gap is visible
+            // rather than silently dropping SIGTERM handling.
+            tracing::error!("failed to install SIGTERM handler: {err}; SIGTERM will kill the process without a clean shutdown");
+            let _ = tokio::signal::ctrl_c().await;
+            tracing::info!("received Ctrl+C; starting graceful shutdown");
+        }
+    }
+}
+
+#[cfg(not(unix))]
+async fn wait_for_os_signal() {
+    let _ = tokio::signal::ctrl_c().await;
+    tracing::info!("received Ctrl+C; starting graceful shutdown");
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nuncio_core::EngineStatus;
+
+    #[tokio::test]
+    async fn trigger_flips_signal_and_event_bus_state() {
+        let event_bus = Arc::new(EventBus::new());
+        let (controller, mut signal) = ShutdownController::new(event_bus.clone());
+
+        assert!(!signal.is_requested());
+        assert_eq!(event_bus.current_state().status, EngineStatus::Idle);
+
+        controller.trigger();
+
+        assert!(signal.is_requested());
+        assert_eq!(event_bus.current_state().status, EngineStatus::ShuttingDown);
+        // Resolves immediately: the signal was already flipped above.
+        signal.wait().await;
+    }
+
+    #[tokio::test]
+    async fn signal_clone_observes_a_trigger_that_happened_before_it_was_created() {
+        let event_bus = Arc::new(EventBus::new());
+        let (controller, signal) = ShutdownController::new(event_bus);
+        controller.trigger();
+
+        // A clone taken AFTER the trigger must still resolve `wait`
+        // immediately rather than hang waiting for a fresh `changed()`.
+        let mut late_clone = signal.clone();
+        late_clone.wait().await;
+        assert!(late_clone.is_requested());
+    }
+
+    #[tokio::test]
+    async fn trigger_is_idempotent() {
+        let event_bus = Arc::new(EventBus::new());
+        let (controller, signal) = ShutdownController::new(event_bus.clone());
+
+        controller.trigger();
+        controller.trigger();
+
+        assert!(signal.is_requested());
+        assert_eq!(event_bus.current_state().status, EngineStatus::ShuttingDown);
+    }
+}

crates/nunciod/src/main.rs

@@ -4,7 +4,16 @@
 
 use nuncio_core::{CoreCommand, CoreEvent, EventBus};
 use nuncio_filter::FilterEngine;
+use nunciod::lifecycle::{install_signal_handlers, ShutdownController};
 use std::sync::Arc;
+use std::time::Duration;
+
+/// Bound on how long shutdown waits for in-flight work (gRPC requests still
+/// being served, an outbox drain pass in progress, a background task's
+/// current loop iteration) to finish on its own before the process moves on
+/// to closing the database and exiting anyway. Keeps a stuck task from
+/// turning a graceful shutdown into a hang.
+const SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(10);
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -17,6 +26,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let mut event_bus_owned = EventBus::new();
     let command_rx = event_bus_owned.take_command_receiver();
     let event_bus = Arc::new(event_bus_owned);
+
+    // Single shutdown signal shared by every long-running task spawned
+    // below (the real-sync command consumer, the outbox worker, the
+    // update-check loop, and the gRPC server itself), so a Ctrl+C/SIGTERM
+    // drains all of them cleanly instead of the OS killing the process
+    // mid-write. `install_signal_handlers` also drives the existing
+    // `CoreCommand::Shutdown` / `CoreEvent::ShuttingDown` event-bus path,
+    // previously wired but never triggered outside tests.
+    let (shutdown_controller, shutdown_signal) = ShutdownController::new(event_bus.clone());
+    let shutdown_controller = Arc::new(shutdown_controller);
+    let signal_task = tokio::spawn(install_signal_handlers(shutdown_controller));
     // PERSISTENT database path: defaults to `~/.nuncio/nuncio.db` -- NOT a
     // temp/ephemeral path -- so accounts (and everything else) survive a
     // daemon restart.
@@ -52,13 +72,24 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // trigger_background_resync`'s post-recovery `send_command` calls a
     // consumer: previously nothing read this channel, so that resync
     // trigger was inert.
-    if let Some(mut command_rx) = command_rx {
+    let sync_command_task = if let Some(mut command_rx) = command_rx {
         let db_sync = db.clone();
         let secrets_sync = account_secrets.clone();
         let event_bus_sync = event_bus.clone();
         let filter_engine_sync = filter_engine.clone();
-        let _sync_command_task = tokio::spawn(async move {
-            while let Some(cmd) = command_rx.recv().await {
+        let mut shutdown = shutdown_signal.clone();
+        Some(tokio::spawn(async move {
+            loop {
+                let cmd = tokio::select! {
+                    _ = shutdown.wait() => {
+                        tracing::info!("sync command consumer exiting on shutdown signal");
+                        break;
+                    }
+                    received = command_rx.recv() => match received {
+                        Some(cmd) => cmd,
+                        None => break,
+                    },
+                };
                 match cmd {
                     CoreCommand::SyncAll => {
                         let synced = nunciod::sync::run_all_accounts_sync(
@@ -93,12 +124,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                     _ => {}
                 }
             }
-        });
+        }))
     } else {
         tracing::error!(
             "EventBus command receiver was already taken; real sync command processing will not run"
         );
-    }
+        None
+    };
 
     // Background Outbox Worker Task. Each tick drains the pending remote
     // mutations the filter engine enqueued (move/copy/flag/unflag/delete on the
@@ -112,20 +144,34 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         db.clone(),
         account_secrets.clone(),
     ));
-    let _outbox_task = tokio::spawn(async move {
+    let mut outbox_shutdown = shutdown_signal.clone();
+    let outbox_task = tokio::spawn(async move {
         let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
         loop {
-            interval.tick().await;
-            let summary =
-                nunciod::outbox::execute_pending_mutations(&db_outbox, outbox_env.as_ref(), 50)
+            tokio::select! {
+                _ = outbox_shutdown.wait() => {
+                    tracing::info!("outbox worker exiting on shutdown signal");
+                    break;
+                }
+                _ = interval.tick() => {
+                    // Runs to completion once selected: a drain pass already
+                    // in progress is never aborted mid-write by a shutdown
+                    // signal that arrives after this branch was chosen.
+                    let summary = nunciod::outbox::execute_pending_mutations(
+                        &db_outbox,
+                        outbox_env.as_ref(),
+                        50,
+                    )
                     .await;
-            if summary.completed > 0 || summary.failed > 0 {
-                tracing::info!(
-                    "outbox drain: {} completed, {} retried, {} failed",
-                    summary.completed,
-                    summary.retried,
-                    summary.failed
-                );
+                    if summary.completed > 0 || summary.failed > 0 {
+                        tracing::info!(
+                            "outbox drain: {} completed, {} retried, {} failed",
+                            summary.completed,
+                            summary.retried,
+                            summary.failed
+                        );
+                    }
+                }
             }
         }
     });
@@ -138,21 +184,29 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     // closed, nunciod must never autonomously check for or install
     // updates, so this loop only runs when an operator explicitly opts in
     // via `NUNCIO_AUTO_UPDATE_ENABLED=1`.
-    let _update_task = if nunciod::auto_update_task_enabled() {
+    let update_task = if nunciod::auto_update_task_enabled() {
         let event_bus_update = event_bus.clone();
+        let mut update_shutdown = shutdown_signal.clone();
         Some(tokio::spawn(async move {
             let mut interval = tokio::time::interval(std::time::Duration::from_secs(86400));
             interval.tick().await;
             loop {
-                interval.tick().await;
-                if let Ok(updater) = nuncio_core::UpdateEngine::new() {
-                    if let Ok(result) = updater.check_for_updates().await {
-                        if result.update_available {
-                            if let Some(info) = result.release_info {
-                                event_bus_update.publish_event(CoreEvent::UpdateAvailable {
-                                    version: info.version,
-                                    release_notes: info.release_notes,
-                                });
+                tokio::select! {
+                    _ = update_shutdown.wait() => {
+                        tracing::info!("update-check loop exiting on shutdown signal");
+                        break;
+                    }
+                    _ = interval.tick() => {
+                        if let Ok(updater) = nuncio_core::UpdateEngine::new() {
+                            if let Ok(result) = updater.check_for_updates().await {
+                                if result.update_available {
+                                    if let Some(info) = result.release_info {
+                                        event_bus_update.publish_event(CoreEvent::UpdateAvailable {
+                                            version: info.version,
+                                            release_notes: info.release_notes,
+                                        });
+                                    }
+                                }
                             }
                         }
                     }
@@ -192,17 +246,57 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
 
     // The gRPC server is the long-running foreground task that keeps this
     // process alive; every other subsystem above is a background worker
-    // spawned on top of it.
-    nunciod::grpc::serve(
-        &grpc_addr,
-        event_bus,
-        db,
-        filter_engine,
-        grpc_secrets,
-        grpc_token,
-    )
-    .await
-    .map_err(|e| format!("nunciod gRPC server failed: {e}"))?;
+    // spawned on top of it. `serve_with_shutdown` stops accepting new
+    // connections and starts draining in-flight requests as soon as
+    // `shutdown_signal` fires; the `select!` below bounds how long that
+    // drain is allowed to take before this process moves on regardless.
+    let db_for_serve = db.clone();
+    let mut grpc_shutdown_future = shutdown_signal.clone();
+    let mut grace_timer_signal = shutdown_signal.clone();
+    let grpc_result = tokio::select! {
+        result = nunciod::grpc::serve_with_shutdown(
+            &grpc_addr,
+            event_bus,
+            db_for_serve,
+            filter_engine,
+            grpc_secrets,
+            grpc_token,
+            async move { grpc_shutdown_future.wait().await },
+        ) => result,
+        _ = async move {
+            grace_timer_signal.wait().await;
+            tokio::time::sleep(SHUTDOWN_GRACE_PERIOD).await;
+        } => {
+            tracing::warn!(
+                "gRPC server did not finish draining in-flight requests within the {:?} \
+                 shutdown grace period; closing the database and exiting anyway",
+                SHUTDOWN_GRACE_PERIOD
+            );
+            Ok(())
+        }
+    };
+
+    // Bound how long the background workers get to notice the shutdown
+    // signal and exit their loops before the database is closed out from
+    // under them.
+    let _ = tokio::time::timeout(SHUTDOWN_GRACE_PERIOD, async {
+        if let Some(task) = sync_command_task {
+            let _ = task.await;
+        }
+        let _ = outbox_task.await;
+        if let Some(task) = update_task {
+            let _ = task.await;
+        }
+    })
+    .await;
+    signal_task.abort();
+
+    // Force the WAL checkpoint before the process exits so no committed
+    // page is left for SQLite's own deferred "last connection closes"
+    // teardown to race (see `DatabaseEngine::close`'s doc comment).
+    db.close().await;
+
+    grpc_result.map_err(|e| format!("nunciod gRPC server failed: {e}"))?;
 
     Ok(())
 }

crates/nunciod/tests/graceful_shutdown_test.rs

@@ -0,0 +1,176 @@
+//! Integration test for `nunciod`'s graceful shutdown path.
+//!
+//! Before this, `nunciod` installed no signal handler and never called
+//! [`DatabaseEngine::close`], so every real exit was an OS kill racing the
+//! WAL checkpoint that `close` exists to force. This test drives the same
+//! building blocks `main.rs` wires together -- a [`ShutdownController`]
+//! shared by the gRPC server and a background worker, triggered directly
+//! (never a real OS signal, for determinism) -- and asserts:
+//!   (a) the gRPC server stops serving once the shutdown signal fires;
+//!   (b) a background worker exits its loop on the same signal instead of
+//!       being aborted mid-iteration;
+//!   (c) `DatabaseEngine::close()` runs and leaves the pool unusable for
+//!       any subsequent query.
+
+#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+
+use nuncio_core::EventBus;
+use nuncio_filter::FilterEngine;
+use nuncio_proto::v1::system_client::SystemClient;
+use nuncio_proto::v1::GetStatusRequest;
+use nuncio_store::db::DatabaseEngine;
+use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT};
+use nunciod::grpc::serve_on_listener_with_shutdown;
+use nunciod::lifecycle::ShutdownController;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::net::TcpListener;
+use tonic::Request;
+
+const JOIN_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[tokio::test]
+async fn shutdown_signal_stops_grpc_server_exits_background_worker_and_closes_the_database() {
+    let event_bus = Arc::new(EventBus::new());
+    let (controller, shutdown_signal) = ShutdownController::new(event_bus.clone());
+
+    let secrets = Arc::new(SecretManager::mock());
+    let token = hex::encode(
+        secrets
+            .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32)
+            .expect("token provisioned from mock vault"),
+    );
+    let filter_engine = Arc::new(FilterEngine::new(Vec::new()).expect("empty rule set"));
+    let (db, _dir) = DatabaseEngine::connect_ephemeral()
+        .await
+        .expect("connect ephemeral test db");
+    let db = Arc::new(db);
+
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind ephemeral loopback port");
+    let addr = listener.local_addr().expect("listener has local addr");
+
+    // (a) the gRPC server: mirrors main.rs's `serve_with_shutdown` wiring --
+    // the future it's given resolves from the SAME `ShutdownSignal` the
+    // worker below observes.
+    let mut grpc_shutdown = shutdown_signal.clone();
+    let grpc_task = tokio::spawn(serve_on_listener_with_shutdown(
+        listener,
+        event_bus.clone(),
+        db.clone(),
+        filter_engine,
+        secrets,
+        token.clone(),
+        async move { grpc_shutdown.wait().await },
+    ));
+
+    // Confirm the server actually accepts a call before shutdown, so a
+    // trivially-passing "stopped" observation later isn't just "it never
+    // started".
+    let mut client = SystemClient::connect(format!("http://{addr}"))
+        .await
+        .expect("client connects before shutdown");
+    let mut request = Request::new(GetStatusRequest {});
+    request.metadata_mut().insert(
+        "authorization",
+        format!("Bearer {token}")
+            .parse()
+            .expect("valid ascii metadata value"),
+    );
+    client
+        .get_status(request)
+        .await
+        .expect("authenticated call succeeds before shutdown");
+
+    // (b) a background worker built the same way the outbox worker and
+    // update-check loop are in `main.rs`: a `tokio::select!` between the
+    // shared shutdown signal and a repeating unit of work. `tick_count`
+    // proves the loop body actually ran, ruling out the trivial case where
+    // the select immediately picks the shutdown branch every time.
+    let tick_count = Arc::new(AtomicUsize::new(0));
+    let tick_count_worker = tick_count.clone();
+    let mut worker_shutdown = shutdown_signal.clone();
+    let worker_task = tokio::spawn(async move {
+        let mut interval = tokio::time::interval(Duration::from_millis(20));
+        loop {
+            tokio::select! {
+                _ = worker_shutdown.wait() => break,
+                _ = interval.tick() => {
+                    tick_count_worker.fetch_add(1, Ordering::SeqCst);
+                }
+            }
+        }
+    });
+
+    // Let the worker tick a few times before shutdown is requested.
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert!(
+        tick_count.load(Ordering::SeqCst) > 0,
+        "worker must have run at least one iteration before shutdown"
+    );
+
+    // Trigger shutdown once, from a single controller shared by both tasks
+    // -- exactly the production wiring, minus the real OS signal.
+    controller.trigger();
+
+    let grpc_result = tokio::time::timeout(JOIN_TIMEOUT, grpc_task)
+        .await
+        .expect("gRPC server task must exit within the join timeout, not hang")
+        .expect("gRPC server task must not panic");
+    assert!(
+        grpc_result.is_ok(),
+        "serve_on_listener_with_shutdown must return Ok once it finishes draining, got {grpc_result:?}"
+    );
+
+    tokio::time::timeout(JOIN_TIMEOUT, worker_task)
+        .await
+        .expect("background worker must exit on the shutdown signal within the join timeout, not be killed")
+        .expect("background worker task must not panic");
+
+    // A dial attempt after the server task has exited must fail: the
+    // listener is gone, proving the server actually stopped accepting
+    // connections rather than merely returning while still bound.
+    assert!(
+        SystemClient::connect(format!("http://{addr}"))
+            .await
+            .is_err(),
+        "gRPC server must no longer be reachable after shutdown"
+    );
+
+    // (c) `DatabaseEngine::close()` -- forces the WAL checkpoint and closes
+    // the pool. A query issued afterward must fail, proving `close()` ran
+    // rather than merely being reachable dead code.
+    db.close().await;
+    let err = db
+        .list_filter_rules()
+        .await
+        .expect_err("a query after close() must fail because the pool is closed");
+    let message = err.to_string();
+    assert!(
+        message.to_lowercase().contains("closed") || message.to_lowercase().contains("pool"),
+        "expected a pool-closed style error after DatabaseEngine::close(), got: {message}"
+    );
+}
+
+#[tokio::test]
+async fn controller_trigger_is_observed_by_every_clone_of_the_signal() {
+    let event_bus = Arc::new(EventBus::new());
+    let (controller, shutdown_signal) = ShutdownController::new(event_bus);
+
+    let mut first = shutdown_signal.clone();
+    let mut second = shutdown_signal.clone();
+    assert!(!first.is_requested());
+    assert!(!second.is_requested());
+
+    controller.trigger();
+
+    tokio::time::timeout(JOIN_TIMEOUT, first.wait())
+        .await
+        .expect("first clone must observe the trigger");
+    tokio::time::timeout(JOIN_TIMEOUT, second.wait())
+        .await
+        .expect("second clone must observe the trigger");
+    assert!(second.is_requested());
+}