diff --git a/crates/fractal-reconcile/src/reconcile.rs b/crates/fractal-reconcile/src/reconcile.rs index 52bf4ce18eb..1127ab259fc 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()), @@ -1857,6 +1861,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()), @@ -10601,6 +10606,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") { @@ -11489,9 +11499,8 @@ impl Reconciler { if !self.step_declarations_hold(run, &view.subject)? { return Ok(changed); } - // The program engine alone decides a program step's outcome: nothing here - // completes it, so it stays working until the engine runs it. - if step.spec.program.is_some() { + if let Some(program) = &step.spec.program { + changed |= self.dispatch_program(run, &step, view, program)?; return Ok(changed); } if !view.agentless && !view.worker_reported { @@ -14365,6 +14374,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, @@ -21184,7 +21240,7 @@ pub mod tests { mod start_preemption; mod seat_admission_queue; mod terminal_step_guard; - mod program_step_hold; + 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..815817283fe --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/program.rs @@ -0,0 +1,764 @@ +//! 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 = /* 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); +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(); + // 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()) + .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 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. + (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 { + 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, 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:?}")), + } + } + + /// 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, generation: u64) -> Result { + let Some(reason) = self.left(step, attempt)? else { return Ok(false) }; + let mut inner = self.lock(); + if forget(&mut inner, step, attempt, generation) { + send_to(&inner, generation, &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, generation: u64, program: &str, message: &Value) -> Result<()> { + if self.stop_if_left(step, attempt, generation)? { + return Ok(()); + } + 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(); + // 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_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_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, 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(()); + } + 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, 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(); + 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}"); + 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(); + } +} + +/// 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. + 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. +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("/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()) + } +} + +/// 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("/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) + } +} + +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..bda84a45fa1 --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/program/tests.rs @@ -0,0 +1,164 @@ +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); + // 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] +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 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())); + // 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..f45d5b266fa --- /dev/null +++ b/crates/fractal-reconcile/src/reconcile/tests/program_engine.rs @@ -0,0 +1,223 @@ +//! 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}; +use serde_json::json; +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. +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/crates/fractal-reconcile/src/reconcile/tests/program_step_hold.rs b/crates/fractal-reconcile/src/reconcile/tests/program_step_hold.rs deleted file mode 100644 index 6d72292fe08..00000000000 --- a/crates/fractal-reconcile/src/reconcile/tests/program_step_hold.rs +++ /dev/null @@ -1,37 +0,0 @@ -//! A program step's outcome is the program engine's: no reconciler pass completes it. -use super::*; - -#[test] -fn a_published_program_step_stays_working_until_the_engine_decides() { - let store = Arc::new(Store::open_memory("node").unwrap()); - let hash = store.put_blob(b"export default {}").unwrap(); - apply_source( - &store, - &format!( - r#"version 2 -mission "program-hold" state="ready" {{ - goal "Run a program." - step "program" {{ program "/bundles/hold.mjs" sha256="{hash}" }} -}}"# - ), - "program-hold-definition", - ); - let run = store - .create_mission_run(&MissionRunRequest { - mission: "program-hold".into(), - revision: None, - workspace: "/tmp".into(), - requester: None, - mode: None, - inputs: BTreeMap::new(), - idempotency_key: "program-hold-run".into(), - }) - .unwrap(); - let step = run.steps.iter().find(|step| step.step == "program").unwrap().subject.clone(); - let reconciler = Reconciler::new(store.clone(), Arc::new(FakeRuntime::default()), "node".into(), Arc::new(Notify::new())); - for _ in 0..4 { - reconciler.reconcile_once().unwrap(); - } - assert_eq!(store.step_run(&step).unwrap().unwrap().status, "working"); - assert_eq!(store.mission_run_status(&run.subject).unwrap().as_deref(), Some("running")); -} diff --git a/docs/st3/large-value-sites.json b/docs/st3/large-value-sites.json index 6adaa294d2e..a59eb5f0d75 100644 --- a/docs/st3/large-value-sites.json +++ b/docs/st3/large-value-sites.json @@ -1687,6 +1687,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/%'"}, diff --git a/docs/st3/program-engine.md b/docs/st3/program-engine.md index 6f012f80e31..34399b934e8 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 @@ -356,7 +365,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."} +}