diff --git a/api/src/lib.rs b/api/src/lib.rs index 8d9ca3e9..388ff6ba 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -4,7 +4,7 @@ //! Oxide Support Shell API -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::fmt; use dropshot::{ @@ -147,7 +147,7 @@ pub trait SushApi { params: PathParams, query: QueryParams, body: TypedBody, - ) -> Result; + ) -> Result, HttpError>; /// Stop a (running) job. #[endpoint { method = POST, path = "/jobs/{job_id}/stop" }] @@ -287,6 +287,13 @@ pub struct SessionStartBody { pub seq: SessionSeq, } +/// Acknowledgement that a job was accepted. +#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] +pub struct JobStarted { + /// Which sleds the job will run on. + pub sleds: BTreeSet, +} + /// Job parameters _not_ specified in the signed job request. #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(default)] diff --git a/client/src/cli.rs b/client/src/cli.rs index 46319a43..0148bc7f 100644 --- a/client/src/cli.rs +++ b/client/src/cli.rs @@ -4,7 +4,7 @@ //! Possibly-interactive command-line interface. -use std::collections::BTreeMap; +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}; @@ -35,7 +35,7 @@ use sush_common::jobs::{ SignedJob, job_status_to_json_map, }; use sush_common::keys::{KeyId, Signature, SshPublicKey}; -use sush_common::targets::{MAX_CUBBY, SledHealth, SledId, SledVersion}; +use sush_common::targets::{MAX_CUBBY, SledHealth, SledId, SledVersion, Target}; use sush_common::version::VersionInfo; use crate::AuthzSigner; @@ -481,7 +481,7 @@ impl CommandContext for Cli { // Job management - fn job_started(&mut self, job: &SignedJob, show: bool) { + fn job_started(&mut self, job: &SignedJob, sleds: Option<&BTreeSet>, show: bool) { if let Some(session) = self.session.lock().unwrap().as_mut() { session.job_started(job.to_owned()); self.save_session(Some(session)); @@ -491,12 +491,26 @@ impl CommandContext for Cli { } let job_id = job.job_id(); match self.get_output_format() { - OutputFormat::Json => println!("{}", json!({"job_started": job_id})), + OutputFormat::Json => println!("{}", json!({"job_started": job_id, "sleds": sleds})), OutputFormat::Text => { - if let Some(watch) = self.watch.lock().unwrap().as_ref() { - let _ = watch.multi.println(format!("✅ Started job `{job_id}`")); - } else { - println!("✅ Started job `{job_id}`"); + let watch = self.watch.lock().unwrap(); + let print = |line: String| match watch.as_ref() { + Some(watch) => { + let _ = watch.multi.println(line); + } + None => println!("{line}"), + }; + 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!( + "❗ Job `{job_id}` will not run on unreachable sled `{baseboard}`" + )); + } + } } } } diff --git a/client/src/commands.rs b/client/src/commands.rs index ef1ebe5b..f5522bb0 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -6,6 +6,7 @@ //! //! May be executed via either the main CLI or the interactive REPL. +use std::collections::BTreeSet; use std::fs::{File, OpenOptions, read}; use std::io::ErrorKind; use std::io::{Seek as _, SeekFrom, Write as _, stdin}; @@ -302,56 +303,49 @@ fn watch_settling() { time_started: Utc::now(), }; - let all = Target::All; let mut settling = Settling::default(); let mut status = JobStatusMap::new(); - assert!(!settling.done(&all, None, &status), "empty never settles"); + assert!(!settling.done(None, None, &status), "empty never settles"); status.insert(sled("A"), terminal.clone()); - assert!(!settling.done(&all, None, &status), "new sled resets"); - assert!(!settling.done(&all, None, &status), "first quiet poll"); + assert!(!settling.done(None, None, &status), "new sled resets"); + assert!(!settling.done(None, None, &status), "first quiet poll"); assert!( - !settling.done(&all, None, &status), + !settling.done(None, None, &status), "quiet but too young for gossip" ); - assert!(settling.done(&all, None, &status), "old enough, quiet"); + assert!(settling.done(None, None, &status), "old enough, quiet"); let mut settling = Settling::default(); status.insert(sled("B"), running.clone()); - assert!(!settling.done(&all, None, &status), "running sled holds"); + assert!(!settling.done(None, None, &status), "running sled holds"); status.insert(sled("B"), terminal.clone()); assert!( - !settling.done(&all, Some(2), &status), + !settling.done(None, Some(2), &status), "the rack count must hold" ); assert!( - settling.done(&all, Some(2), &status), + settling.done(None, Some(2), &status), "the rack count held twice" ); - let named: Target = format!("{},{}", sled("A"), sled("B")).parse().unwrap(); + let known = BTreeSet::from([sled("A"), sled("B")]); let mut settling = Settling::default(); assert!( - settling.done(&named, None, &status), - "named sleds settle on the first terminal poll" - ); - let duplicates: Target = format!("{},{}", sled("A"), sled("A")).parse().unwrap(); - let mut settling = Settling::default(); - assert!( - settling.done(&duplicates, None, &status), - "duplicate baseboards collapse" + settling.done(Some(&known), None, &status), + "known sleds settle on the first terminal poll" ); status.insert(sled("B"), running); assert!( - !settling.done(&named, None, &status), - "a running named sled holds" + !settling.done(Some(&known), None, &status), + "a running known sled holds" ); status.insert(sled("B"), terminal); - let missing: Target = format!("{},{}", sled("A"), sled("C")).parse().unwrap(); + let missing = BTreeSet::from([sled("A"), sled("C")]); let mut settling = Settling::default(); assert!( - !settling.done(&missing, None, &status), - "a missing named sled holds" + !settling.done(Some(&missing), None, &status), + "a missing known sled holds" ); } @@ -1376,7 +1370,7 @@ async fn job( job_start(ctx, client, job, start_args.to_owned()).await } else { ctx.job_signed(&job, true); - ctx.job_started(&job, false); + ctx.job_started(&job, None, false); Ok(()) } } @@ -1418,7 +1412,7 @@ async fn job( StatusDisplayStyle::Short }; if wait { - let status = job_watch(ctx, client, &job_id, &Target::All).await?; + let status = job_watch(ctx, client, &job_id, None).await?; ctx.job_status(&job_id, &status, style); Ok(()) } else { @@ -1535,8 +1529,8 @@ async fn job_start( // Start the job. if interactive { - start.await?; - ctx.job_started(&job, true); + 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? @@ -1557,10 +1551,10 @@ async fn job_start( let file = options .open(&path) .map_err(|error| CommandError::io(&path, error))?; - start.await?; - ctx.job_started(&job, true); + let sleds = start.await?.into_inner().sleds; + ctx.job_started(&job, Some(&sleds), true); let hasher = job_stream(ctx, client, &job_id, &target, &path, file).await?; - let status = job_watch(ctx, client, &job_id, &target.clone().into()).await?; + let status = job_watch(ctx, client, &job_id, Some(sleds)).await?; ctx.job_status(&job_id, &status, StatusDisplayStyle::Short); let Some(JobStatus::Stopped { result: Ok(0), @@ -1593,6 +1587,7 @@ async fn job_start( // 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); let mut last = JobStatusMap::new(); @@ -1606,7 +1601,12 @@ async fn job_start( // Wait for the start request to finish. start_result = &mut start, if !started => { match start_result { - Ok(_) | Err(CommandError::TimedOut) => ctx.job_started(&job, true), + Ok(response) => { + let started = response.into_inner().sleds; + ctx.job_started(&job, Some(&started), true); + sleds = Some(started); + } + Err(CommandError::TimedOut) => ctx.job_started(&job, None, true), Err(error) => { ctx.job_watch_finished(&job_id); return Err(error); @@ -1627,8 +1627,11 @@ async fn job_start( } }; ctx.job_watch_update(&last); - let rack = rack_sleds(client, &job_target).await; - if settling.done(&job_target, rack, &last) && started { + let rack = match sleds { + Some(_) => None, + None => rack_sleds(client).await, + }; + if settling.done(sleds.as_ref(), rack, &last) && started { break last; } if !stalled && last.is_empty() && settling.polls >= WATCH_STALL_POLLS { @@ -1708,8 +1711,8 @@ async fn job_start( } } } else { - start.await?; - ctx.job_started(&job, true); + let sleds = start.await?.into_inner().sleds; + ctx.job_started(&job, Some(&sleds), true); } Ok(()) } @@ -1918,17 +1921,22 @@ 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. +/// `sleds`, if known, are the sleds the job runs on. async fn job_watch( ctx: &mut impl CommandContext, client: &Client, job_id: &JobId, - target: &Target, + sleds: Option>, ) -> Result { ctx.job_watch_started(job_id); + let via = sleds + .as_ref() + .filter(|sleds| sleds.len() == 1) + .and_then(BTreeSet::first); let mut settling = Settling::default(); let mut sigint = signal(SignalKind::interrupt())?; let status = loop { - let status = match job_status_map(ctx, client, job_id, target.single_baseboard()).await { + let status = match job_status_map(ctx, client, job_id, via).await { Ok(status) => status, Err(error) => { ctx.job_watch_finished(job_id); @@ -1936,8 +1944,11 @@ async fn job_watch( } }; ctx.job_watch_update(&status); - let rack = rack_sleds(client, target).await; - if settling.done(target, rack, &status) { + let rack = match sleds { + Some(_) => None, + None => rack_sleds(client).await, + }; + if settling.done(sleds.as_ref(), rack, &status) { break status; } select! { @@ -1969,12 +1980,17 @@ struct Settling { impl Settling { /// A watched job has settled once every sled it runs on reports a - /// terminal status. A target naming only baseboards is settled as - /// soon as they all report. Against the rack inventory, the count + /// terminal status. When those `sleds` are known, the job is settled + /// as soon as they all report. Against the rack inventory, the count /// must match on consecutive polls. Either way, the watch is done /// once it is old enough for gossip to have named every sled and /// the set of sleds has been stable for a couple of polls. - fn done(&mut self, target: &Target, rack: Option, status: &JobStatusMap) -> bool { + fn done( + &mut self, + sleds: Option<&BTreeSet>, + rack: Option, + status: &JobStatusMap, + ) -> bool { self.polls += 1; let terminal = !status.is_empty() && status.values().all(JobStatus::is_terminal); if terminal && status.len() == self.sleds { @@ -1988,9 +2004,9 @@ impl Settling { } else { self.met = 0; } - if let Some(named) = target.named_baseboards() - && named - .into_iter() + if let Some(sleds) = sleds + && sleds + .iter() .all(|sled| status.get(sled).is_some_and(JobStatus::is_terminal)) { return true; @@ -2000,11 +2016,7 @@ impl Settling { } /// The number of sleds in the rack's own inventory, where available. -/// Skips the fetch for targets whose sleds are named. -async fn rack_sleds(client: &Client, target: &Target) -> Option { - if target.named_baseboards().is_some() { - return None; - } +async fn rack_sleds(client: &Client) -> Option { let sleds = client.versions().send().await.ok()?.into_inner().len(); (sleds > 0).then_some(sleds) } diff --git a/client/src/context.rs b/client/src/context.rs index 6260a614..88943642 100644 --- a/client/src/context.rs +++ b/client/src/context.rs @@ -4,6 +4,7 @@ //! Command context tracking. +use std::collections::BTreeSet; use std::fmt; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -149,7 +150,7 @@ pub trait CommandContext: Clone + Send + Sync { // Job management fn really_target(&mut self, sled: &SledId) -> Result<(), CommandError>; - fn job_started(&mut self, job: &SignedJob, show: bool); + 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; diff --git a/client/src/lib.rs b/client/src/lib.rs index 3b90ca87..0639f61b 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -75,6 +75,7 @@ progenitor::generate_api!( JobLimits = sush_common::jobs::JobLimits, JobOutputHash = sush_common::jobs::JobOutputHash, JobOutputStream = sush_common::jobs::JobOutputStream, + JobStarted = sush_api::JobStarted, JobStatus = sush_common::jobs::JobStatus, JobWait = sush_api::JobWait, KeyId = sush_common::keys::KeyId, diff --git a/client/src/repl.rs b/client/src/repl.rs index b3f4d8e0..81380a19 100644 --- a/client/src/repl.rs +++ b/client/src/repl.rs @@ -6,6 +6,7 @@ //! //! Inherits most of its behavior from the (non-interactive) CLI. +use std::collections::BTreeSet; use std::env; use std::ffi::OsString; use std::fs::{Permissions, set_permissions}; @@ -297,9 +298,9 @@ impl CommandContext for Repl { // Job management - fn job_started(&mut self, job: &SignedJob, show: bool) { + fn job_started(&mut self, job: &SignedJob, sleds: Option<&BTreeSet>, show: bool) { self.set_job_id(Some(job.job_id().to_owned())); - self.cli.job_started(job, show); + self.cli.job_started(job, sleds, show); } fn job_stopped(&mut self, job_id: &JobId) { diff --git a/common/src/jobs.rs b/common/src/jobs.rs index a8587506..c2eed34a 100644 --- a/common/src/jobs.rs +++ b/common/src/jobs.rs @@ -4,7 +4,7 @@ //! Signed job requests. -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::fmt; use std::io::Error as IoError; use std::ops::Deref; @@ -24,7 +24,7 @@ use crate::codephrases::Codephrase; use crate::hash::{Hash, Hasher, hash}; use crate::interactive::InteractiveJobError; use crate::keys::{KeyId, Signed, ToBeSigned, ToBeSignedFields, Verified}; -use crate::targets::{Cubbies, Target}; +use crate::targets::Target; codephrase_newtype! { /// A globally unique identifier for a job within a session. @@ -421,13 +421,13 @@ impl JobStartRequest { } } + /// Whether this job runs on `baseboard`. The signed target must match + /// `sleds`, so the sled that accepted the job cannot widen it. /// Interactive jobs run only on their single named baseboard. - pub fn runs_on(&self, baseboard: &BaseboardId, cubbies: &Cubbies) -> bool { - if self.mode.is_interactive() { - self.target.single_baseboard() == Some(baseboard) - } else { - self.target.includes(baseboard, cubbies) - } + pub fn runs_on(&self, baseboard: &BaseboardId, sleds: &BTreeSet) -> bool { + sleds.contains(baseboard) + && self.target.matches(sleds) + && (!self.mode.is_interactive() || self.target.single_baseboard() == Some(baseboard)) } pub fn job_id(&self) -> &JobId { @@ -1009,19 +1009,42 @@ mod test { serial_number: serial.to_string(), }; let me = sled("me"); - let cubbies = Cubbies::from([(14, me.clone())]); + let sleds = BTreeSet::from([me.clone()]); let job = |mode, target| { JobStartRequest::new(JobId::random(), SessionId::random(), "true", mode, target) }; use JobMode::*; let just_me = Target::Sleds(vec![SledId::Baseboard(me.clone())]); - assert!(job(Interactive, just_me.clone()).runs_on(&me, &cubbies)); - assert!(job(Batch, just_me).runs_on(&me, &cubbies)); - assert!(!job(Interactive, Target::All).runs_on(&me, &cubbies)); - assert!(job(Batch, Target::All).runs_on(&me, &cubbies)); + assert!(job(Interactive, just_me.clone()).runs_on(&me, &sleds)); + assert!(job(Batch, just_me).runs_on(&me, &sleds)); + assert!(!job(Interactive, Target::All).runs_on(&me, &sleds)); + assert!(job(Batch, Target::All).runs_on(&me, &sleds)); let my_cubby = Target::Sleds(vec![SledId::Cubby(14)]); - assert!(!job(Interactive, my_cubby.clone()).runs_on(&me, &cubbies)); - assert!(job(Batch, my_cubby).runs_on(&me, &cubbies)); + assert!(!job(Interactive, my_cubby.clone()).runs_on(&me, &sleds)); + assert!(job(Batch, my_cubby).runs_on(&me, &sleds)); + } + + /// A job runs only on the sleds its target resolved to, and only + /// if the signed target matches them. + #[test] + fn resolved_targets() { + let sled = |serial: &str| BaseboardId { + part_number: "913".to_string(), + serial_number: serial.to_string(), + }; + let (me, you) = (sled("me"), sled("you")); + let job = |target| { + JobStartRequest::new( + JobId::random(), + SessionId::random(), + "true", + JobMode::Batch, + target, + ) + }; + assert!(!job(Target::All).runs_on(&me, &BTreeSet::from([you.clone()]))); + let just_you = Target::from(you.clone()); + assert!(!job(just_you).runs_on(&me, &BTreeSet::from([me.clone(), you]))); } /// Requested limits may narrow the default ceiling, never widen it. diff --git a/common/src/targets.rs b/common/src/targets.rs index d1eb8e02..d51233f1 100644 --- a/common/src/targets.rs +++ b/common/src/targets.rs @@ -6,9 +6,9 @@ //! //! One grammar serves signed job requests, the client, and the proxy: //! `*` means every sled in the rack, and a comma-separated list names -//! sleds by baseboard ID or cubby number. Cubby numbers resolve -//! against a mapping the rack learns at runtime, so a target is -//! evaluated lazily, against what is known when it is asked. +//! sleds by baseboard ID or cubby number. The sled that accepts a job +//! resolves its target once, to the baseboards it can reach, and every +//! sled runs the job against that resolution. use std::collections::{BTreeMap, BTreeSet}; use std::fmt; @@ -74,16 +74,52 @@ pub enum SledId { #[error("invalid target `{0}`")] pub struct InvalidTarget(String); +/// A target names a cubby that no known sled occupies. +#[derive(Debug, Error)] +#[error("no known sled occupies cubby {0}")] +pub struct UnknownCubby(pub u8); + impl Target { - /// Whether `baseboard` is targeted, as far as `cubbies` can say. - /// A cubby the mapping does not know targets nothing. - pub fn includes(&self, baseboard: &BaseboardId, cubbies: &Cubbies) -> bool { + /// The sleds in `reachable` that this target names, with cubbies + /// looked up in `cubbies`. + pub fn resolve( + &self, + reachable: &BTreeSet, + cubbies: &Cubbies, + ) -> Result, UnknownCubby> { + match self { + Self::All => Ok(reachable.clone()), + Self::Sleds(sleds) => { + let mut named = BTreeSet::new(); + for sled in sleds { + named.insert(match sled { + SledId::Baseboard(baseboard) => baseboard.clone(), + SledId::Cubby(cubby) => { + cubbies.get(cubby).ok_or(UnknownCubby(*cubby))?.clone() + } + }); + } + Ok(named.intersection(reachable).cloned().collect()) + } + } + } + + /// Whether every sled in `sleds` is one this target names. Each cubby + /// matches any one sled. + pub fn matches(&self, sleds: &BTreeSet) -> bool { match self { Self::All => true, - Self::Sleds(sleds) => sleds.iter().any(|sled| match sled { - SledId::Baseboard(named) => named == baseboard, - SledId::Cubby(cubby) => cubbies.get(cubby) == Some(baseboard), - }), + Self::Sleds(named) => { + let mut baseboards = BTreeSet::new(); + let mut cubbies = BTreeSet::new(); + for sled in named { + match sled { + SledId::Baseboard(baseboard) => baseboards.insert(baseboard.clone()), + SledId::Cubby(cubby) => cubbies.insert(*cubby), + }; + } + sleds.difference(&baseboards).count() <= cubbies.len() + } } } @@ -101,20 +137,6 @@ impl Target { }, } } - - /// Every sled this target names, when it names only baseboards. - pub fn named_baseboards(&self) -> Option> { - match self { - Self::All => None, - Self::Sleds(sleds) => sleds - .iter() - .map(|sled| match sled { - SledId::Baseboard(baseboard) => Some(baseboard), - SledId::Cubby(_) => None, - }) - .collect(), - } - } } impl From for Target { @@ -248,23 +270,48 @@ mod test { } #[test] - fn targets_include() { + fn targets_resolve() { let brm31 = baseboard("913-0000019:BRM42220031"); let brm40 = baseboard("913-0000019:BRM42220040"); - let cubbies = Cubbies::from([(14, brm31.clone())]); - - let all = Target::All; - assert!(all.includes(&brm31, &cubbies)); - assert!(all.includes(&brm40, &cubbies)); - - let by_baseboard: Target = "913-0000019:BRM42220031".parse().unwrap(); - assert!(by_baseboard.includes(&brm31, &cubbies)); - assert!(!by_baseboard.includes(&brm40, &cubbies)); + let brm41 = baseboard("913-0000019:BRM42220041"); + let reachable = BTreeSet::from([brm31.clone(), brm40.clone()]); + let cubbies = Cubbies::from([(14, brm31.clone()), (16, brm41.clone())]); + let resolve = |target: &str| { + target + .parse::() + .unwrap() + .resolve(&reachable, &cubbies) + }; + + assert_eq!(resolve("*").unwrap(), reachable); + assert_eq!( + resolve("913-0000019:BRM42220031,913-0000019:BRM42220041").unwrap(), + BTreeSet::from([brm31.clone()]), + ); + assert_eq!(resolve("14").unwrap(), BTreeSet::from([brm31])); + // Cubby 16 holds a sled we cannot reach. + assert!(resolve("16").unwrap().is_empty()); + assert!(matches!(resolve("14,17"), Err(UnknownCubby(17)))); + } - let by_cubby: Target = "14,16".parse().unwrap(); - assert!(by_cubby.includes(&brm31, &cubbies)); - // Cubby 16 is not in the mapping, so it targets nothing. - assert!(!by_cubby.includes(&brm40, &cubbies)); - assert!(!by_cubby.includes(&brm40, &Cubbies::new())); + #[test] + fn targets_match() { + let brm31 = baseboard("913-0000019:BRM42220031"); + let brm40 = baseboard("913-0000019:BRM42220040"); + let both = BTreeSet::from([brm31.clone(), brm40.clone()]); + let matches = |target: &str, sleds: &BTreeSet| { + target.parse::().unwrap().matches(sleds) + }; + + assert!(matches("*", &both)); + assert!(matches( + "913-0000019:BRM42220031", + &BTreeSet::from([brm31.clone()]) + )); + assert!(matches("913-0000019:BRM42220031", &BTreeSet::new())); + assert!(!matches("913-0000019:BRM42220031", &both)); + assert!(matches("14", &BTreeSet::from([brm40.clone()]))); + assert!(!matches("14", &both)); + assert!(matches("913-0000019:BRM42220031,14", &both)); } } diff --git a/server/src/error.rs b/server/src/error.rs index ba16f63e..fd08e80e 100644 --- a/server/src/error.rs +++ b/server/src/error.rs @@ -68,6 +68,8 @@ pub enum JobError { Key(#[from] KeyError), #[error("Only one session may be running at a time")] MultipleSessions, + #[error("No sled this job targets is reachable")] + NoReachableTarget, #[error("No current session")] NoSession, #[error("Job `{0}` is not the session's next job")] @@ -96,6 +98,8 @@ pub enum JobError { TooManyOutputRequests(usize), #[error("Unauthorized request")] Unauthorized(Nonce), + #[error("Which sled is in cubby {0} has not yet been reported")] + UnknownCubby(u8), #[error("Unable to wait for job end")] Wait, #[error("The provided session ID is invalid")] @@ -199,7 +203,7 @@ impl From for HttpError { HttpError::for_client_error(None, ClientErrorStatusCode::NOT_FOUND, message) } DuplicateJobId(_) | InteractiveTarget | InvalidCommand | Json(_) | MultipleSessions - | InvalidSessionId | UnverifiedJob(_) => { + | InvalidSessionId | NoReachableTarget | UnknownCubby(_) | UnverifiedJob(_) => { HttpError::for_client_error(None, ClientErrorStatusCode::BAD_REQUEST, message) } NotNextJob(_) => { diff --git a/server/src/gossip.rs b/server/src/gossip.rs index 3a72e902..f992f105 100644 --- a/server/src/gossip.rs +++ b/server/src/gossip.rs @@ -21,6 +21,7 @@ use std::cmp::Ordering; use std::collections::{BTreeSet, HashMap, HashSet}; +use std::future::pending; use std::io; use std::net::{SocketAddr, SocketAddrV6}; use std::time::Duration; @@ -59,6 +60,18 @@ impl LinkedBaseboards { pub fn borrow(&self) -> watch::Ref<'_, BTreeSet> { self.0.borrow() } + + pub fn borrow_and_update(&mut self) -> watch::Ref<'_, BTreeSet> { + self.0.borrow_and_update() + } + + /// Wait for the set to change. Once the gossip manager is gone, the + /// set can no longer change, so this never returns. + pub async fn changed(&mut self) { + if self.0.changed().await.is_err() { + pending().await + } + } } /// Manager timing. The defaults suit a rack, tests shrink them. diff --git a/server/src/manager.rs b/server/src/manager.rs index 2b58682f..9bfb4aa0 100644 --- a/server/src/manager.rs +++ b/server/src/manager.rs @@ -17,6 +17,7 @@ use lru::LruCache; use sled_hardware_types::BaseboardId; use slog::{Logger, debug, info, o, warn}; use tokio::fs::read; +use tokio::select; use tokio::sync::{Mutex, mpsc, watch}; use tokio::task::JoinHandle; use tokio::time::timeout; @@ -36,7 +37,7 @@ use sush_common::jobs::{ SignedJob, }; use sush_common::keys::{KeyError, KeyId, SshPublicKey}; -use sush_common::targets::{Cubbies, SledHealth, SledVersion}; +use sush_common::targets::{Cubbies, SledHealth, SledVersion, Target, UnknownCubby}; use sush_common::version::LONG_VERSION; use crate::error::JobError; @@ -399,18 +400,39 @@ impl JobManager { self.join_state.take() } - fn wait_for_job(&self, job_id: &JobId, wait: JobWait) -> impl FnMut(&State) -> bool { - move |state| { - if wait.is_some() { - state - .get_job_status(job_id) - .and_then(|map| map.get(self.own_baseboard())) - .map(|status| wait.matches_status(status)) - .unwrap_or(false) - } else { - true - } + /// Wait until the job's status matches `wait` on each of `sleds` + /// that this sled still links to. + async fn wait_for_job( + &self, + job_id: &JobId, + wait: JobWait, + sleds: BTreeSet, + ) -> Result<(), JobError> { + if !wait.is_some() { + return Ok(()); } + let settled = |state: &State, linked: &BTreeSet| { + let statuses = state.get_job_status(job_id); + sleds.iter().all(|sled| { + let gone = *sled != self.own_baseboard && !linked.contains(sled); + gone || statuses + .and_then(|map| map.get(sled)) + .is_some_and(|status| wait.matches_status(status)) + }) + }; + let mut state = self.state.clone(); + let mut linked = self.linked.clone(); + timeout(WAIT_TIMEOUT, async { + while !settled(&state.borrow_and_update(), &linked.borrow_and_update()) { + select! { + changed = state.changed() => changed.map_err(JobError::closed)?, + () = linked.changed() => (), + } + } + Ok(()) + }) + .await + .map_err(|_| JobError::Timeout)? } async fn wait_for(&self, predicate: impl FnMut(&State) -> bool) -> Result<(), JobError> { @@ -597,12 +619,15 @@ impl JobManager { // Job management. + /// Start `job` on the sleds its target names that this sled can + /// reach, wait until the job's status matches `params.wait` on all + /// of them, and return them. pub async fn job_start( &self, authn: &Identity, job: SignedJob, params: JobStartParams, - ) -> Result<(), JobError> { + ) -> Result, JobError> { // A forged job must not take a signed job's place in the chain. self.state .borrow() @@ -628,10 +653,26 @@ impl JobManager { } // Submit the job for execution. + let sleds = self.reachable_targets(payload.target())?; let wait = params.wait.to_owned(); - self.job_request(authn, JobRequest::Start(job, params)) + self.job_request(authn, JobRequest::Start(job, params, sleds.clone())) .await?; - self.wait_for(self.wait_for_job(&job_id, wait)).await + self.wait_for_job(&job_id, wait, sleds.clone()).await?; + Ok(sleds) + } + + /// The sleds `target` names that this sled can reach: itself and + /// the sleds it holds live gossip links to. + fn reachable_targets(&self, target: &Target) -> Result, JobError> { + let mut reachable = self.linked.borrow().clone(); + reachable.insert(self.own_baseboard.clone()); + let sleds = target + .resolve(&reachable, self.state.borrow().rack().cubbies()) + .map_err(|UnknownCubby(cubby)| JobError::UnknownCubby(cubby))?; + if sleds.is_empty() { + return Err(JobError::NoReachableTarget); + } + Ok(sleds) } pub async fn job_stop( @@ -644,7 +685,13 @@ impl JobManager { self.known_job_check(job_id)?; self.job_request(authn, JobRequest::Stop(job_id.to_owned())) .await?; - self.wait_for(self.wait_for_job(job_id, wait)).await + let sleds = self + .state + .borrow() + .get_job_status(job_id) + .map(|statuses| statuses.keys().cloned().collect()) + .unwrap_or_default(); + self.wait_for_job(job_id, wait, sleds).await } pub async fn job_status( diff --git a/server/src/messages.rs b/server/src/messages.rs index d29b8664..b7d84a70 100644 --- a/server/src/messages.rs +++ b/server/src/messages.rs @@ -4,6 +4,8 @@ //! Messages gossiped via rumors. +use std::collections::BTreeSet; + use chrono::{DateTime, Utc}; use ciborium::{Value, de::from_reader as from_cbor, ser::into_writer as into_cbor}; use serde::{Deserialize, Serialize}; @@ -223,7 +225,7 @@ pub mod v0 { #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[allow(clippy::large_enum_variant)] pub enum JobRequest { - Start(SignedJob, JobStartParams), + Start(SignedJob, JobStartParams, BTreeSet), Stop(JobId), } @@ -333,7 +335,7 @@ mod wire_format { SessionRequest::DenyAttach(..) => "session-deny-attach-request", }, Message::Request(Request::Job(r)) => match &r.request { - JobRequest::Start(_, params) if params.term.is_some() => { + JobRequest::Start(_, params, _) if params.term.is_some() => { "job-start-interactive-request" } JobRequest::Start(..) => "job-start-request", @@ -375,6 +377,10 @@ mod wire_format { part_number: "913-0000019".to_string(), serial_number: "BRM42220030".to_string(), }; + let other = BaseboardId { + part_number: "913-0000019".to_string(), + serial_number: "BRM42220031".to_string(), + }; let signature = |flags, counter| EncodedSignature { r: "abandon".parse().unwrap(), s: "zoo".parse().unwrap(), @@ -457,11 +463,19 @@ mod wire_format { )), Message::Request(Request::job( actor(), - JobRequest::Start(batch, JobStartParams::default()), + JobRequest::Start( + batch, + JobStartParams::default(), + BTreeSet::from([baseboard.clone(), other.clone()]), + ), )), Message::Request(Request::job( actor(), - JobRequest::Start(interactive, interactive_params), + JobRequest::Start( + interactive, + interactive_params, + BTreeSet::from([baseboard.clone()]), + ), )), Message::Request(Request::job(actor(), JobRequest::Stop(long_job))), Message::Request(Request::identity( diff --git a/server/src/rack.rs b/server/src/rack.rs index 6ec50cf8..9ed163b6 100644 --- a/server/src/rack.rs +++ b/server/src/rack.rs @@ -37,8 +37,8 @@ impl Rack { } } - pub fn runs_here(&self, job: &SignedJob) -> bool { - job.runs_on(&self.own_baseboard, &self.cubbies) + pub fn runs_here(&self, job: &SignedJob, sleds: &BTreeSet) -> bool { + job.runs_on(&self.own_baseboard, sleds) } pub fn own_baseboard(&self) -> &BaseboardId { diff --git a/server/src/server.rs b/server/src/server.rs index 3b0a2e49..c1fdee60 100644 --- a/server/src/server.rs +++ b/server/src/server.rs @@ -23,8 +23,8 @@ use tokio_tungstenite::tungstenite::protocol::{Role, WebSocketConfig}; use sush_api::{ AccessParam, Authorization, AuthorizedRangeRequest, JobHistoryParams, JobIdParam, - JobOutputParams, JobStartParams, JobStopParams, JobTargetParams, KeyIdParam, RoutingParam, - SessionAndJobIds, SessionAndKeyIds, SessionIdParam, SessionStartBody, SushApi, + JobOutputParams, JobStartParams, JobStarted, JobStopParams, JobTargetParams, KeyIdParam, + RoutingParam, SessionAndJobIds, SessionAndKeyIds, SessionIdParam, SessionStartBody, SushApi, }; use sush_common::authn::Identity; use sush_common::jobs::{ @@ -224,7 +224,7 @@ impl SushApi for ApiServer { params: PathParams, query: QueryParams, body: TypedBody, - ) -> Result { + ) -> Result, HttpError> { let mgr = ctx.context(); let Authorization { authorization } = headers.into_inner(); let authn = mgr @@ -239,8 +239,8 @@ impl SushApi for ApiServer { String::from("Path parameter job ID does not match body"), )); } - mgr.job_start(&authn, job, query.into_inner()).await?; - Ok(HttpResponseUpdatedNoContent()) + let sleds = mgr.job_start(&authn, job, query.into_inner()).await?; + Ok(HttpResponseOk(JobStarted { sleds })) } async fn job_stop( diff --git a/server/src/state.rs b/server/src/state.rs index 8e6d7f6c..c174dd08 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -64,6 +64,7 @@ pub const MAX_QUEUED_JOBS: usize = 1_000; pub struct QueuedJob { pub job: VerifiedJob, pub params: JobStartParams, + pub sleds: BTreeSet, pub replayed: bool, } @@ -403,7 +404,7 @@ impl State { fn skip_queued_jobs(&self, now: DateTime) -> Vec { let mut outputs = Vec::new(); for (job_id, queued) in self.session.queued_jobs().into_iter().flatten() { - if !queued.replayed && self.rack.runs_here(&queued.job) { + if !queued.replayed && self.rack.runs_here(&queued.job, &queued.sleds) { outputs.push(Output::job_event( self.rack.own_baseboard(), JobEvent::Skipped(*job_id, now, SkipReason::SessionEnded), @@ -803,7 +804,7 @@ impl State { ) -> Vec { let mut outputs = Vec::new(); match request { - JobRequest::Start(signed, params) => { + JobRequest::Start(signed, params, sleds) => { let job = match self.certs.verify(signed) { Ok(job) => job, Err(error) => { @@ -831,7 +832,13 @@ impl State { .session() .is_some_and(|s| s.session_id() == session_id) { - self.enqueue_job(log, now, job, params.clone(), actor, !live); + let queued = QueuedJob { + job, + params: params.clone(), + sleds: sleds.clone(), + replayed: !live, + }; + self.enqueue_job(log, now, queued, actor); outputs.extend(self.release_ready_jobs(log, now, network)); } else { let job_id = *signed.job_id(); @@ -841,7 +848,8 @@ impl State { "session_id" => %session_id, "actor" => %actor, ); - if live && self.rack.runs_here(signed) && !self.history.contains(&job_id) { + if live && self.rack.runs_here(signed, sleds) && !self.history.contains(&job_id) + { outputs.push(Output::job_event( self.rack.own_baseboard(), JobEvent::Skipped(job_id, now, SkipReason::SessionEnded), @@ -1030,20 +1038,13 @@ impl State { outputs } - fn enqueue_job( - &mut self, - log: &Logger, - now: DateTime, - job: VerifiedJob, - params: JobStartParams, - actor: &KeyId, - replayed: bool, - ) { + fn enqueue_job(&mut self, log: &Logger, now: DateTime, queued: QueuedJob, actor: &KeyId) { let SessionState::Active { queued_jobs, .. } = &mut self.session else { return; }; - let job_id = *job.job_id(); - let targeted = self.rack.runs_here(&job); + let job_id = *queued.job.job_id(); + let replayed = queued.replayed; + let targeted = self.rack.runs_here(&queued.job, &queued.sleds); // Adjudication can give the boundary job a status before its // request replays. The queue releases jobs in order, so @@ -1064,14 +1065,7 @@ impl State { // queue to keep the causal chain whole, but only jobs // targeting this sled that can actually run here record // a local status. - queued_jobs.insert( - job_id, - QueuedJob { - job, - params, - replayed, - }, - ); + queued_jobs.insert(job_id, queued); if targeted && !replayed { self.history.set_job_status( &job_id, @@ -1093,7 +1087,7 @@ impl State { return; }; let cancel_here = match queued_jobs.remove(job_id) { - Some(queued) => self.rack.runs_here(&queued.job), + Some(queued) => self.rack.runs_here(&queued.job, &queued.sleds), None => !self.history.contains(job_id), }; if !replayed && cancel_here { @@ -1161,10 +1155,11 @@ impl State { let QueuedJob { job: request, params, + sleds, replayed, } = queued_jobs.remove(&next_id).expect("checked above"); let job_id = request.payload().job_id().to_owned(); - if self.rack.runs_here(&request) { + if self.rack.runs_here(&request, &sleds) { if matches!(admission, Admission::Refuse) { if !replayed { warn!(log, "skipping a superseded session's job"; "job_id" => %job_id); @@ -1693,6 +1688,15 @@ mod test { .unwrap() } + /// A request to start `signed` on this test's sled. + fn start(signed: SignedJob) -> JobRequest { + JobRequest::Start( + signed, + JobStartParams::default(), + BTreeSet::from([baseboard()]), + ) + } + fn discard() -> Logger { Logger::root(slog::Discard, o!()) } @@ -1983,10 +1987,7 @@ mod test { let job_id = session.next_job_id(); let signed = signed_job(&mut root, job_id, session_id, "true").await; - let outputs = sim.request(Request::job( - actor.clone(), - JobRequest::Start(signed, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor.clone(), start(signed))); assert!( outputs .iter() @@ -2050,10 +2051,7 @@ mod test { let job_id = session.next_job_id(); let signed = signed_job(&mut root, job_id, session_id, "true").await; - sim.request(Request::job( - resubmitter, - JobRequest::Start(signed, JobStartParams::default()), - )); + sim.request(Request::job(resubmitter, start(signed))); assert_eq!( sim.state.session.session().and_then(Session::started_by), Some(&starter) @@ -2076,19 +2074,13 @@ mod test { signed.key_id().clone(), other.signature().clone(), ); - let outputs = sim.request(Request::job( - actor.clone(), - JobRequest::Start(forged, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor.clone(), start(forged))); assert!(outputs.is_empty()); assert!(sim.state.session.session().is_none()); assert!(sim.state.pending_rank(session_id).is_some()); assert!(!sim.state.history.contains(&job_id)); - let outputs = sim.request(Request::job( - actor, - JobRequest::Start(signed, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor, start(signed))); assert!( outputs .iter() @@ -2108,10 +2100,7 @@ mod test { assert!(!sim.state.history.contains(&job_id)); let signed = signed_job(&mut root, job_id, session_id, "true").await; - let outputs = sim.request(Request::job( - actor, - JobRequest::Start(signed, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor, start(signed))); assert!( outputs .iter() @@ -2141,7 +2130,11 @@ mod test { .unwrap(); sim.request(Request::job( actor.clone(), - JobRequest::Start(signed, JobStartParams::default()), + JobRequest::Start( + signed, + JobStartParams::default(), + BTreeSet::from([other.clone()]), + ), )); sim.send(Message::Event( other.clone(), @@ -2165,10 +2158,7 @@ mod test { let actor = root.key_id().clone(); let job_id = session.next_job_id(); let signed = signed_job(&mut root, job_id, session_id, "true").await; - sim.request(Request::job( - actor, - JobRequest::Start(signed, JobStartParams::default()), - )); + sim.request(Request::job(actor, start(signed))); let guest = self::root().key_id().clone(); let outputs = sim.request(Request::job(guest, JobRequest::Stop(job_id))); @@ -2219,10 +2209,7 @@ mod test { // If the committed job arrives again, it confirms the start but // skips as superseded instead of running. - let outputs = sim.request(Request::job( - actor.clone(), - JobRequest::Start(job_a, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor.clone(), start(job_a))); assert!( sim.state .session @@ -2240,10 +2227,7 @@ mod test { // That job's successor also does not run. let job_b = signed_job(&mut root, successor_id, session_id, "true").await; - let outputs = sim.request(Request::job( - actor.clone(), - JobRequest::Start(job_b, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor.clone(), start(job_b))); assert!(!outputs.iter().any(|o| matches!(o, Output::Run(_)))); // A newer session outranks the committed one and runs. @@ -2252,10 +2236,7 @@ mod test { sim.request(Request::session(actor.clone(), SessionRequest::Start(next))); let job_c_id = Session::new(next_id).next_job_id(); let job_c = signed_job(&mut root, job_c_id, next_id, "true").await; - let outputs = sim.request(Request::job( - actor, - JobRequest::Start(job_c, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor, start(job_c))); assert!(outputs.iter().any(|o| matches!(o, Output::Run(_)))); } @@ -2292,20 +2273,14 @@ mod test { let job_id = session.next_job_id(); let signed = signed_job(&mut root, job_id, session_id, "true").await; - let outputs = sim.request(Request::job( - actor.clone(), - JobRequest::Start(signed.clone(), JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor.clone(), start(signed.clone()))); assert!(!outputs.iter().any(|o| matches!(o, Output::Run(_)))); assert!(skipped(&outputs, job_id)); session.job_started(signed); let next_id = session.next_job_id(); let next = signed_job(&mut root, next_id, session_id, "true").await; - let outputs = sim.request(Request::job( - actor, - JobRequest::Start(next, JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor, start(next))); assert!(skipped(&outputs, next_id)); } @@ -2319,10 +2294,7 @@ mod test { let job_id = session.next_job_id(); let signed = signed_job(&mut root, job_id, session_id, "true").await; - sim.send(Message::Request(Request::job( - actor, - JobRequest::Start(signed.clone(), JobStartParams::default()), - ))); + sim.send(Message::Request(Request::job(actor, start(signed.clone())))); let outputs = sim.pump_draining(); assert!(!outputs.iter().any(|o| matches!(o, Output::Run(_)))); @@ -2340,10 +2312,7 @@ mod test { let job_id = session.next_job_id(); let signed = signed_job(&mut root, job_id, session_id, "-rf").await; - let outputs = sim.request(Request::job( - actor, - JobRequest::Start(signed.clone(), JobStartParams::default()), - )); + let outputs = sim.request(Request::job(actor, start(signed.clone()))); assert!(!outputs.iter().any(|o| matches!(o, Output::Run(_)))); assert!(outputs.iter().any(|o| matches!( o, @@ -2376,10 +2345,7 @@ mod test { actor.clone(), SessionRequest::Start(preimage), )), - Message::Request(Request::job( - actor, - JobRequest::Start(signed, JobStartParams::default()), - )), + Message::Request(Request::job(actor, start(signed))), Message::Event(baseboard(), Event::Job(JobEvent::Start(job_id, Utc::now()))), ]; diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index b8ee04e9..c4231082 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -18,11 +18,11 @@ use sled_hardware_types::BaseboardId; use slog::{Logger, o}; use tempfile::TempDir; use tokio::sync::watch; -use tokio::time::sleep; +use tokio::time::{sleep, timeout}; use tokio_util::sync::CancellationToken; use chrono::Utc; -use sush_api::{JobStartParams, JobWait}; +use sush_api::{JobStartParams, JobStopParams, JobWait}; use sush_common::authn::Identity; use sush_common::jobs::{ JobId, JobMode, JobOutputState, JobStartRequest, JobStatus, ProcessError, Session, @@ -52,6 +52,7 @@ struct Sled { peers: watch::Sender>, addr: SocketAddrV6, baseboard: BaseboardId, + cubbies: watch::Sender, _output: TempDir, } @@ -93,13 +94,13 @@ impl Sled { // The manager's baseboard must be the one sprockets attests, // as it is on a sled, or the health join can never match. let baseboard = baseboard(identity); - let (_cubbies, cubbies) = watch::channel(Cubbies::new()); + let (cubbies, cubbies_rx) = watch::channel(Cubbies::new()); let mgr = JobManager::new( log.clone(), PathIsolation::InsecureDisable, JobOutputDir::fixed(output.path()), baseboard.clone(), - cubbies, + cubbies_rx, universe.clone(), linked, &locker, @@ -114,6 +115,7 @@ impl Sled { peers, addr, baseboard, + cubbies, _output: output, } } @@ -175,11 +177,12 @@ async fn jobs_gossip_between_sleds() { .await .unwrap(); - // That job, submitted to A, runs on both sleds, and each sled - // learns the other's result. + // That job, submitted to A, runs on both sleds. A waits for both + // results before it returns, and B learns A's result too. let job_id = session.next_job_id(); let job = sign_job(&mut root, job_id, session_id, "true").await; - a.mgr + let sleds = a + .mgr .job_start( &authn_a, job, @@ -190,6 +193,10 @@ async fn jobs_gossip_between_sleds() { ) .await .unwrap(); + assert_eq!( + sleds, + BTreeSet::from([a.baseboard.clone(), b.baseboard.clone()]) + ); eventually("session gossips to B", 60, async || { b.mgr .session(&authn_b) @@ -202,14 +209,11 @@ async fn jobs_gossip_between_sleds() { .is_some_and(|s| matches!(s, JobStatus::Stopped { .. })) }) }; + assert!(stopped(&a.mgr, &authn_a, &b.baseboard).await); eventually("A's run visible on B", 60, async || { stopped(&b.mgr, &authn_b, &a.baseboard).await }) .await; - eventually("B's run visible on A", 60, async || { - stopped(&a.mgr, &authn_a, &b.baseboard).await - }) - .await; // A session started on B supersedes A's everywhere. let Ceremony { @@ -295,6 +299,157 @@ async fn first_job_confirms_start_on_another_sled() { shutdown.cancel(); } +/// A job that names a cubby runs on the sled in that cubby even while +/// that sled's own cubby map is empty, because the sled that accepts +/// the job resolves the cubby. +#[named] +#[tokio::test] +async fn cubby_resolves_on_accepting_sled() { + let (_tmp, dir) = pki("sush-distributed-", 2); + let mut root = common::ephemeral_root(); + let root_pem = dir.join("job-root.pem"); + write( + &root_pem, + pem_cert_chain(vec![root.cert().to_owned()]).unwrap(), + ) + .unwrap(); + + let log = test_logger(function_name!()); + let shutdown = CancellationToken::new(); + let a = Sled::start(&log, &dir, 1, &root_pem, &shutdown).await; + let b = Sled::start(&log, &dir, 2, &root_pem, &shutdown).await; + a.peers.send(BTreeSet::from([b.addr])).unwrap(); + b.peers.send(BTreeSet::from([a.addr])).unwrap(); + eventually("universe convergence", 120, async || { + a.universe.borrow().rumors.network() == b.universe.borrow().rumors.network() + }) + .await; + sees(&a, &b).await; + + // Only A knows that B is in cubby 14. + a.cubbies + .send(Cubbies::from([(14, b.baseboard.clone())])) + .unwrap(); + + let authn = fake_identity(&mut root).await; + let Ceremony { + session_id, + client_nonce, + signer_nonce, + seq, + } = ceremony(&a.mgr, 1); + a.mgr + .session_start(&authn, session_id, client_nonce, signer_nonce, seq) + .await + .unwrap(); + let job_id = Session::new(session_id).next_job_id(); + let job = root + .sign(JobStartRequest::new( + job_id, + session_id, + "true", + JobMode::Batch, + "14".parse().unwrap(), + )) + .await + .unwrap(); + + // A's map update lands whenever its state task gets to it, so + // retry the job until A accepts it. + let wait = JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }; + let sleds = loop { + match a.mgr.job_start(&authn, job.clone(), wait.clone()).await { + Ok(sleds) => break sleds, + Err(JobError::UnknownCubby(14)) => sleep(Duration::from_millis(50)).await, + Err(error) => panic!("job start failed: {error}"), + } + }; + assert_eq!(sleds, BTreeSet::from([b.baseboard.clone()])); + let status = a.mgr.job_status(&authn, &job_id).await.unwrap(); + assert!(matches!( + status.get(&b.baseboard), + Some(JobStatus::Stopped { .. }) + )); + assert!(!status.contains_key(&a.baseboard)); + + shutdown.cancel(); +} + +/// A stop waits for the sleds the job runs on, even when the sled that +/// takes the stop is not one of them. +#[named] +#[tokio::test] +async fn stop_waits_for_remote_sleds() { + let (_tmp, dir) = pki("sush-distributed-", 2); + let mut root = common::ephemeral_root(); + let root_pem = dir.join("job-root.pem"); + write( + &root_pem, + pem_cert_chain(vec![root.cert().to_owned()]).unwrap(), + ) + .unwrap(); + + let log = test_logger(function_name!()); + let shutdown = CancellationToken::new(); + let a = Sled::start(&log, &dir, 1, &root_pem, &shutdown).await; + let b = Sled::start(&log, &dir, 2, &root_pem, &shutdown).await; + a.peers.send(BTreeSet::from([b.addr])).unwrap(); + b.peers.send(BTreeSet::from([a.addr])).unwrap(); + eventually("universe convergence", 120, async || { + a.universe.borrow().rumors.network() == b.universe.borrow().rumors.network() + }) + .await; + sees(&a, &b).await; + + let authn = fake_identity(&mut root).await; + let Ceremony { + session_id, + client_nonce, + signer_nonce, + seq, + } = ceremony(&a.mgr, 1); + a.mgr + .session_start(&authn, session_id, client_nonce, signer_nonce, seq) + .await + .unwrap(); + + // A takes a job that runs only on B, and waits for it to start. + let job_id = Session::new(session_id).next_job_id(); + let job = sign_job_for(&mut root, job_id, session_id, "sleep 1000", &b.baseboard).await; + a.mgr + .job_start( + &authn, + job, + JobStartParams { + wait: JobWait::Start, + ..Default::default() + }, + ) + .await + .unwrap(); + + // A takes the stop, and returns once B has stopped the job. + let stop = JobStopParams { + wait: JobWait::Stop, + ..Default::default() + }; + timeout( + Duration::from_secs(60), + a.mgr.job_stop(&authn, &job_id, stop), + ) + .await + .expect("the stop waits only for B") + .unwrap(); + let status = a.mgr.job_status(&authn, &job_id).await.unwrap(); + assert!(status[&b.baseboard].is_terminal()); + assert!(!status.contains_key(&a.baseboard)); + + shutdown.cancel(); +} + /// The front door refuses a forged job, and the signed job it mimics /// still runs. #[named] @@ -977,12 +1132,12 @@ fn ceremony(mgr: &JobManager, seq: u64) -> Ceremony { } async fn sees(anchor: &Sled, joiner: &Sled) { - eventually("the anchor sees the joiner", 60, async || { + eventually("the anchor links the joiner", 60, async || { anchor .mgr .versions() .iter() - .any(|row| row.baseboard == joiner.baseboard) + .any(|row| row.baseboard == joiner.baseboard && row.health == Some(SledHealth::Linked)) }) .await; } @@ -1686,6 +1841,7 @@ async fn witnessed_session_survives_restart() { a.universe.borrow().rumors.network() == b.universe.borrow().rumors.network() }) .await; + sees(&a, &b).await; session.job_started(j1); let j2_id = session.next_job_id(); diff --git a/server/tests/output/job-start-interactive-request.bin b/server/tests/output/job-start-interactive-request.bin index 126d9c57..bb27ad91 100644 Binary files a/server/tests/output/job-start-interactive-request.bin and b/server/tests/output/job-start-interactive-request.bin differ diff --git a/server/tests/output/job-start-request.bin b/server/tests/output/job-start-request.bin index 845023a0..846dd10f 100644 Binary files a/server/tests/output/job-start-request.bin and b/server/tests/output/job-start-request.bin differ diff --git a/sush.json b/sush.json index 2e8167a2..0bcca7f1 100644 --- a/sush.json +++ b/sush.json @@ -439,8 +439,15 @@ "required": true }, "responses": { - "204": { - "description": "resource updated" + "200": { + "description": "successful operation", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/JobStarted" + } + } + } }, "4XX": { "$ref": "#/components/responses/Error" @@ -1139,6 +1146,23 @@ "session_id" ] }, + "JobStarted": { + "description": "Acknowledgement that a job was accepted.", + "type": "object", + "properties": { + "sleds": { + "description": "Which sleds the job will run on.", + "type": "array", + "items": { + "$ref": "#/components/schemas/BaseboardId" + }, + "uniqueItems": true + } + }, + "required": [ + "sleds" + ] + }, "JobStatus": { "oneOf": [ { diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index f0239e3b..10e52924 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -4,6 +4,7 @@ //! Job manager tests. +use std::collections::BTreeSet; use std::mem::MaybeUninit; use std::os::unix::fs::PermissionsExt as _; use std::path::Path; @@ -1014,10 +1015,11 @@ async fn job_output_perms() { assert_eq!(mode(&out.job_output_path(&job_id, Stderr)).await, 0o600); } +/// A cubby resolves through the rack's cubby map when a job is +/// accepted. #[named] #[tokio::test] async fn cubby_targets() { - // A cubby target only matches through the rack's cubby map. let log = test_logger(function_name!()); let dir = TempDir::with_prefix("sush-").unwrap(); let mut root = ephemeral_test_root(); @@ -1049,74 +1051,40 @@ async fn cubby_targets() { .await .unwrap(); let target: Target = "14".parse().unwrap(); - - // With the map empty, a cubby-targeted job records no status here. - // The all-target job behind it in the session's job chain proves - // it was processed, not merely still queued. - let skipped_id = session.next_job_id(); - let job = root - .sign_job_request_for(skipped_id, session_id, "true", false, target.clone()) - .await; - mgr.job_start(&authn, job.clone().into_signed(), JobStartParams::default()) - .await - .unwrap(); - session.job_started(job.into_signed()); let job_id = session.next_job_id(); let job = root - .sign_job_request(job_id, session_id, "true", false) + .sign_job_request_for(job_id, session_id, "true", false, target) .await; - mgr.job_start( - &authn, - job.clone().into_signed(), - JobStartParams { - wait: JobWait::Stop, - ..Default::default() - }, - ) - .await - .unwrap(); - session.job_started(job.into_signed()); + let wait = JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }; + + // With the map empty, no known sled occupies the cubby, so the + // job is refused. assert!(matches!( - mgr.job_status(&authn, &skipped_id).await.unwrap_err(), - JobError::JobNotFound(jid) if jid == skipped_id + mgr.job_start(&authn, job.clone().into_signed(), wait.clone()) + .await, + Err(JobError::UnknownCubby(14)), )); // Name this baseboard in the map. The update lands whenever the - // state task gets to it, so probe with fresh jobs until one takes. + // state task gets to it, so retry the job until it is accepted. cubbies .send(Cubbies::from([(14, test_baseboard_id())])) .unwrap(); - loop { - let job_id = session.next_job_id(); - let job = root - .sign_job_request_for(job_id, session_id, "true", false, target.clone()) - .await; - mgr.job_start(&authn, job.clone().into_signed(), JobStartParams::default()) + let sleds = loop { + match mgr + .job_start(&authn, job.clone().into_signed(), wait.clone()) .await - .unwrap(); - session.job_started(job.into_signed()); - sleep(Duration::from_millis(50)).await; - if mgr.job_status(&authn, &job_id).await.is_ok() { - break; + { + Ok(sleds) => break sleds, + Err(JobError::UnknownCubby(14)) => sleep(Duration::from_millis(50)).await, + Err(error) => panic!("job start failed: {error}"), } - } - - // Now cubby-targeted jobs run here. - let job_id = session.next_job_id(); - let job = root - .sign_job_request_for(job_id, session_id, "true", false, target) - .await; - mgr.job_start( - &authn, - job.clone().into_signed(), - JobStartParams { - wait: JobWait::Stop, - ..Default::default() - }, - ) - .await - .unwrap(); + }; session.job_started(job.into_signed()); + assert_eq!(sleds, BTreeSet::from([test_baseboard_id()])); let status = mgr.job_status(&authn, &job_id).await.unwrap()[mgr.own_baseboard()].clone(); check_status_stopped(status, &job_id, Ok(0), Some(0), Some(0)); } @@ -1593,9 +1561,9 @@ async fn attribution() { .unwrap(); } -/// A job runs only on the sleds its signed target names. Jobs naming -/// another baseboard, or a cubby while no mapping is known, are -/// processed but never run here, and the causal chain moves on. +/// A job runs only on the sleds its target names that the accepting +/// sled can reach. A job left with none, because it names only an +/// unreachable baseboard or an unknown cubby, is refused. #[named] #[tokio::test] async fn job_targets() { @@ -1629,29 +1597,30 @@ async fn job_targets() { .unwrap() .verify_with_cert(root.cert()) .unwrap(); - mgr.job_start(&authn, job.clone().into_signed(), JobStartParams::default()) - .await - .unwrap(); + let sleds = mgr + .job_start(&authn, job.clone().into_signed(), JobStartParams::default()) + .await?; session.job_started(job.into_signed()); - job_id + Ok::<_, JobError>((job_id, sleds)) }; - let elsewhere = start("true", "913-0000019:BRM00000000").await; - let by_cubby = start("true", "14").await; - let here = start("true", &test_baseboard_id().to_string()).await; - - // The targeted job runs, which proves the untargeted ones were - // already processed and skipped. + let elsewhere = "913-0000019:BRM00000000"; + assert!(matches!( + start("true", elsewhere).await, + Err(JobError::NoReachableTarget), + )); + assert!(matches!( + start("true", "14").await, + Err(JobError::UnknownCubby(14)), + )); + let (here, sleds) = start("true", &format!("{elsewhere},{}", test_baseboard_id())) + .await + .unwrap(); + assert_eq!(sleds, BTreeSet::from([test_baseboard_id()])); timeout(Duration::from_secs(30), mgr.wait_for_job_status(&here)) .await .expect("timed out waiting for targeted job") .unwrap(); - for skipped in [elsewhere, by_cubby] { - assert!(matches!( - mgr.job_status(&authn, &skipped).await, - Err(JobError::JobNotFound(_)), - )); - } } /// A login gossips as evidence every sled verifies for itself: real @@ -2670,7 +2639,9 @@ async fn streaming_validation() { expect_invalid(JobMode::StreamOutput, Target::All).await; expect_invalid( JobMode::StreamOutput, - format!("{},14", test_baseboard_id()).parse().unwrap(), + format!("{},913-0000019:BRM00000000", test_baseboard_id()) + .parse() + .unwrap(), ) .await; expect_invalid(JobMode::StreamInput, test_baseboard_id().into()).await;