From 7f9d32adb7de9817b47a2490221e06dbced2cb04 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Tue, 11 Aug 2026 23:50:05 +0000 Subject: [PATCH 1/4] Shorten job status and fetch output rack-wide One status line per sled by default (--full restores the blocks), and bare `job stdout` walks every sled with recorded status instead of resolving to one. --- client/src/cli.rs | 66 ++++++++++++++++++++++++++++++- client/src/commands.rs | 90 ++++++++++++++++++++++++++++++++++++++---- client/src/context.rs | 12 +++++- client/src/repl.rs | 11 ++++-- 4 files changed, 166 insertions(+), 13 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index 76aad00..4768a57 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -15,6 +15,7 @@ use humantime::format_duration; use indicatif::{ProgressBar, ProgressStyle}; use rustix::io::ioctl_fionread; use serde_json::{json, to_string as to_json_string, to_string_pretty as to_json_string_pretty}; +use sled_hardware_types::BaseboardId; use x509_cert::Certificate; use x509_cert::der::Encode as _; @@ -27,7 +28,7 @@ use sush_common::keys::{KeyId, Signature, SshPublicKey}; use crate::AuthzSigner; use crate::commands::{CommandError, GlobalArgs}; -use crate::context::{CommandContext, OutputFormat}; +use crate::context::{CommandContext, OutputFormat, StatusDisplayStyle}; #[derive(Clone, Debug, Default)] pub struct Cli { @@ -218,6 +219,13 @@ impl CommandContext for Cli { error } + 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} ⟸"), + } + } + fn job_output( &mut self, _job_id: &JobId, @@ -397,7 +405,7 @@ impl CommandContext for Cli { } } - fn job_status(&mut self, _job_id: &JobId, status: &JobStatusMap) { + fn job_status(&mut self, job_id: &JobId, status: &JobStatusMap, style: StatusDisplayStyle) { fn format_elapsed_duration(duration: TimeDelta) -> String { if let Ok(duration) = duration.to_std() { format_duration(duration).to_string() @@ -409,6 +417,60 @@ impl CommandContext for Cli { // TODO: parallel status display match self.get_output_format() { OutputFormat::Json => println!("{}", json!(job_status_to_json_map(status.clone()))), + 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 { + "✅" + }; + let width = status + .keys() + .map(|b| b.to_string().len()) + .max() + .unwrap_or(0); + println!("{icon} Job ID:\t{job_id}"); + for (baseboard_id, status) in status { + let id = baseboard_id.to_string(); + match status { + JobStatus::Cancelled { + time_cancelled, + actor, + .. + } => { + println!(" {id: println!(" {id: { + println!(" {id: { + let duration = format_elapsed_duration(status.time_elapsed()); + let stdout_len = byte_size(output.stdout_len); + let stderr_len = byte_size(output.stderr_len); + let result = match result { + Ok(exit_status) => format!("exit {exit_status}"), + Err(err) => err.to_string(), + }; + println!( + " {id: println!(" {id: { for (baseboard_id, status) in status { match status { diff --git a/client/src/commands.rs b/client/src/commands.rs index aa26def..cac6108 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -52,7 +52,7 @@ use sush_common::keys::{KeyError, KeyId, Signer as _}; use sush_common::targets::{SledId, Target}; use crate::ByteStream; -use crate::context::{Authz, CommandContext, OutputFormat}; +use crate::context::{Authz, CommandContext, OutputFormat, StatusDisplayStyle}; use crate::identity::{IdentityError, SshAgentConnection}; use crate::interactive::interactive_job; #[cfg(feature = "permslip")] @@ -195,6 +195,31 @@ impl ClientArgs { } } +/// A rack-wide target can't combine with `--binary` or `--file`. +#[tokio::test] +async fn output_needs_target() { + use crate::cli::Cli; + + let client = + Client::new_with_client("http://[::1]:1", reqwest::Client::new(), Default::default()); + let output = JobOutput::try_parse_from([ + "job-stdout", + "--binary", + "sea-say-sting-palm-tunnel-festival-pull-bid", + ]) + .unwrap(); + let err = job_output( + &mut Cli::default(), + &client, + &"*".parse::().unwrap(), + Stdout, + output, + ) + .await + .unwrap_err(); + assert!(matches!(err, CommandError::OutputNeedsTarget)); +} + /// [`ClientArgs`] must satisfy Clap's internal consistency asserts /// (unique shorts per subcommand, valid references). #[test] @@ -397,6 +422,10 @@ pub enum JobCommand { /// The job whose status should be fetched. #[clap(env = SUSH_JOB_ID)] job_id: JobId, + + /// Show full per-sled status instead of one line per sled. + #[arg(short, long)] + full: bool, }, /// Get the standard output of a job. @@ -977,15 +1006,20 @@ async fn job( Ok(()) } - (JobCommand::Status { job_id }, Some(client)) => job_status(ctx, client, &job_id).await, + (JobCommand::Status { job_id, full }, Some(client)) => { + let style = if full { + StatusDisplayStyle::Full + } else { + StatusDisplayStyle::Short + }; + job_status(ctx, client, &job_id, style).await + } (JobCommand::Stdout { target, output }, Some(client)) => { - let target = resolve_target(client, &target).await?; job_output(ctx, client, &target, Stdout, output).await } (JobCommand::Stderr { target, output }, Some(client)) => { - let target = resolve_target(client, &target).await?; job_output(ctx, client, &target, Stderr, output).await } @@ -1010,7 +1044,7 @@ async fn job( let status = job_status_try_from_json_map(job) .map_err(CommandError::BaseboardIdParseError)?; if let Some(s) = status.values().next() { - ctx.job_status(s.job_id(), &status); + ctx.job_status(s.job_id(), &status, StatusDisplayStyle::Short); } } Ok(()) @@ -1078,7 +1112,9 @@ async fn job_start( start.await?; ctx.job_started(&job); match job_attach(ctx, client, &job_id, &target).await { - Ok(()) | Err(CommandError::NotFound) => job_status(ctx, client, &job_id).await?, + Ok(()) | Err(CommandError::NotFound) => { + job_status(ctx, client, &job_id, StatusDisplayStyle::Short).await? + } Err(error) => return Err(error), } } else if wait.is_some() { @@ -1127,7 +1163,7 @@ async fn job_start( } // Show the job status and output. - job_status(ctx, client, &job_id).await?; + job_status(ctx, client, &job_id, StatusDisplayStyle::Short).await?; for stream in [Stdout, Stderr] { match with_login_via(ctx, client, Some(&target), async || { client @@ -1175,6 +1211,7 @@ async fn job_status( ctx: &mut impl CommandContext, client: &Client, job_id: &JobId, + style: StatusDisplayStyle, ) -> Result<(), CommandError> { let status = with_login(ctx, client, async || { client.job_status().job_id(job_id).send().await @@ -1184,6 +1221,7 @@ async fn job_status( ctx.job_status( job_id, &job_status_try_from_json_map(status).map_err(CommandError::BaseboardIdParseError)?, + style, ); Ok(()) } @@ -1222,6 +1260,42 @@ type FutureChunk<'a> = dyn Future> + Send + /// Download job output. async fn job_output( + ctx: &mut impl CommandContext, + client: &Client, + target: &Target, + stream: JobOutputStream, + args: JobOutput, +) -> Result<(), CommandError> { + if !target.is_all() { + let baseboard = resolve_target(client, target).await?; + return job_output_from(ctx, client, &baseboard, stream, args).await; + } + if args.binary || args.file.is_some() { + return Err(CommandError::OutputNeedsTarget); + } + + // Fetch output from every sled with a recorded status. + let status = job_status_try_from_json_map( + with_login(ctx, client, async || { + client.job_status().job_id(&args.job_id).send().await + }) + .await? + .into_inner(), + ) + .map_err(CommandError::BaseboardIdParseError)?; + if status.is_empty() { + return Err(CommandError::NotFound); + } + 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); + } + } + Ok(()) +} + +async fn job_output_from( ctx: &mut impl CommandContext, client: &Client, target: &BaseboardId, @@ -1550,6 +1624,8 @@ pub enum CommandError { expected: JobOutputHash, received: JobOutputHash, }, + #[error("❌ `--binary` and `--file` need a specific `--target`")] + OutputNeedsTarget, #[cfg(feature = "permslip")] #[error("❌ permslip error: {0}")] Permslip(#[from] PermslipError), diff --git a/client/src/context.rs b/client/src/context.rs index 281a32c..f3dea9f 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -11,6 +11,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use clap::ValueEnum; +use sled_hardware_types::BaseboardId; use x509_cert::Certificate; use sush_common::authn::{BoundRequest, Credentials, Identity, RequestKey}; @@ -62,6 +63,14 @@ pub enum OutputFormat { Json, } +/// How much of a job's per-sled status to show. +#[derive(Clone, Copy, Debug, Default)] +pub enum StatusDisplayStyle { + #[default] + Short, + Full, +} + impl OutputFormat { pub fn as_str(&self) -> &'static str { match self { @@ -116,6 +125,7 @@ pub trait CommandContext: Clone + Send + Sync { fn job_stopped(&mut self, id: &JobId); fn job_error(&mut self, error: CommandError) -> CommandError; fn job_output(&mut self, id: &JobId, stream: JobOutputStream, output: &[u8], binary: bool); + fn job_output_target(&mut self, target: &BaseboardId); fn job_output_started( &mut self, id: &JobId, @@ -134,7 +144,7 @@ pub trait CommandContext: Clone + Send + Sync { fn job_signing_update(&mut self, id: &JobId); fn job_signing_finished(&mut self, id: &JobId); fn job_signed(&mut self, job: &SignedJob, show: bool); - fn job_status(&mut self, id: &JobId, status: &JobStatusMap); + fn job_status(&mut self, id: &JobId, status: &JobStatusMap, style: StatusDisplayStyle); fn read_signed_job(&mut self) -> Result; // SSH agent and identity diff --git a/client/src/repl.rs b/client/src/repl.rs index 8afe8ef..81666d5 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -15,6 +15,7 @@ use clap::Parser; use rustyline::DefaultEditor; use rustyline::error::ReadlineError; use shlex::split as split_command; +use sled_hardware_types::BaseboardId; use x509_cert::Certificate; use xdg::BaseDirectories; @@ -29,7 +30,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}; +use crate::context::{CommandContext, OutputFormat, StatusDisplayStyle}; use crate::{AuthzSigner, Client}; const PREFIX: &str = "sush"; @@ -260,6 +261,10 @@ impl CommandContext for Repl { self.cli.job_error(error) } + fn job_output_target(&mut self, target: &BaseboardId) { + self.cli.job_output_target(target) + } + fn job_output(&mut self, job_id: &JobId, stream: JobOutputStream, output: &[u8], binary: bool) { self.set_job_id(Some(job_id.to_owned())); self.cli.job_output(job_id, stream, output, binary); @@ -325,9 +330,9 @@ impl CommandContext for Repl { self.cli.job_signed(job, show); } - fn job_status(&mut self, job_id: &JobId, status: &JobStatusMap) { + fn job_status(&mut self, job_id: &JobId, status: &JobStatusMap, style: StatusDisplayStyle) { self.set_job_id(Some(job_id.to_owned())); - self.cli.job_status(job_id, status); + self.cli.job_status(job_id, status, style); } fn read_signed_job(&mut self) -> Result { From c9362c48644ec3ea757d3a4d0ab1ff59016ea425 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Wed, 12 Aug 2026 00:05:03 +0000 Subject: [PATCH 2/4] Accept bare serials as output and attach targets The client resolves a serial (case-insensitively) against the sleds that have a status for the job. Signed targets are unchanged. --- client/src/commands.rs | 116 +++++++++++++++++++++++++++++++++-------- 1 file changed, 94 insertions(+), 22 deletions(-) diff --git a/client/src/commands.rs b/client/src/commands.rs index cac6108..1440723 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -11,6 +11,7 @@ use std::io::{Read as _, Seek as _, SeekFrom, Write as _, stdin}; use std::num::{NonZeroU8, NonZeroU64}; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::str::FromStr; use std::time::Duration; use async_recursion::async_recursion; @@ -45,8 +46,8 @@ use sush_common::jobs::JobOutputStream::{self, Stderr, Stdout}; #[cfg(feature = "permslip")] use sush_common::jobs::JobStartRequest; use sush_common::jobs::{ - Access, JobId, JobLimits, JobOutputHash, JobOutputState, JobStatus, Session, SessionId, - SignedJob, job_status_try_from_json_map, + Access, JobId, JobLimits, JobOutputHash, JobOutputState, JobStatus, JobStatusMap, Session, + SessionId, SignedJob, job_status_try_from_json_map, }; use sush_common::keys::{KeyError, KeyId, Signer as _}; use sush_common::targets::{SledId, Target}; @@ -211,7 +212,7 @@ async fn output_needs_target() { let err = job_output( &mut Cli::default(), &client, - &"*".parse::().unwrap(), + &"*".parse::().unwrap(), Stdout, output, ) @@ -220,6 +221,20 @@ async fn output_needs_target() { assert!(matches!(err, CommandError::OutputNeedsTarget)); } +/// Anything the target grammar accepts stays a target; bare serials +/// fall through; everything else still fails. +#[test] +fn target_arg() { + assert!(matches!("*".parse(), Ok(TargetArg::Target(_)))); + assert!(matches!("14".parse(), Ok(TargetArg::Target(_)))); + assert!(matches!( + "913-0000019:BRM42220030".parse(), + Ok(TargetArg::Target(_)) + )); + assert!(matches!("brm42220030".parse(), Ok(TargetArg::Serial(s)) if s == "brm42220030")); + assert!("not,a:target!".parse::().is_err()); +} + /// [`ClientArgs`] must satisfy Clap's internal consistency asserts /// (unique shorts per subcommand, valid references). #[test] @@ -433,7 +448,7 @@ pub enum JobCommand { Stdout { /// The sled from which output should be fetched. #[arg(short = 'T', long, default_value = "*")] - target: Target, + target: TargetArg, #[clap(flatten)] output: JobOutput, @@ -444,7 +459,7 @@ pub enum JobCommand { Stderr { /// The sled from which output should be fetched. #[arg(short = 'T', long, default_value = "*")] - target: Target, + target: TargetArg, #[clap(flatten)] output: JobOutput, @@ -458,7 +473,7 @@ pub enum JobCommand { /// The sled to attach to. #[arg(short = 'T', long, default_value = "*")] - target: Target, + target: TargetArg, }, /// Show status of previously started jobs. @@ -1024,7 +1039,10 @@ async fn job( } (JobCommand::Attach { job_id, target }, Some(client)) => { - let target = resolve_target(client, &target).await?; + let target = match &target { + TargetArg::Target(target) => resolve_target(client, target).await?, + TargetArg::Serial(serial) => resolve_serial(ctx, client, &job_id, serial).await?, + }; job_attach(ctx, client, &job_id, &target).await?; Ok(()) } @@ -1213,17 +1231,23 @@ async fn job_status( job_id: &JobId, style: StatusDisplayStyle, ) -> Result<(), CommandError> { + let status = job_status_map(ctx, client, job_id).await?; + ctx.job_status(job_id, &status, style); + Ok(()) +} + +/// Fetch a job's rack-wide status map. +async fn job_status_map( + ctx: &mut impl CommandContext, + client: &Client, + job_id: &JobId, +) -> Result { let status = with_login(ctx, client, async || { client.job_status().job_id(job_id).send().await }) .await? .into_inner(); - ctx.job_status( - job_id, - &job_status_try_from_json_map(status).map_err(CommandError::BaseboardIdParseError)?, - style, - ); - Ok(()) + job_status_try_from_json_map(status).map_err(CommandError::BaseboardIdParseError) } /// Stream some bytes into a vector. @@ -1262,10 +1286,17 @@ type FutureChunk<'a> = dyn Future> + Send + async fn job_output( ctx: &mut impl CommandContext, client: &Client, - target: &Target, + target: &TargetArg, stream: JobOutputStream, args: JobOutput, ) -> Result<(), CommandError> { + let target = match target { + TargetArg::Serial(serial) => { + let baseboard = resolve_serial(ctx, client, &args.job_id, serial).await?; + return job_output_from(ctx, client, &baseboard, stream, args).await; + } + TargetArg::Target(target) => target, + }; if !target.is_all() { let baseboard = resolve_target(client, target).await?; return job_output_from(ctx, client, &baseboard, stream, args).await; @@ -1275,14 +1306,7 @@ async fn job_output( } // Fetch output from every sled with a recorded status. - let status = job_status_try_from_json_map( - with_login(ctx, client, async || { - client.job_status().job_id(&args.job_id).send().await - }) - .await? - .into_inner(), - ) - .map_err(CommandError::BaseboardIdParseError)?; + let status = job_status_map(ctx, client, &args.job_id).await?; if status.is_empty() { return Err(CommandError::NotFound); } @@ -1545,9 +1569,55 @@ async fn resolve_target(client: &Client, target: &Target) -> Result::Err; + + fn from_str(s: &str) -> Result { + match s.parse() { + Ok(target) => Ok(Self::Target(target)), + Err(_) if !s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric()) => { + Ok(Self::Serial(s.to_owned())) + } + Err(err) => Err(err), + } + } +} + +/// Resolve a bare serial number against the sleds that have a status +/// for a job. +async fn resolve_serial( + ctx: &mut impl CommandContext, + client: &Client, + job_id: &JobId, + serial: &str, +) -> Result { + let status = job_status_map(ctx, client, job_id).await?; + let mut matches = status + .keys() + .filter(|b| b.serial_number.eq_ignore_ascii_case(serial)); + match (matches.next(), matches.next()) { + (Some(baseboard), None) => Ok(baseboard.clone()), + (None, _) => Err(CommandError::UnknownSerial { + serial: serial.to_owned(), + job_id: job_id.to_owned(), + }), + (Some(_), Some(_)) => Err(CommandError::AmbiguousSerial(serial.to_owned())), + } +} + /// 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")] + AmbiguousSerial(String), #[error("❌ Authentication error")] Authn(#[from] AuthnError), #[error("❌ Canceled")] @@ -1654,6 +1724,8 @@ pub enum CommandError { TimedOut, #[error("❌ Too much output to display on terminal, try `--file`")] TooMuchOutput, + #[error("❌ No sled with serial `{serial}` has a status for job `{job_id}`")] + UnknownSerial { serial: String, job_id: JobId }, #[error("❌ Chain root does not match any supplied root certificate")] UntrustedRoot, #[error("❌ Can't start interactive session: {0}")] From bdced1d887abaed86f0c02543e16de1d8b1e6787 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Wed, 12 Aug 2026 04:11:17 +0000 Subject: [PATCH 3/4] Watch job status with a line per sled `job start -w` and the new `job status --wait` poll the rack-wide status map and render a live line per sled, settling once every known sled is terminal and gossip has had a grace period to name stragglers. --- client/src/cli.rs | 178 +++++++++++++++++++++++-------------- client/src/commands.rs | 195 +++++++++++++++++++++++++++++++++++------ client/src/context.rs | 7 +- client/src/repl.rs | 25 +++--- common/src/jobs.rs | 8 ++ 5 files changed, 305 insertions(+), 108 deletions(-) diff --git a/client/src/cli.rs b/client/src/cli.rs index 4768a57..97fe1ec 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -4,6 +4,8 @@ //! Possibly-interactive command-line interface. +use std::collections::BTreeMap; +use std::fmt; use std::io::{self, BufRead as _, Read as _, Write as _, stderr, stdin, stdout}; use std::path::Path; use std::sync::{Arc, Mutex}; @@ -12,7 +14,7 @@ use std::time::{Duration, SystemTime}; use bytesize::ByteSize; use chrono::TimeDelta; use humantime::format_duration; -use indicatif::{ProgressBar, ProgressStyle}; +use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; use rustix::io::ioctl_fionread; use serde_json::{json, to_string as to_json_string, to_string_pretty as to_json_string_pretty}; use sled_hardware_types::BaseboardId; @@ -35,6 +37,7 @@ pub struct Cli { globals: Arc>, output: Arc>, progress: Arc>>, + watch: Arc>>, session: Arc>>, credentials: AuthzSigner, } @@ -43,6 +46,57 @@ fn byte_size(len: u64) -> bytesize::Display { ByteSize::b(len).display().si() } +fn format_elapsed_duration(duration: TimeDelta) -> String { + if let Ok(duration) = duration.to_std() { + format_duration(duration).to_string() + } else { + String::from("negative duration, times may be unreliable") + } +} + +/// One sled's worth of job status, without the sled's name. +fn short_status_row(status: &JobStatus) -> String { + match status { + JobStatus::Cancelled { + time_cancelled, + actor, + .. + } => format!("Cancelled at {time_cancelled} by {actor}"), + JobStatus::Queued { + time_queued, actor, .. + } => format!("Queued at {time_queued} by {actor}"), + JobStatus::Started { time_started, .. } => format!("Started at {time_started}"), + JobStatus::Stopped { result, output, .. } => { + let duration = format_elapsed_duration(status.time_elapsed()); + let stdout_len = byte_size(output.stdout_len); + let stderr_len = byte_size(output.stderr_len); + let result = match result { + Ok(exit_status) => format!("exit {exit_status}"), + Err(err) => err.to_string(), + }; + format!("Stopped, {result} ({duration}), {stdout_len} out, {stderr_len} err") + } + JobStatus::Error { + time_error, error, .. + } => format!("Error at {time_error}: {error}"), + } +} + +/// Live per-sled status lines for a watched job. +struct Watch { + multi: MultiProgress, + bars: BTreeMap, + width: usize, +} + +impl fmt::Debug for Watch { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Watch") + .field("bars", &self.bars.keys()) + .finish_non_exhaustive() + } +} + impl CommandContext for Cli { // Context management @@ -210,7 +264,13 @@ impl CommandContext for Cli { fn job_stopped(&mut self, job_id: &JobId) { match self.get_output_format() { OutputFormat::Json => println!("{}", json!(job_id)), - OutputFormat::Text => println!("\r✅ Stopped job `{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}`"); + } + } } } @@ -309,34 +369,61 @@ impl CommandContext for Cli { } } - fn job_polling_started(&mut self, job_id: &JobId, elapsed: Duration) { - let mut progress = self.progress.lock().unwrap(); - if matches!(self.get_output_format(), OutputFormat::Text) && progress.is_none() { - let bar = ProgressBar::new_spinner(); - bar.set_elapsed(elapsed); - bar.set_prefix(format!("Waiting for job `{job_id}`")); - bar.set_style( - ProgressStyle::with_template( - "{spinner} \ - {prefix} \ - [{elapsed_precise}] \ - {msg}", - ) - .unwrap(), - ); - *progress = Some(bar); + 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, + }); + } + } + + fn job_watch_update(&mut self, status: &JobStatusMap) { + let mut guard = self.watch.lock().unwrap(); + let Some(watch) = guard.as_mut() else { return }; + + // Widen every name column if a longer baseboard ID appears. + let width = status + .keys() + .map(|b| b.to_string().len()) + .max() + .unwrap_or(0); + if width > watch.width { + watch.width = width; + for (baseboard_id, bar) in &watch.bars { + bar.set_prefix(format!("{: String { - if let Ok(duration) = duration.to_std() { - format_duration(duration).to_string() - } else { - String::from("negative duration, times may be unreliable") - } - } - - // TODO: parallel status display match self.get_output_format() { OutputFormat::Json => println!("{}", json!(job_status_to_json_map(status.clone()))), OutputFormat::Text if matches!(style, StatusDisplayStyle::Short) => { @@ -438,37 +516,7 @@ impl CommandContext for Cli { println!("{icon} Job ID:\t{job_id}"); for (baseboard_id, status) in status { let id = baseboard_id.to_string(); - match status { - JobStatus::Cancelled { - time_cancelled, - actor, - .. - } => { - println!(" {id: println!(" {id: { - println!(" {id: { - let duration = format_elapsed_duration(status.time_elapsed()); - let stdout_len = byte_size(output.stdout_len); - let stderr_len = byte_size(output.stderr_len); - let result = match result { - Ok(exit_status) => format!("exit {exit_status}"), - Err(err) => err.to_string(), - }; - println!( - " {id: println!(" {id: { diff --git a/client/src/commands.rs b/client/src/commands.rs index 1440723..ad29c6b 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -89,7 +89,6 @@ const PARALLEL_CHUNKS: NonZeroU8 = NonZeroU8::new(8).unwrap(); const MAX_PARALLEL_CHUNKS: NonZeroU8 = NonZeroU8::new(64).unwrap(); // Job polling and spinner update intervals. -const JOB_START_UPDATE_INTERVAL: Duration = Duration::from_millis(250); const JOB_STOP_RETRY_INTERVAL: Duration = Duration::from_millis(100); #[cfg(feature = "permslip")] const SIGNING_UPDATE_INTERVAL: Duration = Duration::from_millis(100); @@ -221,6 +220,60 @@ async fn output_needs_target() { assert!(matches!(err, CommandError::OutputNeedsTarget)); } +/// The watch settles only after the sled set is stable and every sled +/// is terminal for consecutive polls. +#[test] +fn watch_settling() { + use chrono::Utc; + + fn sled(serial: &str) -> BaseboardId { + BaseboardId { + part_number: "913-0000019".to_owned(), + serial_number: serial.to_owned(), + } + } + let job_id: JobId = "sea-say-sting-palm-tunnel-festival-pull-bid" + .parse() + .unwrap(); + let terminal = JobStatus::Cancelled { + job_id, + time_cancelled: Utc::now(), + actor: KeyId::random(), + }; + let running = JobStatus::Started { + job_id, + time_started: Utc::now(), + }; + + let mut quiet = 0; + let mut status = JobStatusMap::new(); + assert!( + !watch_done(1, 0, &status, &mut quiet), + "empty never settles" + ); + + status.insert(sled("A"), terminal.clone()); + assert!(!watch_done(2, 0, &status, &mut quiet), "new sled resets"); + assert!(!watch_done(3, 1, &status, &mut quiet), "first quiet poll"); + assert!( + !watch_done(4, 1, &status, &mut quiet), + "quiet but too young for gossip" + ); + assert!(watch_done(5, 1, &status, &mut quiet), "old enough, quiet"); + + let mut quiet = 0; + status.insert(sled("B"), running); + assert!(!watch_done(6, 2, &status, &mut quiet), "running sled holds"); + assert_eq!(quiet, 0); + + status.insert(sled("B"), terminal); + assert!(!watch_done(7, 2, &status, &mut quiet)); + assert!( + watch_done(8, 2, &status, &mut quiet), + "settles once terminal" + ); +} + /// Anything the target grammar accepts stays a target; bare serials /// fall through; everything else still fails. #[test] @@ -441,6 +494,10 @@ pub enum JobCommand { /// Show full per-sled status instead of one line per sled. #[arg(short, long)] full: bool, + + /// Watch the status until the job settles on every sled. + #[arg(short, long)] + wait: bool, }, /// Get the standard output of a job. @@ -1021,13 +1078,19 @@ async fn job( Ok(()) } - (JobCommand::Status { job_id, full }, Some(client)) => { + (JobCommand::Status { job_id, full, wait }, Some(client)) => { let style = if full { StatusDisplayStyle::Full } else { StatusDisplayStyle::Short }; - job_status(ctx, client, &job_id, style).await + if wait { + let status = job_watch(ctx, client, &job_id).await?; + ctx.job_status(&job_id, &status, style); + Ok(()) + } else { + job_status(ctx, client, &job_id, style).await + } } (JobCommand::Stdout { target, output }, Some(client)) => { @@ -1136,36 +1199,60 @@ async fn job_start( Err(error) => return Err(error), } } else if wait.is_some() { - let mut interval = interval(JOB_START_UPDATE_INTERVAL); + // Watch the whole rack while the start request runs, and keep + // watching until the job settles everywhere. + ctx.job_watch_started(&job_id); + let mut ticker = interval(WATCH_INTERVAL); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + let mut last = JobStatusMap::new(); + let mut sleds = 0; + let mut quiet = 0; + let mut polls = 0; + let mut started = false; let mut stopped = false; - interval.set_missed_tick_behavior(MissedTickBehavior::Skip); - ctx.job_polling_started(&job_id, interval.period()); - loop { + let status = loop { select! { // Wait for the start request to finish. - start_result = &mut start => { - if !stopped { - ctx.job_polling_finished(&job_id); + start_result = &mut start, if !started => { + if let Err(error) = start_result { + ctx.job_watch_finished(&job_id); + return Err(error); } - start_result?; ctx.job_started(&job); - break; + started = true; } - // Periodically update the spinner. - _ = interval.tick() => { - ctx.job_polling_update(&job_id); + _ = ticker.tick() => { + last = match job_status_map(ctx, client, &job_id).await { + Ok(status) => status, + // The job may not be visible anywhere yet. + Err(CommandError::NotFound) => JobStatusMap::new(), + Err(error) => { + ctx.job_watch_finished(&job_id); + return Err(error); + } + }; + ctx.job_watch_update(&last); + polls += 1; + if started && watch_done(polls, sleds, &last, &mut quiet) { + break last; + } + sleds = last.len(); } - // Stop the job on interrupt, but don't break out of - // the select loop; we must wait for the start future - // to resolve. But there is a race here with the start, - // so retry the stop a few times if needed. - _ = ctrl_c(), if !stopped => { + // 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. + _ = ctrl_c() => { + if started || stopped { + break last; + } for _ in 0..3 { match job_stop(ctx, client, &job_id).await { Ok(_) => { - ctx.job_polling_finished(&job_id); ctx.job_stopped(&job_id); stopped = true; break; @@ -1173,15 +1260,17 @@ async fn job_start( Err(CommandError::NotFound) => { sleep(JOB_STOP_RETRY_INTERVAL).await; } - Err(error) => return Err(error), + Err(error) => { + ctx.job_watch_finished(&job_id); + return Err(error); + } } } } } - } - - // Show the job status and output. - job_status(ctx, client, &job_id, StatusDisplayStyle::Short).await?; + }; + ctx.job_watch_finished(&job_id); + ctx.job_status(&job_id, &status, StatusDisplayStyle::Short); for stream in [Stdout, Stderr] { match with_login_via(ctx, client, Some(&target), async || { client @@ -1250,6 +1339,60 @@ async fn job_status_map( job_status_try_from_json_map(status).map_err(CommandError::BaseboardIdParseError) } +/// How often a watched job's status is refreshed. +const WATCH_INTERVAL: Duration = Duration::from_secs(1); + +/// Watch a job's status across the rack, one live line per sled, until +/// it settles or the user interrupts. Returns the last status map. +async fn job_watch( + ctx: &mut impl CommandContext, + client: &Client, + job_id: &JobId, +) -> Result { + ctx.job_watch_started(job_id); + let mut sleds = 0; + let mut quiet = 0; + let mut polls = 0; + let status = loop { + let status = match job_status_map(ctx, client, job_id).await { + Ok(status) => status, + Err(error) => { + ctx.job_watch_finished(job_id); + return Err(error); + } + }; + ctx.job_watch_update(&status); + polls += 1; + if watch_done(polls, sleds, &status, &mut quiet) { + break status; + } + sleds = status.len(); + select! { + _ = sleep(WATCH_INTERVAL) => {} + _ = ctrl_c() => break status, + } + }; + ctx.job_watch_finished(job_id); + Ok(status) +} + +/// The fewest polls a watch may run: gossip needs a few seconds to +/// fan a job out across the rack, so a map that looks settled early +/// is likely still missing sleds. +const WATCH_MIN_POLLS: usize = 5; + +/// A watched job has settled once the watch is old enough for gossip +/// to have named every sled, every known sled reports a terminal +/// status, and the set of sleds has been stable for a couple of polls. +fn watch_done(polls: usize, sleds: usize, status: &JobStatusMap, quiet: &mut usize) -> bool { + if !status.is_empty() && status.len() == sleds && status.values().all(JobStatus::is_terminal) { + *quiet += 1; + } else { + *quiet = 0; + } + polls >= WATCH_MIN_POLLS && *quiet >= 2 +} + /// Stream some bytes into a vector. async fn byte_stream_to_vec(mut stream: ByteStream) -> Result, CommandError> { let mut output = Vec::new(); diff --git a/client/src/context.rs b/client/src/context.rs index f3dea9f..b8d6bbe 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -8,7 +8,6 @@ use std::fmt; use std::path::Path; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::Duration; use clap::ValueEnum; use sled_hardware_types::BaseboardId; @@ -135,9 +134,9 @@ pub trait CommandContext: Clone + Send + Sync { ); fn job_output_update(&mut self, id: &JobId, stream: JobOutputStream, bytes: u64); fn job_output_finished(&mut self, id: &JobId, stream: JobOutputStream, stage: Option<&str>); - fn job_polling_started(&mut self, id: &JobId, duration: Duration); - fn job_polling_update(&mut self, id: &JobId); - fn job_polling_finished(&mut self, id: &JobId); + fn job_watch_started(&mut self, id: &JobId); + fn job_watch_update(&mut self, status: &JobStatusMap); + fn job_watch_finished(&mut self, id: &JobId); fn job_attached(&mut self, id: &JobId); fn job_detached(&mut self, id: &JobId); fn job_signing_started(&mut self, id: &JobId); diff --git a/client/src/repl.rs b/client/src/repl.rs index 81666d5..2e101ad 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -9,7 +9,6 @@ use std::env; use std::ffi::OsString; use std::path::Path; -use std::time::Duration; use clap::Parser; use rustyline::DefaultEditor; @@ -294,18 +293,6 @@ impl CommandContext for Repl { self.cli.job_output_finished(job_id, stream, stage); } - fn job_polling_started(&mut self, job_id: &JobId, duration: Duration) { - self.cli.job_polling_started(job_id, duration); - } - - fn job_polling_update(&mut self, job_id: &JobId) { - self.cli.job_polling_update(job_id); - } - - fn job_polling_finished(&mut self, job_id: &JobId) { - self.cli.job_polling_finished(job_id); - } - fn job_attached(&mut self, job_id: &JobId) { self.cli.job_attached(job_id); } @@ -330,6 +317,18 @@ impl CommandContext for Repl { self.cli.job_signed(job, show); } + fn job_watch_started(&mut self, job_id: &JobId) { + self.cli.job_watch_started(job_id) + } + + fn job_watch_update(&mut self, status: &JobStatusMap) { + self.cli.job_watch_update(status) + } + + fn job_watch_finished(&mut self, job_id: &JobId) { + self.cli.job_watch_finished(job_id) + } + fn job_status(&mut self, job_id: &JobId, status: &JobStatusMap, style: StatusDisplayStyle) { self.set_job_id(Some(job_id.to_owned())); self.cli.job_status(job_id, status, style); diff --git a/common/src/jobs.rs b/common/src/jobs.rs index 78ff42c..2cf079c 100644 --- a/common/src/jobs.rs +++ b/common/src/jobs.rs @@ -415,6 +415,14 @@ impl ExecutionError { } impl JobStatus { + /// Whether this status can ever change again. + pub fn is_terminal(&self) -> bool { + matches!( + self, + Self::Cancelled { .. } | Self::Error { .. } | Self::Stopped { .. } + ) + } + /// The most recent timestamp recorded for this status. pub fn time(&self) -> DateTime { match self { From c22f24e18ac9f8583921e51afc881e1ae985e331 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Wed, 12 Aug 2026 14:51:45 +0000 Subject: [PATCH 4/4] Freshen the README job example The --wait transcript shows the short per-sled status now, and the prose mentions the live watch and the --wait/--full status flags. --- README.md | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index f049c2b..4a69aba 100644 --- a/README.md +++ b/README.md @@ -48,23 +48,18 @@ 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 - Target: a part:0001 - Job status: Stopped - Started at: 2026-08-05 16:27:58.824606210 UTC - Stopped at: 2026-08-05 16:27:58.892796324 UTC (68ms 190us 114ns) - Exit status: 0 - Stdout len: 50 B - Stderr len: 0 B - Stdout hash: ae5a3df9c5c629590af52baceb12021525c0be16994738977b2ed049f4f374bb - Stderr hash: af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262 +✅ Job ID: 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: install-there-mutual-warfare-sound-live-order-man ``` The `--wait` (`-w`) flag tells `sush` to wait for the job to stop before -returning; the default behavior is to start the job and immediately return. -You can check the status of a (running) job with `job status`. +returning, showing a live status line per sled while it runs; the default +behavior is to start the job and immediately return. You can check the +status of a (running) job with `job status`, watch it the same way with +`job status --wait`, or see the complete job status with +`job status --full`. To run an interactive job with a pseudoterminal, you can use `job start --interactive` (`-i`):