crates/hub/src/flow.rs at 8a3334b7a364 · andymccall/oxygenhub

# 8a3334b7a364 oxygenhub / crates / hub / src / flow.rs
43.5 KiB · Rust · last changed Scheduled runs: on { schedule } jobs start on the default bookmark when due 1 month ago source blame history raw
//! OxygenFlow on the hub (hub spec §13.2): runs started by pushes, the
//! runner API, job tokens and the checks that mirror jobs. Pipelines are
//! parsed by `flow-core`, the same code `flow check` runs.

use crate::api::ApiError;
use crate::db::{self, flow as fdb};
use crate::repos::RepoName;
use crate::Shared;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::Json;
use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, MAX_RECORDS, MAX_RECORD_TEXT};
use flow_core::{Pipeline, PIPELINE_PATH};
use oxy_core::hash::ChangeHash;

/// Prefix of job tokens, so the wire protocol can tell them from personal tokens.
pub const JOB_TOKEN_PREFIX: &str = "flowjob_";
/// A job token is refused this long after its job started, whatever its state.
pub const JOB_TOKEN_MAX_AGE_MS: i64 = 2 * 3600 * 1000;

/// Whether pushes to this repository start runs: its own setting when a
/// site administrator made one, else hub.toml's `[flow] repos`.
pub fn enabled(app: &Shared, row: &db::repos::RepoRow) -> bool {
    match row.flow.as_str() {
        "on" => true,
        "off" => false,
        _ => app.config.flow.allows(&row.owner, &row.name),
    }
}

/// A bookmark moved by a push: start a run when the repository is allowed
/// and the new state has a pipeline.
pub async fn on_push(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, state: &str, heads: &[ChangeHash], actor: &str) {
    if heads.is_empty() || !enabled(app, row) {
        return;
    }
    // A pushed tag runs the jobs whose `on` names it (oxygenflow#24).
    let started = match bookmark.strip_prefix("tags/") {
        Some(tag) => start_run_with(app, row, bookmark, state, heads, actor, &format!("tag:{tag}"), None).await,
        None => {
            let touched = touched_since_last_run(app, row, bookmark, heads).await;
            start_run_scoped(app, row, bookmark, state, heads, actor, "push", touched.as_deref()).await
        }
    };
    if let Err(e) = started {
        tracing::warn!("flow: starting a run of {}/{}: {e}", row.owner, row.name);
    }
}

/// "Run pipeline" (hub spec §13.2): run a bookmark's current state without
/// moving it. Returns the new run's number.
pub async fn run_bookmark(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, actor: &str) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let b = bookmark.to_string();
    let found = app
        .repos
        .read(&repo, move |r, _| {
            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
            Ok(Some((st, r.store().get_state(&st)?.heads)))
        })
        .await
        .map_err(|e| e.to_string())?;
    let Some((state, heads)) = found else { return Err(format!("no bookmark {bookmark}")) };
    if heads.is_empty() {
        return Err(format!("{bookmark} has no changes yet"));
    }
    start_run(app, row, bookmark, &state.to_hex(), &heads, actor, "manual")
        .await?
        .ok_or_else(|| format!("{bookmark} has no {PIPELINE_PATH}"))
}

/// "Run pipeline" on one committed change: the state whose only head is
/// `change`, whatever bookmarks hold it. Returns the new run's number.
pub async fn run_change(app: &Shared, row: &db::repos::RepoRow, change: &str, actor: &str) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    let h = ChangeHash::from_hex(change.trim().trim_start_matches('#')).ok_or("give the full hash of a change")?;
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let known = app.repos.read(&repo, move |r, _| Ok(r.store().get_change_meta(&h).is_ok())).await.map_err(|e| e.to_string())?;
    if !known {
        return Err(format!("no change {} in this repository", &change.trim()[..12.min(change.trim().len())]));
    }
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(vec![h]).encode());
    start_run(app, row, "", &state.to_hex(), &[h], actor, "manual")
        .await?
        .ok_or_else(|| format!("that change's state has no {PIPELINE_PATH}"))
}

/// Re-run a finished run: the same state, bookmark and pipeline, as a new
/// run whose checks replace the old ones. Returns the new run's number.
pub async fn rerun(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, actor: &str) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    if run.live() {
        return Err(format!("run #{} is still {}", run.number, run.status));
    }
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    if heads.is_empty() {
        return Err("the run names no changes".into());
    }
    // A rerun repeats its run's event: the jobs it had are the jobs it gets.
    let trigger = format!("rerun:{}:{}", run.number, original_event(run));
    start_run(app, row, &run.bookmark, &run.state, &heads, actor, &trigger)
        .await?
        .ok_or_else(|| format!("that state has no {PIPELINE_PATH}"))
}

