diff --git a/crates/stdiod/PROTOCOL.md b/crates/stdiod/PROTOCOL.md index c63a0e60..cbed5962 100644 --- a/crates/stdiod/PROTOCOL.md +++ b/crates/stdiod/PROTOCOL.md @@ -306,6 +306,30 @@ 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. + +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`).* + **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 +385,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/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/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..d0e0b980 100644 --- a/crates/stdiod/crates/edison-stdiod/src/proc.rs +++ b/crates/stdiod/crates/edison-stdiod/src/proc.rs @@ -32,6 +32,16 @@ use crate::child_diagnostics::ChildDiagnostics; use crate::state::StateWriter; use crate::tunnel::OutgoingHandle; +/// 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 @@ -352,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<()>, @@ -427,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()), )); @@ -451,6 +464,7 @@ impl ChildServer { child, pid, outbound_tx, + tunnel_outgoing, stdin_pump, stdout_pump, stderr_pump, @@ -485,34 +499,121 @@ 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 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 mut child = self.child.lock().await; - if let Some(pid) = child.id() { - #[cfg(unix)] - { - let _ = Command::new("kill") - .args(["-KILL", "--", &format!("-{pid}")]) - .status() - .await; + let Self { + server_id, + child, + outbound_tx, + 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)] + { + 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; + } } - #[cfg(windows)] + let _ = guard.start_kill(); + 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`, so the + // common case is a clean exit rather than a cancellation. + drop(outbound_tx); + + // 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, + pump = which, + "pump did not finish within the shutdown budget; aborting it", + ); + 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() { - let _ = Command::new("taskkill") - .args(["/PID", &pid.to_string(), "/T", "/F"]) - .status() - .await; + report_was_cancelled = true; + } + } + 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", + ); } } - let _ = child.start_kill(); - let _ = child.wait().await; - // Reaped here, 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(); } } @@ -591,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 a55d8015..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