From 10bb1f6889011372f44f24653808a348fdc4e6c6 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Thu, 8 Oct 2026 01:31:31 +0300 Subject: [PATCH] [main] stop cleanly on SIGTERM and SIGINT the binary had no signal handler, so every stop killed it mid write and the next start recovered like after a crash. now the first SIGTERM or SIGINT stops it through Hydrant::shutdown, the same path the ffi uses: the runtime is shut down, the persist thread saves the cursors one last time, and the database closes (its journal syncs on drop). it waits up to 30s and exits 0. a second signal exits right away. the signals get their own thread and runtime, because the main runtime is gone while the database closes and the second signal still has to land. --- AGENTS.md | 2 +- docs/getting-started.md | 2 + src/main.rs | 105 ++++++++++++++++++++++++++++++++-------- tests/cli.rs | 94 ++++++++++++++++++++++++++++++----- 4 files changed, 171 insertions(+), 32 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 7ba3a08..5a6c04f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -177,7 +177,7 @@ Hydrant uses multiple `fjall` keyspaces: - `nu tests/pds_host_status_transitions.nu` - Tests PDS host status transitions through offline, active, throttled, and active. - `nu tests/pds_tier_rule_status.nu` - Tests glob tier rules in PDS host status calculation. - `nu tests/harness_plc_fallback.nu` - Tests that the harness skips a dead PLC mirror, against a local mock. -- `cargo test --test cli` - Tests that the binary prints `--version`/`--help` and refuses any other argument instead of booting. +- `cargo test --test cli` - Tests that the binary prints `--version`/`--help` and refuses any other argument instead of booting, and that SIGTERM and SIGINT stop it cleanly so the next start opens the database. ## Rust code style diff --git a/docs/getting-started.md b/docs/getting-started.md index 88308e7..7a1d4a5 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -33,6 +33,8 @@ export HYDRANT_DATABASE_PATH=./hydrant.db see [configuration](configuration.md) for all available variables. if a `.env` file exists in the working directory it will be loaded automatically. +to stop it, send `SIGTERM` or `SIGINT` (ctrl-c). hydrant saves its firehose cursors, writes out and syncs the database journal and closes the database, waiting up to 30 seconds for that, then exits with 0. running backfills are dropped and start over from the queue on the next start. firehose messages still queued in memory are dropped too, and since the saved cursor can already be past some of them, they may not be read again (the same as after a crash). the next start also replays the journal into memory, since the storage engine doesn't flush its write buffers on close. a second signal exits right away without closing the database, which hydrant recovers from like a crash. + ## reverse proxying it is **highly recommended** to run hydrant behind a reverse proxy (like nginx or caddy) if you intend to expose the XRPC or event stream APIs to the public. hydrant's API includes several management endpoints that do not require or support authentication. **you MUST NOT expose these management endpoints to the public internet.** diff --git a/src/main.rs b/src/main.rs index 17b02c1..04de7ee 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,13 @@ use futures::FutureExt; use hydrant::config::Config; use hydrant::control::{ApiBind, ApiBinds, DatabasePoisoned, Hydrant}; +use miette::IntoDiagnostic; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::sync::Arc; +use std::time::Duration; +use tokio::signal::unix::{SignalKind, signal}; +use tokio::sync::Notify; +use tracing::{error, info}; mod allocator; @@ -76,8 +82,11 @@ impl AppConfig { } } -#[tokio::main] -async fn main() -> miette::Result<()> { +// long enough for a busy database to finish what it's writing, short enough that a +// supervisor's own stop timeout (systemd's is 90s) doesn't kill it first +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); + +fn main() -> miette::Result<()> { // config lives in env vars, so a stray argument (a `--version` probe, a typo) would // otherwise boot a full default instance without anyone noticing match parse_args(std::env::args_os().skip(1))? { @@ -107,22 +116,34 @@ async fn main() -> miette::Result<()> { .parse_lossy(directives.as_deref().unwrap_or(hydrant::DEFAULT_LOG_FILTER)); tracing_subscriber::fmt().with_env_filter(env_filter).init(); - let hydrant = Hydrant::new(cfg).await?; - - let api_fut = app - .api_binds - .map(|binds| hydrant.serve(binds).boxed()) - .unwrap_or_else(|| std::future::pending().boxed()); - let debug_fut = app - .enable_debug - .then(|| hydrant.serve_debug(app.debug_port).boxed()) - .unwrap_or_else(|| std::future::pending().boxed()); - - let result = tokio::select! { - r = hydrant.run()? => r, - r = api_fut => r, - r = debug_fut => r, - }; + let stop = watch_stop_signals()?; + let runtime = tokio::runtime::Runtime::new().into_diagnostic()?; + let hydrant = runtime.block_on(Hydrant::new(cfg))?; + + let result = runtime.block_on(async { + let api_fut = app + .api_binds + .map(|binds| hydrant.serve(binds).boxed()) + .unwrap_or_else(|| std::future::pending().boxed()); + let debug_fut = app + .enable_debug + .then(|| hydrant.serve_debug(app.debug_port).boxed()) + .unwrap_or_else(|| std::future::pending().boxed()); + tokio::select! { + r = hydrant.run()? => r, + r = api_fut => r, + r = debug_fut => r, + _ = stop.notified() => Ok(()), + } + }); + + // dropping the runtime would wait on blocking tasks that only `Hydrant::shutdown` ends + runtime.shutdown_background(); + info!("stopping, waiting up to {SHUTDOWN_TIMEOUT:?} for the database to close"); + let closed = hydrant.shutdown(SHUTDOWN_TIMEOUT); + if let (Err(_), Err(e)) = (&result, &closed) { + error!("{e}"); + } // a supervisor can tell this apart from other failures by the exit code if let Err(e) = &result && e.downcast_ref::().is_some() @@ -130,5 +151,51 @@ async fn main() -> miette::Result<()> { eprintln!("{e:?}"); std::process::exit(10); } - result + result?; + closed?; + info!("stopped"); + Ok(()) +} + +/// the first SIGINT or SIGTERM fires the returned notify, a second one exits on the spot. +/// it gets its own thread and runtime, since the main runtime is shut down while the +/// database closes and a signal then still has to reach someone. +fn watch_stop_signals() -> miette::Result> { + let stop = Arc::new(Notify::new()); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .into_diagnostic()?; + // registered before returning, so a signal right after startup isn't the default kill + let (mut int, mut term) = runtime + .block_on(async { + Ok::<_, std::io::Error>(( + signal(SignalKind::interrupt())?, + signal(SignalKind::terminate())?, + )) + }) + .into_diagnostic()?; + std::thread::Builder::new() + .name("signals".into()) + .spawn({ + let stop = stop.clone(); + move || { + runtime.block_on(async { + let mut next = async || { + tokio::select! { + _ = int.recv() => {} + _ = term.recv() => {} + } + }; + next().await; + info!("got a stop signal, stopping (send another to exit right away)"); + stop.notify_one(); + next().await; + eprintln!("got a second stop signal, exiting without closing the database"); + std::process::exit(1); + }) + } + }) + .into_diagnostic()?; + Ok(stop) } diff --git a/tests/cli.rs b/tests/cli.rs index 12986bb..c34d156 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -1,26 +1,32 @@ -use std::process::{Command, Output, Stdio}; +use std::path::Path; +use std::process::{Child, Command, Output, Stdio}; use std::time::{Duration, Instant}; -// config is env only, so anything on argv has to be refused before hydrant boots. -// if it does boot, it gets a throwaway database and no network, and we kill it. -fn hydrant(args: &[&str]) -> Output { - let tmp = tempfile::tempdir().unwrap(); - let mut child = Command::new(env!("CARGO_BIN_EXE_hydrant")) - .args(args) - .env("HYDRANT_DATABASE_PATH", tmp.path().join("db")) +/// a hydrant on a database in `dir` that never touches the network +fn offline_hydrant(dir: &Path) -> Command { + let mut cmd = Command::new(env!("CARGO_BIN_EXE_hydrant")); + cmd.env("HYDRANT_DATABASE_PATH", dir.join("db")) .env( "HYDRANT_API_BIND", - format!("unix:{}", tmp.path().join("api.sock").display()), + format!("unix:{}", dir.join("api.sock").display()), ) .env("HYDRANT_ENABLE_FIREHOSE", "false") .env("HYDRANT_ENABLE_CRAWLER", "false") .env("HYDRANT_ENABLE_BACKFILL", "false") .env("HYDRANT_RELAY_HOSTS", "") + .env("HYDRANT_SEED_HOSTS", "") + .env("HYDRANT_CRAWLER_URLS", "") .stdin(Stdio::null()) .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(); + .stderr(Stdio::piped()); + cmd +} + +// config is env only, so anything on argv has to be refused before hydrant boots. +// if it does boot, it gets a throwaway database and no network, and we kill it. +fn hydrant(args: &[&str]) -> Output { + let tmp = tempfile::tempdir().unwrap(); + let mut child = offline_hydrant(tmp.path()).args(args).spawn().unwrap(); let deadline = Instant::now() + Duration::from_secs(30); while child.try_wait().unwrap().is_none() { if Instant::now() > deadline { @@ -69,3 +75,67 @@ fn any_other_argument_is_an_error() { ); } } + +/// waits for `child` to exit on its own, killing it after `timeout` +fn wait_for_exit(mut child: Child, timeout: Duration) -> Option { + let deadline = Instant::now() + timeout; + while child.try_wait().unwrap().is_none() { + if Instant::now() > deadline { + child.kill().unwrap(); + return None; + } + std::thread::sleep(Duration::from_millis(50)); + } + Some(child.wait_with_output().unwrap()) +} + +fn send_signal(child: &Child, signal: &str) { + let status = Command::new("kill") + .args([signal, &child.id().to_string()]) + .status() + .unwrap(); + assert!(status.success(), "kill {signal} failed"); +} + +/// starts hydrant on `dir` and returns once its api socket is up, so it's fully running +fn start_and_wait_for_api(dir: &Path) -> Child { + let socket = dir.join("api.sock"); + let _ = std::fs::remove_file(&socket); + let mut child = offline_hydrant(dir).spawn().unwrap(); + let deadline = Instant::now() + Duration::from_secs(60); + while !socket.exists() { + if let Some(status) = child.try_wait().unwrap() { + let out = child.wait_with_output().unwrap(); + panic!( + "hydrant exited with {status} before serving:\n{}", + String::from_utf8_lossy(&out.stderr) + ); + } + assert!( + Instant::now() < deadline, + "hydrant never bound its api socket" + ); + std::thread::sleep(Duration::from_millis(50)); + } + child +} + +#[test] +fn sigterm_and_sigint_stop_cleanly_and_the_database_opens_again() { + let tmp = tempfile::tempdir().unwrap(); + // the second and third start only get past opening the database if the one before let go + for signal in ["-TERM", "-INT", "-TERM"] { + let child = start_and_wait_for_api(tmp.path()); + send_signal(&child, signal); + let out = wait_for_exit(child, Duration::from_secs(45)) + .unwrap_or_else(|| panic!("hydrant didn't stop within 45s of {signal}")); + let stdout = String::from_utf8_lossy(&out.stdout); + let stderr = String::from_utf8_lossy(&out.stderr); + assert!( + out.status.success(), + "{signal}: {:?}\n{stdout}{stderr}", + out.status + ); + assert!(stdout.contains("stopped"), "{signal}: {stdout}{stderr}"); + } +} -- 2.54.0