diff --git a/CHANGELOG.md b/CHANGELOG.md index 04f2d4f6..ee1c8d27 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file. Each entry lists the date and the crate versions that were released. +## 2026-07-30 — mqdb-cli 0.8.23, mqdb-agent 0.8.16, mqdb-cluster 0.4.7 + +### Fixed + +- **Agent and cluster nodes shut down gracefully on SIGINT/SIGTERM, and wipe temp secret files.** Neither `mqdb agent start` nor `mqdb cluster start` installed signal handlers, so `Ctrl-C`/`docker stop`/`systemctl stop` killed the process immediately — skipping graceful shutdown, and leaving inline-secret temp files (passwd/ACL/SCRAM/JWT/QUIC content passed via `MQDB_*` env vars, written to `${TMPDIR}/mqdb-env-secrets-{pid}/` with `0o600`) on disk until reboot. Both commands now handle `Ctrl-C` (all platforms) and `SIGTERM` (unix): they signal the agent's existing shutdown channel, await graceful task drain, and remove the process-scoped temp secret directory. The agent's `run`/`start` also now race the broker against the shutdown channel, so `MqdbAgent::shutdown()` actually stops the broker instead of only signalling the auxiliary tasks. `ClusteredAgent` gains a `shutdown_handle()` so a signal task can trigger shutdown while `run(&mut self)` is executing. + ## 2026-07-27 — mqdb-cli 0.8.22, mqdb-core 0.7.7, mqdb-agent 0.8.15, mqdb-cluster 0.4.6 ### Changed diff --git a/Cargo.lock b/Cargo.lock index c03b6ac9..076fc7bb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1420,7 +1420,7 @@ dependencies = [ [[package]] name = "mqdb-agent" -version = "0.8.15" +version = "0.8.16" dependencies = [ "arc-swap", "argon2", @@ -1455,7 +1455,7 @@ dependencies = [ [[package]] name = "mqdb-cli" -version = "0.8.22" +version = "0.8.23" dependencies = [ "base64", "bebytes", @@ -1482,7 +1482,7 @@ dependencies = [ [[package]] name = "mqdb-cluster" -version = "0.4.6" +version = "0.4.7" dependencies = [ "arc-swap", "bebytes", diff --git a/README.md b/README.md index 63d26b11..257b0f18 100644 --- a/README.md +++ b/README.md @@ -955,6 +955,8 @@ docker run -d \ mqdb:latest agent start ``` +**Graceful shutdown.** `agent start` and `cluster start` handle `SIGINT` (Ctrl-C) and `SIGTERM` (`docker stop`, `systemctl stop`, Kubernetes pod termination): the node drains its tasks and shuts down cleanly instead of being killed mid-operation. Inline secrets (the `MQDB_*` variables above, without the `_FILE` suffix) are written to a process-scoped `${TMPDIR}/mqdb-env-secrets-{pid}/` directory with `0600` permissions and removed on shutdown. A `SIGKILL` (e.g. exceeding the orchestrator's termination grace period) cannot be intercepted, so leave enough grace time for a clean stop; any temp secret files left by a hard kill are reaped by the OS temp cleaner. + ### Authentication in CLI Commands When the broker requires authentication, every CLI command needs credentials. Pass them with `--user` and `--pass`, or set `MQDB_USER` and `MQDB_PASS` to avoid repeating them: diff --git a/crates/mqdb-agent/Cargo.toml b/crates/mqdb-agent/Cargo.toml index 91688731..88a13617 100644 --- a/crates/mqdb-agent/Cargo.toml +++ b/crates/mqdb-agent/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqdb-agent" -version = "0.8.15" +version = "0.8.16" edition.workspace = true license = "Apache-2.0" authors.workspace = true diff --git a/crates/mqdb-agent/src/agent/mod.rs b/crates/mqdb-agent/src/agent/mod.rs index e8903231..f34c829e 100644 --- a/crates/mqdb-agent/src/agent/mod.rs +++ b/crates/mqdb-agent/src/agent/mod.rs @@ -250,6 +250,11 @@ impl MqdbAgent { /// # Errors /// Returns an error if the broker fails to start or encounters a runtime error. pub async fn run(&self) -> Result<(), Box> { + // Subscribe before any startup await so a shutdown() racing the startup + // window (e.g. a SIGTERM during a slow bind/cert load) is buffered rather + // than lost by the broadcast channel. + let mut shutdown_rx = self.shutdown_tx.subscribe(); + let (mut config, service_username, service_password, needs_composite, admin_users) = self.build_broker_config().await?; @@ -302,7 +307,10 @@ impl MqdbAgent { }; let license_task = self.spawn_license_check_task(); - broker.run().await?; + tokio::select! { + result = broker.run() => result?, + _ = shutdown_rx.recv() => info!("MQDB Agent shutting down"), + } let _ = self.shutdown_tx.send(()); let _ = handler_task.await; @@ -393,9 +401,15 @@ impl MqdbAgent { let _ = ready_tx.send(true); }); + let mut shutdown_rx = shutdown_tx.subscribe(); let handle = tokio::spawn(async move { - if let Err(e) = broker.run().await { - tracing::error!("broker error: {e}"); + tokio::select! { + result = broker.run() => { + if let Err(e) = result { + tracing::error!("broker error: {e}"); + } + } + _ = shutdown_rx.recv() => info!("MQDB Agent shutting down"), } let _ = shutdown_tx.send(()); let _ = handler_task.await; diff --git a/crates/mqdb-agent/tests/admin_test.rs b/crates/mqdb-agent/tests/admin_test.rs index 3bec39db..0e1e6011 100644 --- a/crates/mqdb-agent/tests/admin_test.rs +++ b/crates/mqdb-agent/tests/admin_test.rs @@ -363,3 +363,24 @@ async fn test_list_query_complexity_limits_via_mqtt() { client.disconnect().await.unwrap(); agent_handle.abort(); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn agent_run_task_stops_on_shutdown() { + let port = next_test_port(); + let tmp = TempDir::new().unwrap(); + let db = Database::open(tmp.path()).await.unwrap(); + let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + let agent = MqdbAgent::new(db) + .with_bind_address(addr) + .with_anonymous(true); + let (handle, mut ready_rx, shutdown) = agent.start().await.unwrap(); + let _ = ready_rx.changed().await; + + let _ = shutdown.send(()); + + tokio::time::timeout(Duration::from_secs(5), handle) + .await + .expect("run task must return after shutdown instead of blocking on the broker") + .expect("run task must not panic"); + drop(tmp); +} diff --git a/crates/mqdb-cli/Cargo.toml b/crates/mqdb-cli/Cargo.toml index 3d84f00b..fa877df2 100644 --- a/crates/mqdb-cli/Cargo.toml +++ b/crates/mqdb-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqdb-cli" -version = "0.8.22" +version = "0.8.23" publish = false edition.workspace = true license = "AGPL-3.0-only" diff --git a/crates/mqdb-cli/src/commands/agent.rs b/crates/mqdb-cli/src/commands/agent.rs index 813c666c..bf1b67f7 100644 --- a/crates/mqdb-cli/src/commands/agent.rs +++ b/crates/mqdb-cli/src/commands/agent.rs @@ -207,7 +207,17 @@ pub(crate) async fn cmd_agent_start( } let agent = Arc::new(agent); - agent.run().await.map_err(|e| e.to_string())?; + let signal_agent = Arc::clone(&agent); + let signal_task = tokio::spawn(async move { + crate::commands::wait_for_shutdown_signal().await; + tracing::info!("shutdown signal received, stopping agent"); + signal_agent.shutdown(); + }); + + let run_result = agent.run().await; + signal_task.abort(); + crate::commands::env_secret::cleanup(); + run_result.map_err(|e| e.to_string())?; Ok(()) } diff --git a/crates/mqdb-cli/src/commands/cluster.rs b/crates/mqdb-cli/src/commands/cluster.rs index 69ea2f7f..76106305 100644 --- a/crates/mqdb-cli/src/commands/cluster.rs +++ b/crates/mqdb-cli/src/commands/cluster.rs @@ -214,7 +214,17 @@ pub(crate) async fn cmd_cluster_start( } let mut agent = ClusteredAgent::new(config)?; - Box::pin(agent.run()).await.map_err(|e| e.to_string())?; + let shutdown = agent.shutdown_handle(); + let signal_task = tokio::spawn(async move { + crate::commands::wait_for_shutdown_signal().await; + tracing::info!("shutdown signal received, stopping cluster node"); + let _ = shutdown.send(()); + }); + + let run_result = Box::pin(agent.run()).await; + signal_task.abort(); + crate::commands::env_secret::cleanup(); + run_result.map_err(|e| e.to_string())?; Ok(()) } diff --git a/crates/mqdb-cli/src/commands/env_secret.rs b/crates/mqdb-cli/src/commands/env_secret.rs index 28f56508..8b50dfad 100644 --- a/crates/mqdb-cli/src/commands/env_secret.rs +++ b/crates/mqdb-cli/src/commands/env_secret.rs @@ -11,6 +11,20 @@ fn secret_dir() -> PathBuf { std::env::temp_dir().join(format!("mqdb-env-secrets-{}", std::process::id())) } +/// Remove the process-scoped temp directory holding inline secret files written +/// from `MQDB_*` env vars, if it was created. Best-effort; called on shutdown. +pub(crate) fn cleanup() { + let dir = secret_dir(); + if dir.exists() + && let Err(e) = std::fs::remove_dir_all(&dir) + { + tracing::warn!( + "failed to remove temp secret directory {}: {e}; secret files may remain on disk", + dir.display() + ); + } +} + pub(crate) fn write_temp_file( name: &str, content: &str, @@ -83,3 +97,25 @@ pub(crate) fn resolve_federated_jwt_content( data.map(String::from) .or_else(|| file.as_ref().and_then(|p| std::fs::read_to_string(p).ok())) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cleanup_removes_process_secret_dir() { + let dir = secret_dir(); + let _ = std::fs::remove_dir_all(&dir); + + cleanup(); + assert!(!dir.exists(), "cleanup on an absent dir must be a no-op"); + + let path = write_temp_file("shutdown-cleanup-test", "value").unwrap(); + assert!(dir.exists(), "secret dir must exist after write_temp_file"); + assert!(path.exists()); + + cleanup(); + assert!(!dir.exists(), "cleanup must remove the secret dir"); + assert!(!path.exists()); + } +} diff --git a/crates/mqdb-cli/src/commands/mod.rs b/crates/mqdb-cli/src/commands/mod.rs index b9b7ded6..85ab8ed6 100644 --- a/crates/mqdb-cli/src/commands/mod.rs +++ b/crates/mqdb-cli/src/commands/mod.rs @@ -14,3 +14,28 @@ pub(crate) mod dev; #[cfg(feature = "cluster")] pub(crate) mod dev_bench; pub(crate) mod env_secret; + +/// Resolve once a shutdown signal arrives: Ctrl-C (SIGINT) on every platform, +/// plus SIGTERM on unix so `docker stop`/`systemctl stop` shut down gracefully. +pub(crate) async fn wait_for_shutdown_signal() { + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + match signal(SignalKind::terminate()) { + Ok(mut sigterm) => { + tokio::select! { + _ = tokio::signal::ctrl_c() => {} + _ = sigterm.recv() => {} + } + } + Err(e) => { + tracing::warn!("failed to install SIGTERM handler: {e}; using SIGINT only"); + let _ = tokio::signal::ctrl_c().await; + } + } + } + #[cfg(not(unix))] + { + let _ = tokio::signal::ctrl_c().await; + } +} diff --git a/crates/mqdb-cluster/Cargo.toml b/crates/mqdb-cluster/Cargo.toml index 0fcdf3dd..c7ebe7f6 100644 --- a/crates/mqdb-cluster/Cargo.toml +++ b/crates/mqdb-cluster/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqdb-cluster" -version = "0.4.6" +version = "0.4.7" publish = false edition.workspace = true license = "AGPL-3.0-only" diff --git a/crates/mqdb-cluster/src/cluster_agent/mod.rs b/crates/mqdb-cluster/src/cluster_agent/mod.rs index b750d6dd..46e9ac36 100644 --- a/crates/mqdb-cluster/src/cluster_agent/mod.rs +++ b/crates/mqdb-cluster/src/cluster_agent/mod.rs @@ -174,6 +174,13 @@ impl ClusteredAgent { let _ = self.shutdown_tx.send(()); } + /// A handle that can trigger graceful shutdown from another task while `run` + /// holds `&mut self`. + #[must_use] + pub fn shutdown_handle(&self) -> broadcast::Sender<()> { + self.shutdown_tx.clone() + } + #[must_use] pub fn node_id(&self) -> NodeId { self.node_id