/// The event a run's trigger stands for: `push`, `tag:NAME`, `manual`,
/// `schedule`, and `rerun:N:EVENT` repeating EVENT.
fn event_of<'a>(trigger: &'a str, bookmark: &'a str) -> flow_core::pipeline::Event<'a> {
    use flow_core::pipeline::Event;
    let t = match trigger.strip_prefix("rerun:") {
        Some(rest) => rest.split_once(':').map(|(_, e)| e).unwrap_or("manual"),
        None => trigger,
    };
    if let Some(tag) = t.strip_prefix("tag:") {
        Event::Tag(tag)
    } else if t == "push" {
        Event::Push(bookmark)
    } else if t == "schedule" {
        Event::Schedule
    } else {
        Event::Manual
    }
}

/// The event to repeat when rerunning `run`.
fn original_event(run: &fdb::Run) -> String {
    let t = run.trigger.as_str();
    match t.strip_prefix("rerun:") {
        Some(rest) => rest.split_once(':').map(|(_, e)| e.to_string()).unwrap_or_else(|| "manual".into()),
        None => t.to_string(),
    }
}

/// The paths the changes a push brought touch: the changes in `heads`'
/// state that the bookmark's previous run did not have, and the files their
/// edits and tree operations name, as the new state places them. `None`
/// (run every job) for the bookmark's first run, or when it cannot be told.
async fn touched_since_last_run(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Option<Vec<String>> {
    let last = fdb::latest_on(&app.db, row.id, bookmark).await.ok()??;
    let old: Vec<ChangeHash> = last.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    if old.is_empty() {
        return None;
    }
    let repo = RepoName::new(&row.owner, &row.name)?;
    let new = heads.to_vec();
    app.repos
        .read(&repo, move |r, cache| {
            let fresh = r.negotiate(&old, &new)?;
            let loaded = cache.load(r, &new)?;
            let by_node: std::collections::HashMap<oxy_core::ids::NodeId, &String> =
                loaded.snapshot.files.iter().map(|(p, f)| (f.node, p)).collect();
            let mut paths = std::collections::BTreeSet::new();
            for h in fresh {
                let (c, _) = r.store().get_change_lazy(&h)?;
                let nodes = c.edits.iter().map(|e| e.node()).chain(c.tree_ops.iter().map(|o| o.node()));
                for n in nodes {
                    if let Some(p) = by_node.get(&n) {
                        paths.insert((*p).clone());
                    }
                }
            }
            Ok(paths.into_iter().collect::<Vec<String>>())
        })
        .await
        .ok()
}

/// Whether a job runs for a push that touched `touched`: always without a
/// scope or without knowing what was touched, else when a path matches.
fn in_scope(scope: &[String], touched: Option<&[String]>) -> bool {
    let Some(touched) = touched else { return true };
    if scope.is_empty() {
        return true;
    }
    let mut b = globset::GlobSetBuilder::new();
    for g in scope {
        match globset::Glob::new(g) {
            Ok(g) => {
                b.add(g);
            }
            Err(_) => return true,
        }
    }
    let Ok(set) = b.build() else { return true };
    touched.iter().any(|p| set.is_match(p))
}

/// [`start_run`] for a push: jobs whose `scope` the push did not touch are
/// left out of the run, unless a job that runs depends on them.
#[allow(clippy::too_many_arguments)]
pub async fn start_run_scoped(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
    touched: Option<&[String]>,
) -> Result<Option<i64>, String> {
    start_run_with(app, row, bookmark, state, heads, actor, trigger, touched).await
}

/// Start a run of the pipeline in the state `heads` make (whose state
/// hash is `state`), recorded on `bookmark`: for a push, "Run pipeline"
/// (`trigger` = `manual`) and re-runs (`rerun:N`). `actor` is the account
/// whose access the run's subrepo reads are checked against. Returns the
/// run's number, or `None` when the state has no pipeline. The caller
/// checks that Flow is enabled.
pub async fn start_run(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
) -> Result<Option<i64>, String> {
    start_run_with(app, row, bookmark, state, heads, actor, trigger, None).await
}

#[allow(clippy::too_many_arguments)]
async fn start_run_with(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
    touched: Option<&[String]>,
) -> Result<Option<i64>, String> {
    start_run_at(app, row, bookmark, state, heads, actor, trigger, touched, None).await
}

/// [`start_run_with`], for a scheduled run due at `at` on the hub's clock.
#[allow(clippy::too_many_arguments)]
async fn start_run_at(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
    touched: Option<&[String]>,
    at: Option<flow_core::pipeline::When>,
) -> Result<Option<i64>, String> {
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let h = heads.to_vec();
    let text = app
        .repos
        .read(&repo, move |r, cache| {
            let loaded = cache.load(r, &h)?;
            Ok(match loaded.snapshot.files.get(PIPELINE_PATH).map(|f| &f.content) {
                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),
                _ => None,
            })
        })
        .await
        .map_err(|e| format!("reading the pipeline: {e}"))?;
    let Some(text) = text else { return Ok(None) };
    let mut hex: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();
    hex.sort();
    let heads_s = hex.join("+");
    let base = db::flow::NewRun { repo_id: row.id, bookmark, state, heads: &heads_s, pipeline_hash: "", actor, trigger, error: None };
    let id = match Pipeline::parse(&text) {
        Err(problems) => {
            let msg = problems.to_string();
            db::flow::create_run(&app.db, db::flow::NewRun { error: Some(&msg), ..base }, &[]).await.map_err(|e| e.to_string())?
        }
        Ok(p) => {
            let depths = p.depths();
            // Jobs out of the push's scope stay out, unless a kept job
            // needs them.
            // The jobs this run's event starts (`on`), out of those its
            // push touched (`scope`); then what they depend on.
            let event = match at {
                Some(w) => flow_core::pipeline::Event::ScheduleAt(w),
                None => event_of(trigger, bookmark),
            };
            let mut keep: std::collections::HashSet<String> = p
                .order()
                .into_iter()
                .filter(|j| p.on_of(j).matches(event) && in_scope(&j.scope, touched))
                .map(|j| j.name.clone())
                .collect();
            loop {
                let needed: Vec<String> = p
                    .order()
                    .into_iter()
                    .filter(|j| keep.contains(&j.name))
                    .flat_map(|j| j.depends_on.iter().cloned())
                    .filter(|d| !keep.contains(d))
                    .collect();
                if needed.is_empty() {
                    break;
                }
                keep.extend(needed);
            }
            let jobs: Vec<fdb::NewJob> = p
                .order()
                .into_iter()
                .filter(|j| keep.contains(&j.name))
                .map(|j| {
                    let sb = &p.sandboxes[&j.sandbox];
                    fdb::NewJob {
                        name: j.name.clone(),
                        depth: depths[j.name.as_str()] as i64,
                        depends: j.depends_on.clone(),
                        spec: serde_json::to_string(j).unwrap_or_default(),
                        sandbox: serde_json::to_string(sb).unwrap_or_default(),
                        runtime: sb.runtime.as_str().into(),
                    }
                })
                .collect();
            if jobs.is_empty() {
                // Nothing this event starts (or this push touched).
                return Ok(None);
            }
            let hash = hex::encode(p.hash());
            let id =
                db::flow::create_run(&app.db, db::flow::NewRun { pipeline_hash: &hash, ..base }, &jobs).await.map_err(|e| e.to_string())?;
            if let Ok(Some(run)) = fdb::run_by_id(&app.db, id).await {
                for j in &jobs {
                    report_check(app, row, state, run.number, &j.name, "queued", "waiting for a runner").await;
                }
            }
            id
        }
    };
    let run = fdb::run_by_id(&app.db, id).await.map_err(|e| e.to_string())?.ok_or("the run vanished")?;
    Ok(Some(run.number))
}

