From f9f189f89615c08eb1e278b9ebee0f33432760b4 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 15:22:49 +0100 Subject: [PATCH 1/6] Run program steps on a supervised Bun host ProgramEngine dispatches each program attempt to one Bun host per member (program_host.mjs, protocol v1): a worker per attempt, effects in recorded process groups (boot id + leader start time, verified before the next host starts), worker and host liveness pings, modules read from the store blob into a verified cache file, one fenced result per key and attempt, and the program's $result journaled on done. Every run evaluation releases program attempts whose step left them, so cancellation, timeouts and retries stop the program and its effects at once. Bun joins the dev shell and the Rust check environment. --- crates/fractal-reconcile/src/reconcile.rs | 62 ++ .../src/reconcile/program.rs | 725 ++++++++++++++++++ .../src/reconcile/program/tests.rs | 156 ++++ .../src/reconcile/program_host.mjs | 248 ++++++ .../src/reconcile/tests/program_engine.rs | 221 ++++++ docs/st3/program-engine.md | 20 +- flake.nix | 4 + release-notes/program-engine-host.json | 11 + 8 files changed, 1442 insertions(+), 5 deletions(-) create mode 100644 crates/fractal-reconcile/src/reconcile/program.rs create mode 100644 crates/fractal-reconcile/src/reconcile/program/tests.rs create mode 100644 crates/fractal-reconcile/src/reconcile/program_host.mjs create mode 100644 crates/fractal-reconcile/src/reconcile/tests/program_engine.rs create mode 100644 release-notes/program-engine-host.json diff --git a/crates/fractal-reconcile/src/reconcile.rs b/crates/fractal-reconcile/src/reconcile.rs index 7313428822b..8fb43d58276 100644 --- a/crates/fractal-reconcile/src/reconcile.rs +++ b/crates/fractal-reconcile/src/reconcile.rs @@ -62,6 +62,7 @@ pub mod idle_nudge; pub mod placement; pub mod run_report; pub mod start_spacing; +pub mod program; pub mod start_jobs; pub mod seat_admission; mod seat_admission_queue; @@ -1421,6 +1422,8 @@ pub struct Reconciler { file_observations: Arc>>, file_watch_revision: Arc, launch_input_wake: std::sync::OnceLock>, + /// Runs `program` steps; started with the first one (docs/st3/program-engine.md). + program_engine: std::sync::OnceLock>, resource_provider: Arc, /// Open faults by subject and scope, loaded from the graph on first use. faults: Mutex>>, @@ -1637,6 +1640,7 @@ impl Reconciler { file_observations: Arc::new(Mutex::new(HashMap::new())), file_watch_revision: Default::default(), launch_input_wake: Default::default(), + program_engine: Default::default(), resource_provider: Arc::new(RegisteredResourceProvider), faults: Mutex::new(None), unrecorded_faults: Mutex::new(Vec::new()), @@ -1842,6 +1846,7 @@ impl Reconciler { file_observations: Arc::new(Mutex::new(HashMap::new())), file_watch_revision: Default::default(), launch_input_wake: Default::default(), + program_engine: Default::default(), resource_provider: Arc::new(RegisteredResourceProvider), faults: Mutex::new(None), unrecorded_faults: Mutex::new(Vec::new()), @@ -10585,6 +10590,11 @@ impl Reconciler { smallclaims::touched::record_due(|| { self.isolate("mission-run", &subject, || { let (run, mission) = self.store.mission_run_for_evaluation(id)?; + // Eager cancellation (§6): a program step that left its attempt in this run's + // latest transition stops now, before an inactive run returns below. + if let Some(engine) = self.program_engine.get() { + engine.release_left(&run); + } if active_only && !matches!(run.status.as_str(), "running" | "standing" | "blocked") { @@ -11473,6 +11483,10 @@ impl Reconciler { if !self.step_declarations_hold(run, &view.subject)? { return Ok(changed); } + if let Some(program) = &step.spec.program { + changed |= self.dispatch_program(run, &step, view, program)?; + return Ok(changed); + } if !view.agentless && !view.worker_reported { return Ok(changed); } @@ -14344,6 +14358,53 @@ impl Reconciler { Ok(true) } + /// The program engine, started with the first program step this daemon runs. + fn program_engine(&self) -> &Arc { + self.program_engine.get_or_init(|| { + let incremental = self.incremental.clone(); + let notify = self.notify.clone(); + let event_notify = self.event_notify.clone(); + program::ProgramEngine::new( + self.store.clone(), + Arc::new(move |step: &str| { + incremental.touch(step); + signal_changed(¬ify, &event_notify); + }), + self.driver_state_dir.join("programs"), + program::Liveness::default(), + ) + }) + } + + /// Hand an active program step to the program engine, which completes or fails it. A step + /// whose bundle this member does not hold fails here. + fn dispatch_program( + &self, + run: &MissionRunView, + step: &RuntimeStep<'_>, + view: &crate::model::StepRunView, + program: &crate::model::ProgramSpec, + ) -> Result { + let dispatch = program::Dispatch { + step: view.subject.clone(), + attempt: view.attempt, + run: run.subject.clone(), + program: program.clone(), + payload: run.inputs.iter().map(|(name, input)| (name.clone(), input.value.clone())).collect(), + env: run_variables(run, step, view).into_iter().filter(|(name, _)| name.starts_with("ST_")).collect(), + cwd: run.workspace.clone(), + }; + match self.program_engine().dispatch(dispatch)? { + program::Dispatched::Running | program::Dispatched::Started => Ok(false), + // The step keeps its state; `ProgramEngine::host_fault` names the reason. + program::Dispatched::HostDown(reason) => { + eprintln!("fractal: program {}: {reason}", view.subject); + Ok(false) + } + program::Dispatched::Refused(reason) => self.store.set_step_state(&view.subject, "failed", Some(&reason)), + } + } + fn materialize_step_declarations( &self, run: &MissionRunView, @@ -21128,6 +21189,7 @@ pub mod tests { mod start_preemption; mod seat_admission_queue; mod terminal_step_guard; + mod program_engine; #[test] fn native_exec_and_gate_shell_resolve_the_declared_path() { use super::{NativeRuntime, RuntimeControl}; diff --git a/crates/fractal-reconcile/src/reconcile/program.rs b/crates/fractal-reconcile/src/reconcile/program.rs new file mode 100644 index 00000000000..246c976b04e --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -0,0 +1,725 @@ +//! The program engine: runs `program` steps (docs/st3/program-engine.md). +//! +//! The daemon owns everything durable about a program run: which step runs, its journal (one +//! `program.step` claim per inner step result), replay after a restart, and the step's outcome. +//! Program bodies are evaluated by one supervised host process per member +//! (`program_host.mjs`, on Bun), which talks to this module over its stdin and stdout only (host +//! protocol v1, §2). That pipe is the host's only write path, so a host orphaned by a dead daemon +//! can never journal anything. +//! +//! - daemon → host: `start`, `ack`, `nack`, `cancel`, `resolved`, `ping`. +//! - host → daemon: `journal`, `run`, `ask`, `done`, `unresponsive`, `pong`. +//! +//! Every message after `start` names its step and attempt; both sides drop a message for an +//! attempt or host generation they no longer run. Supervision: the engine pings the host and +//! replaces one that does not answer within its deadline; a replaced or crashed host's programs +//! resume from their journals on the next host. Before it starts a host, the engine kills the +//! effect process groups an earlier host recorded and never ended (DQ2). + +use std::collections::{BTreeMap, HashMap}; +use std::io::{BufRead as _, BufReader, Write as _}; +use std::path::{Path, PathBuf}; +use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; +use std::sync::mpsc; +use std::sync::{Arc, Mutex, PoisonError, Weak}; +use std::time::{Duration, Instant}; + +use anyhow::{Context as _, Result, anyhow}; +use serde_json::{Value, json}; +use sha2::Digest as _; + +use crate::MissionsStore; +use crate::model::{ClaimInput, ClaimRecord, FractalError, MissionRunView, ProgramSpec}; +use crate::store::Store; + +/// The claim kind of a program journal entry, on the program's `step-run` subject. +pub const JOURNAL_KIND: &str = "program.step"; +/// The journal key of a program's own encoded success value, written when it ends. +pub const RESULT_KEY: &str = "$result"; +/// The step states in which the engine runs a program step's current attempt. A step waiting on +/// a person keeps its program loaded (§11). +pub const RUNNING_STATES: [&str; 4] = ["ready", "working", "blocked", "waiting-person"]; +const HOST_SOURCE: &str = include_str!("program_host.mjs"); +/// How long a crashed host waits before its replacement starts, so a host that cannot start does +/// not spin. +const HOST_RESTART_DELAY: Duration = Duration::from_millis(500); +const JOURNAL_PAGE: usize = 500; + +/// The liveness deadlines of DQ8. +#[derive(Clone, Copy, Debug)] +pub struct Liveness { + /// A program worker that does not answer the host's ping within this is terminated and its + /// step fails. + pub worker_deadline: Duration, + /// How often the engine pings the host. + pub host_ping_every: Duration, + /// A host that does not answer a ping within this is replaced. + pub host_deadline: Duration, +} + +impl Default for Liveness { + fn default() -> Self { + Self { + worker_deadline: Duration::from_secs(30), + host_ping_every: Duration::from_secs(5), + host_deadline: Duration::from_secs(30), + } + } +} + +/// One program step this daemon runs. +#[derive(Clone, Debug)] +pub struct Dispatch { + pub step: String, + pub attempt: u32, + /// The `mission-run/…` subject the step belongs to. + pub run: String, + pub program: ProgramSpec, + pub payload: BTreeMap, + pub env: BTreeMap, + pub cwd: String, +} + +#[derive(Debug, Eq, PartialEq)] +pub enum Dispatched { + /// The host already runs this attempt. + Running, + /// The host was asked to start (or resume) it. + Started, + /// The step cannot run; the reason fails it. + Refused(String), + /// No host could be started to run it; the step keeps its state and the engine reports the + /// reason through [`ProgramEngine::host_fault`]. + HostDown(String), +} + +struct Host { + child: Child, + /// Lines for the host's stdin. A writer thread owns the pipe, so neither the reconciler nor the + /// reader thread ever blocks on a host that stopped reading. + lines: mpsc::Sender>, + generation: u64, + /// The unanswered ping and when it was sent. + ping: Option<(u64, Instant)>, +} + +struct Running { + dispatch: Dispatch, + /// The host generation running it; a host that exits takes exactly its own programs with it. + generation: u64, +} + +#[derive(Default)] +struct Inner { + host: Option, + generations: u64, + pings: u64, + supervised: bool, + running: HashMap, + /// By step: the attempt and the last host or liveness failure that stopped it. + faults: HashMap, +} + +pub struct ProgramEngine { + store: Arc, + wake: Arc, + dir: PathBuf, + liveness: Liveness, + inner: Mutex, +} + +impl ProgramEngine { + /// `wake(step)` tells the reconciler a program step changed outside its evaluation. + pub fn new(store: Arc, wake: Arc, dir: PathBuf, liveness: Liveness) -> Arc { + Arc::new(Self { store, wake, dir, liveness, inner: Mutex::default() }) + } + + fn lock(&self) -> std::sync::MutexGuard<'_, Inner> { + self.inner.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Start or resume `dispatch` unless the live host already runs that attempt. + pub fn dispatch(self: &Arc, dispatch: Dispatch) -> Result { + if self.runs(&dispatch) { + return Ok(Dispatched::Running); + } + let module = match self.pinned_module(&dispatch.program) { + Ok(module) => module, + Err(reason) => return Ok(Dispatched::Refused(reason)), + }; + let journal = self.journal(&dispatch.step, dispatch.attempt)?; + let mut inner = self.lock(); + if inner.running.get(&dispatch.step).is_some_and(|running| { + running.dispatch.attempt == dispatch.attempt && inner.host.as_ref().is_some_and(|host| host.generation == running.generation) + }) { + return Ok(Dispatched::Running); + } + let generation = match self.ensure_host(&mut inner) { + Ok(generation) => generation, + Err(error) => { + let reason = format!("the program host cannot start: {error:#}"); + inner.faults.insert(dispatch.step.clone(), (dispatch.attempt, reason.clone())); + return Ok(Dispatched::HostDown(reason)); + } + }; + inner.faults.remove(&dispatch.step); + eprintln!( + "fractal: program {} attempt {} started with {} journaled steps", + dispatch.step, + dispatch.attempt, + journal.len() + ); + let message = json!({ + "t": "start", + "step": dispatch.step, + "attempt": dispatch.attempt, + "run": dispatch.run, + "module": module, + "sha256": dispatch.program.bundle_hash, + "payload": dispatch.payload, + "env": dispatch.env, + "cwd": dispatch.cwd, + "journal": journal, + }); + inner.running.insert(dispatch.step.clone(), Running { dispatch, generation }); + send(&inner, &message); + Ok(Dispatched::Started) + } + + fn runs(&self, dispatch: &Dispatch) -> bool { + let inner = self.lock(); + let live = inner.host.as_ref().map(|host| host.generation); + inner + .running + .get(&dispatch.step) + .is_some_and(|running| running.dispatch.attempt == dispatch.attempt && Some(running.generation) == live) + } + + /// The last host or liveness failure that stopped this attempt, until a dispatch of the step + /// starts again. + pub fn host_fault(&self, step: &str, attempt: u32) -> Option { + self.lock().faults.get(step).filter(|(at, _)| *at == attempt).map(|(_, reason)| reason.clone()) + } + + /// Eager cancellation (§6), called from every evaluation of a run: each program attempt of + /// `run` this engine runs whose step left it (ended, failed, timed out, retried, cancelled with + /// its run or replaced by a new generation) is stopped now, with its effects. + pub fn release_left(&self, run: &MissionRunView) { + let terminal = matches!(run.status.as_str(), "completed" | "failed" | "cancelled"); + let mut inner = self.lock(); + let left = inner + .running + .iter() + .filter(|(_, running)| running.dispatch.run == run.subject) + .filter_map(|(step, running)| { + let attempt = running.dispatch.attempt; + let reason = match run.steps.iter().find(|view| view.subject == *step) { + None => Some("the run moved to another generation".to_owned()), + Some(_) if terminal => Some(format!("the run is {}", run.status)), + Some(view) if view.attempt != attempt => Some(format!("the step moved to attempt {}", view.attempt)), + Some(view) if !RUNNING_STATES.contains(&view.status.as_str()) => Some(format!("the step is {}", view.status)), + Some(_) => None, + }; + reason.map(|reason| (step.clone(), attempt, reason)) + }) + .collect::>(); + for (step, attempt, reason) in left { + inner.running.remove(&step); + eprintln!("fractal: program {step} attempt {attempt} cancelled: {reason}"); + send(&inner, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + } + } + + /// The bundle as this revision pinned it, read from the store's blob of its hash (never from + /// the author's path) into a verified, content-addressed file the host imports. + fn pinned_module(&self, program: &ProgramSpec) -> std::result::Result { + let hash = &program.bundle_hash; + let pinned = self.dir.join("bundles").join(format!("{hash}.mjs")); + if std::fs::read(&pinned).is_ok_and(|bytes| sha256(&bytes) == *hash) { + return Ok(pinned); + } + let bytes = self + .store + .get_blob(hash) + .map_err(|error| format!("reading program bundle {hash}: {error:#}"))? + .ok_or_else(|| format!("program bundle `{}` (sha256 {hash}) is not in this member's store", program.bundle))?; + let actual = sha256(&bytes); + if actual != *hash { + return Err(format!("the stored program bundle {hash} has sha256 {actual}")); + } + write_atomically(&pinned, &bytes).map_err(|error| format!("pinning program bundle {hash}: {error:#}"))?; + Ok(pinned) + } + + /// The entries replay returns: the completed steps and typed failures of this attempt, in + /// store order. A retried attempt starts from the top, because an earlier attempt's + /// compensation may have undone what its completed steps did. + fn journal(&self, step: &str, attempt: u32) -> Result> { + let mut claims = Vec::new(); + let mut before = None; + loop { + // Newest first; each page continues below the oldest index of the previous one. + let page = self.store.claims_for_subject_kind_at(step, JOURNAL_KIND, before, true, JOURNAL_PAGE)?; + claims.extend(page.claims); + match page.next_cursor { + Some(cursor) => before = Some(cursor), + None => break, + } + } + claims.sort_by_key(|claim| claim.store_index); + let mut entries = Vec::new(); + for claim in claims { + let fields = &claim.body["fields"]; + if fields["attempt"].as_u64() != Some(u64::from(attempt)) || fields["key"] == RESULT_KEY { + continue; + } + match fields["status"].as_str() { + Some("completed") => entries.push(json!({"key": fields["key"], "_tag": "Success", "value": fields["value"]})), + Some("failed") => entries.push(json!({"key": fields["key"], "_tag": "Failure", "error": fields["error"]})), + _ => {} + } + } + Ok(entries) + } + + /// The live host's generation, starting one when there is none or the last one exited. + fn ensure_host(self: &Arc, inner: &mut Inner) -> Result { + if let Some(host) = inner.host.as_mut() { + if host.child.try_wait()?.is_none() { + return Ok(host.generation); + } + // Its reader thread resumes the programs it ran (see `host_exited`). + if let Some(mut host) = inner.host.take() { + let _ = host.child.wait(); + } + } + // No host runs now, so every recorded group belongs to one that is gone (DQ2). + let ended = self.reap_recorded_groups()?; + if ended > 0 { + eprintln!("fractal: ended {ended} effect process groups an earlier program host left"); + } + let script = self.dir.join(format!("host-{}.mjs", &sha256(HOST_SOURCE.as_bytes())[..12])); + if !script.is_file() { + write_atomically(&script, HOST_SOURCE.as_bytes())?; + } + let runtime = runtime()?; + let mut child = Command::new(&runtime) + .arg(&script) + .current_dir(&self.dir) + .env("FRACTAL_PROGRAM_GROUPS", self.groups_dir()) + .env("FRACTAL_PROGRAM_BOOT_ID", boot_id()?) + .env("FRACTAL_PROGRAM_WORKER_DEADLINE_MS", self.liveness.worker_deadline.as_millis().to_string()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .with_context(|| format!("starting the program host with {}", runtime.display()))?; + let stdin = child.stdin.take().ok_or_else(|| anyhow!("program host has no stdin"))?; + let stdout = child.stdout.take().ok_or_else(|| anyhow!("program host has no stdout"))?; + inner.generations += 1; + let generation = inner.generations; + eprintln!("fractal: program host {generation} started (pid {})", child.id()); + let (lines, queue) = mpsc::channel(); + std::thread::Builder::new() + .name(format!("program-host-{generation}-in")) + .spawn(move || write_host(stdin, queue))?; + let engine = Arc::clone(self); + std::thread::Builder::new() + .name(format!("program-host-{generation}-out")) + .spawn(move || engine.read_host(stdout, generation))?; + inner.host = Some(Host { child, lines, generation, ping: None }); + if !inner.supervised { + inner.supervised = true; + let engine = Arc::downgrade(self); + let every = self.liveness.host_ping_every; + std::thread::Builder::new() + .name("program-host-liveness".into()) + .spawn(move || supervise(engine, every))?; + } + Ok(generation) + } + + fn groups_dir(&self) -> PathBuf { + self.dir.join("groups") + } + + /// End the effect process groups recorded by hosts that are gone (DQ2). A record from an + /// earlier boot is dropped without a kill, so a reused pid is never signalled; a group is + /// killed when its leader still has the recorded start time, or when its leader is gone while + /// the group lives on. Returns how many groups it killed. + pub fn reap_recorded_groups(&self) -> Result { + let dir = self.groups_dir(); + std::fs::create_dir_all(&dir)?; + let boot = boot_id()?; + let mut killed = 0; + for entry in std::fs::read_dir(&dir)? { + let path = entry?.path(); + if path.extension().and_then(|extension| extension.to_str()) != Some("json") { + // A record the host was still writing when it died names a group it never ran. + std::fs::remove_file(&path)?; + continue; + } + let record: Value = std::fs::read(&path) + .ok() + .and_then(|bytes| serde_json::from_slice(&bytes).ok()) + .unwrap_or(Value::Null); + let pid = record["pid"].as_i64().and_then(|pid| i32::try_from(pid).ok()).filter(|pid| *pid > 1); + if let Some(pid) = pid + && record["boot_id"].as_str() == Some(boot.as_str()) + { + let leader = start_time(pid); + let ours = match (&leader, record["start_time"].as_str()) { + (Some(now), Some(recorded)) => now == recorded, + // The leader is gone: anything left in its group is the effect's. + (None, _) => true, + (Some(_), None) => false, + }; + // SAFETY: signalling a process group has no memory effects. + if ours && unsafe { libc::kill(-pid, libc::SIGKILL) } == 0 { + killed += 1; + } + } + std::fs::remove_file(&path)?; + } + Ok(killed) + } + + /// The host process's pid, while one runs. + pub fn host_pid(&self) -> Option { + self.lock().host.as_ref().map(|host| host.child.id()) + } + + fn read_host(self: Arc, stdout: ChildStdout, generation: u64) { + for line in BufReader::new(stdout).lines() { + let Ok(line) = line else { break }; + let message = match serde_json::from_str::(&line) { + Ok(message) => message, + Err(error) => { + eprintln!("fractal: program host sent an unreadable line ({error}): {line}"); + continue; + } + }; + if let Err(error) = self.handle(&message, generation) { + eprintln!("fractal: program host message {line}: {error:#}"); + } + } + self.host_exited(generation); + } + + fn handle(&self, message: &Value, generation: u64) -> Result<()> { + if message["t"] == "pong" { + let mut inner = self.lock(); + if let Some(host) = inner.host.as_mut().filter(|host| host.generation == generation) + && host.ping.is_some_and(|(seq, _)| message["seq"].as_u64() == Some(seq)) + { + host.ping = None; + } + return Ok(()); + } + let step = message["step"].as_str().ok_or_else(|| anyhow!("message without a step"))?; + let attempt = message["attempt"].as_u64().ok_or_else(|| anyhow!("message without an attempt"))?; + let program = { + let inner = self.lock(); + match inner.running.get(step) { + // A message from an attempt or host this engine no longer runs changes nothing. + Some(running) if u64::from(running.dispatch.attempt) == attempt && running.generation == generation => { + running.dispatch.program.bundle_hash.clone() + } + _ => return Ok(()), + } + }; + let attempt = u32::try_from(attempt)?; + match message["t"].as_str() { + Some("journal") => self.journal_entry(step, attempt, &program, message), + Some("done") => self.done(step, attempt, &program, message), + Some("unresponsive") => self.unresponsive(step, attempt, message), + other => Err(anyhow!("unknown message type {other:?}")), + } + } + + /// Why the step no longer runs this attempt, if it does not. + fn left(&self, step: &str, attempt: u32) -> Result> { + Ok(match self.store.step_run(step)? { + Some(view) => (view.attempt != attempt || !RUNNING_STATES.contains(&view.status.as_str())) + .then(|| format!("the step is {} at attempt {}", view.status, view.attempt)), + None => Some("the step run no longer exists".to_owned()), + }) + } + + /// The backstop to [`Self::release_left`]: stop the program if its step left this attempt. + /// True when it was stopped. + fn stop_if_left(&self, step: &str, attempt: u32) -> Result { + let Some(reason) = self.left(step, attempt)? else { return Ok(false) }; + let mut inner = self.lock(); + inner.running.remove(step); + send(&inner, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + Ok(true) + } + + /// Append one journal entry. A completed or failed entry is fenced by its key: one result per + /// key and attempt, so a second, different result is refused (R03). + #[allow(clippy::too_many_arguments)] + fn write_journal( + &self, + step: &str, + attempt: u32, + program: &str, + key: &str, + status: &str, + result: &Value, + child: Option<&str>, + ) -> std::result::Result { + let mut fields = BTreeMap::from([ + ("key".to_owned(), Value::from(key)), + ("status".to_owned(), Value::from(status)), + ("attempt".to_owned(), Value::from(attempt)), + ("program".to_owned(), Value::from(program)), + ]); + for name in ["value", "error", "ms"] { + if let Some(value) = result.get(name).filter(|value| !value.is_null()) { + fields.insert(name.to_owned(), value.clone()); + } + } + if let Some(child) = child { + fields.insert("child".to_owned(), Value::from(child)); + } + self.store.append_claim(&ClaimInput { + subject: step.to_owned(), + kind: JOURNAL_KIND.to_owned(), + actor: None, + fields, + evidence: Vec::new(), + expected_subject: None, + idempotency_key: (status != "started").then(|| format!("program:{step}:{attempt}:{key}:result")), + }) + } + + fn journal_entry(&self, step: &str, attempt: u32, program: &str, message: &Value) -> Result<()> { + if self.stop_if_left(step, attempt)? { + return Ok(()); + } + let key = message["key"].as_str().ok_or_else(|| anyhow!("journal entry without a key"))?; + let status = message["status"].as_str().ok_or_else(|| anyhow!("journal entry without a status"))?; + if key == RESULT_KEY { + return Err(anyhow!("a program may not journal the reserved key {RESULT_KEY}")); + } + let written = self.write_journal(step, attempt, program, key, status, message, None); + if status == "started" { + return written.map(|_| ()).map_err(Into::into); + } + let inner = self.lock(); + let seq = &message["seq"]; + match written { + Ok(_) => send(&inner, &json!({"t": "ack", "step": step, "attempt": attempt, "seq": seq})), + // The program must not continue past a result that is not durable: the step fails. + Err(error) => send( + &inner, + &json!({"t": "nack", "step": step, "attempt": attempt, "seq": seq, "error": format!("journal write refused: {error}")}), + ), + } + Ok(()) + } + + fn done(&self, step: &str, attempt: u32, program: &str, message: &Value) -> Result<()> { + self.lock().running.remove(step); + if self.left(step, attempt)?.is_some() { + return Ok(()); + } + let exit = &message["exit"]; + let (status, reason) = match exit["_tag"].as_str() { + Some("Success") => match self.write_journal(step, attempt, program, RESULT_KEY, "completed", &json!({"value": exit["value"]}), None) { + Ok(_) => ("completed", None), + Err(error) => ("failed", Some(format!("the program's result could not be journaled: {error}"))), + }, + _ => ("failed", Some(failure_reason(&exit["error"]))), + }; + self.store.set_step_state_unless_terminal(step, status, reason.as_deref())?; + eprintln!("fractal: program {step} {status}"); + (self.wake)(step); + Ok(()) + } + + /// The host terminated this attempt's worker after it missed its liveness deadline (DQ8): the + /// step fails once instead of looping across restarts. + fn unresponsive(&self, step: &str, attempt: u32, message: &Value) -> Result<()> { + let ms = message["ms"].as_u64().unwrap_or_default(); + let reason = format!("the program's worker missed its {ms} ms liveness deadline and was terminated"); + { + let mut inner = self.lock(); + inner.running.remove(step); + inner.faults.insert(step.to_owned(), (attempt, reason.clone())); + } + eprintln!("fractal: program {step} attempt {attempt}: {reason}"); + if self.left(step, attempt)?.is_none() { + self.store.set_step_state_unless_terminal(step, "failed", Some(&reason))?; + } + (self.wake)(step); + Ok(()) + } + + /// Ping the host; replace one whose last ping went unanswered past its deadline. + fn ping_host(&self) { + let mut inner = self.lock(); + inner.pings += 1; + let seq = inner.pings; + let deadline = self.liveness.host_deadline; + let Some(host) = inner.host.as_mut() else { return }; + match host.ping { + Some((_, sent)) if sent.elapsed() > deadline => { + eprintln!( + "fractal: program host {} did not answer a ping within {} ms; replacing it", + host.generation, + deadline.as_millis() + ); + // Its reader thread sees the pipe close and resumes its programs on a new host. + let _ = host.child.kill(); + } + Some(_) => {} + None => { + host.ping = Some((seq, Instant::now())); + let _ = host.lines.send(format!("{{\"t\":\"ping\",\"seq\":{seq}}}\n").into_bytes()); + } + } + } + + /// A host that exits takes its programs with it; a new host resumes each from its journal. + fn host_exited(self: Arc, generation: u64) { + let resume = { + let mut inner = self.lock(); + if inner.host.as_ref().is_some_and(|host| host.generation == generation) + && let Some(mut host) = inner.host.take() + { + let _ = host.child.kill(); + let status = host.child.wait(); + eprintln!("fractal: program host {generation} exited ({status:?})"); + } + let gone = inner + .running + .iter() + .filter(|(_, running)| running.generation == generation) + .map(|(step, _)| step.clone()) + .collect::>(); + gone.iter().filter_map(|step| inner.running.remove(step)).map(|running| running.dispatch).collect::>() + }; + if resume.is_empty() { + return; + } + std::thread::sleep(HOST_RESTART_DELAY); + for dispatch in resume { + let (step, attempt) = (dispatch.step.clone(), dispatch.attempt); + if !matches!(self.left(&step, attempt), Ok(None)) { + continue; + } + let failure = match self.dispatch(dispatch) { + Ok(Dispatched::Refused(reason)) => { + if let Err(error) = self.store.set_step_state_unless_terminal(&step, "failed", Some(&reason)) { + eprintln!("fractal: failing program {step}: {error:#}"); + } + None + } + Ok(Dispatched::HostDown(reason)) => Some(reason), + Ok(Dispatched::Running | Dispatched::Started) => continue, + Err(error) => Some(format!("resuming the program after its host exited: {error:#}")), + }; + if let Some(reason) = failure { + eprintln!("fractal: program {step}: {reason}"); + self.lock().faults.insert(step.clone(), (attempt, reason)); + } + (self.wake)(&step); + } + } +} + +fn supervise(engine: Weak, every: Duration) { + loop { + std::thread::sleep(every); + let Some(engine) = engine.upgrade() else { return }; + engine.ping_host(); + } +} + +/// Queue one line for the live host. A host that is gone drops it; its programs resume on the next. +fn send(inner: &Inner, message: &Value) { + if let Some(host) = inner.host.as_ref() { + let mut line = serde_json::to_vec(message).unwrap_or_default(); + line.push(b'\n'); + let _ = host.lines.send(line); + } +} + +fn write_host(mut stdin: ChildStdin, queue: mpsc::Receiver>) { + for line in queue { + if stdin.write_all(&line).and_then(|()| stdin.flush()).is_err() { + return; + } + } +} + +/// A typed failure keeps its tag and fields, so `fractal missions show` says which error ended +/// the program. +fn failure_reason(error: &Value) -> String { + let text = match error { + Value::String(text) => text.clone(), + other => other.to_string(), + }; + text.chars().take(900).collect() +} + +fn sha256(bytes: &[u8]) -> String { + hex::encode(sha2::Sha256::digest(bytes)) +} + +/// The JavaScript runtime the host runs on: `FRACTAL_PROGRAM_RUNTIME`, else `bun` on the daemon's +/// `PATH`. +fn runtime() -> Result { + if let Some(runtime) = std::env::var_os("FRACTAL_PROGRAM_RUNTIME") { + return Ok(PathBuf::from(runtime)); + } + std::env::var_os("PATH") + .iter() + .flat_map(std::env::split_paths) + .map(|dir| dir.join("bun")) + .find(|candidate| candidate.is_file()) + .ok_or_else(|| anyhow!("no `bun` on the daemon's PATH and no FRACTAL_PROGRAM_RUNTIME to run program steps")) +} + +/// This boot's identity, as the host records it with every effect group. +fn boot_id() -> Result { + #[cfg(target_os = "linux")] + { + Ok(std::fs::read_to_string("/proc/sys/kernel/random/boot_id")?.trim().to_owned()) + } + #[cfg(not(target_os = "linux"))] + { + let output = Command::new("sysctl").args(["-n", "kern.boottime"]).output()?; + anyhow::ensure!(output.status.success(), "sysctl kern.boottime failed"); + Ok(String::from_utf8_lossy(&output.stdout).trim().to_owned()) + } +} + +/// A live process's start time, read the way the host reads it (`program_host.mjs`). +fn start_time(pid: i32) -> Option { + #[cfg(target_os = "linux")] + { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + stat.get(stat.rfind(')')? + 2..)?.split(' ').nth(19).map(str::to_owned) + } + #[cfg(not(target_os = "linux"))] + { + let output = Command::new("ps").args(["-o", "lstart=", "-p", &pid.to_string()]).output().ok()?; + let text = String::from_utf8_lossy(&output.stdout).trim().to_owned(); + (!text.is_empty()).then_some(text) + } +} + +fn write_atomically(path: &Path, bytes: &[u8]) -> Result<()> { + let dir = path.parent().ok_or_else(|| anyhow!("{} has no parent", path.display()))?; + std::fs::create_dir_all(dir)?; + let temporary = dir.join(format!(".{}.{}", path.file_name().unwrap_or_default().to_string_lossy(), std::process::id())); + std::fs::write(&temporary, bytes)?; + std::fs::rename(&temporary, path)?; + Ok(()) +} + +#[cfg(test)] +mod tests; diff --git a/crates/fractal-reconcile/src/reconcile/program/tests.rs b/crates/fractal-reconcile/src/reconcile/program/tests.rs new file mode 100644 index 00000000000..5fe8c6b44a3 --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/program/tests.rs @@ -0,0 +1,156 @@ +use super::*; + +fn engine(dir: &Path) -> Arc { + ProgramEngine::new(Arc::new(Store::open_memory("node").unwrap()), Arc::new(|_: &str| {}), dir.to_owned(), Liveness::default()) +} + +fn entry(engine: &ProgramEngine, step: &str, key: &str, status: &str, attempt: u32, value: Value) { + let name = if status == "failed" { "error" } else { "value" }; + let result = Value::Object(serde_json::Map::from_iter([(name.to_owned(), value)])); + engine.write_journal(step, attempt, &"a".repeat(64), key, status, &result, None).unwrap(); +} + +fn journal_claims(engine: &ProgramEngine, step: &str) -> usize { + engine.store.claims_for(step, Some(JOURNAL_KIND)).unwrap().len() +} + +#[test] +fn replay_returns_this_attempts_results_in_order() { + let dir = tempfile::tempdir().unwrap(); + let engine = engine(dir.path()); + let step = "step-run/generation/program"; + entry(&engine, step, "merge", "completed", 1, json!({"commit": "c1"})); + entry(&engine, step, "publish-local", "started", 1, Value::Null); + entry(&engine, step, "publish-local", "failed", 1, json!({"_tag": "StepFailed", "step": "publish"})); + entry(&engine, step, "revert", "completed", 1, json!({"reverted": ["c1"]})); + entry(&engine, step, RESULT_KEY, "completed", 1, json!({"done": true})); + entry(&engine, step, "merge", "completed", 2, json!({"commit": "c2"})); + assert_eq!( + engine.journal(step, 1).unwrap(), + vec![ + json!({"key": "merge", "_tag": "Success", "value": {"commit": "c1"}}), + json!({"key": "publish-local", "_tag": "Failure", "error": {"_tag": "StepFailed", "step": "publish"}}), + json!({"key": "revert", "_tag": "Success", "value": {"reverted": ["c1"]}}), + ] + ); + // A retried attempt never replays the first attempt, whose revert undid its merge. + assert_eq!(engine.journal(step, 2).unwrap(), vec![json!({"key": "merge", "_tag": "Success", "value": {"commit": "c2"}})]); +} + +#[test] +fn replay_pages_through_long_journals_in_order() { + let dir = tempfile::tempdir().unwrap(); + let engine = engine(dir.path()); + let step = "step-run/generation/program"; + for n in 1..=(JOURNAL_PAGE + 3) { + entry(&engine, step, &format!("poll#{n}"), "completed", 1, Value::from(n)); + } + let journal = engine.journal(step, 1).unwrap(); + assert_eq!(journal.len(), JOURNAL_PAGE + 3); + assert!(journal.iter().enumerate().all(|(index, entry)| entry["value"] == Value::from(index + 1))); +} + +#[test] +fn a_revision_runs_the_stored_blob_it_pins() { + let dir = tempfile::tempdir().unwrap(); + let engine = engine(dir.path()); + let hash = engine.store.put_blob(b"export default 1").unwrap(); + // The author's path is provenance only: the member runs the store's bytes. + let program = ProgramSpec { bundle: "/nowhere/land.mjs".into(), bundle_hash: hash.clone(), stalled_after_ms: None }; + let pinned = engine.pinned_module(&program).unwrap(); + assert_eq!(std::fs::read(&pinned).unwrap(), b"export default 1"); + // A damaged pinned copy is replaced from the blob, never run. + std::fs::write(&pinned, b"export default 2").unwrap(); + assert_eq!(engine.pinned_module(&program).unwrap(), pinned); + assert_eq!(std::fs::read(&pinned).unwrap(), b"export default 1"); + // A revision whose blob this member does not hold fails its step. + let missing = ProgramSpec { bundle_hash: "0".repeat(64), ..program }; + assert!(engine.pinned_module(&missing).unwrap_err().contains("is not in this member's store")); +} + +#[test] +fn a_key_has_one_result_per_attempt() { + let dir = tempfile::tempdir().unwrap(); + let engine = engine(dir.path()); + let step = "step-run/generation/program"; + let program = "a".repeat(64); + let write = |attempt, status, value: Value| { + engine.write_journal(step, attempt, &program, "merge", status, &json!({"value": value, "ms": 3}), None) + }; + write(1, "started", Value::Null).unwrap(); + write(1, "started", Value::Null).unwrap(); + write(1, "completed", json!({"commit": "c1"})).unwrap(); + // The same result again (a resent entry) is the same durable result. + write(1, "completed", json!({"commit": "c1"})).unwrap(); + // A double execution with a different result, or a failure for a completed key, is refused. + assert!(write(1, "completed", json!({"commit": "c2"})).is_err()); + assert!(engine.write_journal(step, 1, &program, "merge", "failed", &json!({"error": {"_tag": "X"}}), None).is_err()); + // A retried attempt has its own result. + write(2, "completed", json!({"commit": "c2"})).unwrap(); + assert_eq!(journal_claims(&engine, step), 4); + assert_eq!(engine.journal(step, 1).unwrap(), vec![json!({"key": "merge", "_tag": "Success", "value": {"commit": "c1"}})]); +} + +#[test] +fn messages_for_another_attempt_or_host_change_nothing() { + let dir = tempfile::tempdir().unwrap(); + let engine = engine(dir.path()); + let step = "step-run/generation/program"; + let dispatch = Dispatch { + step: step.into(), + attempt: 2, + run: "mission-run/run".into(), + program: ProgramSpec { bundle: "/b.mjs".into(), bundle_hash: "a".repeat(64), stalled_after_ms: None }, + payload: BTreeMap::new(), + env: BTreeMap::new(), + cwd: "/".into(), + }; + engine.lock().running.insert(step.into(), Running { dispatch, generation: 1 }); + let message = |attempt: u32, t: &str| { + json!({"t": t, "step": step, "attempt": attempt, "seq": 1, "key": "merge", "status": "completed", "value": 1, + "exit": {"_tag": "Success", "value": 1}, "ms": 5}) + }; + for t in ["journal", "done", "unresponsive"] { + // An earlier attempt of the step, then the right attempt from a replaced host. + engine.handle(&message(1, t), 1).unwrap(); + engine.handle(&message(2, t), 2).unwrap(); + } + assert_eq!(journal_claims(&engine, step), 0); + assert!(engine.lock().running.contains_key(step)); + assert_eq!(engine.host_fault(step, 2), None); +} + +#[test] +fn recorded_groups_are_killed_only_when_their_leader_is_verified() { + use std::os::unix::process::CommandExt as _; + let dir = tempfile::tempdir().unwrap(); + let engine = engine(dir.path()); + let groups = engine.groups_dir(); + std::fs::create_dir_all(&groups).unwrap(); + let boot = boot_id().unwrap(); + let leader = || Command::new("sleep").arg("1000").process_group(0).spawn().unwrap(); + let alive = |child: &mut Child| child.try_wait().unwrap().is_none(); + let record = |child: &Child, boot: &str, start: Option| { + let pid = child.id(); + std::fs::write(groups.join(format!("{pid}.json")), json!({"pid": pid, "boot_id": boot, "start_time": start}).to_string()).unwrap(); + }; + // Recorded by this boot with the leader's start time: an effect a dead host left behind. + let mut ours = leader(); + record(&ours, &boot, start_time(ours.id() as i32)); + // The same pid with another start time is a process that reused it. + let mut reused = leader(); + record(&reused, &boot, Some("0".into())); + // A record from an earlier boot is dropped without a kill: its pid may be anyone's now. + let mut earlier = leader(); + record(&earlier, "an-earlier-boot", start_time(earlier.id() as i32)); + std::fs::write(groups.join("half-written.json.tmp"), "{").unwrap(); + assert_eq!(engine.reap_recorded_groups().unwrap(), 1); + let status = ours.wait().unwrap(); + assert_eq!(std::os::unix::process::ExitStatusExt::signal(&status), Some(libc::SIGKILL)); + assert!(alive(&mut reused) && alive(&mut earlier)); + assert_eq!(std::fs::read_dir(&groups).unwrap().count(), 0); + for mut child in [reused, earlier] { + child.kill().unwrap(); + child.wait().unwrap(); + } +} diff --git a/crates/fractal-reconcile/src/reconcile/program_host.mjs b/crates/fractal-reconcile/src/reconcile/program_host.mjs new file mode 100644 index 00000000000..2fade8ea860 --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/program_host.mjs @@ -0,0 +1,248 @@ +// The Fractal program host (docs/st3/program-engine.md §2, protocol v1). The daemon's program +// engine starts one per member and talks to it over stdin/stdout, one JSON object per line. The +// host evaluates program bundles and nothing else: it holds no credential, and its only durable +// write is a journal entry the daemon acknowledges. It has no dependencies, so its behaviour is +// fixed by the daemon build; a program's own runtime (the `fractal` program API, Effect) is inside +// its pinned bundle. +// +// The main thread owns the pipe, every effect's process group and liveness. Each program attempt +// runs in its own worker, so one program's busy loop cannot starve the others: a worker that does +// not answer a ping within FRACTAL_PROGRAM_WORKER_DEADLINE_MS is terminated (DQ8). Effects run as +// their own process groups; the main thread records each in FRACTAL_PROGRAM_GROUPS with the boot +// id and the leader's start time, kills the group when its leader exits, when its attempt ends or +// is cancelled, and when stdin closes, and removes the record once it has (DQ2). +import { spawn, spawnSync } from 'node:child_process' +import { createHash } from 'node:crypto' +import { readFileSync, renameSync, rmSync, writeFileSync } from 'node:fs' +import { join } from 'node:path' +import { createInterface } from 'node:readline' +import { fileURLToPath } from 'node:url' +import { Worker, isMainThread, parentPort, workerData } from 'node:worker_threads' + +/** The entry point a bundle's default export carries, version 1 of the host contract. */ +const ENTRY = Symbol.for('fractal.program.v1') + +const defect = (error) => ({ _tag: 'Failure', error: { _tag: 'Defect', message: String(error?.stack ?? error).slice(0, 2000) } }) + +/** The leader's start time as the daemon reads it before it kills a recorded group. */ +const startTime = (pid) => { + try { + if (process.platform === 'linux') { + const stat = readFileSync(`/proc/${pid}/stat`, 'utf8') + return stat.slice(stat.lastIndexOf(')') + 2).split(' ')[19] + } + const out = spawnSync('ps', ['-o', 'lstart=', '-p', String(pid)], { encoding: 'utf8' }).stdout.trim() + return out === '' ? undefined : out + } catch { + return undefined + } +} + +const main = () => { + const groupsDir = process.env.FRACTAL_PROGRAM_GROUPS + const bootId = process.env.FRACTAL_PROGRAM_BOOT_ID + const deadline = Number(process.env.FRACTAL_PROGRAM_WORKER_DEADLINE_MS) + const send = (message) => process.stdout.write(`${JSON.stringify(message)}\n`) + /** By step-run subject: the attempt this host runs. */ + const runs = new Map() + + const record = (run, pid) => { + const path = join(groupsDir, `${pid}.json`) + const body = JSON.stringify({ pid, boot_id: bootId, start_time: startTime(pid), step: run.step, attempt: run.attempt }) + writeFileSync(`${path}.tmp`, body) + renameSync(`${path}.tmp`, path) + } + const killGroup = (pid) => { + try { + process.kill(-pid, 'SIGKILL') + } catch {} + rmSync(join(groupsDir, `${pid}.json`), { force: true }) + } + /** End an attempt: its worker and every effect group it started. */ + const end = (run) => { + run.ended = true + if (runs.get(run.step) === run) runs.delete(run.step) + clearInterval(run.ping) + void run.worker.terminate() + for (const pid of run.groups) killGroup(pid) + run.groups.clear() + } + + const spawnEffect = (run, { id, argv: requested, options = {} }) => { + const reply = (message) => run.ended || run.worker.postMessage({ t: 'spawned', id, ...message }) + const argv = + options.priority === 'background' + ? process.platform === 'darwin' + ? ['taskpolicy', '-b', ...requested] + : ['nice', '-n', '19', ...requested] + : requested + let child + try { + child = spawn(argv[0], argv.slice(1), { + cwd: options.cwd ?? run.cwd, + env: { ...process.env, ...options.env, ...run.env }, + stdio: ['ignore', 'pipe', 'pipe'], + detached: true, + }) + } catch (error) { + return reply({ error: String(error) }) + } + const pid = child.pid + if (pid !== undefined) { + record(run, pid) + run.groups.add(pid) + } + const out = [] + const err = [] + child.stdout.on('data', (chunk) => out.push(chunk)) + child.stderr.on('data', (chunk) => err.push(chunk)) + child.on('error', (error) => reply({ error: String(error) })) + // A group ends with its leader, so nothing it started outlives the step's effect. + child.on('exit', () => { + if (pid === undefined) return + run.groups.delete(pid) + killGroup(pid) + }) + child.on('close', (code, signal) => + reply({ result: { code: code ?? 128, signal, stdout: Buffer.concat(out).toString(), stderr: Buffer.concat(err).toString() } }), + ) + } + + const start = (message) => { + const previous = runs.get(message.step) + if (previous !== undefined) end(previous) + const worker = new Worker(fileURLToPath(import.meta.url), { workerData: message }) + const run = { + step: message.step, + attempt: message.attempt, + cwd: message.cwd, + env: message.env, + worker, + groups: new Set(), + ended: false, + answered: Date.now(), + ping: undefined, + } + runs.set(message.step, run) + const ids = { step: run.step, attempt: run.attempt } + worker.on('message', (message) => { + if (run.ended) return + switch (message.t) { + case 'pong': + run.answered = Date.now() + return + case 'spawn': + return spawnEffect(run, message) + case 'done': + send({ t: 'done', ...ids, exit: message.exit }) + return end(run) + default: + // journal, run, ask: the daemon decides; the worker waits for its answer. + return send({ ...message, ...ids }) + } + }) + worker.on('error', (error) => { + if (run.ended) return + send({ t: 'done', ...ids, exit: defect(error) }) + end(run) + }) + worker.on('exit', () => { + if (run.ended) return + send({ t: 'done', ...ids, exit: defect('the program worker exited without a result') }) + end(run) + }) + run.ping = setInterval(() => { + if (run.ended) return + if (Date.now() - run.answered > deadline) { + // A busy loop fails its own step once instead of looping across restarts. + end(run) + return send({ t: 'unresponsive', ...ids, ms: deadline }) + } + run.worker.postMessage({ t: 'ping' }) + }, Math.max(10, Math.floor(deadline / 4))) + } + + const lines = createInterface({ input: process.stdin }) + lines.on('line', (line) => { + const message = JSON.parse(line) + if (message.t === 'ping') return send({ t: 'pong', seq: message.seq }) + if (message.t === 'start') return start(message) + const run = runs.get(message.step) + if (run === undefined || run.attempt !== message.attempt) return // another attempt's message + if (message.t === 'cancel') return end(run) + run.worker.postMessage(message) // ack, nack, resolved + }) + // The daemon is gone: its replacement resumes every program from the journal. End all effects now. + lines.on('close', () => { + for (const run of [...runs.values()]) end(run) + process.exit(0) + }) +} + +const worker = async () => { + const start = workerData + /** By sequence number: journal entries waiting for ack, requests waiting for `resolved`. */ + const waiting = new Map() + const spawns = new Map() + let seq = 0 + let spawned = 0 + const controller = new AbortController() + parentPort.on('message', (message) => { + if (message.t === 'ping') return parentPort.postMessage({ t: 'pong' }) + if (message.t === 'spawned') { + const pending = spawns.get(message.id) + spawns.delete(message.id) + return message.error === undefined ? pending?.resolve(message.result) : pending?.reject(new Error(message.error)) + } + const pending = waiting.get(message.seq) + waiting.delete(message.seq) + if (message.t === 'ack') pending?.resolve() + else if (message.t === 'nack') pending?.reject(new Error(message.error)) + else if (message.t === 'resolved') pending?.resolve(message.exit) + }) + /** Ask the daemon and wait: a journal entry is durable only once the daemon acknowledges it. */ + const request = (message) => + new Promise((resolve, reject) => { + const n = ++seq + waiting.set(n, { resolve, reject }) + parentPort.postMessage({ ...message, seq: n }) + }) + let exit + try { + const actual = createHash('sha256').update(readFileSync(start.module)).digest('hex') + if (actual !== start.sha256) throw new Error(`bundle ${start.module} has sha256 ${actual}, the run pins ${start.sha256}`) + const program = (await import(start.module)).default?.[ENTRY] + if (typeof program !== 'function') throw new Error(`${start.module} does not export a Fractal program`) + exit = await program({ + step: start.step, + attempt: start.attempt, + run: start.run, + payload: start.payload, + env: start.env, + cwd: start.cwd, + journal: new Map( + start.journal.map((entry) => [ + entry.key, + entry._tag === 'Success' ? { _tag: 'Success', value: entry.value } : { _tag: 'Failure', error: entry.error }, + ]), + ), + signal: controller.signal, + started: (key) => parentPort.postMessage({ t: 'journal', key, status: 'started' }), + record: (entry) => request({ t: 'journal', ...entry }), + child: (child) => request({ t: 'run', key: child.key, mission: child.mission, input: child.input }), + ask: (ask) => request({ t: 'ask', ...ask }), + spawn: (argv, options) => + new Promise((resolve, reject) => { + const id = ++spawned + spawns.set(id, { resolve, reject }) + parentPort.postMessage({ t: 'spawn', id, argv, options }) + }), + }) + } catch (error) { + exit = defect(error) + } + parentPort.postMessage({ t: 'done', exit }) +} + +if (isMainThread) main() +else void worker() diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs b/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs new file mode 100644 index 00000000000..71dbb9c7f1e --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs @@ -0,0 +1,221 @@ +//! The program engine with a real Bun host (docs/st3/program-engine.md): journal and replay +//! across a host crash, eager cancellation from the run's evaluation, and worker liveness. +//! Program steps need the runtime: these tests fail, never skip, when no `bun` is on `PATH`. +use super::*; +use crate::reconcile::program::{JOURNAL_KIND, Liveness, ProgramEngine, RESULT_KEY}; + +/// A hand-written bundle against HostContext v1: one journaled step, then either a busy loop or +/// an effect that waits for a gate file while a background child shares its process group. +const BUNDLE: &str = r#" +const step = async (ctx, key, effect) => { + const replayed = ctx.journal.get(key) + if (replayed !== undefined) return replayed.value + ctx.started(key) + const at = Date.now() + const value = await effect() + await ctx.record({ key, status: 'completed', value, ms: Date.now() - at }) + return value +} +export default { + [Symbol.for('fractal.program.v1')]: async (ctx) => { + const first = await step(ctx, 'first', async () => ({ attempt: ctx.attempt })) + if (ctx.payload.mode === 'busy') for (;;) {} + const waited = await step(ctx, 'wait', async () => { + const script = 'sleep 1000 & echo $$ $! > "$PIDS.tmp"; mv "$PIDS.tmp" "$PIDS"; while [ ! -e "$GATE" ]; do sleep 0.05; done; echo opened' + const out = await ctx.spawn(['sh', '-c', script], { env: { PIDS: ctx.payload.pids, GATE: ctx.payload.gate } }) + return out.stdout.trim() + }) + return { _tag: 'Success', value: { first, waited } } + }, +} +"#; + +struct Fixture { + store: Arc, + reconciler: Reconciler, + engine: Arc, + dir: tempfile::TempDir, +} + +struct Started { + run: String, + step: String, + pids: std::path::PathBuf, + gate: std::path::PathBuf, +} + +fn fixture(liveness: Liveness) -> Fixture { + let dir = tempfile::tempdir().unwrap(); + let store = Arc::new(Store::open_memory("node").unwrap()); + let bundle = dir.path().join("program.mjs"); + std::fs::write(&bundle, BUNDLE).unwrap(); + let hash = store.put_blob(BUNDLE.as_bytes()).unwrap(); + apply_source( + &store, + &format!( + r#"version 2 +mission "program-test" state="ready" {{ + goal "Run a program." + concurrent-runs max=4 + input "mode" kind="text" + input "pids" kind="text" + input "gate" kind="text" + step "program" {{ program "{}" sha256="{hash}" }} +}}"#, + bundle.display() + ), + "program-test-definition", + ); + let reconciler = Reconciler::new(store.clone(), Arc::new(FakeRuntime::default()), "node".into(), Arc::new(Notify::new())); + let engine = ProgramEngine::new(store.clone(), Arc::new(|_: &str| {}), dir.path().join("programs"), liveness); + assert!(reconciler.program_engine.set(engine.clone()).is_ok()); + Fixture { store, reconciler, engine, dir } +} + +impl Fixture { + fn start(&self, name: &str, mode: &str) -> Started { + let pids = self.dir.path().join(format!("{name}.pids")); + let gate = self.dir.path().join(format!("{name}.gate")); + let inputs = BTreeMap::from([ + ("mode".to_owned(), mode.to_owned()), + ("pids".to_owned(), pids.display().to_string()), + ("gate".to_owned(), gate.display().to_string()), + ]); + let run = self + .store + .create_mission_run(&MissionRunRequest { + mission: "program-test".into(), + revision: None, + workspace: self.dir.path().display().to_string(), + requester: None, + mode: None, + inputs, + idempotency_key: format!("program-test-{name}"), + }) + .unwrap(); + let step = run.steps.iter().find(|step| step.step == "program").unwrap().subject.clone(); + Started { run: run.subject, step, pids, gate } + } + + /// Run passes until `done` holds; the engine itself works off the reconciler thread. + fn until(&self, what: &str, done: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(60); + while !done() { + assert!(Instant::now() < deadline, "timed out waiting until {what}"); + self.reconciler.reconcile_once().unwrap(); + std::thread::sleep(Duration::from_millis(50)); + } + } + + /// Wait without any reconciler pass: what happens now is the engine's and the host's alone. + fn without_passes(&self, what: &str, done: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(10); + while !done() { + assert!(Instant::now() < deadline, "timed out waiting until {what}"); + std::thread::sleep(Duration::from_millis(20)); + } + } + + fn status(&self, step: &str) -> String { + self.store.step_run(step).unwrap().unwrap().status + } + + /// `(key, status)` of each journal entry, in store order. + fn journal(&self, step: &str) -> Vec<(String, String)> { + let mut claims = self.store.claims_for(step, Some(JOURNAL_KIND)).unwrap(); + claims.sort_by_key(|claim| claim.store_index); + claims + .iter() + .map(|claim| { + let fields = &claim.body["fields"]; + (fields["key"].as_str().unwrap().to_owned(), fields["status"].as_str().unwrap().to_owned()) + }) + .collect() + } +} + +fn pids(path: &std::path::Path) -> Option> { + let text = std::fs::read_to_string(path).ok()?; + Some(text.split_whitespace().map(|pid| pid.parse().unwrap()).collect()) +} + +fn alive(pid: i32) -> bool { + // SAFETY: signal 0 only checks that the process exists. + unsafe { libc::kill(pid, 0) == 0 } +} + +fn entries(pairs: &[(&str, &str)]) -> Vec<(String, String)> { + pairs.iter().map(|(key, status)| ((*key).to_owned(), (*status).to_owned())).collect() +} + +#[test] +fn a_program_resumes_from_its_journal_after_its_host_is_killed() { + let fixture = fixture(Liveness::default()); + let started = fixture.start("resume", "wait"); + fixture.until("the effect runs", || pids(&started.pids).is_some()); + let before = pids(&started.pids).unwrap(); + assert_eq!(fixture.journal(&started.step), entries(&[("first", "started"), ("first", "completed"), ("wait", "started")])); + let host = fixture.engine.host_pid().unwrap(); + std::fs::remove_file(&started.pids).unwrap(); + // SAFETY: the host is this test's own child process. + assert_eq!(unsafe { libc::kill(host as i32, libc::SIGKILL) }, 0); + // A new host resumes the attempt: `first` replays from the journal and `wait` runs again. + fixture.without_passes("the effect runs on a new host", || pids(&started.pids).is_some()); + assert_ne!(fixture.engine.host_pid(), Some(host)); + // The killed host's effect group, leader and background child, ended before the new host started. + assert!(before.iter().all(|pid| !alive(*pid)), "survivors of the killed host: {before:?}"); + std::fs::write(&started.gate, "").unwrap(); + fixture.until("the program completes", || fixture.status(&started.step) == "completed"); + assert_eq!( + fixture.journal(&started.step), + entries(&[ + ("first", "started"), + ("first", "completed"), + ("wait", "started"), + ("wait", "started"), + ("wait", "completed"), + (RESULT_KEY, "completed"), + ]) + ); + let result = fixture.store.claims_for(&started.step, Some(JOURNAL_KIND)).unwrap().into_iter().max_by_key(|claim| claim.store_index).unwrap(); + assert_eq!(result.body["fields"]["value"], json!({"first": {"attempt": 1}, "waited": "opened"})); +} + +#[test] +fn a_step_that_leaves_its_attempt_stops_its_program_and_effects_at_once() { + let fixture = fixture(Liveness::default()); + let cancelled = fixture.start("cancelled", "wait"); + let failed = fixture.start("failed", "wait"); + fixture.until("both effects run", || pids(&cancelled.pids).is_some() && pids(&failed.pids).is_some()); + let group = |started: &Started| pids(&started.pids).unwrap(); + let (cancelled_group, failed_group) = (group(&cancelled), group(&failed)); + fixture.store.request_mission_run_cancellation(&cancelled.run, "not wanted").unwrap(); + fixture.store.set_step_state(&failed.step, "failed", Some("the active execution timeout expired")).unwrap(); + // One evaluation of each run tells the engine; nothing waits for a journal write. + fixture.reconciler.evaluate_changed_mission_runs().unwrap(); + fixture.without_passes("both effect groups end", || cancelled_group.iter().chain(&failed_group).all(|pid| !alive(*pid))); + std::fs::write(&cancelled.gate, "").unwrap(); + std::fs::write(&failed.gate, "").unwrap(); + std::thread::sleep(Duration::from_millis(300)); + for started in [&cancelled, &failed] { + assert_eq!(fixture.journal(&started.step), entries(&[("first", "started"), ("first", "completed"), ("wait", "started")])); + assert_ne!(fixture.status(&started.step), "completed"); + } +} + +#[test] +fn an_unresponsive_worker_fails_only_its_own_step() { + let fixture = fixture(Liveness { worker_deadline: Duration::from_millis(1_000), ..Liveness::default() }); + let busy = fixture.start("busy", "busy"); + let waiting = fixture.start("waiting", "wait"); + fixture.until("the busy program fails", || fixture.status(&busy.step) == "failed"); + let reason = fixture.store.latest_claim(&busy.step, Some("step-run.state")).unwrap().unwrap().body["fields"]["reason"].clone(); + assert_eq!(reason, "the program's worker missed its 1000 ms liveness deadline and was terminated"); + assert_eq!(fixture.engine.host_fault(&busy.step, 1).as_deref(), reason.as_str()); + let host = fixture.engine.host_pid(); + fixture.until("the other program's effect runs", || pids(&waiting.pids).is_some()); + std::fs::write(&waiting.gate, "").unwrap(); + fixture.until("the other program completes", || fixture.status(&waiting.step) == "completed"); + // The host itself stayed up: only the busy worker was terminated. + assert_eq!(fixture.engine.host_pid(), host); +} diff --git a/docs/st3/program-engine.md b/docs/st3/program-engine.md index 5a9aed9cc87..1e510c506b8 100644 --- a/docs/st3/program-engine.md +++ b/docs/st3/program-engine.md @@ -135,7 +135,7 @@ can run an effect to its end but can never journal it. | Direction | Message | Fields | Meaning | |---|---|---|---| -| daemon → host | `start` | `step`, `attempt`, `module`, `sha256`, `payload`, `env`, `cwd`, `journal[]` | start or resume one program step | +| daemon → host | `start` | `step`, `attempt`, `run`, `module`, `sha256`, `payload`, `env`, `cwd`, `journal[]` | start or resume one program step | | host → daemon | `journal` | `step`, `attempt`, `seq`, `key`, `status` (`started`/`completed`/`failed`), `value`/`error`, `ms` | an inner step began or ended | | daemon → host | `ack` | `step`, `attempt`, `seq` | the completed/failed entry is durable (R01) | | daemon → host | `nack` | `step`, `attempt`, `seq`, `error` | the entry could not be written; the program fails with a defect | @@ -154,9 +154,18 @@ the next host resumes each from its journal. A bundle is one self-contained ES module whose default export carries the entry `Symbol.for('fractal.program.v1')`. The program API (`Fractal.mission/step/command`) and Effect are inside the bundle, so the host is version-free and the bundle hash pins the runtime library -too. The host verifies the hash before loading and runs each program attempt in its own worker. It -starts each step effect in its own process group and kills those groups on `cancel` and when its -stdin closes; groups that outlive a crashed host or daemon are ended by the next daemon (DQ2). +too. The engine reads the module from the store's blob of the revision's hash into a verified, +content-addressed file; the host verifies the hash again before loading and runs each program +attempt in its own worker, while its main thread owns the pipe, the effect process groups and +liveness. It starts each step effect in its own process group and kills those groups when the +group's leader exits, on `cancel`, when it terminates the attempt's worker (an ended or +unresponsive program) and when its stdin closes; groups that outlive a crashed host or daemon are +ended by the next daemon (DQ2). + +The daemon starts the host with `FRACTAL_PROGRAM_RUNTIME`, or else `bun` on its `PATH`; a member +without either cannot run program steps. Cancellation is eager: every evaluation of a run tells the +engine which of the run's program attempts left their step (ended, failed, timed out, retried, +cancelled with the run, or replaced by a new generation), and the engine sends `cancel` at once. ### 3. Declaration and authority @@ -296,7 +305,8 @@ Choice: a child run. It needs no standing owner and reuses §6. - `missions waiting-on` names the innermost open key: a running effect, a child run or an ask. - Stalls: a program step whose journal shows no new entry within its `stalled-after` raises the run fault with scope `program/stalled`, naming the open key. A dead host that the engine cannot - restart raises `program/host` on every program step it held. + restart raises `program/host` on every program step it held, and so does a program worker the + host terminated for missing its liveness deadline (DQ8), on that worker's step. ### 9. One execution model diff --git a/flake.nix b/flake.nix index ca930d50ea2..180a66cdd9c 100644 --- a/flake.nix +++ b/flake.nix @@ -597,6 +597,8 @@ # Boot cleanup fixtures query their isolated daemon with curl. pkgs.curl pkgs.nodejs + # Program steps run on a Bun host; the engine tests start a real one. + pkgs.bun ptyPackage effect-utils.packages.${system}.otelite messagingCompat @@ -2200,6 +2202,8 @@ # st3's messaging fault matrix runs the omp channel hook (TypeScript) under the # provider stand-in with Node's built-in type stripping, which Node 24 enables. pkgs.nodejs + # Program steps run on a Bun host; the engine tests start a real one. + pkgs.bun # st3's completion tests drive the installed stub in each supported shell. pkgs.fish pkgs.zsh diff --git a/release-notes/program-engine-host.json b/release-notes/program-engine-host.json new file mode 100644 index 00000000000..0df6e506d83 --- /dev/null +++ b/release-notes/program-engine-host.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "The daemon runs `program` steps: one supervised Bun host per member runs each program attempt in its own worker, journals each inner step as a `program.step` claim, replays after host or daemon restarts, cancels eagerly from the run's evaluation, ends recorded effect process groups a dead host left, and fails a step whose worker misses its liveness deadline", + "replay": {"status": "changed", "detail": "The engine writes `program.step` claims (registered in the declaration change) on program step runs: one fenced result per key and attempt (idempotency key program:STEP:ATTEMPT:KEY:result) and the reserved `$result` key with the program's success value. Replay reads the current attempt's completed and failed entries in pages of 500."}, + "schema": {"status": "none", "detail": "No table or migration. The engine keeps pinned bundles, the host script and effect process-group records under the daemon's drivers/programs state directory."}, + "rules": {"status": "none", "detail": "No rule or authority change; journal claims are written by the daemon (system-only). Mission-run authority for effects is a separate change."}, + "compatibility": {"status": "changed", "detail": "A program step needs FRACTAL_PROGRAM_RUNTIME or `bun` on the daemon's PATH; without either the step keeps its state and the engine reports the host failure. Liveness: a worker that does not answer the host's ping within 30 s is terminated and its step fails; a host that does not answer the daemon's ping within 30 s is replaced and its programs resume. The development shell and the Rust check environment now include Bun."}, + "downtime": {"status": "none", "detail": "No restart or migration beyond a normal deploy; programs interrupted by the deploy resume from their journals."}, + "manual": {"status": "none", "detail": "No operator action is required on members that run no program steps."}, + "recovery": {"status": "none", "detail": "A crashed or replaced host is restarted and each program resumes from its journal; effect groups it left are killed only after their leader's boot id and start time are verified."} +} From a537eda54cc45a7856256f1001b9ce346aba5a3a Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 15:35:20 +0100 Subject: [PATCH 2/6] Import Instant in the program engine tests --- crates/fractal-reconcile/src/reconcile/tests/program_engine.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs b/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs index 71dbb9c7f1e..e42f8293e58 100644 --- a/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs +++ b/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs @@ -3,6 +3,7 @@ //! Program steps need the runtime: these tests fail, never skip, when no `bun` is on `PATH`. use super::*; use crate::reconcile::program::{JOURNAL_KIND, Liveness, ProgramEngine, RESULT_KEY}; +use std::time::Instant; /// A hand-written bundle against HostContext v1: one journaled step, then either a busy loop or /// an effect that waits for a gate file while a background child shares its process group. From efe0519c773a783db2a28b76a6d16e30a694d748 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 15:36:33 +0100 Subject: [PATCH 3/6] Import json in the program engine tests --- crates/fractal-reconcile/src/reconcile/tests/program_engine.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs b/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs index e42f8293e58..f45d5b266fa 100644 --- a/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs +++ b/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs @@ -3,6 +3,7 @@ //! Program steps need the runtime: these tests fail, never skip, when no `bun` is on `PATH`. use super::*; use crate::reconcile::program::{JOURNAL_KIND, Liveness, ProgramEngine, RESULT_KEY}; +use serde_json::json; use std::time::Instant; /// A hand-written bundle against HostContext v1: one journaled step, then either a busy loop or From 20a7701b2988762d1273fc78c84f28fe39602be3 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 15:47:50 +0100 Subject: [PATCH 4/6] Tie engine state changes to the sending attempt and host A late message for an earlier attempt or a replaced host no longer removes the current attempt; ack, nack and cancel go only to the host generation that sent the entry while its attempt still runs; refused completed/failed entries (reserved key, missing key or status) are nacked. DQ2 reaping treats a leader as gone only on ESRCH, keeps records whose live leader's start time cannot be read, verifies half-renamed .tmp records, and uses absolute ps/sysctl paths on macOS. --- .../src/reconcile/program.rs | 108 ++++++++++++------ .../src/reconcile/program/tests.rs | 8 ++ 2 files changed, 81 insertions(+), 35 deletions(-) diff --git a/crates/fractal-reconcile/src/reconcile/program.rs b/crates/fractal-reconcile/src/reconcile/program.rs index 246c976b04e..c03d843a741 100644 --- a/crates/fractal-reconcile/src/reconcile/program.rs +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -354,11 +354,8 @@ impl ProgramEngine { let mut killed = 0; for entry in std::fs::read_dir(&dir)? { let path = entry?.path(); - if path.extension().and_then(|extension| extension.to_str()) != Some("json") { - // A record the host was still writing when it died names a group it never ran. - std::fs::remove_file(&path)?; - continue; - } + // A `.tmp` record is one the host was renaming into place when it died: its group was + // already running, so it is verified like any other record. let record: Value = std::fs::read(&path) .ok() .and_then(|bytes| serde_json::from_slice(&bytes).ok()) @@ -367,12 +364,14 @@ impl ProgramEngine { if let Some(pid) = pid && record["boot_id"].as_str() == Some(boot.as_str()) { - let leader = start_time(pid); - let ours = match (&leader, record["start_time"].as_str()) { - (Some(now), Some(recorded)) => now == recorded, + let ours = match (leader_gone(pid), start_time(pid), record["start_time"].as_str()) { // The leader is gone: anything left in its group is the effect's. - (None, _) => true, - (Some(_), None) => false, + (true, _, _) => true, + (false, Some(now), Some(recorded)) => now == recorded, + // A live leader whose start time cannot be read is never signalled; the + // record stays for the next host start. + (false, None, _) => continue, + (false, Some(_), None) => false, }; // SAFETY: signalling a process group has no memory effects. if ours && unsafe { libc::kill(-pid, libc::SIGKILL) } == 0 { @@ -430,9 +429,9 @@ impl ProgramEngine { }; let attempt = u32::try_from(attempt)?; match message["t"].as_str() { - Some("journal") => self.journal_entry(step, attempt, &program, message), - Some("done") => self.done(step, attempt, &program, message), - Some("unresponsive") => self.unresponsive(step, attempt, message), + Some("journal") => self.journal_entry(step, attempt, generation, &program, message), + Some("done") => self.done(step, attempt, generation, &program, message), + Some("unresponsive") => self.unresponsive(step, attempt, generation, message), other => Err(anyhow!("unknown message type {other:?}")), } } @@ -448,11 +447,12 @@ impl ProgramEngine { /// The backstop to [`Self::release_left`]: stop the program if its step left this attempt. /// True when it was stopped. - fn stop_if_left(&self, step: &str, attempt: u32) -> Result { + fn stop_if_left(&self, step: &str, attempt: u32, generation: u64) -> Result { let Some(reason) = self.left(step, attempt)? else { return Ok(false) }; let mut inner = self.lock(); - inner.running.remove(step); - send(&inner, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + if forget(&mut inner, step, attempt, generation) { + send_to(&inner, generation, &json!({"t": "cancel", "step": step, "attempt": attempt, "reason": reason})); + } Ok(true) } @@ -494,34 +494,44 @@ impl ProgramEngine { }) } - fn journal_entry(&self, step: &str, attempt: u32, program: &str, message: &Value) -> Result<()> { - if self.stop_if_left(step, attempt)? { + fn journal_entry(&self, step: &str, attempt: u32, generation: u64, program: &str, message: &Value) -> Result<()> { + if self.stop_if_left(step, attempt, generation)? { return Ok(()); } - let key = message["key"].as_str().ok_or_else(|| anyhow!("journal entry without a key"))?; - let status = message["status"].as_str().ok_or_else(|| anyhow!("journal entry without a status"))?; - if key == RESULT_KEY { - return Err(anyhow!("a program may not journal the reserved key {RESULT_KEY}")); - } - let written = self.write_journal(step, attempt, program, key, status, message, None); - if status == "started" { - return written.map(|_| ()).map_err(Into::into); + let key = message["key"].as_str().unwrap_or_default(); + let status = message["status"].as_str().unwrap_or_default(); + let written = if key.is_empty() || !["started", "completed", "failed"].contains(&status) { + Err(format!("a journal entry needs a key and a status, got key `{key}` status `{status}`")) + } else if key == RESULT_KEY { + Err(format!("a program may not journal the reserved key {RESULT_KEY}")) + } else { + self.write_journal(step, attempt, program, key, status, message, None).map(|_| ()).map_err(|error| error.to_string()) + }; + let seq = &message["seq"]; + if status == "started" || seq.is_null() { + return written.map_err(|error| anyhow!(error)); } let inner = self.lock(); - let seq = &message["seq"]; + // Only the attempt and host that sent the entry hear about it (R01). + if !matches!(inner.running.get(step), Some(running) if running.dispatch.attempt == attempt && running.generation == generation) { + return Ok(()); + } match written { - Ok(_) => send(&inner, &json!({"t": "ack", "step": step, "attempt": attempt, "seq": seq})), + Ok(()) => send_to(&inner, generation, &json!({"t": "ack", "step": step, "attempt": attempt, "seq": seq})), // The program must not continue past a result that is not durable: the step fails. - Err(error) => send( + Err(error) => send_to( &inner, + generation, &json!({"t": "nack", "step": step, "attempt": attempt, "seq": seq, "error": format!("journal write refused: {error}")}), ), } Ok(()) } - fn done(&self, step: &str, attempt: u32, program: &str, message: &Value) -> Result<()> { - self.lock().running.remove(step); + fn done(&self, step: &str, attempt: u32, generation: u64, program: &str, message: &Value) -> Result<()> { + if !forget(&mut self.lock(), step, attempt, generation) { + return Ok(()); + } if self.left(step, attempt)?.is_some() { return Ok(()); } @@ -541,12 +551,14 @@ impl ProgramEngine { /// The host terminated this attempt's worker after it missed its liveness deadline (DQ8): the /// step fails once instead of looping across restarts. - fn unresponsive(&self, step: &str, attempt: u32, message: &Value) -> Result<()> { + fn unresponsive(&self, step: &str, attempt: u32, generation: u64, message: &Value) -> Result<()> { let ms = message["ms"].as_u64().unwrap_or_default(); let reason = format!("the program's worker missed its {ms} ms liveness deadline and was terminated"); { let mut inner = self.lock(); - inner.running.remove(step); + if !forget(&mut inner, step, attempt, generation) { + return Ok(()); + } inner.faults.insert(step.to_owned(), (attempt, reason.clone())); } eprintln!("fractal: program {step} attempt {attempt}: {reason}"); @@ -638,6 +650,32 @@ fn supervise(engine: Weak, every: Duration) { } } +/// Forget `step` only while this engine still runs exactly that attempt on that host generation, +/// so a late message never removes a newer attempt. True when it did. +fn forget(inner: &mut Inner, step: &str, attempt: u32, generation: u64) -> bool { + let current = inner + .running + .get(step) + .is_some_and(|running| running.dispatch.attempt == attempt && running.generation == generation); + if current { + inner.running.remove(step); + } + current +} + +/// Queue one line for host `generation` only; a replaced host's messages never reach its successor. +fn send_to(inner: &Inner, generation: u64, message: &Value) { + if inner.host.as_ref().is_some_and(|host| host.generation == generation) { + send(inner, message); + } +} + +/// True only when no process has this pid (`ESRCH`); any other answer keeps the leader alive. +fn leader_gone(pid: i32) -> bool { + // SAFETY: signal 0 only checks that the process exists. + unsafe { libc::kill(pid, 0) } == -1 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) +} + /// Queue one line for the live host. A host that is gone drops it; its programs resume on the next. fn send(inner: &Inner, message: &Value) { if let Some(host) = inner.host.as_ref() { @@ -691,7 +729,7 @@ fn boot_id() -> Result { } #[cfg(not(target_os = "linux"))] { - let output = Command::new("sysctl").args(["-n", "kern.boottime"]).output()?; + let output = Command::new("/usr/sbin/sysctl").args(["-n", "kern.boottime"]).output()?; anyhow::ensure!(output.status.success(), "sysctl kern.boottime failed"); Ok(String::from_utf8_lossy(&output.stdout).trim().to_owned()) } @@ -706,7 +744,7 @@ fn start_time(pid: i32) -> Option { } #[cfg(not(target_os = "linux"))] { - let output = Command::new("ps").args(["-o", "lstart=", "-p", &pid.to_string()]).output().ok()?; + let output = Command::new("/bin/ps").args(["-o", "lstart=", "-p", &pid.to_string()]).output().ok()?; let text = String::from_utf8_lossy(&output.stdout).trim().to_owned(); (!text.is_empty()).then_some(text) } diff --git a/crates/fractal-reconcile/src/reconcile/program/tests.rs b/crates/fractal-reconcile/src/reconcile/program/tests.rs index 5fe8c6b44a3..bda84a45fa1 100644 --- a/crates/fractal-reconcile/src/reconcile/program/tests.rs +++ b/crates/fractal-reconcile/src/reconcile/program/tests.rs @@ -118,6 +118,12 @@ fn messages_for_another_attempt_or_host_change_nothing() { assert_eq!(journal_claims(&engine, step), 0); assert!(engine.lock().running.contains_key(step)); assert_eq!(engine.host_fault(step, 2), None); + // A late message for the earlier attempt never removes the attempt that replaced it, and a + // refused entry is answered rather than left waiting (no host here, so nothing is sent). + engine.done(step, 1, 1, &"a".repeat(64), &message(1, "done")).unwrap(); + engine.unresponsive(step, 1, 1, &message(1, "unresponsive")).unwrap(); + assert!(engine.lock().running.contains_key(step)); + assert_eq!(engine.host_fault(step, 1), None); } #[test] @@ -137,6 +143,8 @@ fn recorded_groups_are_killed_only_when_their_leader_is_verified() { // Recorded by this boot with the leader's start time: an effect a dead host left behind. let mut ours = leader(); record(&ours, &boot, start_time(ours.id() as i32)); + // The host died between writing this record and renaming it into place: the group runs. + std::fs::rename(groups.join(format!("{}.json", ours.id())), groups.join(format!("{}.json.tmp", ours.id()))).unwrap(); // The same pid with another start time is a process that reused it. let mut reused = leader(); record(&reused, &boot, Some("0".into())); From 07f6ab3e6a680cf5fd5c2fedc947bdf2d01706ca Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 15:49:24 +0100 Subject: [PATCH 5/6] Bind the existence check before comparing it --- crates/fractal-reconcile/src/reconcile/program.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/fractal-reconcile/src/reconcile/program.rs b/crates/fractal-reconcile/src/reconcile/program.rs index c03d843a741..74665a83de8 100644 --- a/crates/fractal-reconcile/src/reconcile/program.rs +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -673,7 +673,8 @@ fn send_to(inner: &Inner, generation: u64, message: &Value) { /// True only when no process has this pid (`ESRCH`); any other answer keeps the leader alive. fn leader_gone(pid: i32) -> bool { // SAFETY: signal 0 only checks that the process exists. - unsafe { libc::kill(pid, 0) } == -1 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) + let missing = unsafe { libc::kill(pid, 0) } == -1; + missing && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) } /// Queue one line for the live host. A host that is gone drops it; its programs resume on the next. From 9fb2ba4c7d9da5c6c278e28359b16cd78cace1c6 Mon Sep 17 00:00:00 2001 From: Johannes Schickling Date: Tue, 10 Nov 2026 16:23:34 +0100 Subject: [PATCH 6/6] Register the program host source in the large-value inventory --- crates/fractal-reconcile/src/reconcile/program.rs | 2 +- docs/st3/large-value-sites.json | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/fractal-reconcile/src/reconcile/program.rs b/crates/fractal-reconcile/src/reconcile/program.rs index 74665a83de8..815817283fe 100644 --- a/crates/fractal-reconcile/src/reconcile/program.rs +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -39,7 +39,7 @@ pub const RESULT_KEY: &str = "$result"; /// The step states in which the engine runs a program step's current attempt. A step waiting on /// a person keeps its program loaded (§11). pub const RUNNING_STATES: [&str; 4] = ["ready", "working", "blocked", "waiting-person"]; -const HOST_SOURCE: &str = include_str!("program_host.mjs"); +const HOST_SOURCE: &str = /* sqlite-large-value: allow-for-now lv-h344eb4debbcf owner=fractal_reconcile::reconcile::program TODO=trace-statement-and-value-representation */ include_str!("program_host.mjs"); /// How long a crashed host waits before its replacement starts, so a host that cannot start does /// not spin. const HOST_RESTART_DELAY: Duration = Duration::from_millis(500); diff --git a/docs/st3/large-value-sites.json b/docs/st3/large-value-sites.json index e20ccceb97b..0624ecb1c95 100644 --- a/docs/st3/large-value-sites.json +++ b/docs/st3/large-value-sites.json @@ -1678,6 +1678,7 @@ {"id": "lv-h341199aca241", "path": "crates/fractal-seats/src/store/reconcile_work_context/tests.rs", "kind": "sql-template", "fingerprint": "c26ae3724e16e5c700f4b62ee9634ab5ad4ff5b469c40cd1ff641134be24cc2e", "owner": "fractal_seats::store::reconcile_work_context::tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 548, "excerpt": "UPDATE mission_runs SET phase=?1"}, {"id": "lv-h3411e5a95aa3", "path": "crates/fractal-store/src/store.rs", "kind": "sql-template", "fingerprint": "ee05258ea1abb951413a0b71a4bf6a58fc669d5378455d5f208d58212ee44132", "owner": "fractal_store::store", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 4405, "excerpt": "INSERT INTO client_diagnostic_receipts(device,event_id,accepted_at_unix_ms) VALUES (?1,?2,?3)"}, {"id": "lv-h342a097047a2", "path": "crates/fractal-store/src/store/usage_credits/tests.rs", "kind": "sql-template", "fingerprint": "26fa646c5c653b98318b6c8f4137ffd58b331a8f8b2d03cc8526784c70daf09d", "owner": "fractal_store::store::usage_credits::tests", "todo": "prepare-body-and-predecessor-values", "disposition": "allow-for-now", "line_hint": 89, "excerpt": "INSERT INTO local_usage_credit_pending SELECT id FROM claims WHERE kind='usage.limits.observed'"}, + {"id": "lv-h344eb4debbcf", "path": "crates/fractal-reconcile/src/reconcile/program.rs", "kind": "included-source", "fingerprint": "14bc51f07a0ad35c7aa75195f68b8c33031b72ab5b6719b7bff00753450e6501", "owner": "fractal_reconcile::reconcile::program", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 42, "excerpt": "include_str ! ( \"program_host.mjs\" )"}, {"id": "lv-h3452c033a82d", "path": "crates/fractal-api-client-v0/src/api/client_v0/observer_subscription_detail_tests.rs", "kind": "sql-template", "fingerprint": "74721c8b24e28aea05585fb9a62bfc30baf41b32584982467f4eabe19d4c8836", "owner": "fractal_api_client_v0::api::client_v0::observer_subscription_detail_tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 743, "excerpt": "UPDATE desired SET owner_run=NULL WHERE subject=?1"}, {"id": "lv-h3458778f2d90", "path": "crates/fractal-store/src/store/local_trim_tests.rs", "kind": "sql-template", "fingerprint": "8faf044a2a9cc7ec4a2a8c3be41da9b789b4d6eee03d27c6f796564528beaf97", "owner": "fractal_store::store::local_trim_tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 1934, "excerpt": "INSERT OR IGNORE INTO local_trim_citations(cited, citer) VALUES (?1, ?2)"}, {"id": "lv-h345fc67acafe", "path": "crates/fractal-attention/src/store/attention.rs", "kind": "sql-template", "fingerprint": "4365d3b5b8a68200370e8fe8e010a45a20592286a9600d5265474a4e1e287927", "owner": "fractal_attention::store::attention", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 195, "excerpt": "DELETE FROM desired WHERE owner_run=?1 AND subject NOT LIKE 'resource/uri/%'"},