From 40cd77eca3c611af665f30b9dc1f5dd651ca9b7b Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 27 Aug 2026 09:34:46 -0600 Subject: [PATCH 1/9] Don't execute jobs from the past A sled that joins an established universe receives its entire history, and until now executed every retained job as its causal chain became ready. The gossip manager now publishes the causal frontier of the set received at join alongside the rumors handle, and the state machine executes only jobs that arrive as strict causal descendants of that frontier. Co-Authored-By: Claude Mythos 5 --- server/src/gossip.rs | 44 ++++++++--- server/src/manager.rs | 6 +- server/src/state.rs | 153 +++++++++++++++++++++++++++--------- server/tests/distributed.rs | 121 +++++++++++++++++++++++++++- server/tests/gossip.rs | 8 +- tests/src/manager_tests.rs | 6 +- 6 files changed, 276 insertions(+), 62 deletions(-) diff --git a/server/src/gossip.rs b/server/src/gossip.rs index 93debfb..497811d 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, Network, Peer, Rumors, Ticks, Version}; use serde::Serialize; use serde::de::DeserializeOwned; use slog::{Logger, debug, info, o, warn}; @@ -65,11 +66,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 } @@ -102,7 +122,7 @@ pub async fn spawn_gossip( peers: watch::Receiver>, seed: Rumors, shutdown: CancellationToken, -) -> io::Result<(SocketAddrV6, watch::Receiver>)> +) -> io::Result<(SocketAddrV6, watch::Receiver>)> where T: DeserializeOwned + Serialize + Send + Sync + 'static, { @@ -131,11 +151,11 @@ pub fn spawn_gossip_manager( peers: watch::Receiver>, seed: Rumors, 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, @@ -173,7 +193,7 @@ struct Manager { transport: Transport, peers: watch::Receiver>, rumors: Rumors, - publish: watch::Sender>, + publish: watch::Sender>, drivers: JoinSet<(SocketAddr, Stopped)>, live: HashMap, dials: JoinSet, @@ -315,7 +335,11 @@ where 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()); + 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()); } 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/state.rs b/server/src/state.rs index 3bc239b..1d53d7e 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -35,6 +35,7 @@ use sush_common::targets::Cubbies; use sush_common::version::{VersionInfo, VersionMap}; use crate::executor::{Executor, PathIsolation}; +use crate::gossip::Universe; use crate::history::JobHistory; use crate::job::SocketSender; use crate::messages::v0::{ @@ -47,7 +48,8 @@ use crate::output::JobOutputDir; pub type AttachmentPoints = BTreeMap>>; pub type Certificates = BTreeMap; pub type GossipNetwork = Rumors; -pub type QueuedJobs = BTreeMap; +pub type GossipUniverse = Universe; +pub type QueuedJobs = BTreeMap; pub type RunningJobs = BTreeMap<(JobId, BaseboardId), DateTime>; /// Maximum certificate chain length. @@ -83,6 +85,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 +205,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 +220,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 +235,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, @@ -272,8 +293,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 +304,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 { + info!(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 +356,10 @@ 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. Messages at or concurrent with it are + /// replayed history: they rebuild state but never execute here. + join_frontier: Option, /// Message versions from newer builds, each warned about once. unknown_versions: BTreeSet, /// Build provenance by sled. @@ -343,6 +378,7 @@ impl State { own_baseboard: BaseboardId, root_certs: &[Certificate], session_sush_nonce: Arc>, + join_frontier: Option, ) -> Result { let certs = root_certs .iter() @@ -369,6 +405,7 @@ impl State { unknown_versions: Default::default(), identities: LruCache::new(MAX_REGISTERED_IDENTITIES), revoked_keys: LruCache::new(MAX_REVOKED_KEYS), + join_frontier, }; new.validate_certs(&roots); for root in &roots { @@ -663,6 +700,7 @@ impl State { &self.running, ); session.execute_ready_jobs( + log, &self.own_baseboard, &self.cubbies, &mut self.certs, @@ -714,6 +752,10 @@ 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 + .join_frontier + .as_ref() + .is_none_or(|frontier| incoming_version > frontier); match self.session.active_session() { Some(mut session) if session.session_id() == session_id => { session.enqueue_job( @@ -725,8 +767,10 @@ impl State { signed.clone(), params.clone(), actor, + !live, ); session.execute_ready_jobs( + log, &self.own_baseboard, &self.cubbies, &mut self.certs, @@ -743,7 +787,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( @@ -904,6 +949,34 @@ 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); + if let Some(rumors) = rumors { + debug!(log, "sending error to gossip network"; "error" => ?error); + rumors.send(Message::Event(own_baseboard.clone(), Event::Error(error)).into()); + } + } + }); +} + /// 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 @@ -938,7 +1011,7 @@ impl StateManager { own_baseboard: BaseboardId, mut requests: R, mut cubbies: watch::Receiver, - universe: watch::Receiver, + universe: watch::Receiver, roots: &[Certificate], session_sush_nonce: Arc>, shutdown: CancellationToken, @@ -946,22 +1019,29 @@ 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(); + let Universe { + rumors: initial, + frontier, + } = universe.borrow().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, + )?; + 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")), @@ -1051,21 +1131,15 @@ 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()); - } - } - }); + apply_message( + &log, + &tx_state, + &mut executor, + gossip.as_ref().map(|(rumors, _)| rumors), + &own_baseboard, + &version, + &message, + ); }, }, @@ -1095,19 +1169,20 @@ 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(); // 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(), diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index 74e4314..94bdda1 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -24,7 +24,7 @@ use sush_common::version::VersionInfo; use sush_server::executor::PathIsolation; use sush_server::gossip::spawn_gossip; 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 +34,7 @@ use common::{ struct Sled { mgr: JobManager, - universe: watch::Receiver, + universe: watch::Receiver, peers: watch::Sender>, addr: SocketAddrV6, baseboard: BaseboardId, @@ -111,7 +111,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 +200,118 @@ 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(); +} diff --git a/server/tests/gossip.rs b/server/tests/gossip.rs index 4e69750..8fe975d 100644 --- a/server/tests/gossip.rs +++ b/server/tests/gossip.rs @@ -16,7 +16,7 @@ use slog::Logger; use tokio::sync::watch; use tokio_util::sync::CancellationToken; -use sush_server::gossip::spawn_gossip; +use sush_server::gossip::{Universe, spawn_gossip}; use common::{corpus, eventually, gossip_config, localhost, pki, sprockets_config, test_logger}; @@ -24,7 +24,7 @@ struct Node { addr: SocketAddrV6, initial: Network, peers: watch::Sender>, - universe: watch::Receiver>, + universe: watch::Receiver>, shutdown: CancellationToken, } @@ -56,11 +56,11 @@ impl Node { } fn network(&self) -> Network { - self.universe.borrow().network() + self.universe.borrow().rumors.network() } fn rumors(&self) -> Rumors { - self.universe.borrow().clone() + self.universe.borrow().rumors.clone() } fn contains(&self, message: &str) -> bool { diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index 3b8e9af..a2b7f4b 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -34,7 +34,7 @@ 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::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}; @@ -855,7 +855,7 @@ 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())); let mgr = JobManager::with_root_certs( log, PathIsolation::InsecureDisable, @@ -900,7 +900,7 @@ 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())).unwrap(); timeout(Duration::from_secs(30), async { while mgr.session(&authn).is_some() { sleep(Duration::from_millis(50)).await; From 895f616d495f265a6b58b93eaa8c8b4deb465531 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 27 Aug 2026 09:40:40 -0600 Subject: [PATCH 2/9] Declare jobs from previous lives interrupted A sled that reboots mid-job leaves the rack believing the job runs forever: the executor died with the process, so no stop event ever comes. Now, when replayed history at a universe join shows jobs running on our own baseboard that this incarnation is not actually running, we gossip an error event for each with the new ProcessError::Interrupted, closing them out honestly for every seat in the rack (sush#9). The scan runs only at local quiescence, after draining every message already delivered, so a job whose terminal event is present is never falsely interrupted. A stop that is still in flight elsewhere in the rack can race the declaration; the audit trail then carries both, and honestly. Jobs this incarnation really is running when its universe migrates are excluded: they continue, and their events land in the new universe. Co-Authored-By: Claude Mythos 5 --- common/src/jobs.rs | 2 + server/src/messages.rs | 9 +- server/src/state.rs | 129 +++++++++++++++++- server/tests/distributed.rs | 68 ++++++++- server/tests/output/job-interrupted-event.bin | Bin 0 -> 143 bytes sush.json | 3 +- 6 files changed, 205 insertions(+), 6 deletions(-) create mode 100644 server/tests/output/job-interrupted-event.bin 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/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 1d53d7e..3e0727d 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; @@ -442,6 +442,15 @@ 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() + } + pub fn get_job_status(&self, job_id: &JobId) -> Option<&JobStatusMap> { self.history.get_job_status(job_id) } @@ -977,6 +986,63 @@ fn apply_message( }); } +/// 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 historical jobs interrupted, excluding `survivors` +/// (jobs this incarnation is really running across a universe swap) +/// and jobs already so declared. Call only at local quiescence +/// (after [`drain_ready`]), so a job whose terminal event is +/// already here is never falsely interrupted. +fn interrupt_orphans( + log: &Logger, + tx_state: &watch::Sender, + rumors: Option<&GossipNetwork>, + own_baseboard: &BaseboardId, + survivors: &BTreeSet, + interrupted: &mut BTreeSet, +) { + let mut orphans = tx_state.borrow().own_running_jobs(); + orphans.retain(|job_id| !survivors.contains(job_id) && !interrupted.contains(job_id)); + for job_id in orphans { + interrupted.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 @@ -1036,7 +1102,7 @@ impl StateManager { own_baseboard.clone(), roots, session_sush_nonce.clone(), - frontier, + frontier.clone(), )?; initial_state.cubbies = cubbies.borrow_and_update().clone(); let (tx_state, rx_state) = watch::channel(initial_state); @@ -1061,6 +1127,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; `interrupted` are jobs of + // ours the replay showed running, already declared dead. + let mut frontier = frontier; + let mut survivors: BTreeSet = BTreeSet::new(); + let mut interrupted: BTreeSet = BTreeSet::new(); + // Announce our build. if let Some((rumors, _)) = &gossip { rumors.send( @@ -1131,6 +1206,8 @@ impl StateManager { break; } Some((version, message)) => { + // 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, @@ -1140,6 +1217,29 @@ impl StateManager { &version, &message, ); + // Replayed history can show jobs of ours + // running that died with a previous life. + // Drain to local quiescence first, so a + // job whose terminal event is already here + // is never declared interrupted. + if !live { + drain_ready( + &log, + &mut causal_messages, + &tx_state, + &mut executor, + gossip.as_ref().map(|(rumors, _)| rumors), + &own_baseboard, + ); + interrupt_orphans( + &log, + &tx_state, + gossip.as_ref().map(|(rumors, _)| rumors), + &own_baseboard, + &survivors, + &mut interrupted, + ); + } }, }, @@ -1171,6 +1271,9 @@ impl StateManager { let fresh = universe.borrow_and_update().clone(); info!(log, "gossip universe changed, resetting state"; "network" => %fresh.rumors.network()); causal_messages = fresh.rumors.causal_messages(); + survivors = tx_state.borrow().own_running_jobs(); + interrupted = BTreeSet::new(); + frontier = fresh.frontier.clone(); // TODO: re-inject local job state (policy pending). tx_state.send_modify(|state| { *state = State::new( @@ -1190,6 +1293,26 @@ impl StateManager { ) .into(), ); + // The set received at join is already local: drain + // it, then declare interrupted any job of ours the + // replayed history still shows running. A previous + // life started them; nothing will ever stop them. + drain_ready( + &log, + &mut causal_messages, + &tx_state, + &mut executor, + Some(&*rumors), + &own_baseboard, + ); + interrupt_orphans( + &log, + &tx_state, + Some(&*rumors), + &own_baseboard, + &survivors, + &mut interrupted, + ); } } }), diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index 94bdda1..e829772 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -16,13 +16,15 @@ 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, 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::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::GossipUniverse; use sush_server::{JobManager, seed_gossip}; @@ -315,3 +317,67 @@ async fn rejoining_replays_without_reexecuting() { 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; + + shutdown.cancel(); +} diff --git a/server/tests/output/job-interrupted-event.bin b/server/tests/output/job-interrupted-event.bin new file mode 100644 index 0000000000000000000000000000000000000000..cf36b59e866e43cd3bb9871619d0391f29d9c44b GIT binary patch literal 143 zcmZ3O6lSn6)wL`&ucT>Fc0poMNqk;uZc=Jdwxyx5t^o)bTILp~7G);pz+{|)d`*mu zj0_Bn4HhPQ Date: Thu, 27 Aug 2026 15:29:25 -0600 Subject: [PATCH 3/9] Add the gossip bookmark store A rumors bookmark persists a peer's identity so a restarted sled reclaims it instead of stranding it. We owe rumors raw byte storage with two guarantees: stores commit atomically, and a load never returns a record older than the newest store we acknowledged, since a stale record claims coverage the identity already transmitted past and corrupts causality on reclamation, whereas a lost record merely strands. Records live in a small sequence-numbered envelope, one file per boot M.2 slot. Loads read every slot and the highest sequence wins, tolerating boot-device flips and disk swaps; stores go only to the winning slot, so one disk's health is in the write path rather than two. A BookmarkSource hands out one handle per peer: minting a handle supersedes all earlier ones, so a straggling store from an abandoned universe cannot clobber its successor's record. Shed handles persist nothing, for peers that must keep gossiping after their storage failed, and the null source persists nothing at all, for the standalone server and tests. Co-Authored-By: Claude Mythos 5 --- Cargo.lock | 1 + server/Cargo.toml | 1 + server/src/bookmark.rs | 447 +++++++++++++++++++++++++++++++++++++++++ server/src/lib.rs | 1 + 4 files changed, 450 insertions(+) create mode 100644 server/src/bookmark.rs diff --git a/Cargo.lock b/Cargo.lock index ac94b0c..99af468 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4925,6 +4925,7 @@ name = "sush-server" version = "0.1.0" dependencies = [ "async-trait", + "atomicwrites", "attest-mock", "bytes", "bytesize", 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..fa073cb --- /dev/null +++ b/server/src/bookmark.rs @@ -0,0 +1,447 @@ +// 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"; + +/// 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 fenced handle per +/// peer. +#[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, +} + +#[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. + 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()), + }), + } + } + + /// 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 a real one, but without incrementing + /// the generation. + fn probe_handle(&self) -> SushBookmark { + SushBookmark { + shared: self.shared.clone(), + generation: 0, + shed: false, + } + } + + /// Does the storage look reasonable? + pub async fn probe(&self) -> Result<(), BookmarkIoError> { + if self.shared.slots.is_empty() { + return Ok(()); + } + let usable = match self.probe_handle().load().await { + Ok(_) => { + if self.shared.slots.iter().any(|path| { + path.parent() + .is_some_and(|parent| parent.as_std_path().is_dir()) + }) { + 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 an envelope into its sequence number and record. + fn parse(bytes: &[u8]) -> Option<(u64, Vec)> { + let payload = bytes.strip_prefix(MAGIC)?; + let (seq, record) = payload.split_first_chunk::<8>()?; + Some((u64::from_be_bytes(*seq), record.to_vec())) + } +} + +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); + } + 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)) => { + 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: a cancelled write may + // still land and must be outnumbered. + state.seq += 1; + let mut envelope = Vec::with_capacity(MAGIC.len() + 8 + record.len()); + envelope.extend_from_slice(MAGIC); + envelope.extend_from_slice(&state.seq.to_be_bytes()); + envelope.extend_from_slice(&record); + + let target = path.clone(); + let written = spawn_blocking(move || { + AtomicFile::new(target, 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 + } + }) + }) + .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 { + let mut bytes = MAGIC.to_vec(); + bytes.extend_from_slice(&seq.to_be_bytes()); + bytes.extend_from_slice(record); + bytes + } + + 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 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/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; From 92b970f5035808d9567420541b49ae799487499e Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 27 Aug 2026 15:29:38 -0600 Subject: [PATCH 4/9] Persist gossip identities across restarts Every gossip peer now carries a bookmark handle. The seed's is attached at creation, where a pristine peer costs nothing; a migration mints a fresh handle for the joined peer, superseding the abandoned universe's. Identities recorded by previous incarnations are reclaimed at the first gossip after a migration returns us to the universe that knew them. Persistence must never cost availability. A bookmark failure inside a session aborts that session before any wire traffic, so a seed on bad storage could never even learn it lost dominance: probe the storage at seeding and shed the bookmark up front. A join whose identity cannot be persisted likewise sheds and gossips on. Both degradations merely strand identities, which is what every restart did before this commit. A store that fails after a successful join still stops that sled's gossip until its next migration or restart; the planned rumors API for shedding a live peer's bookmark is the remaining fix. Co-Authored-By: Claude Mythos 5 --- server/src/gossip.rs | 77 +++++++++++----- server/src/main.rs | 3 +- server/src/state.rs | 23 ++++- server/tests/distributed.rs | 174 +++++++++++++++++++++++++++++++++++- server/tests/gossip.rs | 11 ++- tests/src/manager_tests.rs | 25 ++++-- tests/src/test_utils.rs | 3 +- 7 files changed, 278 insertions(+), 38 deletions(-) diff --git a/server/src/gossip.rs b/server/src/gossip.rs index 497811d..2beeb5b 100644 --- a/server/src/gossip.rs +++ b/server/src/gossip.rs @@ -29,7 +29,7 @@ use std::net::{SocketAddr, SocketAddrV6}; use std::time::Duration; use futures::StreamExt as _; -use rumors::{Error, Network, Peer, Rumors, Ticks, Version}; +use rumors::{Error, Joined, Network, Peer, Rumors, Ticks, Version}; use serde::Serialize; use serde::de::DeserializeOwned; use slog::{Logger, debug, info, o, warn}; @@ -42,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. @@ -70,14 +71,14 @@ impl Default for GossipConfig { #[derive(Clone, Debug)] pub struct Universe { /// The gossiped set. - pub rumors: Rumors, + 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 { + pub fn genesis(rumors: Rumors) -> Self { Self { rumors, frontier: None, @@ -88,7 +89,7 @@ impl Universe { /// 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> { +pub fn isolated(seed: Rumors) -> watch::Receiver> { let (_tx, rx) = watch::channel(Universe::genesis(seed)); rx } @@ -120,7 +121,8 @@ 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>)> where @@ -136,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)) } @@ -144,12 +146,14 @@ 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> where @@ -163,6 +167,7 @@ where transport, peers, rumors: seed, + bookmarks, publish, drivers: JoinSet::new(), live: HashMap::new(), @@ -192,7 +197,8 @@ struct Manager { endpoint: Endpoint, transport: Transport, peers: watch::Receiver>, - rumors: Rumors, + rumors: Rumors, + bookmarks: BookmarkSource, publish: watch::Sender>, drivers: JoinSet<(SocketAddr, Stopped)>, live: HashMap, @@ -325,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(); @@ -332,28 +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 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()); + 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, { 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/state.rs b/server/src/state.rs index 3e0727d..a772118 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -34,6 +34,7 @@ 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; @@ -47,7 +48,7 @@ use crate::output::JobOutputDir; pub type AttachmentPoints = BTreeMap>>; pub type Certificates = BTreeMap; -pub type GossipNetwork = Rumors; +pub type GossipNetwork = Rumors; pub type GossipUniverse = Universe; pub type QueuedJobs = BTreeMap; pub type RunningJobs = BTreeMap<(JobId, BaseboardId), DateTime>; @@ -1049,8 +1050,24 @@ fn interrupt_orphans( /// 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; 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)] diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index e829772..22c2a7a 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -22,6 +22,7 @@ use sush_common::jobs::{JobId, JobStatus, ProcessError, Session, SessionId, Sess 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}; @@ -50,6 +51,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( @@ -59,7 +79,8 @@ impl Sled { corpus(dir), localhost(), peers_rx, - seed_gossip(), + seed_gossip(&bookmarks).await, + bookmarks, shutdown.clone(), ) .await @@ -381,3 +402,154 @@ async fn interrupted_jobs_get_stopped() { 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 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 8fe975d..a6dcab5 100644 --- a/server/tests/gossip.rs +++ b/server/tests/gossip.rs @@ -16,6 +16,7 @@ use slog::Logger; use tokio::sync::watch; use tokio_util::sync::CancellationToken; +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}; @@ -31,7 +32,12 @@ struct Node { 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 @@ -59,7 +66,7 @@ impl Node { self.universe.borrow().rumors.network() } - fn rumors(&self) -> Rumors { + fn rumors(&self) -> Rumors { self.universe.borrow().rumors.clone() } diff --git a/tests/src/manager_tests.rs b/tests/src/manager_tests.rs index a2b7f4b..34a0130 100644 --- a/tests/src/manager_tests.rs +++ b/tests/src/manager_tests.rs @@ -34,6 +34,7 @@ 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::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}; @@ -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(Universe::genesis(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(Universe::genesis(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(); From 9ed1c3d1703496ea33cf8a780521a7025af8e7dd Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 27 Aug 2026 16:21:13 -0600 Subject: [PATCH 5/9] Interrupt only jobs from previous lives The adversarial review found the orphan scan drawing candidates from every job running on our baseboard, so a replayed straggler arriving while a live job ran would declare that live job interrupted, break its attachment, and discard its eventual result. Candidates now come only from start events that themselves arrived as replayed history: a job this incarnation started can never be a candidate, no matter when the scan runs. Three more review findings ride along. A genuine stop and an interrupted declaration now converge to the real result in either arrival order (errors no longer displace terminal statuses, and a stop supersedes an interrupted error), instead of the interrupt winning both races. A job we declared interrupted no longer counts as a survivor at the next universe swap, so back-to-back migrations re-declare it rather than exempting it forever. And replayed messages no longer mint local cancelled statuses or re-gossip concurrent-session errors, which grew the set a little more on every rejoin by every sled. Co-Authored-By: Claude Mythos 5 --- server/src/state.rs | 114 +++++++++++++++++++++++++----- server/tests/distributed.rs | 135 +++++++++++++++++++++++++++++++++++- 2 files changed, 230 insertions(+), 19 deletions(-) diff --git a/server/src/state.rs b/server/src/state.rs index a772118..48c9052 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -270,8 +270,14 @@ impl<'a> SessionGuard<'a> { own_baseboard: &BaseboardId, history: &mut JobHistory, running: &RunningJobs, + replayed: bool, ) { self.queued_jobs.remove(job_id); + // A replayed cancellation keeps the queue and chain honest but + // records no local status: whatever happened is in the history. + if replayed { + return; + } history.transition_job_status( job_id, own_baseboard, @@ -315,7 +321,7 @@ impl<'a> SessionGuard<'a> { let job_id = request.payload().job_id().to_owned(); if request.payload().runs_on(own_baseboard, cubbies) { if replayed { - info!(log, "not executing replayed job"; "job_id" => %job_id); + warn!(log, "not executing replayed job"; "job_id" => %job_id); } else if history .get_job_status(&job_id) .map(|status| { @@ -361,6 +367,10 @@ pub struct State { /// rather than seeded it. Messages at or concurrent with it are /// replayed history: they rebuild state but never execute here. join_frontier: Option, + /// Jobs whose start event on our own baseboard arrived as replayed + /// history: a previous life started them, so no executor of ours + /// will ever stop them. Cleared per job by a terminal event. + replayed_started: BTreeSet, /// Message versions from newer builds, each warned about once. unknown_versions: BTreeSet, /// Build provenance by sled. @@ -407,6 +417,7 @@ impl State { identities: LruCache::new(MAX_REGISTERED_IDENTITIES), revoked_keys: LruCache::new(MAX_REVOKED_KEYS), join_frontier, + replayed_started: Default::default(), }; new.validate_certs(&roots); for root in &roots { @@ -452,6 +463,29 @@ impl State { .collect() } + /// Jobs a previous life of this server started and left running: + /// their start events arrived as replayed history and no terminal + /// event has. Jobs this incarnation started never appear here. + pub fn orphaned_jobs(&self) -> BTreeSet { + self.replayed_started + .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) } @@ -689,6 +723,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 { @@ -708,6 +743,7 @@ impl State { &self.own_baseboard, &mut self.history, &self.running, + replayed, ); session.execute_ready_jobs( log, @@ -762,10 +798,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 - .join_frontier - .as_ref() - .is_none_or(|frontier| incoming_version > frontier); + let live = self.is_live(incoming_version); match self.session.active_session() { Some(mut session) if session.session_id() == session_id => { session.enqueue_job( @@ -813,6 +846,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, @@ -820,6 +854,7 @@ impl State { &self.own_baseboard, &mut self.history, &self.running, + replayed, ); } } @@ -872,6 +907,12 @@ 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 are ever declared + // interrupted. + if *baseboard_id == self.own_baseboard && !self.is_live(incoming_version) { + self.replayed_started.insert(*job_id); + } self.history.set_job_status( job_id, baseboard_id, @@ -888,6 +929,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.replayed_started.remove(job_id); } self.running.remove(&(*job_id, baseboard_id.clone())); self.history.transition_job_status( @@ -904,6 +946,21 @@ impl State { output: output.clone(), }) } + // An interrupted declaration raced this + // genuine stop; the real result wins, in + // either arrival order. The start time + // died with the interrupted incarnation. + 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(), @@ -917,17 +974,26 @@ 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.replayed_started.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: + // a genuine 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, ); @@ -979,7 +1045,11 @@ fn apply_message( tx_state.send_modify(|state| { if let Err(error) = state.update(log, executor, version, message) { error!(log, "state update failed"; "error" => ?error); - if let Some(rumors) = rumors { + // Replayed messages already had their say: re-gossiping + // their errors would grow the set on every rejoin. + 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()); } @@ -1012,9 +1082,11 @@ fn drain_ready( /// Declare historical jobs interrupted, excluding `survivors` /// (jobs this incarnation is really running across a universe swap) -/// and jobs already so declared. Call only at local quiescence -/// (after [`drain_ready`]), so a job whose terminal event is -/// already here is never falsely interrupted. +/// and jobs already so declared. Candidates come only from replayed +/// start events, so a job this incarnation is running is never a +/// candidate no matter when the scan runs. Call only at local +/// quiescence (after [`drain_ready`]), so a job whose terminal event +/// is already here is never falsely interrupted. fn interrupt_orphans( log: &Logger, tx_state: &watch::Sender, @@ -1023,7 +1095,7 @@ fn interrupt_orphans( survivors: &BTreeSet, interrupted: &mut BTreeSet, ) { - let mut orphans = tx_state.borrow().own_running_jobs(); + let mut orphans = tx_state.borrow().orphaned_jobs(); orphans.retain(|job_id| !survivors.contains(job_id) && !interrupted.contains(job_id)); for job_id in orphans { interrupted.insert(job_id); @@ -1094,7 +1166,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, @@ -1108,10 +1180,12 @@ impl StateManager { // 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. + // `borrow_and_update` marks the value seen: a migration that + // landed before we subscribed must not replay as a swap. let Universe { rumors: initial, frontier, - } = universe.borrow().clone(); + } = universe.borrow_and_update().clone(); let mut causal_messages = initial.causal_messages(); // We report our current state through a watch channel. @@ -1288,7 +1362,11 @@ impl StateManager { let fresh = universe.borrow_and_update().clone(); info!(log, "gossip universe changed, resetting state"; "network" => %fresh.rumors.network()); causal_messages = fresh.rumors.causal_messages(); + // A job we declared interrupted 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| !interrupted.contains(job_id)); interrupted = BTreeSet::new(); frontier = fresh.frontier.clone(); // TODO: re-inject local job state (policy pending). diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index 22c2a7a..eb4fc82 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -18,7 +18,9 @@ use tokio_util::sync::CancellationToken; use chrono::Utc; use sush_api::{JobStartParams, JobWait}; -use sush_common::jobs::{JobId, JobStatus, ProcessError, 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; @@ -400,6 +402,133 @@ async fn interrupted_jobs_get_stopped() { }) .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-classified + // traffic and triggers the orphan scan. 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(); } @@ -457,6 +586,10 @@ async fn bookmarks_survive_restart() { b_shutdown.cancel(); drop(b); a.peers.send(BTreeSet::new()).unwrap(); + // Let the dead incarnation's tasks quiesce: a real reboot does, + // and 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, From 8e283891712e6c7796f33a0900fcd68d0ca299b6 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 27 Aug 2026 16:21:24 -0600 Subject: [PATCH 6/9] Commit bookmark records in sequence order The adversarial review found the one write path the atomic rename does not serialize: a store future dropped at its await (a migrating manager aborts session drivers; a join can time out) detaches the blocking write, whose rename then lands on top of a newer record that already returned Ok. Renames now go through a commit fence: a write lands only if its sequence number exceeds everything this process has committed, so a straggler loses instead of clobbering. The envelope also gains a digest over the sequence number and record, since the whole safety argument rides on an integer that was previously trusted straight off the disk. Loads no longer regress the in-memory sequence reservation or serve superseded handles, the probe proves a slot writable by writing instead of guessing from directory metadata, and a bookmark failure inside a gossip session now warns loudly: it stops every later session at the persist gate, so unlike routine link churn it must not hide at debug level. Co-Authored-By: Claude Mythos 5 --- server/src/bookmark.rs | 193 +++++++++++++++++++++++++++++++++-------- server/src/gossip.rs | 7 ++ 2 files changed, 164 insertions(+), 36 deletions(-) diff --git a/server/src/bookmark.rs b/server/src/bookmark.rs index fa073cb..91fe82f 100644 --- a/server/src/bookmark.rs +++ b/server/src/bookmark.rs @@ -43,6 +43,20 @@ use tokio::task::spawn_blocking; /// 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; the +/// whole safety argument rides on that integer. +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 { @@ -81,6 +95,12 @@ struct SharedStore { /// 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 this process, and + /// the lock every rename takes. A store future dropped mid-write + /// leaves a detached blocking task whose rename would otherwise + /// land *after* a newer store's; renames commit in sequence order + /// or not at all. + committed: std::sync::Mutex, } #[derive(Debug, Default)] @@ -93,6 +113,14 @@ struct StoreState { impl BookmarkSource { /// A source persisting to `slots`, each on its own device. + /// + /// The contract for integrators: construct exactly one source per + /// slot set per process (the fences and sequence numbers that keep + /// the store safe live inside it), and feed that same source to + /// both [`seed_gossip`](crate::seed_gossip) and + /// [`spawn_gossip`](crate::gossip::spawn_gossip). The caller + /// creates the parent directories, writable by this server's user, + /// one per boot M.2; the record files are created and owned here. pub fn new(log: &Logger, slots: Vec) -> Self { Self { shared: Arc::new(SharedStore { @@ -100,6 +128,7 @@ impl BookmarkSource { slots, generation: AtomicU64::new(0), state: Mutex::new(StoreState::default()), + committed: std::sync::Mutex::new(0), }), } } @@ -130,27 +159,34 @@ impl BookmarkSource { } } - /// A probing handle: reads like a real one, but without incrementing - /// the generation. + /// A probing handle: reads like the current peer's, but without + /// superseding anything. fn probe_handle(&self) -> SushBookmark { SushBookmark { shared: self.shared.clone(), - generation: 0, + generation: self.shared.generation.load(Ordering::SeqCst), shed: false, } } - /// Does the storage look reasonable? + /// 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(_) => { - if self.shared.slots.iter().any(|path| { - path.parent() - .is_some_and(|parent| parent.as_std_path().is_dir()) - }) { + 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) @@ -179,12 +215,45 @@ impl SushBookmark { self.generation < self.shared.generation.load(Ordering::SeqCst) } - /// Split an envelope into its sequence number and record. + /// 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, record) = payload.split_first_chunk::<8>()?; - Some((u64::from_be_bytes(*seq), record.to_vec())) + 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. Runs on the blocking pool; the lock is +/// the commit point, so a straggling write detached from a dropped +/// store future cannot land on top of the newer record that beat it. +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 { @@ -198,6 +267,9 @@ impl Bookmark for SushBookmark { 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; @@ -229,8 +301,13 @@ impl Bookmark for SushBookmark { } match newest { Some((seq, home, record)) => { - state.home = Some(home); - state.seq = seq; + // Never regress: a reserved sequence number outranks a + // re-read of the disk, or a cancelled write's straggler + // could 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: @@ -275,27 +352,15 @@ impl Bookmark for SushBookmark { // Reserve the sequence number first: a cancelled write may // still land and must be outnumbered. state.seq += 1; - let mut envelope = Vec::with_capacity(MAGIC.len() + 8 + record.len()); - envelope.extend_from_slice(MAGIC); - envelope.extend_from_slice(&state.seq.to_be_bytes()); - envelope.extend_from_slice(&record); + let seq = state.seq; + let envelope = Self::envelope(seq, &record); + let shared = self.shared.clone(); let target = path.clone(); - let written = spawn_blocking(move || { - AtomicFile::new(target, 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 - } - }) - }) - .await - .map_err(|join| io::Error::other(join.to_string())) - .and_then(|result| result); + 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(()) => { @@ -337,10 +402,7 @@ mod test { } fn envelope(seq: u64, record: &[u8]) -> Vec { - let mut bytes = MAGIC.to_vec(); - bytes.extend_from_slice(&seq.to_be_bytes()); - bytes.extend_from_slice(record); - bytes + SushBookmark::envelope(seq, record) } async fn read_back(handle: &SushBookmark) -> Option> { @@ -425,6 +487,65 @@ mod test { )); } + /// 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] diff --git a/server/src/gossip.rs b/server/src/gossip.rs index 2beeb5b..68fb2ee 100644 --- a/server/src/gossip.rs +++ b/server/src/gossip.rs @@ -417,6 +417,13 @@ where Stopped::Failed }; } + // A bookmark failure stops every future session too (the + // persist gate runs before any wire traffic), so unlike + // routine link churn it must be loud. + 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; From c8877830a8c2e3686f3416a4391caa036681855c Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 27 Aug 2026 16:47:29 -0600 Subject: [PATCH 7/9] Call jobs from previous lives zombies A job whose process died with a previous incarnation but whose bookkeeping says it still runs is exactly what Unix calls a zombie, and recording its death is reaping. Rename replayed_started, orphaned_jobs, and interrupt_orphans accordingly, and sweep the recently added comments for style: no more spliced clauses, one home per argument, and the supersede vocabulary instead of fences. Co-Authored-By: Claude Mythos 5 --- server/src/bookmark.rs | 44 ++++++------- server/src/gossip.rs | 6 +- server/src/state.rs | 125 ++++++++++++++++-------------------- server/tests/distributed.rs | 10 +-- 4 files changed, 85 insertions(+), 100 deletions(-) diff --git a/server/src/bookmark.rs b/server/src/bookmark.rs index 91fe82f..7d5142d 100644 --- a/server/src/bookmark.rs +++ b/server/src/bookmark.rs @@ -45,8 +45,7 @@ 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; the -/// whole safety argument rides on that integer. +/// damaged sequence number from silently reordering the slots. const SEQ_LEN: usize = 8; const DIGEST_LEN: usize = 32; @@ -76,8 +75,8 @@ pub enum BookmarkIoError { Fenced, } -/// This server's bookmark storage. Hands out one fenced handle per -/// peer. +/// This server's bookmark storage. Hands out one handle per peer, +/// each superseding the last. #[derive(Clone, Debug)] pub struct BookmarkSource { shared: Arc, @@ -95,11 +94,9 @@ struct SharedStore { /// 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 this process, and - /// the lock every rename takes. A store future dropped mid-write - /// leaves a detached blocking task whose rename would otherwise - /// land *after* a newer store's; renames commit in sequence order - /// or not at all. + /// 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, } @@ -114,13 +111,12 @@ struct StoreState { impl BookmarkSource { /// A source persisting to `slots`, each on its own device. /// - /// The contract for integrators: construct exactly one source per - /// slot set per process (the fences and sequence numbers that keep - /// the store safe live inside it), and feed that same source to - /// both [`seed_gossip`](crate::seed_gossip) and - /// [`spawn_gossip`](crate::gossip::spawn_gossip). The caller - /// creates the parent directories, writable by this server's user, - /// one per boot M.2; the record files are created and owned here. + /// 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 { @@ -236,9 +232,9 @@ impl SushBookmark { } /// Rename `envelope` into place iff `seq` is newer than everything -/// committed by this process. Runs on the blocking pool; the lock is -/// the commit point, so a straggling write detached from a dropped -/// store future cannot land on top of the newer record that beat it. +/// 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 { @@ -301,9 +297,9 @@ impl Bookmark for SushBookmark { } match newest { Some((seq, home, record)) => { - // Never regress: a reserved sequence number outranks a - // re-read of the disk, or a cancelled write's straggler - // could collide with a fresh reservation. + // 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; @@ -349,8 +345,8 @@ impl Bookmark for SushBookmark { }; let path = self.shared.slots[home].clone(); - // Reserve the sequence number first: a cancelled write may - // still land and must be outnumbered. + // 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); diff --git a/server/src/gossip.rs b/server/src/gossip.rs index 68fb2ee..6fca713 100644 --- a/server/src/gossip.rs +++ b/server/src/gossip.rs @@ -417,9 +417,9 @@ where Stopped::Failed }; } - // A bookmark failure stops every future session too (the - // persist gate runs before any wire traffic), so unlike - // routine link churn it must be loud. + // 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; diff --git a/server/src/state.rs b/server/src/state.rs index 48c9052..219732c 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -273,8 +273,6 @@ impl<'a> SessionGuard<'a> { replayed: bool, ) { self.queued_jobs.remove(job_id); - // A replayed cancellation keeps the queue and chain honest but - // records no local status: whatever happened is in the history. if replayed { return; } @@ -364,13 +362,13 @@ pub struct State { /// 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. Messages at or concurrent with it are - /// replayed history: they rebuild state but never execute here. + /// rather than seeded it. join_frontier: Option, /// Jobs whose start event on our own baseboard arrived as replayed - /// history: a previous life started them, so no executor of ours - /// will ever stop them. Cleared per job by a terminal event. - replayed_started: BTreeSet, + /// 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. @@ -417,7 +415,7 @@ impl State { identities: LruCache::new(MAX_REGISTERED_IDENTITIES), revoked_keys: LruCache::new(MAX_REVOKED_KEYS), join_frontier, - replayed_started: Default::default(), + zombies: Default::default(), }; new.validate_certs(&roots); for root in &roots { @@ -463,11 +461,9 @@ impl State { .collect() } - /// Jobs a previous life of this server started and left running: - /// their start events arrived as replayed history and no terminal - /// event has. Jobs this incarnation started never appear here. - pub fn orphaned_jobs(&self) -> BTreeSet { - self.replayed_started + /// Jobs a previous life of this server started and left running. + pub fn zombies(&self) -> BTreeSet { + self.zombies .iter() .filter(|job_id| { self.running @@ -478,7 +474,7 @@ impl State { } /// Whether a message at `version` is live traffic rather than - /// replayed history: only strict causal descendants of the join + /// replayed history. Only strict causal descendants of the join /// frontier are live. fn is_live(&self, version: &Version) -> bool { self.join_frontier @@ -908,10 +904,9 @@ impl State { 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 are ever declared - // interrupted. + // previous life's. Only these can be zombies. if *baseboard_id == self.own_baseboard && !self.is_live(incoming_version) { - self.replayed_started.insert(*job_id); + self.zombies.insert(*job_id); } self.history.set_job_status( job_id, @@ -929,7 +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.replayed_started.remove(job_id); + self.zombies.remove(job_id); } self.running.remove(&(*job_id, baseboard_id.clone())); self.history.transition_job_status( @@ -947,9 +942,9 @@ impl State { }) } // An interrupted declaration raced this - // genuine stop; the real result wins, in - // either arrival order. The start time - // died with the interrupted incarnation. + // 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, @@ -974,16 +969,17 @@ 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.replayed_started.remove(job_id); + self.zombies.remove(job_id); } self.running.remove(&(*job_id, baseboard_id.clone())); self.history.transition_job_status( job_id, baseboard_id, Some(incoming_version.rank()), - // An error never displaces a terminal status: - // a genuine stop that raced an interrupted - // declaration keeps the real result. + // 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 { .. }) @@ -1045,8 +1041,8 @@ fn apply_message( tx_state.send_modify(|state| { if let Err(error) = state.update(log, executor, version, message) { error!(log, "state update failed"; "error" => ?error); - // Replayed messages already had their say: re-gossiping - // their errors would grow the set on every rejoin. + // 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 { @@ -1080,25 +1076,25 @@ fn drain_ready( } } -/// Declare historical jobs interrupted, excluding `survivors` -/// (jobs this incarnation is really running across a universe swap) -/// and jobs already so declared. Candidates come only from replayed -/// start events, so a job this incarnation is running is never a -/// candidate no matter when the scan runs. Call only at local -/// quiescence (after [`drain_ready`]), so a job whose terminal event -/// is already here is never falsely interrupted. -fn interrupt_orphans( +/// 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, - interrupted: &mut BTreeSet, + reaped: &mut BTreeSet, ) { - let mut orphans = tx_state.borrow().orphaned_jobs(); - orphans.retain(|job_id| !survivors.contains(job_id) && !interrupted.contains(job_id)); - for job_id in orphans { - interrupted.insert(job_id); + 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( @@ -1123,11 +1119,11 @@ fn interrupt_orphans( /// server's sessions. This stands in for joining the rack's network over /// sprockets on the bootstrap network. /// -/// A pristine seed's bookmark touches no storage; 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. +/// 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(), @@ -1180,8 +1176,8 @@ impl StateManager { // 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. - // `borrow_and_update` marks the value seen: a migration that - // landed before we subscribed must not replay as a swap. + // `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, @@ -1221,11 +1217,11 @@ impl StateManager { // 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; `interrupted` are jobs of - // ours the replay showed running, already declared dead. + // runs across a universe swap; `reaped` are zombies + // already declared interrupted. let mut frontier = frontier; let mut survivors: BTreeSet = BTreeSet::new(); - let mut interrupted: BTreeSet = BTreeSet::new(); + let mut reaped: BTreeSet = BTreeSet::new(); // Announce our build. if let Some((rumors, _)) = &gossip { @@ -1308,11 +1304,7 @@ impl StateManager { &version, &message, ); - // Replayed history can show jobs of ours - // running that died with a previous life. - // Drain to local quiescence first, so a - // job whose terminal event is already here - // is never declared interrupted. + // Replayed traffic can reveal zombies. if !live { drain_ready( &log, @@ -1322,13 +1314,13 @@ impl StateManager { gossip.as_ref().map(|(rumors, _)| rumors), &own_baseboard, ); - interrupt_orphans( + reap_zombies( &log, &tx_state, gossip.as_ref().map(|(rumors, _)| rumors), &own_baseboard, &survivors, - &mut interrupted, + &mut reaped, ); } }, @@ -1362,12 +1354,11 @@ impl StateManager { let fresh = universe.borrow_and_update().clone(); info!(log, "gossip universe changed, resetting state"; "network" => %fresh.rumors.network()); causal_messages = fresh.rumors.causal_messages(); - // A job we declared interrupted may still show as - // running (its error event races the swap); it is - // no survivor. + // 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| !interrupted.contains(job_id)); - interrupted = BTreeSet::new(); + 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| { @@ -1388,10 +1379,8 @@ impl StateManager { ) .into(), ); - // The set received at join is already local: drain - // it, then declare interrupted any job of ours the - // replayed history still shows running. A previous - // life started them; nothing will ever stop them. + // The set received at join is already local: + // drain it, then reap. drain_ready( &log, &mut causal_messages, @@ -1400,13 +1389,13 @@ impl StateManager { Some(&*rumors), &own_baseboard, ); - interrupt_orphans( + reap_zombies( &log, &tx_state, Some(&*rumors), &own_baseboard, &survivors, - &mut interrupted, + &mut reaped, ); } } diff --git a/server/tests/distributed.rs b/server/tests/distributed.rs index eb4fc82..48d9222 100644 --- a/server/tests/distributed.rs +++ b/server/tests/distributed.rs @@ -402,7 +402,7 @@ async fn interrupted_jobs_get_stopped() { }) .await; - // The job's genuine stop was in flight all along: when it lands, + // 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( @@ -511,8 +511,8 @@ async fn stragglers_do_not_interrupt_live_jobs() { }) .await; - // C reconnects; its marooned message reaches B as replayed-classified - // traffic and triggers the orphan scan. The live job must survive. + // 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(); @@ -586,8 +586,8 @@ async fn bookmarks_survive_restart() { b_shutdown.cancel(); drop(b); a.peers.send(BTreeSet::new()).unwrap(); - // Let the dead incarnation's tasks quiesce: a real reboot does, - // and two live sources over one slot are the store's one + // 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( From 48c7620709a2804bc30bb7d9aa90aabe3a9377ea Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 27 Aug 2026 19:42:24 -0600 Subject: [PATCH 8/9] Pin sprockets at the merged feature gate The ipcc feature gate merged upstream as 98615b86, identical in content to the branch rev we carried; the pin now names the commit on sprockets main. Co-Authored-By: Claude Mythos 5 --- Cargo.lock | 20 ++++++++++---------- Cargo.toml | 4 ++-- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 99af468..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)", @@ -5102,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]] @@ -5883,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" From fb12aa9ad4700787bcb13059c7ee6382729bee0d Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Fri, 28 Aug 2026 09:04:44 -0600 Subject: [PATCH 9/9] Blame build skew when a login fails across builds Co-Authored-By: Claude Mythos 5 --- client/src/commands.rs | 56 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 55 insertions(+), 1 deletion(-) 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(_) + )); + } +}