/// Record a job's check; it links to the run's page by the run's number.
async fn report_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_number: i64, job: &str, status: &str, summary: &str) {
    let url = format!("{}/{}/{}/flow/{run_number}", app.config.base_url(), row.owner, row.name);
    let name = format!("flow/{job}");
    let c = db::hooks::upsert_check(
        &app.db,
        row.id,
        db::hooks::CheckReport { state, name: &name, status, summary, url: &url, reporter: None, seal: "" },
    )
    .await;
    if let Ok(c) = c {
        let api = crate::api::check_of(&c);
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
    }
}

/// The check status a job's status shows as.
fn check_status(job_status: &str) -> &'static str {
    match job_status {
        "queued" => "queued",
        "running" => "running",
        "passed" => "passed",
        "skipped" => "skipped",
        _ => "failed",
    }
}

/// The runner a bearer token belongs to: a registered runner, or one
/// configured with a hub.toml token (tracked from its first request). A
/// disabled runner is refused.
async fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<db::runners::Runner, ApiError> {
    let Some(t) = oxy_server::bearer_token(headers) else {
        return Err(ApiError(StatusCode::UNAUTHORIZED, "a runner token is required".into()));
    };
    let hash = blake3::hash(t.as_bytes()).to_hex().to_string();
    let runner = match db::runners::by_token(&app.db, &hash).await? {
        Some(r) => r,
        None if app.config.flow.runner_tokens.iter().any(|x| crate::auth::mac_equal(x, t)) => {
            db::runners::configured(&app.db, &hash, "runner").await?
        }
        None => return Err(ApiError(StatusCode::UNAUTHORIZED, "not a runner token".into())),
    };
    if runner.state == "disabled" {
        return Err(ApiError(StatusCode::FORBIDDEN, format!("runner {} is disabled on this hub", runner.name)));
    }
    Ok(runner)
}

