diff --git a/crates/Cargo.lock b/crates/Cargo.lock index 57f4acbc0..13c42b7fb 100644 --- a/crates/Cargo.lock +++ b/crates/Cargo.lock @@ -2651,6 +2651,7 @@ dependencies = [ "khive-gate-rego", "khive-query", "khive-request", + "khive-runtime", "khive-score", "khive-storage", "khive-types", diff --git a/crates/khive-runtime/Cargo.toml b/crates/khive-runtime/Cargo.toml index 470540076..98666e1c3 100644 --- a/crates/khive-runtime/Cargo.toml +++ b/crates/khive-runtime/Cargo.toml @@ -49,6 +49,11 @@ sha2 = { workspace = true } [dev-dependencies] khive-db = { version = "0.8.0", path = "../khive-db", features = ["test-support"] } khive-request = { version = "0.8.0", path = "../khive-request" } +# Self-dependency to enable `fault-injection` for integration tests under +# `tests/`: those link the lib as an external crate, so `#[cfg(test)]` items +# gated `any(test, feature = "fault-injection")` (e.g. `run_daemon_in_process_test`) +# are otherwise unreachable from there. +khive-runtime = { path = ".", features = ["fault-injection"] } khive-gate-rego = { version = "0.8.0", path = "../khive-gate-rego" } khive-storage = { version = "0.8.0", path = "../khive-storage", features = ["test-support"] } tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/khive-runtime/src/daemon.rs b/crates/khive-runtime/src/daemon.rs index ebfee838a..a6918c57a 100644 --- a/crates/khive-runtime/src/daemon.rs +++ b/crates/khive-runtime/src/daemon.rs @@ -1915,9 +1915,44 @@ async fn run_daemon_with_boot_guard_inner( // same process, which would self-deadlock on `flock`. let _startup_lock = boot_guard; - if !cleanup_stale_daemon(&sock, &pid_file, allow_same_process_incumbent).await { - tracing::info!("a live process already owns the daemon PID file; exiting"); - return Ok(()); + // #1874: a second daemon must refuse loudly (non-zero exit, pid named) rather + // than exit `Ok(())` — a silent success here is what let two detached daemons + // coexist on one store with neither side nor its caller ever noticing. Both + // live outcomes below refuse; only the message differs. + match cleanup_stale_daemon( + &sock, + &pid_file, + allow_same_process_incumbent, + dispatcher.config_id(), + ) + .await + { + Incumbent::Serving(incumbent_pid) => { + tracing::error!( + pid = incumbent_pid, + socket = ?sock, + "refusing to start: a khived instance is already serving this socket" + ); + anyhow::bail!( + "refusing to start: khived is already running as pid {incumbent_pid}, \ + serving socket {}. Stop that instance first if you intend to replace it.", + sock.display() + ); + } + Incumbent::Live(incumbent_pid) => { + tracing::error!( + pid = incumbent_pid, + socket = ?sock, + "refusing to start: a live process owns the PID file but no khived answered" + ); + anyhow::bail!( + "refusing to start: pid {incumbent_pid} owns the daemon PID file and is alive, \ + but nothing answered the khived protocol on {}. It may be draining. Nothing \ + was removed; stop that process first if you intend to replace it.", + sock.display() + ); + } + Incumbent::Stale => {} } let listener = UnixListener::bind(&sock)?; @@ -1951,8 +1986,13 @@ async fn run_daemon_with_boot_guard_inner( drop(listener); let _ = std::fs::remove_file(&sock); } - if pid_file_names_a_reachable_daemon(&pid_file, &sock, allow_same_process_incumbent) - .await + if pid_file_names_a_reachable_daemon( + &pid_file, + &sock, + allow_same_process_incumbent, + dispatcher.config_id(), + ) + .await { tracing::info!( "a replacement khived already claimed the pid/socket rendezvous; exiting" @@ -2248,20 +2288,117 @@ fn pid_can_name_incumbent(pid: u32, current_pid: u32, allow_same_process_incumbe allow_same_process_incumbent || pid != current_pid } +/// Bounded timeout for the protocol-identity probe used by duplicate-daemon +/// detection. Short enough that a hung or foreign listener does not stall +/// startup; long enough for a live khived under normal load to answer a +/// `probe_only` frame. +#[cfg(unix)] +const DUPLICATE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(500); + +/// Whether the listener at `sock` actually speaks the khived wire protocol +/// **as the same khived this process would defer to** — identified by +/// `expected_config_id`. +/// +/// A live PID plus an accepting Unix socket is not proof of khived: any +/// unrelated process that happens to have bound the same path also answers +/// `connect()`. Nor is any well-formed [`DaemonResponseFrame`] proof: a +/// `config_mismatch`/`version_mismatch` response, a `metrics_only` snapshot +/// response, or a legacy pre-probe daemon that falls through to normal +/// dispatch on the empty `ops` string, all deserialize cleanly without being +/// the unambiguous "yes, alive and identity-matching" answer this check +/// needs — the daemon's `metrics_only` arm in particular echoes the same +/// `ok=true, result=None, error=None`, all-mismatch-flags-false, matching +/// protocol version and `served_config_id` shape as the probe-ack arm, and +/// is distinguished only by carrying `metrics: Some(...)`. This sends a +/// bounded `probe_only` frame (the same identity probe the client-side +/// recovery path uses, `crates/khive-mcp/src/daemon.rs::probe_daemon_identity`) +/// carrying this process's own `config_id`, and requires the exact +/// probe-branch shape back: `ok=true`, `result=None`, `error=None`, +/// `metrics=None`, `request_id=None` (this probe frame never sets one), no +/// mismatch flags, matching protocol version, and matching +/// `served_config_id` — mirroring the client probe's `is_probe_ack` check so +/// both sides of the protocol agree on what "alive" means. Connect, write, +/// and read are all inside the one bounded timeout: `UnixStream::connect` +/// itself awaits write readiness, so a listener with a saturated accept +/// backlog could otherwise hold this call open past the advertised bound. +/// A connect that succeeds but never answers, times out, or answers with +/// non-protocol bytes, a mismatched identity, a `metrics_only` snapshot, or +/// any other non-probe-shaped response is not treated as the same khived +/// and falls through to the stale-socket recovery path instead. +#[cfg(unix)] +async fn socket_speaks_khived_protocol(sock: &std::path::Path, expected_config_id: &str) -> bool { + let probe = DaemonRequestFrame { + probe_only: true, + protocol_version: PROTOCOL_VERSION, + config_id: expected_config_id.to_string(), + ..Default::default() + }; + let Ok(payload) = serde_json::to_vec(&probe) else { + return false; + }; + let response = tokio::time::timeout(DUPLICATE_PROBE_TIMEOUT, async { + let mut stream = UnixStream::connect(sock).await.ok()?; + write_frame(&mut stream, &payload).await.ok()?; + let raw = read_frame(&mut stream).await.ok()?; + serde_json::from_slice::(&raw).ok() + }) + .await + .ok() + .flatten(); + + let Some(resp) = response else { + return false; + }; + let is_probe_ack = resp.ok + && resp.result.is_none() + && resp.error.is_none() + && resp.metrics.is_none() + && resp.request_id.is_none(); + is_probe_ack + && !resp.version_mismatch + && !resp.namespace_mismatch + && !resp.config_mismatch + && resp.daemon_protocol_version == PROTOCOL_VERSION + && resp.served_config_id.as_deref() == Some(expected_config_id) +} + +/// What owns the daemon PID file, from the point of view of a process that wants +/// to start. A live owner is never cleaned up: a draining incumbent closes its +/// listener before it releases writers, so an unanswered socket is ambiguous and +/// deleting its PID file is how two daemons end up on one store. +#[cfg(unix)] +enum Incumbent { + /// A live process that answered the khived protocol on the socket. + Serving(u32), + /// A live process owns the PID file and nothing answered. Nothing removed. + Live(u32), + /// Nothing live owns the store; the socket and PID file were removed. + Stale, +} + +/// Check whether `pid_file`/`sock` already name a live daemon and, if not, +/// remove the stale rendezvous files so the caller may bind fresh. +/// +/// Both live outcomes mean the caller must not bind and must refuse to start +/// rather than silently deferring (#1874: a quiet `Ok(())` here is exactly what +/// let two detached daemons coexist on one store). Only `Stale` clears the +/// rendezvous and lets the caller proceed. #[cfg(unix)] async fn cleanup_stale_daemon( sock: &std::path::Path, pid_file: &std::path::Path, allow_same_process_incumbent: bool, -) -> bool { + expected_config_id: &str, +) -> Incumbent { if let Ok(pid_str) = std::fs::read_to_string(pid_file) { if let Ok(pid) = pid_str.trim().parse::() { if pid_can_name_incumbent(pid, std::process::id(), allow_same_process_incumbent) && is_process_running(pid) { - // A draining incumbent closes its listener before releasing writers. - // Ambiguous live PIDs are left for client recovery to classify. - return false; + if sock.exists() && socket_speaks_khived_protocol(sock, expected_config_id).await { + return Incumbent::Serving(pid); + } + return Incumbent::Live(pid); } } } @@ -2275,7 +2412,7 @@ async fn cleanup_stale_daemon( tracing::warn!(error = %e, path = ?pid_file, "failed to remove stale PID file"); } } - true + Incumbent::Stale } /// Create `pid_file` exclusively (`O_EXCL`) and write this process's PID. @@ -2309,6 +2446,7 @@ async fn pid_file_names_a_reachable_daemon( pid_file: &std::path::Path, sock: &std::path::Path, allow_same_process_incumbent: bool, + expected_config_id: &str, ) -> bool { let Ok(pid_str) = std::fs::read_to_string(pid_file) else { return false; @@ -2319,7 +2457,7 @@ async fn pid_file_names_a_reachable_daemon( pid_can_name_incumbent(pid, std::process::id(), allow_same_process_incumbent) && is_process_running(pid) && sock.exists() - && UnixStream::connect(sock).await.is_ok() + && socket_speaks_khived_protocol(sock, expected_config_id).await } #[cfg(unix)] @@ -2793,7 +2931,10 @@ mod tests { // Harness eligibility makes our own stable PID an incumbent; // ordinary same-PID rejection is covered separately above. assert!( - !cleanup_stale_daemon(&sock, &pid_file, true).await, + matches!( + cleanup_stale_daemon(&sock, &pid_file, true, "probe-test").await, + Incumbent::Live(_) | Incumbent::Serving(_) + ), "live incumbent must retain ownership with socket_exists={socket_exists}" ); assert_eq!( @@ -3482,6 +3623,218 @@ mod tests { serde_json::from_slice(&raw).expect("decode response frame") } + /// #2230 review (Medium): duplicate-daemon detection must not treat any + /// accepting Unix listener as khived. A real khived (`handle_conn` behind + /// a bound socket) must still be recognized by the protocol probe. + #[tokio::test] + async fn socket_speaks_khived_protocol_accepts_a_real_khived() { + let dir = tempfile::tempdir().expect("tempdir"); + let sock_path = dir.path().join("real.sock"); + let listener = UnixListener::bind(&sock_path).expect("bind real listener"); + let dispatcher = MockDispatch { + namespace: "local".to_string(), + config_id: "probe-test".to_string(), + dispatch_calls: Arc::new(std::sync::atomic::AtomicUsize::new(0)), + pool: None, + dispatch_err: None, + }; + let accept_task = tokio::spawn(async move { + if let Ok((stream, _)) = listener.accept().await { + handle_conn(stream, dispatcher).await; + } + }); + + assert!( + socket_speaks_khived_protocol(&sock_path, "probe-test").await, + "a real khived answering the probe_only frame with a matching config_id must be recognized" + ); + + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), accept_task).await; + } + + /// #2230 review (Medium): a listener that accepts a connection but never + /// answers with a well-formed daemon response — e.g. an unrelated process + /// that happens to have bound the same socket path — must not be + /// classified as khived, and the probe must not hang past its own + /// bounded timeout. + #[tokio::test] + async fn socket_speaks_khived_protocol_rejects_a_non_protocol_listener() { + let dir = tempfile::tempdir().expect("tempdir"); + let sock_path = dir.path().join("fake.sock"); + let listener = UnixListener::bind(&sock_path).expect("bind fake listener"); + let held = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let held_for_task = held.clone(); + let accept_task = tokio::spawn(async move { + if let Ok((stream, _)) = listener.accept().await { + // Accept but never write anything back — the connection stays + // open exactly like a foreign process that speaks a different + // (or no) protocol on this socket. + held_for_task.lock().await.push(stream); + } + }); + + let before = tokio::time::Instant::now(); + let speaks = socket_speaks_khived_protocol(&sock_path, "probe-test").await; + let elapsed = before.elapsed(); + + assert!( + !speaks, + "a listener that accepts but never answers the probe frame must not be treated as khived" + ); + assert!( + elapsed < std::time::Duration::from_secs(2), + "the probe must be bounded by its own timeout, not hang indefinitely; took {elapsed:?}" + ); + + accept_task.abort(); + let _ = accept_task.await; + drop(held); + } + + /// Regression (#2230): a well-formed [`DaemonResponseFrame`] + /// that is not the unambiguous probe-ack sentinel — e.g. one reporting a + /// `config_mismatch` for a *different* config_id, exactly what a live + /// khived serving another store would send back — must not be treated as + /// the same live, identity-matching duplicate. Before this fix, any + /// frame that merely deserialized was accepted, so this response would + /// have been misclassified as "alive" and refused a legitimate boot. + #[tokio::test] + async fn socket_speaks_khived_protocol_rejects_a_non_ack_or_mismatched_response() { + let dir = tempfile::tempdir().expect("tempdir"); + let sock_path = dir.path().join("mismatched.sock"); + let listener = UnixListener::bind(&sock_path).expect("bind fake listener"); + let accept_task = tokio::spawn(async move { + if let Ok((mut stream, _)) = listener.accept().await { + let _raw = read_frame(&mut stream).await.expect("read probe frame"); + let resp = DaemonResponseFrame { + ok: false, + result: None, + error: None, + namespace_mismatch: false, + config_mismatch: true, + served_config_id: Some("someone-elses-config".to_string()), + version_mismatch: false, + daemon_protocol_version: PROTOCOL_VERSION, + error_detail: None, + metrics: None, + request_id: None, + }; + let payload = serde_json::to_vec(&resp).expect("encode response"); + write_frame(&mut stream, &payload) + .await + .expect("write response"); + } + }); + + let speaks = socket_speaks_khived_protocol(&sock_path, "expected-config").await; + assert!( + !speaks, + "a well-formed but non-ack / identity-mismatched response must not be treated as \ + the same live khived" + ); + + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), accept_task).await; + } + + /// Regression (#2230): the daemon's `metrics_only` arm answers with + /// `ok=true, result=None, error=None`, every mismatch flag false, the + /// current protocol version, and a matching `served_config_id` — the + /// exact same shape the probe-ack arm produces, differing only in + /// carrying `metrics: Some(...)`. A well-formed metrics snapshot + /// response must not be misread as a probe acknowledgement; otherwise a + /// client whose only interaction with the socket happened to be a + /// metrics poll would be classified as the same live, identity-matching + /// khived. This response carries `request_id: None`, so it isolates the + /// `metrics.is_none()` conjunct — see the sibling test below for the + /// `request_id.is_none()` conjunct. + #[tokio::test] + async fn socket_speaks_khived_protocol_rejects_a_metrics_only_response() { + let dir = tempfile::tempdir().expect("tempdir"); + let sock_path = dir.path().join("metrics-only.sock"); + let listener = UnixListener::bind(&sock_path).expect("bind fake listener"); + let accept_task = tokio::spawn(async move { + if let Ok((mut stream, _)) = listener.accept().await { + let _raw = read_frame(&mut stream).await.expect("read probe frame"); + let resp = DaemonResponseFrame { + ok: true, + result: None, + error: None, + namespace_mismatch: false, + config_mismatch: false, + served_config_id: Some("expected-config".to_string()), + version_mismatch: false, + daemon_protocol_version: PROTOCOL_VERSION, + error_detail: None, + metrics: Some(MetricsSnapshot::default()), + request_id: None, + }; + let payload = serde_json::to_vec(&resp).expect("encode response"); + write_frame(&mut stream, &payload) + .await + .expect("write response"); + } + }); + + let speaks = socket_speaks_khived_protocol(&sock_path, "expected-config").await; + assert!( + !speaks, + "an otherwise-matching response carrying a metrics snapshot must not be treated as \ + a probe acknowledgement" + ); + + tokio::time::timeout(std::time::Duration::from_secs(2), accept_task) + .await + .expect("fake listener accept task timed out") + .expect("fake listener accept task panicked"); + } + + /// Regression (#2230): sibling of the metrics-only test above, isolating + /// the `request_id.is_none()` conjunct. A response with `metrics: None` + /// but an echoed `request_id: Some(_)` is otherwise identical to a probe + /// acknowledgement and must not be misread as one — a probe frame never + /// sets `request_id`, so an echo of one is proof the peer answered a + /// different, non-probe request. + #[tokio::test] + async fn socket_speaks_khived_protocol_rejects_a_response_with_request_id() { + let dir = tempfile::tempdir().expect("tempdir"); + let sock_path = dir.path().join("request-id.sock"); + let listener = UnixListener::bind(&sock_path).expect("bind fake listener"); + let accept_task = tokio::spawn(async move { + if let Ok((mut stream, _)) = listener.accept().await { + let _raw = read_frame(&mut stream).await.expect("read probe frame"); + let resp = DaemonResponseFrame { + ok: true, + result: None, + error: None, + namespace_mismatch: false, + config_mismatch: false, + served_config_id: Some("expected-config".to_string()), + version_mismatch: false, + daemon_protocol_version: PROTOCOL_VERSION, + error_detail: None, + metrics: None, + request_id: Some(42), + }; + let payload = serde_json::to_vec(&resp).expect("encode response"); + write_frame(&mut stream, &payload) + .await + .expect("write response"); + } + }); + + let speaks = socket_speaks_khived_protocol(&sock_path, "expected-config").await; + assert!( + !speaks, + "an otherwise-matching response carrying an echoed request_id must not be treated \ + as a probe acknowledgement" + ); + + tokio::time::timeout(std::time::Duration::from_secs(2), accept_task) + .await + .expect("fake listener accept task timed out") + .expect("fake listener accept task panicked"); + } + #[derive(Clone)] struct DetailedDispatch { calls: Arc, diff --git a/crates/khive-runtime/tests/duplicate_daemon_refusal.rs b/crates/khive-runtime/tests/duplicate_daemon_refusal.rs new file mode 100644 index 000000000..e52e5518d --- /dev/null +++ b/crates/khive-runtime/tests/duplicate_daemon_refusal.rs @@ -0,0 +1,103 @@ +//! Regression test for #1874 — a second `khived` boot attempt against a +//! store already served by a live daemon must refuse loudly (an `Err` naming +//! the incumbent pid) rather than silently exit `Ok(())`. +//! +//! Before the fix, `run_daemon_with_boot_guard_inner` treated "a live, +//! responsive incumbent already owns this socket" as ordinary success: it +//! logged at `info` and returned `Ok(())`. A human (or a supervisor) +//! starting a second daemon against a store already served by one had no +//! signal that anything was wrong — both processes ran, each holding its own +//! WAL connection and read marks, exactly the two-daemon state #1874 +//! describes. +//! +//! Unix-only: daemon boot/socket/pid-file machinery is `#[cfg(unix)]` only. + +#![cfg(unix)] + +use async_trait::async_trait; +use khive_runtime::daemon::run_daemon_in_process_test; +use khive_runtime::{DaemonDispatch, RequestIdentity}; +use serial_test::serial; + +#[derive(Clone)] +struct NeverDispatch; + +#[async_trait] +impl DaemonDispatch for NeverDispatch { + fn plan(&self, ops: &str) -> String { + khive_request::plan_request(ops, &Default::default()).to_string() + } + + async fn dispatch( + &self, + _ops: String, + _presentation: Option, + _presentation_per_op: Option>>, + _format: Option, + _format_per_op: Option>>, + _from_wire: bool, + _identity: Option, + ) -> Result { + Err("dispatch not exercised by this test".to_string()) + } + + async fn warm_all(&self) {} + + fn namespace(&self) -> &str { + "test" + } + + fn config_id(&self) -> &str { + "test-config" + } +} + +/// Second boot attempt must fail loudly (`Err`) and name the incumbent's pid +/// — never silently succeed while a live daemon already owns the socket. +#[tokio::test] +#[serial] +async fn second_daemon_boot_refuses_loudly_while_first_is_live() { + let dir = tempfile::tempdir().expect("tempdir"); + std::env::set_var("KHIVE_SOCKET", dir.path().join("khived.sock")); + std::env::set_var("KHIVE_PID", dir.path().join("khived.pid")); + std::env::set_var("KHIVE_LOCK", dir.path().join("khived.recovery.lock")); + + let first = tokio::spawn(run_daemon_in_process_test(NeverDispatch)); + + // Poll for the socket to appear rather than a fixed sleep: bind happens + // asynchronously inside the spawned task. + let sock = dir.path().join("khived.sock"); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); + while !sock.exists() { + if tokio::time::Instant::now() >= deadline { + panic!("first daemon never bound its socket within the deadline"); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + + let second = run_daemon_in_process_test(NeverDispatch).await; + + let err = second.expect_err( + "a second boot attempt while the first is live and responsive must return Err, \ + not silently succeed", + ); + let message = format!("{err:#}"); + assert!( + message.contains(&std::process::id().to_string()), + "the refusal must name the incumbent's pid so an operator can act on it, got: {message}" + ); + assert!( + message.to_lowercase().contains("already running"), + "the refusal must say plainly that a daemon is already running, got: {message}" + ); + + // The in-process harness serves until aborted (no SIGTERM channel in a + // test process) — the same teardown contract used elsewhere for this + // entrypoint (see khive-mcp's `InProcessDaemonHandle::stop`). + first.abort(); + let _ = first.await; + + std::env::remove_var("KHIVE_SOCKET"); + std::env::remove_var("KHIVE_PID"); + std::env::remove_var("KHIVE_LOCK"); +}