diff --git a/Cargo.lock b/Cargo.lock index ac94b0c..f3d4a8a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1477,7 +1477,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2328,7 +2328,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3535,7 +3535,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3917,7 +3917,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3984,7 +3984,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4617,7 +4617,7 @@ dependencies = [ [[package]] name = "sprockets-tls" version = "0.1.0" -source = "git+https://github.com/oxidecomputer/sprockets.git?rev=b8782742d1bd91546887fbc1757a7d024a5469ef#b8782742d1bd91546887fbc1757a7d024a5469ef" +source = "git+https://github.com/oxidecomputer/sprockets.git?rev=98615b86d51f74ce220198eb16c57932fd24e7cb#98615b86d51f74ce220198eb16c57932fd24e7cb" dependencies = [ "anyhow", "attest-data 0.5.0 (git+https://github.com/oxidecomputer/dice-util?rev=4a39ef08d81e5177edee0bddb1146032aa21074d)", @@ -4649,7 +4649,7 @@ dependencies = [ [[package]] name = "sprockets-tls-test-utils" version = "0.1.0" -source = "git+https://github.com/oxidecomputer/sprockets.git?rev=b8782742d1bd91546887fbc1757a7d024a5469ef#b8782742d1bd91546887fbc1757a7d024a5469ef" +source = "git+https://github.com/oxidecomputer/sprockets.git?rev=98615b86d51f74ce220198eb16c57932fd24e7cb#98615b86d51f74ce220198eb16c57932fd24e7cb" dependencies = [ "camino", "pki-playground 0.2.0 (git+https://github.com/oxidecomputer/pki-playground?rev=7600756029ce046a02c6234aa84ce230cc5eaa04)", @@ -4925,6 +4925,7 @@ name = "sush-server" version = "0.1.0" dependencies = [ "async-trait", + "atomicwrites", "attest-mock", "bytes", "bytesize", @@ -5101,10 +5102,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5882,7 +5883,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 0a769df..36381c3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,8 +60,8 @@ signature = "2" sled-hardware-types = { git = "https://github.com/oxidecomputer/omicron", tag = "rel/v21/rc1" } slog = "2" slog-term = "2" -sprockets-tls = { git = "https://github.com/oxidecomputer/sprockets.git", rev = "b8782742d1bd91546887fbc1757a7d024a5469ef", default-features = false } -sprockets-tls-test-utils = { git = "https://github.com/oxidecomputer/sprockets.git", rev = "b8782742d1bd91546887fbc1757a7d024a5469ef" } +sprockets-tls = { git = "https://github.com/oxidecomputer/sprockets.git", rev = "98615b86d51f74ce220198eb16c57932fd24e7cb", default-features = false } +sprockets-tls-test-utils = { git = "https://github.com/oxidecomputer/sprockets.git", rev = "98615b86d51f74ce220198eb16c57932fd24e7cb" } ssh-key = { version = "0.6", features = ["ed25519", "p256", "serde"] } tempfile = "3" thiserror = "1" diff --git a/client/src/commands.rs b/client/src/commands.rs index 2aba0fd..d1b3c3e 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -872,10 +872,29 @@ async fn authenticate( if let Some(via) = via { request = request.via(via.to_string()); } - let identity = request.send().await?.into_inner(); + let identity = match request.send().await { + Ok(identity) => identity.into_inner(), + Err(error) if error.status() == Some(StatusCode::UNAUTHORIZED) => { + let server = client.version().send().await.ok().map(|v| v.into_inner()); + return Err(login_refused(server, error.into())); + } + Err(error) => return Err(error.into()), + }; Ok((identity, Authz::new(credentials, key))) } +/// A refused login usually indicates build skew rather than a bad key. +/// Blame the builds when they differ, otherwise return the server error. +fn login_refused(server: Option, error: CommandError) -> CommandError { + let client = VersionInfo::current(); + match server { + Some(server) if server.commit != client.commit => { + CommandError::AuthnBuildMismatch { client, server } + } + _ => error, + } +} + /// Make a request as someone who is logged in, logging in if needed. /// The client's pre-send hook signs each attempt. async fn with_login( @@ -2357,6 +2376,15 @@ pub enum CommandError { AmbiguousSerial(String), #[error("❌ Authentication error")] Authn(#[from] AuthnError), + #[error( + "❌ Authentication failed, client and server are different builds\n \ + Client: {client}\n \ + Server: {server}" + )] + AuthnBuildMismatch { + client: VersionInfo, + server: VersionInfo, + }, #[error("❌ Canceled")] Canceled, #[error("❌ Certificate for `{0}` is outside its validity window")] @@ -2554,3 +2582,29 @@ impl From> for CommandError { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn login_refused_blames_build_skew() { + let refused = || CommandError::Client(String::from("Authentication required, try `iam`")); + let skewed = VersionInfo { + version: String::from("0.1.0"), + commit: String::from("not-this-build"), + }; + assert!(matches!( + login_refused(Some(skewed), refused()), + CommandError::AuthnBuildMismatch { .. } + )); + assert!(matches!( + login_refused(Some(VersionInfo::current()), refused()), + CommandError::Client(_) + )); + assert!(matches!( + login_refused(None, refused()), + CommandError::Client(_) + )); + } +} diff --git a/common/src/jobs.rs b/common/src/jobs.rs index 7203c9f..186def1 100644 --- a/common/src/jobs.rs +++ b/common/src/jobs.rs @@ -422,6 +422,8 @@ pub enum ProcessError { OutputLimitExceeded { stream: JobOutputStream, limit: u64 }, #[error("Unable to join job process: {0}")] Join(String), + #[error("The server restarted while the job was running")] + Interrupted, } impl ProcessError { diff --git a/server/Cargo.toml b/server/Cargo.toml index 27044d5..ce03ab9 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -16,6 +16,7 @@ embedded = [] test-support = [] [dependencies] +atomicwrites.workspace = true bytes.workspace = true camino.workspace = true bytesize.workspace = true diff --git a/server/src/bookmark.rs b/server/src/bookmark.rs new file mode 100644 index 0000000..7d5142d --- /dev/null +++ b/server/src/bookmark.rs @@ -0,0 +1,564 @@ +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at https://mozilla.org/MPL/2.0/. + +//! Durable gossip peer identity across restarts. +//! +//! A rumors [`Bookmark`] records who a peer is and how far it has +//! advanced, so a restarted sled reclaims its old identity instead of +//! stranding it. Rumors owns the record format and decides when to load +//! and store. We supply raw byte storage obeying two constraints: stores +//! are atomic, and a load never returns a record older than the newest +//! store we reported `Ok` (stale records corrupt causality, whereas +//! lost records merely strand identities). +//! +//! Storage is one small file per configured (M.2) slot. Loads read every +//! slot, and take the record with the highest sequence number; that slot +//! becomes the *home*. Stores go only to the home, since writing both +//! would either make the server dependent on the health of both or, done +//! merely best-effort, let a stale record load after a fresher disk dies, +//! violating the constraint above. The slots must never both be written +//! by live peers, and a record must never be restored from a backup. +//! A [`BookmarkSource`] hands out one handle per peer, with generation +//! numbers ensuring that a straggler from an abandoned universe can't +//! clobber its successor's record. + +use std::fs::Permissions; +use std::io::{self, Cursor, Write as _}; +use std::os::unix::fs::PermissionsExt as _; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use atomicwrites::{AtomicFile, OverwriteBehavior}; +use camino::{Utf8Path, Utf8PathBuf}; +use rumors::{Bookmark, BookmarkError, Serialized}; +use slog::{Logger, o, warn}; +use thiserror::Error; +use tokio::fs::read; +use tokio::io::AsyncWrite; +use tokio::sync::Mutex; +use tokio::task::spawn_blocking; + +/// The envelope magic. The payload includes its own magic. +/// A change to our envelope means a new magic. +const MAGIC: &[u8; 8] = b"SUSHBKMK"; + +/// Envelope layout: magic, big-endian sequence number, digest of the +/// sequence number and record together, record. The digest keeps a +/// damaged sequence number from silently reordering the slots. +const SEQ_LEN: usize = 8; +const DIGEST_LEN: usize = 32; + +fn digest(seq: u64, record: &[u8]) -> [u8; DIGEST_LEN] { + let mut hasher = sush_common::hash::Hasher::new(); + hasher.update(&seq.to_be_bytes()); + hasher.update(record); + *hasher.finalize().as_bytes() +} + +/// What a bookmark load or store failed at. +#[derive(Debug, Error)] +pub enum BookmarkIoError { + #[error("bookmark I/O failed on `{path}`: {error}")] + Io { + path: Utf8PathBuf, + #[source] + error: io::Error, + }, + #[error("every bookmark slot is corrupt (last: `{path}`)")] + Corrupt { path: Utf8PathBuf }, + #[error("serializing the bookmark record failed: {0}")] + Serialize(#[source] io::Error), + #[error("no bookmark slot is writable")] + NoSlot, + #[error("the bookmark was handed to a newer peer")] + Fenced, +} + +/// This server's bookmark storage. Hands out one handle per peer, +/// each superseding the last. +#[derive(Clone, Debug)] +pub struct BookmarkSource { + shared: Arc, +} + +#[derive(Debug)] +struct SharedStore { + log: Logger, + /// Candidate record files, one per boot M.2. Empty means this + /// server persists no identity (the standalone server, tests). + slots: Vec, + /// The newest generation. A handle from an older one may load + /// but not store. + generation: AtomicU64, + /// Serializes loads and stores across handles, so the newest + /// record on disk is always the newest store anyone `Ok`'d. + state: Mutex, + /// The newest sequence number committed to disk by this process, + /// under the lock every rename takes. Renames commit in sequence + /// order or not at all; see [`commit`]. + committed: std::sync::Mutex, +} + +#[derive(Debug, Default)] +struct StoreState { + /// The slot holding the newest record, once known. + home: Option, + /// The sequence number of the newest record. + seq: u64, +} + +impl BookmarkSource { + /// A source persisting to `slots`, each on its own device. + /// + /// Construct exactly one source per slot set per process, and feed + /// that same source to both [`seed_gossip`](crate::seed_gossip) + /// and [`spawn_gossip`](crate::gossip::spawn_gossip): everything + /// serializing the store lives inside it. The caller creates the + /// parent directories, one per boot M.2, writable by this server's + /// user. The record files are created and owned here. + pub fn new(log: &Logger, slots: Vec) -> Self { + Self { + shared: Arc::new(SharedStore { + log: log.new(o!("component" => "bookmark")), + slots, + generation: AtomicU64::new(0), + state: Mutex::new(StoreState::default()), + committed: std::sync::Mutex::new(0), + }), + } + } + + /// A source that loads and persists nothing. + pub fn null() -> Self { + Self::new(&Logger::root(slog::Discard, o!()), Vec::new()) + } + + /// A ratcheting handle for the next peer. + /// All earlier handles are superseded. + pub fn next_handle(&self) -> SushBookmark { + let generation = self.shared.generation.fetch_add(1, Ordering::SeqCst) + 1; + SushBookmark { + shared: self.shared.clone(), + generation, + shed: false, + } + } + + /// A handle that never touches storage, for a peer that must keep + /// gossiping after its real bookmark failed. + pub fn shed_handle(&self) -> SushBookmark { + SushBookmark { + shared: self.shared.clone(), + generation: 0, + shed: true, + } + } + + /// A probing handle: reads like the current peer's, but without + /// superseding anything. + fn probe_handle(&self) -> SushBookmark { + SushBookmark { + shared: self.shared.clone(), + generation: self.shared.generation.load(Ordering::SeqCst), + shed: false, + } + } + + /// Does the storage work? Reads every slot and proves at least one + /// writable by writing. + pub async fn probe(&self) -> Result<(), BookmarkIoError> { + if self.shared.slots.is_empty() { + return Ok(()); + } + let usable = match self.probe_handle().load().await { + Ok(_) => { + let mut writable = false; + for path in &self.shared.slots { + let probe = path.with_extension("probe"); + if tokio::fs::write(&probe, b"").await.is_ok() { + let _ = tokio::fs::remove_file(&probe).await; + writable = true; + break; + } + } + if writable { + Ok(()) + } else { + Err(BookmarkIoError::NoSlot) + } + } + Err(error) => Err(error), + }; + if let Err(error) = &usable { + warn!(self.shared.log, "no usable bookmark storage"; "error" => %error); + } + usable + } +} + +/// One peer's handle on the [`BookmarkSource`]. +#[derive(Debug)] +pub struct SushBookmark { + shared: Arc, + generation: u64, + shed: bool, +} + +impl SushBookmark { + /// Has this bookmark been overtaken by events? + fn obe(&self) -> bool { + self.generation < self.shared.generation.load(Ordering::SeqCst) + } + + /// Split a checksummed envelope into its sequence number and record. + fn parse(bytes: &[u8]) -> Option<(u64, Vec)> { + let payload = bytes.strip_prefix(MAGIC)?; + let (seq, rest) = payload.split_first_chunk::()?; + let (sum, record) = rest.split_first_chunk::()?; + let seq = u64::from_be_bytes(*seq); + (digest(seq, record) == *sum).then(|| (seq, record.to_vec())) + } + + /// Build the checksummed envelope around `record`. + fn envelope(seq: u64, record: &[u8]) -> Vec { + let mut bytes = Vec::with_capacity(MAGIC.len() + SEQ_LEN + DIGEST_LEN + record.len()); + bytes.extend_from_slice(MAGIC); + bytes.extend_from_slice(&seq.to_be_bytes()); + bytes.extend_from_slice(&digest(seq, record)); + bytes.extend_from_slice(record); + bytes + } +} + +/// Rename `envelope` into place iff `seq` is newer than everything +/// committed by this process. A store future dropped at its await +/// detaches the blocking write, whose rename would otherwise land on +/// top of the newer record that beat it. Runs on the blocking pool. +fn commit(shared: &SharedStore, path: &Utf8Path, seq: u64, envelope: &[u8]) -> io::Result<()> { + let mut committed = shared.committed.lock().unwrap(); + if seq <= *committed { + return Err(io::Error::other("superseded by a newer record")); + } + AtomicFile::new(path, OverwriteBehavior::AllowOverwrite) + .write(|file| { + file.set_permissions(Permissions::from_mode(0o600))?; + file.write_all(envelope) + }) + .map_err(|error| match error { + atomicwrites::Error::Internal(error) | atomicwrites::Error::User(error) => error, + })?; + *committed = seq; + Ok(()) +} + +impl BookmarkError for SushBookmark { + type Error = BookmarkIoError; +} + +impl Bookmark for SushBookmark { + type Reader = Cursor>; + + async fn load(&self) -> Result, Self::Error> { + if self.shed || self.shared.slots.is_empty() { + return Ok(None); + } + if self.obe() { + return Err(BookmarkIoError::Fenced); + } + let mut state = self.shared.state.lock().await; + let mut newest: Option<(u64, usize, Vec)> = None; + let mut corrupt: Option<&Utf8Path> = None; + for (index, path) in self.shared.slots.iter().enumerate() { + let bytes = match read(path).await { + Ok(bytes) => bytes, + Err(error) if error.kind() == io::ErrorKind::NotFound => continue, + Err(error) => { + return Err(BookmarkIoError::Io { + path: path.clone(), + error, + }); + } + }; + match Self::parse(&bytes) { + Some((seq, record)) => { + if newest.as_ref().is_none_or(|(newest, ..)| seq > *newest) { + newest = Some((seq, index, record)); + } + } + None => { + warn!( + self.shared.log, "skipping corrupt bookmark slot"; + "path" => %path, + ); + corrupt = Some(path); + } + } + } + match newest { + Some((seq, home, record)) => { + // A reserved sequence number outranks a re-read of the + // disk. Regressing would let a cancelled write's + // straggler collide with a fresh reservation. + if seq >= state.seq { + state.home = Some(home); + state.seq = seq; + } + Ok(Some(Cursor::new(record))) + } + // A present-but-unreadable record is an error: + // rumors must not mistake it for a fresh start. + None => match corrupt { + Some(path) => Err(BookmarkIoError::Corrupt { path: path.into() }), + None => Ok(None), + }, + } + } + + async fn store(&self, write: F) -> Result<(), Self::Error> + where + F: for<'a> FnOnce(&'a mut (dyn AsyncWrite + Unpin + Send)) -> Serialized<'a> + Send, + { + if self.shed || self.shared.slots.is_empty() { + return Ok(()); + } + + let mut buf = Cursor::new(Vec::new()); + write(&mut buf).await.map_err(BookmarkIoError::Serialize)?; + let record = buf.into_inner(); + + let mut state = self.shared.state.lock().await; + if self.obe() { + return Err(BookmarkIoError::Fenced); + } + let home = match state.home { + Some(home) => home, + None => self + .shared + .slots + .iter() + .position(|path| { + path.parent() + .is_some_and(|parent| parent.as_std_path().is_dir()) + }) + .ok_or(BookmarkIoError::NoSlot)?, + }; + let path = self.shared.slots[home].clone(); + + // Reserve the sequence number first, since a cancelled write + // may still land and must be outnumbered. + state.seq += 1; + let seq = state.seq; + let envelope = Self::envelope(seq, &record); + + let shared = self.shared.clone(); + let target = path.clone(); + let written = spawn_blocking(move || commit(&shared, &target, seq, &envelope)) + .await + .map_err(|join| io::Error::other(join.to_string())) + .and_then(|result| result); + + match written { + Ok(()) => { + state.home = Some(home); + Ok(()) + } + Err(error) => Err(BookmarkIoError::Io { path, error }), + } + } +} + +#[cfg(test)] +mod test { + use super::*; + + use std::fs::{create_dir, metadata, read, write}; + + use camino::Utf8PathBuf; + use tempfile::TempDir; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + /// A serializer closure writing fixed bytes, shaped like rumors'. + fn record( + bytes: &'static [u8], + ) -> impl for<'a> FnOnce(&'a mut (dyn AsyncWrite + Unpin + Send)) -> Serialized<'a> + Send { + move |w| Box::pin(async move { w.write_all(bytes).await }) + } + + /// Two slot paths in separate directories, like two M.2s. + fn slots(dir: &TempDir) -> Vec { + ["m2a", "m2b"] + .iter() + .map(|m2| { + let parent = Utf8PathBuf::from_path_buf(dir.path().join(m2)).unwrap(); + create_dir(&parent).unwrap(); + parent.join("bookmark") + }) + .collect() + } + + fn envelope(seq: u64, record: &[u8]) -> Vec { + SushBookmark::envelope(seq, record) + } + + async fn read_back(handle: &SushBookmark) -> Option> { + let mut reader = handle.load().await.unwrap()?; + let mut bytes = Vec::new(); + reader.read_to_end(&mut bytes).await.unwrap(); + Some(bytes) + } + + fn test_log() -> Logger { + Logger::root(slog::Discard, o!()) + } + + /// A stored record loads back verbatim, sequenced and private. + #[tokio::test] + async fn round_trip() { + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let slots = slots(&dir); + let source = BookmarkSource::new(&test_log(), slots.clone()); + + let handle = source.next_handle(); + assert!(read_back(&handle).await.is_none()); + handle.store(record(b"who we are")).await.unwrap(); + assert_eq!(read_back(&handle).await.unwrap(), b"who we are"); + + let bytes = read(&slots[0]).unwrap(); + assert_eq!(bytes, envelope(1, b"who we are")); + let mode = metadata(&slots[0]).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o600); + } + + /// The newest record wins the load regardless of slot, and its + /// slot becomes the home every store then writes. + #[tokio::test] + async fn newest_slot_is_home() { + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let slots = slots(&dir); + write(&slots[0], envelope(5, b"stale")).unwrap(); + write(&slots[1], envelope(9, b"fresh")).unwrap(); + + let source = BookmarkSource::new(&test_log(), slots.clone()); + let handle = source.next_handle(); + assert_eq!(read_back(&handle).await.unwrap(), b"fresh"); + + handle.store(record(b"fresher")).await.unwrap(); + assert_eq!(read(&slots[0]).unwrap(), envelope(5, b"stale")); + assert_eq!(read(&slots[1]).unwrap(), envelope(10, b"fresher")); + } + + /// Minting a new handle fences the old one's stores. + #[tokio::test] + async fn stale_generations_cannot_store() { + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let source = BookmarkSource::new(&test_log(), slots(&dir)); + + let old = source.next_handle(); + old.store(record(b"before")).await.unwrap(); + let new = source.next_handle(); + assert!(matches!( + old.store(record(b"after")).await, + Err(BookmarkIoError::Fenced) + )); + assert_eq!(read_back(&new).await.unwrap(), b"before"); + } + + /// A corrupt slot is skipped when another is valid, and is an + /// error rather than absence when nothing valid remains. + #[tokio::test] + async fn corruption_is_never_absence() { + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let slots = slots(&dir); + write(&slots[0], b"scribble").unwrap(); + write(&slots[1], envelope(3, b"good")).unwrap(); + + let source = BookmarkSource::new(&test_log(), slots.clone()); + assert_eq!(read_back(&source.next_handle()).await.unwrap(), b"good"); + + write(&slots[1], b"more scribble").unwrap(); + assert!(matches!( + source.next_handle().load().await, + Err(BookmarkIoError::Corrupt { .. }) + )); + } + + /// A damaged sequence number fails the digest rather than silently + /// reordering the slots. + #[tokio::test] + async fn a_flipped_sequence_number_is_corruption() { + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let slots = slots(&dir); + let mut bytes = envelope(3, b"good"); + bytes[MAGIC.len()] ^= 0x80; + write(&slots[0], bytes).unwrap(); + + let source = BookmarkSource::new(&test_log(), slots.clone()); + assert!(matches!( + source.next_handle().load().await, + Err(BookmarkIoError::Corrupt { .. }) + )); + } + + /// A straggling write from a dropped store future cannot land on + /// top of a newer committed record. + #[tokio::test] + async fn stragglers_cannot_clobber_newer_commits() { + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let slots = slots(&dir); + let source = BookmarkSource::new(&test_log(), slots.clone()); + + let newer = envelope(7, b"newer"); + let straggler = envelope(6, b"stale"); + commit(&source.shared, &slots[0], 7, &newer).unwrap(); + assert!(commit(&source.shared, &slots[0], 6, &straggler).is_err()); + assert_eq!(read(&slots[0]).unwrap(), newer); + } + + /// A superseded handle can no longer load: its view of home and + /// sequence state belongs to a dead peer. + #[tokio::test] + async fn stale_generations_cannot_load() { + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let source = BookmarkSource::new(&test_log(), slots(&dir)); + let old = source.next_handle(); + old.store(record(b"before")).await.unwrap(); + let _new = source.next_handle(); + assert!(matches!(old.load().await, Err(BookmarkIoError::Fenced))); + } + + /// Probing proves writability by writing, not by guessing from + /// directory metadata. + #[tokio::test] + async fn probe_rejects_unwritable_storage() { + let source = BookmarkSource::new( + &test_log(), + vec![Utf8PathBuf::from("/nonexistent/sush/bookmark")], + ); + assert!(matches!(source.probe().await, Err(BookmarkIoError::NoSlot))); + + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let source = BookmarkSource::new(&test_log(), slots(&dir)); + source.probe().await.unwrap(); + } + + /// A slotless source and a shed handle persist nothing and never + /// fail, and a shed handle ignores even an existing record. + #[tokio::test] + async fn none_and_shed_touch_nothing() { + let source = BookmarkSource::null(); + let handle = source.next_handle(); + assert!(read_back(&handle).await.is_none()); + handle.store(record(b"lost")).await.unwrap(); + assert!(read_back(&handle).await.is_none()); + + let dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let slots = slots(&dir); + write(&slots[0], envelope(7, b"kept")).unwrap(); + let source = BookmarkSource::new(&test_log(), slots.clone()); + let shed = source.shed_handle(); + assert!(read_back(&shed).await.is_none()); + shed.store(record(b"dropped")).await.unwrap(); + assert_eq!(read(&slots[0]).unwrap(), envelope(7, b"kept")); + } +} diff --git a/server/src/gossip.rs b/server/src/gossip.rs index 93debfb..6fca713 100644 --- a/server/src/gossip.rs +++ b/server/src/gossip.rs @@ -14,8 +14,9 @@ //! through a fresh link to the peer that beat it, and the process repeats //! until one universe remains. //! -//! The manager publishes its current [`Rumors`] handle on a watch channel. -//! A migration replaces the handle entirely; consumers must re-subscribe, +//! The manager publishes its current [`Universe`] (the [`Rumors`] handle +//! and the causal frontier we joined it at) on a watch channel. A +//! migration replaces the handle entirely; consumers must re-subscribe, //! and everything the old universe carried is gone. //! //! TODO: re-inject local state into the new universe after a migration @@ -28,7 +29,7 @@ use std::net::{SocketAddr, SocketAddrV6}; use std::time::Duration; use futures::StreamExt as _; -use rumors::{Error, Network, Peer, Rumors, Ticks}; +use rumors::{Error, Joined, Network, Peer, Rumors, Ticks, Version}; use serde::Serialize; use serde::de::DeserializeOwned; use slog::{Logger, debug, info, o, warn}; @@ -41,6 +42,7 @@ use tokio_util::sync::CancellationToken; use rumors::link::routed::Endpoint; +use crate::bookmark::{BookmarkSource, SushBookmark}; use crate::link::{CorpusSource, SprocketsDial, SprocketsLink, Transport}; /// Manager timing. The defaults suit a rack; tests shrink them. @@ -65,11 +67,30 @@ impl Default for GossipConfig { } } +/// A gossip universe and where we entered it. +#[derive(Clone, Debug)] +pub struct Universe { + /// The gossiped set. + pub rumors: Rumors, + /// The causal frontier of the set received when we joined, + /// or `None` if we seeded the universe ourselves. + pub frontier: Option, +} + +impl Universe { + pub fn genesis(rumors: Rumors) -> Self { + Self { + rumors, + frontier: None, + } + } +} + /// A single-peer universe that never changes. The standalone server uses /// this, as does a sled that cannot gossip. The receiver outlives its /// sender. -pub fn isolated(seed: Rumors) -> watch::Receiver> { - let (_tx, rx) = watch::channel(seed); +pub fn isolated(seed: Rumors) -> watch::Receiver> { + let (_tx, rx) = watch::channel(Universe::genesis(seed)); rx } @@ -100,9 +121,10 @@ pub async fn spawn_gossip( corpus: CorpusSource, listen_addr: SocketAddrV6, peers: watch::Receiver>, - seed: Rumors, + seed: Rumors, + bookmarks: BookmarkSource, shutdown: CancellationToken, -) -> io::Result<(SocketAddrV6, watch::Receiver>)> +) -> io::Result<(SocketAddrV6, watch::Receiver>)> where T: DeserializeOwned + Serialize + Send + Sync + 'static, { @@ -116,7 +138,7 @@ where ) .await?; let bound = transport.bound(); - let universe = spawn_gossip_manager(log, config, transport, peers, seed, shutdown); + let universe = spawn_gossip_manager(log, config, transport, peers, seed, bookmarks, shutdown); Ok((bound, universe)) } @@ -124,18 +146,20 @@ where /// the addresses on `peers`, drives gossip on every link, and resolves /// universe collisions. The returned channel follows the current universe, /// starting at `seed`. +#[allow(clippy::too_many_arguments)] pub fn spawn_gossip_manager( log: &Logger, config: GossipConfig, transport: Transport, peers: watch::Receiver>, - seed: Rumors, + seed: Rumors, + bookmarks: BookmarkSource, shutdown: CancellationToken, -) -> watch::Receiver> +) -> watch::Receiver> where T: DeserializeOwned + Serialize + Send + Sync + 'static, { - let (publish, subscribe) = watch::channel(seed.clone()); + let (publish, subscribe) = watch::channel(Universe::genesis(seed.clone())); let manager = Manager { log: log.new(o!("component" => "gossip manager")), config, @@ -143,6 +167,7 @@ where transport, peers, rumors: seed, + bookmarks, publish, drivers: JoinSet::new(), live: HashMap::new(), @@ -172,8 +197,9 @@ struct Manager { endpoint: Endpoint, transport: Transport, peers: watch::Receiver>, - rumors: Rumors, - publish: watch::Sender>, + rumors: Rumors, + bookmarks: BookmarkSource, + publish: watch::Sender>, drivers: JoinSet<(SocketAddr, Stopped)>, live: HashMap, dials: JoinSet, @@ -305,6 +331,12 @@ where /// the swap. On failure our universe is intact and the debt stands, so /// the next link retries; either way all links are rebuilt, since the /// old ones belong to the universe we are leaving. + /// + /// The new peer gets a fresh bookmark handle, fencing off all the + /// abandoned universe's stores. If the received identity cannot be + /// persisted, we keep gossiping with a shed handle rather than take + /// the sled out of gossip; a stranded identity is harmless, unlike + /// a support shell that cannot reach a degraded rack. async fn migrate(&mut self, peer: SocketAddr, mut link: SprocketsLink) { self.drivers.abort_all(); self.live.clear(); @@ -312,24 +344,51 @@ where self.log, "joining the universe that beat ours"; "peer" => %peer, "ours" => %self.rumors.network(), ); - match timeout(self.config.join_timeout, Peer::bootstrap().join(&mut link)).await { - Ok(Ok(Some(joined))) => { - self.rumors = joined.into_rumors(); - let _ = self.publish.send(self.rumors.clone()); - self.joins.clear(); - info!(self.log, "migrated"; "network" => %self.rumors.network()); + let bootstrap = Peer::bootstrap().bookmark(self.bookmarks.next_handle()); + match timeout(self.config.join_timeout, bootstrap.join(&mut link)).await { + Ok(Joined::Joined { peer }) => self.adopt(peer), + Ok(Joined::Unbookmarked(unbookmarked)) => { + warn!( + self.log, "cannot persist our identity, gossiping unbookmarked"; + "error" => %unbookmarked.error, + ); + let peer = match unbookmarked + .peer + .bookmark(self.bookmarks.shed_handle()) + .await + { + Ok(peer) => peer, + Err(_) => unreachable!("a shed bookmark never touches storage"), + }; + self.adopt(peer); } - Ok(Ok(None)) => warn!(self.log, "mutual bootstrap, retrying"), - Ok(Err(err)) => warn!(self.log, "join failed"; "error" => %err), + Ok(Joined::Bailed { .. }) => warn!(self.log, "mutual bootstrap, retrying"), + Ok(Joined::Failed { error, .. }) => warn!(self.log, "join failed"; "error" => %error), Err(_) => warn!(self.log, "join timed out"), } drop(link); self.link_absent(); } + + /// Follow the joined peer into its universe. + fn adopt(&mut self, peer: Peer) { + self.rumors = peer.into_rumors(); + let frontier = self.rumors.snapshot().latest().clone(); + let _ = self.publish.send(Universe { + rumors: self.rumors.clone(), + frontier: Some(frontier), + }); + self.joins.clear(); + info!(self.log, "migrated"; "network" => %self.rumors.network()); + } } /// Drive sessions on one link until it fails, reporting why. -async fn sessions(rumors: &Rumors, mut link: SprocketsLink, log: &Logger) -> Stopped +async fn sessions( + rumors: &Rumors, + mut link: SprocketsLink, + log: &Logger, +) -> Stopped where T: DeserializeOwned + Serialize + Send + Sync + 'static, { @@ -358,6 +417,13 @@ where Stopped::Failed }; } + // A bookmark failure also stops every later session at the + // persist gate, so it deserves a warning where routine + // link churn does not. + Err(Error::Bookmark(error)) => { + warn!(log, "bookmark failure stops gossip"; "error" => %error); + return Stopped::Failed; + } Err(err) => { debug!(log, "session failed"; "error" => %err); return Stopped::Failed; diff --git a/server/src/lib.rs b/server/src/lib.rs index 1df6919..2675827 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -10,6 +10,7 @@ extern crate function_name; #[cfg(all(feature = "embedded", feature = "test-support"))] compile_error!("`test-support` must not be enabled for an embedded server"); +pub mod bookmark; pub mod error; pub mod executor; pub mod gossip; diff --git a/server/src/main.rs b/server/src/main.rs index 1918d01..be837ad 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -20,6 +20,7 @@ use x509_cert::der::DecodePem as _; use sush_api::sush_api_mod::api_description; use sush_common::targets::Cubbies; +use sush_server::bookmark::BookmarkSource; use sush_server::executor::PathIsolation; use sush_server::gossip::isolated; use sush_server::manager::JobManager; @@ -94,7 +95,7 @@ async fn main() -> Result<(), String> { }; // TODO: get/seed Rumors network - let gossip = isolated(seed_gossip()); + let gossip = isolated(seed_gossip(&BookmarkSource::null()).await); #[cfg(feature = "test-support")] let roots = overridable_root_certs(&override_root_certs).await?; diff --git a/server/src/manager.rs b/server/src/manager.rs index 59b6cd9..0c8e296 100644 --- a/server/src/manager.rs +++ b/server/src/manager.rs @@ -43,7 +43,7 @@ use crate::executor::PathIsolation; use crate::job::SocketSender; use crate::messages::v0::{CertRequest, IdentityRequest, JobRequest, Request, SessionRequest}; use crate::output::{JobOutputDir, JobOutputFileStream}; -use crate::state::{GossipNetwork, MAX_CERTS, State, StateManager}; +use crate::state::{GossipUniverse, MAX_CERTS, State, StateManager}; /// Maximum number of cached identities. const MAX_CACHED_IDENTITIES: NonZeroUsize = NonZeroUsize::new(1_000).unwrap(); @@ -103,7 +103,7 @@ impl JobManager { output_dir: JobOutputDir, own_baseboard: BaseboardId, cubbies: watch::Receiver, - universe: watch::Receiver, + universe: watch::Receiver, roots: &[impl AsRef], shutdown: CancellationToken, ) -> Result { @@ -128,7 +128,7 @@ impl JobManager { output_dir: JobOutputDir, own_baseboard: BaseboardId, cubbies: watch::Receiver, - universe: watch::Receiver, + universe: watch::Receiver, roots: &[Certificate], shutdown: CancellationToken, ) -> Result { diff --git a/server/src/messages.rs b/server/src/messages.rs index 9a1f470..e7c732e 100644 --- a/server/src/messages.rs +++ b/server/src/messages.rs @@ -570,7 +570,7 @@ mod wire_format { assert_wire_format("job-stop-event", msg); let msg: VersionedMessage = Message::Event( - baseboard, + baseboard.clone(), Event::Job(JobEvent::Error( job_id, when, @@ -579,6 +579,13 @@ mod wire_format { ) .into(); assert_wire_format("job-error-event", msg); + + let msg: VersionedMessage = Message::Event( + baseboard, + Event::Job(JobEvent::Error(job_id, when, ProcessError::Interrupted)), + ) + .into(); + assert_wire_format("job-interrupted-event", msg); } #[test] diff --git a/server/src/state.rs b/server/src/state.rs index 3bc239b..219732c 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -12,9 +12,9 @@ use std::num::NonZeroUsize; use std::sync::{Arc, Mutex}; use chrono::{DateTime, Utc}; -use futures::{Stream, StreamExt}; +use futures::{FutureExt as _, Stream, StreamExt}; use lru::LruCache; -use rumors::{Peer, Rumors, Version}; +use rumors::{CausalMessages, Peer, Rumors, Version}; use sled_hardware_types::BaseboardId; use slog::{Logger, debug, error, info, o, warn}; use tokio::sync::watch; @@ -34,7 +34,9 @@ use sush_common::keys::{KeyError, KeyId, Signature, SshPublicKey}; use sush_common::targets::Cubbies; use sush_common::version::{VersionInfo, VersionMap}; +use crate::bookmark::{BookmarkSource, SushBookmark}; use crate::executor::{Executor, PathIsolation}; +use crate::gossip::Universe; use crate::history::JobHistory; use crate::job::SocketSender; use crate::messages::v0::{ @@ -46,8 +48,9 @@ use crate::output::JobOutputDir; pub type AttachmentPoints = BTreeMap>>; pub type Certificates = BTreeMap; -pub type GossipNetwork = Rumors; -pub type QueuedJobs = BTreeMap; +pub type GossipNetwork = Rumors; +pub type GossipUniverse = Universe; +pub type QueuedJobs = BTreeMap; pub type RunningJobs = BTreeMap<(JobId, BaseboardId), DateTime>; /// Maximum certificate chain length. @@ -83,6 +86,16 @@ pub struct RegisteredIdentity { pub verifier: RequestVerifier, } +/// A job waiting in the session queue for its turn in the causal chain. +/// Jobs replayed from history that predates our join keep the chain intact +/// but never execute here. +#[derive(Clone, Debug)] +pub struct QueuedJob { + pub job: SignedJob, + pub params: JobStartParams, + pub replayed: bool, +} + #[derive(Clone, Debug)] pub enum SessionState { Inactive { @@ -193,7 +206,7 @@ impl<'a> SessionGuard<'a> { self.inner.skip_job(*job_id) } - pub fn next_queued_job(&mut self) -> Option<(SignedJob, JobStartParams)> { + pub fn next_queued_job(&mut self) -> Option { self.queued_jobs.remove(&self.inner.next_job_id()) } @@ -208,6 +221,7 @@ impl<'a> SessionGuard<'a> { job: SignedJob, params: JobStartParams, actor: &KeyId, + replayed: bool, ) { let job_id = *job.job_id(); let targeted = job.payload().runs_on(own_baseboard, cubbies); @@ -222,9 +236,17 @@ impl<'a> SessionGuard<'a> { } else { // Insert the job into our queue. Every job joins the // queue to keep the causal chain whole, but only jobs - // targeting this sled record a local status. - self.queued_jobs.insert(job_id, (job, params)); - if targeted { + // targeting this sled that can actually run here record + // a local status. + self.queued_jobs.insert( + job_id, + QueuedJob { + job, + params, + replayed, + }, + ); + if targeted && !replayed { history.set_job_status( &job_id, own_baseboard, @@ -248,8 +270,12 @@ impl<'a> SessionGuard<'a> { own_baseboard: &BaseboardId, history: &mut JobHistory, running: &RunningJobs, + replayed: bool, ) { self.queued_jobs.remove(job_id); + if replayed { + return; + } history.transition_job_status( job_id, own_baseboard, @@ -272,8 +298,10 @@ impl<'a> SessionGuard<'a> { /// in the hash chain, and there may be an unbounded number of /// newly-ready-to-run jobs after it in the queue. Once we reach /// a fixed point, we have nothing further to do. + #[allow(clippy::too_many_arguments)] pub fn execute_ready_jobs( &mut self, + log: &Logger, own_baseboard: &BaseboardId, cubbies: &Cubbies, certs: &mut Certificates, @@ -281,19 +309,27 @@ impl<'a> SessionGuard<'a> { executor: &mut Executor, attachments: &mut AttachmentPoints, ) { - while let Some((request, params)) = self.next_queued_job() { + while let Some(QueuedJob { + job: request, + params, + replayed, + }) = self.next_queued_job() + { let (tx_attachment, rx_attachment) = watch::channel(None); let job_id = request.payload().job_id().to_owned(); - if request.payload().runs_on(own_baseboard, cubbies) - && history + if request.payload().runs_on(own_baseboard, cubbies) { + if replayed { + warn!(log, "not executing replayed job"; "job_id" => %job_id); + } else if history .get_job_status(&job_id) .map(|status| { matches!(status.get(own_baseboard), Some(JobStatus::Queued { .. })) }) .unwrap_or(true) - { - executor.job_start(certs, request.clone(), params, tx_attachment); - attachments.insert(job_id, rx_attachment); + { + executor.job_start(certs, request.clone(), params, tx_attachment); + attachments.insert(job_id, rx_attachment); + } } self.job_started(request); } @@ -325,6 +361,14 @@ pub struct State { roots: Box<[KeyId]>, /// Baseboards by cubby number, as much of it as is known. cubbies: Cubbies, + /// The causal frontier we joined this universe at, if we joined + /// rather than seeded it. + join_frontier: Option, + /// Jobs whose start event on our own baseboard arrived as replayed + /// history. A previous life started them, and no executor of ours + /// will ever stop them. A terminal event clears its job, so + /// mid-replay entries are only suspects. + zombies: BTreeSet, /// Message versions from newer builds, each warned about once. unknown_versions: BTreeSet, /// Build provenance by sled. @@ -343,6 +387,7 @@ impl State { own_baseboard: BaseboardId, root_certs: &[Certificate], session_sush_nonce: Arc>, + join_frontier: Option, ) -> Result { let certs = root_certs .iter() @@ -369,6 +414,8 @@ impl State { unknown_versions: Default::default(), identities: LruCache::new(MAX_REGISTERED_IDENTITIES), revoked_keys: LruCache::new(MAX_REVOKED_KEYS), + join_frontier, + zombies: Default::default(), }; new.validate_certs(&roots); for root in &roots { @@ -405,6 +452,36 @@ impl State { &self.history } + /// Jobs running on our own baseboard, according to our history. + pub fn own_running_jobs(&self) -> BTreeSet { + self.running + .keys() + .filter(|(_, baseboard)| *baseboard == self.own_baseboard) + .map(|(job_id, _)| *job_id) + .collect() + } + + /// Jobs a previous life of this server started and left running. + pub fn zombies(&self) -> BTreeSet { + self.zombies + .iter() + .filter(|job_id| { + self.running + .contains_key(&(**job_id, self.own_baseboard.clone())) + }) + .copied() + .collect() + } + + /// Whether a message at `version` is live traffic rather than + /// replayed history. Only strict causal descendants of the join + /// frontier are live. + fn is_live(&self, version: &Version) -> bool { + self.join_frontier + .as_ref() + .is_none_or(|frontier| version > frontier) + } + pub fn get_job_status(&self, job_id: &JobId) -> Option<&JobStatusMap> { self.history.get_job_status(job_id) } @@ -642,6 +719,7 @@ impl State { } } (actor, SessionRequest::Skip(session_id, job_id)) => { + let replayed = !self.is_live(incoming_version); if let Some(mut session) = self.session.active_session() && session.session_id() == *session_id { @@ -661,8 +739,10 @@ impl State { &self.own_baseboard, &mut self.history, &self.running, + replayed, ); session.execute_ready_jobs( + log, &self.own_baseboard, &self.cubbies, &mut self.certs, @@ -714,6 +794,7 @@ impl State { // Refused jobs targeting this sled record an error // status so the submitter learns their fate. let session_id = signed.payload().session_id(); + let live = self.is_live(incoming_version); match self.session.active_session() { Some(mut session) if session.session_id() == session_id => { session.enqueue_job( @@ -725,8 +806,10 @@ impl State { signed.clone(), params.clone(), actor, + !live, ); session.execute_ready_jobs( + log, &self.own_baseboard, &self.cubbies, &mut self.certs, @@ -743,7 +826,8 @@ impl State { "session_id" => %session_id, "actor" => %actor, ); - if signed.payload().runs_on(&self.own_baseboard, &self.cubbies) + if live + && signed.payload().runs_on(&self.own_baseboard, &self.cubbies) && !self.history.contains(&job_id) { executor.job_refused( @@ -758,6 +842,7 @@ impl State { } (actor, JobRequest::Stop(job_id)) => { executor.job_stop(job_id); + let replayed = !self.is_live(incoming_version); if let Some(mut session) = self.session.active_session() { session.cancel_job( job_id, @@ -765,6 +850,7 @@ impl State { &self.own_baseboard, &mut self.history, &self.running, + replayed, ); } } @@ -817,6 +903,11 @@ impl State { JobEvent::Start(job_id, when) => { info!(log, "job started"; "job_id" => %job_id, "when" => %when); self.running.insert((*job_id, baseboard_id.clone()), *when); + // A replayed start on our own baseboard is a + // previous life's. Only these can be zombies. + if *baseboard_id == self.own_baseboard && !self.is_live(incoming_version) { + self.zombies.insert(*job_id); + } self.history.set_job_status( job_id, baseboard_id, @@ -833,6 +924,7 @@ impl State { info!(log, "job stopped"; "job_id" => %job_id, "when" => %when, "result" => ?result); if baseboard_id == &self.own_baseboard { self.attachments.remove(job_id); + self.zombies.remove(job_id); } self.running.remove(&(*job_id, baseboard_id.clone())); self.history.transition_job_status( @@ -849,6 +941,21 @@ impl State { output: output.clone(), }) } + // An interrupted declaration raced this + // genuine stop. The real result wins in + // either arrival order, though the start + // time died with the previous life. + Some(JobStatus::Error { + error: ProcessError::Interrupted, + time_error, + .. + }) => Some(JobStatus::Stopped { + job_id: *job_id, + time_started: *time_error, + time_stopped: *when, + result: result.clone(), + output: output.clone(), + }), _ => None, }, self.session.queued_jobs(), @@ -862,17 +969,27 @@ impl State { error!(log, "job error"; "job_id" => %job_id, "when" => %when, "error" => %error); if baseboard_id == &self.own_baseboard { self.attachments.remove(job_id); + self.zombies.remove(job_id); } self.running.remove(&(*job_id, baseboard_id.clone())); - self.history.set_job_status( + self.history.transition_job_status( job_id, baseboard_id, - JobStatus::Error { - job_id: *job_id, - time_error: *when, - error: error.clone(), - }, Some(incoming_version.rank()), + // An error never displaces a terminal + // status, so a stop that raced an + // interrupted declaration keeps the real + // result. + |old_status| match old_status { + None + | Some(JobStatus::Queued { .. }) + | Some(JobStatus::Started { .. }) => Some(JobStatus::Error { + job_id: *job_id, + time_error: *when, + error: error.clone(), + }), + _ => None, + }, self.session.queued_jobs(), &self.running, ); @@ -904,14 +1021,121 @@ impl State { } } +/// Apply one gossip message to the state, reporting update errors back +/// onto the network. +/// +/// We unconditionally mark the watch sender as modified even though it +/// might not be, because *most* of the messages cause *some* modification +/// of the state, and we would rather be safe against future code changes +/// than manually tracking precisely which messages *don't* modify state. +/// The cost is a few spurious wakeups. +fn apply_message( + log: &Logger, + tx_state: &watch::Sender, + executor: &mut Executor, + rumors: Option<&GossipNetwork>, + own_baseboard: &BaseboardId, + version: &Version, + message: &Arc, +) { + tx_state.send_modify(|state| { + if let Err(error) = state.update(log, executor, version, message) { + error!(log, "state update failed"; "error" => ?error); + // Re-gossiping a replayed message's error would grow the + // set a little more on every rejoin by every sled. + if state.is_live(version) + && let Some(rumors) = rumors + { + debug!(log, "sending error to gossip network"; "error" => ?error); + rumors.send(Message::Event(own_baseboard.clone(), Event::Error(error)).into()); + } + } + }); +} + +/// Apply every message already delivered locally, without waiting for +/// more. Afterwards the local set is quiescent. +fn drain_ready( + log: &Logger, + causal_messages: &mut CausalMessages, + tx_state: &watch::Sender, + executor: &mut Executor, + rumors: Option<&GossipNetwork>, + own_baseboard: &BaseboardId, +) { + while let Some(Some((version, message))) = causal_messages.next().now_or_never() { + apply_message( + log, + tx_state, + executor, + rumors, + own_baseboard, + &version, + &message, + ); + } +} + +/// Declare interrupted every job a previous life left running, +/// excluding `survivors` (jobs this incarnation carried across a +/// universe swap) and jobs already reaped. Zombies come only from +/// replayed start events, so a job this incarnation started is never +/// a candidate. Reap only at local quiescence (after +/// [`drain_ready`]), when a zombie whose terminal event is already +/// here has been cleared. +fn reap_zombies( + log: &Logger, + tx_state: &watch::Sender, + rumors: Option<&GossipNetwork>, + own_baseboard: &BaseboardId, + survivors: &BTreeSet, + reaped: &mut BTreeSet, +) { + let mut zombies = tx_state.borrow().zombies(); + zombies.retain(|job_id| !survivors.contains(job_id) && !reaped.contains(job_id)); + for job_id in zombies { + reaped.insert(job_id); + if let Some(rumors) = rumors { + rumors.send( + Message::Event( + own_baseboard.clone(), + Event::Job(JobEvent::Error( + job_id, + Utc::now(), + ProcessError::Interrupted, + )), + ) + .into(), + ); + } + warn!(log, "interrupted job from a previous life"; "job_id" => %job_id); + } +} + /// Create a fresh gossip network with this server as its only peer. /// /// A peer that seeds its own network has no one to gossip with, so jobs run /// only on the server that accepted them, and no server learns about any other /// server's sessions. This stands in for joining the rack's network over /// sprockets on the bootstrap network. -pub fn seed_gossip() -> GossipNetwork { - Peer::seed().into_rumors() +/// +/// A pristine seed's bookmark touches no storage, and identities +/// recorded there are reclaimed only after a migration returns us to +/// their universe. Bad storage would abort every session at the +/// persist gate, before the seed could even learn to migrate. Probe +/// first and shed on failure. +pub async fn seed_gossip(bookmarks: &BookmarkSource) -> GossipNetwork { + let handle = match bookmarks.probe().await { + Ok(()) => bookmarks.next_handle(), + Err(_) => bookmarks.shed_handle(), + }; + match Peer::seed().bookmark(handle).await { + Ok(peer) => peer.into_rumors(), + Err(unbookmarked) => match unbookmarked.peer.bookmark(bookmarks.shed_handle()).await { + Ok(peer) => peer.into_rumors(), + Err(_) => unreachable!("a shed bookmark never touches storage"), + }, + } } #[derive(Debug)] @@ -938,7 +1162,7 @@ impl StateManager { own_baseboard: BaseboardId, mut requests: R, mut cubbies: watch::Receiver, - universe: watch::Receiver, + mut universe: watch::Receiver, roots: &[Certificate], session_sush_nonce: Arc>, shutdown: CancellationToken, @@ -946,22 +1170,31 @@ impl StateManager { where R: Stream + Send + Unpin + 'static, { - // We report our current state through a watch channel. - let mut initial_state = - State::new(own_baseboard.clone(), roots, session_sush_nonce.clone())?; - initial_state.cubbies = cubbies.borrow_and_update().clone(); - let (tx_state, rx_state) = watch::channel(initial_state); - let roots = roots.to_vec(); - // We process messages in causal order, so that we can rely on // things like "the session stop happens after its corresponding // session start". This costs a little extra in-memory bookkeeping // and computation, but makes it much easier to ensure that our // state machine is correct, because it now only has to be correct // in the face of arbitrary *causal* reorderings. - let initial = universe.borrow().clone(); + // `borrow_and_update` marks the value seen, so a migration + // that landed before we subscribed does not replay as a swap. + let Universe { + rumors: initial, + frontier, + } = universe.borrow_and_update().clone(); let mut causal_messages = initial.causal_messages(); + // We report our current state through a watch channel. + let mut initial_state = State::new( + own_baseboard.clone(), + roots, + session_sush_nonce.clone(), + frontier.clone(), + )?; + initial_state.cubbies = cubbies.borrow_and_update().clone(); + let (tx_state, rx_state) = watch::channel(initial_state); + let roots = roots.to_vec(); + // The executor needs to have access to send messages back. let (mut executor, mut events) = Executor::new( log.new(o!("component" => "executor")), @@ -981,6 +1214,15 @@ impl StateManager { spawn(async move { info!(log, "managing state"); + // Replay bookkeeping. `frontier` classifies incoming + // messages (at or concurrent with it means replayed + // history); `survivors` are jobs this incarnation itself + // runs across a universe swap; `reaped` are zombies + // already declared interrupted. + let mut frontier = frontier; + let mut survivors: BTreeSet = BTreeSet::new(); + let mut reaped: BTreeSet = BTreeSet::new(); + // Announce our build. if let Some((rumors, _)) = &gossip { rumors.send( @@ -1051,21 +1293,36 @@ impl StateManager { break; } Some((version, message)) => { - // We unconditionally mark the watch sender as modified - // even though it might not be, because *most* of the - // messages cause *some* modification of the state, and we - // would rather be safe against future code changes than - // manually tracking precisely which messages *don't* modify - // state. The cost is a few spurious wakeups. - tx_state.send_modify(|state| { - if let Err(error) = state.update(&log, &mut executor, &version, &message) { - error!(log, "state update failed"; "error" => ?error); - if let Some((rumors, _)) = &gossip { - debug!(log, "sending error to gossip network"; "error" => ?error); - rumors.send(Message::Event(own_baseboard.clone(), Event::Error(error)).into()); - } - } - }); + // Past the join frontier means the message is live. + let live = frontier.as_ref().is_none_or(|f| version > f); + apply_message( + &log, + &tx_state, + &mut executor, + gossip.as_ref().map(|(rumors, _)| rumors), + &own_baseboard, + &version, + &message, + ); + // Replayed traffic can reveal zombies. + if !live { + drain_ready( + &log, + &mut causal_messages, + &tx_state, + &mut executor, + gossip.as_ref().map(|(rumors, _)| rumors), + &own_baseboard, + ); + reap_zombies( + &log, + &tx_state, + gossip.as_ref().map(|(rumors, _)| rumors), + &own_baseboard, + &survivors, + &mut reaped, + ); + } }, }, @@ -1095,19 +1352,26 @@ impl StateManager { let (rumors, universe) = gossip.as_mut().expect("gossip present when it changes"); let fresh = universe.borrow_and_update().clone(); - info!(log, "gossip universe changed, resetting state"; "network" => %fresh.network()); - causal_messages = fresh.causal_messages(); + info!(log, "gossip universe changed, resetting state"; "network" => %fresh.rumors.network()); + causal_messages = fresh.rumors.causal_messages(); + // A reaped zombie may still show as running (its + // error event races the swap); it is no survivor. + survivors = tx_state.borrow().own_running_jobs(); + survivors.retain(|job_id| !reaped.contains(job_id)); + reaped = BTreeSet::new(); + frontier = fresh.frontier.clone(); // TODO: re-inject local job state (policy pending). tx_state.send_modify(|state| { *state = State::new( own_baseboard.clone(), &roots, state.session_sush_nonce.clone(), + fresh.frontier.clone(), ) .expect("roots validated at startup"); state.cubbies = cubbies.borrow().clone(); }); - *rumors = fresh; + *rumors = fresh.rumors; rumors.send( Message::Event( own_baseboard.clone(), @@ -1115,6 +1379,24 @@ impl StateManager { ) .into(), ); + // The set received at join is already local: + // drain it, then reap. + drain_ready( + &log, + &mut causal_messages, + &tx_state, + &mut executor, + Some(&*rumors), + &own_baseboard, + ); + reap_zombies( + &log, + &tx_state, + Some(&*rumors), + &own_baseboard, + &survivors, + &mut reaped, + ); } } }), diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index 74e4314..48d9222 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -16,15 +16,20 @@ use tempfile::TempDir; use tokio::sync::watch; use tokio_util::sync::CancellationToken; +use chrono::Utc; use sush_api::{JobStartParams, JobWait}; -use sush_common::jobs::{JobStatus, Session, SessionId, SessionSignerNonce}; +use sush_common::jobs::{ + JobId, JobOutputState, JobStatus, ProcessError, Session, SessionId, SessionSignerNonce, +}; use sush_common::keys::pem_cert_chain; use sush_common::targets::Cubbies; use sush_common::version::VersionInfo; +use sush_server::bookmark::BookmarkSource; use sush_server::executor::PathIsolation; use sush_server::gossip::spawn_gossip; +use sush_server::messages::v0::{Event, JobEvent, Message}; use sush_server::output::JobOutputDir; -use sush_server::state::GossipNetwork; +use sush_server::state::GossipUniverse; use sush_server::{JobManager, seed_gossip}; use common::{ @@ -34,7 +39,7 @@ use common::{ struct Sled { mgr: JobManager, - universe: watch::Receiver, + universe: watch::Receiver, peers: watch::Sender>, addr: SocketAddrV6, baseboard: BaseboardId, @@ -48,6 +53,25 @@ impl Sled { identity: usize, root_pem: &Utf8PathBuf, shutdown: &CancellationToken, + ) -> Sled { + Self::start_with_bookmarks( + log, + dir, + identity, + root_pem, + BookmarkSource::null(), + shutdown, + ) + .await + } + + async fn start_with_bookmarks( + log: &Logger, + dir: &Utf8PathBuf, + identity: usize, + root_pem: &Utf8PathBuf, + bookmarks: BookmarkSource, + shutdown: &CancellationToken, ) -> Sled { let (peers, peers_rx) = watch::channel(BTreeSet::new()); let (addr, universe) = spawn_gossip( @@ -57,7 +81,8 @@ impl Sled { corpus(dir), localhost(), peers_rx, - seed_gossip(), + seed_gossip(&bookmarks).await, + bookmarks, shutdown.clone(), ) .await @@ -111,7 +136,7 @@ async fn jobs_gossip_between_sleds() { // The sleds converge on one universe, resetting the losing job manager. eventually("universe convergence", 120, async || { - a.universe.borrow().network() == b.universe.borrow().network() + a.universe.borrow().rumors.network() == b.universe.borrow().rumors.network() }) .await; @@ -200,3 +225,464 @@ async fn jobs_gossip_between_sleds() { shutdown.cancel(); } + +#[tokio::test] +async fn rejoining_replays_without_reexecuting() { + let (_tmp, dir) = pki("sush-replay-", 2); + let mut root = common::ephemeral_root(); + let root_pem = dir.join("job-root.pem"); + std::fs::write( + &root_pem, + pem_cert_chain(vec![root.cert().to_owned()]).unwrap(), + ) + .unwrap(); + + let log = test_logger("rejoining_replays_without_reexecuting"); + let shutdown = CancellationToken::new(); + + // Sled A runs a whole job before B exists. + let a = Sled::start(&log, &dir, 1, &root_pem, &shutdown).await; + let authn_a = fake_identity(&mut root).await; + let signer_nonce = SessionSignerNonce::random(); + let session_id = SessionId::compute( + a.mgr.own_baseboard(), + a.mgr.session_sush_nonce(), + signer_nonce, + ); + let session = Session::new(session_id); + a.mgr + .session_start(&authn_a, session_id, signer_nonce, true) + .await + .unwrap(); + let job_id = session.next_job_id(); + let job = sign_job(&mut root, job_id, session_id, "true").await; + a.mgr + .job_start( + &authn_a, + job, + JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }, + ) + .await + .unwrap(); + + // B joins later and receives the whole history as replay: it learns + // what happened, but the executed job must not run again here. + 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(); + let authn_b = fake_identity(&mut root).await; + eventually("A's history replays on B", 120, async || { + b.mgr.job_status(&authn_b, &job_id).await.is_ok_and(|map| { + map.get(&a.baseboard) + .is_some_and(|s| matches!(s, JobStatus::Stopped { .. })) + }) + }) + .await; + assert!( + !b.mgr + .job_status(&authn_b, &job_id) + .await + .unwrap() + .contains_key(&b.baseboard), + "replayed job executed on the joining sled" + ); + + // Live traffic still executes everywhere: a fresh session's job runs + // on both sleds. + let successor_nonce = SessionSignerNonce::random(); + let successor = SessionId::compute( + a.mgr.own_baseboard(), + a.mgr.session_sush_nonce(), + successor_nonce, + ); + let fresh = Session::new(successor); + a.mgr + .session_start(&authn_a, successor, successor_nonce, true) + .await + .unwrap(); + let live_job = fresh.next_job_id(); + let job = sign_job(&mut root, live_job, successor, "true").await; + a.mgr + .job_start( + &authn_a, + job, + JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }, + ) + .await + .unwrap(); + eventually("the live job runs on B too", 120, async || { + a.mgr + .job_status(&authn_a, &live_job) + .await + .is_ok_and(|map| { + map.get(&b.baseboard) + .is_some_and(|s| matches!(s, JobStatus::Stopped { .. })) + }) + }) + .await; + + // With B provably executing live jobs, the replayed one still never + // ran there. + assert!( + !b.mgr + .job_status(&authn_b, &job_id) + .await + .unwrap() + .contains_key(&b.baseboard), + "replayed job executed late on the joining sled" + ); + + shutdown.cancel(); +} + +#[tokio::test] +async fn interrupted_jobs_get_stopped() { + let (_tmp, dir) = pki("sush-interrupted-", 2); + let mut root = common::ephemeral_root(); + let root_pem = dir.join("job-root.pem"); + std::fs::write( + &root_pem, + pem_cert_chain(vec![root.cert().to_owned()]).unwrap(), + ) + .unwrap(); + + let log = test_logger("interrupted_jobs_get_stopped"); + let shutdown = CancellationToken::new(); + let a = Sled::start(&log, &dir, 1, &root_pem, &shutdown).await; + let authn_a = fake_identity(&mut root).await; + + // A previous life of sled 2 started a job and died before stopping + // it: the rack's history shows it running forever. + let job_id: JobId = "abandon-abandon-abandon-abandon-abandon-abandon-abandon-ability" + .parse() + .unwrap(); + let ghost = BaseboardId { + part_number: "sled".to_string(), + serial_number: "2".to_string(), + }; + a.universe.borrow().rumors.clone().send( + Message::Event( + ghost.clone(), + Event::Job(JobEvent::Start(job_id, Utc::now())), + ) + .into(), + ); + eventually("A records the orphaned start", 60, async || { + a.mgr.job_status(&authn_a, &job_id).await.is_ok_and(|map| { + map.get(&ghost) + .is_some_and(|s| matches!(s, JobStatus::Started { .. })) + }) + }) + .await; + + // Sled 2's next incarnation joins, finds its own job running in + // replayed history with no executor to ever stop it, and declares + // it interrupted for the whole rack to see. + 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("the orphan is interrupted", 120, async || { + a.mgr.job_status(&authn_a, &job_id).await.is_ok_and(|map| { + map.get(&ghost).is_some_and(|s| { + matches!( + s, + JobStatus::Error { + error: ProcessError::Interrupted, + .. + } + ) + }) + }) + }) + .await; + + // The job's genuine stop was in flight all along. When it lands, + // the real result supersedes the interrupted declaration. + let output = JobOutputState::default(); + a.universe.borrow().rumors.clone().send( + Message::Event( + ghost.clone(), + Event::Job(JobEvent::Stop(job_id, Utc::now(), Ok(0), output)), + ) + .into(), + ); + eventually("the late stop supersedes the interrupt", 120, async || { + a.mgr.job_status(&authn_a, &job_id).await.is_ok_and(|map| { + map.get(&ghost) + .is_some_and(|s| matches!(s, JobStatus::Stopped { result: Ok(0), .. })) + }) + }) + .await; + + shutdown.cancel(); +} + +#[tokio::test] +async fn stragglers_do_not_interrupt_live_jobs() { + let (_tmp, dir) = pki("sush-straggler-", 3); + let mut root = common::ephemeral_root(); + let root_pem = dir.join("job-root.pem"); + std::fs::write( + &root_pem, + pem_cert_chain(vec![root.cert().to_owned()]).unwrap(), + ) + .unwrap(); + + let log = test_logger("stragglers_do_not_interrupt_live_jobs"); + let shutdown = CancellationToken::new(); + + // A and C converge; C then holds a message A never sees. + let a = Sled::start(&log, &dir, 1, &root_pem, &shutdown).await; + let c = Sled::start(&log, &dir, 3, &root_pem, &shutdown).await; + a.peers.send(BTreeSet::from([c.addr])).unwrap(); + c.peers.send(BTreeSet::from([a.addr])).unwrap(); + eventually("A and C converge", 120, async || { + a.universe.borrow().rumors.network() == c.universe.borrow().rumors.network() + }) + .await; + a.peers.send(BTreeSet::new()).unwrap(); + c.peers.send(BTreeSet::new()).unwrap(); + let marooned = BaseboardId { + part_number: "sled".to_string(), + serial_number: "marooned".to_string(), + }; + c.universe + .borrow() + .rumors + .clone() + .send(Message::Event(marooned.clone(), Event::Version(VersionInfo::current())).into()); + // A also advances on its own side of the split, so the marooned + // message is genuinely concurrent with (not under) B's frontier. + let split_marker = BaseboardId { + part_number: "sled".to_string(), + serial_number: "split-marker".to_string(), + }; + a.universe + .borrow() + .rumors + .clone() + .send(Message::Event(split_marker, Event::Version(VersionInfo::current())).into()); + + // B joins through A alone, so C's message is concurrent with B's + // join frontier, and starts a live job that keeps running. + 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("B joins A", 120, async || { + a.universe.borrow().rumors.network() == b.universe.borrow().rumors.network() + }) + .await; + let authn_a = fake_identity(&mut root).await; + let signer_nonce = SessionSignerNonce::random(); + let session_id = SessionId::compute( + a.mgr.own_baseboard(), + a.mgr.session_sush_nonce(), + signer_nonce, + ); + let session = Session::new(session_id); + a.mgr + .session_start(&authn_a, session_id, signer_nonce, true) + .await + .unwrap(); + let job_id = session.next_job_id(); + let job = sign_job(&mut root, job_id, session_id, "sleep 30").await; + a.mgr + .job_start( + &authn_a, + job, + JobStartParams { + wait: JobWait::Start, + ..Default::default() + }, + ) + .await + .unwrap(); + eventually("the live job starts on B", 120, async || { + a.mgr.job_status(&authn_a, &job_id).await.is_ok_and(|map| { + map.get(&b.baseboard) + .is_some_and(|s| matches!(s, JobStatus::Started { .. })) + }) + }) + .await; + + // C reconnects. Its marooned message reaches B as replayed + // traffic and triggers a reap, which the live job must survive. + a.peers.send(BTreeSet::from([b.addr, c.addr])).unwrap(); + b.peers.send(BTreeSet::from([a.addr, c.addr])).unwrap(); + c.peers.send(BTreeSet::from([a.addr, b.addr])).unwrap(); + let authn_b = fake_identity(&mut root).await; + eventually("the marooned message reaches B", 120, async || { + b.mgr.versions().iter().any(|row| row.baseboard == marooned) + }) + .await; + assert!( + b.mgr.job_status(&authn_b, &job_id).await.is_ok_and(|map| { + map.get(&b.baseboard) + .is_some_and(|s| matches!(s, JobStatus::Started { .. })) + }), + "a straggler-triggered scan interrupted a live job" + ); + + shutdown.cancel(); +} + +#[tokio::test] +async fn bookmarks_survive_restart() { + let (_tmp, dir) = pki("sush-bookmark-", 2); + let mut root = common::ephemeral_root(); + let root_pem = dir.join("job-root.pem"); + std::fs::write( + &root_pem, + pem_cert_chain(vec![root.cert().to_owned()]).unwrap(), + ) + .unwrap(); + + let log = test_logger("bookmarks_survive_restart"); + let shutdown = CancellationToken::new(); + + // Sled A holds session history, so it wins every dominance contest. + let a = Sled::start(&log, &dir, 1, &root_pem, &shutdown).await; + let authn_a = fake_identity(&mut root).await; + let signer_nonce = SessionSignerNonce::random(); + let session_id = SessionId::compute( + a.mgr.own_baseboard(), + a.mgr.session_sush_nonce(), + signer_nonce, + ); + a.mgr + .session_start(&authn_a, session_id, signer_nonce, true) + .await + .unwrap(); + + // Sled 2 keeps its identity in a bookmark; joining records it. + let bookmark_dir = TempDir::with_prefix("sush-bookmark-").unwrap(); + let slot = Utf8PathBuf::from_path_buf(bookmark_dir.path().join("bookmark")).unwrap(); + let b_shutdown = CancellationToken::new(); + let b = Sled::start_with_bookmarks( + &log, + &dir, + 2, + &root_pem, + BookmarkSource::new(&log, vec![slot.clone()]), + &b_shutdown, + ) + .await; + a.peers.send(BTreeSet::from([b.addr])).unwrap(); + b.peers.send(BTreeSet::from([a.addr])).unwrap(); + eventually("the joining sled records its identity", 120, async || { + slot.as_std_path().exists() + }) + .await; + let before = std::fs::read(&slot).unwrap(); + + // The next incarnation reads the record back, rejoins, and + // advances it, reclaiming the previous life's identity. + b_shutdown.cancel(); + drop(b); + a.peers.send(BTreeSet::new()).unwrap(); + // Let the dead incarnation's tasks quiesce, as a real reboot + // would. Two live sources over one slot are the store's one + // forbidden misuse. + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + let b = Sled::start_with_bookmarks( + &log, + &dir, + 2, + &root_pem, + BookmarkSource::new(&log, vec![slot.clone()]), + &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; + eventually( + "the record advances past the previous life", + 120, + async || std::fs::read(&slot).unwrap() != before, + ) + .await; + + shutdown.cancel(); +} + +#[tokio::test] +async fn gossip_survives_bookmark_failure() { + let (_tmp, dir) = pki("sush-nobookmark-", 2); + let mut root = common::ephemeral_root(); + let root_pem = dir.join("job-root.pem"); + std::fs::write( + &root_pem, + pem_cert_chain(vec![root.cert().to_owned()]).unwrap(), + ) + .unwrap(); + + let log = test_logger("gossip_survives_bookmark_failure"); + let shutdown = CancellationToken::new(); + + let a = Sled::start(&log, &dir, 1, &root_pem, &shutdown).await; + let authn_a = fake_identity(&mut root).await; + let signer_nonce = SessionSignerNonce::random(); + let session_id = SessionId::compute( + a.mgr.own_baseboard(), + a.mgr.session_sush_nonce(), + signer_nonce, + ); + let session = Session::new(session_id); + a.mgr + .session_start(&authn_a, session_id, signer_nonce, true) + .await + .unwrap(); + + // Sled 2's bookmark points into a directory that does not exist. + // It sheds the bookmark and gossips anyway, stranding identities + // rather than the rack. + let b = Sled::start_with_bookmarks( + &log, + &dir, + 2, + &root_pem, + BookmarkSource::new(&log, vec![Utf8PathBuf::from("/nonexistent/sush/bookmark")]), + &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; + + // Live jobs still run on the degraded sled. + let job_id = session.next_job_id(); + let job = sign_job(&mut root, job_id, session_id, "true").await; + a.mgr + .job_start( + &authn_a, + job, + JobStartParams { + wait: JobWait::Stop, + ..Default::default() + }, + ) + .await + .unwrap(); + eventually("the job runs on the degraded sled", 120, async || { + a.mgr.job_status(&authn_a, &job_id).await.is_ok_and(|map| { + map.get(&b.baseboard) + .is_some_and(|s| matches!(s, JobStatus::Stopped { .. })) + }) + }) + .await; + + shutdown.cancel(); +} diff --git a/server/tests/gossip.rs b/server/tests/gossip.rs index 4e69750..a6dcab5 100644 --- a/server/tests/gossip.rs +++ b/server/tests/gossip.rs @@ -16,7 +16,8 @@ use slog::Logger; use tokio::sync::watch; use tokio_util::sync::CancellationToken; -use sush_server::gossip::spawn_gossip; +use sush_server::bookmark::{BookmarkSource, SushBookmark}; +use sush_server::gossip::{Universe, spawn_gossip}; use common::{corpus, eventually, gossip_config, localhost, pki, sprockets_config, test_logger}; @@ -24,14 +25,19 @@ struct Node { addr: SocketAddrV6, initial: Network, peers: watch::Sender>, - universe: watch::Receiver>, + universe: watch::Receiver>, shutdown: CancellationToken, } impl Node { async fn start(log: &Logger, dir: &Utf8PathBuf, identity: usize) -> Node { let shutdown = CancellationToken::new(); - let seed: Rumors = Peer::seed().into_rumors(); + let bookmarks = BookmarkSource::null(); + let seed: Rumors = Peer::seed() + .bookmark(bookmarks.next_handle()) + .await + .expect("a pristine seed never touches its bookmark") + .into_rumors(); let initial = seed.network(); let (peers, peers_rx) = watch::channel(BTreeSet::new()); let (addr, universe) = spawn_gossip( @@ -42,6 +48,7 @@ impl Node { localhost(), peers_rx, seed, + bookmarks, shutdown.clone(), ) .await @@ -56,11 +63,11 @@ impl Node { } fn network(&self) -> Network { - self.universe.borrow().network() + self.universe.borrow().rumors.network() } - fn rumors(&self) -> Rumors { - self.universe.borrow().clone() + fn rumors(&self) -> Rumors { + self.universe.borrow().rumors.clone() } fn contains(&self, message: &str) -> bool { diff --git a/server/tests/output/job-interrupted-event.bin b/server/tests/output/job-interrupted-event.bin new file mode 100644 index 0000000..cf36b59 Binary files /dev/null and b/server/tests/output/job-interrupted-event.bin differ diff --git a/sush.json b/sush.json index 0314e47..da8853c 100644 --- a/sush.json +++ b/sush.json @@ -1507,7 +1507,8 @@ "type": "string", "enum": [ "Unknown", - "InvalidCommand" + "InvalidCommand", + "Interrupted" ] }, { diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 3b8e9af..34a0130 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -34,7 +34,8 @@ use sush_common::jobs::{ }; use sush_common::keys::{EphemeralKey, KeyError, KeyId, KeyType, Signer as _, pem_cert_chain}; use sush_common::targets::{Cubbies, Target}; -use sush_server::gossip::isolated; +use sush_server::bookmark::BookmarkSource; +use sush_server::gossip::{Universe, isolated}; use sush_server::io::BATCH_OUTPUT_BUFFER_SIZE; use sush_server::messages::v0::{CertRequest, IdentityRequest, Message, Request, SessionRequest}; use sush_server::output::{JobOutputDir, OutputDirs}; @@ -578,7 +579,7 @@ async fn cubby_targets() { JobOutputDir::fixed(dir.path()), test_baseboard_id(), cubbies_rx, - isolated(seed_gossip()), + isolated(seed_gossip(&BookmarkSource::null()).await), &[root.cert().to_owned()], CancellationToken::new(), ) @@ -683,7 +684,7 @@ async fn root_certs_from_files() { JobOutputDir::fixed(dir.path()), test_baseboard_id(), no_cubbies(), - isolated(seed_gossip()), + isolated(seed_gossip(&BookmarkSource::null()).await), &[path], CancellationToken::new(), ) @@ -733,7 +734,7 @@ async fn bad_root_cert_files() { JobOutputDir::fixed(dir.path()), test_baseboard_id(), no_cubbies(), - isolated(seed_gossip()), + isolated(seed_gossip(&BookmarkSource::null()).await), &[path], CancellationToken::new(), ) @@ -763,7 +764,7 @@ async fn job_output_dir_moves() { JobOutputDir::new(rx_dirs), test_baseboard_id(), no_cubbies(), - isolated(seed_gossip()), + isolated(seed_gossip(&BookmarkSource::null()).await), &[root.cert().to_owned()], CancellationToken::new(), ) @@ -855,7 +856,9 @@ async fn universe_swap() { let log = test_logger(function_name!()); let dir = TempDir::with_prefix("sush-").unwrap(); let mut root = ephemeral_test_root(); - let (universe, universe_rx) = watch::channel(seed_gossip()); + let (universe, universe_rx) = watch::channel(Universe::genesis( + seed_gossip(&BookmarkSource::null()).await, + )); let mgr = JobManager::with_root_certs( log, PathIsolation::InsecureDisable, @@ -900,7 +903,11 @@ async fn universe_swap() { assert!(mgr.job_status(&authn, &job_id).await.is_ok()); // Migrate. The session and the job's history are gone. - universe.send(seed_gossip()).unwrap(); + universe + .send(Universe::genesis( + seed_gossip(&BookmarkSource::null()).await, + )) + .unwrap(); timeout(Duration::from_secs(30), async { while mgr.session(&authn).is_some() { sleep(Duration::from_millis(50)).await; @@ -998,7 +1005,7 @@ async fn cert_chain() { part_number: "test part".to_string(), serial_number: "0000".to_string(), }; - let gossip = isolated(seed_gossip()); + let gossip = isolated(seed_gossip(&BookmarkSource::null()).await); let shutdown = CancellationToken::new(); let mgr = JobManager::with_root_certs( log, @@ -1771,7 +1778,7 @@ async fn hostile_imports_cannot_displace() { part_number: "test part".to_string(), serial_number: "0000".to_string(), }; - let seed = seed_gossip(); + let seed = seed_gossip(&BookmarkSource::null()).await; let peer = seed.clone(); let shutdown = CancellationToken::new(); let mgr = JobManager::with_root_certs( @@ -1916,7 +1923,7 @@ async fn homonym_issuer_resolves_to_true_parent() { part_number: "test part".to_string(), serial_number: "0000".to_string(), }; - let seed = seed_gossip(); + let seed = seed_gossip(&BookmarkSource::null()).await; let peer = seed.clone(); let shutdown = CancellationToken::new(); let mgr = JobManager::with_root_certs( diff --git a/tests/src/test_utils.rs b/tests/src/test_utils.rs index 749fc55..aea150d 100644 --- a/tests/src/test_utils.rs +++ b/tests/src/test_utils.rs @@ -29,6 +29,7 @@ use sush_common::codephrases::Codephrase; use sush_common::jobs::{JobId, JobMode, JobStartRequest, SessionId, VerifiedJob}; use sush_common::keys::{EphemeralKey, KeyType, Signer}; use sush_common::targets::{Cubbies, Target}; +use sush_server::bookmark::BookmarkSource; use sush_server::executor::PathIsolation; use sush_server::gossip::isolated; use sush_server::output::{JobOutputDir, JobOutputFileStream}; @@ -227,7 +228,7 @@ pub async fn manager_test_root_and_peer( CancellationToken, ) { let dir = TempDir::with_prefix("sush-").unwrap(); - let seed = seed_gossip(); + let seed = seed_gossip(&BookmarkSource::null()).await; let peer = seed.clone(); let gossip = isolated(seed); let shutdown = CancellationToken::new();