/// `POST /api/v1/flow/runners/register`: a one-time registration token
/// from Admin → Runners buys a runner credential, binding the runner's
/// signing identity (hub spec §13.2).
pub async fn register(
    State(app): State<Shared>,
    Json(req): Json<flow_core::api::Register>,
) -> Result<Json<flow_core::api::Registered>, ApiError> {
    let bytes = hex::decode(req.identity.trim()).map_err(|_| ApiError(StatusCode::BAD_REQUEST, "identity: not hex".into()))?;
    let identity = oxy_core::identity::Identity::decode(&bytes).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("identity: {e}")))?;
    let name: String = req.name.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).take(64).collect();
    if name.is_empty() {
        return Err(ApiError(StatusCode::BAD_REQUEST, "give the runner a name of letters, digits, - _ or .".into()));
    }
    let credential = format!("flowrunner_{}", crate::auth::random_hex(32));
    let hash = blake3::hash(credential.as_bytes()).to_hex().to_string();
    let reg = blake3::hash(req.token.trim().as_bytes()).to_hex().to_string();
    let runner =
        db::runners::register(&app.db, &reg, &name, &hash, &identity.hash().to_hex(), &identity.name, "").await?.ok_or_else(|| {
            ApiError(StatusCode::FORBIDDEN, "that registration token is unknown, used or expired; create one on Admin → Runners".into())
        })?;
    db::runners::seen(&app.db, runner.id, None, Some(&req.info)).await?;
    tracing::info!("flow: runner {} registered (key {})", runner.name, &runner.key[..12]);
    Ok(Json(flow_core::api::Registered { id: runner.id, name: runner.name, credential }))
}

/// `POST /api/v1/flow/claim`.
pub async fn claim(State(app): State<Shared>, headers: HeaderMap, Json(req): Json<ClaimRequest>) -> Result<Response, ApiError> {
    let me = runner_auth(&app, &headers).await?;
    let claimed: String = req.runner.chars().filter(|c| !c.is_control()).take(64).collect();
    db::runners::seen(&app.db, me.id, Some(&claimed), req.info.as_ref()).await?;
    if let Some(holding) = &req.holding {
        end_jobs_not_held(&app, &me, holding).await?;
    }
    // A draining runner finishes what it has and takes nothing new.
    if me.state == "draining" {
        return Ok(StatusCode::NO_CONTENT.into_response());
    }
    let runner = if me.source == "registered" { me.name.clone() } else { claimed };
    let secret = format!("{JOB_TOKEN_PREFIX}{}", crate::auth::random_hex(32));
    let room = fdb::Room {
        memory_mib: req.free_memory_mib,
        cpus: req.free_cpus,
        images: req.images.as_ref().map(|i| i.iter().map(|h| h.to_ascii_lowercase()).collect()),
    };
    // The labels it reports now, or those it registered with.
    let labels: Vec<String> = match &req.info {
        Some(i) => i.labels.clone(),
        None => me.label_list().into_iter().map(String::from).collect(),
    };
    let Some(job) =
        fdb::claim_within(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope(), &labels).await?
    else {
        return Ok(StatusCode::NO_CONTENT.into_response());
    };
    db::runners::set_job_runner(&app.db, job.id, me.id).await?;
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
    let row =
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
    report_check(&app, &row, &run.state, run.number, &job.name, "running", &format!("running on {runner}")).await;
    let bad = |e: serde_json::Error| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}"));
    // The job's limit: its own timeout, else its sandbox's, else the site
    // default, never past the site maximum; recorded for the reaper.
    let limits = limits(&app).await;
    let spec: flow_core::pipeline::Job = serde_json::from_str(&job.spec).map_err(bad)?;
    let sandbox: flow_core::pipeline::Sandbox = serde_json::from_str(&job.sandbox).map_err(bad)?;
    let timeout_s = spec.timeout_s.or(sandbox.timeout_s).unwrap_or(limits.default_s).min(limits.max_s).max(60);
    fdb::set_job_timeout(&app.db, job.id, timeout_s).await?;
    let claim = Claim {
        job_id: job.id,
        run_id: run.id,
        repo: format!("{}/{}", row.owner, row.name),
        clone_url: format!("{}/{}/{}", app.config.base_url(), row.owner, row.name),
        token: secret,
        heads: run.head_list().into_iter().map(String::from).collect(),
        pipeline_hash: run.pipeline_hash.clone(),
        job: spec,
        sandbox,
        timeout_s,
        subrepos: vec![],
        state: run.state.clone(),
        bookmark: run.bookmark.clone(),
        run_number: run.number,
    };
    let mut claim = claim;
    let (inputs, ids) = subrepo_inputs(&app, &row, &run).await;
    if !ids.is_empty() {
        fdb::set_also_read(&app.db, job.id, &ids).await?;
    }
    claim.subrepos = inputs;
    Ok(Json(claim).into_response())
}

