unom/punktfunk
1
0
Fork 0
Code Issues Pull Requests 2 Actions 1 Packages Projects Releases 32 Wiki Activity

feat(store): plugin store host module — signed catalogs, tiered trust, install jobs

Browse Source 
The index is the verification gate: a catalog entry pins one exact version
plus that version's tarball integrity hash, so 'verified on every release'
is a property of the data rather than a promise about process. Nothing can
express 'track latest' for a catalogued plugin.

- store/index.rs   signed index parse + ed25519 verify (ring), validate-and-drop
                   per entry, semver minHost/advisory ranges
- store/sources.rs built-in unom source (compiled-in URL + two key slots for
                   rotation) + operator sources in plugin-sources.json
- store/catalog.rs https fetch with size/timeout/redirect caps, signature before
                   parse, last-good disk cache (stale-but-usable when offline)
- store/jobs.rs    single-flight install/uninstall: registry-integrity preflight
                   against the pin, spawn the runner CLI with live log capture,
                   post-install version check with rollback, provenance record,
                   runner restart (discovery is startup-only)
- store/manifest.rs install provenance; absence means CLI-installed
- mgmt/store.rs    12 routes under /api/v1/store, denied to the plugin token
                   (a plugin that can install plugins is an escalation primitive)

Also generalizes runner discovery and listInstalled from @punktfunk/plugin-*
to ANY scope's plugin-*: catalog entries must be scoped so the scope can map
to that entry's registry, so a third-party plugin necessarily arrives under
its own scope and would otherwise install but never run.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Enrico Bühler
2026-07-20 20:12:18 +02:00
parent 96e19986bc
commit 45c3b96907
19 changed files with 3578 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/punktfunk-host/src/store/jobs.rs
+687
View File
@@ -0,0 +1,687 @@
//! Install / uninstall **jobs** (design `plugin-store.md` §4.3).
//!
//! A package operation takes tens of seconds, so the API hands back a job id immediately (202) and
//! the console polls. One at a time: `bun add` and `bun remove` share a lockfile and a
//! `node_modules` tree, so a second concurrent op is a corruption bug waiting to happen — a
//! request that arrives while a job runs gets 409, not a queue.
//!
//! The pipeline, and why each step is there:
//!
//! ```text
//! resolve → what exactly are we installing (catalog entry ⇒ pkg@version+integrity, or a raw spec)
//! verify → the registry's advertised integrity for that version MUST equal the index pin
//! install → spawn the runner CLI (`bun add`), streaming its output into the job log
//! check → the version on disk is the version we pinned, else roll back
//! record → provenance manifest, so the tier stays visible forever
//! restart → the runner rediscovers units only at startup, so activation is a restart
//! ```
//!
//! **verify** is the step that makes "verified" mean something. A reviewed entry pins a tarball
//! hash; if the registry now advertises a different hash for that same version, someone republished
//! it after review and the install is refused before any code is fetched, let alone run. Tier-3
//! (raw spec) installs skip it — there is no pin to check against, and that asymmetry *is* the
//! tier.
use super::index::{scope_of, Entry};
use super::manifest::{self, Record, Tier};
use anyhow::{bail, Context, Result};
use std::collections::VecDeque;
use std::io::{BufRead, BufReader, Read};
use std::process::{Command, Stdio};
use std::sync::Mutex;
use std::time::{Duration, Instant};
/// How long a package op may run before we kill it. `bun add` over a cold cache on a slow link is
/// the worst case; five minutes is far past it.
const JOB_TIMEOUT: Duration = Duration::from_secs(300);
/// How many finished jobs we keep for the console to read back.
const JOB_HISTORY: usize = 20;
/// Lines of runner output retained per job (the console shows a tail).
const LOG_LINES: usize = 200;
// ---------------------------------------------------------------- job records
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, utoipa::ToSchema)]
#[serde(rename_all = "lowercase")]
pub(crate) enum State {
Running,
Done,
Failed,
}
/// A job as the console sees it. Field names are snake_case like the rest of the management API
/// (the *file* formats — index, sources, manifest — follow npm's camelCase instead).
#[derive(Debug, Clone, serde::Serialize, utoipa::ToSchema)]
pub(crate) struct Job {
pub id: String,
/// `install` or `uninstall`.
pub kind: String,
/// What the operator asked for — a package name, or the raw spec they typed.
pub target: String,
pub state: State,
/// Coarse step name, for a progress line the operator can read.
pub phase: String,
/// Tail of the runner's combined stdout/stderr.
pub log: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
pub started_at: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub finished_at: Option<u64>,
}
struct Jobs {
jobs: VecDeque<Job>,
counter: u64,
}
fn jobs() -> &'static Mutex<Jobs> {
static JOBS: std::sync::OnceLock<Mutex<Jobs>> = std::sync::OnceLock::new();
JOBS.get_or_init(|| {
Mutex::new(Jobs {
jobs: VecDeque::new(),
counter: 0,
})
})
}
fn lock() -> std::sync::MutexGuard<'static, Jobs> {
jobs().lock().unwrap_or_else(|e| e.into_inner())
}
/// Start tracking a job, refusing if one is already in flight (single-flight, §4.3).
fn begin(kind: &str, target: &str) -> Result<String> {
let mut g = lock();
if let Some(active) = g.jobs.iter().find(|j| j.state == State::Running) {
bail!(
"another plugin operation is already running ({} {})",
active.kind,
active.target
);
}
g.counter += 1;
let id = format!("job-{}-{}", super::catalog::unix_now(), g.counter);
let job = Job {
id: id.clone(),
kind: kind.to_string(),
target: target.to_string(),
state: State::Running,
phase: "queued".into(),
log: Vec::new(),
error: None,
started_at: super::catalog::unix_now(),
finished_at: None,
};
g.jobs.push_back(job);
while g.jobs.len() > JOB_HISTORY {
g.jobs.pop_front();
}
Ok(id)
}
fn update(id: &str, f: impl FnOnce(&mut Job)) {
let mut g = lock();
if let Some(job) = g.jobs.iter_mut().find(|j| j.id == id) {
f(job);
}
}
fn set_phase(id: &str, phase: &str) {
tracing::info!(job = %id, phase, "plugin store job");
update(id, |j| j.phase = phase.to_string());
}
fn log_line(id: &str, line: String) {
update(id, |j| {
if j.log.len() >= LOG_LINES {
j.log.remove(0);
}
j.log.push(line);
});
}
fn finish(id: &str, result: Result<()>) {
update(id, |j| {
j.finished_at = Some(super::catalog::unix_now());
match &result {
Ok(()) => {
j.state = State::Done;
j.phase = "done".into();
}
Err(e) => {
j.state = State::Failed;
j.error = Some(format!("{e:#}"));
}
}
});
match result {
Ok(()) => tracing::info!(job = %id, "plugin store job finished"),
Err(e) => tracing::warn!(job = %id, "plugin store job failed: {e:#}"),
}
}
/// One job by id.
pub(crate) fn get(id: &str) -> Option<Job> {
lock().jobs.iter().find(|j| j.id == id).cloned()
}
/// Recent jobs, newest last.
pub(crate) fn list() -> Vec<Job> {
lock().jobs.iter().cloned().collect()
}
/// Is a job in flight? (The console disables install buttons on this.)
pub(crate) fn busy() -> bool {
lock().jobs.iter().any(|j| j.state == State::Running)
}
// ---------------------------------------------------------------- install plans
/// A fully resolved install: everything the executor needs, with the trust decision already made.
pub(crate) struct Plan {
/// Package name without a version.
pub pkg: Option<String>,
/// The argument handed to `bun add` — `pkg@version` for a catalogued entry, the raw spec
/// otherwise.
pub spec: String,
pub version: Option<String>,
/// `(scope, registry_url)` to map in the plugins dir's `bunfig.toml`.
pub registry: Option<(String, String)>,
pub integrity: Option<String>,
pub tier: Tier,
pub source: Option<String>,
pub entry_id: Option<String>,
}
impl Plan {
/// From a catalog entry: exact version, pinned integrity, scope→registry mapping.
pub(crate) fn from_entry(entry: &Entry, source: &str, verified: bool) -> Result<Plan> {
let scope = scope_of(&entry.pkg).context("catalog entry package must be scoped")?;
Ok(Plan {
pkg: Some(entry.pkg.clone()),
spec: format!("{}@{}", entry.pkg, entry.version),
version: Some(entry.version.clone()),
registry: Some((scope, entry.registry.clone())),
integrity: Some(entry.integrity.clone()),
tier: if verified {
Tier::Verified
} else {
Tier::External
},
source: Some(source.to_string()),
entry_id: Some(entry.id.clone()),
})
}
/// From a raw package spec typed into the danger dialog. No pin, no review, no shelf.
pub(crate) fn from_spec(spec: &str) -> Result<Plan> {
let spec = validate_spec(spec)?;
Ok(Plan {
pkg: parse_spec_pkg(&spec),
spec,
version: None,
registry: None,
integrity: None,
tier: Tier::Unverified,
source: None,
entry_id: None,
})
}
}
/// Accept only specs we're willing to hand to `bun add`.
///
/// This is not shell quoting (we always exec with an argument vector, never a shell) — it is about
/// what the *package manager* will do with the string. Two real hazards: a leading `-` turns the
/// operand into a flag, and `file:` specs would let a console session install code from anywhere on
/// the host's filesystem, which is a different (and larger) capability than "install a package".
/// Local paths remain available through the CLI, which is where local development belongs.
fn validate_spec(spec: &str) -> Result<String> {
let s = spec.trim();
if s.is_empty() {
bail!("empty package spec");
}
if s.len() > 400 {
bail!("package spec is too long");
}
if s.starts_with('-') {
bail!("a package spec cannot start with '-'");
}
if s.chars().any(|c| c.is_whitespace() || c.is_control()) {
bail!("a package spec cannot contain whitespace or control characters");
}
let lower = s.to_ascii_lowercase();
for bad in ["file:", "link:", "portal:"] {
if lower.starts_with(bad) {
bail!("`{bad}` specs are not installable from the console — use the `punktfunk-host plugins add` CLI");
}
}
// URL-ish specs: only https and git+https. (http:// and git:// are unauthenticated transports.)
if lower.contains("://")
&& !(lower.starts_with("https://") || lower.starts_with("git+https://"))
{
bail!("only https:// and git+https:// URLs are accepted");
}
Ok(s.to_string())
}
/// Best-effort package name from an npm-style spec (`@scope/name`, `@scope/name@1.2.3`, `name@1`).
/// `None` for URL/git specs — those are identified after the fact by diffing `node_modules`.
fn parse_spec_pkg(spec: &str) -> Option<String> {
if spec.contains("://") {
return None;
}
if let Some(rest) = spec.strip_prefix('@') {
// Scoped: the version separator is the '@' AFTER the scope's slash.
let (scope_and_name, _) = match rest.split_once('/') {
Some((scope, tail)) => match tail.split_once('@') {
Some((name, ver)) => (format!("@{scope}/{name}"), Some(ver)),
None => (format!("@{scope}/{tail}"), None),
},
None => return None,
};
return Some(scope_and_name);
}
Some(
spec.split_once('@')
.map(|(n, _)| n)
.unwrap_or(spec)
.to_string(),
)
}
// ---------------------------------------------------------------- execution
/// Kick off an install. Returns the job id; the work runs on a detached thread.
pub(crate) fn spawn_install(plan: Plan) -> Result<String> {
let target = plan.pkg.clone().unwrap_or_else(|| plan.spec.clone());
let id = begin("install", &target)?;
let job_id = id.clone();
std::thread::Builder::new()
.name("pf-store-install".into())
.spawn(move || {
let result = run_install(&job_id, plan);
finish(&job_id, result);
crate::events::emit(crate::events::EventKind::StoreChanged);
})
.context("spawn the plugin-install worker")?;
Ok(id)
}
/// Kick off an uninstall.
pub(crate) fn spawn_uninstall(pkg: String) -> Result<String> {
if super::valid_installed_pkg(&pkg).is_err() {
bail!("not an installable plugin package name");
}
let id = begin("uninstall", &pkg)?;
let job_id = id.clone();
std::thread::Builder::new()
.name("pf-store-uninstall".into())
.spawn(move || {
let result = run_uninstall(&job_id, &pkg);
finish(&job_id, result);
crate::events::emit(crate::events::EventKind::StoreChanged);
})
.context("spawn the plugin-uninstall worker")?;
Ok(id)
}
fn run_install(id: &str, plan: Plan) -> Result<()> {
let dir = super::plugins_dir();
// ---- verify: the pin must still describe what the registry serves ------------------------
if let (Some(integrity), Some(pkg), Some(version), Some((_, registry))) = (
plan.integrity.as_deref(),
plan.pkg.as_deref(),
plan.version.as_deref(),
plan.registry.as_ref(),
) {
set_phase(id, "verifying");
let advertised = registry_integrity(registry, pkg, version)
.with_context(|| format!("check {pkg}@{version} against {registry}"))?;
if advertised != integrity {
bail!(
"integrity mismatch for {pkg}@{version}: the catalog pins {integrity} but the \
registry now serves {advertised}. This version was republished after it was \
reviewed — refusing to install."
);
}
log_line(id, format!("integrity ok: {pkg}@{version}"));
}
// ---- install ------------------------------------------------------------------------------
set_phase(id, "installing");
let before = super::installed_packages(&dir);
let mut args = vec!["add".to_string(), plan.spec.clone()];
if plan.version.is_some() {
// Pin the dependency range too, so a later `bun install` in this tree can't drift off the
// reviewed version.
args.push("--exact".into());
}
if let Some((scope, url)) = &plan.registry {
args.push("--registry".into());
args.push(format!("{scope}={url}"));
}
if !plan.spec.starts_with("@punktfunk/") {
// The runner CLI's supply-chain gate refuses anything resolving off Punktfunk's own
// registry unless told otherwise. Here the operator already made that decision — either by
// adding the source, or in the danger dialog — so pass the flag rather than making the gate
// unable to express "yes, deliberately".
args.push("--allow-public-registry".into());
}
args.push("--plugins".into());
args.push(dir.to_string_lossy().into_owned());
run_runner(id, &args)?;
// ---- check: is what landed what we asked for? ---------------------------------------------
set_phase(id, "checking");
let after = super::installed_packages(&dir);
let added: Vec<_> = after
.iter()
.filter(|p| !before.iter().any(|b| b.pkg == p.pkg))
.collect();
// For a catalogued install we know the package name; for a URL/git spec we learn it here.
let pkg = plan
.pkg
.clone()
.or_else(|| added.first().map(|p| p.pkg.clone()))
.context(
"the install finished but no new plugin package appeared — is this package a \
punktfunk plugin? (it must be named `@scope/plugin-*` or `punktfunk-plugin-*`)",
)?;
let installed = after
.iter()
.find(|p| p.pkg == pkg)
.with_context(|| format!("{pkg} is not present after install"))?;
if let (Some(want), Some(got)) = (plan.version.as_deref(), installed.version.as_deref()) {
if want != got {
// Roll back rather than leave an unreviewed version installed under a verified badge.
log_line(id, format!("rolling back: expected {want}, found {got}"));
set_phase(id, "rolling back");
let _ = run_runner(
id,
&[
"remove".to_string(),
pkg.clone(),
"--plugins".to_string(),
dir.to_string_lossy().into_owned(),
],
);
bail!("installed {pkg}@{got} but the catalog pinned {want} — rolled back");
}
}
// ---- record + activate ---------------------------------------------------------------------
set_phase(id, "recording");
manifest::record(
&dir,
&pkg,
Record {
tier: plan.tier,
source: plan.source.clone(),
entry_id: plan.entry_id.clone(),
version: installed.version.clone().or(plan.version.clone()),
spec: (plan.tier == Tier::Unverified).then(|| plan.spec.clone()),
installed_at: Some(manifest::now_stamp()),
},
)
.context("record install provenance")?;
restart_runner(id);
Ok(())
}
fn run_uninstall(id: &str, pkg: &str) -> Result<()> {
let dir = super::plugins_dir();
set_phase(id, "removing");
run_runner(
id,
&[
"remove".to_string(),
pkg.to_string(),
"--plugins".to_string(),
dir.to_string_lossy().into_owned(),
],
)?;
set_phase(id, "recording");
manifest::forget(&dir, pkg).context("update install provenance")?;
restart_runner(id);
Ok(())
}
/// Restart the runner so it rediscovers units. Best-effort and non-fatal: the package IS installed
/// at this point, so a restart failure must not present as an install failure — it presents as
/// "installed, runner needs a nudge", which the console says out loud.
fn restart_runner(id: &str) {
set_phase(id, "restarting runner");
match crate::plugins::restart_runtime() {
Ok(true) => log_line(id, "plugin runner restarted".into()),
Ok(false) => log_line(
id,
"the plugin runner is not enabled — enable it to start this plugin".into(),
),
Err(e) => log_line(id, format!("could not restart the plugin runner: {e:#}")),
}
}
/// Run the bun runner CLI with `args`, streaming both streams into the job log.
fn run_runner(id: &str, args: &[String]) -> Result<()> {
let (program, prefix) = crate::plugins::runner_command()?;
tracing::info!(job = %id, program = %program.display(), ?args, "spawning the plugin runner");
let mut child = Command::new(&program)
.args(&prefix)
.args(args)
// No stdin: `bun add` must never sit waiting on a prompt inside a service.
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.with_context(|| format!("run the plugin runner ({})", program.display()))?;
// Pump both streams concurrently: bun writes progress to one and errors to the other, and a
// full pipe on either would block the child.
let mut pumps = Vec::new();
if let Some(out) = child.stdout.take() {
let job = id.to_string();
pumps.push(std::thread::spawn(move || pump(&job, out)));
}
if let Some(err) = child.stderr.take() {
let job = id.to_string();
pumps.push(std::thread::spawn(move || pump(&job, err)));
}
let deadline = Instant::now() + JOB_TIMEOUT;
let status = loop {
match child.try_wait().context("wait for the plugin runner")? {
Some(status) => break status,
None if Instant::now() >= deadline => {
let _ = child.kill();
let _ = child.wait();
bail!(
"the plugin runner timed out after {}s",
JOB_TIMEOUT.as_secs()
);
}
None => std::thread::sleep(Duration::from_millis(150)),
}
};
for p in pumps {
let _ = p.join();
}
if !status.success() {
bail!(
"the plugin runner exited with status {} — see the job log",
status.code().unwrap_or(-1)
);
}
Ok(())
}
/// Copy one child stream into the job log, line by line.
fn pump(job: &str, stream: impl std::io::Read) {
for line in BufReader::new(stream).lines().map_while(Result::ok) {
let line = line.trim_end().to_string();
if !line.is_empty() {
log_line(job, line);
}
}
}
// ---------------------------------------------------------------- registry preflight
/// The integrity hash the registry itself advertises for `pkg@version`.
///
/// This is the check that gives the pin teeth. npm clients already refuse a tarball whose bytes
/// don't match the registry's advertised hash, so the remaining attack is a *republish* — same
/// version, new content, new hash. Comparing the registry's current hash against the reviewed one
/// catches exactly that, before anything is downloaded.
fn registry_integrity(registry: &str, pkg: &str, version: &str) -> Result<String> {
let base = registry.trim_end_matches('/');
// npm registry convention: the scope separator is percent-encoded in the packument path.
let url = format!("{base}/{}", pkg.replace('/', "%2f"));
let agent = ureq::AgentBuilder::new()
.timeout(Duration::from_secs(20))
.redirects(3)
.user_agent(&format!("punktfunk-host/{}", super::index::host_version()))
.build();
let resp = agent
.get(&url)
.set("Accept", "application/json")
.call()
.map_err(|e| match e {
ureq::Error::Status(404, _) => {
anyhow::anyhow!("the registry does not know this package")
}
other => anyhow::anyhow!("registry request failed: {other}"),
})?;
let mut body = Vec::new();
resp.into_reader()
.take(16 * 1024 * 1024)
.read_to_end(&mut body)
.context("read the registry response")?;
let doc: serde_json::Value =
serde_json::from_slice(&body).context("registry returned invalid JSON")?;
let dist = doc
.get("versions")
.and_then(|v| v.get(version))
.and_then(|v| v.get("dist"))
.with_context(|| format!("the registry has no version {version} of this package"))?;
dist.get("integrity")
.and_then(|i| i.as_str())
.map(str::to_string)
.context("the registry did not advertise an integrity hash for this version")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn spec_validation_refuses_the_dangerous_shapes() {
assert!(validate_spec("@scope/plugin-x").is_ok());
assert!(validate_spec("@scope/plugin-x@1.2.3").is_ok());
assert!(validate_spec("punktfunk-plugin-x").is_ok());
assert!(validate_spec("https://e.org/p.tgz").is_ok());
assert!(validate_spec("git+https://e.org/p.git").is_ok());
assert!(validate_spec("").is_err());
assert!(validate_spec(" ").is_err());
// a leading dash would be parsed by bun as a flag, not an operand
assert!(validate_spec("--production").is_err());
assert!(validate_spec("a b").is_err());
// local-path specs are CLI-only: installing from anywhere on the filesystem is a bigger
// capability than installing a package
assert!(validate_spec("file:/etc/passwd").is_err());
assert!(validate_spec("link:../evil").is_err());
// unauthenticated transports
assert!(validate_spec("http://e.org/p.tgz").is_err());
assert!(validate_spec("git://e.org/p.git").is_err());
}
#[test]
fn spec_package_name_parsing() {
assert_eq!(parse_spec_pkg("@a/b").as_deref(), Some("@a/b"));
assert_eq!(parse_spec_pkg("@a/b@1.2.3").as_deref(), Some("@a/b"));
assert_eq!(parse_spec_pkg("plain").as_deref(), Some("plain"));
assert_eq!(parse_spec_pkg("plain@2").as_deref(), Some("plain"));
// URL specs are identified by diffing node_modules after the install, not by parsing
assert_eq!(parse_spec_pkg("https://e.org/p.tgz"), None);
}
#[test]
fn plan_from_entry_pins_version_registry_and_integrity() {
let idx = super::super::index::Index::parse(
br#"{"schema":1,"plugins":[{"id":"rom-manager","pkg":"@punktfunk/plugin-rom-manager",
"registry":"https://git.unom.io/api/packages/unom/npm/","title":"ROM",
"version":"0.3.0","integrity":"sha512-AAAA"}]}"#,
)
.unwrap();
let plan = Plan::from_entry(&idx.plugins[0], "unom", true).unwrap();
assert_eq!(plan.spec, "@punktfunk/plugin-rom-manager@0.3.0");
assert_eq!(plan.version.as_deref(), Some("0.3.0"));
assert_eq!(plan.tier, Tier::Verified);
assert_eq!(
plan.registry.as_ref().unwrap().0,
"@punktfunk",
"the scope drives the bunfig registry mapping"
);
assert_eq!(plan.integrity.as_deref(), Some("sha512-AAAA"));
// The same entry from an operator-added source is External, never Verified (D6).
let ext = Plan::from_entry(&idx.plugins[0], "retro-hub", false).unwrap();
assert_eq!(ext.tier, Tier::External);
assert_eq!(ext.source.as_deref(), Some("retro-hub"));
}
#[test]
fn plan_from_spec_is_unverified_and_unpinned() {
let plan = Plan::from_spec(" @someone/punktfunk-plugin-x ").unwrap();
assert_eq!(plan.tier, Tier::Unverified);
assert_eq!(plan.spec, "@someone/punktfunk-plugin-x");
assert!(plan.version.is_none());
assert!(
plan.integrity.is_none(),
"nothing to check a raw spec against"
);
assert!(plan.registry.is_none());
}
/// `begin` is process-global and single-flight, so tests that create jobs must not overlap.
fn exclusive() -> std::sync::MutexGuard<'static, ()> {
static M: Mutex<()> = Mutex::new(());
M.lock().unwrap_or_else(|e| e.into_inner())
}
#[test]
fn single_flight_refuses_a_second_job() {
let _g = exclusive();
let first = begin("install", "@a/b").expect("first job starts");
assert!(begin("install", "@c/d").is_err(), "second must be refused");
assert!(busy());
finish(&first, Ok(()));
assert!(!busy());
let second = begin("install", "@c/d").expect("a finished job frees the slot");
finish(&second, Ok(()));
}
#[test]
fn job_log_is_bounded_and_tails() {
let _g = exclusive();
let id = begin("install", "@log/test").unwrap();
for i in 0..(LOG_LINES + 50) {
log_line(&id, format!("line {i}"));
}
let job = get(&id).unwrap();
assert_eq!(job.log.len(), LOG_LINES);
assert_eq!(job.log.last().unwrap(), &format!("line {}", LOG_LINES + 49));
finish(&id, Err(anyhow::anyhow!("boom")));
let job = get(&id).unwrap();
assert_eq!(job.state, State::Failed);
assert!(job.error.unwrap().contains("boom"));
assert!(job.finished_at.is_some());
}
}