From 03b71a4a063520b3d06ff3d8be981c8250d5d2e0 Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Sun, 6 Sep 2026 13:02:12 -0700 Subject: [PATCH 1/4] test(mcp): run the daemon restart-after-crash tests on Linux and Windows CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eight resilience integration tests in daemon_tests.rs — including client_report_triggers_restart_after_kill, hyperd_monitor_detects_killed_hyperd_and_restarts, engine_recovers_after_hyperd_killed, and daemon_mode_two_engines_share_same_hyperd — carried an unconditional #[ignore], so Linux and Windows CI silently skipped them even though the stated reason names macOS only. The subsystem whose entire purpose is resilience had zero automated restart-after-crash coverage on any platform. Switch each to #[cfg_attr(target_os = "macos", ignore = ...)] so the existing reason and behavior are preserved on macOS while Linux and Windows now run them. Verified locally (macOS, where the ignore still applies by default): running them explicitly with --ignored passes reliably, 8/8, in ~14-22s total across two consecutive runs — nowhere near the 150s timeout that motivated the ignore, so the ignore itself is a real per-platform accommodation, not evidence the tests are broken. Also corrects a stale comment in health_listener_waits_for_command_after_accept that still described the listener's old 100ms accept-loop poll interval (now 5ms). --- hyperdb-mcp/tests/daemon_tests.rs | 45 ++++++++++++++++++++++++------- 1 file changed, 35 insertions(+), 10 deletions(-) diff --git a/hyperdb-mcp/tests/daemon_tests.rs b/hyperdb-mcp/tests/daemon_tests.rs index 76c6119..30db814 100644 --- a/hyperdb-mcp/tests/daemon_tests.rs +++ b/hyperdb-mcp/tests/daemon_tests.rs @@ -314,8 +314,9 @@ fn health_listener_waits_for_command_after_accept() { } } - // Keep the already-accepted first socket idle for more than two - // additional 100ms listener polls before sending its first command. + // Keep the already-accepted first socket idle for well over the + // listener's 5ms accept-loop poll interval before sending its + // first command. std::thread::sleep(Duration::from_millis(350)); idle_client.ping("delayed first-client") })(); @@ -1232,7 +1233,10 @@ fn takeover_decision_both_unparseable_reuses() { // ─── Integration tests: full daemon lifecycle with real hyperd ───────────────── #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn daemon_mode_engine_connects_to_shared_hyperd() { let _lock = acquire_env_lock(); let daemon = TestDaemon::start(); @@ -1251,7 +1255,10 @@ fn daemon_mode_engine_connects_to_shared_hyperd() { } #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn daemon_mode_two_engines_share_same_hyperd() { let _lock = acquire_env_lock(); let daemon = TestDaemon::start(); @@ -1294,7 +1301,10 @@ fn daemon_mode_two_engines_share_same_hyperd() { } #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn daemon_mode_persistent_database_file_survives_engine_drop() { let _lock = acquire_env_lock(); let _daemon = TestDaemon::start(); @@ -1319,7 +1329,10 @@ fn daemon_mode_persistent_database_file_survives_engine_drop() { } #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn daemon_mode_persistent_engine_data_is_queryable() { let _lock = acquire_env_lock(); let daemon = TestDaemon::start(); @@ -1348,7 +1361,10 @@ fn daemon_mode_persistent_engine_data_is_queryable() { #[cfg(unix)] #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn hyperd_monitor_detects_killed_hyperd_and_restarts() { let _lock = acquire_env_lock(); let daemon = TestDaemon::start(); @@ -1376,7 +1392,10 @@ fn hyperd_monitor_detects_killed_hyperd_and_restarts() { #[cfg(unix)] #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn client_report_triggers_restart_after_kill() { let _lock = acquire_env_lock(); let daemon = TestDaemon::start(); @@ -1404,7 +1423,10 @@ fn client_report_triggers_restart_after_kill() { #[cfg(unix)] #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn engine_recovers_after_hyperd_killed() { // End-to-end test: the user-visible behavior of this whole feature. // 1. Start daemon + create an Engine (= an MCP client connection). @@ -1456,7 +1478,10 @@ fn engine_recovers_after_hyperd_killed() { } #[test] -#[ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout"] +#[cfg_attr( + target_os = "macos", + ignore = "flaky on macOS CI — daemon startup exceeds 150s timeout" +)] fn daemon_mode_ephemeral_database_cleaned_up_on_drop() { let _lock = acquire_env_lock(); let _daemon = TestDaemon::start(); From d3c7424139d39c196a81f4d6dcf620d0cb6a5bcc Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Sun, 6 Sep 2026 13:02:22 -0700 Subject: [PATCH 2/4] test(mcp): close a false-pass race in the post-loss engine-mutex probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit slow_health_report_does_not_hold_engine_mutex asserted that the engine mutex stays available while a REPORT_HYPERD_ERROR round-trip is pending, using a peer-side try_lock probe as the proof. But the probe was only meaningful while the calling worker was still blocked inside report_hyperd_error_to_daemon (200ms budget). If the worker's call returned before the peer reached the probe — e.g. a scheduler delay pushing the peer past 200ms — the worker had already released everything, so try_lock trivially succeeded and the assertion passed without observing the property at all. Reproduced directly: inserting an artificial 300ms delay before the old probe made the test report 'ok' despite proving nothing; the same delay against the new code fails loudly with an explicit 'inconclusive' message instead. Fix: an AtomicBool set by each worker the instant its call returns, read by the peer before try_lock. If the worker already finished, the observation is marked probe_valid: false and the test fails with a clear 'rerun' message instead of silently passing. Also deletes a comment claiming 'production is red here because ensure_engine holds the engine mutex throughout Engine::new' — true at v0.7.2, false today: ensure_engine takes a separate engine_initialization single-flight mutex and drops the engine guard before Engine::new. Left in place, it would send a future maintainer hunting a problem that doesn't exist. --- hyperdb-mcp/tests/recovery_tests.rs | 76 +++++++++++++++++++++++------ 1 file changed, 62 insertions(+), 14 deletions(-) diff --git a/hyperdb-mcp/tests/recovery_tests.rs b/hyperdb-mcp/tests/recovery_tests.rs index 11386b8..5717040 100644 --- a/hyperdb-mcp/tests/recovery_tests.rs +++ b/hyperdb-mcp/tests/recovery_tests.rs @@ -11,6 +11,7 @@ use std::io::{BufRead as _, BufReader, Write as _}; use std::net::TcpListener; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Output, Stdio}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock, TryLockError, mpsc}; use std::thread; use std::time::{Duration, Instant}; @@ -439,6 +440,13 @@ type EngineHandle = Arc>>; struct ReportObservation { sequence: usize, engine_mutex_available: bool, + /// False when the calling worker's `resource_body_for_uri` had already + /// returned before the peer read `worker_finished` (see `worker_finished` + /// below). A probe taken after the worker returns observes nothing about + /// mutex-holding *during* the pending report — it would trivially see the + /// mutex free regardless of production behavior. Callers must treat such + /// an observation as inconclusive, not as a pass. + probe_valid: bool, } fn run_slow_health_mutex_child() { @@ -472,8 +480,13 @@ fn run_slow_health_mutex_child() { let engine_probe = Arc::new(OnceLock::::new()); 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 + // `ReportObservation::probe_valid`). Reset before each worker starts. + let worker_finished = Arc::new(AtomicBool::new(false)); let peer_info = daemon_info.clone(); let peer_engine_probe = Arc::clone(&engine_probe); + let peer_worker_finished = Arc::clone(&worker_finished); let peer = thread::spawn(move || { run_controlled_health_peer( &listener, @@ -481,6 +494,7 @@ fn run_slow_health_mutex_child() { &peer_engine_probe, &report_seen_tx, &report_release_rx, + &peer_worker_finished, ) }); @@ -527,8 +541,13 @@ fn run_slow_health_mutex_child() { let mut failures = Vec::new(); let worker_server = Arc::clone(&server); + let loss_worker_finished = Arc::clone(&worker_finished); let loss_worker = thread::spawn(move || -> Result { - match worker_server.resource_body_for_uri("hyper://workspace") { + let result = worker_server.resource_body_for_uri("hyper://workspace"); + // Must be set the instant the call returns, before the peer's + // REPORT_HYPERD_ERROR handler reads it — see `ReportObservation::probe_valid`. + loss_worker_finished.store(true, Ordering::Release); + match result { Err(error) => Ok(error.code), Ok(value) => Err(format!( "dead Hyper connection unexpectedly returned resource {value:?}" @@ -539,10 +558,18 @@ fn run_slow_health_mutex_child() { let first_report = receive_and_release_report(&report_seen_rx, &report_release_tx, 1, &mut failures); let loss_worker_result = loss_worker.join(); - if let Some(observation) = first_report - && !observation.engine_mutex_available - { - failures.push("engine mutex was unavailable at the first slow loss report".to_string()); + if let Some(observation) = first_report { + if !observation.probe_valid { + failures.push( + "first report probe is inconclusive: the loss worker's call returned before \ + the peer could observe the engine mutex, so the mutex-availability check below \ + proves nothing (likely a scheduler flake — rerun)" + .to_string(), + ); + } + if !observation.engine_mutex_available { + failures.push("engine mutex was unavailable at the first slow loss report".to_string()); + } } match loss_worker_result { Ok(Ok(ErrorCode::ConnectionLost)) => {} @@ -565,13 +592,18 @@ 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 probes `try_lock` synchronously before releasing that response, - // so this cannot pass merely because a scheduler slept past the 200 ms I/O - // budget. Current production is red here because ensure_engine holds the - // engine mutex throughout Engine::new. + // 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. + worker_finished.store(false, Ordering::Release); let reinit_server = Arc::clone(&server); + let reinit_worker_finished = Arc::clone(&worker_finished); let reinit_worker = thread::spawn(move || -> Result { - match reinit_server.resource_body_for_uri("hyper://workspace") { + let result = reinit_server.resource_body_for_uri("hyper://workspace"); + reinit_worker_finished.store(true, Ordering::Release); + match result { Err(error) => Ok(error.code), Ok(value) => Err(format!( "dead daemon endpoint unexpectedly reinitialized to resource {value:?}" @@ -581,13 +613,21 @@ fn run_slow_health_mutex_child() { let second_report = receive_and_release_report(&report_seen_rx, &report_release_tx, 2, &mut failures); let reinit_worker_result = reinit_worker.join(); - if let Some(observation) = second_report - && !observation.engine_mutex_available - { - failures.push( + if let Some(observation) = second_report { + if !observation.probe_valid { + failures.push( + "second report probe is inconclusive: the reinit worker's call returned before \ + the peer could observe the engine mutex, so the mutex-availability check below \ + proves nothing (likely a scheduler flake — rerun)" + .to_string(), + ); + } + if !observation.engine_mutex_available { + failures.push( "engine mutex was held while post-loss Engine initialization waited on REPORT_HYPERD_ERROR" .to_string(), ); + } } match reinit_worker_result { Ok(Ok(ErrorCode::InternalError)) => {} @@ -672,6 +712,7 @@ fn run_controlled_health_peer( engine_probe: &OnceLock, report_seen_tx: &mpsc::Sender, report_release_rx: &mpsc::Receiver, + worker_finished: &AtomicBool, ) -> Result, String> { let mut commands = Vec::new(); let mut report_sequence = 0_usize; @@ -717,6 +758,12 @@ 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); @@ -731,6 +778,7 @@ fn run_controlled_health_peer( .send(ReportObservation { sequence: report_sequence, engine_mutex_available, + probe_valid, }) .map_err(|error| format!("signal observed REPORT_HYPERD_ERROR: {error}"))?; let released_sequence = report_release_rx From 0a3e93aae2d4bbb26f8e555bb133500c7b51d825 Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Sun, 6 Sep 2026 13:02:32 -0700 Subject: [PATCH 3/4] test(mcp): verify the accepted-connection blocking fix by socket state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit health_listener_waits_for_command_after_accept infers HealthListener's set_nonblocking(false) fix (for accepted sockets inheriting the listener's O_NONBLOCK on BSD-derived kernels — macOS included, not Linux) from read timing. On Linux, accepted sockets are blocking regardless of the listener's mode, so that test passes trivially there even with the fix reverted; it can only ever catch a regression on the BSD kernels that actually have the bug. Extract the accept-and-force-blocking step into accept_and_force_blocking (behavior-preserving refactor — same logic, now unit-testable) and add accept_and_force_blocking_clears_nonblocking_flag, which reads the accepted socket's real O_NONBLOCK flag via fcntl(F_GETFL) instead of inferring it from behavior. Verified by reverting the one-line set_nonblocking(false) fix: the new test fails on this machine with O_NONBLOCK observed set (flags=4 vs expected 0); restoring the fix makes it pass again. Also names the BSD/macOS-vs-Linux platform reason in accept_and_force_blocking's doc comment (previously boilerplate that a future simplifier could plausibly delete), and adds the missing CHANGELOG entry for the underlying fix, merged into the existing Unreleased ### Fixed heading. --- hyperdb-mcp/CHANGELOG.md | 10 +++ hyperdb-mcp/src/daemon/health.rs | 112 ++++++++++++++++++++++++++++--- 2 files changed, 113 insertions(+), 9 deletions(-) diff --git a/hyperdb-mcp/CHANGELOG.md b/hyperdb-mcp/CHANGELOG.md index 42791ca..5ef1182 100644 --- a/hyperdb-mcp/CHANGELOG.md +++ b/hyperdb-mcp/CHANGELOG.md @@ -232,6 +232,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/). `daemon status --port` now probes that exact health port, discovered-daemon error reports target the effective health port, and best-effort health I/O no longer retains the engine mutex. +- **Health-listener connections could be torn down before the client sent its + first byte, on macOS and other BSD-derived kernels.** `HealthListener::bind` + puts the listening socket in non-blocking mode so its accept loop can poll + for shutdown; on BSD kernels (unlike Linux) `accept()` propagates that + `O_NONBLOCK` flag to the accepted socket, so the connection handler's first + `read_line` returned `WouldBlock` in microseconds and closed the connection + before a client had a chance to write a command. Liveness checks, restart + reporting, and heartbeats all depend on this connection surviving long + enough to receive one line, so accepted connections are now explicitly + forced back into blocking mode. - **Attachment contention is actionable for persistent *and* user attaches.** A lock conflict (SQLSTATE `55006`, or a legacy "already attached" / "file is locked" phrase from older hyperd) now returns `RESOURCE_BUSY` with the diff --git a/hyperdb-mcp/src/daemon/health.rs b/hyperdb-mcp/src/daemon/health.rs index cba466b..33825b6 100644 --- a/hyperdb-mcp/src/daemon/health.rs +++ b/hyperdb-mcp/src/daemon/health.rs @@ -136,21 +136,20 @@ impl HealthListener { break; } - match self.listener.accept() { - Ok((stream, _addr)) => { - if let Err(error) = stream.set_nonblocking(false) { - warn!( - error = %error, - "could not make accepted health connection blocking" - ); - continue; - } + match accept_and_force_blocking(&self.listener) { + Ok(AcceptedConnection::Ready(stream)) => { let state = Arc::clone(&state); let info = Arc::clone(&info); std::thread::spawn(move || { handle_client(stream, &state, &info); }); } + Ok(AcceptedConnection::ForceBlockingFailed(error)) => { + warn!( + error = %error, + "could not make accepted health connection blocking" + ); + } Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { // Poll tightly: the doctor network phase budgets only a // few hundred ms for a STATUS round-trip, and on slow CI @@ -169,6 +168,40 @@ impl HealthListener { } } +/// Outcome of [`accept_and_force_blocking`] once a connection has actually +/// been accepted (as opposed to `accept()` itself erroring, which is +/// propagated through the outer `io::Result`). +enum AcceptedConnection { + /// Accepted and confirmed blocking; ready to hand to [`handle_client`]. + Ready(TcpStream), + /// Accepted, but the attempt to clear the non-blocking flag failed. The + /// connection is dropped; the caller logs and moves on to the next + /// `accept()`. + ForceBlockingFailed(std::io::Error), +} + +/// Accept one connection and force it into blocking mode. +/// +/// [`HealthListener::bind`] puts the *listening* socket into non-blocking +/// mode so [`HealthListener::run`]'s loop can poll `should_shutdown` between +/// accepts. On BSD-derived kernels — macOS and other BSDs, but **not** +/// Linux, which keeps a newly accepted socket's blocking mode independent of +/// the listener's — `accept()` propagates the listening socket's +/// `O_NONBLOCK` flag to the accepted socket. Left non-blocking, the accepted +/// stream would return `WouldBlock` from `read_line` in +/// [`handle_client`] in microseconds — typically before the client has even +/// written its first byte — tearing the connection down before it ever +/// received a command. `set_nonblocking(false)` below undoes that +/// propagation unconditionally, which is a no-op (not a bug) on platforms +/// that never had the problem. +fn accept_and_force_blocking(listener: &TcpListener) -> std::io::Result { + let (stream, _addr) = listener.accept()?; + match stream.set_nonblocking(false) { + Ok(()) => Ok(AcceptedConnection::Ready(stream)), + Err(error) => Ok(AcceptedConnection::ForceBlockingFailed(error)), + } +} + fn status_json(info: &Mutex) -> String { let snapshot = info.lock().expect("DaemonInfo mutex poisoned").clone(); match DaemonRecord::with_current_identity(&snapshot) { @@ -703,4 +736,65 @@ mod tests { "newline-free health responses beyond the 64 KiB protocol limit must be rejected; {outcome}" ); } + + /// Asserts the accepted socket's actual blocking state via `fcntl`, + /// rather than inferring it from read-timing behavior. On Linux an + /// accepted socket is blocking regardless of the listener's mode, so a + /// behavioral test (send late, expect it to still be read) passes + /// trivially there even with `accept_and_force_blocking`'s + /// `set_nonblocking(false)` reverted — it would only catch a regression + /// on the BSD-derived kernels (macOS and other BSDs) that actually + /// propagate `O_NONBLOCK` to accepted sockets. Reading the `O_NONBLOCK` + /// flag directly makes the test verify the real contract everywhere, + /// rather than a platform-dependent behavioral proxy for it. + #[cfg(unix)] + #[test] + fn accept_and_force_blocking_clears_nonblocking_flag() { + use std::os::unix::io::AsRawFd; + + let listener = HealthListener::bind(0).expect("bind test health listener"); + let port = listener.port; + + let client = std::thread::spawn(move || { + // Held for the duration of the accept below; dropped (and thus + // closed) only once this thread returns. + std::net::TcpStream::connect(("127.0.0.1", port)).expect("connect test client") + }); + + let accept_deadline = Instant::now() + Duration::from_secs(2); + let accepted = loop { + match accept_and_force_blocking(&listener.listener) { + Ok(AcceptedConnection::Ready(stream)) => break stream, + Ok(AcceptedConnection::ForceBlockingFailed(error)) => { + panic!("force-blocking the accepted test connection failed: {error}") + } + Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < accept_deadline, + "timed out waiting to accept the test client connection" + ); + std::thread::sleep(Duration::from_millis(2)); + } + Err(error) => panic!("accept failed: {error}"), + } + }; + client.join().expect("test client thread must not panic"); + + // SAFETY: `accepted` is a live, owned, valid socket for the duration + // of this call; `F_GETFL` only reads flags and mutates nothing. + let flags = unsafe { libc::fcntl(accepted.as_raw_fd(), libc::F_GETFL) }; + assert!( + flags >= 0, + "fcntl(F_GETFL) on the accepted test connection failed: {}", + std::io::Error::last_os_error() + ); + assert_eq!( + flags & libc::O_NONBLOCK, + 0, + "accepted health connection must be blocking (O_NONBLOCK must be clear); \ + reverting accept_and_force_blocking's set_nonblocking(false) call would \ + leave O_NONBLOCK set on BSD-derived kernels that propagate it from the \ + listening socket" + ); + } } From 8b28ecdeeb163a6513ff6da81dc776f793ddab48 Mon Sep 17 00:00:00 2001 From: Stefan Steiner Date: Sun, 6 Sep 2026 13:02:39 -0700 Subject: [PATCH 4/4] test(api): retry past a torn PID-file read instead of failing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The child in callback_connection_shutdowns_hyperd_after_parent_kill reports its hyperd PID via fs::write, which is File::create (truncate or create) then write_all — not atomic. wait_for_reported_pid already retried on NotFound, but treated a successful read of unparseable content as fatal, so a parent poll landing in the create/write_all gap made "".parse::() panic the whole test instead of retrying. Treat an empty or unparseable read the same as 'not created yet' and keep polling to the deadline, only surfacing an error if the file never became parseable in time. Adds wait_for_reported_pid_retries_past_a_torn_write, which reproduces the exact non-atomic sequence (create, sleep, then write real content) against a stretched-out empty window. Verified by reverting the fix: the new test fails with the exact reported symptom (invalid PID report "": cannot parse integer from empty string); restoring the fix makes it pass again. --- hyperdb-api/tests/process_tests.rs | 61 ++++++++++++++++++++++++++---- 1 file changed, 54 insertions(+), 7 deletions(-) diff --git a/hyperdb-api/tests/process_tests.rs b/hyperdb-api/tests/process_tests.rs index 280a21c..1f95d06 100644 --- a/hyperdb-api/tests/process_tests.rs +++ b/hyperdb-api/tests/process_tests.rs @@ -72,24 +72,71 @@ fn callback_connection_shutdowns_hyperd_after_parent_kill() { fn wait_for_reported_pid(pid_file: &std::path::Path, timeout: Duration) -> Result { let deadline = std::time::Instant::now() + timeout; + let mut last_unparseable: Option = None; loop { match fs::read_to_string(pid_file) { - Ok(contents) => { - return contents - .trim() - .parse() - .map_err(|error| format!("invalid PID report {contents:?}: {error}")); - } + 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. + last_unparseable = Some(contents); + } + }, Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} Err(error) => return Err(format!("could not read PID report: {error}")), } if std::time::Instant::now() >= deadline { - return Err(format!("no PID report appeared at {}", pid_file.display())); + return Err(match last_unparseable { + Some(contents) => format!( + "PID report at {} never became parseable before the deadline; last read {contents:?}", + pid_file.display() + ), + None => format!("no PID report appeared at {}", pid_file.display()), + }); } thread::sleep(Duration::from_millis(20)); } } +/// 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. +#[test] +fn wait_for_reported_pid_retries_past_a_torn_write() { + let temp_dir = tempfile::tempdir().expect("create temp dir for torn-write simulation"); + let pid_file = temp_dir.path().join("hyperd-pid"); + + 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. + 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"); + }); + + let result = wait_for_reported_pid(&pid_file, Duration::from_secs(5)); + writer + .join() + .expect("torn-write simulation thread must not panic"); + + assert_eq!( + result, + Ok(4242), + "a read landing in the create/write_all gap must be retried, not treated as a fatal parse error" + ); +} + fn bounded_process_exit_poll(pid: u32, timeout: Duration) -> bool { let deadline = std::time::Instant::now() + timeout; loop {