From 0f1941669463f48df92550df846cd3e6499430ff Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sun, 25 Oct 2026 13:38:33 -0600 Subject: [PATCH] Resolve job targets eagerly Previously, we resolved cubby numbers as job targets at execution time, using the server's currently available cubby map. The problem is that soon after boot, that cubby map is still empty, and so the server silently skipped jobs that named its cubby. In addition, a job that targeted a sled that was down or absent at job submission time could run whenever that sled later rejoined, perhaps quite a lot later (#75). With this change, job targets are resolved once up front by the accepting sled. It filters out the sleds it doesn't have a link to, and gossips the resolved targets with the job start request. Each sled checks that the signed target matches the gossiped target. Jobs that name an unreported cubby or have no reachable sled are refused. Starting a job now returns the sleds it will run on. The client reports any named sled the job will not run on, and a waiting client or server waits for exactly those sleds. Fixes #75. Co-authored-by: Claude --- api/src/lib.rs | 11 +- client/src/cli.rs | 30 ++- client/src/commands.rs | 112 ++++++----- client/src/context.rs | 3 +- client/src/lib.rs | 1 + client/src/repl.rs | 5 +- common/src/jobs.rs | 53 +++-- common/src/targets.rs | 125 ++++++++---- server/src/error.rs | 6 +- server/src/gossip.rs | 13 ++ server/src/manager.rs | 79 ++++++-- server/src/messages.rs | 22 ++- server/src/rack.rs | 4 +- server/src/server.rs | 10 +- server/src/state.rs | 130 +++++-------- server/tests/distributed.rs | 182 ++++++++++++++++-- .../output/job-start-interactive-request.bin | Bin 437 -> 489 bytes server/tests/output/job-start-request.bin | Bin 376 -> 479 bytes sush.json | 28 ++- tests/src/manager_tests.rs | 123 +++++------- 20 files changed, 619 insertions(+), 318 deletions(-) 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 126d9c57cee24802f8d94676936cc55788a81914..bb27ad919b251107cdb6d70d7db5ca9f2d89c777 100644 GIT binary patch delta 65 zcmdnW{E~Ts7i05A?<_`>#zol$iA5#xd8N5YsYThAhQ_)EAYf>jTbx>ynV18UaSHM^ MF)}hTFfcX%016})2mk;8 delta 12 TcmaFKyp?%^7h}^#?<__DAJznE diff --git a/server/tests/output/job-start-request.bin b/server/tests/output/job-start-request.bin index 845023a004eebddfbb620de1b0a4c48d26f6f401..846dd10fc79cb87a75f8ddf4762f9ea2fef9befd 100644 GIT binary patch delta 116 zcmeytbf0;G7i05AZ(YXprbXEWiA5#xd8N5YsYThAhQ_)EAYf>jTbx>ynV18UaSHM^ PF)}hTFfcYCU5g 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;