From f84e763bd0e61483d1cea831a56ebedfc6b71031 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 14:29:25 -0600 Subject: [PATCH 01/20] Return proxy errors as JSON Also tweak error messages to read better now that the client can decode them. Co-authored-by: Claude --- server/src/proxy.rs | 30 +++++++++++++++++++++++------- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/server/src/proxy.rs b/server/src/proxy.rs index 1fc9e909..2d476f07 100644 --- a/server/src/proxy.rs +++ b/server/src/proxy.rs @@ -36,6 +36,7 @@ use rustls::pki_types::{CertificateDer, PrivateKeyDer}; use rustls::server::{ClientHello, ResolvesServerCert}; use rustls::sign::CertifiedKey; use rustls::version::TLS13; +use serde_json::json; use sled_hardware_types::BaseboardId; use slog::{Logger, debug, error, info, o, warn}; use tokio::io::{AsyncRead, AsyncWrite, copy_bidirectional}; @@ -197,12 +198,12 @@ impl Router { Some(Ok(target)) => self.targets.resolve(&target).ok_or_else(|| { Box::new(error_response( StatusCode::BAD_GATEWAY, - format!("no route to target `{target}`"), + format!("No reachable sled matches `{target}`"), )) }), Some(Err(bad)) => Err(Box::new(error_response( StatusCode::BAD_REQUEST, - format!("unable to parse target `{bad}`"), + format!("Invalid target `{bad}`"), ))), None => { let sleds = self.targets.sleds.borrow(); @@ -210,14 +211,14 @@ impl Router { Some(home) => sleds.get(home).copied().ok_or_else(|| { Box::new(error_response( StatusCode::SERVICE_UNAVAILABLE, - format!("home sled `{home}` unknown to the proxy"), + format!("The proxy's own sled `{home}` is not reachable"), )) }), None => match sleds.iter().next() { Some((_, addr)) => Ok(*addr), None => Err(Box::new(error_response( StatusCode::SERVICE_UNAVAILABLE, - "no sleds known to the proxy", + "No sleds are reachable", ))), }, } @@ -358,7 +359,7 @@ async fn forward(log: Logger, addr: SocketAddr, request: Request) -> R Ok(response) => response.map(Either::Left), Err(err) => error_response( StatusCode::BAD_GATEWAY, - format!("unable to reach sled at `{addr}`: {err}"), + format!("Unable to reach sled at `{addr}`: {err}"), ), } } @@ -408,10 +409,12 @@ async fn bridge(log: Logger, client: OnUpgrade, sled: OnUpgrade) { } } -fn error_response(status: StatusCode, message: impl Into) -> Response { +fn error_response(status: StatusCode, message: impl Into) -> Response { + let body = json!({ "request_id": "", "message": message.into() }); Response::builder() .status(status) - .body(Either::Right(Full::new(message.into()))) + .header(CONTENT_TYPE, "application/json") + .body(Either::Right(Full::new(body.to_string().into()))) .expect("error response should build") } @@ -427,6 +430,19 @@ mod test { s.parse().unwrap() } + /// Check that proxy errors carry the fields clients require. + #[tokio::test] + async fn error_body() { + use http_body_util::BodyExt as _; + + let response = error_response(StatusCode::BAD_GATEWAY, "no route"); + assert_eq!(response.headers()[CONTENT_TYPE], "application/json"); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["message"], "no route"); + assert_eq!(body["request_id"], ""); + } + /// Unrouted requests go to the home sled, or fail while the /// targets don't know it. Without a home, any sled will do. #[test] From fb1c1d84090d6a17d3c7b549d41360902350f816 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 14:51:56 -0600 Subject: [PATCH 02/20] Explain why `job attach` has nothing to attach to Fixes #74. Co-authored-by: Claude --- client/src/commands.rs | 40 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index f5522bb0..7c6a9fa4 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1441,8 +1441,14 @@ async fn job( }, TargetArg::Target(target) => resolve_target(client, target).await?, }; - job_attach(ctx, client, &job_id, &target).await?; - Ok(()) + match job_attach(ctx, client, &job_id, &target).await { + Err(CommandError::NotFound(_)) => { + let error = attach_refused(ctx, client, &job_id, &target).await; + let _ = ctx.job_error(error); + Ok(()) + } + result => result, + } } (JobCommand::History { limit, offset }, Some(client)) => { @@ -2298,6 +2304,30 @@ async fn byte_stream_to_file( } } +/// Explain, from the job's status on `target`, why an attach to +/// `job_id` there found nothing to attach to. +async fn attach_refused( + ctx: &mut impl CommandContext, + client: &Client, + job_id: &JobId, + target: &BaseboardId, +) -> CommandError { + let status = match job_status_map(ctx, client, job_id, Some(target)).await { + Ok(status) => status, + Err(error) => return error, + }; + match status.get(target) { + None => CommandError::JobNotOnSled(*job_id, target.clone()), + Some(JobStatus::Queued { .. }) => CommandError::JobNotYetRunning(*job_id), + Some(JobStatus::Started { .. }) => CommandError::JobNotInteractive(*job_id), + Some(JobStatus::Cancelled { .. }) => CommandError::JobCancelled(*job_id), + Some(JobStatus::Skipped { reason, .. }) => CommandError::JobSkipped(*job_id, *reason), + Some(JobStatus::Stopped { .. } | JobStatus::Error { .. }) => { + CommandError::JobFinished(*job_id) + } + } +} + /// Connect via WebSockets to a running interactive job. async fn job_attach( ctx: &mut impl CommandContext, @@ -2589,6 +2619,12 @@ pub enum CommandError { JobCancelled(JobId), #[error("❌ Job `{0}` did not run")] JobDidNotRun(JobId), + #[error("❌ Job `{0}` has finished, try `job status` or `job output`")] + JobFinished(JobId), + #[error("❌ Job `{0}` is not interactive, try `job status` or `job output`")] + JobNotInteractive(JobId), + #[error("❌ Job `{0}` not found on sled `{1}`")] + JobNotOnSled(JobId, BaseboardId), #[error("❌ Job `{0}` is not yet running")] JobNotYetRunning(JobId), #[error("⏩ Job `{0}` was skipped on this sled: {1}")] From eda58bbe816a8c723a5c9d699e7ba32c88983b53 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 15:03:26 -0600 Subject: [PATCH 03/20] Give up on sleds that don't answer the proxy Co-authored-by: Claude --- server/src/proxy.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/server/src/proxy.rs b/server/src/proxy.rs index 2d476f07..edf0f5b9 100644 --- a/server/src/proxy.rs +++ b/server/src/proxy.rs @@ -17,6 +17,7 @@ use std::convert::Infallible; use std::io; use std::net::SocketAddr; use std::sync::Arc; +use std::time::Duration; use camino::{Utf8Path, Utf8PathBuf}; use http::header::CONTENT_TYPE; @@ -42,6 +43,7 @@ use slog::{Logger, debug, error, info, o, warn}; use tokio::io::{AsyncRead, AsyncWrite, copy_bidirectional}; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::watch; +use tokio::time::timeout; use tokio::{select, spawn, try_join}; use tokio_rustls::TlsAcceptor; use tokio_util::sync::CancellationToken; @@ -52,6 +54,9 @@ use sush_common::version::VersionInfo; /// Server addresses by baseboard. pub type Sleds = BTreeMap; +/// How long to wait for a sled to accept a connection. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); + /// The sush servers the proxy may route to. The sled and cubby maps are /// discovered independently (by DDM and MGS, resp.), so each arrives on /// its own channel. @@ -371,7 +376,9 @@ async fn try_forward( addr: SocketAddr, mut request: Request, ) -> Result, Box> { - let stream = TcpStream::connect(addr).await?; + let stream = timeout(CONNECT_TIMEOUT, TcpStream::connect(addr)) + .await + .map_err(|_| format!("no answer within {}s", CONNECT_TIMEOUT.as_secs()))??; let (mut sender, conn) = client_conn::handshake(TokioIo::new(stream)).await?; spawn({ let log = log.clone(); From 39cfc1bd3d9e92482c63c987a21c9a505e115019 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 15:09:35 -0600 Subject: [PATCH 04/20] Let a second interrupt end `job start -w` while a stop is pending Co-authored-by: Claude --- client/src/commands.rs | 50 ++++++++++++++++++++++++------------------ 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 7c6a9fa4..855682ad 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -102,6 +102,7 @@ const MAX_PARALLEL_CHUNKS: NonZeroU8 = NonZeroU8::new(64).unwrap(); // Job polling and spinner update intervals. const JOB_STOP_RETRY_INTERVAL: Duration = Duration::from_millis(100); + #[cfg(feature = "permslip")] const SIGNING_UPDATE_INTERVAL: Duration = Duration::from_millis(100); @@ -1600,8 +1601,22 @@ async fn job_start( let mut settling = Settling::default(); let mut started = false; let mut stalled = false; + let mut stopping = false; let mut stopped = false; let mut sigint = signal(SignalKind::interrupt())?; + let mut stop_ctx = ctx.clone(); + let via = job_target.single_baseboard(); + let stop = async { + // A stop can race the start, so retry it a few times. + for _ in 0..3 { + match job_stop(&mut stop_ctx, client, &job_id, via).await { + Err(CommandError::NotFound(_)) => sleep(JOB_STOP_RETRY_INTERVAL).await, + result => return result.map(|()| true), + } + } + Ok(false) + }; + pin!(stop); let status = loop { select! { // Wait for the start request to finish. @@ -1646,32 +1661,25 @@ async fn job_start( } } - // While the job runs, an interrupt stops it but keeps - // watching: the start future must still resolve, and the - // stops are worth seeing. There is a race with the start, - // so retry the stop a few times if needed. Once the job - // has stopped, or after a first interrupt, an interrupt - // just ends the watch. + // While the job runs, a first interrupt stops it and keeps + // watching; a second ends the watch. _ = sigint.recv() => { - if started || stopped { + if started || stopping { break last; } - for _ in 0..3 { - match job_stop(ctx, client, &job_id, job_target.single_baseboard()).await { - Ok(_) => { - ctx.job_stopped(&job_id); - stopped = true; - break; - } - Err(CommandError::NotFound(_)) => { - sleep(JOB_STOP_RETRY_INTERVAL).await; - } - Err(error) => { - ctx.job_watch_finished(&job_id); - return Err(error); - } + stopping = true; + } + + stop_result = &mut stop, if stopping && !stopped => { + match stop_result { + Ok(true) => ctx.job_stopped(&job_id), + Ok(false) => (), + Err(error) => { + ctx.job_watch_finished(&job_id); + return Err(error); } } + stopped = true; } } }; From 15c24275c07df70d5a68ad23b608e3919ff6c061 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 17:19:21 -0600 Subject: [PATCH 05/20] Make `job skip` default to the next job Co-authored-by: Claude --- client/src/commands.rs | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 855682ad..3ba2dfca 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -617,11 +617,10 @@ pub enum JobCommand { job_id: JobId, }, - /// Decline a job, burning its ID and advancing the session. + /// Decline to run a job, burning its ID and advancing the session. Skip { - /// The job to skip. - #[clap(env = SUSH_JOB_ID)] - job_id: JobId, + /// The job to skip. By default, the next job in the current session. + job_id: Option, }, /// Get the status of a started job. @@ -1368,7 +1367,13 @@ async fn job( }; if let Some(client) = client { ctx.job_signed(&job, false); - job_start(ctx, client, job, start_args.to_owned()).await + let job_id = *job.job_id(); + match job_start(ctx, client, job, start_args.to_owned()).await { + Err(error) if ctx.next_job_id().ok() == Some(job_id) => { + Err(CommandError::JobNotStarted(job_id, Box::new(error))) + } + result => result, + } } else { ctx.job_signed(&job, true); ctx.job_started(&job, None, false); @@ -1388,6 +1393,10 @@ async fn job( let Some(session_id) = ctx.session_id() else { return Err(CommandError::MissingSession); }; + let job_id = match job_id { + Some(job_id) => job_id, + None => ctx.next_job_id()?, + }; if let Some(client) = client { with_login(ctx, client, async || { client @@ -2633,6 +2642,8 @@ pub enum CommandError { JobNotInteractive(JobId), #[error("❌ Job `{0}` not found on sled `{1}`")] JobNotOnSled(JobId, BaseboardId), + #[error("{1}\n❗ Job `{0}` was signed but not started, try `job skip`")] + JobNotStarted(JobId, Box), #[error("❌ Job `{0}` is not yet running")] JobNotYetRunning(JobId), #[error("⏩ Job `{0}` was skipped on this sled: {1}")] From 9d6cfaf356ce292f7fb6204e5b18db9a90fefa45 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 17:29:41 -0600 Subject: [PATCH 06/20] Show interactive job output if it finishes before we attach Fixes #74. Co-authored-by: Claude --- client/src/commands.rs | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 3ba2dfca..7015c80c 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1548,11 +1548,25 @@ async fn job_start( let sleds = start.await?.into_inner().sleds; ctx.job_started(&job, Some(&sleds), true); match job_attach(ctx, client, &job_id, &target).await { - Ok(()) | Err(CommandError::NotFound(_)) => { - job_status(ctx, client, &job_id, StatusDisplayStyle::Short).await? + Ok(()) => (), + Err(CommandError::NotFound(_)) => { + let output = JobOutput { + job_id, + binary: true, + file: None, + force: false, + chunk_size: DEFAULT_CHUNK_SIZE, + parallel: PARALLEL_CHUNKS, + }; + if let Err(error) = job_output_from(ctx, client, &target, Stdout, output).await + && !error.is_unfinished_job() + { + let _ = ctx.job_error(error); + } } Err(error) => return Err(error), } + job_status(ctx, client, &job_id, StatusDisplayStyle::Short).await? } else if streaming { let Some(path) = file else { return Err(CommandError::StreamingNeedsFile); @@ -2744,6 +2758,11 @@ impl CommandError { error, } } + + /// Whether this error says the job has not yet finished. + pub fn is_unfinished_job(&self) -> bool { + matches!(self, Self::JobNotYetRunning(_) | Self::JobStillRunning(_)) + } } impl From> for CommandError { From 035b9568b7cf5a26a90fa457937ad06fce26c88a Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 17:54:57 -0600 Subject: [PATCH 07/20] Exit quietly when stdout is a closed pipe Fixes #72. Co-authored-by: Claude --- client/src/cli.rs | 145 ++++++++++++++++++++++----------------------- client/src/lib.rs | 41 +++++++++++++ client/src/repl.rs | 4 +- clippy.toml | 20 +++++++ common/build.rs | 3 + 5 files changed, 138 insertions(+), 75 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index 0148bc7f..792a70c7 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -7,13 +7,12 @@ use std::collections::{BTreeMap, BTreeSet}; use std::fmt; use std::fs; -use std::io::{self, BufRead as _, ErrorKind, Read as _, Write as _, stderr, stdin, stdout}; +use std::io::{self, BufRead as _, ErrorKind, Read as _, Write as _, stderr, stdin}; use std::os::unix::fs::OpenOptionsExt as _; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime}; -use anstream::print; use anstyle::{AnsiColor, Style}; use atomicwrites::{AtomicFile, OverwriteBehavior}; use bytesize::ByteSize; @@ -38,9 +37,9 @@ use sush_common::keys::{KeyId, Signature, SshPublicKey}; use sush_common::targets::{MAX_CUBBY, SledHealth, SledId, SledVersion, Target}; use sush_common::version::VersionInfo; -use crate::AuthzSigner; use crate::commands::{CommandError, GlobalArgs}; use crate::context::{CommandContext, OutputFormat, StatusDisplayStyle}; +use crate::{AuthzSigner, quiet_write}; pub(crate) const PREFIX: &str = "sush"; const SESSION_FILE_NAME: &str = "session.json"; @@ -248,17 +247,17 @@ impl CommandContext for Cli { sleds: &[SledVersion], ) { match self.get_output_format() { - OutputFormat::Json => println!( + OutputFormat::Json => qprintln!( "{}", json!({"client": client, "server": server, "sleds": sleds}) ), OutputFormat::Text => { - println!("Client:\t{client}"); + qprintln!("Client:\t{client}"); if let Some(server) = server { - println!("Server:\t{server}"); + qprintln!("Server:\t{server}"); } if !sleds.is_empty() { - print!("{}", draw_rack(sleds)); + qprint!("{}", draw_rack(sleds)); } } } @@ -319,7 +318,7 @@ impl CommandContext for Cli { sush_nonce: SessionSushNonce, ) { match self.get_output_format() { - OutputFormat::Json => println!( + OutputFormat::Json => qprintln!( "{}", json!({ "baseboard_id": &baseboard_id.to_string(), @@ -328,16 +327,16 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - println!("✅ Please relay these parameters to support:"); - println!(" Baseboard ID: {baseboard_id}"); - println!(" Sush nonce: {sush_nonce}"); + qprintln!("✅ Please relay these parameters to support:"); + qprintln!(" Baseboard ID: {baseboard_id}"); + qprintln!(" Sush nonce: {sush_nonce}"); } } } fn job_start_params(&self, request: &JobStartRequest) { match self.get_output_format() { - OutputFormat::Json => println!( + OutputFormat::Json => qprintln!( "{}", json!({ "job_id": &request.job_id, @@ -348,12 +347,12 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - println!("✅ Please request a job with these parameters from support:"); - println!(" Job ID: {}", request.job_id); - println!(" Session ID: {}", request.session_id); - println!(" Command: {}", request.command); - println!(" Mode: {}", request.mode.as_str()); - println!(" Target: {}", request.target); + qprintln!("✅ Please request a job with these parameters from support:"); + qprintln!(" Job ID: {}", request.job_id); + qprintln!(" Session ID: {}", request.session_id); + qprintln!(" Command: {}", request.command); + qprintln!(" Mode: {}", request.mode.as_str()); + qprintln!(" Target: {}", request.target); } } } @@ -366,14 +365,14 @@ impl CommandContext for Cli { ) { let session_id = self.adopt_session(session, true); match self.get_output_format() { - OutputFormat::Json => println!( + OutputFormat::Json => qprintln!( "{}", json!({"session_created": session_id, "signer_nonce": signer_nonce, "seq": seq}) ), OutputFormat::Text => { - println!("✅ Session is now `{session_id}`"); - println!(" Signer nonce: {signer_nonce}"); - println!(" Sequence number: {seq}"); + qprintln!("✅ Session is now `{session_id}`"); + qprintln!(" Signer nonce: {signer_nonce}"); + qprintln!(" Sequence number: {seq}"); } } } @@ -381,8 +380,8 @@ impl CommandContext for Cli { fn session_started(&mut self, session: Session, force: bool) { let session_id = self.adopt_session(session, force); match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"session_started": session_id})), - OutputFormat::Text => println!("✅ Session is now `{session_id}`"), + OutputFormat::Json => qprintln!("{}", json!({"session_started": session_id})), + OutputFormat::Text => qprintln!("✅ Session is now `{session_id}`"), } } @@ -395,26 +394,26 @@ impl CommandContext for Cli { self.save_session(None); } match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"session_stopped": session_id})), - OutputFormat::Text => println!("✅ Stopped session `{session_id}`"), + OutputFormat::Json => qprintln!("{}", json!({"session_stopped": session_id})), + OutputFormat::Text => qprintln!("✅ Stopped session `{session_id}`"), } } fn attach_allowed(&mut self, key_id: &KeyId, access: Access) { match self.get_output_format() { OutputFormat::Json => { - println!("{}", json!({"attach_allowed": key_id, "access": access})) + qprintln!("{}", json!({"attach_allowed": key_id, "access": access})) } OutputFormat::Text => { - println!("✅ Allowed {} attach for `{key_id}`", access.as_str()) + qprintln!("✅ Allowed {} attach for `{key_id}`", access.as_str()) } } } fn attach_denied(&mut self, key_id: &KeyId) { match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"attach_denied": key_id})), - OutputFormat::Text => println!("✅ Denied attach for `{key_id}`"), + OutputFormat::Json => qprintln!("{}", json!({"attach_denied": key_id})), + OutputFormat::Text => qprintln!("✅ Denied attach for `{key_id}`"), } } @@ -463,15 +462,15 @@ impl CommandContext for Cli { } match self.get_output_format() { - OutputFormat::Json => println!("{}", json!(certs)), + OutputFormat::Json => qprintln!("{}", json!(certs)), OutputFormat::Text => { if roots.is_empty() { - println!("✅ Verified chain consistency"); + qprintln!("✅ Verified chain consistency"); } else { - println!("✅ Verified chain against supplied root"); + qprintln!("✅ Verified chain against supplied root"); } for cert in &chain { - println!(" {}", cert.tbs_certificate.subject); + qprintln!(" {}", cert.tbs_certificate.subject); } } } @@ -491,14 +490,14 @@ impl CommandContext for Cli { } let job_id = job.job_id(); match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"job_started": job_id, "sleds": sleds})), + OutputFormat::Json => qprintln!("{}", json!({"job_started": job_id, "sleds": sleds})), OutputFormat::Text => { let watch = self.watch.lock().unwrap(); let print = |line: String| match watch.as_ref() { Some(watch) => { let _ = watch.multi.println(line); } - None => println!("{line}"), + None => qprintln!("{line}"), }; print(format!("✅ Started job `{job_id}`")); if let (Target::Sleds(named), Some(sleds)) = (job.payload().target(), sleds) { @@ -518,12 +517,12 @@ impl CommandContext for Cli { fn job_stopped(&mut self, job_id: &JobId) { match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"job_stopped": job_id})), + OutputFormat::Json => qprintln!("{}", json!({"job_stopped": job_id})), OutputFormat::Text => { if let Some(watch) = self.watch.lock().unwrap().as_ref() { let _ = watch.multi.println(format!("✅ Stopped job `{job_id}`")); } else { - println!("\r✅ Stopped job `{job_id}`"); + qprintln!("\r✅ Stopped job `{job_id}`"); } } } @@ -538,8 +537,8 @@ impl CommandContext for Cli { if skipped { self.save_session(session_guard.as_ref()); match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"job_skipped": job_id})), - OutputFormat::Text => println!("✅ Skipped job `{job_id}`"), + OutputFormat::Json => qprintln!("{}", json!({"job_skipped": job_id})), + OutputFormat::Text => qprintln!("✅ Skipped job `{job_id}`"), } } skipped @@ -552,8 +551,8 @@ impl CommandContext for Cli { fn job_output_target(&mut self, target: &BaseboardId) { match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({ "target": target.to_string() })), - OutputFormat::Text => println!(" » {target} «"), + OutputFormat::Json => qprintln!("{}", json!({ "target": target.to_string() })), + OutputFormat::Text => qprintln!(" » {target} «"), } } @@ -565,24 +564,24 @@ impl CommandContext for Cli { binary: bool, ) { match self.get_output_format() { - OutputFormat::Json if binary => println!("{}", json!(output)), - OutputFormat::Json => println!("{}", json!(String::from_utf8_lossy(output))), + OutputFormat::Json if binary => qprintln!("{}", json!(output)), + OutputFormat::Json => qprintln!("{}", json!(String::from_utf8_lossy(output))), OutputFormat::Text if binary => { - if let Err(err) = stdout().write_all(output) { + if let Err(err) = quiet_write(output) { eprintln!("{err}"); } } OutputFormat::Text if output.is_empty() => (), OutputFormat::Text => { match stream { - JobOutputStream::Stdout => println!("✅ Job stdout:"), - JobOutputStream::Stderr => println!("✅ Job stderr:"), + JobOutputStream::Stdout => qprintln!("✅ Job stdout:"), + JobOutputStream::Stderr => qprintln!("✅ Job stderr:"), }; let output = String::from_utf8_lossy(output); if output.ends_with('\n') { - print!("{output}"); + qprint!("{output}"); } else { - println!("{output}"); + qprintln!("{output}"); } } } @@ -648,7 +647,7 @@ impl CommandContext for Cli { if let Some(stage) = stage { let length = ByteSize::b(progress.length().unwrap_or(progress.position())); let elapsed = progress.elapsed(); - println!( + qprintln!( "{stage} {stream}:\t{} in {} ({:.0} MB/s)", length.display().si(), format_duration(progress.elapsed()), @@ -727,17 +726,17 @@ impl CommandContext for Cli { fn job_attached(&mut self, job_id: &JobId) { match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"attached": job_id})), + OutputFormat::Json => qprintln!("{}", json!({"attached": job_id})), OutputFormat::Text => { - println!("✅ Attached to interactive job `{job_id}`, detach with `^]`") + qprintln!("✅ Attached to interactive job `{job_id}`, detach with `^]`") } } } fn job_detached(&mut self, job_id: &JobId) { match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"detached": job_id})), - OutputFormat::Text => println!("\r✅ Detached from interactive job `{job_id}`"), + OutputFormat::Json => qprintln!("{}", json!({"detached": job_id})), + OutputFormat::Text => qprintln!("\r✅ Detached from interactive job `{job_id}`"), } } @@ -771,18 +770,18 @@ impl CommandContext for Cli { progress.finish_and_clear(); } if matches!(self.get_output_format(), OutputFormat::Text) { - println!("✅ Signed request for job `{job_id}`"); + qprintln!("✅ Signed request for job `{job_id}`"); } } fn job_signed(&mut self, job: &SignedJob, show: bool) { if show { match self.get_output_format() { - OutputFormat::Json => println!( + OutputFormat::Json => qprintln!( "{}", to_json_string(&job).expect("job should be JSON-serializable") ), - OutputFormat::Text => println!( + OutputFormat::Text => qprintln!( "{}", to_json_string_pretty(&job).expect("job should be JSON-serializable") ), @@ -792,7 +791,7 @@ impl CommandContext for Cli { fn job_status(&mut self, job_id: &JobId, status: &JobStatusMap, style: StatusDisplayStyle) { match self.get_output_format() { - OutputFormat::Json => println!("{}", json!(job_status_to_json_map(status.clone()))), + OutputFormat::Json => qprintln!("{}", json!(job_status_to_json_map(status.clone()))), OutputFormat::Text if matches!(style, StatusDisplayStyle::Short) => { let icon = if status.values().any(|s| { matches!( @@ -811,10 +810,10 @@ impl CommandContext for Cli { .map(|b| b.to_string().len()) .max() .unwrap_or(0); - println!("{icon} Job ID:\t{job_id}"); + qprintln!("{icon} Job ID:\t{job_id}"); for (baseboard_id, status) in status { let id = baseboard_id.to_string(); - println!(" {id: { @@ -825,7 +824,7 @@ impl CommandContext for Cli { time_cancelled, actor, } => { - println!( + qprintln!( "❌ Job ID:\t{job_id}\n \ Target:\t{baseboard_id}\n \ Job status:\tCancelled\n \ @@ -838,7 +837,7 @@ impl CommandContext for Cli { time_queued, actor, } => { - println!( + qprintln!( "⏳ Job ID:\t{job_id}\n \ Target:\t{baseboard_id}\n \ Job status:\tQueued\n \ @@ -850,7 +849,7 @@ impl CommandContext for Cli { job_id, time_started, } => { - println!( + qprintln!( "✅ Job ID:\t{job_id}\n \ Target:\t{baseboard_id}\n \ Job status:\tStarted\n \ @@ -873,7 +872,7 @@ impl CommandContext for Cli { let duration = format_elapsed_duration(status.time_elapsed()); let stdout_len = byte_size(*stdout_len); let stderr_len = byte_size(*stderr_len); - println!( + qprintln!( "✅ Job ID:\t{job_id}\n \ Target:\t{baseboard_id}\n \ Job status:\tStopped\n \ @@ -902,7 +901,7 @@ impl CommandContext for Cli { let duration = format_elapsed_duration(status.time_elapsed()); let stdout_len = byte_size(*stdout_len); let stderr_len = byte_size(*stderr_len); - println!( + qprintln!( "✅ Job ID:\t{job_id}\n \ Target:\t{baseboard_id}\n \ Job status:\t{err}\n \ @@ -919,7 +918,7 @@ impl CommandContext for Cli { time_error, error, } => { - println!( + qprintln!( "❌ Job ID:\t{job_id}\n \ Target:\t{baseboard_id}\n \ Job status:\tError\n \ @@ -932,7 +931,7 @@ impl CommandContext for Cli { time_skipped, reason, } => { - println!( + qprintln!( "⏩ Job ID:\t{job_id}\n \ Target:\t{baseboard_id}\n \ Job status:\tSkipped\n \ @@ -958,7 +957,7 @@ impl CommandContext for Cli { fn iam(&mut self, identity: &Identity) -> Result<(), CommandError> { match self.get_output_format() { - OutputFormat::Json => println!("{}", to_json_string(identity)?), + OutputFormat::Json => qprintln!("{}", to_json_string(identity)?), OutputFormat::Text => { let Identity { key_id, @@ -969,7 +968,7 @@ impl CommandContext for Cli { let fingerprint = public_key.fingerprint(); let algorithm = public_key.algorithm(); let comment = public_key.comment(); - println!( + qprintln!( "✅ Key ID:\t{key_id}\n \ Fingerprint:\t{fingerprint}\n \ Algorithm:\t{algorithm}\n \ @@ -986,8 +985,8 @@ impl CommandContext for Cli { let n = keys.len(); if n == 0 { match self.get_output_format() { - OutputFormat::Json => println!("[]"), - OutputFormat::Text => println!("❌ No SSH identities found"), + OutputFormat::Json => qprintln!("[]"), + OutputFormat::Text => qprintln!("❌ No SSH identities found"), } } else { for key in keys { @@ -996,7 +995,7 @@ impl CommandContext for Cli { let algorithm = key.algorithm(); let comment = key.comment(); match self.get_output_format() { - OutputFormat::Json => println!( + OutputFormat::Json => qprintln!( "{}", json!({ "key_id": key_id, @@ -1006,7 +1005,7 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - println!( + qprintln!( "✅ Key ID:\t{key_id}\n \ Fingerprint:\t{fingerprint}\n \ Algorithm:\t{algorithm}\n \ diff --git a/client/src/lib.rs b/client/src/lib.rs index 0639f61b..2b907f51 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -2,6 +2,9 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. +use std::fmt::Arguments; +use std::io::{self, ErrorKind, Write as _, stdout}; +use std::process::exit; use std::sync::{Arc, Mutex}; use reqwest::header::AUTHORIZATION; @@ -9,6 +12,44 @@ use sush_common::authn::AuthnError; use crate::context::Authz; +/// Like `print!`, but exit quietly if stdout is a closed pipe. +#[macro_export] +macro_rules! qprint { + ($($arg:tt)*) => { + $crate::quiet_print(::std::format_args!($($arg)*)) + }; +} + +/// Like `println!`, but exit quietly if stdout is a closed pipe. +#[macro_export] +macro_rules! qprintln { + () => { + $crate::quiet_print(::std::format_args!("\n")) + }; + ($($arg:tt)*) => { + $crate::quiet_print(::std::format_args!("{}\n", ::std::format_args!($($arg)*))) + }; +} + +#[doc(hidden)] +pub fn quiet_print(args: Arguments<'_>) { + if let Err(error) = anstream::stdout().lock().write_fmt(args) { + if error.kind() == ErrorKind::BrokenPipe { + exit(0); + } + panic!("failed printing to stdout: {error}"); + } +} + +/// Write `bytes` to standard output (unaltered), but exit quietly if +/// it is a closed pipe. +pub fn quiet_write(bytes: &[u8]) -> io::Result<()> { + match stdout().lock().write_all(bytes) { + Err(error) if error.kind() == ErrorKind::BrokenPipe => exit(0), + result => result, + } +} + pub mod agent; pub mod cli; pub mod commands; diff --git a/client/src/repl.rs b/client/src/repl.rs index 81380a19..4aaf5800 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -197,11 +197,11 @@ impl CommandContext for Repl { if let Some(arg) = $arg { unsafe { env::set_var($var, arg.to_string()) } args.$arg = Some(arg.clone()); - println!("✅ {} set to `{}`", $name, arg); + qprintln!("✅ {} set to `{}`", $name, arg); } else { unsafe { env::remove_var($var) } args.$arg = None; - println!("✅ {} unset", $name); + qprintln!("✅ {} unset", $name); } }}; } diff --git a/clippy.toml b/clippy.toml index 441096f8..82debe19 100644 --- a/clippy.toml +++ b/clippy.toml @@ -11,3 +11,23 @@ replacement = "sush_common::keys::SshPublicKey" path = "ssh_key::Signature" reason = "use wrapper instead" replacement = "sush_common::keys::Signature" + +[[disallowed-macros]] +path = "std::print" +reason = "panics on a closed pipe" +replacement = "qprint" + +[[disallowed-macros]] +path = "std::println" +reason = "panics on a closed pipe" +replacement = "qprintln" + +[[disallowed-macros]] +path = "anstream::print" +reason = "panics on a closed pipe" +replacement = "qprint" + +[[disallowed-macros]] +path = "anstream::println" +reason = "panics on a closed pipe" +replacement = "qprintln" diff --git a/common/build.rs b/common/build.rs index a56eb616..1da53625 100644 --- a/common/build.rs +++ b/common/build.rs @@ -5,6 +5,9 @@ //! Embed the git commit, "-dirty" suffixed when the tree has //! uncommitted changes. +// Cargo reads build script instructions from stdout. +#![allow(clippy::disallowed_macros)] + use std::process::Command; fn main() { From 8164eec1a39b575cec3440db635564a5d756c802 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 18:59:24 -0600 Subject: [PATCH 08/20] Print one keyed JSON object per line in JSON mode Fixes #70. Co-authored-by: Claude --- client/src/cli.rs | 166 +++++++++++++++++++++++++++++------------ client/src/commands.rs | 4 +- client/src/context.rs | 12 ++- client/src/repl.rs | 11 ++- 4 files changed, 140 insertions(+), 53 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index 792a70c7..ab166bb8 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -15,6 +15,8 @@ use std::time::{Duration, SystemTime}; use anstyle::{AnsiColor, Style}; use atomicwrites::{AtomicFile, OverwriteBehavior}; +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD; use bytesize::ByteSize; use chrono::TimeDelta; use humantime::format_duration; @@ -249,7 +251,23 @@ impl CommandContext for Cli { match self.get_output_format() { OutputFormat::Json => qprintln!( "{}", - json!({"client": client, "server": server, "sleds": sleds}) + json!({ + "version": { + "client": client, + "server": server, + "sleds": sleds + .iter() + .map(|sled| { + json!({ + "cubby": sled.cubby, + "baseboard": sled.baseboard.to_string(), + "version": sled.version, + "health": sled.health, + }) + }) + .collect::>(), + } + }) ), OutputFormat::Text => { qprintln!("Client:\t{client}"); @@ -321,9 +339,11 @@ impl CommandContext for Cli { OutputFormat::Json => qprintln!( "{}", json!({ - "baseboard_id": &baseboard_id.to_string(), - "client_nonce": &client_nonce, - "sush_nonce": &sush_nonce, + "session_start_params": { + "baseboard_id": &baseboard_id.to_string(), + "client_nonce": &client_nonce, + "sush_nonce": &sush_nonce, + } }) ), OutputFormat::Text => { @@ -339,11 +359,13 @@ impl CommandContext for Cli { OutputFormat::Json => qprintln!( "{}", json!({ - "job_id": &request.job_id, - "session_id": &request.session_id, - "command": &request.command, - "mode": &request.mode, - "target": &request.target.to_string(), + "job_start_params": { + "job_id": &request.job_id, + "session_id": &request.session_id, + "command": &request.command, + "mode": &request.mode, + "target": &request.target.to_string(), + } }) ), OutputFormat::Text => { @@ -367,7 +389,13 @@ impl CommandContext for Cli { match self.get_output_format() { OutputFormat::Json => qprintln!( "{}", - json!({"session_created": session_id, "signer_nonce": signer_nonce, "seq": seq}) + json!({ + "session_created": { + "session_id": session_id, + "signer_nonce": signer_nonce, + "seq": seq, + } + }) ), OutputFormat::Text => { qprintln!("✅ Session is now `{session_id}`"); @@ -402,7 +430,15 @@ impl CommandContext for Cli { fn attach_allowed(&mut self, key_id: &KeyId, access: Access) { match self.get_output_format() { OutputFormat::Json => { - qprintln!("{}", json!({"attach_allowed": key_id, "access": access})) + qprintln!( + "{}", + json!({ + "attach_allowed": { + "key_id": key_id, + "access": access, + } + }) + ) } OutputFormat::Text => { qprintln!("✅ Allowed {} attach for `{key_id}`", access.as_str()) @@ -462,7 +498,7 @@ impl CommandContext for Cli { } match self.get_output_format() { - OutputFormat::Json => qprintln!("{}", json!(certs)), + OutputFormat::Json => qprintln!("{}", json!({"certs": certs})), OutputFormat::Text => { if roots.is_empty() { qprintln!("✅ Verified chain consistency"); @@ -490,7 +526,17 @@ impl CommandContext for Cli { } let job_id = job.job_id(); match self.get_output_format() { - OutputFormat::Json => qprintln!("{}", json!({"job_started": job_id, "sleds": sleds})), + OutputFormat::Json => qprintln!( + "{}", + json!({ + "job_started": { + "job_id": job_id, + "sleds": sleds.map(|sleds| { + sleds.iter().map(ToString::to_string).collect::>() + }), + } + }) + ), OutputFormat::Text => { let watch = self.watch.lock().unwrap(); let print = |line: String| match watch.as_ref() { @@ -551,21 +597,39 @@ impl CommandContext for Cli { fn job_output_target(&mut self, target: &BaseboardId) { match self.get_output_format() { - OutputFormat::Json => qprintln!("{}", json!({ "target": target.to_string() })), + OutputFormat::Json => (), OutputFormat::Text => qprintln!(" » {target} «"), } } fn job_output( &mut self, - _job_id: &JobId, + job_id: &JobId, + target: &BaseboardId, stream: JobOutputStream, output: &[u8], binary: bool, ) { match self.get_output_format() { - OutputFormat::Json if binary => qprintln!("{}", json!(output)), - OutputFormat::Json => qprintln!("{}", json!(String::from_utf8_lossy(output))), + OutputFormat::Json => { + let (output, encoding) = if binary { + (STANDARD.encode(output), "base64") + } else { + (String::from_utf8_lossy(output).into_owned(), "utf-8") + }; + qprintln!( + "{}", + json!({ + "job_output": { + "job_id": job_id, + "target": target.to_string(), + "stream": stream, + "encoding": encoding, + "output": output, + } + }) + ) + } OutputFormat::Text if binary => { if let Err(err) = quiet_write(output) { eprintln!("{err}"); @@ -791,7 +855,15 @@ impl CommandContext for Cli { fn job_status(&mut self, job_id: &JobId, status: &JobStatusMap, style: StatusDisplayStyle) { match self.get_output_format() { - OutputFormat::Json => qprintln!("{}", json!(job_status_to_json_map(status.clone()))), + OutputFormat::Json => qprintln!( + "{}", + json!({ + "job_status": { + "job_id": job_id, + "sleds": job_status_to_json_map(status.clone()), + } + }) + ), OutputFormat::Text if matches!(style, StatusDisplayStyle::Short) => { let icon = if status.values().any(|s| { matches!( @@ -957,7 +1029,7 @@ impl CommandContext for Cli { fn iam(&mut self, identity: &Identity) -> Result<(), CommandError> { match self.get_output_format() { - OutputFormat::Json => qprintln!("{}", to_json_string(identity)?), + OutputFormat::Json => qprintln!("{}", json!({"identity": identity})), OutputFormat::Text => { let Identity { key_id, @@ -982,36 +1054,34 @@ impl CommandContext for Cli { } fn identities(&mut self, keys: &[SshPublicKey]) -> Result<(), CommandError> { - let n = keys.len(); - if n == 0 { - match self.get_output_format() { - OutputFormat::Json => qprintln!("[]"), - OutputFormat::Text => qprintln!("❌ No SSH identities found"), - } - } else { - for key in keys { - let key_id = key.key_id()?; - let fingerprint = key.fingerprint(); - let algorithm = key.algorithm(); - let comment = key.comment(); - match self.get_output_format() { - OutputFormat::Json => qprintln!( - "{}", - json!({ - "key_id": key_id, - "fingerprint": fingerprint, - "algorithm": algorithm.to_string(), + match self.get_output_format() { + OutputFormat::Json => { + let identities = keys + .iter() + .map(|key| { + Ok(json!({ + "key_id": key.key_id()?, + "fingerprint": key.fingerprint(), + "algorithm": key.algorithm().to_string(), "public_key": key.to_openssh()?, - }) - ), - OutputFormat::Text => { - qprintln!( - "✅ Key ID:\t{key_id}\n \ - Fingerprint:\t{fingerprint}\n \ - Algorithm:\t{algorithm}\n \ - Comment:\t{comment}", - ); - } + })) + }) + .collect::, CommandError>>()?; + qprintln!("{}", json!({"identities": identities})); + } + OutputFormat::Text if keys.is_empty() => qprintln!("❌ No SSH identities found"), + OutputFormat::Text => { + for key in keys { + let key_id = key.key_id()?; + let fingerprint = key.fingerprint(); + let algorithm = key.algorithm(); + let comment = key.comment(); + qprintln!( + "✅ Key ID:\t{key_id}\n \ + Fingerprint:\t{fingerprint}\n \ + Algorithm:\t{algorithm}\n \ + Comment:\t{comment}", + ); } } } diff --git a/client/src/commands.rs b/client/src/commands.rs index 7015c80c..061d852f 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1738,7 +1738,7 @@ async fn job_start( { Ok(byte_stream) => { let output = byte_stream_to_vec(byte_stream.into_inner()).await?; - ctx.job_output(&job_id, stream, &output, binary); + ctx.job_output(&job_id, baseboard, stream, &output, binary); } Err(error) if multiple => { let _ = ctx.job_error(error); @@ -2267,7 +2267,7 @@ async fn job_output_from( .into_inner(); let bytes = byte_stream_to_vec(byte_stream).await?; check_hash!(hash(&bytes), None)?; - ctx.job_output(&job_id, stream, &bytes, binary); + ctx.job_output(&job_id, target, stream, &bytes, binary); Ok(()) } } diff --git a/client/src/context.rs b/client/src/context.rs index 88943642..ee77a186 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -62,8 +62,11 @@ impl Authz { /// communication with a human user (if there is one). #[derive(Clone, Copy, Debug, Default, ValueEnum)] pub enum OutputFormat { + /// UTF-8 encoded text #[default] Text, + + /// One JSON object per line Json, } @@ -154,7 +157,14 @@ pub trait CommandContext: Clone + Send + Sync { fn job_stopped(&mut self, id: &JobId); fn job_skipped(&mut self, id: &JobId) -> bool; fn job_error(&mut self, error: CommandError) -> CommandError; - fn job_output(&mut self, id: &JobId, stream: JobOutputStream, output: &[u8], binary: bool); + fn job_output( + &mut self, + id: &JobId, + target: &BaseboardId, + stream: JobOutputStream, + output: &[u8], + binary: bool, + ); fn job_output_target(&mut self, target: &BaseboardId); fn job_output_started( &mut self, diff --git a/client/src/repl.rs b/client/src/repl.rs index 4aaf5800..3075083d 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -320,9 +320,16 @@ impl CommandContext for Repl { self.cli.job_output_target(target) } - fn job_output(&mut self, job_id: &JobId, stream: JobOutputStream, output: &[u8], binary: bool) { + fn job_output( + &mut self, + job_id: &JobId, + target: &BaseboardId, + stream: JobOutputStream, + output: &[u8], + binary: bool, + ) { self.set_job_id(Some(job_id.to_owned())); - self.cli.job_output(job_id, stream, output, binary); + self.cli.job_output(job_id, target, stream, output, binary); } fn job_output_started( From 7f2f435b7151ba90f8685d0eeaf7e2caf11f343f Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 19:30:27 -0600 Subject: [PATCH 09/20] Report command errors as JSON in JSON mode Co-authored-by: Claude --- client/src/cli.rs | 26 ++++-- client/src/commands.rs | 176 +++++++++++++++++++++++------------------ client/src/context.rs | 4 +- client/src/main.rs | 3 +- client/src/repl.rs | 17 ++-- 5 files changed, 136 insertions(+), 90 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index ab166bb8..7f66a1dd 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -240,6 +240,27 @@ impl CommandContext for Cli { *self.globals.lock().unwrap() = args; } + // Errors + + fn command_error(&mut self, error: &CommandError) { + let errors = error.errors(); + match self.get_output_format() { + OutputFormat::Json => { + let message = errors + .into_iter() + .map(|(_, message)| message) + .collect::>() + .join("\n"); + qprintln!("{}", json!({"error": {"message": message}})) + } + OutputFormat::Text => { + for (icon, message) in errors { + eprintln!("{icon} {message}"); + } + } + } + } + // Build provenance fn versions( @@ -590,11 +611,6 @@ impl CommandContext for Cli { skipped } - fn job_error(&mut self, error: CommandError) -> CommandError { - eprintln!("{error}"); - error - } - fn job_output_target(&mut self, target: &BaseboardId) { match self.get_output_format() { OutputFormat::Json => (), diff --git a/client/src/commands.rs b/client/src/commands.rs index 061d852f..f8e9420c 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1454,7 +1454,7 @@ async fn job( match job_attach(ctx, client, &job_id, &target).await { Err(CommandError::NotFound(_)) => { let error = attach_refused(ctx, client, &job_id, &target).await; - let _ = ctx.job_error(error); + ctx.command_error(&error); Ok(()) } result => result, @@ -1561,7 +1561,7 @@ async fn job_start( if let Err(error) = job_output_from(ctx, client, &target, Stdout, output).await && !error.is_unfinished_job() { - let _ = ctx.job_error(error); + ctx.command_error(&error); } } Err(error) => return Err(error), @@ -1740,9 +1740,7 @@ async fn job_start( let output = byte_stream_to_vec(byte_stream.into_inner()).await?; ctx.job_output(&job_id, baseboard, stream, &output, binary); } - Err(error) if multiple => { - let _ = ctx.job_error(error); - } + Err(error) if multiple => ctx.command_error(&error), Err(error) => return Err(error), } } @@ -2127,7 +2125,7 @@ async fn job_output( for baseboard in status.keys() { ctx.job_output_target(baseboard); if let Err(error) = job_output_from(ctx, client, baseboard, stream, args.clone()).await { - let _ = ctx.job_error(error); + ctx.command_error(&error); } } Ok(()) @@ -2592,161 +2590,161 @@ async fn resolve_serial( /// What went wrong parsing, preparing, or executing a client command. #[derive(Debug, Error)] pub enum CommandError { - #[error("❌ Serial `{0}` matches more than one sled, use a full baseboard ID")] + #[error("Serial `{0}` matches more than one sled, use a full baseboard ID")] AmbiguousSerial(String), - #[error("❌ Authentication error")] + #[error("Authentication error")] Authn(#[from] AuthnError), #[error( - "❌ Authentication failed, client and server are different builds\n \ - Client: {client}\n \ - Server: {server}" + "Authentication failed, client and server are different builds\n \ + Client: {client}\n \ + Server: {server}" )] AuthnBuildMismatch { client: VersionInfo, server: VersionInfo, }, - #[error("❌ Canceled")] + #[error("Canceled")] Canceled, - #[error("❌ Certificate for `{0}` is outside its validity window")] + #[error("Certificate for `{0}` is outside its validity window")] CertExpired(String), - #[error("❌ Chunk size must be positive")] + #[error("Chunk size must be positive")] ChunkSizeZero, - #[error("❓ {0}")] + #[error("{0}")] Clap(#[from] clap::Error), - #[error("❌ {0}")] + #[error("{0}")] Client(String), - #[error("❌ {0}")] + #[error("{0}")] Der(#[from] x509_cert::der::Error), - #[error("❌ {0}")] + #[error("{0}")] DurationOutOfRange(#[from] chrono::OutOfRangeError), - #[error("❌ Empty certificate chain")] + #[error("Empty certificate chain")] EmptyCertChain, - #[error("❌ Identity error: {0}")] + #[error("Identity error: {0}")] Identity(#[from] IdentityError), #[error( - "❌ Identity mismatch, tried to start an interactive job\n \ - for `{interactive}`\n \ - as `{key_id}`" + "Identity mismatch, tried to start an interactive job\n \ + for `{interactive}`\n \ + as `{key_id}`" )] IdentityMismatch { interactive: KeyId, key_id: KeyId }, - #[error("❌ Interactive job error: {0}")] + #[error("Interactive job error: {0}")] Interactive(#[from] InteractiveJobError), - #[error("❌ Interactive jobs must target exactly one sled")] + #[error("Interactive jobs must target exactly one sled")] InteractiveTarget, - #[error("❌ Local I/O error accessing `{path}`: {error}")] + #[error("Local I/O error accessing `{path}`: {error}")] Io { path: PathBuf, error: std::io::Error, }, - #[error("❌ Signal handling error: {0}")] + #[error("Signal handling error: {0}")] Signal(#[from] std::io::Error), - #[error("❌ Unauthorized, try `iam`")] + #[error("Unauthorized, try `iam`")] InvalidAuthorization, - #[error("❌ Leaf certificate does not match key `{0}`")] + #[error("Leaf certificate does not match key `{0}`")] InvalidLeafCert(KeyId), - #[error("❌ Root certificate is not self-signed")] + #[error("Root certificate is not self-signed")] InvalidRootCert, - #[error("❌ Job `{0}` was cancelled before it started")] + #[error("Job `{0}` was cancelled before it started")] JobCancelled(JobId), - #[error("❌ Job `{0}` did not run")] + #[error("Job `{0}` did not run")] JobDidNotRun(JobId), - #[error("❌ Job `{0}` has finished, try `job status` or `job output`")] + #[error("Job `{0}` has finished, try `job status` or `job output`")] JobFinished(JobId), - #[error("❌ Job `{0}` is not interactive, try `job status` or `job output`")] + #[error("Job `{0}` is not interactive, try `job status` or `job output`")] JobNotInteractive(JobId), - #[error("❌ Job `{0}` not found on sled `{1}`")] + #[error("Job `{0}` not found on sled `{1}`")] JobNotOnSled(JobId, BaseboardId), - #[error("{1}\n❗ Job `{0}` was signed but not started, try `job skip`")] + #[error("Job `{0}` was signed but not started, try `job skip`")] JobNotStarted(JobId, Box), - #[error("❌ Job `{0}` is not yet running")] + #[error("Job `{0}` is not yet running")] JobNotYetRunning(JobId), - #[error("⏩ Job `{0}` was skipped on this sled: {1}")] + #[error("Job `{0}` was skipped on this sled: {1}")] JobSkipped(JobId, SkipReason), - #[error("❌ Job `{0}` is still running")] + #[error("Job `{0}` is still running")] JobStillRunning(JobId), - #[error("❌ JSON error: {0}")] + #[error("JSON error: {0}")] Json(#[from] serde_json::Error), - #[error("❌ Key error: {0}")] + #[error("Key error: {0}")] Key(#[from] KeyError), - #[error("❌ Length mismatch: expected {expected} bytes, received {received}")] + #[error("Length mismatch: expected {expected} bytes, received {received}")] LengthMismatch { expected: u64, received: u64 }, - #[error("❌ Missing client nonce: run `session start-params` first, or pass one")] + #[error("Missing client nonce: run `session start-params` first, or pass one")] MissingClientNonce, - #[error("❌ Missing job command, try `--help`")] + #[error("Missing job command, try `--help`")] MissingCommand, #[cfg(feature = "permslip")] - #[error("❌ Missing signing key name, try `--permslip`")] + #[error("Missing signing key name, try `--permslip`")] MissingKeyName, #[cfg(feature = "permslip")] - #[error("❌ Missing permslip URL, try `--permslip-url` or setting `PERMSLIP_URL`")] + #[error("Missing permslip URL, try `--permslip-url` or setting `PERMSLIP_URL`")] MissingPermslipUrl, - #[error("❌ Missing rack ID, try `--rack` (IDs from `oxide system hardware rack list`)")] + #[error("Missing rack ID, try `--rack` (IDs from `oxide system hardware rack list`)")] MissingRack, - #[error("❌ Missing Nexus token, try `--nexus-token` or setting `SUSH_NEXUS_TOKEN`")] + #[error("Missing Nexus token, try `--nexus-token` or setting `SUSH_NEXUS_TOKEN`")] MissingNexusToken, - #[error("❌ Missing session, try `session start`")] + #[error("Missing session, try `session start`")] MissingSession, - #[error("❌ Missing SSH agent socket, try `--ssh-auth-sock`")] + #[error("Missing SSH agent socket, try `--ssh-auth-sock`")] MissingSshAuthSock, - #[error("❌ {0}")] + #[error("{0}")] NotFound(String), - #[error("❌ Job `{0}` is not the session's next job")] + #[error("Job `{0}` is not the session's next job")] NotNextJob(JobId), - #[error("❌ Command not supported in offline mode, try `--url`")] + #[error("Command not supported in offline mode, try `--url`")] Offline, #[error( - "❌ Hash mismatch, output may be truncated or corrupted\n \ - Expected: {expected}\n \ - Received: {received}" + "Hash mismatch, output may be truncated or corrupted\n \ + Expected: {expected}\n \ + Received: {received}" )] OutputHashMismatch { expected: JobOutputHash, received: JobOutputHash, }, - #[error("❌ `--binary` and `--file` need a specific `--target`")] + #[error("`--binary` and `--file` need a specific `--target`")] OutputNeedsTarget, #[cfg(feature = "permslip")] - #[error("❌ permslip error: {0}")] + #[error("permslip error: {0}")] Permslip(#[from] PermslipError), - #[error("❌ Job process error: {0}")] + #[error("Job process error: {0}")] Process(#[from] sush_common::jobs::ProcessError), - #[error("❌ Proxy TLS error: {0}")] + #[error("Proxy TLS error: {0}")] ProxyTls(#[from] tls::ProxyTlsError), - #[error("👋 Goodbye!")] + #[error("Goodbye!")] Quit, - #[error("❌ {0}")] + #[error("{0}")] Readline(#[from] rustyline::error::ReadlineError), #[error(transparent)] Recursive(#[from] Box), - #[error("❌ Reqwest error: {0}")] + #[error("Reqwest error: {0}")] Reqwest(#[from] reqwest::Error), - #[error("❌ SSH signature error on key ID `{0}`")] + #[error("SSH signature error on key ID `{0}`")] Signature(KeyId), - #[error("❌ Streaming jobs need `--file`")] + #[error("Streaming jobs need `--file`")] StreamingNeedsFile, - #[error("❌ Job `{0}` did not exit cleanly, streamed output is unverified")] + #[error("Job `{0}` did not exit cleanly, streamed output is unverified")] StreamUnverified(JobId), - #[error("❌ Can't parse target baseboard ID: {0}")] + #[error("Can't parse target baseboard ID: {0}")] BaseboardIdParseError(sled_hardware_types::BaseboardIdParseError), - #[error("❌ SSH key error: {0}")] + #[error("SSH key error: {0}")] SshKey(#[from] ssh_key::Error), - #[error("❌ Timed out waiting for job")] + #[error("Timed out waiting for job")] TimedOut, - #[error("❌ Too much output to display on terminal, try `--file`")] + #[error("Too much output to display on terminal, try `--file`")] TooMuchOutput, - #[error("❌ Tunnel error: {0}")] + #[error("Tunnel error: {0}")] Tunnel(#[from] TunnelError), - #[error("❌ No sled with serial `{serial}` has a status for job `{job_id}`")] + #[error("No sled with serial `{serial}` has a status for job `{job_id}`")] UnknownSerial { serial: String, job_id: JobId }, - #[error("❌ Serial `{0}` matches no sled in the rack inventory")] + #[error("Serial `{0}` matches no sled in the rack inventory")] UnknownRackSerial(String), - #[error("❌ Chain root does not match any supplied root certificate")] + #[error("Chain root does not match any supplied root certificate")] UntrustedRoot, - #[error("❌ Can't start interactive session: {0}")] + #[error("Can't start interactive session: {0}")] Upgrade(String), - #[error("❌ UTF-8 error: {0}")] + #[error("UTF-8 error: {0}")] Utf8(#[from] std::string::FromUtf8Error), - #[error("❌ WebSocket error: {0}")] + #[error("WebSocket error: {0}")] WebSocket(#[from] WebSocketError), } @@ -2763,6 +2761,30 @@ impl CommandError { pub fn is_unfinished_job(&self) -> bool { matches!(self, Self::JobNotYetRunning(_) | Self::JobStillRunning(_)) } + + fn icon(&self) -> &'static str { + match self { + Self::Clap(_) => "❓", + Self::JobNotStarted(..) => "❗", + Self::JobSkipped(..) => "⏩", + Self::Quit => "👋", + Self::Recursive(error) => error.icon(), + _ => "❌", + } + } + + /// The icon and message of each error in this one, inner first. + pub fn errors(&self) -> Vec<(&'static str, String)> { + match self { + Self::JobNotStarted(_, error) => { + let mut errors = error.errors(); + errors.push((self.icon(), self.to_string())); + errors + } + Self::Recursive(error) => error.errors(), + _ => vec![(self.icon(), self.to_string())], + } + } } impl From> for CommandError { diff --git a/client/src/context.rs b/client/src/context.rs index ee77a186..aa2145e3 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -103,6 +103,9 @@ pub trait CommandContext: Clone + Send + Sync { fn set_globals(&mut self, _args: GlobalArgs) {} fn pre_parse_hook(&mut self, _command: &str) {} + // Errors + fn command_error(&mut self, error: &CommandError); + // Build provenance fn versions( &mut self, @@ -156,7 +159,6 @@ pub trait CommandContext: Clone + Send + Sync { fn job_started(&mut self, job: &SignedJob, sleds: Option<&BTreeSet>, show: bool); fn job_stopped(&mut self, id: &JobId); fn job_skipped(&mut self, id: &JobId) -> bool; - fn job_error(&mut self, error: CommandError) -> CommandError; fn job_output( &mut self, id: &JobId, diff --git a/client/src/main.rs b/client/src/main.rs index 16a13b4d..f2581007 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -12,6 +12,7 @@ use rustix::termios::{OptionalActions, OutputModes, Termios, isatty, tcgetattr, use sush_client::cli::Cli; use sush_client::commands::ClientArgs; +use sush_client::context::CommandContext as _; /// The kernel tab-expansion bits, which rustix does not expose on /// illumos: TABDLY from illumos sys/termios.h. @@ -44,7 +45,7 @@ async fn main() -> ExitCode { let code = match ClientArgs::parse().execute(&mut cli).await { Ok(()) => ExitCode::SUCCESS, Err(error) => { - eprintln!("{error}"); + cli.command_error(&error); ExitCode::FAILURE } }; diff --git a/client/src/repl.rs b/client/src/repl.rs index 3075083d..955bdfa6 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -101,12 +101,15 @@ impl Repl { }; let command = match ReplCommandParser::try_parse(&words, &mut self) { Ok(ReplCommandParser { command }) => command, - Err(err) => perr!("{err}"), + Err(err) => { + self.command_error(&err); + continue; + } }; match command.execute(&mut self).await { Ok(()) => (), Err(CommandError::Quit) => break, - Err(err) => perr!("{err}"), + Err(err) => self.command_error(&err), } } Err(ReadlineError::Interrupted) => continue, @@ -213,6 +216,12 @@ impl CommandContext for Repl { self.cli.set_globals(args); } + // Errors + + fn command_error(&mut self, error: &CommandError) { + self.cli.command_error(error) + } + // Build provenance fn versions( @@ -312,10 +321,6 @@ impl CommandContext for Repl { self.cli.job_skipped(job_id) } - fn job_error(&mut self, error: CommandError) -> CommandError { - self.cli.job_error(error) - } - fn job_output_target(&mut self, target: &BaseboardId) { self.cli.job_output_target(target) } From 5ccfcbae561ba07ff38d560a31df99c8af04e362 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 19:57:09 -0600 Subject: [PATCH 10/20] Log in once for concurrent refused requests Co-authored-by: Claude --- client/src/commands.rs | 31 ++++++++++++++++++++----------- client/src/lib.rs | 33 ++++++++++++++++++++++++++++++--- 2 files changed, 50 insertions(+), 14 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index f8e9420c..b9ebbc3e 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -958,9 +958,9 @@ where with_login_via(ctx, client, None, make_request).await } -/// Like [`with_login`], for requests a proxy routes to a particular -/// sled. Identities live on one sled, so the login must go where the -/// request it retries went. +/// Like [`with_login`], for a request that the proxy routes to sled +/// `via`. A sled that refuses a request answers with a login challenge +/// that only it can verify, so the login goes to `via` as well. async fn with_login_via( ctx: &mut impl CommandContext, client: &Client, @@ -971,14 +971,23 @@ where Req: AsyncFnMut() -> Result>, CommandError: From>, { - match make_request().await { - Err(ClientError::ErrorResponse(err)) if err.status() == StatusCode::UNAUTHORIZED => { - let (_identity, authz) = authenticate(ctx, client, via, err).await?; - ctx.set_credentials(Some(authz)); - Ok(make_request().await?) - } - Err(err) => Err(err.into()), - Ok(res) => Ok(res), + let signer = ctx.authz_signer(); + let mut generation = signer.generation(); + let mut logged_in = false; + loop { + match make_request().await { + Err(ClientError::ErrorResponse(err)) + if err.status() == StatusCode::UNAUTHORIZED && !logged_in => + { + if let Some(_login) = signer.login_needed(generation).await { + let (_identity, authz) = authenticate(ctx, client, via, err).await?; + ctx.set_credentials(Some(authz)); + logged_in = true; + } + generation = signer.generation(); + } + result => return Ok(result?), + } } } diff --git a/client/src/lib.rs b/client/src/lib.rs index 2b907f51..b3b62085 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -9,6 +9,7 @@ use std::sync::{Arc, Mutex}; use reqwest::header::AUTHORIZATION; use sush_common::authn::AuthnError; +use tokio::sync::{Mutex as AsyncMutex, MutexGuard as AsyncMutexGuard}; use crate::context::Authz; @@ -66,15 +67,41 @@ pub mod tunnel; /// client's pre-send hook, which signs every request with the current /// ephemeral key once one exists. #[derive(Clone, Debug, Default)] -pub struct AuthzSigner(Arc>>); +pub struct AuthzSigner(Arc); + +#[derive(Debug, Default)] +struct AuthzState { + /// The current authorization, and how many times it has been set. + authz: Mutex<(Option, u64)>, + + /// Held during a login. When the server refuses several concurrent + /// requests, the first to take this lock logs in, and the others + /// retry with its credentials. + login: AsyncMutex<()>, +} impl AuthzSigner { pub fn get(&self) -> Option { - self.0.lock().unwrap().clone() + self.0.authz.lock().unwrap().0.clone() } pub fn set(&self, authz: Option) { - *self.0.lock().unwrap() = authz; + let mut guard = self.0.authz.lock().unwrap(); + guard.0 = authz; + guard.1 += 1; + } + + /// How many times the authorization has been set. + pub fn generation(&self) -> u64 { + self.0.authz.lock().unwrap().1 + } + + /// Wait for any login in progress. Return the login guard if the + /// authorization has not been set since `generation`, or `None` if + /// another login set it. + pub async fn login_needed(&self, generation: u64) -> Option> { + let login = self.0.login.lock().await; + (self.generation() == generation).then_some(login) } } From 0fc1685dd681d0196c213b99829edc3528f9def5 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Mon, 26 Oct 2026 20:06:45 -0600 Subject: [PATCH 11/20] End watch-time messages with a newline Co-authored-by: Claude --- client/src/cli.rs | 29 ++++++++++++----------------- 1 file changed, 12 insertions(+), 17 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index 7f66a1dd..cc7bc3ef 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -111,6 +111,13 @@ impl Cli { } } + fn print(&self, message: impl fmt::Display) { + match self.watch.lock().unwrap().as_ref() { + Some(watch) => watch.multi.suspend(|| eprintln!("{message}")), + None => qprintln!("{message}"), + } + } + /// Adopt `session` unless one with the same ID is already /// attached and `force` is unset. fn adopt_session(&mut self, session: Session, force: bool) -> SessionId { @@ -559,20 +566,13 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - let watch = self.watch.lock().unwrap(); - let print = |line: String| match watch.as_ref() { - Some(watch) => { - let _ = watch.multi.println(line); - } - None => qprintln!("{line}"), - }; - print(format!("✅ Started job `{job_id}`")); + self.print(format!("✅ Started job `{job_id}`")); if let (Target::Sleds(named), Some(sleds)) = (job.payload().target(), sleds) { for sled in named { if let SledId::Baseboard(baseboard) = sled && !sleds.contains(baseboard) { - print(format!( + self.print(format!( "❗ Job `{job_id}` will not run on unreachable sled `{baseboard}`" )); } @@ -586,11 +586,7 @@ impl CommandContext for Cli { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"job_stopped": job_id})), OutputFormat::Text => { - if let Some(watch) = self.watch.lock().unwrap().as_ref() { - let _ = watch.multi.println(format!("✅ Stopped job `{job_id}`")); - } else { - qprintln!("\r✅ Stopped job `{job_id}`"); - } + self.print(format!("\r✅ Stopped job `{job_id}`")); } } } @@ -787,9 +783,8 @@ 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!( + if matches!(self.get_output_format(), OutputFormat::Text) { + self.print(format!( "❗ No sled has reported a status for job `{job_id}`" )); } From 0564e1f008669ff9060af10b09821f10dc8e1fc9 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 08:53:44 -0600 Subject: [PATCH 12/20] Write advisory messages to stderr Co-authored-by: Claude --- client/src/cli.rs | 79 +++++++++++++++++++++++++----------------- client/src/commands.rs | 16 +++++---- client/src/context.rs | 20 ++++++++++- client/src/repl.rs | 15 +++++--- 4 files changed, 85 insertions(+), 45 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index cc7bc3ef..23c77b2c 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -40,7 +40,7 @@ use sush_common::targets::{MAX_CUBBY, SledHealth, SledId, SledVersion, Target}; use sush_common::version::VersionInfo; use crate::commands::{CommandError, GlobalArgs}; -use crate::context::{CommandContext, OutputFormat, StatusDisplayStyle}; +use crate::context::{CommandContext, OutputFormat, StartedDisplay, StatusDisplayStyle}; use crate::{AuthzSigner, quiet_write}; pub(crate) const PREFIX: &str = "sush"; @@ -111,10 +111,10 @@ impl Cli { } } - fn print(&self, message: impl fmt::Display) { + fn note(&self, message: impl fmt::Display) { match self.watch.lock().unwrap().as_ref() { Some(watch) => watch.multi.suspend(|| eprintln!("{message}")), - None => qprintln!("{message}"), + None => eprintln!("{message}"), } } @@ -375,7 +375,7 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - qprintln!("✅ Please relay these parameters to support:"); + self.note("✅ Please relay these parameters to support:"); qprintln!(" Baseboard ID: {baseboard_id}"); qprintln!(" Sush nonce: {sush_nonce}"); } @@ -397,7 +397,7 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - qprintln!("✅ Please request a job with these parameters from support:"); + self.note("✅ Please request a job with these parameters from support:"); qprintln!(" Job ID: {}", request.job_id); qprintln!(" Session ID: {}", request.session_id); qprintln!(" Command: {}", request.command); @@ -426,7 +426,8 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - qprintln!("✅ Session is now `{session_id}`"); + self.note("✅ Created session"); + qprintln!(" Session ID: {session_id}"); qprintln!(" Signer nonce: {signer_nonce}"); qprintln!(" Sequence number: {seq}"); } @@ -437,7 +438,10 @@ impl CommandContext for Cli { let session_id = self.adopt_session(session, force); match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"session_started": session_id})), - OutputFormat::Text => qprintln!("✅ Session is now `{session_id}`"), + OutputFormat::Text => { + self.note("✅ Session is now current"); + qprintln!(" Session ID: {session_id}"); + } } } @@ -451,7 +455,7 @@ impl CommandContext for Cli { } match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"session_stopped": session_id})), - OutputFormat::Text => qprintln!("✅ Stopped session `{session_id}`"), + OutputFormat::Text => self.note(format!("✅ Stopped session `{session_id}`")), } } @@ -468,16 +472,17 @@ impl CommandContext for Cli { }) ) } - OutputFormat::Text => { - qprintln!("✅ Allowed {} attach for `{key_id}`", access.as_str()) - } + OutputFormat::Text => self.note(format!( + "✅ Allowed {} attach for `{key_id}`", + access.as_str() + )), } } fn attach_denied(&mut self, key_id: &KeyId) { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"attach_denied": key_id})), - OutputFormat::Text => qprintln!("✅ Denied attach for `{key_id}`"), + OutputFormat::Text => self.note(format!("✅ Denied attach for `{key_id}`")), } } @@ -529,9 +534,9 @@ impl CommandContext for Cli { OutputFormat::Json => qprintln!("{}", json!({"certs": certs})), OutputFormat::Text => { if roots.is_empty() { - qprintln!("✅ Verified chain consistency"); + self.note("✅ Verified chain consistency"); } else { - qprintln!("✅ Verified chain against supplied root"); + self.note("✅ Verified chain against supplied root"); } for cert in &chain { qprintln!(" {}", cert.tbs_certificate.subject); @@ -544,12 +549,17 @@ impl CommandContext for Cli { // Job management - fn job_started(&mut self, job: &SignedJob, sleds: Option<&BTreeSet>, show: bool) { + fn job_started( + &mut self, + job: &SignedJob, + sleds: Option<&BTreeSet>, + display: StartedDisplay, + ) { if let Some(session) = self.session.lock().unwrap().as_mut() { session.job_started(job.to_owned()); self.save_session(Some(session)); } - if !show { + if matches!(display, StartedDisplay::None) { return; } let job_id = job.job_id(); @@ -566,13 +576,16 @@ impl CommandContext for Cli { }) ), OutputFormat::Text => { - self.print(format!("✅ Started job `{job_id}`")); + self.note(format!("✅ Started job `{job_id}`")); + if matches!(display, StartedDisplay::Full) { + qprintln!(" Job ID: {job_id}"); + } if let (Target::Sleds(named), Some(sleds)) = (job.payload().target(), sleds) { for sled in named { if let SledId::Baseboard(baseboard) = sled && !sleds.contains(baseboard) { - self.print(format!( + self.note(format!( "❗ Job `{job_id}` will not run on unreachable sled `{baseboard}`" )); } @@ -586,7 +599,7 @@ impl CommandContext for Cli { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"job_stopped": job_id})), OutputFormat::Text => { - self.print(format!("\r✅ Stopped job `{job_id}`")); + self.note(format!("\r✅ Stopped job `{job_id}`")); } } } @@ -601,7 +614,7 @@ impl CommandContext for Cli { self.save_session(session_guard.as_ref()); match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"job_skipped": job_id})), - OutputFormat::Text => qprintln!("✅ Skipped job `{job_id}`"), + OutputFormat::Text => self.note(format!("✅ Skipped job `{job_id}`")), } } skipped @@ -610,7 +623,7 @@ impl CommandContext for Cli { fn job_output_target(&mut self, target: &BaseboardId) { match self.get_output_format() { OutputFormat::Json => (), - OutputFormat::Text => qprintln!(" » {target} «"), + OutputFormat::Text => self.note(format!(" » {target} «")), } } @@ -650,8 +663,8 @@ impl CommandContext for Cli { OutputFormat::Text if output.is_empty() => (), OutputFormat::Text => { match stream { - JobOutputStream::Stdout => qprintln!("✅ Job stdout:"), - JobOutputStream::Stderr => qprintln!("✅ Job stderr:"), + JobOutputStream::Stdout => self.note("✅ Job stdout:"), + JobOutputStream::Stderr => self.note("✅ Job stderr:"), }; let output = String::from_utf8_lossy(output); if output.ends_with('\n') { @@ -723,12 +736,12 @@ impl CommandContext for Cli { if let Some(stage) = stage { let length = ByteSize::b(progress.length().unwrap_or(progress.position())); let elapsed = progress.elapsed(); - qprintln!( + self.note(format!( "{stage} {stream}:\t{} in {} ({:.0} MB/s)", length.display().si(), format_duration(progress.elapsed()), length.as_mb() / elapsed.as_secs_f64(), - ); + )); } } } @@ -784,7 +797,7 @@ impl CommandContext for Cli { fn job_watch_stalled(&mut self, job_id: &JobId) { if matches!(self.get_output_format(), OutputFormat::Text) { - self.print(format!( + self.note(format!( "❗ No sled has reported a status for job `{job_id}`" )); } @@ -802,16 +815,18 @@ impl CommandContext for Cli { fn job_attached(&mut self, job_id: &JobId) { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"attached": job_id})), - OutputFormat::Text => { - qprintln!("✅ Attached to interactive job `{job_id}`, detach with `^]`") - } + OutputFormat::Text => self.note(format!( + "✅ Attached to interactive job `{job_id}`, detach with `^]`" + )), } } fn job_detached(&mut self, job_id: &JobId) { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"detached": job_id})), - OutputFormat::Text => qprintln!("\r✅ Detached from interactive job `{job_id}`"), + OutputFormat::Text => { + self.note(format!("\r✅ Detached from interactive job `{job_id}`")) + } } } @@ -845,7 +860,7 @@ impl CommandContext for Cli { progress.finish_and_clear(); } if matches!(self.get_output_format(), OutputFormat::Text) { - qprintln!("✅ Signed request for job `{job_id}`"); + self.note(format!("✅ Signed request for job `{job_id}`")); } } @@ -1080,7 +1095,7 @@ impl CommandContext for Cli { .collect::, CommandError>>()?; qprintln!("{}", json!({"identities": identities})); } - OutputFormat::Text if keys.is_empty() => qprintln!("❌ No SSH identities found"), + OutputFormat::Text if keys.is_empty() => self.note("❌ No SSH identities found"), OutputFormat::Text => { for key in keys { let key_id = key.key_id()?; diff --git a/client/src/commands.rs b/client/src/commands.rs index b9ebbc3e..637ac983 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -57,7 +57,7 @@ use sush_common::targets::{SledId, Target}; use sush_common::version::VersionInfo; use crate::ByteStream; -use crate::context::{Authz, CommandContext, OutputFormat, StatusDisplayStyle}; +use crate::context::{Authz, CommandContext, OutputFormat, StartedDisplay, StatusDisplayStyle}; use crate::identity::{IdentityError, SshAgentConnection}; use crate::interactive::interactive_job; #[cfg(feature = "permslip")] @@ -1385,7 +1385,7 @@ async fn job( } } else { ctx.job_signed(&job, true); - ctx.job_started(&job, None, false); + ctx.job_started(&job, None, StartedDisplay::None); Ok(()) } } @@ -1555,7 +1555,7 @@ async fn job_start( // Start the job. if interactive { let sleds = start.await?.into_inner().sleds; - ctx.job_started(&job, Some(&sleds), true); + ctx.job_started(&job, Some(&sleds), StartedDisplay::Note); match job_attach(ctx, client, &job_id, &target).await { Ok(()) => (), Err(CommandError::NotFound(_)) => { @@ -1591,7 +1591,7 @@ async fn job_start( .open(&path) .map_err(|error| CommandError::io(&path, error))?; let sleds = start.await?.into_inner().sleds; - ctx.job_started(&job, Some(&sleds), true); + ctx.job_started(&job, Some(&sleds), StartedDisplay::Note); let hasher = job_stream(ctx, client, &job_id, &target, &path, file).await?; let status = job_watch(ctx, client, &job_id, Some(sleds)).await?; ctx.job_status(&job_id, &status, StatusDisplayStyle::Short); @@ -1656,10 +1656,12 @@ async fn job_start( match start_result { Ok(response) => { let started = response.into_inner().sleds; - ctx.job_started(&job, Some(&started), true); + ctx.job_started(&job, Some(&started), StartedDisplay::Note); sleds = Some(started); } - Err(CommandError::TimedOut) => ctx.job_started(&job, None, true), + Err(CommandError::TimedOut) => { + ctx.job_started(&job, None, StartedDisplay::Note) + } Err(error) => { ctx.job_watch_finished(&job_id); return Err(error); @@ -1756,7 +1758,7 @@ async fn job_start( } } else { let sleds = start.await?.into_inner().sleds; - ctx.job_started(&job, Some(&sleds), true); + ctx.job_started(&job, Some(&sleds), StartedDisplay::Full); } Ok(()) } diff --git a/client/src/context.rs b/client/src/context.rs index aa2145e3..7d98e7c8 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -78,6 +78,19 @@ pub enum StatusDisplayStyle { Full, } +/// What to show when a job starts. +#[derive(Clone, Copy, Debug)] +pub enum StartedDisplay { + /// Display nothing; the job was signed offline. + None, + + /// Display a note to stderr only. + Note, + + /// Display a note to stderr, and the job ID to stdout. + Full, +} + impl OutputFormat { pub fn as_str(&self) -> &'static str { match self { @@ -156,7 +169,12 @@ pub trait CommandContext: Clone + Send + Sync { // Job management fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError>; - fn job_started(&mut self, job: &SignedJob, sleds: Option<&BTreeSet>, show: bool); + fn job_started( + &mut self, + job: &SignedJob, + sleds: Option<&BTreeSet>, + display: StartedDisplay, + ); fn job_stopped(&mut self, id: &JobId); fn job_skipped(&mut self, id: &JobId) -> bool; fn job_output( diff --git a/client/src/repl.rs b/client/src/repl.rs index 955bdfa6..0e9fe822 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -34,7 +34,7 @@ use crate::commands::{ ClientCommand, CommandError, GlobalArgs, SSH_AUTH_SOCK, SUSH_JOB_ID, SUSH_KEY_ID, SUSH_OUTPUT_FORMAT, SUSH_URL, }; -use crate::context::{CommandContext, OutputFormat, StatusDisplayStyle}; +use crate::context::{CommandContext, OutputFormat, StartedDisplay, StatusDisplayStyle}; use crate::{AuthzSigner, Client}; const HISTORY_FILE: &str = "history.txt"; @@ -200,11 +200,11 @@ impl CommandContext for Repl { if let Some(arg) = $arg { unsafe { env::set_var($var, arg.to_string()) } args.$arg = Some(arg.clone()); - qprintln!("✅ {} set to `{}`", $name, arg); + eprintln!("✅ {} set to `{}`", $name, arg); } else { unsafe { env::remove_var($var) } args.$arg = None; - qprintln!("✅ {} unset", $name); + eprintln!("✅ {} unset", $name); } }}; } @@ -307,9 +307,14 @@ impl CommandContext for Repl { // Job management - fn job_started(&mut self, job: &SignedJob, sleds: Option<&BTreeSet>, show: bool) { + fn job_started( + &mut self, + job: &SignedJob, + sleds: Option<&BTreeSet>, + display: StartedDisplay, + ) { self.set_job_id(Some(job.job_id().to_owned())); - self.cli.job_started(job, sleds, show); + self.cli.job_started(job, sleds, display); } fn job_stopped(&mut self, job_id: &JobId) { From dde283c6f4de42e4ac25b68efdf7ac654ff66cf3 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 09:26:40 -0600 Subject: [PATCH 13/20] Make session and signing messages accurate Session errors carry an error code and say what to do, and a stale session is 410, not 404. A start refused for its session no longer suggests `job skip`, and a failed signing no longer claims success. Co-authored-by: Claude --- Cargo.lock | 26 ++++++++++++++++++++++++-- Cargo.toml | 1 + api/Cargo.toml | 1 + api/src/lib.rs | 9 +++++++++ client/src/cli.rs | 17 ++++++++++++++--- client/src/commands.rs | 25 +++++++++++++++++++------ client/src/context.rs | 3 ++- client/src/repl.rs | 8 ++++++-- server/src/error.rs | 18 ++++++++++++++---- 9 files changed, 90 insertions(+), 18 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0f8e3869..1568fa84 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3594,7 +3594,7 @@ dependencies = [ "hex", "serde", "serde_with", - "strum", + "strum 0.26.3", "thiserror 2.0.20", ] @@ -4713,7 +4713,16 @@ version = "0.26.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" dependencies = [ - "strum_macros", + "strum_macros 0.26.4", +] + +[[package]] +name = "strum" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" +dependencies = [ + "strum_macros 0.27.2", ] [[package]] @@ -4729,6 +4738,18 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "strum_macros" +version = "0.27.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "suanpan" version = "0.1.0" @@ -4800,6 +4821,7 @@ dependencies = [ "schemars 0.8.22", "serde", "sled-hardware-types", + "strum 0.27.2", "sush-common", "thiserror 1.0.69", ] diff --git a/Cargo.toml b/Cargo.toml index 30c5e683..95187ad7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -67,6 +67,7 @@ slog-term = "2" sprockets-tls = { git = "https://github.com/oxidecomputer/sprockets.git", rev = "98615b86d51f74ce220198eb16c57932fd24e7cb", default-features = false } sprockets-tls-test-utils = { git = "https://github.com/oxidecomputer/sprockets.git", rev = "98615b86d51f74ce220198eb16c57932fd24e7cb" } ssh-key = { version = "0.6", features = ["ed25519", "p256", "serde"] } +strum = { version = "0.27", features = ["derive"] } tempfile = "3" thiserror = "1" tokio = { version = "1", features = ["macros", "process", "rt", "signal", "sync", "time"] } diff --git a/api/Cargo.toml b/api/Cargo.toml index 00024b0d..b68e637e 100644 --- a/api/Cargo.toml +++ b/api/Cargo.toml @@ -15,5 +15,6 @@ hyper.workspace = true schemars.workspace = true serde.workspace = true sled-hardware-types.workspace = true +strum.workspace = true sush-common = { path = "../common" } thiserror.workspace = true diff --git a/api/src/lib.rs b/api/src/lib.rs index 388ff6ba..93fb7a70 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -18,6 +18,7 @@ use schemars::JsonSchema; use serde::de::{Deserializer, Error as DeserializeError, IntoDeserializer, MapAccess, Visitor}; use serde::{Deserialize, Serialize}; use sled_hardware_types::BaseboardId; +use strum::{Display, EnumString}; use sush_common::authn::Identity; use sush_common::jobs::{ @@ -318,6 +319,14 @@ pub struct JobStartParams { pub wait: JobWait, } +/// Codes for the errors a client may act on. `Display` gives the +/// `error_code` string of the response body. +#[derive(Clone, Copy, Debug, Display, EnumString, Eq, PartialEq)] +pub enum ErrorCode { + NoSession, + SessionNotCurrent, +} + /// Whether and until what state is reached to wait for a job start/stop request. #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "lowercase")] diff --git a/client/src/cli.rs b/client/src/cli.rs index 23c77b2c..954df353 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -439,7 +439,18 @@ impl CommandContext for Cli { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"session_started": session_id})), OutputFormat::Text => { - self.note("✅ Session is now current"); + self.note("✅ Started session"); + qprintln!(" Session ID: {session_id}"); + } + } + } + + fn session_joined(&mut self, session: Session, force: bool) { + let session_id = self.adopt_session(session, force); + match self.get_output_format() { + OutputFormat::Json => qprintln!("{}", json!({"session_joined": session_id})), + OutputFormat::Text => { + self.note("✅ Joined session"); qprintln!(" Session ID: {session_id}"); } } @@ -855,11 +866,11 @@ impl CommandContext for Cli { } } - fn job_signing_finished(&mut self, job_id: &JobId) { + fn job_signing_finished(&mut self, job_id: &JobId, signed: bool) { if let Some(progress) = self.progress.lock().unwrap().take() { progress.finish_and_clear(); } - if matches!(self.get_output_format(), OutputFormat::Text) { + if signed && matches!(self.get_output_format(), OutputFormat::Text) { self.note(format!("✅ Signed request for job `{job_id}`")); } } diff --git a/client/src/commands.rs b/client/src/commands.rs index 637ac983..71762576 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -41,7 +41,7 @@ use tokio_tungstenite::tungstenite::protocol::Role; use x509_cert::Certificate; use x509_cert::der::DecodePem as _; -use sush_api::JobWait; +use sush_api::{ErrorCode, JobWait}; use sush_common::authn::{ AuthnError, Challenge, ChallengeResponse, Credentials, Identity, RequestKey, }; @@ -1081,7 +1081,7 @@ async fn session( return Err(CommandError::MissingSession); } // The server's session state is authoritative. - ctx.session_started(session, true); + ctx.session_joined(session, true); Ok(()) } @@ -1092,7 +1092,7 @@ async fn session( }, None, ) => { - ctx.session_started(Session::new(session_id), force); + ctx.session_joined(Session::new(session_id), force); Ok(()) } @@ -1364,12 +1364,12 @@ async fn job( let job = loop { select! { job = &mut sign => { - ctx.job_signing_finished(&job_id); + ctx.job_signing_finished(&job_id, job.is_ok()); break job?; } _ = interval.tick() => ctx.job_signing_update(&job_id), _ = sigint.recv() => { - ctx.job_signing_finished(&job_id); + ctx.job_signing_finished(&job_id, false); return Err(CommandError::Canceled); } } @@ -1378,7 +1378,10 @@ async fn job( ctx.job_signed(&job, false); let job_id = *job.job_id(); match job_start(ctx, client, job, start_args.to_owned()).await { - Err(error) if ctx.next_job_id().ok() == Some(job_id) => { + Err(error) + if !matches!(error, CommandError::Session(_)) + && ctx.next_job_id().ok() == Some(job_id) => + { Err(CommandError::JobNotStarted(job_id, Box::new(error))) } result => result, @@ -2729,6 +2732,8 @@ pub enum CommandError { Recursive(#[from] Box), #[error("Reqwest error: {0}")] Reqwest(#[from] reqwest::Error), + #[error("{0}")] + Session(String), #[error("SSH signature error on key ID `{0}`")] Signature(KeyId), #[error("Streaming jobs need `--file`")] @@ -2806,6 +2811,14 @@ impl From> for CommandError { CommunicationError(e) if e.is_timeout() => CommandError::TimedOut, CommunicationError(e) => CommandError::Client(format!("Communication error: {e}")), InvalidUpgrade(e) => CommandError::Client(e.to_string()), + ErrorResponse(e) + if matches!( + e.error_code.as_deref().map(str::parse), + Some(Ok(ErrorCode::NoSession | ErrorCode::SessionNotCurrent)) + ) => + { + CommandError::Session(e.message.to_owned()) + } ErrorResponse(e) if e.status() == StatusCode::NOT_FOUND => { CommandError::NotFound(e.message.to_owned()) } diff --git a/client/src/context.rs b/client/src/context.rs index 7d98e7c8..1377e6be 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -155,6 +155,7 @@ pub trait CommandContext: Clone + Send + Sync { seq: SessionSeq, ); fn session_started(&mut self, session: Session, force: bool); + fn session_joined(&mut self, session: Session, force: bool); fn session_stopped(&mut self, session_id: &SessionId); fn attach_allowed(&mut self, key_id: &KeyId, access: Access); fn attach_denied(&mut self, key_id: &KeyId); @@ -203,7 +204,7 @@ pub trait CommandContext: Clone + Send + Sync { fn job_detached(&mut self, id: &JobId); fn job_signing_started(&mut self, id: &JobId); fn job_signing_update(&mut self, id: &JobId); - fn job_signing_finished(&mut self, id: &JobId); + fn job_signing_finished(&mut self, id: &JobId, signed: bool); fn job_signed(&mut self, job: &SignedJob, show: bool); fn job_status(&mut self, id: &JobId, status: &JobStatusMap, style: StatusDisplayStyle); fn read_signed_job(&mut self) -> Result; diff --git a/client/src/repl.rs b/client/src/repl.rs index 0e9fe822..03403113 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -282,6 +282,10 @@ impl CommandContext for Repl { self.cli.session_started(session, force) } + fn session_joined(&mut self, session: Session, force: bool) { + self.cli.session_joined(session, force) + } + fn session_stopped(&mut self, session_id: &SessionId) { self.cli.session_stopped(session_id) } @@ -382,8 +386,8 @@ impl CommandContext for Repl { self.cli.job_signing_update(job_id); } - fn job_signing_finished(&mut self, job_id: &JobId) { - self.cli.job_signing_finished(job_id); + fn job_signing_finished(&mut self, job_id: &JobId, signed: bool) { + self.cli.job_signing_finished(job_id, signed); } fn job_signed(&mut self, job: &SignedJob, show: bool) { diff --git a/server/src/error.rs b/server/src/error.rs index fd08e80e..4f63d338 100644 --- a/server/src/error.rs +++ b/server/src/error.rs @@ -12,6 +12,7 @@ use std::path::{Path, PathBuf}; use dropshot::{ClientErrorStatusCode, HttpError}; use thiserror::Error; +use sush_api::ErrorCode; use sush_common::authn::{Challenge, Nonce}; use sush_common::interactive::InteractiveJobError; use sush_common::jobs::{ExecutionError, JobId, JobOutputHash, SessionId}; @@ -70,13 +71,13 @@ pub enum JobError { MultipleSessions, #[error("No sled this job targets is reachable")] NoReachableTarget, - #[error("No current session")] + #[error("No current session, try `session start`")] NoSession, #[error("Job `{0}` is not the session's next job")] NotNextJob(JobId), #[error("Only the session starter may manage the session")] NotSessionStarter, - #[error("Session `{0}` is no longer current")] + #[error("Session `{0}` is no longer current, try `session attach` or `session start`")] SessionNotCurrent(SessionId), #[error("Job output hash mismatch, file may be corrupt")] OutputHashMismatch(JobId, JobOutputHash), @@ -198,10 +199,19 @@ impl From for HttpError { AttachDenied | NotSessionStarter => { HttpError::for_client_error(None, ClientErrorStatusCode::FORBIDDEN, message) } - IdentityNotFound(_) | JobNotFound(_) | NoSession | OutputFileMissing(_) - | SessionNotCurrent(_) => { + IdentityNotFound(_) | JobNotFound(_) | OutputFileMissing(_) => { HttpError::for_client_error(None, ClientErrorStatusCode::NOT_FOUND, message) } + NoSession => HttpError::for_client_error( + Some(ErrorCode::NoSession.to_string()), + ClientErrorStatusCode::NOT_FOUND, + message, + ), + SessionNotCurrent(_) => HttpError::for_client_error( + Some(ErrorCode::SessionNotCurrent.to_string()), + ClientErrorStatusCode::GONE, + message, + ), DuplicateJobId(_) | InteractiveTarget | InvalidCommand | Json(_) | MultipleSessions | InvalidSessionId | NoReachableTarget | UnknownCubby(_) | UnverifiedJob(_) => { HttpError::for_client_error(None, ClientErrorStatusCode::BAD_REQUEST, message) From 17dcde4ea2ed399cd7c8c25f4012864fab89584b Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 09:31:26 -0600 Subject: [PATCH 14/20] Fetch an interactive job's final status from its sled Co-authored-by: Claude --- client/src/commands.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 71762576..c1c640bf 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1578,7 +1578,8 @@ async fn job_start( } Err(error) => return Err(error), } - job_status(ctx, client, &job_id, StatusDisplayStyle::Short).await? + let status = job_status_map(ctx, client, &job_id, Some(&target)).await?; + ctx.job_status(&job_id, &status, StatusDisplayStyle::Short); } else if streaming { let Some(path) = file else { return Err(CommandError::StreamingNeedsFile); From 165ad9bd61fa4c9f2bd68231e45f67f246bcdee7 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 09:31:26 -0600 Subject: [PATCH 15/20] Stop calling attachable jobs interactive Co-authored-by: Claude --- client/src/cli.rs | 10 ++++------ client/src/commands.rs | 6 +++--- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index 954df353..a0c49ed6 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -826,18 +826,16 @@ impl CommandContext for Cli { fn job_attached(&mut self, job_id: &JobId) { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"attached": job_id})), - OutputFormat::Text => self.note(format!( - "✅ Attached to interactive job `{job_id}`, detach with `^]`" - )), + OutputFormat::Text => { + self.note(format!("✅ Attached to job `{job_id}`, detach with `^]`")) + } } } fn job_detached(&mut self, job_id: &JobId) { match self.get_output_format() { OutputFormat::Json => qprintln!("{}", json!({"detached": job_id})), - OutputFormat::Text => { - self.note(format!("\r✅ Detached from interactive job `{job_id}`")) - } + OutputFormat::Text => self.note(format!("\r✅ Detached from job `{job_id}`")), } } diff --git a/client/src/commands.rs b/client/src/commands.rs index c1c640bf..3653b697 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -2363,7 +2363,7 @@ async fn attach_refused( match status.get(target) { None => CommandError::JobNotOnSled(*job_id, target.clone()), Some(JobStatus::Queued { .. }) => CommandError::JobNotYetRunning(*job_id), - Some(JobStatus::Started { .. }) => CommandError::JobNotInteractive(*job_id), + Some(JobStatus::Started { .. }) => CommandError::JobStartingOrStopping(*job_id), Some(JobStatus::Cancelled { .. }) => CommandError::JobCancelled(*job_id), Some(JobStatus::Skipped { reason, .. }) => CommandError::JobSkipped(*job_id, *reason), Some(JobStatus::Stopped { .. } | JobStatus::Error { .. }) => { @@ -2665,8 +2665,6 @@ pub enum CommandError { JobDidNotRun(JobId), #[error("Job `{0}` has finished, try `job status` or `job output`")] JobFinished(JobId), - #[error("Job `{0}` is not interactive, try `job status` or `job output`")] - JobNotInteractive(JobId), #[error("Job `{0}` not found on sled `{1}`")] JobNotOnSled(JobId, BaseboardId), #[error("Job `{0}` was signed but not started, try `job skip`")] @@ -2675,6 +2673,8 @@ pub enum CommandError { JobNotYetRunning(JobId), #[error("Job `{0}` was skipped on this sled: {1}")] JobSkipped(JobId, SkipReason), + #[error("Job `{0}` is starting or stopping, try again")] + JobStartingOrStopping(JobId), #[error("Job `{0}` is still running")] JobStillRunning(JobId), #[error("JSON error: {0}")] From c9b7563b81e2417a5009771159f736014ab0126d Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 09:35:45 -0600 Subject: [PATCH 16/20] Announce a watched job when it starts, not when it stops Co-authored-by: Claude --- client/src/commands.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 3653b697..3a74e635 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1521,10 +1521,9 @@ async fn job_start( } = start_args; let interactive = job.payload().is_interactive(); let streaming = matches!(job.payload().mode, JobMode::StreamOutput); - let wait = if interactive || streaming { + let watch = wait; + let wait = if interactive || streaming || watch { JobWait::Start - } else if wait { - JobWait::Stop } else { JobWait::None }; @@ -1625,7 +1624,7 @@ async fn job_start( received, }); } - } else if wait.is_some() { + } else if watch { // Watch the whole rack while the start request runs, and keep // watching until the job settles everywhere. ctx.job_watch_started(&job_id); @@ -1702,7 +1701,10 @@ async fn job_start( // While the job runs, a first interrupt stops it and keeps // watching; a second ends the watch. _ = sigint.recv() => { - if started || stopping { + let finished = started + && !last.is_empty() + && last.values().all(JobStatus::is_terminal); + if finished || stopping { break last; } stopping = true; From e0aa5339031b890526e892936df60b7105896678 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 09:40:21 -0600 Subject: [PATCH 17/20] Don't echo ^C into a job watch Co-authored-by: Claude --- client/src/cli.rs | 35 ++++++++++++++++++++++++++++++----- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index a0c49ed6..59bf6183 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -22,6 +22,7 @@ use chrono::TimeDelta; use humantime::format_duration; use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; use rustix::io::ioctl_fionread; +use rustix::termios::{LocalModes, OptionalActions, Termios, isatty, tcgetattr, tcsetattr}; use serde::{Deserialize, Serialize}; use serde_json::{json, to_string as to_json_string, to_string_pretty as to_json_string_pretty}; use sled_hardware_types::BaseboardId; @@ -218,6 +219,31 @@ struct Watch { multi: MultiProgress, bars: BTreeMap, width: usize, + saved: Option, +} + +impl Watch { + fn new() -> Self { + Self { + multi: MultiProgress::new(), + bars: BTreeMap::new(), + width: 0, + saved: no_echo_control(), + } + } +} + +/// Stop the terminal from echoing control sequences. +fn no_echo_control() -> Option { + let input = stdin(); + if !isatty(&input) { + return None; + } + let old = tcgetattr(&input).ok()?; + let mut new = old.clone(); + new.local_modes &= !LocalModes::ECHOCTL; + tcsetattr(&input, OptionalActions::Now, &new).ok()?; + Some(old) } impl fmt::Debug for Watch { @@ -759,11 +785,7 @@ impl CommandContext for Cli { fn job_watch_started(&mut self, _job_id: &JobId) { if matches!(self.get_output_format(), OutputFormat::Text) { - *self.watch.lock().unwrap() = Some(Watch { - multi: MultiProgress::new(), - bars: BTreeMap::new(), - width: 0, - }); + *self.watch.lock().unwrap() = Some(Watch::new()); } } @@ -820,6 +842,9 @@ impl CommandContext for Cli { bar.finish_and_clear(); } let _ = watch.multi.clear(); + if let Some(saved) = watch.saved { + let _ = tcsetattr(stdin(), OptionalActions::Now, &saved); + } } } From 41f532a7711681604e177861e7f68e25b82fc5be Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 15:37:27 -0600 Subject: [PATCH 18/20] Fix a login deadlock in job start -w Co-authored-by: Claude --- client/src/commands.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 3a74e635..69e98d27 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -20,6 +20,7 @@ use std::time::Duration; use async_recursion::async_recursion; use bytesize::ByteSize; use clap::{Parser, Subcommand}; +use futures::future::{Fuse, FusedFuture as _}; use futures::stream; use futures::{FutureExt as _, StreamExt as _}; use http::header::WWW_AUTHENTICATE; @@ -1652,6 +1653,8 @@ async fn job_start( Ok(false) }; pin!(stop); + let poll = Fuse::terminated(); + pin!(poll); let status = loop { select! { // Wait for the start request to finish. @@ -1674,10 +1677,18 @@ async fn job_start( ticker.reset_immediately(); } - _ = ticker.tick() => { - last = match job_status_map(ctx, client, &job_id, job_target.single_baseboard()).await { + _ = ticker.tick(), if poll.is_terminated() => { + let mut poll_ctx = ctx.clone(); + let job_id = &job_id; + poll.set( + async move { job_status_map(&mut poll_ctx, client, job_id, via).await } + .fuse(), + ); + } + + status = &mut poll, if !poll.is_terminated() => { + last = match status { Ok(status) => status, - // The job may not be visible anywhere yet. Err(CommandError::NotFound(_)) => JobStatusMap::new(), Err(error) => { ctx.job_watch_finished(&job_id); From 463d7d317ac384584941a8561e8a575e640c4007 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 16:30:18 -0600 Subject: [PATCH 19/20] Watch a single-sled job on its own sled Co-authored-by: Claude --- client/src/commands.rs | 3 +-- common/src/targets.rs | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index 69e98d27..e9cf682b 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1629,7 +1629,6 @@ async fn job_start( // Watch the whole rack while the start request runs, and keep // watching until the job settles everywhere. ctx.job_watch_started(&job_id); - let job_target = job.payload().target().clone(); let mut sleds = None; let mut ticker = interval(WATCH_INTERVAL); ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); @@ -1641,7 +1640,7 @@ async fn job_start( let mut stopped = false; let mut sigint = signal(SignalKind::interrupt())?; let mut stop_ctx = ctx.clone(); - let via = job_target.single_baseboard(); + let via = job.payload().target().is_single().then_some(&target); let stop = async { // A stop can race the start, so retry it a few times. for _ in 0..3 { diff --git a/common/src/targets.rs b/common/src/targets.rs index d51233f1..5d582a05 100644 --- a/common/src/targets.rs +++ b/common/src/targets.rs @@ -127,6 +127,11 @@ impl Target { matches!(self, Self::All) } + /// Whether this target names exactly one sled, by baseboard or cubby. + pub fn is_single(&self) -> bool { + matches!(self, Self::Sleds(sleds) if sleds.len() == 1) + } + /// A singular target or `None`. pub fn single_baseboard(&self) -> Option<&BaseboardId> { match self { @@ -269,6 +274,15 @@ mod test { assert_eq!("14, 16".parse::().unwrap().to_string(), "14,16"); } + #[test] + fn targets_single() { + let single = |target: &str| target.parse::().unwrap().is_single(); + assert!(single("14")); + assert!(single("913-0000019:BRM42220031")); + assert!(!single("14,16")); + assert!(!single("*")); + } + #[test] fn targets_resolve() { let brm31 = baseboard("913-0000019:BRM42220031"); From e135468783d874032003fe3979a1885815ae1ffb Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 27 Oct 2026 19:22:01 -0600 Subject: [PATCH 20/20] Tidy job start -w output Co-authored-by: Claude --- README.md | 18 +++++-------- client/src/cli.rs | 61 +++++++++++++++++++----------------------- client/src/commands.rs | 10 +++---- client/src/context.rs | 4 +-- client/src/repl.rs | 4 --- 5 files changed, 39 insertions(+), 58 deletions(-) diff --git a/README.md b/README.md index ea75c945..665d1fcb 100644 --- a/README.md +++ b/README.md @@ -37,22 +37,18 @@ sush# job start -w hostname 👋 Please confirm user presence to sign with key `federal-worth-fee-seed-skin-interest-road-luggage` ✅ Session is now `front-edit-bless-suggest-defy-bacon-retire-person` ✅ Signed request for job `kingdom-owner-dilemma-craft-soda-hungry-lumber-festival` -✅ Job ID: kingdom-owner-dilemma-craft-soda-hungry-lumber-festival +✅ Started job `kingdom-owner-dilemma-craft-soda-hungry-lumber-festival` 913-0000019:BRM23230010 Stopped, exit 0 (5ms 545us 307ns), 12 B out, 0 B err 913-0000019:BRM23230018 Stopped, exit 0 (6ms 165us 887ns), 12 B out, 0 B err 913-0000019:BRM27230037 Stopped, exit 0 (5ms 927us 316ns), 12 B out, 0 B err 913-0000023:2F8JEXDK Stopped, exit 0 (5ms 282us 774ns), 9 B out, 0 B err - » 913-0000019:BRM23230010 « -✅ Job stdout: + » 913-0000019:BRM23230010 stdout « BRM23230010 - » 913-0000019:BRM23230018 « -✅ Job stdout: + » 913-0000019:BRM23230018 stdout « BRM23230018 - » 913-0000019:BRM27230037 « -✅ Job stdout: + » 913-0000019:BRM27230037 stdout « BRM27230037 - » 913-0000023:2F8JEXDK « -✅ Job stdout: + » 913-0000023:2F8JEXDK stdout « 2F8JEXDK ``` @@ -100,9 +96,9 @@ test try running the command `echo $SUSH_JOB_ID`: ``` sush# job start --wait "echo $SUSH_JOB_ID" ✅ Signed request for job `install-there-mutual-warfare-sound-live-order-man` -✅ Job ID: install-there-mutual-warfare-sound-live-order-man +✅ Started job `install-there-mutual-warfare-sound-live-order-man` a part:0001 Stopped, exit 0 (68ms 190us 114ns), 50 B out, 0 B err -✅ Job stdout: + » a part:0001 stdout « install-there-mutual-warfare-sound-live-order-man ``` diff --git a/client/src/cli.rs b/client/src/cli.rs index 59bf6183..98228c1b 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -657,13 +657,6 @@ impl CommandContext for Cli { skipped } - fn job_output_target(&mut self, target: &BaseboardId) { - match self.get_output_format() { - OutputFormat::Json => (), - OutputFormat::Text => self.note(format!(" » {target} «")), - } - } - fn job_output( &mut self, job_id: &JobId, @@ -692,22 +685,20 @@ impl CommandContext for Cli { }) ) } - OutputFormat::Text if binary => { - if let Err(err) = quiet_write(output) { - eprintln!("{err}"); - } - } OutputFormat::Text if output.is_empty() => (), OutputFormat::Text => { - match stream { - JobOutputStream::Stdout => self.note("✅ Job stdout:"), - JobOutputStream::Stderr => self.note("✅ Job stderr:"), - }; - let output = String::from_utf8_lossy(output); - if output.ends_with('\n') { - qprint!("{output}"); + self.note(format!(" » {target} {stream} «")); + if binary { + if let Err(err) = quiet_write(output) { + eprintln!("{err}"); + } } else { - qprintln!("{output}"); + let output = String::from_utf8_lossy(output); + if output.ends_with('\n') { + qprint!("{output}"); + } else { + qprintln!("{output}"); + } } } } @@ -924,25 +915,27 @@ impl CommandContext for Cli { } }) ), - OutputFormat::Text if matches!(style, StatusDisplayStyle::Short) => { - let icon = if status.values().any(|s| { - matches!( - s, - JobStatus::Error { .. } - | JobStatus::Cancelled { .. } - | JobStatus::Stopped { result: Err(_), .. } - ) - }) { - "❌" - } else { - "✅" - }; + OutputFormat::Text if !matches!(style, StatusDisplayStyle::Full) => { + if matches!(style, StatusDisplayStyle::Medium) { + let icon = if status.values().any(|s| { + matches!( + s, + JobStatus::Error { .. } + | JobStatus::Cancelled { .. } + | JobStatus::Stopped { result: Err(_), .. } + ) + }) { + "❌" + } else { + "✅" + }; + qprintln!("{icon} Job ID:\t{job_id}"); + } let width = status .keys() .map(|b| b.to_string().len()) .max() .unwrap_or(0); - qprintln!("{icon} Job ID:\t{job_id}"); for (baseboard_id, status) in status { let id = baseboard_id.to_string(); qprintln!(" {id: return Err(error), } let status = job_status_map(ctx, client, &job_id, Some(&target)).await?; - ctx.job_status(&job_id, &status, StatusDisplayStyle::Short); + ctx.job_status(&job_id, &status, StatusDisplayStyle::Medium); } else if streaming { let Some(path) = file else { return Err(CommandError::StreamingNeedsFile); @@ -1748,9 +1748,6 @@ async fn job_start( // Show the output of every sled the job ran on. let multiple = status.len() > 1; for baseboard in status.keys() { - if multiple { - ctx.job_output_target(baseboard); - } for stream in [Stdout, Stderr] { match with_login_via(ctx, client, Some(baseboard), async || { client @@ -2150,7 +2147,6 @@ async fn job_output( ))); } for baseboard in status.keys() { - ctx.job_output_target(baseboard); if let Err(error) = job_output_from(ctx, client, baseboard, stream, args.clone()).await { ctx.command_error(&error); } diff --git a/client/src/context.rs b/client/src/context.rs index 1377e6be..805677fc 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -73,8 +73,9 @@ pub enum OutputFormat { /// How much of a job's per-sled status to show. #[derive(Clone, Copy, Debug, Default)] pub enum StatusDisplayStyle { - #[default] Short, + #[default] + Medium, Full, } @@ -186,7 +187,6 @@ pub trait CommandContext: Clone + Send + Sync { output: &[u8], binary: bool, ); - fn job_output_target(&mut self, target: &BaseboardId); fn job_output_started( &mut self, id: &JobId, diff --git a/client/src/repl.rs b/client/src/repl.rs index 03403113..32226c54 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -330,10 +330,6 @@ impl CommandContext for Repl { self.cli.job_skipped(job_id) } - fn job_output_target(&mut self, target: &BaseboardId) { - self.cli.job_output_target(target) - } - fn job_output( &mut self, job_id: &JobId,