git ents: feat: add exec-sprites and an effect dispatcher draining the postgres queue

Cargo.lock @@ -1130,6 +1130,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "effect-dispatcher" +version = "0.0.0" +dependencies = [ + "git-backend", + "gix-hash", + "refstore-postgres", + "uuid", +] + [[package]] name = "either" version = "1.16.0" @@ -1173,6 +1183,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25929004897f2bbab309121a60400d36992f6d911d09baa6c172f6cc55706601" +[[package]] +name = "exec-sprites" +version = "0.0.0" +dependencies = [ + "git-backend", + "gix-hash", +] + [[package]] name = "facet" version = "0.50.0-rc.5"

Cargo.toml @@ -2,6 +2,8 @@ resolver = "3" members = [ "crates/backend-conformance", + "crates/effect-dispatcher", + "crates/exec-sprites", "crates/git-anchor", "crates/git-backend", "crates/git-comment", @@ -65,6 +67,8 @@ facet-git-tree = { git = "https://github.com/git-ents/facet-git-tree" } form_urlencoded = "1" backend-conformance = { path = "crates/backend-conformance" } +effect-dispatcher = { path = "crates/effect-dispatcher" } +exec-sprites = { path = "crates/exec-sprites" } git-anchor = { path = "crates/git-anchor" } git-backend = { path = "crates/git-backend" } git-comment = { path = "crates/git-comment" }

crates/effect-dispatcher/Cargo.toml @@ -1,0 +1,17 @@ +[package] +name = "effect-dispatcher" +version = "0.0.0" +edition.workspace = true +publish.workspace = true +license.workspace = true + +[dependencies] +git-backend = { workspace = true } +gix-hash = { workspace = true } +refstore-postgres = { workspace = true } + +[dev-dependencies] +uuid = { workspace = true } + +[lints] +workspace = true

crates/effect-dispatcher/src/job.rs @@ -1,0 +1,173 @@ +//! The queued work order: one effect plus its materialized inputs, encoded +//! as the text `git_ents_effect_queue.payload` carries. A hand-rolled line +//! format (like `git-effect`'s job files) rather than a serialization +//! dependency, per the dependency policy. +//! +//! One `key value` pair per line; `command` — the only field that can +//! legitimately contain newlines — is escaped ([`escape`]/[`unescape`]). +//! A payload [`decode`] cannot read is a poison row: the dispatcher +//! completes it without running anything (mirroring how +//! `git_effect::engine` drops a malformed job file), rather than retrying +//! it forever. + +use std::collections::BTreeMap; + +use git_backend::{EffectDef, MaterializedInputs}; +use gix_hash::ObjectId; + +/// One queue row's decoded work order. +#[derive(Debug, Clone)] +pub struct Job { + /// What to run. + pub effect: EffectDef, + /// The materialized inputs to run it against. + pub inputs: MaterializedInputs, +} + +/// Encode `job` as the queue payload text [`decode`] reads. +#[must_use] +pub fn encode(job: &Job) -> String { + let mut out = String::new(); + out.push_str(&format!("name {}\n", job.effect.name)); + if let Some(command) = &job.effect.command { + out.push_str(&format!("command {}\n", escape(command))); + } + if let Some(image) = &job.effect.image { + out.push_str(&format!("image {image}\n")); + } + out.push_str(&format!("tree {}\n", job.inputs.tree)); + for (name, path) in &job.inputs.toolchain_paths { + out.push_str(&format!("toolchain {name} {path}\n")); + } + if let Some(cache) = &job.inputs.cache { + out.push_str(&format!("cache {cache}\n")); + } + out +} + +/// Decode a queue payload, or `None` when it is malformed (an unknown key, +/// a missing `name`/`tree`, an invalid tree oid). +#[must_use] +pub fn decode(payload: &str) -> Option<Job> { + let mut name = None; + let mut command = None; + let mut image = None; + let mut tree = None; + let mut toolchain_paths = BTreeMap::new(); + let mut cache = None; + for line in payload.lines() { + if line.is_empty() { + continue; + } + let (key, rest) = line.split_once(' ')?; + match key { + "name" => name = Some(rest.to_owned()), + "command" => command = Some(unescape(rest)), + "image" => image = Some(rest.to_owned()), + "tree" => tree = Some(ObjectId::from_hex(rest.as_bytes()).ok()?), + "toolchain" => { + let (toolchain, path) = rest.split_once(' ')?; + toolchain_paths.insert(toolchain.to_owned(), path.to_owned()); + } + "cache" => cache = Some(rest.to_owned()), + _ => return None, + } + } + Some(Job { + effect: EffectDef { + name: name?, + command, + image, + }, + inputs: MaterializedInputs { + tree: tree?, + toolchain_paths, + cache, + }, + }) +} + +/// Escape backslashes and newlines so a multi-line command survives the +/// one-pair-per-line format. +fn escape(value: &str) -> String { + value.replace('\\', "\\\\").replace('\n', "\\n") +} + +/// Invert [`escape`]. +fn unescape(value: &str) -> String { + let mut out = String::with_capacity(value.len()); + let mut chars = value.chars(); + while let Some(c) = chars.next() { + if c != '\\' { + out.push(c); + continue; + } + match chars.next() { + Some('n') => out.push('\n'), + Some(other) => out.push(other), + None => out.push('\\'), + } + } + out +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use super::*; + + fn tree() -> ObjectId { + ObjectId::from_hex(b"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb").unwrap() + } + + #[test] + fn a_job_round_trips_through_the_payload_format() { + let mut toolchain_paths = BTreeMap::new(); + toolchain_paths.insert("rust".to_owned(), "/toolchains/aaa/bin".to_owned()); + let job = Job { + effect: EffectDef { + name: "test".to_owned(), + command: Some("cargo fmt --check\ncargo test".to_owned()), + image: Some("debian:stable-slim".to_owned()), + }, + inputs: MaterializedInputs { + tree: tree(), + toolchain_paths, + cache: Some("sccache".to_owned()), + }, + }; + let decoded = decode(&encode(&job)).unwrap(); + assert_eq!(decoded.effect, job.effect); + assert_eq!(decoded.inputs.tree, job.inputs.tree); + assert_eq!(decoded.inputs.toolchain_paths, job.inputs.toolchain_paths); + assert_eq!(decoded.inputs.cache, job.inputs.cache); + } + + #[test] + fn a_minimal_job_round_trips() { + let job = Job { + effect: EffectDef { + name: "test".to_owned(), + command: None, + image: None, + }, + inputs: MaterializedInputs { + tree: tree(), + toolchain_paths: BTreeMap::new(), + cache: None, + }, + }; + let decoded = decode(&encode(&job)).unwrap(); + assert_eq!(decoded.effect, job.effect); + assert_eq!(decoded.inputs.cache, None); + } + + #[test] + fn malformed_payloads_decode_to_none() { + assert!(decode("").is_none()); + assert!(decode("name only-a-name\n").is_none()); // no tree + assert!(decode("tree bbbb\nname x\n").is_none()); // bad oid + assert!(decode("unknown key\n").is_none()); + } +}

crates/effect-dispatcher/src/lib.rs @@ -1,0 +1,654 @@ +//! The WS7 effect dispatcher (`docs/scale-out.adoc`, "WS7 — Effects and +//! Sprites"): one small machine that drains the Postgres effect queue and +//! spawns each claimed effect through an injected +//! [`git_backend::EffectExecutor`] — `exec-local` in a local deployment, +//! `exec-sprites` hosted; the loop cannot tell which, and must not be able +//! to (application code branches on trait capabilities, never on +//! deployment identity). +//! +//! # At-least-once, from the queue table +//! +//! [`git_backend::RefStore::watch`] is a wakeup *hint* only. Every wakeup — +//! a watch hint, a worker slot freeing up, or the periodic poll — runs a +//! full [`Dispatcher::tick`], which requeues stale claims and then claims +//! until the queue is empty or a cap is hit; the poll doubles as the +//! reconnect backstop the watch contract demands, so a dropped +//! LISTEN/NOTIFY notification delays a drain by at most one +//! [`DispatcherConfig::poll_interval`], never loses one. Claims carry a +//! claimant and timestamp; a claim older than +//! [`DispatcherConfig::claim_timeout`] is returned to the queue, so a +//! dispatcher that dies mid-effect redelivers rather than loses — possibly +//! running an effect twice, which is the at-least-once trade: effects are +//! recorded per commit, so a duplicate run re-records the same outcome. +//! +//! # Caps +//! +//! Two knobs bound concurrency: a global cap (cost — every running effect +//! is a machine or a container) and a per-repo cap (fairness — one +//! repository's backlog must not starve the rest). Both are enforced +//! exactly: the drain claims one row per query, recomputing the exclusion +//! set (repositories at their per-repo cap) between claims, so a burst +//! from one repository can never overshoot its cap inside a single batch. +//! One `UPDATE … SKIP LOCKED` round trip per claimed effect is cheap next +//! to what an effect costs to run. + +pub mod job; +mod queue; + +pub use queue::{EffectQueue, QueuedJob}; + +use std::collections::HashMap; +use std::sync::mpsc::{Receiver, Sender}; +use std::sync::{Arc, Mutex as StdMutex, PoisonError}; +use std::time::Duration; + +use git_backend::{EffectExecutor, RefEventStream}; + +/// The dispatcher's knobs. Both caps are enforced exactly (see the crate +/// docs on claiming one row at a time). +#[derive(Debug, Clone)] +pub struct DispatcherConfig { + /// The most effects running at once across every repository (cost). + pub global_cap: usize, + /// The most effects running at once for one repository (fairness). + pub per_repo_cap: usize, + /// How old a claim must be before [`Dispatcher::tick`] returns it to + /// the queue. Must comfortably exceed the longest legitimate effect + /// run (the executors' own timeout is 30 minutes), or a slow effect is + /// redelivered while still running. + pub claim_timeout: Duration, + /// The periodic-poll interval: the ceiling on how long a dropped watch + /// hint can delay a drain. + pub poll_interval: Duration, + /// Warm-pool size — always 0 today, and nothing implements a warm pool + /// beyond this knob. Q3 (`docs/scale-out.adoc`): revisit only if + /// measured Sprite cold start (image pull included) is *not* ≪ effect + /// duration; until that measurement exists, a warm pool is cost + /// without evidence. + pub warm_pool: usize, +} + +impl Default for DispatcherConfig { + fn default() -> Self { + Self { + global_cap: 8, + per_repo_cap: 2, + claim_timeout: Duration::from_secs(45 * 60), + poll_interval: Duration::from_secs(10), + warm_pool: 0, + } + } +} + +/// In-flight accounting: how many effects are running globally and per +/// repository. Updated when a worker starts and when it settles; the drain +/// derives its claim budget and exclusion set from it. +#[derive(Debug, Default)] +struct Running { + global: usize, + per_repo: HashMap<String, usize>, +} + +/// The dispatcher loop: [`Dispatcher::run`] forever in production, +/// [`Dispatcher::tick`] once per wakeup (and directly from tests). +pub struct Dispatcher { + queue: Arc<dyn EffectQueue>, + executor: Arc<dyn EffectExecutor>, + config: DispatcherConfig, + claimed_by: String, + running: Arc<StdMutex<Running>>, + wake_tx: Sender<()>, + wake_rx: StdMutex<Receiver<()>>, +} + +/// Lock `mutex`, recovering the guard from a poisoned lock rather than +/// panicking: losing one wakeup or one count to a poisoned lock is +/// recoverable (the periodic poll re-drains); tearing the dispatcher down +/// is not. +fn lock<T>(mutex: &StdMutex<T>) -> std::sync::MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +} + +impl Dispatcher { + /// A dispatcher draining `queue` into `executor` under `config`'s caps. + #[must_use] + pub fn new( + queue: Arc<dyn EffectQueue>, + executor: Arc<dyn EffectExecutor>, + config: DispatcherConfig, + ) -> Self { + let (wake_tx, wake_rx) = std::sync::mpsc::channel(); + Self { + queue, + executor, + config, + claimed_by: format!("dispatcher-{}", std::process::id()), + running: Arc::new(StdMutex::new(Running::default())), + wake_tx, + wake_rx: StdMutex::new(wake_rx), + } + } + + /// Run forever: drain now, then re-drain on every wakeup — a `hints` + /// event, a worker slot freeing up, or the periodic poll (the + /// reconnect backstop; see the crate docs). + pub fn run(&self, hints: RefEventStream) -> ! { + let forward = self.wake_tx.clone(); + std::thread::spawn(move || { + while hints.recv().is_some() { + if forward.send(()).is_err() { + break; + } + } + }); + let wake_rx = lock(&self.wake_rx); + loop { + self.tick(); + // A hint, a completion, or the poll timeout: which one woke us + // is deliberately not distinguished — every wakeup re-drains. + let _wakeup = wake_rx.recv_timeout(self.config.poll_interval); + } + } + + /// One full drain: requeue stale claims, then claim-and-start until + /// the queue is empty or a cap is hit. Idempotent and safe to call on + /// every wakeup; claims one row per query so both caps are exact (see + /// the crate docs). + pub fn tick(&self) { + if let Err(e) = self.queue.requeue_stale(self.config.claim_timeout) { + eprintln!("dispatcher: could not requeue stale claims: {e}"); + } + loop { + let exclude = { + let running = lock(&self.running); + if running.global >= self.config.global_cap { + return; + } + running + .per_repo + .iter() + .filter(|(_, count)| **count >= self.config.per_repo_cap) + .map(|(repo, _)| repo.clone()) + .collect::<Vec<_>>() + }; + let claimed = match self.queue.claim(&self.claimed_by, 1, &exclude) { + Ok(claimed) => claimed, + Err(e) => { + eprintln!("dispatcher: could not claim from the queue: {e}"); + return; + } + }; + let Some(claimed_job) = claimed.into_iter().next() else { + return; + }; + self.start(claimed_job); + } + } + + /// Decode and spawn one claimed row, handing its wait to a worker + /// thread that completes the row and frees the slot when the effect + /// settles. + /// + /// Failure semantics, per the at-least-once contract: + /// - an *undecodable* payload is poison: completed immediately, never + /// retried (mirroring how the engine drops a malformed job file); + /// - a payload that decodes but will not `spawn` (the sandbox is down, + /// the launcher errored) stays `claimed`, so the stale-claim timeout + /// redelivers it — the work never started, so redelivery is safe; + /// - a spawned effect is completed once `wait` settles, *whatever* it + /// settles to: an executor error after the spawn is a recorded + /// outcome, not grounds to run the effect again in-process. + fn start(&self, claimed_job: QueuedJob) { + let Some(work) = job::decode(&claimed_job.payload) else { + eprintln!( + "dispatcher: dropping malformed payload on queue row {} ({})", + claimed_job.id, claimed_job.repo + ); + if let Err(e) = self.queue.complete(claimed_job.id) { + eprintln!( + "dispatcher: could not complete poison row {}: {e}", + claimed_job.id + ); + } + return; + }; + let handle = match self.executor.spawn(&work.effect, work.inputs) { + Ok(handle) => handle, + Err(e) => { + eprintln!( + "dispatcher: could not spawn {} for {} (left claimed for redelivery): {e}", + work.effect.name, claimed_job.repo + ); + return; + } + }; + + { + let mut running = lock(&self.running); + running.global = running.global.saturating_add(1); + let count = running + .per_repo + .entry(claimed_job.repo.clone()) + .or_insert(0); + *count = count.saturating_add(1); + } + + let queue = Arc::clone(&self.queue); + let executor = Arc::clone(&self.executor); + let running = Arc::clone(&self.running); + let wake = self.wake_tx.clone(); + let effect_name = work.effect.name; + std::thread::spawn(move || { + match executor.wait(&handle) { + Ok(status) => eprintln!( + "dispatcher: {effect_name} settled for {}: {status:?}", + claimed_job.repo + ), + Err(e) => eprintln!( + "dispatcher: could not observe {effect_name} for {}: {e}", + claimed_job.repo + ), + } + if let Err(e) = queue.complete(claimed_job.id) { + eprintln!( + "dispatcher: could not complete queue row {}: {e}", + claimed_job.id + ); + } + { + let mut running = lock(&running); + running.global = running.global.saturating_sub(1); + if let Some(count) = running.per_repo.get_mut(&claimed_job.repo) { + *count = count.saturating_sub(1); + if *count == 0 { + running.per_repo.remove(&claimed_job.repo); + } + } + } + // A slot freed: wake the loop so remaining queue rows are + // claimed now, not on the next poll. + let _woken = wake.send(()); + }); + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use std::collections::{BTreeMap, HashSet}; + use std::sync::Condvar; + use std::time::Instant; + + use git_backend::{EffectDef, EffectHandle, EffectStatus, MaterializedInputs}; + + use super::*; + + /// Poll `condition` for up to five seconds — worker settlement runs on + /// its own threads, so assertions on it are eventual. + fn eventually(condition: impl Fn() -> bool) -> bool { + let deadline = Instant::now().checked_add(Duration::from_secs(5)).unwrap(); + while Instant::now() < deadline { + if condition() { + return true; + } + std::thread::sleep(Duration::from_millis(5)); + } + condition() + } + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum State { + Enqueued, + Claimed, + Done, + } + + #[derive(Debug, Clone)] + struct Row { + id: i64, + repo: String, + payload: String, + state: State, + claimed_at: Option<Instant>, + claimed_by: Option<String>, + } + + /// In-memory [`EffectQueue`] with the table's exact state machine. + struct FakeQueue { + rows: StdMutex<Vec<Row>>, + } + + impl FakeQueue { + fn new(rows: Vec<Row>) -> Self { + Self { + rows: StdMutex::new(rows), + } + } + + fn states(&self) -> Vec<State> { + lock(&self.rows).iter().map(|row| row.state).collect() + } + + fn all_done(&self) -> bool { + self.states().iter().all(|state| *state == State::Done) + } + } + + impl EffectQueue for FakeQueue { + fn claim( + &self, + claimed_by: &str, + limit: usize, + exclude_repos: &[String], + ) -> git_backend::Result<Vec<QueuedJob>> { + let mut rows = lock(&self.rows); + let mut out = Vec::new(); + for row in rows.iter_mut() { + if out.len() >= limit { + break; + } + if row.state == State::Enqueued && !exclude_repos.contains(&row.repo) { + row.state = State::Claimed; + row.claimed_at = Some(Instant::now()); + row.claimed_by = Some(claimed_by.to_owned()); + out.push(QueuedJob { + id: row.id, + repo: row.repo.clone(), + payload: row.payload.clone(), + }); + } + } + Ok(out) + } + + fn complete(&self, id: i64) -> git_backend::Result<()> { + for row in lock(&self.rows).iter_mut() { + if row.id == id { + row.state = State::Done; + } + } + Ok(()) + } + + fn requeue_stale(&self, older_than: Duration) -> git_backend::Result<u64> { + let mut requeued = 0u64; + for row in lock(&self.rows).iter_mut() { + let stale = row.state == State::Claimed + && row.claimed_at.is_none_or(|at| at.elapsed() > older_than); + if stale { + row.state = State::Enqueued; + row.claimed_at = None; + row.claimed_by = None; + requeued = requeued.saturating_add(1); + } + } + Ok(requeued) + } + } + + /// [`EffectExecutor`] whose spawns are recorded and whose completions + /// the test releases one by one. + struct FakeExecutor { + started: StdMutex<Vec<String>>, + released: StdMutex<HashSet<String>>, + settle: Condvar, + } + + impl FakeExecutor { + fn new() -> Self { + Self { + started: StdMutex::new(Vec::new()), + released: StdMutex::new(HashSet::new()), + settle: Condvar::new(), + } + } + + fn started(&self) -> Vec<String> { + lock(&self.started).clone() + } + + fn release(&self, name: &str) { + lock(&self.released).insert(name.to_owned()); + self.settle.notify_all(); + } + } + + impl EffectExecutor for FakeExecutor { + fn spawn( + &self, + effect: &EffectDef, + _inputs: MaterializedInputs, + ) -> git_backend::Result<EffectHandle> { + lock(&self.started).push(effect.name.clone()); + Ok(EffectHandle { + id: effect.name.clone(), + }) + } + + fn wait(&self, handle: &EffectHandle) -> git_backend::Result<EffectStatus> { + let deadline = Duration::from_secs(5); + let mut released = lock(&self.released); + while !released.contains(&handle.id) { + let (guard, timeout) = self + .settle + .wait_timeout(released, deadline) + .unwrap_or_else(PoisonError::into_inner); + released = guard; + if timeout.timed_out() { + return Err(git_backend::Error::Effect(format!( + "test effect {} was never released", + handle.id + ))); + } + } + Ok(EffectStatus::Pass) + } + } + + fn payload(name: &str) -> String { + job::encode(&job::Job { + effect: EffectDef { + name: name.to_owned(), + command: Some("true".to_owned()), + image: None, + }, + inputs: MaterializedInputs { + tree: gix_hash::ObjectId::from_hex(b"cccccccccccccccccccccccccccccccccccccccc") + .unwrap(), + toolchain_paths: BTreeMap::new(), + cache: None, + }, + }) + } + + fn row(id: i64, repo: &str, name: &str) -> Row { + Row { + id, + repo: repo.to_owned(), + payload: payload(name), + state: State::Enqueued, + claimed_at: None, + claimed_by: None, + } + } + + fn dispatcher( + rows: Vec<Row>, + config: DispatcherConfig, + ) -> (Dispatcher, Arc<FakeQueue>, Arc<FakeExecutor>) { + let queue = Arc::new(FakeQueue::new(rows)); + let executor = Arc::new(FakeExecutor::new()); + let dispatcher = Dispatcher::new( + Arc::clone(&queue) as Arc<dyn EffectQueue>, + Arc::clone(&executor) as Arc<dyn EffectExecutor>, + config, + ); + (dispatcher, queue, executor) + } + + #[test] + fn a_tick_claims_spawns_and_completes() { + let (dispatcher, queue, executor) = dispatcher( + vec![row(1, "repo-a", "fmt"), row(2, "repo-a", "test")], + DispatcherConfig::default(), + ); + dispatcher.tick(); + assert_eq!( + executor.started(), + vec!["fmt".to_owned(), "test".to_owned()] + ); + assert_eq!(queue.states(), vec![State::Claimed, State::Claimed]); + { + let rows = lock(&queue.rows); + assert!( + rows.iter() + .all(|row| row.claimed_by.as_deref() == Some(dispatcher.claimed_by.as_str())) + ); + } + + executor.release("fmt"); + executor.release("test"); + assert!(eventually(|| queue.all_done())); + } + + #[test] + fn a_stale_claim_is_requeued_and_redelivered() { + let mut stale = row(1, "repo-a", "fmt"); + stale.state = State::Claimed; + stale.claimed_at = Instant::now().checked_sub(Duration::from_secs(600)); + stale.claimed_by = Some("dispatcher-that-died".to_owned()); + let mut fresh = row(2, "repo-b", "test"); + fresh.state = State::Claimed; + fresh.claimed_at = Some(Instant::now()); + fresh.claimed_by = Some("dispatcher-still-alive".to_owned()); + + let config = DispatcherConfig { + claim_timeout: Duration::from_secs(60), + ..DispatcherConfig::default() + }; + let (dispatcher, queue, executor) = dispatcher(vec![stale, fresh], config); + dispatcher.tick(); + + // The stale claim came back and ran; the fresh claim was left with + // its (living) claimant, not double-delivered. + assert_eq!(executor.started(), vec!["fmt".to_owned()]); + executor.release("fmt"); + assert!(eventually(|| queue.states().first() == Some(&State::Done))); + assert_eq!(queue.states().get(1), Some(&State::Claimed)); + } + + #[test] + fn the_global_cap_bounds_concurrency() { + let rows = (1..=5) + .map(|n| row(n, "repo-a", &format!("effect-{n}"))) + .collect(); + let config = DispatcherConfig { + global_cap: 2, + per_repo_cap: 10, + ..DispatcherConfig::default() + }; + let (dispatcher, queue, executor) = dispatcher(rows, config); + + dispatcher.tick(); + assert_eq!(executor.started().len(), 2); + // Re-ticking while saturated claims nothing more. + dispatcher.tick(); + assert_eq!(executor.started().len(), 2); + + // A freed slot admits exactly one more on the next drain. + executor.release("effect-1"); + assert!(eventually(|| lock(&dispatcher.running).global == 1)); + dispatcher.tick(); + assert_eq!(executor.started().len(), 3); + + for n in 2..=5 { + executor.release(&format!("effect-{n}")); + assert!(eventually( + || lock(&dispatcher.running).global < dispatcher.config.global_cap + )); + dispatcher.tick(); + } + assert!(eventually(|| queue.all_done())); + assert_eq!(executor.started().len(), 5); + } + + #[test] + fn the_per_repo_cap_keeps_a_backlogged_repo_from_starving_others() { + // repo-a's three jobs are older (lower ids) than repo-b's one; with + // a per-repo cap of 1, repo-b must still run immediately. + let rows = vec![ + row(1, "repo-a", "a-1"), + row(2, "repo-a", "a-2"), + row(3, "repo-a", "a-3"), + row(4, "repo-b", "b-1"), + ]; + let config = DispatcherConfig { + global_cap: 8, + per_repo_cap: 1, + ..DispatcherConfig::default() + }; + let (dispatcher, queue, executor) = dispatcher(rows, config); + + dispatcher.tick(); + assert_eq!(executor.started(), vec!["a-1".to_owned(), "b-1".to_owned()]); + + // repo-a proceeds FIFO as its slot frees; repo-b's completion + // doesn't admit more repo-a work beyond its cap. + executor.release("a-1"); + assert!(eventually(|| { + lock(&dispatcher.running).per_repo.get("repo-a").copied() != Some(1) + })); + dispatcher.tick(); + assert_eq!( + executor.started(), + vec!["a-1".to_owned(), "b-1".to_owned(), "a-2".to_owned()] + ); + + executor.release("a-2"); + executor.release("b-1"); + assert!(eventually(|| lock(&dispatcher.running).per_repo.is_empty())); + dispatcher.tick(); + executor.release("a-3"); + assert!(eventually(|| queue.all_done())); + } + + #[test] + fn a_malformed_payload_is_completed_without_running() { + let mut poison = row(1, "repo-a", "unused"); + poison.payload = "not a payload".to_owned(); + let (dispatcher, queue, executor) = dispatcher(vec![poison], DispatcherConfig::default()); + dispatcher.tick(); + assert!(executor.started().is_empty()); + assert_eq!(queue.states(), vec![State::Done]); + } + + #[test] + fn a_failed_spawn_leaves_the_row_claimed_for_redelivery() { + /// An executor that refuses every spawn. + struct DownExecutor; + impl EffectExecutor for DownExecutor { + fn spawn( + &self, + _effect: &EffectDef, + _inputs: MaterializedInputs, + ) -> git_backend::Result<EffectHandle> { + Err(git_backend::Error::Effect("the sandbox is down".to_owned())) + } + fn wait(&self, _handle: &EffectHandle) -> git_backend::Result<EffectStatus> { + Err(git_backend::Error::Effect("nothing ever spawns".to_owned())) + } + } + + let queue = Arc::new(FakeQueue::new(vec![row(1, "repo-a", "fmt")])); + let dispatcher = Dispatcher::new( + Arc::clone(&queue) as Arc<dyn EffectQueue>, + Arc::new(DownExecutor), + DispatcherConfig::default(), + ); + dispatcher.tick(); + // Claimed, not done: the work never started, so the stale-claim + // timeout will redeliver it. + assert_eq!(queue.states(), vec![State::Claimed]); + } +}

crates/effect-dispatcher/src/queue.rs @@ -1,0 +1,77 @@ +//! The dispatcher's queue seam: [`EffectQueue`] abstracts +//! `git_ents_effect_queue`'s claim/complete/requeue triangle so the +//! dispatcher loop is tested against an in-memory fake, with +//! [`refstore_postgres::PostgresRefStore`]'s `dispatcher_*` surface as the +//! real implementation. + +use std::time::Duration; + +use git_backend::Result; + +/// One claimed queue row: the id [`EffectQueue::complete`] takes back, the +/// repository it belongs to (per-repo fairness accounting), and the +/// payload [`crate::job::decode`] reads. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct QueuedJob { + /// The row's id. + pub id: i64, + /// The repository the row was enqueued for. + pub repo: String, + /// The enqueued payload. + pub payload: String, +} + +/// The at-least-once effect queue (`docs/scale-out.adoc`, "RefStore": the +/// queue table, not the watch channel, carries the guarantee). `claim` +/// transitions rows `enqueued → claimed` with a claimant and timestamp; +/// `complete` transitions `claimed → done`; `requeue_stale` returns claims +/// older than a timeout to `enqueued` — redelivery for a dispatcher that +/// died with claims outstanding. At-least-once, not exactly-once: a +/// redelivered row can run its effect twice, and effects are recorded per +/// commit, so the duplicate re-records the same outcome. +pub trait EffectQueue: Send + Sync { + /// Atomically claim up to `limit` of the oldest `enqueued` rows for + /// `claimed_by`, skipping rows whose repository is in `exclude_repos` + /// (repositories at their fairness cap). + fn claim( + &self, + claimed_by: &str, + limit: usize, + exclude_repos: &[String], + ) -> Result<Vec<QueuedJob>>; + + /// Mark a claimed row done. + fn complete(&self, id: i64) -> Result<()>; + + /// Return every claim older than `older_than` to `enqueued`, returning + /// how many rows were requeued. + fn requeue_stale(&self, older_than: Duration) -> Result<u64>; +} + +impl EffectQueue for refstore_postgres::PostgresRefStore { + fn claim( + &self, + claimed_by: &str, + limit: usize, + exclude_repos: &[String], + ) -> Result<Vec<QueuedJob>> { + let limit = i64::try_from(limit).unwrap_or(i64::MAX); + Ok(self + .dispatcher_claim(claimed_by, limit, exclude_repos)? + .into_iter() + .map(|row| QueuedJob { + id: row.id.into(), + repo: row.repo_id, + payload: row.payload, + }) + .collect()) + } + + fn complete(&self, id: i64) -> Result<()> { + self.dispatcher_complete(id.into()) + } + + fn requeue_stale(&self, older_than: Duration) -> Result<u64> { + self.dispatcher_requeue_stale(older_than) + } +}

crates/effect-dispatcher/tests/postgres_queue.rs @@ -1,0 +1,217 @@ +//! The dispatcher's claim/requeue/complete SQL against a real Postgres, +//! exercised through the [`effect_dispatcher::EffectQueue`] impl for +//! [`refstore_postgres::PostgresRefStore`]. Gated on a reachable Postgres +//! exactly like `refstore-postgres`' own suites (whose harness this +//! duplicates, as `odb_ws5_conformance` already does): +//! +//! 1. `GIT_ENTS_TEST_POSTGRES_URL`, if set — an already-running Postgres. +//! 2. A throwaway `docker run` Postgres container, if docker is available. +//! 3. Otherwise: a visible skip. +//! +//! One caveat the SQL makes unavoidable: `dispatcher_*` queries span every +//! `repo_id` by design, so against a *shared* external database +//! (`GIT_ENTS_TEST_POSTGRES_URL`) this test can claim rows other suites +//! enqueued. The docker path — one container per test — is fully isolated; +//! assertions below filter to this test's own repo ids rather than assert +//! global counts, so an externally shared database perturbs nothing here. + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + reason = "test harness and assertions, not application code" +)] + +use std::process::{Command, Stdio}; +use std::time::Duration; + +use effect_dispatcher::EffectQueue as _; +use refstore_postgres::PostgresRefStore; + +/// A reachable test Postgres: either an externally supplied instance or a +/// throwaway docker container this harness starts and stops. +enum TestPostgres { + External(String), + Docker { container_id: String, url: String }, +} + +impl TestPostgres { + fn url(&self) -> &str { + match self { + Self::External(url) | Self::Docker { url, .. } => url, + } + } +} + +impl Drop for TestPostgres { + fn drop(&mut self) { + if let Self::Docker { container_id, .. } = self { + let _ignored = Command::new("docker") + .args(["rm", "-f", container_id]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } + } +} + +/// Obtain a test Postgres per the priority order in the module doc, or +/// `None` if neither an external URL nor docker is available. +fn test_postgres() -> Option<TestPostgres> { + if let Ok(url) = std::env::var("GIT_ENTS_TEST_POSTGRES_URL") { + return Some(TestPostgres::External(url)); + } + if !docker_available() { + return None; + } + start_docker_postgres() +} + +fn docker_available() -> bool { + Command::new("docker") + .arg("version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|status| status.success()) + .unwrap_or(false) +} + +fn start_docker_postgres() -> Option<TestPostgres> { + let output = Command::new("docker") + .args([ + "run", + "-d", + "--rm", + "-e", + "POSTGRES_PASSWORD=postgres", + "-p", + "127.0.0.1::5432", + "postgres:16-alpine", + ]) + .output() + .ok()?; + if !output.status.success() { + eprintln!( + "effect-dispatcher postgres_queue: docker run failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + return None; + } + let container_id = String::from_utf8_lossy(&output.stdout).trim().to_owned(); + + for _ in 0..120 { + let ready = Command::new("docker") + .args(["exec", &container_id, "pg_isready", "-U", "postgres"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|status| status.success()) + .unwrap_or(false); + if ready { + break; + } + std::thread::sleep(Duration::from_millis(250)); + } + + let port_output = Command::new("docker") + .args(["port", &container_id, "5432"]) + .output() + .ok()?; + let mapping = String::from_utf8_lossy(&port_output.stdout); + let port = mapping + .lines() + .next()? + .rsplit(':') + .next()? + .trim() + .to_owned(); + + Some(TestPostgres::Docker { + url: format!("host=127.0.0.1 port={port} user=postgres password=postgres dbname=postgres"), + container_id, + }) +} + +#[test] +fn dispatcher_sql_claims_across_repos_requeues_stale_and_completes() { + let Some(pg) = test_postgres() else { + eprintln!( + "skipping dispatcher_sql_claims_across_repos_requeues_stale_and_completes: \ + set GIT_ENTS_TEST_POSTGRES_URL, or make docker available" + ); + return; + }; + let repo_a = format!("dispatch-a-{}", uuid::Uuid::new_v4()); + let repo_b = format!("dispatch-b-{}", uuid::Uuid::new_v4()); + let store_a = PostgresRefStore::connect(pg.url(), repo_a.clone()).expect("connect a"); + let store_b = PostgresRefStore::connect(pg.url(), repo_b.clone()).expect("connect b"); + let a_id: i64 = store_a + .enqueue_effect("payload-a") + .expect("enqueue a") + .into(); + let b_id: i64 = store_b + .enqueue_effect("payload-b") + .expect("enqueue b") + .into(); + let mine = |id: i64, repo: &str| repo == repo_a && id == a_id || repo == repo_b && id == b_id; + + // The dispatcher claim spans repos: one query sees both stores' rows, + // each attributed to its repo. + let claimed = store_a.claim("dispatcher-1", 100, &[]).expect("claim"); + let claimed: Vec<_> = claimed + .into_iter() + .filter(|job| mine(job.id, &job.repo)) + .collect(); + assert_eq!(claimed.len(), 2); + assert!( + claimed + .iter() + .any(|job| job.id == a_id && job.repo == repo_a && job.payload == "payload-a") + ); + assert!( + claimed + .iter() + .any(|job| job.id == b_id && job.repo == repo_b && job.payload == "payload-b") + ); + + // A claimed row is not claimable again while its claim is fresh... + let reclaimed = store_a + .claim("dispatcher-2", 100, &[]) + .expect("claim while claimed"); + assert!(reclaimed.iter().all(|job| !mine(job.id, &job.repo))); + + // ...but a zero timeout makes every claim stale: both rows come back. + let requeued = store_a + .requeue_stale(Duration::ZERO) + .expect("requeue stale"); + assert!(requeued >= 2); + + // The per-repo exclusion (a repo at its fairness cap) skips that + // repo's rows and still claims the rest. + let claimed = store_a + .claim("dispatcher-3", 100, std::slice::from_ref(&repo_b)) + .expect("claim excluding b"); + let ids: Vec<i64> = claimed + .iter() + .filter(|job| mine(job.id, &job.repo)) + .map(|job| job.id) + .collect(); + assert_eq!(ids, vec![a_id]); + + // Done is terminal: a completed row never comes back, even through a + // zero-timeout requeue. + store_a.complete(a_id).expect("complete a"); + let _requeued = store_a + .requeue_stale(Duration::ZERO) + .expect("requeue stale again"); + let claimed = store_a + .claim("dispatcher-4", 100, &[]) + .expect("claim after complete"); + let ids: Vec<i64> = claimed + .iter() + .filter(|job| mine(job.id, &job.repo)) + .map(|job| job.id) + .collect(); + assert_eq!(ids, vec![b_id]); + store_b.complete(b_id).expect("complete b"); +}

crates/exec-sprites/Cargo.toml @@ -1,0 +1,15 @@ +[package] +name = "exec-sprites" +version = "0.0.0" +edition.workspace = true +publish.workspace = true +license.workspace = true + +[dependencies] +git-backend = { workspace = true } + +[dev-dependencies] +gix-hash = { workspace = true } + +[lints] +workspace = true

crates/exec-sprites/src/fly.rs @@ -1,0 +1,257 @@ +//! The real [`SpriteLauncher`]: shell out to the `fly` (flyctl) CLI's +//! `machine` commands — the same pattern as `git-effect`'s `docker` and +//! `sprite` backends, and deliberately not an HTTP client against the +//! Machines REST API (dependency policy: no new external dependencies). +//! +//! The `sprite` CLI the checks engine already drives was considered and +//! passed over here: it manages one persistent sandbox per repository and +//! has no image flag, while `exec-sprites`' whole point is one throwaway +//! machine per effect booted from a WS8-baked image. flyctl's `machine +//! run` expresses exactly that. +//! +//! Honesty about coverage: argv assembly ([`run_args`]) and output parsing +//! ([`parse_machine_id`], [`machine_settled`]) are pure and unit-tested; +//! *validating them against a live flyctl* is deploy-only work — flyctl's +//! human-oriented output is unversioned, and nothing in this repository +//! can pin it. Each parsing site carries the caveat. + +use std::process::Command; +use std::time::{Duration, Instant}; + +use git_backend::{Error, Result}; + +use crate::{MachineSpec, SpriteLauncher}; + +/// How long [`FlyLauncher::wait`] polls a machine before giving up — +/// matches the effect engine's own 30-minute per-effect timeout, so a +/// wedged machine is abandoned on the same clock as a wedged local run. +const WAIT_TIMEOUT: Duration = Duration::from_secs(30 * 60); + +/// How often [`FlyLauncher::wait`] polls `fly machine status`. +const POLL: Duration = Duration::from_secs(2); + +/// [`SpriteLauncher`] over the `fly` CLI: `fly machine run --rm --detach` +/// to create, `fly machine status` polling to wait. Authentication is +/// flyctl's own (`FLY_API_TOKEN`, or its config file) — this launcher +/// passes nothing secret on any command line. +pub struct FlyLauncher { + bin: String, + app: String, + poll: Duration, + wait_timeout: Duration, +} + +impl FlyLauncher { + /// A launcher creating machines in the Fly app `app` via the `fly` + /// binary on `PATH`. + #[must_use] + pub fn new(app: impl Into<String>) -> Self { + Self { + bin: "fly".to_owned(), + app: app.into(), + poll: POLL, + wait_timeout: WAIT_TIMEOUT, + } + } +} + +/// `fly machine run`'s argv for `spec` — pure, so the exact invocation is +/// unit-tested without flyctl (the same pattern as +/// `git_effect::docker::run_args`). Flags precede the positional image and +/// command so a command word can never be mistaken for a flag; `--rm` +/// reaps the machine on exit, `--detach` returns once it is created (the +/// executor's `spawn` must not block for completion). +#[must_use] +pub fn run_args(app: &str, spec: &MachineSpec) -> Vec<String> { + let mut args = vec![ + "machine".to_owned(), + "run".to_owned(), + "--app".to_owned(), + app.to_owned(), + "--name".to_owned(), + spec.name.clone(), + "--rm".to_owned(), + "--detach".to_owned(), + ]; + for (key, value) in &spec.env { + args.push("--env".to_owned()); + args.push(format!("{key}={value}")); + } + args.push(spec.image.clone()); + args.push("sh".to_owned()); + args.push("-c".to_owned()); + args.push(spec.command.clone()); + args +} + +/// The machine id out of `fly machine run --detach`'s output: the value of +/// its `Machine ID: <id>` line, or, failing that, the first token shaped +/// like a machine id (14 lowercase hex characters). Deploy-only caveat: +/// this matches the output shape current flyctl releases print; a live +/// `fly machine run` is the only authority on whether it still holds. +#[must_use] +pub fn parse_machine_id(output: &str) -> Option<String> { + for line in output.lines() { + if let Some(rest) = line.trim().strip_prefix("Machine ID:") { + let id = rest.trim(); + if !id.is_empty() { + return Some(id.to_owned()); + } + } + } + output + .split_whitespace() + .find(|token| { + token.len() == 14 + && token + .chars() + .all(|c| c.is_ascii_digit() || c.is_ascii_lowercase() && c.is_ascii_hexdigit()) + }) + .map(str::to_owned) +} + +/// Whether a `fly machine status` output describes a settled machine +/// (stopped or destroyed). Same deploy-only caveat as +/// [`parse_machine_id`]. +#[must_use] +pub fn machine_settled(status_output: &str) -> bool { + let lowered = status_output.to_lowercase(); + lowered.contains("stopped") || lowered.contains("destroyed") +} + +impl SpriteLauncher for FlyLauncher { + fn launch(&self, spec: &MachineSpec) -> Result<String> { + let output = Command::new(&self.bin) + .args(run_args(&self.app, spec)) + .output() + .map_err(|e| { + Error::Effect(format!( + "could not run the fly CLI (is flyctl installed?): {e}" + )) + })?; + if !output.status.success() { + return Err(Error::Effect(format!( + "fly machine run failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ))); + } + let stdout = String::from_utf8_lossy(&output.stdout); + parse_machine_id(&stdout).ok_or_else(|| { + Error::Effect( + "fly machine run succeeded but no machine id was found in its output".to_owned(), + ) + }) + } + + fn wait(&self, machine: &str) -> Result<()> { + let deadline = Instant::now() + .checked_add(self.wait_timeout) + .ok_or_else(|| Error::Effect("wait timeout overflowed the clock".to_owned()))?; + loop { + let output = Command::new(&self.bin) + .args(["machine", "status", machine, "--app", &self.app]) + .output() + .map_err(|e| { + Error::Effect(format!( + "could not run the fly CLI (is flyctl installed?): {e}" + )) + })?; + let text = format!( + "{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + if !output.status.success() { + // `--rm` reaps the machine on exit, so "not found" after a + // successful launch means it ran and was already destroyed: + // settled. (Deploy-only caveat as above.) + let lowered = text.to_lowercase(); + if lowered.contains("not found") || lowered.contains("could not find") { + return Ok(()); + } + return Err(Error::Effect(format!( + "fly machine status failed for {machine}: {}", + text.trim() + ))); + } + if machine_settled(&text) { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(Error::Effect(format!( + "machine {machine} did not settle within {:?}", + self.wait_timeout + ))); + } + std::thread::sleep(self.poll); + } + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::indexing_slicing, reason = "unit test")] + + use std::collections::BTreeMap; + + use super::*; + + fn spec() -> MachineSpec { + let mut env = BTreeMap::new(); + env.insert("GIT_ENTS_EFFECT".to_owned(), "test".to_owned()); + MachineSpec { + name: "effect-test-aaaaaaaaaaaa".to_owned(), + image: "registry.fly.io/git-ents-effects:baked".to_owned(), + env, + command: "cargo test".to_owned(), + } + } + + #[test] + fn run_args_put_flags_before_the_positional_image_and_command() { + let args = run_args("git-ents-effects", &spec()); + assert_eq!( + args, + vec![ + "machine", + "run", + "--app", + "git-ents-effects", + "--name", + "effect-test-aaaaaaaaaaaa", + "--rm", + "--detach", + "--env", + "GIT_ENTS_EFFECT=test", + "registry.fly.io/git-ents-effects:baked", + "sh", + "-c", + "cargo test", + ] + ); + } + + #[test] + fn parse_machine_id_prefers_the_labeled_line() { + let output = "Success! A Machine has been successfully launched\n\ + Machine ID: 148ed599c14189\n\ + Instance ID: 01HXYZ\n"; + assert_eq!(parse_machine_id(output), Some("148ed599c14189".to_owned())); + } + + #[test] + fn parse_machine_id_falls_back_to_an_id_shaped_token() { + assert_eq!( + parse_machine_id("launched 148ed599c14189 in yyz"), + Some("148ed599c14189".to_owned()) + ); + assert_eq!(parse_machine_id("no ids here"), None); + } + + #[test] + fn machine_settled_matches_stopped_and_destroyed() { + assert!(machine_settled("State: stopped")); + assert!(machine_settled("machine was destroyed")); + assert!(!machine_settled("State: started")); + } +}

crates/exec-sprites/src/lib.rs @@ -1,0 +1,383 @@ +//! `exec-sprites`: [`git_backend::EffectExecutor`] over Fly Machines +//! ("Sprites") — the hosted row of `docs/scale-out.adoc`'s "EffectExecutor" +//! table (WS7: the dispatcher drains the effect queue and creates Sprites +//! through this crate). +//! +//! One machine per effect, created through a [`SpriteLauncher`]. +//! [`FlyLauncher`] is the real one: it shells out to the `fly` (flyctl) +//! CLI, the same way the workspace's other sandbox backends shell out to +//! `docker` and `sprite` — no HTTP client dependency, per the dependency +//! policy. +//! +//! # What comes back, and how +//! +//! Nothing returns through the machine. Results and cache entries return +//! via *attested push* signed with the worker member key the machine is +//! provisioned with (`docs/scale-out.adoc`, WS7 and "Attested push": key +//! availability in Sprites is exactly the enrollment cost uniform-strong +//! attestation already pays). [`WorkerKey`] models that provisioning: the +//! machine spec carries only the *name* of the secret-provisioned +//! environment variable holding the key material — the material itself is +//! set out-of-band (`fly secrets set`, or machine secrets at deploy time) +//! and never travels through a machine-create argument, where +//! `fly machine status` would echo it. The launcher consequently observes +//! only machine lifecycle; [`git_backend::EffectExecutor::wait`] here +//! settles [`EffectStatus::SettledRemotely`], and the recorded run refs are +//! the outcome's source of truth. +//! +//! # The image (WS8) +//! +//! [`SpriteConfig::image`] (or an effect's own `image` override) is +//! expected to carry a baked toolchain object store (WS8, "Hydration and +//! toolchains"): materialization inside the machine stays the one code +//! path of correctness rule 6, the baked store merely being the tier that +//! answers `read` on a hit, with a miss falling through to fetch. Nothing +//! here bakes or verifies images; this crate only names what to boot. +//! +//! # What needs a real deployment +//! +//! Everything assembled here — machine specs, argv, env plumbing — is pure +//! and unit-tested against a fake launcher. What is *not* claimable +//! in-repo: flyctl's output shapes ([`fly::parse_machine_id`], the status +//! text [`FlyLauncher`] polls) and the end-to-end attested results push, +//! which need a deployed Fly app and a provisioned worker member key to +//! exercise. Those spots carry their own deploy-only notes. + +mod fly; + +pub use fly::FlyLauncher; + +use std::collections::BTreeMap; + +use git_backend::{ + EffectDef, EffectExecutor, EffectHandle, EffectStatus, Error, MaterializedInputs, Result, +}; + +/// Env var carrying the effect's name into the machine. +pub const EFFECT_ENV: &str = "GIT_ENTS_EFFECT"; + +/// Env var carrying the tree the effect runs against (full hex OID). +pub const TREE_ENV: &str = "GIT_ENTS_TREE"; + +/// Env var carrying the remote the in-machine runner pushes its results +/// and cache refs to (the attested push's destination). +pub const RESULTS_REMOTE_ENV: &str = "GIT_ENTS_RESULTS_REMOTE"; + +/// Env var carrying the worker member name whose key signs the results +/// push. +pub const WORKER_MEMBER_ENV: &str = "GIT_ENTS_WORKER_MEMBER"; + +/// Env var carrying the *name* of the secret-provisioned env var that +/// holds the worker member's private key material (see [`WorkerKey`]). +pub const WORKER_KEY_ENV: &str = "GIT_ENTS_WORKER_KEY_ENV"; + +/// Env var carrying the colon-joined `PATH` entries of the activated +/// toolchains, in toolchain-name order. +pub const TOOLCHAIN_PATH_ENV: &str = "GIT_ENTS_TOOLCHAIN_PATH"; + +/// Env var carrying the effect's cache name, when it declares one. +pub const CACHE_ENV: &str = "GIT_ENTS_CACHE"; + +/// Everything a launcher needs to create one machine: which image to boot, +/// what to run in it, and the environment the in-machine runner reads its +/// work order from. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MachineSpec { + /// The machine's name (derived from the effect and its tree). + pub name: String, + /// The image to boot — expected to carry the baked toolchain object + /// store (WS8). + pub image: String, + /// The environment the runner reads its work order from. Never carries + /// key material, only the name of the secret that does. + pub env: BTreeMap<String, String>, + /// The effect's shell command, run under `sh -c`. + pub command: String, +} + +/// How a Sprite actually gets created and reaped. [`FlyLauncher`] shells +/// out to flyctl; tests substitute a fake to assert the [`MachineSpec`] +/// without any Fly dependency. +pub trait SpriteLauncher: Send + Sync { + /// Create and start a machine per `spec`, returning its backend id. + /// Must not block for the effect's completion. + fn launch(&self, spec: &MachineSpec) -> Result<String>; + + /// Block until machine `machine` has settled (stopped, or already + /// reaped). + fn wait(&self, machine: &str) -> Result<()>; +} + +/// The worker member identity a machine pushes results back as: an +/// enrolled member (`refs/meta/members/*`) whose key material is +/// provisioned to the machine as a secret env var named +/// [`WorkerKey::key_env`]. Modeled as configuration because the material +/// itself must stay out of machine-create arguments; provisioning the +/// secret is a deploy step this crate cannot perform. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerKey { + /// The enrolled worker member's name. + pub member: String, + /// The name of the env var (a Fly secret) holding the member's private + /// key material inside the machine. + pub key_env: String, +} + +/// The executor's fixed configuration: the default image, where results +/// push back to, and the worker member identity that signs the push. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SpriteConfig { + /// The default image to boot when an effect names none — expected to + /// carry the baked toolchain object store (WS8). + pub image: String, + /// The remote the in-machine runner pushes results and cache refs to. + pub results_remote: String, + /// The worker member identity signing that push. + pub worker_key: WorkerKey, +} + +/// [`EffectExecutor`] creating one machine per spawned effect through a +/// [`SpriteLauncher`]. +pub struct SpriteExecutor<L> { + launcher: L, + config: SpriteConfig, +} + +impl<L: SpriteLauncher> SpriteExecutor<L> { + /// An executor creating machines through `launcher` per `config`. + #[must_use] + pub fn new(launcher: L, config: SpriteConfig) -> Self { + Self { launcher, config } + } +} + +/// Assemble the [`MachineSpec`] for one effect — pure, so exactly what a +/// machine is created with (image selection, env plumbing, key-material +/// indirection) is unit-tested without a launcher. +/// +/// # Errors +/// +/// Returns [`Error::Effect`] for a composite effect (no command): the +/// engine derives its outcome from its dependencies instead of spawning it. +pub fn machine_spec( + config: &SpriteConfig, + effect: &EffectDef, + inputs: &MaterializedInputs, +) -> Result<MachineSpec> { + let Some(command) = effect.command.clone() else { + return Err(Error::Effect(format!( + "effect {} is composite (no command); its outcome derives from its \ + dependencies instead of a spawn", + effect.name + ))); + }; + + let tree = inputs.tree.to_string(); + let mut env = BTreeMap::new(); + env.insert(EFFECT_ENV.to_owned(), effect.name.clone()); + env.insert(TREE_ENV.to_owned(), tree.clone()); + env.insert(RESULTS_REMOTE_ENV.to_owned(), config.results_remote.clone()); + env.insert( + WORKER_MEMBER_ENV.to_owned(), + config.worker_key.member.clone(), + ); + env.insert(WORKER_KEY_ENV.to_owned(), config.worker_key.key_env.clone()); + if !inputs.toolchain_paths.is_empty() { + let path = inputs + .toolchain_paths + .values() + .map(String::as_str) + .collect::<Vec<_>>() + .join(":"); + env.insert(TOOLCHAIN_PATH_ENV.to_owned(), path); + } + if let Some(cache) = &inputs.cache { + env.insert(CACHE_ENV.to_owned(), cache.clone()); + } + + Ok(MachineSpec { + name: machine_name(&effect.name, &tree), + image: effect.image.clone().unwrap_or_else(|| config.image.clone()), + env, + command, + }) +} + +/// A machine name for `effect` at `tree_hex`, kept to the `[a-z0-9-]` a +/// machine name allows (mirroring `git_effect::engine::sprite_name`'s +/// sanitization): `effect-<name>-<tree prefix>`. +fn machine_name(effect: &str, tree_hex: &str) -> String { + let sanitized: String = effect + .chars() + .map(|c| { + if c.is_ascii_alphanumeric() { + c.to_ascii_lowercase() + } else { + '-' + } + }) + .collect(); + let trimmed = sanitized.trim_matches('-'); + let name = if trimmed.is_empty() { + "effect" + } else { + trimmed + }; + let short = tree_hex.get(..12).unwrap_or(tree_hex); + format!("effect-{name}-{short}") +} + +impl<L: SpriteLauncher> EffectExecutor for SpriteExecutor<L> { + fn spawn(&self, effect: &EffectDef, inputs: MaterializedInputs) -> Result<EffectHandle> { + let spec = machine_spec(&self.config, effect, &inputs)?; + let id = self.launcher.launch(&spec)?; + Ok(EffectHandle { id }) + } + + fn wait(&self, handle: &EffectHandle) -> Result<EffectStatus> { + self.launcher.wait(&handle.id)?; + // The machine's termination is all this executor can observe; the + // outcome itself returns via the attested results push (crate + // docs). Exit-code sniffing through flyctl is deliberately not + // attempted — it would duplicate, and could contradict, the + // recorded run refs. + Ok(EffectStatus::SettledRemotely) + } +} + +#[cfg(test)] +mod tests { + #![allow( + clippy::unwrap_used, + clippy::unwrap_in_result, + clippy::indexing_slicing, + clippy::assertions_on_result_states, + reason = "unit test" + )] + + use std::sync::Mutex; + + use super::*; + + fn tree() -> gix_hash::ObjectId { + gix_hash::ObjectId::from_hex(b"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa").unwrap() + } + + fn config() -> SpriteConfig { + SpriteConfig { + image: "registry.fly.io/git-ents-effects:baked".to_owned(), + results_remote: "https://ents.example/repo.git".to_owned(), + worker_key: WorkerKey { + member: "worker-1".to_owned(), + key_env: "WORKER_SSH_KEY".to_owned(), + }, + } + } + + fn effect(command: Option<&str>, image: Option<&str>) -> EffectDef { + EffectDef { + name: "Build & Test".to_owned(), + command: command.map(str::to_owned), + image: image.map(str::to_owned), + } + } + + fn inputs() -> MaterializedInputs { + let mut toolchain_paths = BTreeMap::new(); + toolchain_paths.insert("rust".to_owned(), "/toolchains/aaa/bin".to_owned()); + toolchain_paths.insert("zig".to_owned(), "/toolchains/bbb/bin".to_owned()); + MaterializedInputs { + tree: tree(), + toolchain_paths, + cache: Some("sccache".to_owned()), + } + } + + /// Captures every launched spec; `wait` records the machine id it was + /// asked about. + struct FakeLauncher { + launched: Mutex<Vec<MachineSpec>>, + waited: Mutex<Vec<String>>, + } + + impl FakeLauncher { + fn new() -> Self { + Self { + launched: Mutex::new(Vec::new()), + waited: Mutex::new(Vec::new()), + } + } + } + + impl SpriteLauncher for FakeLauncher { + fn launch(&self, spec: &MachineSpec) -> Result<String> { + self.launched.lock().unwrap().push(spec.clone()); + Ok(format!("machine-{}", self.launched.lock().unwrap().len())) + } + + fn wait(&self, machine: &str) -> Result<()> { + self.waited.lock().unwrap().push(machine.to_owned()); + Ok(()) + } + } + + #[test] + fn machine_spec_plumbs_image_env_and_key_material_indirection() { + let spec = machine_spec(&config(), &effect(Some("cargo test"), None), &inputs()).unwrap(); + + assert_eq!(spec.image, "registry.fly.io/git-ents-effects:baked"); + assert_eq!(spec.command, "cargo test"); + assert_eq!(spec.name, "effect-build---test-aaaaaaaaaaaa"); + assert_eq!(spec.env[EFFECT_ENV], "Build & Test"); + assert_eq!(spec.env[TREE_ENV], tree().to_string()); + assert_eq!( + spec.env[RESULTS_REMOTE_ENV], + "https://ents.example/repo.git" + ); + assert_eq!(spec.env[WORKER_MEMBER_ENV], "worker-1"); + // Only the *name* of the secret-provisioned variable travels in + // the spec — never key bytes. + assert_eq!(spec.env[WORKER_KEY_ENV], "WORKER_SSH_KEY"); + assert_eq!( + spec.env[TOOLCHAIN_PATH_ENV], + "/toolchains/aaa/bin:/toolchains/bbb/bin" + ); + assert_eq!(spec.env[CACHE_ENV], "sccache"); + } + + #[test] + fn an_effects_own_image_overrides_the_default() { + let spec = machine_spec( + &config(), + &effect(Some("true"), Some("registry.fly.io/custom:1")), + &inputs(), + ) + .unwrap(); + assert_eq!(spec.image, "registry.fly.io/custom:1"); + } + + #[test] + fn a_composite_effect_is_never_launched() { + assert!(machine_spec(&config(), &effect(None, None), &inputs()).is_err()); + } + + #[test] + fn spawn_launches_and_wait_settles_remotely() { + let executor = SpriteExecutor::new(FakeLauncher::new(), config()); + let handle = executor + .spawn(&effect(Some("cargo test"), None), inputs()) + .unwrap(); + assert_eq!(handle.id, "machine-1"); + assert_eq!( + executor.wait(&handle).unwrap(), + EffectStatus::SettledRemotely + ); + assert_eq!( + *executor.launcher.waited.lock().unwrap(), + vec!["machine-1".to_owned()] + ); + let launched = executor.launcher.launched.lock().unwrap(); + assert_eq!(launched.len(), 1); + assert_eq!(launched.first().unwrap().command, "cargo test"); + } +}

feat: add exec-sprites and an effect dispatcher draining the postgres queue

Reviews

Start a review