diff --git a/crates/omachatd/src/main.rs b/crates/omachatd/src/main.rs index 84af7f1..22186c9 100644 --- a/crates/omachatd/src/main.rs +++ b/crates/omachatd/src/main.rs @@ -1,5 +1,6 @@ use omachatd::{DaemonConfig, DaemonCore, EventHub, IpcServer, NostrService}; use std::{env, ffi::OsStr, fs, os::unix::fs::PermissionsExt, path::PathBuf, process::ExitCode}; +use tokio::signal::unix::{SignalKind, signal}; use tokio::sync::watch; #[tokio::main] @@ -28,6 +29,16 @@ async fn main() -> ExitCode { } async fn run(options: Options) -> Result<(), Box> { + // Install handlers before publishing the IPC socket. Once a client can + // observe readiness, both terminal interrupts and systemd's SIGTERM must + // reach the same draining/cleanup path. Registration failure is fatal. + let mut interrupt = signal(SignalKind::interrupt())?; + let mut terminate = signal(SignalKind::terminate())?; + let reload_signal = options + .config + .as_ref() + .map(|path| signal(SignalKind::hangup()).map(|signal| (path.clone(), signal))) + .transpose()?; let mut config = if let Some(path) = &options.config { DaemonConfig::load(path)? } else { @@ -94,26 +105,23 @@ async fn run(options: Options) -> Result<(), Box> { let server = IpcServer::bind(&options.socket, core.clone(), events)?; let (shutdown_tx, shutdown_rx) = watch::channel(false); - if let Some(config_path) = options.config.clone() { + let reload_task = reload_signal.map(|(config_path, mut signal)| { let reload_core = core.clone(); tokio::spawn(async move { - let Ok(mut signal) = - tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup()) - else { - return; - }; while signal.recv().await.is_some() { if let Err(error) = reload_core.reload(&config_path) { eprintln!("omachatd: rejected SIGHUP reload: {error}"); } } - }); - } + }) + }); let signal_shutdown = shutdown_tx.clone(); - tokio::spawn(async move { - if tokio::signal::ctrl_c().await.is_ok() { - let _ = signal_shutdown.send(true); + let signal_task = tokio::spawn(async move { + tokio::select! { + _ = interrupt.recv() => {}, + _ = terminate.recv() => {}, } + let _ = signal_shutdown.send(true); }); let panic_shutdown = shutdown_tx.clone(); let panic_core = core.clone(); @@ -122,6 +130,14 @@ async fn run(options: Options) -> Result<(), Box> { let _ = panic_shutdown.send(true); }); let server_result = server.run(shutdown_rx).await; + // Join cancellation before dismantling services so SIGHUP cannot race + // shutdown by applying a new configuration or restarting subscriptions. + if let Some(task) = reload_task { + task.abort(); + let _ = task.await; + } + signal_task.abort(); + let _ = signal_task.await; core.prepare_for_shutdown().await; if let Some(service) = geo_relays { service.shutdown().await; diff --git a/crates/omachatd/tests/process_lifecycle.rs b/crates/omachatd/tests/process_lifecycle.rs new file mode 100644 index 0000000..a0bd5ec --- /dev/null +++ b/crates/omachatd/tests/process_lifecycle.rs @@ -0,0 +1,317 @@ +//! Exercise the installed daemon entry point, not just DaemonCore methods. +use omachat_proto::ipc::{Command, Request, Response, ResponseOutcome, VERSION, encode_line}; +use serde_json::{Value, json}; +use std::{ + fs, + io::{BufRead, BufReader, Write}, + os::unix::net::UnixStream, + path::{Path, PathBuf}, + process::{Child, Command as Process, Stdio}, + thread, + time::{Duration, Instant}, +}; +use tempfile::{TempDir, tempdir}; + +const DEADLINE: Duration = Duration::from_secs(10); + +struct Fixture { + dir: TempDir, + config: PathBuf, + socket: PathBuf, +} + +impl Fixture { + fn new() -> Self { + let dir = tempdir().unwrap(); + let config = dir.path().join("config.json"); + let socket = dir.path().join("runtime/omachat.sock"); + let fixture = Self { + dir, + config, + socket, + }; + fixture.configure(json!({"storage_provider":"file", "joined_geohashes":["gcpvj"]})); + fixture + } + + fn configure(&self, config: Value) { + fs::write(&self.config, serde_json::to_vec(&config).unwrap()).unwrap(); + } + + fn start(&self) -> Daemon { + self.start_with_config(true) + } + + fn start_with_config(&self, with_config: bool) -> Daemon { + let log = self.dir.path().join("stderr.log"); + let mut command = Process::new(env!("CARGO_BIN_EXE_omachatd")); + if with_config { + command.arg("--config").arg(&self.config); + } + let child = command + .arg("--file-key") + .arg("--state") + .arg(self.dir.path().join("state")) + .arg("--socket") + .arg(&self.socket) + .env("XDG_RUNTIME_DIR", self.dir.path()) + .env("XDG_STATE_HOME", self.dir.path()) + .env("TOKIO_WORKER_THREADS", "2") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(fs::File::create(&log).unwrap()) + .spawn() + .unwrap(); + let mut daemon = Daemon { child, log }; + let start = Instant::now(); + loop { + assert!( + daemon.child.try_wait().unwrap().is_none(), + "daemon exited: {}", + daemon.log() + ); + if let Ok(stream) = UnixStream::connect(&self.socket) { + // The socket exists only after startup. Complete hello/status + // as the readiness barrier before sending a process signal. + let mut client = Client::new(stream); + client.request(Command::Status); + return daemon; + } + assert!( + start.elapsed() < DEADLINE, + "startup timed out: {}", + daemon.log() + ); + thread::sleep(Duration::from_millis(20)); + } + } + + fn client(&self) -> Client { + Client::new(UnixStream::connect(&self.socket).unwrap()) + } +} + +struct Daemon { + child: Child, + log: PathBuf, +} + +impl Daemon { + fn log(&self) -> String { + fs::read_to_string(&self.log).unwrap() + } + + fn signal(&mut self, signal: &str) { + assert!( + self.child.try_wait().unwrap().is_none(), + "daemon already exited: {}", + self.log() + ); + // This Child remains unreaped throughout signal delivery, so its PID + // cannot be reused for an unrelated process. + assert!( + Process::new("kill") + .arg(format!("-{signal}")) + .arg(self.child.id().to_string()) + .status() + .unwrap() + .success() + ); + } + + fn stop(&mut self, signal: &str, socket: &Path) { + self.signal(signal); + let start = Instant::now(); + loop { + if let Some(status) = self.child.try_wait().unwrap() { + assert!( + status.success(), + "unclean {signal} exit: {status}; {}", + self.log() + ); + assert!(!socket.exists(), "shutdown left an IPC socket"); + assert!( + !socket.with_extension("lock").exists(), + "shutdown left an instance lock path" + ); + return; + } + assert!( + start.elapsed() < DEADLINE, + "shutdown timed out: {}", + self.log() + ); + thread::sleep(Duration::from_millis(20)); + } + } + + fn wait_for_rejected_reload(&self, previous: usize) { + let start = Instant::now(); + while self.log().matches("rejected SIGHUP reload").count() <= previous { + assert!( + start.elapsed() < DEADLINE, + "reload not observed: {}", + self.log() + ); + thread::sleep(Duration::from_millis(20)); + } + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + // Reap on assertion failures too; never leave a background daemon. + if matches!(self.child.try_wait(), Ok(None)) { + let _ = self.child.kill(); + } + let _ = self.child.wait(); + } +} + +struct Client { + reader: BufReader, + next_id: u64, +} + +impl Client { + fn new(stream: UnixStream) -> Self { + stream.set_read_timeout(Some(DEADLINE)).unwrap(); + stream.set_write_timeout(Some(DEADLINE)).unwrap(); + let mut client = Self { + reader: BufReader::new(stream), + next_id: 0, + }; + client.request(Command::Hello { + minimum_version: VERSION, + maximum_version: VERSION, + }); + client + } + + fn request(&mut self, command: Command) -> Value { + self.next_id += 1; + let id = self.next_id.to_string(); + let request = Request { + version: VERSION, + id: id.clone(), + command, + }; + self.reader + .get_mut() + .write_all(&encode_line(&request).unwrap()) + .unwrap(); + let mut line = String::new(); + assert!( + self.reader.read_line(&mut line).unwrap() > 0, + "unexpected IPC EOF" + ); + let response: Response = serde_json::from_str(&line).unwrap(); + assert_eq!(response.version, VERSION); + assert_eq!(response.id, id); + match response.outcome { + ResponseOutcome::Ok { result } => result, + other => panic!("request failed: {other:?}"), + } + } + + fn assert_eof(&mut self) { + assert_eq!(self.reader.read_line(&mut String::new()).unwrap(), 0); + } +} + +#[test] +fn sigterm_drains_clients_and_restart_preserves_identity_and_sealed_outbox() { + let fixture = Fixture::new(); + let mut daemon = fixture.start(); + let mut first = fixture.client(); + let mut second = fixture.client(); + let before = first.request(Command::Status); + assert_eq!(second.request(Command::Status), before); + let sent = first.request(Command::Send { + conversation: "07e1870bb208e66b5189c2dc7b1c0018e26871920148706534dd74ee5a126ff4".into(), + text: "private process restart message".into(), + }); + assert_eq!(sent["delivery"], "queued"); + assert_eq!(second.request(Command::Status)["outbox_pending"], 1); + let outbox = fixture.dir.path().join("state/records/nostr-outbox-v1"); + let sealed = fs::read(&outbox).unwrap(); + let plaintext = b"private process restart message"; + assert!( + !sealed + .windows(plaintext.len()) + .any(|bytes| bytes == plaintext) + ); + daemon.stop("TERM", &fixture.socket); + first.assert_eof(); + second.assert_eof(); + + let mut restarted = fixture.start(); + let after = fixture.client().request(Command::Status); + for field in [ + "/fingerprint", + "/nostr_public_key", + "/peer_id", + "/account/account_id", + "/account/device_id", + ] { + let identity = before.pointer(field).and_then(Value::as_str).unwrap(); + assert!(!identity.is_empty(), "empty {field}"); + assert_eq!( + after.pointer(field).and_then(Value::as_str), + Some(identity), + "changed {field}" + ); + } + assert_eq!(after["outbox_pending"], 1); + assert_eq!(after["outbox_failed"], 0); + assert_eq!( + fs::read(outbox).unwrap(), + sealed, + "restart changed the queued ciphertext" + ); + restarted.stop("TERM", &fixture.socket); +} + +#[test] +fn sigint_without_a_config_file_uses_the_same_clean_shutdown_path() { + let fixture = Fixture::new(); + let mut daemon = fixture.start_with_config(false); + let mut client = fixture.client(); + daemon.stop("INT", &fixture.socket); + client.assert_eof(); +} + +#[test] +fn sighup_applies_valid_config_and_rejects_invalid_or_restart_only_changes() { + let fixture = Fixture::new(); + let mut daemon = fixture.start(); + let mut client = fixture.client(); + fixture.configure(json!({"storage_provider":"file", "joined_geohashes":["u4pruy"]})); + daemon.signal("HUP"); + let start = Instant::now(); + loop { + if client.request(Command::Status)["joined_geohashes"] == json!(["u4pruy"]) { + break; + } + assert!(start.elapsed() < DEADLINE, "valid reload was not applied"); + thread::sleep(Duration::from_millis(20)); + } + let before = client.request(Command::Status); + for (index, invalid) in [ + "{broken JSON".to_owned(), + json!({"joined_geohashes":["invalid!"]}).to_string(), + json!({"relays":["wss://relay.example"], "joined_geohashes":["gcpvj"]}).to_string(), + ] + .iter() + .enumerate() + { + fs::write(&fixture.config, invalid).unwrap(); + daemon.signal("HUP"); + // Wait for acknowledgement of the rejected signal, not a sleep + // that could inspect the old state before the reload even happened. + daemon.wait_for_rejected_reload(index); + assert_eq!(client.request(Command::Status), before); + assert_eq!(fixture.client().request(Command::Status), before); + } + daemon.stop("TERM", &fixture.socket); +} diff --git a/docs/installation.md b/docs/installation.md index c499976..58e57c2 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -42,6 +42,20 @@ Use `omachat-ctl status`, launch `omachat`, and stop with `systemctl --user disable --now omachatd.service`. No install or removal action edits Hyprland, Waybar, shell.json, user config, polkit policy, or linger state. +SIGTERM (the normal systemd stop signal) and SIGINT use the same graceful +shutdown path: stop accepting IPC, drain active requests within the bounded +IPC deadline, remove the socket, and shut down the owned services. Neither +signal erases identity or the sealed outbox. Signal handlers are registered +before the IPC socket becomes available; handler-registration errors fail +startup rather than leaving a running daemon without its shutdown handlers. + +When launched with `--config`, SIGHUP validates and applies supported hot +changes such as joined geohashes. Malformed/invalid configuration and relay +policy changes requiring restart are rejected with an error in the journal; +the prior active configuration stays in effect. The reload task is stopped +and joined before service teardown. This does not replace the target-host +login/keyring/linger lifecycle checks in OC-007. + The JSON daemon config may set `account_handle` (for example `"@tom"`) and `account_display_name`. Both are sealed into a root-signed local binding and survive restart if later omitted from configuration. Until the central registry diff --git a/packaging/man/omachatd.8 b/packaging/man/omachatd.8 index 4dabc67..f816193 100644 --- a/packaging/man/omachatd.8 +++ b/packaging/man/omachatd.8 @@ -11,7 +11,12 @@ socket. Only the Nostr transport is currently wired into the daemon; the mesh transport is not yet integrated. SIGHUP validates a replacement configuration before applying it. -SIGINT stops cleanly. +SIGINT and SIGTERM stop cleanly: stop accepting IPC, drain active requests +within a bounded deadline, remove the socket, and shut down owned services. +Neither signal erases identity or the sealed outbox. +Signal handlers are installed before the IPC socket becomes available. +With --config, malformed or invalid SIGHUP replacements and relay-policy +changes requiring restart leave the prior active configuration in effect. .SH ACCOUNT The JSON configuration may set account_handle and account_display_name. These values are sealed into a root-signed local account binding and survive restart.