From 42d556bae1c6b8618a56960b5d19fdf77829f24c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 01:57:12 +0000 Subject: [PATCH 1/2] Make child shutdown deterministic: terminal report precedes any respawn ack ChildServer::shutdown now kills the child, drops the stdin channel, and JOINS the stdout pump (2s bound, abort fallback) instead of aborting it blindly - so the old child's terminal server_offline is queued on the outbound channel before shutdown returns, and every respawn path in the supervisor awaits shutdown before try_spawn sends the replacement's server_spawn_result. This turns the backend's spawn-ack error clearing from a heuristic into a rule, specced as PROTOCOL.md T-74. Two new tests pin the guarantee at both seams (pump join, offline-before-ack on the outbound channel) and fail against the old blind-abort behavior. Also moves daemon.rs's inline test module to a sibling daemon_tests.rs per crate convention. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NgN7YdfLT41kWnWqe1MPZa --- crates/stdiod/PROTOCOL.md | 14 ++- .../edison-stdiod/src/daemon_reconcile.rs | 13 +++ .../crates/edison-stdiod/src/daemon_tests.rs | 54 +++++++++- .../stdiod/crates/edison-stdiod/src/proc.rs | 99 ++++++++++++++----- .../crates/edison-stdiod/src/proc_tests.rs | 46 +++++++++ 5 files changed, 200 insertions(+), 26 deletions(-) diff --git a/crates/stdiod/PROTOCOL.md b/crates/stdiod/PROTOCOL.md index c63a0e60..353ea0e6 100644 --- a/crates/stdiod/PROTOCOL.md +++ b/crates/stdiod/PROTOCOL.md @@ -306,6 +306,18 @@ pump EOF, input pump write error, a frame addressed to an exited child). The reference client uses a one-shot latch. *Source: `proc.rs::ChildDiagnostics::take_terminal_error` (`reported.swap`).* +**T-74** When a client stops a child and then reports a spawn result for the +same `server_id` (a kill-and-respawn: env update, spec update, a changed +desired spec, or a restart of an unresponsive child), the terminal +`server_offline` for the old child MUST be sent before the new +`server_spawn_result`. Receivers MAY therefore treat +`server_spawn_result{ok:true}` as clearing any terminal error they hold for +that `server_id`. See T-42. +*Source: `proc.rs::ChildServer::shutdown` (joins the stdout pump, so the +terminal report is queued before it returns); `daemon.rs::Supervisor::try_spawn` +(every respawn path awaits `shutdown` first); `registry.py::_dispatch_inbound` +(`ServerSpawnResult` arm clears `last_error_by_server`).* + **T-44** A spawn failure MUST produce **both** `server_spawn_result{ok:false}` and `tunnel_error{code:"spawn_failed", server_id}`, in that order. The backend treats `spawn_failed` and `server_offline` as terminal for the current child: @@ -361,7 +373,7 @@ carry on) rather than failing the frame. | Code | Direction | Meaning | Governed by | |------|-----------|---------|-------------| -| `server_offline` | client → backend | The child for `server_id` is gone: its output pump hit EOF or errored, its input pump failed to write, or a frame arrived for a child already known dead. Terminal for that child, and emitted at most once per child lifetime | T-42, T-43, T-47 | +| `server_offline` | client → backend | The child for `server_id` is gone: its output pump hit EOF or errored, its input pump failed to write, or a frame arrived for a child already known dead. Terminal for that child, and emitted at most once per child lifetime | T-42, T-43, T-47, T-74 | | `spawn_failed` | client → backend | The client tried to start `server_id` and could not (binary missing, exec refused, module URI unrecognised). Always accompanies a `server_spawn_result{ok:false}` and follows it | T-44 | | `server_unresponsive` | client → backend | The child for `server_id` stopped consuming its input: the per-child queue filled, so this request could not be delivered. The client then kills and respawns the child | T-51 | | `stdio_tunnel_disabled` | backend → client | Device-wide (`server_id` null): the org has the `stdio_tunnel_enabled` feature flag off. The backend closes with 1008 straight after | T-07, T-49 | diff --git a/crates/stdiod/crates/edison-stdiod/src/daemon_reconcile.rs b/crates/stdiod/crates/edison-stdiod/src/daemon_reconcile.rs index 541d200d..edc332fc 100644 --- a/crates/stdiod/crates/edison-stdiod/src/daemon_reconcile.rs +++ b/crates/stdiod/crates/edison-stdiod/src/daemon_reconcile.rs @@ -60,6 +60,19 @@ impl Supervisor { /// Spawn one raw desired server after applying current local values. Reports /// the spawn result and retains placeholders so later respawns re-enrich cleanly. + /// + /// Ordering contract (PROTOCOL.md T-74): every kill-and-respawn path here + /// (`apply_snapshot`, `apply_delta`, `apply_spec_update`, + /// `apply_env_update`, `restart_unresponsive`) MUST `await` + /// [`ChildServer::shutdown`] for the outgoing child *before* calling this, + /// and this method's `server_spawn_result` send must stay after the + /// spawn. That is what makes the old child's terminal `server_offline` + /// reach the outbound channel first: `shutdown` returns only once the + /// stdout pump's report has been queued, and the channel behind + /// [`OutgoingHandle`] preserves the order in which sends complete, so the + /// WS writer emits the two frames in that order. The backend relies on it + /// to treat a successful ack as clearing a stored terminal error + /// (`registry.py::_dispatch_inbound`). async fn try_spawn(&mut self, raw: DesiredServer) { let server_id = raw.server_id.clone(); let sensitive_arg_values = self diff --git a/crates/stdiod/crates/edison-stdiod/src/daemon_tests.rs b/crates/stdiod/crates/edison-stdiod/src/daemon_tests.rs index 0732fd4f..8c980b1e 100644 --- a/crates/stdiod/crates/edison-stdiod/src/daemon_tests.rs +++ b/crates/stdiod/crates/edison-stdiod/src/daemon_tests.rs @@ -23,8 +23,16 @@ fn env_store_for(test: &str) -> EnvStore { } fn supervisor_with(test: &str, children: HashMap) -> Supervisor { + supervisor_with_outgoing(test, children, OutgoingHandle::new()) +} + +fn supervisor_with_outgoing( + test: &str, + children: HashMap, + outgoing: OutgoingHandle, +) -> Supervisor { let mut supervisor = Supervisor::new( - OutgoingHandle::new(), + outgoing, StateWriter::new(State::default()), env_store_for(test), ); @@ -91,6 +99,50 @@ async fn snapshot_reports_an_exited_child_as_crashed() { assert!(matches!(entries[0].state, ServerStatus::Crashed)); } +/// PROTOCOL.md T-74: a kill-and-respawn MUST put the old child's terminal +/// `server_offline` on the outbound channel before the replacement's +/// `server_spawn_result`, so the backend can treat a successful ack as +/// clearing the stored error. The outbound channel is the observation point +/// the WS writer drains in order, so frame order here is wire order. +/// +/// Both frames are read with `try_recv`: they must already be queued by the +/// time the respawn path returns, not merely arrive eventually. +#[tokio::test] +async fn respawn_queues_terminal_offline_before_the_spawn_ack() { + let spec = desired("filesystem", "sleep 30"); + let outgoing = OutgoingHandle::new(); + let (wire_tx, mut wire_rx) = mpsc::channel(8); + outgoing.set(wire_tx); + let child = ChildServer::spawn(&spec, &spec, outgoing.clone(), Vec::new(), None).unwrap(); + let mut supervisor = supervisor_with_outgoing( + "respawn-order", + HashMap::from([("filesystem".to_string(), child)]), + outgoing, + ); + + supervisor.restart_unresponsive("filesystem").await; + + let TunnelFrame::TunnelError(error) = wire_rx + .try_recv() + .expect("the old child's terminal error should already be queued") + else { + panic!("first frame should be the old child's terminal tunnel_error"); + }; + assert_eq!(error.code, "server_offline"); + assert_eq!(error.server_id.as_deref(), Some("filesystem")); + + let TunnelFrame::ServerSpawnResult(result) = wire_rx + .try_recv() + .expect("the replacement's spawn ack should follow it") + else { + panic!("second frame should be the replacement's server_spawn_result"); + }; + assert!(result.ok, "the replacement should have spawned"); + assert_eq!(result.server_id, "filesystem"); + + supervisor.shutdown_children().await; +} + /// A child whose stdin has broken is terminal for MCP - the backend gets its /// `server_offline` - but the process may still be running, and the snapshot /// must not call a live PID crashed. It stays `running` until the supervisor diff --git a/crates/stdiod/crates/edison-stdiod/src/proc.rs b/crates/stdiod/crates/edison-stdiod/src/proc.rs index 4056b3d9..6a354f16 100644 --- a/crates/stdiod/crates/edison-stdiod/src/proc.rs +++ b/crates/stdiod/crates/edison-stdiod/src/proc.rs @@ -32,6 +32,11 @@ use crate::child_diagnostics::ChildDiagnostics; use crate::state::StateWriter; use crate::tunnel::OutgoingHandle; +/// How long [`ChildServer::shutdown`] waits for the stdout pump to finish its +/// terminal report before falling back to aborting it. See the comment at the +/// join site for why this is a join rather than an abort. +const PUMP_JOIN_TIMEOUT: Duration = Duration::from_secs(2); + /// Build the base `Command` for a child MCP server. /// /// On Unix this is just `Command::new(program).args(args)` - the inherited PATH @@ -485,34 +490,80 @@ impl ChildServer { self.diagnostics.has_observed_exit() } - /// Kill the child and abort the pumps. + /// Kill the child and wind the pumps down. + /// + /// Returns only once the stdout pump has run its terminal + /// [`report_terminal`] to completion, so by the time this returns the + /// child's `server_offline` is already queued on the outbound channel + /// (or the one-shot latch says it will never be sent). Callers that + /// respawn the same `server_id` therefore cannot enqueue the new child's + /// `server_spawn_result` ahead of the old child's terminal error - see + /// PROTOCOL.md T-74 and `daemon.rs::Supervisor::try_spawn`. pub async fn shutdown(self) { - let mut child = self.child.lock().await; - if let Some(pid) = child.id() { - #[cfg(unix)] - { - let _ = Command::new("kill") - .args(["-KILL", "--", &format!("-{pid}")]) - .status() - .await; - } - #[cfg(windows)] - { - let _ = Command::new("taskkill") - .args(["/PID", &pid.to_string(), "/T", "/F"]) - .status() - .await; + let Self { + server_id, + child, + outbound_tx, + stdin_pump, + mut stdout_pump, + stderr_pump, + diagnostics, + .. + } = self; + { + let mut guard = child.lock().await; + if let Some(pid) = guard.id() { + #[cfg(unix)] + { + let _ = Command::new("kill") + .args(["-KILL", "--", &format!("-{pid}")]) + .status() + .await; + } + #[cfg(windows)] + { + let _ = Command::new("taskkill") + .args(["/PID", &pid.to_string(), "/T", "/F"]) + .status() + .await; + } } + let _ = guard.start_kill(); + let _ = guard.wait().await; + // Released before joining the stdout pump: its terminal report + // reads the exit status through this same lock and would + // otherwise deadlock until the timeout below. } - let _ = child.start_kill(); - let _ = child.wait().await; - // Reaped here, so the process is observably gone even if no pump + // Reaped just above, so the process is observably gone even if no pump // ever managed to read an exit status for it. - self.diagnostics.mark_observed_exit(); - drop(child); - self.stdin_pump.abort(); - self.stdout_pump.abort(); - self.stderr_pump.abort(); + diagnostics.mark_observed_exit(); + + // Closing the frame channel ends the stdin pump at its `recv`. It is + // aborted below regardless; this just makes the common case a clean + // exit rather than a cancellation. + drop(outbound_tx); + + // Join, don't abort: the stdout pump's last act is the terminal + // `server_offline` report, and aborting it mid-flight is what made + // the ordering against a subsequent respawn a coin flip. The dead + // child's pipe is already at EOF, so the pump only has to finish + // `report_terminal`: at most ~100ms waiting for stderr to drain, + // ~100ms polling for the exit status, then one channel send. Two + // seconds is an order of magnitude of headroom over that, and it + // still bounds the pathological case where the send blocks because + // the outbound channel is full and the WS writer is wedged. + if tokio::time::timeout(PUMP_JOIN_TIMEOUT, &mut stdout_pump) + .await + .is_err() + { + warn!( + server_id = %server_id, + "stdout pump did not finish reporting within the shutdown budget; aborting it", + ); + stdout_pump.abort(); + } + stdin_pump.abort(); + stderr_pump.abort(); } } diff --git a/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs b/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs index a55d8015..8567bee1 100644 --- a/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs +++ b/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs @@ -109,6 +109,52 @@ async fn broken_stdin_replays_actionable_terminal_error() { ); } +/// `shutdown` must not return until the stdout pump has finished reporting +/// the child's death: the terminal `server_offline` is read here with a +/// non-blocking `try_recv`, so it can only pass if the frame was queued +/// before `shutdown` returned. That is what lets a caller respawn the same +/// `server_id` without racing its `server_spawn_result` ahead of this frame +/// (PROTOCOL.md T-74). +#[cfg(unix)] +#[tokio::test] +async fn shutdown_returns_only_after_the_terminal_report_is_queued() { + let desired = DesiredServer { + server_id: "server".into(), + name: "server".into(), + command: "/bin/sh".into(), + // Silent and long-lived: nothing reaches the wire until the kill + // closes stdout, so anything received afterwards is the pump's + // terminal report. + args: vec!["-c".into(), "sleep 30".into()], + env: Default::default(), + working_dir: None, + enabled: true, + }; + let outgoing = OutgoingHandle::new(); + let (wire_tx, mut wire_rx) = mpsc::channel(4); + outgoing.set(wire_tx); + let child = ChildServer::spawn(&desired, &desired, outgoing, Vec::new(), None).unwrap(); + assert!( + wire_rx.try_recv().is_err(), + "a live child should not have reported anything yet" + ); + + child.shutdown().await; + + let frame = wire_rx + .try_recv() + .expect("terminal report should be queued before shutdown returns"); + let TunnelFrame::TunnelError(error) = frame else { + panic!("expected terminal tunnel error"); + }; + assert_eq!(error.code, "server_offline"); + assert_eq!(error.server_id.as_deref(), Some("server")); + assert!( + wire_rx.try_recv().is_err(), + "the terminal report is one-shot per child" + ); +} + #[cfg(unix)] #[tokio::test] async fn exited_process_reports_final_stderr_once() { From 8404f81d233e7955cc5fb2d7db6be86858d7793c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 02:33:52 +0000 Subject: [PATCH 2/2] Make the terminal report survive a pump that has to be aborted Review follow-ups on the shutdown-ordering work. - shutdown joins both frame pumps under one shutdown budget instead of joining stdout and aborting stdin. Either pump can hold the one-shot terminal report: stdout takes it at EOF, stdin takes it when a write to the child fails. Aborting the stdin pump while it was mid-send dropped the report and left the respawn ack with nothing in front of it, which is the hang T-42 exists to prevent. Sequential joins share one deadline so the total wait is unchanged; stderr is still aborted outright. - When the budget does run out, shutdown produces the terminal report itself rather than leaving it stranded in the aborted pump. The latch that hands out the report is consumed by the pump that took it, so nothing else can ever send it once that pump is gone. Semantics of the fallback, since the interesting case is a pump that was cancelled with a send in flight: - report_terminal flips a report_sent flag immediately after the send completes, with no await in between, so cancellation cannot land between queueing the frame and recording it. An aborted pump has therefore either sent the report or left it unsent, never an unknown. - abort only requests cancellation, so shutdown waits for the pump to actually stop (200ms grace) before deciding. A pump that will not cancel in that window is left alone and no fallback is sent: at most once is preserved, and the backend's staleness teardown fails the in-flight calls. - The fallback send is try_send, so a wedged outbound channel cannot block shutdown. A full channel means the report is dropped, which the same staleness teardown covers. PROTOCOL.md T-74 said the ordering guarantee rested on joining the stdout pump. It now states the guarantee as ordering rather than delivery, names the fallback, and records why T-43's at-most-once still holds across it. Tests: two shutdown tests wedge the outbound channel so a pump parks mid-report, one with stdout holding the report and one with stdin, and assert exactly one server_offline reaches the wire. The stdin one fails against a shutdown that joins only stdout; both fail without the fallback. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NgN7YdfLT41kWnWqe1MPZa --- crates/stdiod/PROTOCOL.md | 16 +- .../edison-stdiod/src/child_diagnostics.rs | 12 ++ .../stdiod/crates/edison-stdiod/src/proc.rs | 132 +++++++++++----- .../crates/edison-stdiod/src/proc_tests.rs | 142 +++++++++++++++--- .../stdiod/crates/edison-stdiod/src/tunnel.rs | 10 ++ 5 files changed, 248 insertions(+), 64 deletions(-) diff --git a/crates/stdiod/PROTOCOL.md b/crates/stdiod/PROTOCOL.md index 353ea0e6..cbed5962 100644 --- a/crates/stdiod/PROTOCOL.md +++ b/crates/stdiod/PROTOCOL.md @@ -313,8 +313,20 @@ desired spec, or a restart of an unresponsive child), the terminal `server_spawn_result`. Receivers MAY therefore treat `server_spawn_result{ok:true}` as clearing any terminal error they hold for that `server_id`. See T-42. -*Source: `proc.rs::ChildServer::shutdown` (joins the stdout pump, so the -terminal report is queued before it returns); `daemon.rs::Supervisor::try_spawn` + +This is an ordering requirement, not a delivery guarantee. A client whose +outbound channel cannot accept the report (full, or disconnected mid-shutdown) +MAY drop it, in which case the backend sees no `server_offline` at all and +falls back to its own staleness handling (T-50); what a client MUST NOT do is +let the report arrive after the ack. The reference client bounds its own +shutdown: it joins both frame pumps under one budget, since either may hold the +one-shot report, and if it has to abort a pump that has taken the report it +sends the report itself, non-blocking, before returning. T-43 still holds +across that fallback: the latch that hands out the report and the flag that +records a completed send are updated without an await between them, so an +aborted pump has either sent the report or left it unsent, never both. +*Source: `proc.rs::ChildServer::shutdown` (joins both pumps, with a fallback +send after an abort); `daemon.rs::Supervisor::try_spawn` (every respawn path awaits `shutdown` first); `registry.py::_dispatch_inbound` (`ServerSpawnResult` arm clears `last_error_by_server`).* diff --git a/crates/stdiod/crates/edison-stdiod/src/child_diagnostics.rs b/crates/stdiod/crates/edison-stdiod/src/child_diagnostics.rs index 9f030f60..3f7dd8bf 100644 --- a/crates/stdiod/crates/edison-stdiod/src/child_diagnostics.rs +++ b/crates/stdiod/crates/edison-stdiod/src/child_diagnostics.rs @@ -36,6 +36,13 @@ pub(crate) struct ChildDiagnostics { /// as crashed next to its own live PID. observed_exit: Arc, reported: Arc, + /// Set once a terminal report has actually been queued on the outbound + /// channel. `reported` says a path took ownership of the report; + /// this says the wire has it. Nothing awaits between the send completing + /// and this store, so a cancelled pump can never leave the pair + /// disagreeing, which is what lets `shutdown` decide whether it has to + /// send the report itself. + pub(crate) report_sent: Arc, /// One-shot guard so a single death produces a single `crashed` write, /// whichever pump gets there first. crash_published: Arc, @@ -87,6 +94,11 @@ impl ChildDiagnostics { self.observed_exit.load(Ordering::Acquire) } + /// Whether a terminal report has reached the outbound channel. + pub(crate) fn report_sent(&self) -> bool { + self.report_sent.load(Ordering::Acquire) + } + /// Flip this child's `state.json` entry to `crashed`, once, and only for /// a process that was seen to exit. /// diff --git a/crates/stdiod/crates/edison-stdiod/src/proc.rs b/crates/stdiod/crates/edison-stdiod/src/proc.rs index 6a354f16..d0e0b980 100644 --- a/crates/stdiod/crates/edison-stdiod/src/proc.rs +++ b/crates/stdiod/crates/edison-stdiod/src/proc.rs @@ -32,11 +32,16 @@ use crate::child_diagnostics::ChildDiagnostics; use crate::state::StateWriter; use crate::tunnel::OutgoingHandle; -/// How long [`ChildServer::shutdown`] waits for the stdout pump to finish its -/// terminal report before falling back to aborting it. See the comment at the -/// join site for why this is a join rather than an abort. +/// How long [`ChildServer::shutdown`] waits for the frame pumps to finish +/// their terminal report before falling back to aborting them. Shared by both +/// joins. See the comment at the join site for why this is a join rather than +/// an abort. const PUMP_JOIN_TIMEOUT: Duration = Duration::from_secs(2); +/// How long [`ChildServer::shutdown`] waits for an aborted pump to actually +/// stop before deciding whether it has to send the terminal report itself. +const PUMP_ABORT_GRACE: Duration = Duration::from_millis(200); + /// Build the base `Command` for a child MCP server. /// /// On Unix this is just `Command::new(program).args(args)` - the inherited PATH @@ -357,6 +362,9 @@ pub struct ChildServer { /// behind an async lock. pub pid: Option, pub outbound_tx: mpsc::Sender, + /// Kept so [`shutdown`](Self::shutdown) can emit the terminal report + /// itself when it has to abort the pump that would have sent it. + tunnel_outgoing: OutgoingHandle, pub stdin_pump: JoinHandle<()>, pub stdout_pump: JoinHandle<()>, stderr_pump: JoinHandle<()>, @@ -432,7 +440,7 @@ impl ChildServer { let stdout_pump = tokio::spawn(stdout_pump( enriched.server_id.clone(), stdout, - tunnel_outgoing, + tunnel_outgoing.clone(), diagnostics.clone(), Some(child.clone()), )); @@ -456,6 +464,7 @@ impl ChildServer { child, pid, outbound_tx, + tunnel_outgoing, stdin_pump, stdout_pump, stderr_pump, @@ -492,25 +501,27 @@ impl ChildServer { /// Kill the child and wind the pumps down. /// - /// Returns only once the stdout pump has run its terminal - /// [`report_terminal`] to completion, so by the time this returns the - /// child's `server_offline` is already queued on the outbound channel - /// (or the one-shot latch says it will never be sent). Callers that - /// respawn the same `server_id` therefore cannot enqueue the new child's - /// `server_spawn_result` ahead of the old child's terminal error - see - /// PROTOCOL.md T-74 and `daemon.rs::Supervisor::try_spawn`. + /// Returns only once the child's terminal `server_offline` has been queued + /// on the outbound channel, so a caller that respawns the same `server_id` + /// cannot enqueue the new child's `server_spawn_result` ahead of it - see + /// PROTOCOL.md T-74 and `daemon.rs::Supervisor::try_spawn`. Both frame + /// pumps are joined under one budget, because either of them can be the + /// one holding the terminal report. If the budget runs out, the report is + /// sent from here instead so that aborting a wedged pump cannot swallow it + /// (the exact hang T-42 exists to prevent). pub async fn shutdown(self) { let Self { server_id, child, outbound_tx, - stdin_pump, + mut stdin_pump, mut stdout_pump, stderr_pump, + tunnel_outgoing, diagnostics, .. } = self; - { + let status = { let mut guard = child.lock().await; if let Some(pid) = guard.id() { #[cfg(unix)] @@ -529,41 +540,80 @@ impl ChildServer { } } let _ = guard.start_kill(); - let _ = guard.wait().await; - // Released before joining the stdout pump: its terminal report - // reads the exit status through this same lock and would - // otherwise deadlock until the timeout below. - } + let status = guard.wait().await.ok(); + // Released before joining the pumps: a terminal report reads the + // exit status through this same lock and would otherwise deadlock + // until the budget below runs out. + status + }; // Reaped just above, so the process is observably gone even if no pump // ever managed to read an exit status for it. diagnostics.mark_observed_exit(); - // Closing the frame channel ends the stdin pump at its `recv`. It is - // aborted below regardless; this just makes the common case a clean - // exit rather than a cancellation. + // Closing the frame channel ends the stdin pump at its `recv`, so the + // common case is a clean exit rather than a cancellation. drop(outbound_tx); - // Join, don't abort: the stdout pump's last act is the terminal - // `server_offline` report, and aborting it mid-flight is what made - // the ordering against a subsequent respawn a coin flip. The dead - // child's pipe is already at EOF, so the pump only has to finish - // `report_terminal`: at most ~100ms waiting for stderr to drain, - // ~100ms polling for the exit status, then one channel send. Two - // seconds is an order of magnitude of headroom over that, and it - // still bounds the pathological case where the send blocks because - // the outbound channel is full and the WS writer is wedged. - if tokio::time::timeout(PUMP_JOIN_TIMEOUT, &mut stdout_pump) - .await - .is_err() - { + // Join, don't abort: a pump's last act is the terminal + // `server_offline` report, and aborting it mid-flight is what made the + // ordering against a subsequent respawn a coin flip. Both pumps are + // joined because either can own that report - stdout on EOF, stdin on + // a failed write - and a one-shot latch means the one that got there + // first is the only one that will ever send it. The dead child's pipes + // are already at EOF, so the work left is `report_terminal`: at most + // ~100ms waiting for stderr to drain, ~100ms polling for the exit + // status, then one channel send. The budget is shared by both joins + // and bounds the pathological case where the send blocks because the + // outbound channel is full and the WS writer is wedged. + let deadline = tokio::time::Instant::now() + PUMP_JOIN_TIMEOUT; + let mut report_was_cancelled = false; + for (which, pump) in [("stdout", &mut stdout_pump), ("stdin", &mut stdin_pump)] { + if tokio::time::timeout_at(deadline, &mut *pump).await.is_ok() { + continue; + } warn!( server_id = %server_id, - "stdout pump did not finish reporting within the shutdown budget; aborting it", + pump = which, + "pump did not finish within the shutdown budget; aborting it", ); - stdout_pump.abort(); + pump.abort(); + // `abort` only requests cancellation, which lands at the task's + // next await point. Wait for it: if the pump were still runnable + // it could complete a send between the check below and the + // fallback, and the backend would see two terminal errors for one + // child. When cancellation does not land in the grace period we + // leave the report alone rather than risk that duplicate - the + // backend's staleness teardown fails the calls either way. + if tokio::time::timeout(PUMP_ABORT_GRACE, &mut *pump) + .await + .is_ok() + { + report_was_cancelled = true; + } } - stdin_pump.abort(); stderr_pump.abort(); + + // The pump that owned the report was cancelled before it queued one, + // so send it from here. `reported` is consumed by whichever path took + // the error and `report_sent` only flips once a send has completed, so + // between them they say exactly what still has to go out: nothing when + // a send got through, a fresh error when no path ever took one, and a + // reconstruction when a path took one and died before sending it. + // Non-blocking on purpose - shutdown must not block on a wedged + // channel, and a full channel means the session is being torn down + // anyway. + if report_was_cancelled && !diagnostics.report_sent() { + let error = diagnostics + .take_terminal_error(&server_id, status.as_ref()) + .unwrap_or_else(|| diagnostics.terminal_error(&server_id, status.as_ref())); + if !tunnel_outgoing.try_send(TunnelFrame::TunnelError(error)) { + warn!( + server_id = %server_id, + "could not queue the terminal report during shutdown; the backend will \ + fail in-flight calls when the session goes stale", + ); + } + } } } @@ -642,7 +692,13 @@ async fn report_terminal( let error = diagnostics.take_terminal_error(server_id, status.as_ref()); diagnostics.publish_crashed(server_id).await; if let Some(error) = error { - tunnel_outgoing.send(TunnelFrame::TunnelError(error)).await; + if tunnel_outgoing.send(TunnelFrame::TunnelError(error)).await { + // No await between the send completing and this store, so a pump + // cancelled at any point either queued the frame and recorded it + // or did neither. `shutdown` relies on that to avoid both a lost + // report and a duplicate one. + diagnostics.report_sent.store(true, Ordering::Release); + } } } diff --git a/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs b/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs index 8567bee1..f088c2e5 100644 --- a/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs +++ b/crates/stdiod/crates/edison-stdiod/src/proc_tests.rs @@ -1,10 +1,25 @@ use super::*; +use edison_tunnel_protocol::Ping; use serde_json::json; use tokio::io::duplex; use crate::child_diagnostics::mark_entry_crashed; use crate::state::{ServerEntry, ServerStatus}; +/// A `/bin/sh -c