From 41bb73354e3449bdeb0df0029f6faf52f70b0a74 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 13:34:05 -0600 Subject: [PATCH 01/15] Never block the job loop on a client socket A connected but unresponsive client could stall the initial window size and playback sends inside the job select loop. That blocked the stop arm, so job stop cancelled the token to no effect and the job stayed running rack-wide. The mux now accepts a batch of initial messages delivered from the client's own writer task, and the select loop no longer awaits any client socket. Cancellation also interrupts a writer task mid send instead of waiting for the next message. Streamed output sends get the same treatment via a 60 second timeout. A consumer that makes no TCP progress for that long is treated as dead and the job is cancelled, matching the existing send error policy. The close handshake is bounded too, so the streaming task cannot outlive the job while holding the socket. Co-Authored-By: Claude Mythos 5 --- server/src/job.rs | 46 ++++++++++++++++++++++++++++------------------ server/src/mux.rs | 23 ++++++++++++++--------- 2 files changed, 42 insertions(+), 27 deletions(-) diff --git a/server/src/job.rs b/server/src/job.rs index 9873f785..6f14dcaf 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -35,7 +35,7 @@ use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _, AsyncWriteExt as _}; use tokio::process::Child; use tokio::sync::mpsc; use tokio::task::{JoinError, JoinHandle}; -use tokio::time::{Duration, Instant, sleep}; +use tokio::time::{Duration, Instant, sleep, timeout}; use tokio::{pin, select, spawn}; use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::tungstenite::protocol::Message as WebSocketMessage; @@ -60,6 +60,10 @@ pub type SocketReceiver = mpsc::Receiver<(SocketStream, Access)>; /// attach or make progress. const STREAMING_LINGER: Duration = Duration::from_secs(600); +/// How long a single send to a streaming consumer may stall +/// before the job is cancelled. +const STREAMING_SEND_TIMEOUT: Duration = Duration::from_secs(60); + pub struct Job { task: JoinHandle<(Result, JobOutputState)>, tx_client: SocketSender, @@ -242,7 +246,7 @@ async fn job( // Attach a new client, send it the current window size, and play back the last buffer. // Streaming jobs accept one read-write consumer, even posthumously. - Some((mut client, access)) = rx_client.recv(), if !dead || streaming => { + Some((client, access)) = rx_client.recv(), if !dead || streaming => { if streaming { if access != Access::ReadWrite || attached || tx_socket.try_send(client).is_err() { debug!(log, "refused streaming client"); @@ -251,25 +255,21 @@ async fn job( } continue; } + let mut initial = Vec::new(); match io.get_window_size() { Err(error) => error!(log, "failed to get pseudoterminal window size"; "error" => %error), Ok(size) => { - match client.send(Message::Control(Control::WindowChange(size.clone())).try_into().unwrap()).await { - Err(error) => error!(log, "failed to send pty window size"; "error" => %error), - Ok(()) => debug!(log, "sent pty window size"; "size" => ?size), - } + debug!(log, "sending pty window size"; "size" => ?size); + initial.push(Message::Control(Control::WindowChange(size)).try_into().unwrap()); } } if let Ok(Some(playback)) = playback_buffer(&mut stdout_file, INTERACTIVE_JOB_BUFFER_SIZE).await { - let playback_len = playback.len(); - match client.send(Message::Data(playback).try_into().unwrap()).await { - Err(error) => error!(log, "failed to play back job output"; "error" => %error), - Ok(()) => debug!(log, "played back output"; "bytes" => playback_len), - } + debug!(log, "playing back output"; "bytes" => playback.len()); + initial.push(Message::Data(playback).try_into().unwrap()); } - clients.add(client, access, stop.child_token()); + clients.add(client, access, initial, stop.child_token()); } // Handle a message from a client. @@ -398,14 +398,24 @@ async fn stream_output( return; }; while let Some(buf) = rx_output.recv().await { - if let Err(error) = client.send(Message::Data(buf).try_into().unwrap()).await { - error!(log, "failed to stream output to client"; "error" => %error); - stop.cancel(); - return; + match timeout( + STREAMING_SEND_TIMEOUT, + client.send(Message::Data(buf).try_into().unwrap()), + ) + .await + { + Ok(Ok(())) => continue, + Ok(Err(error)) => error!(log, "failed to stream output to client"; "error" => %error), + Err(_) => error!(log, "timed out streaming output to client"), } + stop.cancel(); + return; } - let _ = client.send(Message::Close.try_into().unwrap()).await; - let _ = client.close(None).await; + let _ = timeout(STREAMING_SEND_TIMEOUT, async { + let _ = client.send(Message::Close.try_into().unwrap()).await; + let _ = client.close(None).await; + }) + .await; } fn process_exit(exit_status: ExitStatus) -> Result { diff --git a/server/src/mux.rs b/server/src/mux.rs index 8994b992..60c90e98 100644 --- a/server/src/mux.rs +++ b/server/src/mux.rs @@ -61,6 +61,7 @@ impl WebSocketMux { &mut self, stream: SocketStream, access: Access, + initial: Vec, stop: CancellationToken, ) -> ClientId { let client_id = self.next_client_id; @@ -71,18 +72,22 @@ impl WebSocketMux { self.stop.insert(client_id, stop.clone()); let mut rx = self.send.subscribe(); let handle = spawn(async move { - loop { - select! { - recvd = rx.recv() => { - let Ok(message) = recvd else { break }; - if to.send(message).await.is_err() { - break - } + let io = async { + for message in initial { + if to.send(message).await.is_err() { + return; } - _ = stop.cancelled() => { - break; + } + loop { + let Ok(message) = rx.recv().await else { return }; + if to.send(message).await.is_err() { + return; } } + }; + select! { + _ = io => {} + _ = stop.cancelled() => {} } let _ = to.close().await; stop.cancel(); From 46bf16d889e552a5ed98fa9beddf816494236b85 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 13:49:34 -0600 Subject: [PATCH 02/15] Route unrouted requests to the home sled only The proxy preferred its home sled but fell back to an arbitrary sticky default when home was missing from the sled map. If the home sush server is unreachable, the sled is effectively dead, and there is a second switch. Untargeted requests now go to home or fail with 503, and the sticky default is gone. A proxy with no configured home, which happens only in development, still picks any known sled. Co-Authored-By: Claude Mythos 5 --- server/src/proxy.rs | 61 ++++++++++++++++++++------------------------- 1 file changed, 27 insertions(+), 34 deletions(-) diff --git a/server/src/proxy.rs b/server/src/proxy.rs index ce65dc61..9aa01b0b 100644 --- a/server/src/proxy.rs +++ b/server/src/proxy.rs @@ -7,9 +7,8 @@ //! Terminates client connections and routes each request to a sled, //! answering only `/version` itself. //! A request that names a target goes to the first sled the target -//! resolves to. Anything else goes to the sled hosting the proxy -//! when known, else a sticky default, because identities are cached -//! on the sled that authenticated them. +//! resolves to. Anything else goes to the sled hosting the proxy, +//! which any request may reach. //! Requests are forwarded untouched: bound request signatures cover //! the exact request line, so the proxy may never rewrite one. @@ -17,7 +16,7 @@ use std::collections::BTreeMap; use std::convert::Infallible; use std::io; use std::net::SocketAddr; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use http::header::CONTENT_TYPE; use http::{Method, StatusCode}; @@ -106,11 +105,7 @@ impl ProxyServer { let listener = TcpListener::bind(local_addr).await?; let local_addr = listener.local_addr()?; let acceptor = tls.map(|config| TlsAcceptor::from(Arc::new(config))); - let router = Arc::new(Router { - targets, - home, - default: Mutex::new(None), - }); + let router = Arc::new(Router { targets, home }); spawn(listen( log.new(o!("component" => "proxy")), listener, @@ -138,7 +133,6 @@ impl ProxyServer { struct Router { targets: watch::Receiver, home: Option, - default: Mutex>, } impl Router { @@ -155,29 +149,21 @@ impl Router { StatusCode::BAD_REQUEST, format!("unable to parse target `{bad}`"), ))), - None => { - if let Some(home) = self.home.as_ref() - && let Some(addr) = targets.sleds.get(home) - { - return Ok(*addr); - } - let mut default = self.default.lock().unwrap(); - if let Some(baseboard) = default.as_ref() - && let Some(addr) = targets.sleds.get(baseboard) - { - return Ok(*addr); - } - match targets.sleds.iter().next() { - Some((baseboard, addr)) => { - *default = Some(baseboard.clone()); - Ok(*addr) - } + None => match self.home.as_ref() { + Some(home) => targets.sleds.get(home).copied().ok_or_else(|| { + Box::new(error_response( + StatusCode::SERVICE_UNAVAILABLE, + format!("home sled `{home}` unknown to the proxy"), + )) + }), + None => match targets.sleds.iter().next() { + Some((_, addr)) => Ok(*addr), None => Err(Box::new(error_response( StatusCode::SERVICE_UNAVAILABLE, "no sleds known to the proxy", ))), - } - } + }, + }, } } } @@ -377,8 +363,8 @@ mod test { s.parse().unwrap() } - /// Unrouted requests go to the home sled while the targets know - /// it, and otherwise to a sticky default. + /// Unrouted requests go to the home sled, or fail while the + /// targets don't know it. Without a home, any sled will do. #[test] fn home_preference() { let sled = |serial: &str| BaseboardId { @@ -391,20 +377,27 @@ mod test { targets.sleds.insert(sled("home"), addr(2)); let (tx, rx) = watch::channel(targets); let router = Router { - targets: rx, + targets: rx.clone(), home: Some(sled("home")), - default: Mutex::new(None), }; let get = request("/versions"); assert_eq!(router.route(&get).unwrap(), addr(2)); tx.send_modify(|t| { t.sleds.remove(&sled("home")); }); - assert_eq!(router.route(&get).unwrap(), addr(1)); + assert_eq!( + router.route(&get).unwrap_err().status(), + StatusCode::SERVICE_UNAVAILABLE + ); tx.send_modify(|t| { t.sleds.insert(sled("home"), addr(2)); }); assert_eq!(router.route(&get).unwrap(), addr(2)); + let homeless = Router { + targets: rx, + home: None, + }; + assert_eq!(homeless.route(&get).unwrap(), addr(1)); } /// The proxy answers `GET /version` itself unless `via` routes it. From e0596033cbf3f31b51396fca6ba15cae4b5e4d22 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 13:55:24 -0600 Subject: [PATCH 03/15] Bound the drain of a dead streaming job Reads reset the streaming drain timeout so a slow consumer can finish the tail, but that let an escaped descendant that keeps writing run the job forever. Process death now also arms an absolute one minute limit, after which the job ends with a warning and its recorded exit status. Co-Authored-By: Claude Mythos 5 --- server/src/job.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/server/src/job.rs b/server/src/job.rs index 6f14dcaf..c8ad9deb 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -64,6 +64,10 @@ const STREAMING_LINGER: Duration = Duration::from_secs(600); /// before the job is cancelled. const STREAMING_SEND_TIMEOUT: Duration = Duration::from_secs(60); +/// How long a dead streaming job may drain in total, since an +/// escaped descendant can hold the output pipe open forever. +const STREAMING_DRAIN_LIMIT: Duration = Duration::from_secs(60); + pub struct Job { task: JoinHandle<(Result, JobOutputState)>, tx_client: SocketSender, @@ -137,6 +141,8 @@ async fn job( pin!(drain_timeout); let linger = sleep(Duration::default()); pin!(linger); + let drain_limit = sleep(Duration::default()); + pin!(drain_limit); loop { // A finished stream is done once its consumer has taken every chunk. if eof && attached && pending.is_none() { @@ -330,9 +336,18 @@ async fn job( _ = child.wait(), if !dead => { debug!(log, "reaped job process"); drain_timeout.as_mut().reset(Instant::now() + io.drain_timeout()); + if streaming { + drain_limit.as_mut().reset(Instant::now() + STREAMING_DRAIN_LIMIT); + } dead = true; } + // Cut off a dead stream that never reaches EOF. + _ = &mut drain_limit, if streaming && dead && !eof => { + warn!(log, "streaming job output never closed"); + break; + } + // Give up on a stream whose consumer never drained it. _ = &mut linger, if eof => { warn!(log, "streaming consumer never drained the output"); From f931cad50aec3fd4c69e4267b35ce30deca7bec9 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 14:14:33 -0600 Subject: [PATCH 04/15] Warn when a job's target may not exist A job whose target matches no sled enqueues rack-wide but runs nowhere, and both server and client waits are ten silent minutes. Job start now checks explicit cubby and baseboard targets against the rack inventory before the signing ceremony and asks for confirmation on a miss, since a sled cut off from gossip may still be directly reachable. Offline signing, JSON output, and an unavailable inventory skip the prompt. A watch that has seen no status from any sled after fifteen polls says so once instead of spinning silently. Co-Authored-By: Claude Mythos 5 --- client/src/cli.rs | 26 +++++++++++++++++++++++++- client/src/commands.rs | 38 ++++++++++++++++++++++++++++++++++++++ client/src/context.rs | 4 +++- client/src/repl.rs | 10 +++++++++- 4 files changed, 75 insertions(+), 3 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index 13588734..59e95642 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -27,7 +27,7 @@ use sush_common::jobs::{ SignedJob, job_status_to_json_map, }; use sush_common::keys::{KeyId, Signature, SshPublicKey}; -use sush_common::targets::{MAX_CUBBY, SledVersion}; +use sush_common::targets::{MAX_CUBBY, SledId, SledVersion}; use sush_common::version::VersionInfo; use crate::AuthzSigner; @@ -463,6 +463,15 @@ impl CommandContext for Cli { } } + fn job_watch_stalled(&mut self, job_id: &JobId) { + let guard = self.watch.lock().unwrap(); + if let Some(watch) = guard.as_ref() { + let _ = watch.multi.println(format!( + "❗ No sled has reported a status for job `{job_id}`" + )); + } + } + fn job_watch_finished(&mut self, _job_id: &JobId) { if let Some(watch) = self.watch.lock().unwrap().take() { for bar in watch.bars.values() { @@ -757,6 +766,21 @@ impl CommandContext for Cli { Ok(()) } + fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError> { + match self.get_output_format() { + OutputFormat::Json => Ok(()), + OutputFormat::Text => { + let prompt = + format!("❓ Sled `{sled}` is not in the rack inventory. Proceed (yes/no)? "); + if read_bool(&prompt)? { + Ok(()) + } else { + Err(CommandError::Canceled) + } + } + } + } + fn really_revoke(&mut self, what: &str, key_id: KeyId) -> Result { match self.get_output_format() { OutputFormat::Json => Ok(key_id), diff --git a/client/src/commands.rs b/client/src/commands.rs index 8f04422b..25d49226 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1169,6 +1169,9 @@ async fn job( return Err(CommandError::io(path, ErrorKind::AlreadyExists.into())); } } + if let Some(client) = client.as_ref() { + preflight_target(ctx, client, &target).await?; + } let streaming = if *streaming { Streaming::Output } else { @@ -1403,6 +1406,7 @@ async fn job_start( let mut last = JobStatusMap::new(); let mut settling = Settling::default(); let mut started = false; + let mut stalled = false; let mut stopped = false; let mut sigint = signal(SignalKind::interrupt())?; let status = loop { @@ -1435,6 +1439,10 @@ async fn job_start( if settling.done(&job_target, rack, &last) && started { break last; } + if !stalled && last.is_empty() && settling.polls >= WATCH_STALL_POLLS { + ctx.job_watch_stalled(&job_id); + stalled = true; + } } // While the job runs, an interrupt stops it but keeps @@ -1597,6 +1605,10 @@ async fn job_watch( /// is likely still missing sleds. const WATCH_MIN_POLLS: usize = 5; +/// How many polls a watch may go without any sled reporting a status +/// before warning that the job may never run. +const WATCH_STALL_POLLS: usize = 15; + /// Rolling settlement state for a watched job. #[derive(Default)] struct Settling { @@ -2081,6 +2093,32 @@ impl FromStr for TargetArg { } } +/// Confirm before signing a job for a sled that doesn't appear in inventory. +#[cfg(feature = "permslip")] +async fn preflight_target( + ctx: &mut impl CommandContext, + client: &Client, + target: &Target, +) -> Result<(), CommandError> { + let Target::Sleds(sleds) = target else { + return Ok(()); + }; + let Ok(inventory) = client.versions().send().await else { + return Ok(()); + }; + let inventory = inventory.into_inner(); + for sled in sleds { + let known = match sled { + SledId::Baseboard(baseboard) => inventory.iter().any(|s| &s.baseboard == baseboard), + SledId::Cubby(cubby) => inventory.iter().any(|s| s.cubby == Some(*cubby)), + }; + if !known { + ctx.really_target(sled)?; + } + } + Ok(()) +} + /// Resolve a target argument to a target, matching bare serial /// numbers against the rack's sled inventory. async fn resolve_target_arg(client: &Client, target: &TargetArg) -> Result { diff --git a/client/src/context.rs b/client/src/context.rs index 62a7010c..c6562016 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -18,7 +18,7 @@ use sush_common::jobs::{ Access, JobId, JobOutputStream, JobStatusMap, Session, SessionId, SignedJob, }; use sush_common::keys::{KeyId, SshPublicKey}; -use sush_common::targets::SledVersion; +use sush_common::targets::{SledId, SledVersion}; use sush_common::version::VersionInfo; use crate::AuthzSigner; @@ -130,6 +130,7 @@ pub trait CommandContext: Clone + Send + Sync { fn cert_imported(&mut self, path: &Path, key_id: KeyId) -> Result<(), CommandError>; // Job management + fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError>; fn job_started(&mut self, job: &SignedJob); fn job_stopped(&mut self, id: &JobId); fn job_error(&mut self, error: CommandError) -> CommandError; @@ -146,6 +147,7 @@ pub trait CommandContext: Clone + Send + Sync { fn job_output_finished(&mut self, id: &JobId, stream: JobOutputStream, stage: Option<&str>); fn job_watch_started(&mut self, id: &JobId); fn job_watch_update(&mut self, status: &JobStatusMap); + fn job_watch_stalled(&mut self, id: &JobId); fn job_watch_finished(&mut self, id: &JobId); fn job_attached(&mut self, id: &JobId); fn job_detached(&mut self, id: &JobId); diff --git a/client/src/repl.rs b/client/src/repl.rs index 29cd7c22..c86229d0 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -23,7 +23,7 @@ use sush_common::jobs::{ Access, JobId, JobOutputStream, JobStatusMap, Session, SessionId, SignedJob, }; use sush_common::keys::{KeyId, SshPublicKey}; -use sush_common::targets::SledVersion; +use sush_common::targets::{SledId, SledVersion}; use sush_common::version::VersionInfo; use crate::cli::Cli; @@ -338,6 +338,10 @@ impl CommandContext for Repl { self.cli.job_watch_update(status) } + fn job_watch_stalled(&mut self, job_id: &JobId) { + self.cli.job_watch_stalled(job_id) + } + fn job_watch_finished(&mut self, job_id: &JobId) { self.cli.job_watch_finished(job_id) } @@ -365,6 +369,10 @@ impl CommandContext for Repl { self.cli.please_touch(identity) } + fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError> { + self.cli.really_target(sled) + } + fn really_revoke(&mut self, what: &str, key_id: KeyId) -> Result { self.cli.really_revoke(what, key_id) } From 42d91577a1edb5656acf270a7eeb5761fd1cac1b Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 14:24:44 -0600 Subject: [PATCH 05/15] Deny warnings in the lint recipe Clippy warnings that only fail in CI make just lint an unreliable gate. Match the CI flags. Co-Authored-By: Claude Mythos 5 --- justfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/justfile b/justfile index 862a860d..616d73dc 100644 --- a/justfile +++ b/justfile @@ -4,7 +4,7 @@ check: cargo check --workspace --all-targets lint: - cargo fmt --check && cargo clippy --tests + cargo fmt --check && cargo clippy --tests -- --no-deps --deny warnings test *FILTER: cargo nextest run --workspace {{FILTER}} From e5f3c6e177509c98ca47de4ff267789360e06c78 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 14:52:51 -0600 Subject: [PATCH 06/15] Bound mux writer teardown A client that stopped reading could park its writer task in an untimed close, and a job that ended naturally never cancelled its writers at all, leaking a task and a socket per stalled client. Closing is now bounded by a timeout and dropping the mux cancels every writer, so a wedged client costs at most a few seconds of cleanup. Co-Authored-By: Claude Mythos 5 --- server/src/mux.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/server/src/mux.rs b/server/src/mux.rs index 60c90e98..c90d6f87 100644 --- a/server/src/mux.rs +++ b/server/src/mux.rs @@ -16,6 +16,7 @@ use futures::stream::{FuturesUnordered, SplitStream}; use futures::{SinkExt as _, Stream, StreamExt}; use tokio::sync::broadcast; use tokio::task::JoinHandle; +use tokio::time::{Duration, timeout}; use tokio::{select, spawn}; use tokio_stream::{StreamMap, StreamNotifyClose}; use tokio_tungstenite::tungstenite::error::Error as WebSocketError; @@ -30,6 +31,9 @@ use crate::job::SocketStream; /// before it is disconnected. pub const MUX_CLIENT_CHANNEL_CAPACITY: usize = 100; +/// How long a writer task may spend closing its socket. +const CLOSE_TIMEOUT: Duration = Duration::from_secs(5); + /// Client identifiers are opaque and ephemeral. We use monotonically /// increasing integers because they're simple and cheap. pub type ClientId = usize; @@ -89,7 +93,7 @@ impl WebSocketMux { _ = io => {} _ = stop.cancelled() => {} } - let _ = to.close().await; + let _ = timeout(CLOSE_TIMEOUT, to.close()).await; stop.cancel(); client_id }); @@ -121,6 +125,14 @@ impl WebSocketMux { } } +impl Drop for WebSocketMux { + fn drop(&mut self) { + for stop in self.stop.values() { + stop.cancel(); + } + } +} + impl Stream for WebSocketMux { type Item = (ClientId, Result); From 3e177045817bf8c75a2d53780aaf9c0061787cd0 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 14:53:31 -0600 Subject: [PATCH 07/15] Abandon a stream whose consumer never attaches A streaming job with no consumer fills the pipe and blocks with the process alive, so neither the drain timers nor the linger ever arm, and the job stays Running forever. Kill the job if no consumer attaches within the linger period. The linger still governs after EOF, so posthumous attachment is unaffected. Co-Authored-By: Claude Mythos 5 --- server/src/job.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/server/src/job.rs b/server/src/job.rs index c8ad9deb..d36c935c 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -143,6 +143,8 @@ async fn job( pin!(linger); let drain_limit = sleep(Duration::default()); pin!(drain_limit); + let attach_deadline = sleep(STREAMING_LINGER); + pin!(attach_deadline); loop { // A finished stream is done once its consumer has taken every chunk. if eof && attached && pending.is_none() { @@ -348,9 +350,17 @@ async fn job( break; } + // Give up on a live stream whose consumer never attached. + _ = &mut attach_deadline, if streaming && !attached && !eof => { + warn!(log, "streaming consumer never attached"); + stop.cancel(); + break; + } + // Give up on a stream whose consumer never drained it. _ = &mut linger, if eof => { warn!(log, "streaming consumer never drained the output"); + stop.cancel(); break; } From 025ed3deae87bf1c0116cbf89208b7461ae535b5 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 14:53:31 -0600 Subject: [PATCH 08/15] Stop jobs with SIGTERM before SIGKILL Job stop sent SIGKILL immediately and gave up after one attempt, even if the kill failed and left the process alive with the stop arm permanently disabled. Send SIGTERM first so the job can clean up, upgrade to SIGKILL after a grace period, and retry a failed kill a few times before giving up. Resolves the 2-stage stop TODO. Co-Authored-By: Claude Mythos 5 --- server/src/executor.rs | 32 ++++++++++++++++++-------------- server/src/job.rs | 28 ++++++++++++++++++++++++---- tests/src/manager_tests.rs | 6 +++--- 3 files changed, 45 insertions(+), 21 deletions(-) diff --git a/server/src/executor.rs b/server/src/executor.rs index b8b8b849..2bf2a5f7 100644 --- a/server/src/executor.rs +++ b/server/src/executor.rs @@ -18,7 +18,7 @@ use std::sync::{Arc, RwLock}; use chrono::Utc; use futures::Stream; use pwd::Passwd; -use rustix::io::close; +use rustix::io::{Errno, close}; use rustix::process::{Pid, Signal, ioctl_tiocsctty, kill_process_group, setsid}; use slog::{Logger, debug, error, o, warn}; use tokio::fs::{DirBuilder, OpenOptions}; @@ -488,20 +488,24 @@ async fn send_error( } } -/// Kill a job's whole process group, of which `child` should be the leader. -/// -/// TODO: 2-stage stop with `SIGTERM` and a grace period. -pub fn kill_job(log: &Logger, child: &Child) { - if let Some(pid) = child.id() - && let Ok(pid) = pid.try_into() - && let Some(pid) = Pid::from_raw(pid) - { - match kill_process_group(pid, Signal::KILL) { - Ok(()) => debug!(log, "killed job processes"), - Err(error) => error!(log, "unable to kill job"; "error" => %error), +/// A job's process group ID, which its `child` leads. Capture it +/// before the leader is reaped, after which `id` returns nothing. +pub fn job_pgid(child: &Child) -> Option { + Pid::from_raw(child.id()?.try_into().ok()?) +} + +/// Send a signal to a job's whole process group. +pub fn kill_job(log: &Logger, pgid: Option, signal: Signal) { + let Some(pgid) = pgid else { + debug!(log, "job has no process group"); + return; + }; + match kill_process_group(pgid, signal) { + Ok(()) => debug!(log, "signalled job processes"; "signal" => ?signal), + Err(Errno::SRCH) => debug!(log, "job processes are already dead"), + Err(error) => { + error!(log, "unable to signal job"; "signal" => ?signal, "error" => %error) } - } else { - debug!(log, "process is already dead or has an invalid PID"); } } diff --git a/server/src/job.rs b/server/src/job.rs index d36c935c..bfdea85b 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -29,6 +29,7 @@ use blake3::Hasher; use bytes::{Bytes, BytesMut}; use dropshot::WebsocketConnectionRaw; use futures::{SinkExt as _, StreamExt as _}; +use rustix::process::Signal; use slog::{Logger, debug, error, info, warn}; use tokio::fs::File; use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _, AsyncWriteExt as _}; @@ -48,7 +49,7 @@ use sush_common::jobs::{ }; use tokio_util::sync::CancellationToken; -use crate::executor::kill_job; +use crate::executor::{job_pgid, kill_job}; use crate::io::JobIo; use crate::mux::WebSocketMux; @@ -68,6 +69,12 @@ const STREAMING_SEND_TIMEOUT: Duration = Duration::from_secs(60); /// escaped descendant can hold the output pipe open forever. const STREAMING_DRAIN_LIMIT: Duration = Duration::from_secs(60); +/// How long a stopped job may clean up after SIGTERM before SIGKILL. +const KILL_GRACE: Duration = Duration::from_secs(5); + +/// How many times a failed SIGKILL is retried before giving up. +const KILL_ATTEMPTS: usize = 3; + pub struct Job { task: JoinHandle<(Result, JobOutputState)>, tx_client: SocketSender, @@ -121,7 +128,9 @@ async fn job( let mut stderr_hasher = Hasher::new(); let mut clients = WebSocketMux::new(); let mut fatal = Option::::None; + let pgid = job_pgid(&child); let mut killed = false; + let mut kills = 0; let mut dead = false; let streaming = matches!(streaming, Streaming::Output); let mut pending = Option::::None; @@ -145,6 +154,8 @@ async fn job( pin!(drain_limit); let attach_deadline = sleep(STREAMING_LINGER); pin!(attach_deadline); + let kill_timeout = sleep(Duration::default()); + pin!(kill_timeout); loop { // A finished stream is done once its consumer has taken every chunk. if eof && attached && pending.is_none() { @@ -326,12 +337,21 @@ async fn job( } } - // Stop job on cancellation signal, but only once. + // Stop job on cancellation signal: terminate first, then kill. _ = stop.cancelled(), if !killed => { - kill_job(&log, &child); + kill_job(&log, pgid, Signal::TERM); + kill_timeout.as_mut().reset(Instant::now() + KILL_GRACE); killed = true; } + // Upgrade to SIGKILL a stopped group that outlives its grace + // period, even if its leader is already dead. + _ = &mut kill_timeout, if killed && kills < KILL_ATTEMPTS => { + kill_job(&log, pgid, Signal::KILL); + kill_timeout.as_mut().reset(Instant::now() + KILL_GRACE); + kills += 1; + } + // Notice when the job dies, but do not exit the loop; // we must continue reading output until we hit EOF or // the drain timeout expires. @@ -387,7 +407,7 @@ async fn job( let exit_status = select! { status = child.wait() => status, _ = stop.cancelled(), if !killed => { - kill_job(&log, &child); + kill_job(&log, pgid, Signal::KILL); child.wait().await } }; diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 55d0e99b..54f440dc 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -14,7 +14,7 @@ use std::time::Duration; use chrono::Utc; use function_name::named; use http_range_header::{EndPosition, StartPosition, SyntacticallyCorrectRange as Range}; -use libc::{SIGKILL, SIGXCPU}; +use libc::{SIGTERM, SIGXCPU}; use pwd::Passwd; use sled_hardware_types::BaseboardId; use slog::{Discard, Logger, o}; @@ -329,7 +329,7 @@ async fn job_stop() { check_status_stopped( status, &job_id, - Err(ProcessError::Killed(SIGKILL)), + Err(ProcessError::Killed(SIGTERM)), Some(0), Some(0), ); @@ -820,7 +820,7 @@ async fn shutdown() { check_status_stopped( status, &job_id, - Err(ProcessError::Killed(SIGKILL)), + Err(ProcessError::Killed(SIGTERM)), Some(0), Some(0), ); From 00e575aace47bd117878fa244435177523c34c1f Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 17:23:26 -0600 Subject: [PATCH 09/15] Route watch status polls via a single-sled target Watch polls were untargeted, so the proxy sent them to its home sled: settlement waited on gossip instead of the executing sled, and alternating requests between sleds could bounce 401s while a login gossips. The job_status endpoint now accepts the routing hint, and both watch loops pass it when the target names exactly one baseboard. Co-Authored-By: Claude Mythos 5 --- api/src/lib.rs | 1 + client/src/commands.rs | 23 +++++++++++++++-------- server/src/server.rs | 1 + sush.json | 9 +++++++++ 4 files changed, 26 insertions(+), 8 deletions(-) diff --git a/api/src/lib.rs b/api/src/lib.rs index 528640c7..78ae794d 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -189,6 +189,7 @@ pub trait SushApi { ctx: RequestContext, headers: Header, params: PathParams, + query: QueryParams, ) -> Result, HttpError>; /// Get (a subset of) the standard output or standard error of a job. diff --git a/client/src/commands.rs b/client/src/commands.rs index 25d49226..8a14f2ae 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1425,7 +1425,7 @@ async fn job_start( } _ = ticker.tick() => { - last = match job_status_map(ctx, client, &job_id).await { + last = match job_status_map(ctx, client, &job_id, job_target.single_baseboard()).await { Ok(status) => status, // The job may not be visible anywhere yet. Err(CommandError::NotFound(_)) => JobStatusMap::new(), @@ -1545,19 +1545,26 @@ async fn job_status( job_id: &JobId, style: StatusDisplayStyle, ) -> Result<(), CommandError> { - let status = job_status_map(ctx, client, job_id).await?; + let status = job_status_map(ctx, client, job_id, None).await?; ctx.job_status(job_id, &status, style); Ok(()) } -/// Fetch a job's rack-wide status map. +/// Fetch a job's rack-wide status map. Routing `via` a single-sled +/// target gets its authoritative status and keeps the login on the +/// sled that already knows it. async fn job_status_map( ctx: &mut impl CommandContext, client: &Client, job_id: &JobId, + via: Option<&BaseboardId>, ) -> Result { - let status = with_login(ctx, client, async || { - client.job_status().job_id(job_id).send().await + let status = with_login_via(ctx, client, via, async || { + let mut request = client.job_status().job_id(job_id); + if let Some(via) = via { + request = request.via(via.to_string()); + } + request.send().await }) .await? .into_inner(); @@ -1579,7 +1586,7 @@ async fn job_watch( let mut settling = Settling::default(); let mut sigint = signal(SignalKind::interrupt())?; let status = loop { - let status = match job_status_map(ctx, client, job_id).await { + let status = match job_status_map(ctx, client, job_id, target.single_baseboard()).await { Ok(status) => status, Err(error) => { ctx.job_watch_finished(job_id); @@ -1719,7 +1726,7 @@ async fn job_output( } // Fetch output from every sled with a recorded status. - let status = job_status_map(ctx, client, &args.job_id).await?; + let status = job_status_map(ctx, client, &args.job_id, None).await?; if status.is_empty() { return Err(CommandError::NotFound(format!( "Job `{}` not found", @@ -2158,7 +2165,7 @@ async fn resolve_serial( job_id: &JobId, serial: &str, ) -> Result { - let status = job_status_map(ctx, client, job_id).await?; + let status = job_status_map(ctx, client, job_id, None).await?; let mut matches = status .keys() .filter(|b| b.serial_number.eq_ignore_ascii_case(serial)); diff --git a/server/src/server.rs b/server/src/server.rs index 3568ac36..71979a61 100644 --- a/server/src/server.rs +++ b/server/src/server.rs @@ -299,6 +299,7 @@ impl SushApi for ApiServer { ctx: RequestContext, headers: Header, params: PathParams, + _query: QueryParams, ) -> Result, HttpError> { let mgr = ctx.context(); let Authorization { authorization } = headers.into_inner(); diff --git a/sush.json b/sush.json index 7e975112..fca7e282 100644 --- a/sush.json +++ b/sush.json @@ -613,6 +613,15 @@ "schema": { "$ref": "#/components/schemas/JobId" } + }, + { + "in": "query", + "name": "via", + "description": "Where a proxy should route this request.", + "schema": { + "nullable": true, + "type": "string" + } } ], "responses": { From 315d1bc4afca45bd60cd2bc073a9828af4617ed9 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 21:03:28 -0600 Subject: [PATCH 10/15] Restore the output file position after a failed playback A playback error left the shared output file cursor mid-file, so the next recorded write would overwrite the tail of the record and silently diverge from the audit hash. Re-seek to the end on error, and cancel the job if even that fails. Co-Authored-By: Claude Mythos 5 --- server/src/job.rs | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/server/src/job.rs b/server/src/job.rs index bfdea85b..ae6e9a86 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -283,9 +283,19 @@ async fn job( } } - if let Ok(Some(playback)) = playback_buffer(&mut stdout_file, INTERACTIVE_JOB_BUFFER_SIZE).await { - debug!(log, "playing back output"; "bytes" => playback.len()); - initial.push(Message::Data(playback).try_into().unwrap()); + match playback_buffer(&mut stdout_file, INTERACTIVE_JOB_BUFFER_SIZE).await { + Ok(Some(playback)) => { + debug!(log, "playing back output"; "bytes" => playback.len()); + initial.push(Message::Data(playback).try_into().unwrap()); + } + Ok(None) => (), + Err(error) => { + error!(log, "failed to play back job output"; "error" => %error); + if let Err(error) = stdout_file.seek(SeekFrom::End(0)).await { + error!(log, "failed to restore output file position"; "error" => %error); + stop.cancel(); + } + } } clients.add(client, access, initial, stop.child_token()); @@ -479,8 +489,8 @@ fn process_exit(exit_status: ExitStatus) -> Result { } /// Fetch the last few bytes of the output file for client play back. -/// Errors here indicate problems with the output file, and so should -/// be treated as job-ending. +/// On error the file position may be anywhere; the caller must restore +/// it before the file is written again, or end the job. async fn playback_buffer( output_file: &mut File, output_bytes: usize, From f64ade7c90bd38010a129c378b850b4c174658ec Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 21:07:08 -0600 Subject: [PATCH 11/15] Fix inaccurate comment Co-Authored-By: Claude Mythos 5 --- server/src/io.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/server/src/io.rs b/server/src/io.rs index 610def50..e98dcc1e 100644 --- a/server/src/io.rs +++ b/server/src/io.rs @@ -169,8 +169,8 @@ impl JobIo { /// How long to keep reading output after the child dies. /// - /// For a pty there is no portable EOF signal; a short quiet period - /// is the only way to know we've drained (OpenSSH does this too). + /// For a pty there is no portable EOF signal; a short window + /// after death is the best we can do (OpenSSH does this too). /// Pipes deliver EOF once every writer exits, so this is only a /// backstop against a descendant that escaped the process group /// while holding the inherited pipe open. From 0e7e08c102465a6bb31b0c06cf8db86f94353959 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 21:33:40 -0600 Subject: [PATCH 12/15] Route the output status fetch via its sled The fetch logged in via the target sled but sent the status request unrouted, recreating the login bounce that watch polls just fixed. Co-Authored-By: Claude Mythos 5 --- client/src/commands.rs | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 8a14f2ae..d0633f71 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1757,14 +1757,7 @@ async fn job_output_from( }: JobOutput, ) -> Result<(), CommandError> { // Fetch job status for output length and hash. - let status = job_status_try_from_json_map( - with_login_via(ctx, client, Some(target), async || { - client.job_status().job_id(job_id).send().await - }) - .await? - .into_inner(), - ) - .map_err(CommandError::BaseboardIdParseError)?; + let status = job_status_map(ctx, client, &job_id, Some(target)).await?; let JobOutputState { stdout_len, From 32f1f1cc90b7acfdf66353cf237361d4a5280465 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 21:33:40 -0600 Subject: [PATCH 13/15] Upgrade to SIGKILL even after the job loop exits A stopped job that ignored SIGTERM but closed its pipes broke out of the loop before the grace period expired, and the reap then waited forever. The reap select now escalates too. Co-Authored-By: Claude Mythos 5 --- server/src/job.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/server/src/job.rs b/server/src/job.rs index ae6e9a86..99a5fa4c 100644 --- a/server/src/job.rs +++ b/server/src/job.rs @@ -357,6 +357,7 @@ async fn job( // Upgrade to SIGKILL a stopped group that outlives its grace // period, even if its leader is already dead. _ = &mut kill_timeout, if killed && kills < KILL_ATTEMPTS => { + warn!(log, "job survived SIGTERM grace period, upgrading to SIGKILL"); kill_job(&log, pgid, Signal::KILL); kill_timeout.as_mut().reset(Instant::now() + KILL_GRACE); kills += 1; @@ -413,13 +414,18 @@ async fn job( let _ = tx_output.try_send(buf); } - // Reap the process. + // Reap the process. A stopped job that exited the loop alive + // still gets its SIGKILL upgrade after the grace period. let exit_status = select! { status = child.wait() => status, _ = stop.cancelled(), if !killed => { kill_job(&log, pgid, Signal::KILL); child.wait().await } + _ = sleep(KILL_GRACE), if killed && !dead => { + kill_job(&log, pgid, Signal::KILL); + child.wait().await + } }; let result = match fatal { Some(error) => Err(error), From 51698229c8b7aba3e02971aece84e9a6fe89a1e1 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 21:33:40 -0600 Subject: [PATCH 14/15] Log proxy routing failures A proxy answering 502 or 503 was invisible in the switch zone logs. Co-Authored-By: Claude Mythos 5 --- server/src/proxy.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/server/src/proxy.rs b/server/src/proxy.rs index 9aa01b0b..f5275355 100644 --- a/server/src/proxy.rs +++ b/server/src/proxy.rs @@ -274,7 +274,13 @@ where Some(response) => response, None => match router.route(&request) { Ok(addr) => forward(log, addr, request).await, - Err(response) => *response, + Err(response) => { + warn!( + log, "request not routable"; + "uri" => %request.uri(), "status" => %response.status() + ); + *response + } }, }) } From f7b7c40b6f76083ef65a1c230770dd3fe7ad1b12 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 22 Aug 2026 21:43:35 -0600 Subject: [PATCH 15/15] Route job stop via a single-sled target The Ctrl-C stop from a routed watch went unrouted to the home sled, costing a second login. The via hint rides JobStopParams because an endpoint gets one query struct; sleds ignore it as usual. Co-Authored-By: Claude Mythos 5 --- api/src/lib.rs | 2 ++ client/src/commands.rs | 18 +++++++++--------- server/src/manager.rs | 2 +- sush.json | 9 +++++++++ tests/src/manager_tests.rs | 7 +++++++ 5 files changed, 28 insertions(+), 10 deletions(-) diff --git a/api/src/lib.rs b/api/src/lib.rs index 78ae794d..8c11b4c3 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -388,6 +388,8 @@ impl JobWait { pub struct JobStopParams { /// Wait for the job process to end. pub wait: JobWait, + /// Where a proxy should route this request. Sleds ignore it. + pub via: Option, } /// Simple pagination for history list. diff --git a/client/src/commands.rs b/client/src/commands.rs index d0633f71..2f962aea 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1214,7 +1214,7 @@ async fn job( } (JobCommand::Stop { job_id }, Some(client)) => { - job_stop(ctx, client, &job_id).await?; + job_stop(ctx, client, &job_id, None).await?; ctx.job_stopped(&job_id); Ok(()) } @@ -1456,7 +1456,7 @@ async fn job_start( break last; } for _ in 0..3 { - match job_stop(ctx, client, &job_id).await { + match job_stop(ctx, client, &job_id, job_target.single_baseboard()).await { Ok(_) => { ctx.job_stopped(&job_id); stopped = true; @@ -1526,14 +1526,14 @@ async fn job_stop( ctx: &mut impl CommandContext, client: &Client, job_id: &JobId, + via: Option<&BaseboardId>, ) -> Result<(), CommandError> { - with_login(ctx, client, async || { - client - .job_stop() - .job_id(job_id) - .wait(JobWait::Stop) - .send() - .await + with_login_via(ctx, client, via, async || { + let mut request = client.job_stop().job_id(job_id).wait(JobWait::Stop); + if let Some(via) = via { + request = request.via(via.to_string()); + } + request.send().await }) .await?; Ok(()) diff --git a/server/src/manager.rs b/server/src/manager.rs index d5f114e1..82aa8296 100644 --- a/server/src/manager.rs +++ b/server/src/manager.rs @@ -622,7 +622,7 @@ impl JobManager { &self, authn: &Identity, job_id: &JobId, - JobStopParams { wait }: JobStopParams, + JobStopParams { wait, .. }: JobStopParams, ) -> Result<(), JobError> { self.job_request(authn, JobRequest::Stop(job_id.to_owned())) .await?; diff --git a/sush.json b/sush.json index fca7e282..e36a0790 100644 --- a/sush.json +++ b/sush.json @@ -669,6 +669,15 @@ "$ref": "#/components/schemas/JobId" } }, + { + "in": "query", + "name": "via", + "description": "Where a proxy should route this request. Sleds ignore it.", + "schema": { + "nullable": true, + "type": "string" + } + }, { "in": "query", "name": "wait", diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 54f440dc..96b90b53 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -271,6 +271,7 @@ async fn job_stop() { &job_id, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -313,6 +314,7 @@ async fn job_stop() { &job_id, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -390,6 +392,7 @@ async fn cancel_queued_job() { &job_id_b, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -405,6 +408,7 @@ async fn cancel_queued_job() { &job_id_a, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -980,6 +984,7 @@ async fn attribution() { &job_id_b, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -994,6 +999,7 @@ async fn attribution() { &job_id_a, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await @@ -1399,6 +1405,7 @@ async fn attach_grants() { &job_id, JobStopParams { wait: JobWait::Stop, + ..Default::default() }, ) .await