/// The Oxygen subrepo pins in `name`'s state `heads`: (path, repository id, pinned heads).
async fn pins_in(
    app: &Shared,
    name: crate::repos::RepoName,
    heads: Vec<oxy_core::hash::ChangeHash>,
) -> Vec<(String, [u8; 16], Vec<oxy_core::hash::ChangeHash>)> {
    app.repos
        .read(&name, move |repo, cache| {
            let loaded = cache.load(repo, &heads)?;
            let mut out = Vec::new();
            for (path, f) in &loaded.snapshot.files {
                if f.kind != oxy_core::objects::NodeKind::Subrepo {
                    continue;
                }
                if let oxy_core::state::FileContent::Opaque { blobs } = &f.content {
                    if let Some(b) = blobs.first() {
                        if let Ok(oxy_core::subrepo::Pin::Oxy { repo: id, heads, .. }) = repo
                            .read_blob(b)
                            .map_err(crate::repos::RepoError::Oxy)
                            .and_then(|bytes| oxy_core::subrepo::Pin::decode(&bytes).map_err(|e| crate::repos::RepoError::Oxy(e.into())))
                        {
                            out.push((path.clone(), id, heads));
                        }
                    }
                }
            }
            Ok(out)
        })
        .await
        .unwrap_or_default()
}

/// How deep subrepos of subrepos are followed.
const MAX_SUBREPO_DEPTH: usize = 3;

/// The Oxygen subrepos the run's state pins, and those their states pin in
/// turn (parents first, paths joined), that live on this hub and that the
/// run's pusher may read: what the job's workspace needs beside its own
/// tree. A pin the hub cannot serve is left out, with what it pins; the job
/// then fails on the missing files with an ordinary error.
async fn subrepo_inputs(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>) {
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![]) };
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
    let rows = db::repos::list(&app.db).await.unwrap_or_default();
    let mut inputs = Vec::new();
    let mut ids = Vec::new();
    // (path prefix, repository, heads, depth), parents before children.
    let mut queue = std::collections::VecDeque::from([(String::new(), name, heads, 0usize)]);
    while let Some((prefix, name, heads, depth)) = queue.pop_front() {
        if depth == MAX_SUBREPO_DEPTH {
            continue;
        }
        for (path, id, pinned) in pins_in(app, name, heads).await {
            let path = if prefix.is_empty() { path } else { format!("{prefix}/{path}") };
            for other in &rows {
                let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };
                let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);
                if !matches {
                    continue;
                }
                let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();
                if crate::auth::roles::allows(role, crate::auth::Access::Read) {
                    inputs.push(flow_core::api::SubrepoInput {
                        path: path.clone(),
                        repo: format!("{}/{}", other.owner, other.name),
                        clone_url: format!("{}/{}/{}", app.config.base_url(), other.owner, other.name),
                        heads: pinned.iter().map(|h| h.to_hex()).collect(),
                    });
                    if !ids.contains(&other.id) {
                        ids.push(other.id);
                    }
                    queue.push_back((path.clone(), other_name, pinned.clone(), depth + 1));
                }
                break;
            }
        }
    }
    (inputs, ids)
}

async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {
    match fdb::job(&app.db, id).await? {
        Some(j) if j.status == "running" => Ok(j),
        Some(_) => Err(ApiError(StatusCode::CONFLICT, "this job is not running".into())),
        None => Err(ApiError(StatusCode::NOT_FOUND, "no such job".into())),
    }
}

/// `POST /api/v1/flow/jobs/{id}/log`.
pub async fn log(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(id): Path<i64>,
    Json(mut batch): Json<LogBatch>,
) -> Result<StatusCode, ApiError> {
    runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    for r in &mut batch.records {
        if r.text.len() > MAX_RECORD_TEXT {
            let mut cut = MAX_RECORD_TEXT;
            while !r.text.is_char_boundary(cut) {
                cut -= 1;
            }
            r.text.truncate(cut);
        }
        if !matches!(r.stream.as_str(), "out" | "err" | "console" | "flow") {
            r.stream = "console".into();
        }
    }
    let kept = fdb::append_log(&app.db, job.id, &batch.records, MAX_RECORDS as i64).await?;
    if kept < batch.records.len() && job.log_count < MAX_RECORDS as i64 {
        tracing::info!("flow: job {} reached the log limit", job.id);
    }
    Ok(StatusCode::NO_CONTENT)
}

/// Largest artifact the hub keeps; the runner refuses larger ones first.
pub const MAX_ARTIFACT_BYTES: usize = 512 << 20;

