diff --git a/hyperdb-api/tests/process_tests.rs b/hyperdb-api/tests/process_tests.rs index 1f95d06..272d350 100644 --- a/hyperdb-api/tests/process_tests.rs +++ b/hyperdb-api/tests/process_tests.rs @@ -28,7 +28,16 @@ fn callback_connection_shutdowns_hyperd_after_parent_kill() { let pid = hyper .pid() .expect("child must report HyperProcess public PID"); - fs::write(pid_file, pid.to_string()).expect("child must report Hyper PID to parent"); + // Publish the PID atomically — stage a sibling temp file, then rename + // it into place. A plain `fs::write` is `File::create` (truncate) then + // `write_all`, so a poll can observe not just an empty file but a + // truncated-yet-parseable prefix (`"123"` of `"12345"`), which would + // hand the parent a plausible but wrong PID with nothing to catch it. + let pid_file = std::path::PathBuf::from(pid_file); + let staged_pid_file = pid_file.with_extension("tmp"); + fs::write(&staged_pid_file, pid.to_string()).expect("child must stage its Hyper PID"); + fs::rename(&staged_pid_file, &pid_file) + .expect("child must publish its Hyper PID to parent"); loop { thread::park(); @@ -78,12 +87,13 @@ fn wait_for_reported_pid(pid_file: &std::path::Path, timeout: Duration) -> Resul Ok(contents) => match contents.trim().parse::() { Ok(pid) => return Ok(pid), Err(_) => { - // The child reports its PID with `fs::write`, which is - // `File::create` (truncates/creates) then `write_all` — - // not atomic. A poll can land in the window where the - // file exists but is still empty or partially written; - // treat that the same as "not created yet" rather than a - // fatal parse error, and keep polling to the deadline. + // The child publishes its PID with a rename, so a partial + // read should not be observable at all. Tolerate one + // anyway instead of panicking the whole test on it: this + // retry is what keeps a writer that ever drops back to a + // plain, non-atomic `fs::write` from resurrecting the + // flake. Treat it as "not created yet" and keep polling to + // the deadline. last_unparseable = Some(contents); } }, @@ -103,11 +113,14 @@ fn wait_for_reported_pid(pid_file: &std::path::Path, timeout: Duration) -> Resul } } -/// The parent's `fs::write` of the PID report is `File::create` (truncates or -/// creates, leaving an empty file momentarily) followed by `write_all` — not -/// atomic. A poll landing in that window used to make `"".parse::()` -/// panic the whole test; it must instead be treated as "not ready yet" and -/// retried to the deadline. +/// The *child* writes the PID report; the parent is the one polling for it. +/// The child now publishes atomically (temp file plus `fs::rename`), but a +/// plain `fs::write` is `File::create` (truncates or creates, leaving an empty +/// file momentarily) followed by `write_all` — not atomic. A poll landing in +/// that window used to make `"".parse::()` panic the whole test; it must +/// instead be treated as "not ready yet" and retried to the deadline, so a +/// writer that ever drops back to a non-atomic `fs::write` cannot resurrect +/// the flake. #[test] fn wait_for_reported_pid_retries_past_a_torn_write() { let temp_dir = tempfile::tempdir().expect("create temp dir for torn-write simulation"); @@ -115,11 +128,11 @@ fn wait_for_reported_pid_retries_past_a_torn_write() { let writer_pid_file = pid_file.clone(); let writer = thread::spawn(move || { - // Reproduces the exact non-atomic sequence: create (truncate) first, - // leaving the file present-but-empty for a deliberate window, then - // write the real content — same as the production `fs::write` call - // this helper polls for, just with the empty window stretched out - // long enough that a fast poller is guaranteed to observe it. + // Reproduces the non-atomic sequence a plain `fs::write` performs: + // create (truncate) first, leaving the file present-but-empty for a + // deliberate window, then write the real content — just with the empty + // window stretched out long enough that a fast poller is guaranteed to + // observe it. fs::File::create(&writer_pid_file).expect("create pid file (torn-write simulation)"); thread::sleep(Duration::from_millis(150)); fs::write(&writer_pid_file, "4242").expect("finish torn-write simulation"); diff --git a/hyperdb-mcp/CHANGELOG.md b/hyperdb-mcp/CHANGELOG.md index 11c31ce..f1ebd73 100644 --- a/hyperdb-mcp/CHANGELOG.md +++ b/hyperdb-mcp/CHANGELOG.md @@ -340,6 +340,78 @@ and this project adheres to [Semantic Versioning](https://semver.org/). than losing scratch tables. Load with `persist: true` if data must survive a panicking tool call. Fixes [issue #266](https://github.com/tableau/hyper-api-rust/issues/266). +- **`discover()` rejected a live daemon over an unrecognized `identity` + block, defeating forward compatibility.** The client fast path deserialized + the whole `DaemonRecord` (added in 0.7.3) — including the nested `identity` + object — and discarded the enrichment immediately after + (`record.info().clone()`), so the added strictness bought nothing while + costing compatibility across daemon/client version skew: a retyped + `identity`, a reshaped `executable_path`, or an `identity` missing + `executable_path` each made a well-formed, live daemon's `daemon.json` fail + to parse, and `discover()` returned `None` for a healthy process. + `wait_for_daemon` (`spawn.rs`) then timed out and silently fell back to a + private `hyperd`, `status_degraded` (`server.rs`) reported no daemon, and + `Engine::is_running` returned false. `discover()` now reads the file once + and parses those bytes twice: first as the strict `DaemonRecord`, which + remains the primary path, and — only when that fails — as a bare + `DaemonInfo`, which ignores unknown fields and never trips over a nested + object this fast path does not need. A record that only the tolerant + fallback could read is deliberately left on disk when its health check + fails, instead of being stale-cleaned: such a record is exactly what a + *newer* daemon looks like to an older client, the daemon rewrites + `daemon.json` only on `hyperd` restart, and deleting it over one failed + 300 ms PING would hide a live daemon from every client on the machine, not + just this one. A strictly parsed record on a dead port is still cleaned up + as before. The doctor's separate raw reader is unaffected and keeps + the strict contract, since it deliberately surfaces a malformed `identity` + block as a diagnostic. Fixes + [issue #270](https://github.com/tableau/hyper-api-rust/issues/270). +- **The "atomic" discovery-file write wasn't atomic, on the strength of a + false premise about Windows.** `write_discovery_record` deleted the + existing `daemon.json` before renaming the temp file into place, with a + comment claiming "`rename` fails if target exists" on Windows — but + `std::fs::rename` replaces an existing target on both Windows + (`MoveFileExW` with `MOVEFILE_REPLACE_EXISTING`, falling back to + `SetFileInformationByHandle`) and Unix. The unnecessary delete opened a + window where the file did not exist at all; since `try_restart_hyperd` + rewrites this file on every `hyperd` restart, a concurrent `discover()` + could observe `Missing` mid-restart. The write is now a single `rename`, + matching what the function's doc comment already promised. +- **An oversized-but-well-formed discovery record was reported as + `Malformed`.** The doctor's bounded raw reader rejected any file over 64 + KiB before attempting to parse it, so a file that was perfectly valid JSON + — just unexpectedly large — sent the user to fix "invalid JSON" that + parsed fine. `RawDiscoveryRead` gained an `Oversized` variant so the doctor + reports what actually happened instead. +- **A daemon peer that closed a health-command connection without writing + anything was reported as a successful, empty response.** `daemon_stop` + printed `Daemon responded:` with nothing after it and exited 0, which a + calling script would read as a successful stop. `send_command_with_timeout` + now returns `UnexpectedEof` when the connection ends before any bytes are + read. `doctor`'s separate `send_doctor_command` read loop had the same + shape and now classifies a silent close identically. This changes one + `doctor` warning: a recorded daemon candidate that accepts the connection + and then closes without answering is now reported as + `daemon_discovery_candidate_unreachable` ("did not return fresh enriched + STATUS") rather than `daemon_status_malformed` ("returned malformed or + unenriched STATUS"), which is what actually happened — the peer sent no + response to be malformed. Fixes 3 of the 5 items in + [issue #275](https://github.com/tableau/hyper-api-rust/issues/275). +- **After restarting `hyperd`, the daemon advertised the new endpoint over the + health port before writing it to the discovery file.** `try_restart_hyperd` + copied the endpoint into the in-memory `DaemonInfo` that `STATUS` serves and + released that lock *before* rewriting `daemon.json`. Since + `discovery::discover()` — the path every client's `Engine::new` takes — + reads the file, a client discovering inside that window connected to the + `hyperd` the daemon had just dropped. The write failing was worse than the + window: the error return happened with the new endpoint already published to + `STATUS` and the replacement `HyperProcess` dropped on the way out, leaving + `STATUS` durably naming a `hyperd` the daemon itself had killed. The + discovery file is now persisted first and the `DaemonInfo` flipped second, + both under one lock, so observing the new endpoint through either channel + implies a live `hyperd` behind it and no `STATUS` reader can see an endpoint + the file has not committed. Fixes + [issue #284](https://github.com/tableau/hyper-api-rust/issues/284). ## [0.5.0] - 2026-06-07 diff --git a/hyperdb-mcp/src/daemon/discovery.rs b/hyperdb-mcp/src/daemon/discovery.rs index 04c7d70..9bcea72 100644 --- a/hyperdb-mcp/src/daemon/discovery.rs +++ b/hyperdb-mcp/src/daemon/discovery.rs @@ -79,6 +79,32 @@ impl DaemonBuildIdentity { } } +/// Bytes carried alongside a [`RawDiscoveryRead`] classification so a caller +/// willing to try a looser shape can re-parse them without a second `read`. +/// +/// `Debug` deliberately renders only the length. `RawDiscoveryRead` is +/// `Debug`-formatted into diagnostics, and a discovery file this process could +/// not parse may contain anything at all — a `Vec`'s derived `Debug` would +/// echo every one of those bytes into a doctor report or log line, just as a +/// decimal list rather than as text. +pub(crate) struct DiscoveryBytes(Vec); + +impl DiscoveryBytes { + pub(crate) fn new(bytes: Vec) -> Self { + Self(bytes) + } + + fn as_slice(&self) -> &[u8] { + &self.0 + } +} + +impl std::fmt::Debug for DiscoveryBytes { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "<{} bytes>", self.0.len()) + } +} + #[derive(Debug)] pub(crate) enum RawDiscoveryRead { Missing { @@ -90,6 +116,23 @@ pub(crate) enum RawDiscoveryRead { }, Malformed { path: crate::diagnostics::ReportedPath, + /// The bytes that failed to parse, carried so a caller willing to try + /// a looser shape (see [`discover`]) can do so without a second + /// `read` — which would cost redundant I/O and let a rewrite land + /// between the two parses. + contents: DiscoveryBytes, + }, + /// The file exceeded [`MAX_DISCOVERY_FILE_BYTES`] before JSON parsing was + /// even attempted. Distinct from [`Self::Malformed`]: the content may be + /// perfectly well-formed JSON — it's merely larger than any legitimate + /// discovery record should be — so callers (the doctor) should report + /// what actually happened instead of sending the user to fix "invalid + /// JSON" that isn't. + Oversized { + path: crate::diagnostics::ReportedPath, + /// The bytes read before the cap was hit, carried for the same reason + /// as [`Self::Malformed`]'s. + contents: DiscoveryBytes, }, Parsed { path: crate::diagnostics::ReportedPath, @@ -139,14 +182,18 @@ pub(crate) fn read_discovery_file_raw(path: &Path) -> RawDiscoveryRead { }; } if contents.len() > MAX_DISCOVERY_FILE_BYTES { - return RawDiscoveryRead::Malformed { + return RawDiscoveryRead::Oversized { path: reported_path, + contents: DiscoveryBytes::new(contents), }; } - parse_discovery_contents(reported_path, &contents) + parse_discovery_contents(reported_path, contents) } +/// Read the discovery file the way the client fast path does: follow +/// symlinks and accept any valid record size (unlike the doctor's bounded, +/// no-follow [`read_discovery_file_raw`]). fn read_discovery_file_legacy(path: &Path) -> RawDiscoveryRead { let reported_path = crate::diagnostics::ReportedPath::from_os_str(path.as_os_str()); let contents = match std::fs::read(path) { @@ -163,20 +210,21 @@ fn read_discovery_file_legacy(path: &Path) -> RawDiscoveryRead { }; } }; - parse_discovery_contents(reported_path, &contents) + parse_discovery_contents(reported_path, contents) } fn parse_discovery_contents( reported_path: crate::diagnostics::ReportedPath, - contents: &[u8], + contents: Vec, ) -> RawDiscoveryRead { - match serde_json::from_slice(contents) { + match serde_json::from_slice(&contents) { Ok(record) => RawDiscoveryRead::Parsed { path: reported_path, record, }, Err(_) => RawDiscoveryRead::Malformed { path: reported_path, + contents: DiscoveryBytes::new(contents), }, } } @@ -254,8 +302,14 @@ fn write_discovery_record(record: &(impl Serialize + ?Sized)) -> io::Result<()> let tmp_path = dir.join("daemon.json.tmp"); let json = serde_json::to_string_pretty(record).map_err(|e| io::Error::other(e.to_string()))?; std::fs::write(&tmp_path, json.as_bytes())?; - // On Windows, rename fails if target exists. Remove stale target first. - let _ = std::fs::remove_file(&path); + // `std::fs::rename` already replaces an existing target atomically on + // both Unix (`rename(2)`) and Windows (`MoveFileExW` with + // `MOVEFILE_REPLACE_EXISTING`, falling back to `SetFileInformationByHandle` + // with `FILE_RENAME_FLAG_REPLACE_IF_EXISTS`). Pre-deleting the target here + // would reintroduce exactly the window this function's doc comment + // promises not to have: a concurrent `discover()` could observe the file + // as `Missing` mid-restart (see `try_restart_hyperd`, which rewrites this + // file on every `hyperd` restart). std::fs::rename(&tmp_path, &path)?; Ok(()) } @@ -263,12 +317,25 @@ fn write_discovery_record(record: &(impl Serialize + ?Sized)) -> io::Result<()> /// Read the discovery file and validate that the daemon is still alive. /// Returns `None` if no daemon is running (file missing, stale, or unreachable). pub fn discover() -> Option { - let path = discovery_file_path().ok()?; + let discovery_path = discovery_file_path().ok()?; // Preserve the historical client-discovery contract: normal discovery // follows symlinks and accepts any valid record size. Doctor uses the // separate bounded, no-follow raw reader above because it must never // mutate or block on a special file. - let record = match read_discovery_file_legacy(&path) { + // + // The `RawDiscoveryRead` classification below chooses a debug-log message + // and short-circuits a missing/unreadable file. It must never gate whether + // a *live* daemon is discovered: an unrecognized, reshaped, or absent + // `identity` block is forward-compatible noise to this fast path — see + // docs/superpowers/specs/2026-08-13-hyperdb-mcp-agent-ux-design.md + // ("Unknown fields remain forward compatible"). So when the strict + // `DaemonRecord` parse fails, we don't give up: we re-parse the bytes the + // classification already carries as a tolerant `DaemonInfo`, which ignores + // unknown fields and never fails on a nested object (like `identity`) that + // this fast path doesn't need. (The doctor's raw reader keeps the strict + // `DaemonRecord` contract because it deliberately wants to know about a + // malformed `identity` block.) + let (info, from_fallback) = match read_discovery_file_legacy(&discovery_path) { RawDiscoveryRead::Missing { path } => { tracing::debug!(encoding = ?path.encoding, "daemon discovery file is missing"); return None; @@ -277,25 +344,50 @@ pub fn discover() -> Option { tracing::debug!(?kind, encoding = ?path.encoding, "daemon discovery file is unreadable"); return None; } - RawDiscoveryRead::Malformed { path } => { - tracing::debug!(encoding = ?path.encoding, "daemon discovery file is malformed"); - return None; + // `read_discovery_file_legacy` applies no size cap, so it does not + // produce `Oversized` today. Folding it in with `Malformed` keeps that + // from mattering: were this reader ever given a cap, a large-but-valid + // record would take the tolerant fallback rather than silently + // becoming an undiscoverable daemon — the exact defect this fallback + // exists to prevent. + RawDiscoveryRead::Oversized { path, contents } + | RawDiscoveryRead::Malformed { path, contents } => { + let Ok(info) = serde_json::from_slice::(contents.as_slice()) else { + tracing::debug!(encoding = ?path.encoding, "daemon discovery file is malformed"); + return None; + }; + tracing::debug!(encoding = ?path.encoding, "daemon discovery file failed the strict parse; accepted as legacy DaemonInfo"); + (info, true) } RawDiscoveryRead::Parsed { path, record } => { - tracing::debug!(encoding = ?path.encoding, "daemon discovery file parsed"); - record + tracing::debug!(encoding = ?path.encoding, "daemon discovery file parsed (enriched)"); + (record.info().clone(), false) } }; - let info = record.info().clone(); // Validate liveness by connecting to the health port if is_daemon_alive(info.health_port) { - Some(info) - } else { - // Stale file — daemon crashed. Clean up. - let _ = std::fs::remove_file(&path); - None + return Some(info); } + + if from_fallback { + // Never stale-clean a record we could not strictly parse. This branch + // exists precisely because a *newer* daemon may write a record this + // client cannot fully understand, and a single failed 300 ms PING is + // not grounds for destroying it: the daemon rewrites `daemon.json` + // only on `hyperd` restart, so deleting here would hide a live newer + // daemon from every client, not just this one. Leaving it also keeps + // the promise doctor already makes to operators about a record it + // could not parse ("it was left unchanged"). + tracing::debug!( + "leaving the leniently parsed discovery record in place after a failed health check" + ); + return None; + } + + // Stale file — daemon crashed. Clean up. + let _ = std::fs::remove_file(&discovery_path); + None } /// Remove the discovery file (called during graceful shutdown). @@ -464,7 +556,7 @@ mod tests { use std::panic::{AssertUnwindSafe, catch_unwind}; use std::sync::{Arc, Mutex}; - use serde_json::json; + use serde_json::{Value, json}; use tempfile::TempDir; use crate::daemon::health::{DaemonState, HealthListener}; @@ -669,6 +761,228 @@ mod tests { ); } + /// Exercises the actual scan, health check, and cleanup logic across a + /// process boundary so `discover()`'s `HYPERDB_STATE_DIR` env override + /// can't race other tests in this binary. + fn run_identity_forward_compat_scenario() { + assert!( + std::env::var_os("HYPERDB_STATE_DIR").is_some(), + "child scenario requires an isolated state directory" + ); + + // Three real-world shapes a client fast path must still discover a + // live daemon through, per the forward-compatibility contract in + // docs/superpowers/specs/2026-08-13-hyperdb-mcp-agent-ux-design.md + // ("Unknown fields remain forward compatible"). None of these is + // valid input for `DaemonBuildIdentity`, so a strict `DaemonRecord` + // parse of the whole file must fail for every one of them — that's + // asserted below as the fixture sanity check. + let shapes: [(&str, Value); 3] = [ + ("identity retyped as a string", json!("not-an-object")), + ( + "executable_path reshaped", + json!({ + "mcp_version": "0.7.0.rabc123", + "executable_path": "/opt/hyperdb/bin/hyperdb-mcp" + }), + ), + ( + "identity missing executable_path", + json!({ "mcp_version": "0.7.0.rabc123" }), + ), + ]; + + let mut failures = Vec::new(); + for (index, (label, identity_json)) in shapes.into_iter().enumerate() { + let pid = 9_000 + u32::try_from(index).expect("fixture index fits in u32"); + let health_listener = HealthListener::bind(0).unwrap(); + let info = DaemonInfo { + pid, + hyperd_endpoint: "127.0.0.1:54321".to_string(), + health_port: health_listener.port, + started_at: "2026-08-13T12:34:56Z".to_string(), + version: "0.7.0".to_string(), + }; + + let wire = json!({ + "pid": info.pid, + "hyperd_endpoint": info.hyperd_endpoint, + "health_port": info.health_port, + "started_at": info.started_at, + "version": info.version, + "identity": identity_json, + }); + + if serde_json::from_value::(wire.clone()).is_ok() { + failures.push(format!( + "{label}: fixture unexpectedly satisfied the strict DaemonRecord parse; \ + it no longer exercises the forward-compatibility regression" + )); + } + + std::fs::create_dir_all(state_dir().unwrap()).unwrap(); + let path = discovery_file_path().unwrap(); + std::fs::write(&path, serde_json::to_vec(&wire).unwrap()).unwrap(); + + let health_state = Arc::new(DaemonState::new()); + let health_info = Arc::new(Mutex::new(info.clone())); + let run_state = Arc::clone(&health_state); + let run_info = Arc::clone(&health_info); + let health_server = + std::thread::spawn(move || health_listener.run(run_state, run_info)); + + match discover() { + Some(discovered) if discovered == info => {} + Some(other) => failures.push(format!( + "{label}: discover() returned different facts than the live daemon: {other:?}" + )), + None => failures.push(format!( + "{label}: discover() returned None for a live, healthy daemon whose only \ + defect is an unrecognized `identity` block" + )), + } + + health_state.request_shutdown(); + health_server.join().unwrap(); + let _ = std::fs::remove_file(&path); + } + + assert!( + failures.is_empty(), + "identity forward-compatibility discover failures:\n{}", + failures.join("\n") + ); + } + + #[test] + fn discover_tolerates_forward_incompatible_identity_shapes() { + const CHILD_SENTINEL_ENV: &str = "HYPERDB_MCP_IDENTITY_FORWARD_COMPAT_CHILD"; + const TEST_NAME: &str = + "daemon::discovery::tests::discover_tolerates_forward_incompatible_identity_shapes"; + + let _process_guard = crate::diagnostics::real_network_test_guard(); + if let Some(marker) = std::env::var_os(CHILD_SENTINEL_ENV) { + std::fs::write(std::path::PathBuf::from(marker), b"started").unwrap(); + run_identity_forward_compat_scenario(); + return; + } + run_discovery_compatibility_child(TEST_NAME, CHILD_SENTINEL_ENV); + } + + /// Returns a port that nothing is listening on, by binding one and + /// releasing it immediately. Anything that later camps on it still fails + /// `ping_identified` (no PONG token), so the port reads as dead either way. + fn released_port() -> u16 { + let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap(); + listener.local_addr().unwrap().port() + } + + fn run_lenient_stale_cleanup_scenario() { + assert!( + std::env::var_os("HYPERDB_STATE_DIR").is_some(), + "child scenario requires an isolated state directory" + ); + + let path = discovery_file_path().unwrap(); + let mut failures = Vec::new(); + + // A record whose `identity` block this client cannot understand, but + // whose legacy fields are perfectly readable — exactly what a *newer* + // daemon looks like to an older client, which is the only reason the + // lenient fallback exists. + let lenient_wire = json!({ + "pid": 9_100, + "hyperd_endpoint": "127.0.0.1:54321", + "health_port": released_port(), + "started_at": "2026-08-13T12:34:56Z", + "version": "0.7.0", + "identity": "not-an-object", + }); + if serde_json::from_value::(lenient_wire.clone()).is_ok() { + failures.push( + "fixture satisfied the strict DaemonRecord parse; it no longer reaches the \ + lenient fallback this test is about" + .to_string(), + ); + } + if serde_json::from_value::(lenient_wire.clone()).is_err() { + failures.push( + "fixture is not readable as a legacy DaemonInfo either; it would be rejected \ + before the branch this test pins" + .to_string(), + ); + } + + std::fs::create_dir_all(state_dir().unwrap()).unwrap(); + let lenient_bytes = serde_json::to_vec(&lenient_wire).unwrap(); + std::fs::write(&path, &lenient_bytes).unwrap(); + + if let Some(info) = discover() { + failures.push(format!( + "discover() reported a live daemon on a released port: {info:?}" + )); + } + // The pin: a failed health check must not destroy a record this + // client could only read leniently. The daemon rewrites `daemon.json` + // only on `hyperd` restart, so deleting it here would hide a live, + // newer daemon from every client on the machine. + match std::fs::read(&path) { + Ok(after) if after == lenient_bytes => {} + Ok(_) => { + failures.push("discover() changed the leniently parsed record's bytes".to_string()); + } + Err(error) => failures.push(format!( + "discover() removed the leniently parsed discovery record it could not \ + strictly parse: {error}" + )), + } + + // Contrast: a strictly parseable record on a dead port is still + // stale-cleaned, so the guard above is narrow rather than a blanket + // disabling of cleanup. + let mut strict_info = legacy_info(); + strict_info.pid = 9_101; + strict_info.health_port = released_port(); + write_discovery_file(&strict_info).unwrap(); + if let Some(info) = discover() { + failures.push(format!( + "discover() reported a live daemon for the strict record on a released port: {info:?}" + )); + } + if path.exists() { + failures.push( + "discover() did not stale-clean a strictly parsed record on a dead port" + .to_string(), + ); + } + + assert!( + failures.is_empty(), + "lenient stale-cleanup failures:\n{}", + failures.join("\n") + ); + } + + /// A record that only the lenient `DaemonInfo` fallback could read must + /// survive a failed health check. Before the fallback existed, a strict + /// parse failure returned from `discover()` before ever reaching the + /// stale-cleanup `remove_file`, so tolerating the record must not also + /// start deleting it. + #[test] + fn discover_preserves_a_leniently_parsed_stale_record() { + const CHILD_SENTINEL_ENV: &str = "HYPERDB_MCP_LENIENT_STALE_CLEANUP_CHILD"; + const TEST_NAME: &str = + "daemon::discovery::tests::discover_preserves_a_leniently_parsed_stale_record"; + + let _process_guard = crate::diagnostics::real_network_test_guard(); + if let Some(marker) = std::env::var_os(CHILD_SENTINEL_ENV) { + std::fs::write(std::path::PathBuf::from(marker), b"started").unwrap(); + run_lenient_stale_cleanup_scenario(); + return; + } + run_discovery_compatibility_child(TEST_NAME, CHILD_SENTINEL_ENV); + } + #[test] fn raw_discovery_read_is_non_mutating_and_distinguishes_io() { const SECRET_SENTINEL: &str = "RAW_DISCOVERY_SECRET_MUST_NOT_LEAK"; @@ -722,11 +1036,28 @@ mod tests { match catch_raw_read(&malformed_path) { Ok(state) => { - if format!("{state:?}").contains(SECRET_SENTINEL) { + let rendered = format!("{state:?}"); + if rendered.contains(SECRET_SENTINEL) { failures.push("malformed state leaked discovery contents".to_string()); } + // `Malformed` carries the unparsed bytes so `discover()` can + // retry them without a second `read`. A derived `Debug` on a + // `Vec` would print every byte as a decimal list — the same + // leak, merely unsearchable — so check for that form too. + let sentinel_as_debug_bytes = SECRET_SENTINEL + .as_bytes() + .iter() + .map(u8::to_string) + .collect::>() + .join(", "); + if rendered.contains(&sentinel_as_debug_bytes) { + failures.push( + "malformed state leaked discovery contents as a debug byte list" + .to_string(), + ); + } match state { - RawDiscoveryRead::Malformed { path } + RawDiscoveryRead::Malformed { path, .. } if path == ReportedPath::from_os_str(malformed_path.as_os_str()) => {} other => failures.push(format!( "malformed JSON was not reported as Malformed with its ReportedPath: {other:?}" @@ -797,7 +1128,7 @@ mod tests { } #[test] - fn raw_discovery_rejects_oversized_valid_json() { + fn raw_discovery_reports_oversized_valid_json_as_oversized() { const MAX_EXPECTED_DISCOVERY_BYTES: usize = 64 * 1024; let tmp = TempDir::new().unwrap(); @@ -845,10 +1176,10 @@ mod tests { )), } match read_discovery_file_raw(&oversized_path) { - RawDiscoveryRead::Malformed { path: reported } + RawDiscoveryRead::Oversized { path: reported, .. } if reported == ReportedPath::from_os_str(oversized_path.as_os_str()) => {} other => failures.push(format!( - "oversized valid JSON was not rejected as malformed: {other:?}" + "oversized valid JSON was not honestly classified as Oversized (not Malformed): {other:?}" )), } for (label, path, expected) in [ @@ -908,7 +1239,7 @@ mod tests { )), } match read_discovery_file_raw(&path) { - RawDiscoveryRead::Malformed { path: reported } + RawDiscoveryRead::Oversized { path: reported, .. } if reported == ReportedPath::from_os_str(path.as_os_str()) => {} RawDiscoveryRead::Missing { .. } => { failures @@ -918,6 +1249,12 @@ mod tests { "doctor raw reader reported the oversized record unreadable: {kind:?}" )), RawDiscoveryRead::Malformed { .. } => { + failures.push( + "doctor raw reader misclassified the well-formed oversized record as malformed" + .to_string(), + ); + } + RawDiscoveryRead::Oversized { .. } => { failures.push("doctor raw reader reported the wrong oversized path".to_string()); } RawDiscoveryRead::Parsed { .. } => { diff --git a/hyperdb-mcp/src/daemon/health.rs b/hyperdb-mcp/src/daemon/health.rs index 33825b6..43c084c 100644 --- a/hyperdb-mcp/src/daemon/health.rs +++ b/hyperdb-mcp/src/daemon/health.rs @@ -336,6 +336,12 @@ pub fn send_command_with_timeout( stream.set_read_timeout(Some(remaining_io_time(io_deadline)?))?; let mut byte = [0]; match stream.read(&mut byte) { + Ok(0) if response.is_empty() => { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "health command connection closed before any response was sent", + )); + } Ok(0) => break, Ok(_) if response.len() == MAX_HEALTH_RESPONSE_BYTES => { return Err(std::io::Error::new( @@ -797,4 +803,43 @@ mod tests { listening socket" ); } + + /// A peer that reads the request and then closes without writing anything + /// must be reported as a failed command, not a successful empty response. + /// `daemon_stop` in `main.rs` treats `Ok(_)` as "the daemon responded" and + /// exits 0, so an `Ok("")` here would silently mask a peer that never + /// answered at all. + #[test] + fn peer_close_without_any_response_is_an_error_not_empty_success() { + let peer = spawn_test_peer(|stream, _stop| { + let request = read_test_command(&stream)?; + if request != "PING\n" { + return Err(format!("unexpected health command: {request:?}")); + } + // Close the connection immediately, writing nothing at all. + drop(stream); + Ok(()) + }); + + let call = catch_unwind(AssertUnwindSafe(|| { + send_command_with_timeout( + peer.port, + "PING", + Duration::from_secs(1), + Duration::from_secs(1), + ) + })); + peer.finish() + .expect("silent-close peer must shut down cleanly"); + let result = match call { + Ok(outcome) => outcome, + Err(payload) => resume_unwind(payload), + }; + + assert_eq!( + result.as_ref().err().map(std::io::Error::kind), + Some(std::io::ErrorKind::UnexpectedEof), + "a peer that closes having sent nothing must be reported as an error, not {result:?}" + ); + } } diff --git a/hyperdb-mcp/src/daemon/run.rs b/hyperdb-mcp/src/daemon/run.rs index 19df656..4dc9155 100644 --- a/hyperdb-mcp/src/daemon/run.rs +++ b/hyperdb-mcp/src/daemon/run.rs @@ -287,7 +287,9 @@ async fn hyperd_monitor( } /// Attempt one restart of hyperd. Drops the old process, spawns a new one, -/// updates `DaemonInfo`, and rewrites the discovery file. +/// rewrites the discovery file, and only then updates the `DaemonInfo` that +/// STATUS serves — see the publish-order comment in the body for why that +/// sequence is load-bearing rather than incidental. /// /// Every call (success or spawn-failure) consumes one slot from the rate-limit /// window — a broken hyperd binary should not spin forever. @@ -317,16 +319,37 @@ fn try_restart_hyperd( .ok_or_else(|| RestartError::SpawnFailed("hyperd did not report endpoint".into()))? .to_string(); - // Publish the new endpoint to STATUS readers and to the discovery file. - // We snapshot the updated DaemonInfo while holding info_arc's lock, then - // write the file outside the lock to keep the critical section small. - let snapshot = { + // Publish the new endpoint, discovery file first and STATUS second. + // + // Two independent channels advertise the endpoint: the `daemon.json` + // discovery file that `discovery::discover()` reads — the path every + // client's `Engine::new` takes — and the in-memory `DaemonInfo` that the + // health port's STATUS command serves. Updating STATUS first and the file + // afterwards left a window in which STATUS announced the freshly spawned + // `hyperd` while the file still named the one we just dropped, so a client + // discovering during that window connected to a dead port. Persisting + // first inverts that: whichever channel an observer reads, seeing the new + // endpoint now implies a live `hyperd` behind it. + // + // Both steps happen under a single `info_arc` lock so the pair is atomic + // to STATUS readers — no reader can catch STATUS already advertising an + // endpoint the file has not committed yet. The critical section is one + // small serialize-and-rename, and the only contender is `status_json`, + // whose own critical section is a single clone. + // `write_enriched_discovery_file` never touches `info_arc`, so holding it + // here cannot deadlock. + // + // Failing the write before mutating `DaemonInfo` also matters: the `?` + // below drops `new_hyper` on the way out, and STATUS must not be left + // advertising the endpoint of a `hyperd` this function just killed. + { let mut info_guard = info_arc.lock().expect("DaemonInfo mutex poisoned"); + let mut snapshot = info_guard.clone(); + snapshot.hyperd_endpoint.clone_from(&new_endpoint); + discovery::write_enriched_discovery_file(&snapshot) + .map_err(|e| RestartError::SpawnFailed(format!("discovery write: {e}")))?; info_guard.hyperd_endpoint.clone_from(&new_endpoint); - info_guard.clone() - }; - discovery::write_enriched_discovery_file(&snapshot) - .map_err(|e| RestartError::SpawnFailed(format!("discovery write: {e}")))?; + } guard.hyper = Some(new_hyper); Ok(new_endpoint) diff --git a/hyperdb-mcp/src/diagnostics.rs b/hyperdb-mcp/src/diagnostics.rs index 464917f..6ad60f4 100644 --- a/hyperdb-mcp/src/diagnostics.rs +++ b/hyperdb-mcp/src/diagnostics.rs @@ -346,6 +346,10 @@ pub(crate) enum DoctorDaemonState { Missing, Unreadable, Malformed, + /// The discovery file is well-formed but larger than any legitimate + /// record should be; distinct from `Malformed` so the user isn't told to + /// fix "invalid JSON" that parses just fine. + Oversized, ParsedUnreachable, LiveFromDiscovery, LiveFromScan, @@ -375,6 +379,7 @@ pub(crate) enum DoctorDaemonWarning { kind: io::ErrorKind, }, MalformedDiscovery, + OversizedDiscovery, DiscoveryCandidateUnreachable { responding_port: u16, }, @@ -436,6 +441,10 @@ pub(crate) fn collect_doctor_daemon( warnings.push(DoctorDaemonWarning::MalformedDiscovery); (DoctorDaemonState::Malformed, None) } + RawDiscoveryRead::Oversized { .. } => { + warnings.push(DoctorDaemonWarning::OversizedDiscovery); + (DoctorDaemonState::Oversized, None) + } RawDiscoveryRead::Parsed { record, .. } => { (DoctorDaemonState::ParsedUnreachable, Some(record)) } @@ -1178,6 +1187,17 @@ fn send_doctor_command( } let read_capacity = remaining_capacity.min(chunk.len()); match stream.read(&mut chunk[..read_capacity]) { + // Classify a silent close exactly as `send_command_with_timeout` + // does. Returning `Ok("")` here would reach `verify_status_candidate` + // as an unparseable STATUS body and be reported as + // `MalformedStatus` — but a peer that never wrote a byte did not + // send a malformed response, it failed to answer at all. + Ok(0) if response.is_empty() => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "doctor health peer closed the connection before any response was sent", + )); + } Ok(0) => break, Ok(count) => { let line_end = chunk[..count] @@ -1304,6 +1324,10 @@ fn daemon_doctor_warning(warning: &DoctorDaemonWarning) -> DoctorWarning { "daemon_discovery_malformed", "The daemon discovery file was malformed; it was left unchanged.", ), + DoctorDaemonWarning::OversizedDiscovery => doctor_warning( + "daemon_discovery_oversized", + "The daemon discovery file is valid but larger than any legitimate record should be; it was left unchanged.", + ), DoctorDaemonWarning::DiscoveryCandidateUnreachable { responding_port } => doctor_warning( "daemon_discovery_candidate_unreachable", format!( @@ -1439,6 +1463,7 @@ const fn daemon_state_label(state: DoctorDaemonState) -> &'static str { DoctorDaemonState::Missing => "missing", DoctorDaemonState::Unreadable => "unreadable", DoctorDaemonState::Malformed => "malformed", + DoctorDaemonState::Oversized => "oversized", DoctorDaemonState::ParsedUnreachable => "parsed_unreachable", DoctorDaemonState::LiveFromDiscovery => "live_from_discovery", DoctorDaemonState::LiveFromScan => "live_from_scan", @@ -1671,7 +1696,7 @@ mod tests { use tempfile::TempDir; use crate::daemon::discovery::{ - DaemonBuildIdentity, DaemonInfo, DaemonRecord, PortScan, RawDiscoveryRead, + DaemonBuildIdentity, DaemonInfo, DaemonRecord, DiscoveryBytes, PortScan, RawDiscoveryRead, read_discovery_file_raw, }; use crate::daemon::health::{DaemonState, HealthListener}; @@ -1688,6 +1713,7 @@ mod tests { Missing, Unreadable(io::ErrorKind), Malformed, + Oversized, Parsed(Value), } @@ -1697,7 +1723,17 @@ mod tests { match self { Self::Missing => RawDiscoveryRead::Missing { path }, Self::Unreadable(kind) => RawDiscoveryRead::Unreadable { path, kind: *kind }, - Self::Malformed => RawDiscoveryRead::Malformed { path }, + // The doctor never reads the carried bytes — they exist for + // `discover()`'s tolerant fallback — so any non-record payload + // is a faithful fixture here. + Self::Malformed => RawDiscoveryRead::Malformed { + path, + contents: DiscoveryBytes::new(b"{ unterminated".to_vec()), + }, + Self::Oversized => RawDiscoveryRead::Oversized { + path, + contents: DiscoveryBytes::new(b"{ well-formed but enormous }".to_vec()), + }, Self::Parsed(value) => RawDiscoveryRead::Parsed { path, record: serde_json::from_value(value.clone()).unwrap(), @@ -2292,6 +2328,99 @@ mod tests { ); } + /// A recorded daemon candidate that accepts the connection and then closes + /// without writing a byte never sent a response, so it cannot have sent a + /// malformed one. `send_doctor_command` used to return `Ok("")` there, which + /// reached `verify_status_candidate` as an unparseable STATUS body and was + /// reported to the operator as `daemon_status_malformed`. It must classify + /// the silent close the way `send_command_with_timeout` does — as an EOF — + /// so the candidate is reported unreachable instead. + #[test] + fn real_doctor_reports_a_silent_peer_close_as_unreachable_not_malformed() { + use std::io::{BufRead as _, BufReader}; + use std::net::TcpListener; + + let _network_guard = real_network_test_guard(); + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = std::thread::spawn(move || -> Result { + let (stream, _) = listener + .accept() + .map_err(|error| format!("silent-close peer accept failed: {error}"))?; + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .map_err(|error| error.to_string())?; + let mut command = String::new(); + BufReader::new(&stream) + .read_line(&mut command) + .map_err(|error| format!("silent-close peer read failed: {error}"))?; + // Close having written nothing at all. + drop(stream); + Ok(command.trim().to_string()) + }); + + let tmp = TempDir::new().unwrap(); + let discovery_path = tmp.path().join("daemon.json"); + std::fs::write( + &discovery_path, + serde_json::to_vec(&enriched_status( + 4_242, + port, + "0.7.0.rsilent", + "/opt/hyperdb/bin/hyperdb-mcp", + )) + .unwrap(), + ) + .unwrap(); + + // `span: 0` leaves the port scan with no candidates, so the report + // reflects the discovery candidate's outcome alone. + let report = collect_real_doctor_daemon( + Some(&discovery_path), + None, + PortScan { + base: port, + span: 0, + }, + ); + let observed_command = server.join().unwrap(); + + let mut failures = Vec::new(); + match observed_command { + Ok(command) if command == "STATUS" => {} + Ok(command) => failures.push(format!( + "discovery candidate was probed with {command:?}, expected a direct STATUS" + )), + Err(error) => failures.push(error), + } + if report.verified.is_some() { + failures.push("a peer that answered nothing was accepted as a daemon".to_string()); + } + if report.state != DoctorDaemonState::ParsedUnreachable { + failures.push(format!( + "expected state parsed_unreachable, got {:?}", + report.state + )); + } + if report.warnings + != vec![DoctorDaemonWarning::DiscoveryCandidateUnreachable { + responding_port: port, + }] + { + failures.push(format!( + "a silent close must be reported as an unreachable candidate, not as malformed \ + STATUS; got {:?}", + report.warnings + )); + } + + assert!( + failures.is_empty(), + "silent-close doctor classification failures:\n{}", + failures.join("\n") + ); + } + #[test] fn collect_doctor_state_matrix_is_pure() { struct Case { @@ -2341,6 +2470,15 @@ mod tests { expected_live: None, expected_warnings: vec![DoctorDaemonWarning::MalformedDiscovery], }, + Case { + name: "oversized", + raw: RawFixture::Oversized, + scan_ports: vec![], + status_responses: vec![], + expected_state: DoctorDaemonState::Oversized, + expected_live: None, + expected_warnings: vec![DoctorDaemonWarning::OversizedDiscovery], + }, Case { name: "parsed-unreachable", raw: RawFixture::Parsed(enriched_status( diff --git a/hyperdb-mcp/tests/daemon_tests.rs b/hyperdb-mcp/tests/daemon_tests.rs index 30db814..db0b5a3 100644 --- a/hyperdb-mcp/tests/daemon_tests.rs +++ b/hyperdb-mcp/tests/daemon_tests.rs @@ -1378,8 +1378,9 @@ fn hyperd_monitor_detects_killed_hyperd_and_restarts() { // Wait up to 12 seconds for the monitor to fire and restart hyperd. // (5s monitor tick + spawn time + slack.) - let new_endpoint = wait_for_endpoint_change_or_recovery(daemon.info.health_port, 12) - .expect("daemon should restart hyperd within 12s"); + let new_endpoint = + wait_for_live_hyperd_after_kill(daemon.info.health_port, &daemon.info.hyperd_endpoint, 12) + .expect("daemon should restart hyperd within 12s"); // The new endpoint must be reachable. Don't assert it differs from the old — // port reuse is permitted by the OS. @@ -1411,8 +1412,9 @@ fn client_report_triggers_restart_after_kill() { let response = health::send_command(daemon.info.health_port, "REPORT_HYPERD_ERROR").unwrap(); assert_eq!(response.trim(), "OK"); - let new_endpoint = wait_for_endpoint_change_or_recovery(daemon.info.health_port, 12) - .expect("daemon should restart hyperd within 12s after report"); + let new_endpoint = + wait_for_live_hyperd_after_kill(daemon.info.health_port, &daemon.info.hyperd_endpoint, 12) + .expect("daemon should restart hyperd within 12s after report"); let probe = std::net::TcpStream::connect_timeout( &new_endpoint.parse().expect("valid endpoint"), @@ -1462,8 +1464,10 @@ fn engine_recovers_after_hyperd_killed() { find_hyperd_pid_for_endpoint(&daemon.info.hyperd_endpoint).expect("should find hyperd pid"); kill_pid(pid); - // Wait for daemon-side restart (discovery file gets a new endpoint). - wait_for_endpoint_change_or_recovery(daemon.info.health_port, 12) + // Wait for daemon-side restart. The daemon persists `daemon.json` before + // it flips what STATUS serves, so a live endpoint from STATUS means the + // discovery file `Engine::new` reads below is already committed. + wait_for_live_hyperd_after_kill(daemon.info.health_port, &daemon.info.hyperd_endpoint, 12) .expect("daemon should restart hyperd within 12s"); // Engine #2: post-restart. This mirrors what `with_engine` does after a @@ -1858,12 +1862,48 @@ fn kill_pid(pid: u32) { assert!(status.success(), "kill -9 {pid} failed"); } -/// Poll the daemon's `STATUS` endpoint until the reported `hyperd_endpoint` is -/// reachable (i.e. a fresh hyperd has been spawned after a kill). Returns the -/// endpoint string, or `None` if the timeout expires. +/// Wait for the daemon to advertise a *live* hyperd after the process serving +/// `killed_endpoint` was killed, and return the endpoint it settled on, or +/// `None` if the timeout expires. +/// +/// Asking only "is the endpoint STATUS reports connectable?" is not enough, +/// and was the cause of a ~20%-per-run failure on this file's restart tests. +/// For a few milliseconds after `SIGKILL` the doomed hyperd's listening socket +/// still completes inbound connections, so a single-phase gate polled +/// immediately after the kill returns on its first attempt — carrying the +/// *pre-kill* endpoint, because the daemon's 5s monitor tick has not fired and +/// STATUS still advertises the old process. The caller then raced the kernel +/// tearing that socket down and failed with `Connection refused` a few +/// milliseconds later. +/// +/// So gate on the old hyperd being observably gone first, then on the +/// advertised endpoint being connectable. That ordering is also why this +/// cannot simply require the endpoint to *change*: the OS is free to hand the +/// replacement the same port, and a changed-endpoint gate would then wait out +/// its whole timeout on a perfectly healthy restart. #[cfg(unix)] -fn wait_for_endpoint_change_or_recovery(health_port: u16, timeout_secs: u64) -> Option { +fn wait_for_live_hyperd_after_kill( + health_port: u16, + killed_endpoint: &str, + timeout_secs: u64, +) -> Option { let deadline = Instant::now() + Duration::from_secs(timeout_secs); + + // Phase 1: the killed hyperd must stop accepting connections. Until it + // does, any endpoint we observe could still be its soon-to-close socket. + let killed_addr = killed_endpoint.parse::().ok()?; + loop { + if Instant::now() >= deadline { + return None; + } + if std::net::TcpStream::connect_timeout(&killed_addr, Duration::from_millis(500)).is_err() { + break; + } + std::thread::sleep(Duration::from_millis(25)); + } + + // Phase 2: whatever STATUS advertises now must be reachable. With the old + // socket proven down, a successful connect means a live replacement. while Instant::now() < deadline { if let Ok(response) = health::send_command(health_port, "STATUS") && let Ok(parsed) = serde_json::from_str::(response.trim()) diff --git a/hyperdb-mcp/tests/recovery_tests.rs b/hyperdb-mcp/tests/recovery_tests.rs index 5717040..22a198c 100644 --- a/hyperdb-mcp/tests/recovery_tests.rs +++ b/hyperdb-mcp/tests/recovery_tests.rs @@ -481,7 +481,7 @@ fn run_slow_health_mutex_child() { let (report_seen_tx, report_seen_rx) = mpsc::channel(); let (report_release_tx, report_release_rx) = mpsc::channel(); // Set by each worker closure the instant its `resource_body_for_uri` call - // returns, and read by the peer *before* it probes the engine mutex (see + // returns, and read by the peer *after* it probes the engine mutex (see // `ReportObservation::probe_valid`). Reset before each worker starts. let worker_finished = Arc::new(AtomicBool::new(false)); let peer_info = daemon_info.clone(); @@ -592,11 +592,11 @@ fn run_slow_health_mutex_child() { // A second public call now takes the post-loss initialization path. The // dead endpoint makes Engine::try_daemon_mode emit another slow report. - // The peer reads `worker_finished` and probes `try_lock` synchronously - // before releasing that response, so this cannot pass merely because a - // scheduler slept past the 200 ms I/O budget: if the worker's call had - // already returned by the time the peer checked, `probe_valid` is false - // and the observation is treated as inconclusive rather than a pass. + // The peer probes `try_lock` and then reads `worker_finished`, both + // synchronously before releasing that response, so this cannot pass merely + // because a scheduler slept past the 200 ms I/O budget: if the worker's + // call had already returned by the time the peer checked, `probe_valid` is + // false and the observation is treated as inconclusive rather than a pass. worker_finished.store(false, Ordering::Release); let reinit_server = Arc::clone(&server); let reinit_worker_finished = Arc::clone(&worker_finished); @@ -758,12 +758,6 @@ fn run_controlled_health_peer( let engine_handle = engine_probe .get() .ok_or_else(|| "engine probe was not installed before report".to_string())?; - // Read BEFORE probing the mutex. If the calling worker's - // public call had already returned, it already released - // (and would trivially show as released regardless of - // whether production ever held it during the pending - // report) — the probe below cannot then prove anything. - let probe_valid = !worker_finished.load(Ordering::Acquire); let engine_mutex_available = match engine_handle.try_lock() { Ok(guard) => { drop(guard); @@ -774,6 +768,17 @@ fn run_controlled_health_peer( return Err("engine mutex was poisoned during report probe".to_string()); } }; + // Read AFTER probing the mutex. If the calling worker's public + // call had already returned, it already released the mutex + // (and would trivially show as released regardless of whether + // production ever held it during the pending report), so the + // probe above cannot then prove anything. The flag is + // monotonic false -> true, so observing `false` here implies + // it was also false before the probe; reading it first would + // instead leave a window for the worker to return in between, + // marking a vacuous probe valid. A late read can only produce + // a false *inconclusive*, never a false pass. + let probe_valid = !worker_finished.load(Ordering::Acquire); report_seen_tx .send(ReportObservation { sequence: report_sequence,