diff --git a/crates/stdiod/PROTOCOL.md b/crates/stdiod/PROTOCOL.md index 14b3b51..808e97d 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/sealgate-stdiod/src/child_diagnostics.rs b/crates/stdiod/crates/sealgate-stdiod/src/child_diagnostics.rs index 7926d23..47f08e7 100644 --- a/crates/stdiod/crates/sealgate-stdiod/src/child_diagnostics.rs +++ b/crates/stdiod/crates/sealgate-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/sealgate-stdiod/src/proc.rs b/crates/stdiod/crates/sealgate-stdiod/src/proc.rs index 0ec1eb8..f4db0dd 100644 --- a/crates/stdiod/crates/sealgate-stdiod/src/proc.rs +++ b/crates/stdiod/crates/sealgate-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 `supervisor.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/sealgate-stdiod/src/proc_tests.rs b/crates/stdiod/crates/sealgate-stdiod/src/proc_tests.rs index a55d801..a29a765 100644 --- a/crates/stdiod/crates/sealgate-stdiod/src/proc_tests.rs +++ b/crates/stdiod/crates/sealgate-stdiod/src/proc_tests.rs @@ -1,10 +1,25 @@ use super::*; +use sealgate_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