/// `POST /api/v1/flow/jobs/{id}/artifacts/{name}`: the bytes of an artifact
/// the job declares (OxygenFlow spec §5.2), kept as a blob of the run's
/// repository. Answers `{"blob"}`.
pub async fn artifact(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path((id, name)): Path<(i64, String)>,
    body: axum::body::Bytes,
) -> Result<Response, ApiError> {
    runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    let spec: flow_core::Job =
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
    let Some(out) = spec.outputs.iter().find(|o| o.name == name) else {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no artifact {name:?}", job.name)));
    };
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
    let row =
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
    let content = blake3::hash(&body).to_hex().to_string();
    let size = body.len() as i64;
    let blob = app
        .repos
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the artifact: {e}")))?;
    let a = fdb::Artifact { job_id: job.id, name, path: out.path.clone(), blob: blob.to_hex(), size, content, created_ms: db::now_ms() };
    fdb::add_artifact(&app.db, &a).await?;
    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())
}

/// `POST /api/v1/flow/jobs/{id}/finish`.
pub async fn finish(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(id): Path<i64>,
    Json(f): Json<Finish>,
) -> Result<StatusCode, ApiError> {
    let me = runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    let status = match (f.exit_code, &f.error) {
        (Some(0), None) => "passed",
        (Some(_), _) => "failed",
        (None, _) => "error",
    };
    let (receipt, receipt_hash) = match &f.receipt {
        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {
            // The receipt names exactly the artifacts uploaded, in name order.
            let kept: Vec<String> = fdb::artifacts(&app.db, job.id).await?.into_iter().map(|a| a.blob).collect();
            if r.outputs.iter().map(hex::encode).collect::<Vec<_>>() != kept {
                return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt names other outputs than the artifacts uploaded".into()));
            }
            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))
        }
        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),
        None => (String::new(), String::new()),
    };
    let error = f.error.clone().unwrap_or_default();
    let changed = fdb::finish(
        &app.db,
        job.id,
        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },
    )
    .await?;
    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
            if let (Some(r), Some(signed)) = (&f.receipt, &f.signed_receipt) {
                let (attest, signer, note) = match keep_receipt(&app, &me, &row, &run, r, signed).await {
                    Ok((attest, signer)) => (attest, signer, String::new()),
                    Err(e) => {
                        tracing::warn!("flow: job {} receipt not kept: {e}", job.id);
                        (String::new(), String::new(), e)
                    }
                };
                fdb::set_receipt(&app.db, job.id, &attest, &signer, &note).await?;
            }
        }
    }
    report_changed(&app, job.run_id, changed).await?;
    Ok(StatusCode::NO_CONTENT)
}

/// Report the checks of jobs whose status changed.
async fn report_changed(app: &Shared, run_id: i64, changed: Vec<fdb::Job>) -> sqlx::Result<()> {
    let Some(run) = fdb::run_by_id(&app.db, run_id).await? else {
        return Ok(());
    };
    let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? else {
        return Ok(());
    };
    for j in changed {
        let summary = match j.status.as_str() {
            "passed" => format!("passed in {}", duration(&j)),
            "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),
            "skipped" => "skipped: a dependency did not pass".into(),
            _ => format!("could not run: {}", j.error),
        };
        report_check(app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;
    }
    Ok(())
}

/// The site's job time limits (oxygenhub#89): set on Admin → Runners →
/// Settings, else from `hub.toml`.
pub struct Limits {
    /// For a job whose pipeline names none.
    pub default_s: u64,
    /// The most a pipeline may ask for.
    pub max_s: u64,
    /// Grace past a job's limit before the hub gives it up: a runner stops
    /// a job at its limit and reports it within moments, so one still
    /// running this long after is one the runner no longer reports (it was
    /// restarted, or lost the hub while the job ended, oxygenhub#71).
    pub grace_ms: i64,
}

pub const SETTING_DEFAULT: &str = "flow.job_timeout_minutes";
pub const SETTING_MAX: &str = "flow.job_timeout_max_minutes";
pub const SETTING_GRACE: &str = "flow.lost_grace_minutes";

pub async fn limits(app: &Shared) -> Limits {
    let minutes = |key: &'static str| async move {
        db::accounts::setting(&app.db, key).await.ok().flatten().and_then(|v| v.parse::<u64>().ok()).filter(|m| *m > 0)
    };
    let default_m = minutes(SETTING_DEFAULT).await.unwrap_or(app.config.flow.job_timeout_minutes.max(1));
    let max_m = minutes(SETTING_MAX).await.unwrap_or(default_m.max(6 * 60)).max(default_m);
    let grace_m = minutes(SETTING_GRACE).await.unwrap_or(5);
    Limits { default_s: default_m * 60, max_s: max_m * 60, grace_ms: grace_m as i64 * 60_000 }
}

