diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 23f0f66..5aa2251 100644 --- a/crates/nunciod/src/grpc.rs +++ b/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, + db: Arc, + filter_engine: Arc, + secrets: Arc, + token: impl Into>, + shutdown: impl std::future::Future + Send + 'static, +) -> Result<(), GrpcServeError> { + let socket_addr = addr + .parse::() + .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, + db: Arc, + filter_engine: Arc, + secrets: Arc, + token: impl Into>, + shutdown: impl std::future::Future + 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, + db: Arc, + filter_engine: Arc, + secrets: Arc, + token: impl Into>, + overrides: MailEngineOverrides, + calendar_overrides: CalendarEngineOverrides, + contacts_overrides: ContactsEngineOverrides, + accounts_overrides: AccountsEngineOverrides, + shutdown: impl std::future::Future + Send + 'static, ) -> Result<(), GrpcServeError> { let token: Arc = 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) } diff --git a/crates/nunciod/src/lib.rs b/crates/nunciod/src/lib.rs index 7b3413b..9a10ad9 100644 --- a/crates/nunciod/src/lib.rs +++ b/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; diff --git a/crates/nunciod/src/lifecycle.rs b/crates/nunciod/src/lifecycle.rs new file mode 100644 index 0000000..dcef665 --- /dev/null +++ b/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, +} + +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, + event_bus: Arc, +} + +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) -> (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) { + 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); + } +} diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index c98aa72..d50f90b 100644 --- a/crates/nunciod/src/main.rs +++ b/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> { @@ -17,6 +26,17 @@ async fn main() -> Result<(), Box> { 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> { // 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> { _ => {} } } - }); + })) } 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> { 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> { // 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> { // 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(()) } diff --git a/crates/nunciod/tests/graceful_shutdown_test.rs b/crates/nunciod/tests/graceful_shutdown_test.rs new file mode 100644 index 0000000..40d784c --- /dev/null +++ b/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()); +}