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 { 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" + ); + } } 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(); 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