/// End every job still running past its time limit plus the grace:
/// it is marked an error with the reason, the jobs that depended on it are
/// skipped and its run settled, as when a runner reports an error.
pub async fn reap_lost_jobs(app: &Shared) -> sqlx::Result<usize> {
    let limits = limits(app).await;
    let lost = fdb::running_past_limit(&app.db, db::now_ms(), limits.default_s, limits.grace_ms).await?;
    let n = lost.len();
    for job in lost {
        let error = format!(
            "lost: still running {} minutes after it started, past the time limit; the runner{} stopped reporting it",
            (db::now_ms() - job.started_ms.unwrap_or(0)) / 60_000,
            if job.runner.is_empty() { String::new() } else { format!(" {}", job.runner) },
        );
        let changed =
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
                .await?;
        tracing::warn!("flow: job {} {error}", job.id);
        report_changed(app, job.run_id, changed).await?;
    }
    Ok(n)
}

/// End the jobs running on `runner` that it does not hold (oxygenhub#71):
/// it restarted since claiming them, or never received the claim. A
/// runner claims one job at a time, so a job it holds is always listed.
async fn end_jobs_not_held(app: &Shared, runner: &db::runners::Runner, holding: &[i64]) -> sqlx::Result<()> {
    for job in fdb::running_on(&app.db, runner.id).await? {
        if holding.contains(&job.id) {
            continue;
        }
        let error = format!("lost: the runner {} no longer holds this job (it restarted, or never received it)", runner.name);
        let changed =
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
                .await?;
        tracing::warn!("flow: job {} {error}", job.id);
        report_changed(app, job.run_id, changed).await?;
    }
    Ok(())
}

/// Start the scheduled runs due at `when` (oxygenflow#24): on each
/// repository that runs Flow, the jobs of its default bookmark's pipeline
/// with a `schedule` line due then. Returns how many runs started.
pub async fn run_due(app: &Shared, when: flow_core::pipeline::When) -> usize {
    let Ok(rows) = db::repos::list(&app.db).await else { return 0 };
    let mut started = 0;
    for row in rows.into_iter().filter(|r| enabled(app, r)) {
        let Some(repo) = RepoName::new(&row.owner, &row.name) else { continue };
        let b = row.default_bookmark.clone();
        let found = app
            .repos
            .read(&repo, move |r, _| {
                let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
                Ok(Some((st, r.store().get_state(&st)?.heads)))
            })
            .await;
        let Ok(Some((state, heads))) = found else { continue };
        if heads.is_empty() {
            continue;
        }
        match start_run_at(app, &row, &row.default_bookmark, &state.to_hex(), &heads, &row.owner, "schedule", None, Some(when)).await {
            Ok(Some(_)) => started += 1,
            Ok(None) => {}
            Err(e) => tracing::warn!("flow: scheduled run of {}/{}: {e}", row.owner, row.name),
        }
    }
    started
}

/// The hub's clock as a [`When`](flow_core::pipeline::When).
pub fn when_now() -> flow_core::pipeline::When {
    let now = time::OffsetDateTime::now_utc();
    flow_core::pipeline::When {
        minute: now.minute() as u32,
        hour: now.hour() as u32,
        day: now.day() as u32,
        month: u8::from(now.month()) as u32,
        weekday: now.weekday().number_days_from_sunday() as u32,
    }
}

/// Start the scheduled runs due each minute, once a minute.
pub async fn scheduler(app: Shared) {
    let mut last = None;
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
        let minute = time::OffsetDateTime::now_utc().unix_timestamp() / 60;
        if last == Some(minute) {
            continue;
        }
        last = Some(minute);
        let n = run_due(&app, when_now()).await;
        if n > 0 {
            tracing::info!("flow: {n} scheduled run(s) started");
        }
    }
}

/// Look for lost jobs once a minute.
pub async fn reaper(app: Shared) {
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
        if let Err(e) = reap_lost_jobs(&app).await {
            tracing::warn!("flow: looking for lost jobs: {e}");
        }
    }
}

/// Check a runner's signed receipt for a job of `run` and keep it in the
/// repository (Oxygen spec §14.6): the attestation's hex and the runner's
/// name, or why it was not kept. It must be the job's receipt, for the
/// run's state and repository, signed by the key the runner's token first
/// signed with.
pub async fn keep_receipt(
    app: &Shared,
    runner: &db::runners::Runner,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    receipt: &flow_core::Receipt,
    signed_hex: &str,
) -> Result<(String, String), String> {
    use oxy_core::receipt::SignedReceipt;
    let bytes = hex::decode(signed_hex).map_err(|_| "the signed receipt is not hex".to_string())?;
    let sr = SignedReceipt::decode(&bytes).map_err(|e| format!("not a signed receipt: {e}"))?;
    sr.verify().map_err(|e| format!("it does not verify: {e}"))?;
    if sr.receipt.hash().0 != receipt.hash() {
        return Err("it signs another receipt than the job's".into());
    }
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(heads).encode());
    if sr.state() != state {
        return Err("it is for another state than the run's".into());
    }
    let key = sr.runner.hash().to_hex();
    let bound = db::runners::bind_key(&app.db, runner.id, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;
    if bound != key {
        return Err(format!("signed by {}, not the key this runner signs with ({})", &key[..12], &bound[..12.min(bound.len())]));
    }
    let name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let signer = sr.runner.name.clone();
    let attest = app.repos.write(&name, move |repo, _| Ok(repo.add_receipt(&sr)?)).await.map_err(|e| e.to_string())?;
    Ok((attest.to_hex(), signer))
}

/// A job's wall time, as the pages show it.
pub fn duration(j: &fdb::Job) -> String {
    match (j.started_ms, j.finished_ms) {
        (Some(a), Some(b)) => human_ms(b - a),
        (Some(a), None) => human_ms(db::now_ms() - a),
        _ => "–".into(),
    }
}

pub fn human_ms(ms: i64) -> String {
    let ms = ms.max(0);
    if ms < 1000 {
        format!("{ms} ms")
    } else if ms < 60_000 {
        format!("{:.1} s", ms as f64 / 1000.0)
    } else {
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
    }
}

/// Where a live run stands among the runners' work: what its viewers may
/// know of it. Jobs of other repositories are named only when those are
/// public; the rest are counted.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct QueueInfo {
    /// 1-based place of the run's first queued job among all queued jobs.
    pub position: Option<usize>,
    /// Jobs running now, and queued jobs ahead of the run's first one.
    pub running: Vec<String>,
    pub ahead: Vec<String>,
    /// Of those, jobs in repositories the page may not name.
    pub hidden: usize,
}

impl QueueInfo {
    /// "3rd in line" and the like.
    pub fn place(&self) -> Option<String> {
        self.position.map(|n| {
            let suffix = match (n % 10, n % 100) {
                (1, 11) | (2, 12) | (3, 13) => "th",
                (1, _) => "st",
                (2, _) => "nd",
                (3, _) => "rd",
                _ => "th",
            };
            format!("{n}{suffix} in line")
        })
    }
}

/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.
pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {
    let mut info = QueueInfo::default();
    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);
    info.position = first.map(|i| i + 1);
    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);
    let mut queued_seen = 0;
    for e in queue {
        if e.run_id == run_id {
            continue;
        }
        let ahead = match e.status.as_str() {
            "running" => true,
            _ => {
                queued_seen += 1;
                first.is_some_and(|f| queued_seen <= f)
            }
        };
        if !ahead {
            continue;
        }
        if e.repo_id != repo_id && e.visibility != "public" {
            info.hidden += 1;
        } else if e.status == "running" {
            info.running.push(label(e));
        } else {
            info.ahead.push(label(e));
        }
    }
    info
}

#[cfg(test)]
mod queue_tests {
    use super::*;

    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {
        fdb::QueueEntry {
            job_id,
            run_id,
            repo_id,
            owner: "o".into(),
            name: format!("r{repo_id}"),
            visibility: vis.into(),
            run_number: run_id,
            job: job.into(),
            status: status.into(),
            runner: String::new(),
            spec: String::new(),
            sandbox: String::new(),
        }
    }

    #[test]
    fn a_run_learns_its_place_and_what_is_ahead_of_it() {
        // Running first, then queued in claim order (as `fdb::queue` returns them).
        let q = vec![
            e(1, 10, 1, "public", "build", "running"),
            e(2, 11, 2, "private", "test", "running"),
            e(3, 10, 1, "public", "test", "queued"),
            e(4, 12, 3, "private", "lint", "queued"),
            e(5, 13, 4, "public", "build", "queued"),
            e(6, 13, 4, "public", "test", "queued"),
            e(7, 14, 5, "public", "late", "queued"),
        ];
        let info = queue_info(&q, 4, 13);
        assert_eq!(info.position, Some(3));
        assert_eq!(info.place().as_deref(), Some("3rd in line"));
        assert_eq!(info.running, ["o/r1 #10 build"]);
        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");
        assert_eq!(info.hidden, 2, "private repositories are counted, not named");
        // A private repository's own jobs are named to its viewers.
        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);
        assert_eq!(queue_info(&q, 3, 12).position, Some(2));
        // A run with nothing queued has no place.
        assert_eq!(queue_info(&q, 1, 99).position, None);
        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();
        assert_eq!(
            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],
            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]
        );
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Keyboard
