OXYGENHUB
# 554ccacfc123 oxygenhub / crates / hub / src / flow.rs
63.2 KiB · Rust · last changed Coverage reports, built in 1 month ago source blame history raw
1 
//! OxygenFlow on the hub (hub spec §13.2): runs started by pushes, the
2 
//! runner API, job tokens and the checks that mirror jobs. Pipelines are
3 
//! parsed by `flow-core`, the same code `flow check` runs.
4
5 
use crate::api::ApiError;
6 
use crate::db::{self, flow as fdb};
7 
use crate::repos::RepoName;
8 
use crate::Shared;
9 
use axum::extract::{Path, State};
10 
use axum::http::{HeaderMap, StatusCode};
11 
use axum::response::{IntoResponse, Response};
12 
use axum::Json;
13 
use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, MAX_RECORDS, MAX_RECORD_TEXT};
14 
use flow_core::{Pipeline, PIPELINE_PATH};
15 
use oxy_core::hash::ChangeHash;
16
17 
/// Prefix of job tokens, so the wire protocol can tell them from personal tokens.
18 
pub const JOB_TOKEN_PREFIX: &str = "flowjob_";
19 
/// A job token is refused this long after its job started, whatever its state.
20 
pub const JOB_TOKEN_MAX_AGE_MS: i64 = 2 * 3600 * 1000;
21
22 
/// Whether pushes to this repository start runs: its own setting when a
23 
/// site administrator made one, else hub.toml's `[flow] repos`.
24 
pub fn enabled(app: &Shared, row: &db::repos::RepoRow) -> bool {
25 
    match row.flow.as_str() {
26 
        "on" => true,
27 
        "off" => false,
28 
        _ => app.config.flow.allows(&row.owner, &row.name),
29 
    }
30 
}
31
32 
/// A bookmark moved by a push: start a run when the repository is allowed
33 
/// and the new state has a pipeline.
34 
pub async fn on_push(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, state: &str, heads: &[ChangeHash], actor: &str) {
35 
    if heads.is_empty() || !enabled(app, row) {
36 
        return;
37 
    }
38 
    // A pushed tag runs the jobs whose `on` names it (oxygenflow#24).
39 
    let started = match bookmark.strip_prefix("tags/") {
40 
        Some(tag) => start_run_with(app, row, bookmark, state, heads, actor, &format!("tag:{tag}"), None).await,
41 
        None => {
42 
            let touched = touched_since_last_run(app, row, bookmark, heads).await;
43 
            let started = start_run_scoped(app, row, bookmark, state, heads, actor, "push", touched.as_deref()).await;
44 
            if let Ok(Some(n)) = started {
45 
                supersede(app, row, bookmark, n).await;
46 
            }
47 
            started
48 
        }
49 
    };
50 
    if let Err(e) = started {
51 
        tracing::warn!("flow: starting a run of {}/{}: {e}", row.owner, row.name);
52 
    }
53 
}
54
55 
/// "Run pipeline" (hub spec §13.2): run a bookmark's current state without
56 
/// moving it. Returns the new run's number.
57 
pub async fn run_bookmark(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, actor: &str) -> Result<i64, String> {
58 
    if !enabled(app, row) {
59 
        return Err("OxygenFlow is not enabled for this repository".into());
60 
    }
61 
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
62 
    let b = bookmark.to_string();
63 
    let found = app
64 
        .repos
65 
        .read(&repo, move |r, _| {
66 
            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
67 
            Ok(Some((st, r.store().get_state(&st)?.heads)))
68 
        })
69 
        .await
70 
        .map_err(|e| e.to_string())?;
71 
    let Some((state, heads)) = found else { return Err(format!("no bookmark {bookmark}")) };
72 
    if heads.is_empty() {
73 
        return Err(format!("{bookmark} has no changes yet"));
74 
    }
75 
    start_run(app, row, bookmark, &state.to_hex(), &heads, actor, "manual")
76 
        .await?
77 
        .ok_or_else(|| format!("{bookmark} has no {PIPELINE_PATH}"))
78 
}
79
80 
/// "Run pipeline" on one committed change: the state whose only head is
81 
/// `change`, whatever bookmarks hold it. Returns the new run's number.
82 
pub async fn run_change(app: &Shared, row: &db::repos::RepoRow, change: &str, actor: &str) -> Result<i64, String> {
83 
    if !enabled(app, row) {
84 
        return Err("OxygenFlow is not enabled for this repository".into());
85 
    }
86 
    let h = ChangeHash::from_hex(change.trim().trim_start_matches('#')).ok_or("give the full hash of a change")?;
87 
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
88 
    let known = app.repos.read(&repo, move |r, _| Ok(r.store().get_change_meta(&h).is_ok())).await.map_err(|e| e.to_string())?;
89 
    if !known {
90 
        return Err(format!("no change {} in this repository", &change.trim()[..12.min(change.trim().len())]));
91 
    }
92 
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(vec![h]).encode());
93 
    start_run(app, row, "", &state.to_hex(), &[h], actor, "manual")
94 
        .await?
95 
        .ok_or_else(|| format!("that change's state has no {PIPELINE_PATH}"))
96 
}
97
98 
/// Re-run a finished run: the same state, bookmark and pipeline, as a new
99 
/// run whose checks replace the old ones. Returns the new run's number.
100 
pub async fn rerun(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, actor: &str) -> Result<i64, String> {
101 
    if !enabled(app, row) {
102 
        return Err("OxygenFlow is not enabled for this repository".into());
103 
    }
104 
    if run.live() {
105 
        return Err(format!("run #{} is still {}", run.number, run.status));
106 
    }
107 
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
108 
    if heads.is_empty() {
109 
        return Err("the run names no changes".into());
110 
    }
111 
    // A rerun repeats its run's event: the jobs it had are the jobs it gets.
112 
    let trigger = format!("rerun:{}:{}", run.number, original_event(run));
113 
    start_run(app, row, &run.bookmark, &run.state, &heads, actor, &trigger)
114 
        .await?
115 
        .ok_or_else(|| format!("that state has no {PIPELINE_PATH}"))
116 
}
117
118 
/// Cancel a run still going (oxygenflow#32), its checks with it.
119 
pub async fn cancel(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, actor: &str) -> Result<(), String> {
120 
    if !run.live() {
121 
        return Err(format!("run #{} is already {}", run.number, run.status));
122 
    }
123 
    cancel_with(app, run, &format!("cancelled by {actor}")).await.map_err(|e| {
124 
        tracing::warn!("flow: cancelling run #{} of {}/{}: {e}", run.number, row.owner, row.name);
125 
        "the run could not be cancelled".to_string()
126 
    })
127 
}
128
129 
async fn cancel_with(app: &Shared, run: &fdb::Run, why: &str) -> sqlx::Result<()> {
130 
    let changed = fdb::cancel_run(&app.db, run.id, why).await?;
131 
    report_changed(app, run.id, changed).await
132 
}
133
134 
/// A push makes the push runs still going on its bookmark moot: they are
135 
/// cancelled, and the new run's scope reaches back past them.
136 
async fn supersede(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, number: i64) {
137 
    let Ok(older) = fdb::live_push_runs_before(&app.db, row.id, bookmark, number).await else { return };
138 
    for run in older {
139 
        if let Err(e) = cancel_with(app, &run, &format!("superseded by run #{number}")).await {
140 
            tracing::warn!("flow: superseding run #{} of {}/{}: {e}", run.number, row.owner, row.name);
141 
        }
142 
    }
143 
}
144
145 
/// The event a run's trigger stands for: `push`, `tag:NAME`, `manual`,
146 
/// `schedule`, and `rerun:N:EVENT` repeating EVENT.
147 
fn event_of<'a>(trigger: &'a str, bookmark: &'a str) -> flow_core::pipeline::Event<'a> {
148 
    use flow_core::pipeline::Event;
149 
    let t = match trigger.strip_prefix("rerun:") {
150 
        Some(rest) => rest.split_once(':').map(|(_, e)| e).unwrap_or("manual"),
151 
        None => trigger,
152 
    };
153 
    if let Some(tag) = t.strip_prefix("tag:") {
154 
        Event::Tag(tag)
155 
    } else if t == "push" {
156 
        Event::Push(bookmark)
157 
    } else if t == "schedule" {
158 
        Event::Schedule
159 
    } else {
160 
        Event::Manual
161 
    }
162 
}
163
164 
/// The event to repeat when rerunning `run`.
165 
fn original_event(run: &fdb::Run) -> String {
166 
    let t = run.trigger.as_str();
167 
    match t.strip_prefix("rerun:") {
168 
        Some(rest) => rest.split_once(':').map(|(_, e)| e.to_string()).unwrap_or_else(|| "manual".into()),
169 
        None => t.to_string(),
170 
    }
171 
}
172
173 
/// The paths the changes a push brought touch: the changes in `heads`'
174 
/// state that the bookmark's previous run did not have, and the files their
175 
/// edits and tree operations name, as the new state places them. `None`
176 
/// (run every job) for the bookmark's first run, or when it cannot be told.
177 
async fn touched_since_last_run(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Option<Vec<String>> {
178 
    let last = fdb::latest_on(&app.db, row.id, bookmark).await.ok()??;
179 
    let old: Vec<ChangeHash> = last.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
180 
    if old.is_empty() {
181 
        return None;
182 
    }
183 
    let repo = RepoName::new(&row.owner, &row.name)?;
184 
    let new = heads.to_vec();
185 
    app.repos
186 
        .read(&repo, move |r, cache| {
187 
            let fresh = r.negotiate(&old, &new)?;
188 
            let loaded = cache.load(r, &new)?;
189 
            let by_node: std::collections::HashMap<oxy_core::ids::NodeId, &String> =
190 
                loaded.snapshot.files.iter().map(|(p, f)| (f.node, p)).collect();
191 
            let mut paths = std::collections::BTreeSet::new();
192 
            for h in fresh {
193 
                let (c, _) = r.store().get_change_lazy(&h)?;
194 
                let nodes = c.edits.iter().map(|e| e.node()).chain(c.tree_ops.iter().map(|o| o.node()));
195 
                for n in nodes {
196 
                    if let Some(p) = by_node.get(&n) {
197 
                        paths.insert((*p).clone());
198 
                    }
199 
                }
200 
            }
201 
            Ok(paths.into_iter().collect::<Vec<String>>())
202 
        })
203 
        .await
204 
        .ok()
205 
}
206
207 
/// Whether a job runs for a push that touched `touched`: always without a
208 
/// scope or without knowing what was touched, else when a path matches.
209 
fn in_scope(scope: &[String], touched: Option<&[String]>) -> bool {
210 
    let Some(touched) = touched else { return true };
211 
    if scope.is_empty() {
212 
        return true;
213 
    }
214 
    let mut b = globset::GlobSetBuilder::new();
215 
    for g in scope {
216 
        match globset::Glob::new(g) {
217 
            Ok(g) => {
218 
                b.add(g);
219 
            }
220 
            Err(_) => return true,
221 
        }
222 
    }
223 
    let Ok(set) = b.build() else { return true };
224 
    touched.iter().any(|p| set.is_match(p))
225 
}
226
227 
/// [`start_run`] for a push: jobs whose `scope` the push did not touch are
228 
/// left out of the run, unless a job that runs depends on them.
229 
#[allow(clippy::too_many_arguments)]
230 
pub async fn start_run_scoped(
231 
    app: &Shared,
232 
    row: &db::repos::RepoRow,
233 
    bookmark: &str,
234 
    state: &str,
235 
    heads: &[ChangeHash],
236 
    actor: &str,
237 
    trigger: &str,
238 
    touched: Option<&[String]>,
239 
) -> Result<Option<i64>, String> {
240 
    start_run_with(app, row, bookmark, state, heads, actor, trigger, touched).await
241 
}
242
243 
/// Start a run of the pipeline in the state `heads` make (whose state
244 
/// hash is `state`), recorded on `bookmark`: for a push, "Run pipeline"
245 
/// (`trigger` = `manual`) and re-runs (`rerun:N`). `actor` is the account
246 
/// whose access the run's subrepo reads are checked against. Returns the
247 
/// run's number, or `None` when the state has no pipeline. The caller
248 
/// checks that Flow is enabled.
249 
pub async fn start_run(
250 
    app: &Shared,
251 
    row: &db::repos::RepoRow,
252 
    bookmark: &str,
253 
    state: &str,
254 
    heads: &[ChangeHash],
255 
    actor: &str,
256 
    trigger: &str,
257 
) -> Result<Option<i64>, String> {
258 
    start_run_with(app, row, bookmark, state, heads, actor, trigger, None).await
259 
}
260
261 
#[allow(clippy::too_many_arguments)]
262 
async fn start_run_with(
263 
    app: &Shared,
264 
    row: &db::repos::RepoRow,
265 
    bookmark: &str,
266 
    state: &str,
267 
    heads: &[ChangeHash],
268 
    actor: &str,
269 
    trigger: &str,
270 
    touched: Option<&[String]>,
271 
) -> Result<Option<i64>, String> {
272 
    start_run_at(app, row, bookmark, state, heads, actor, trigger, touched, None).await
273 
}
274
275 
/// [`start_run_with`], for a scheduled run due at `at` on the hub's clock.
276 
#[allow(clippy::too_many_arguments)]
277 
async fn start_run_at(
278 
    app: &Shared,
279 
    row: &db::repos::RepoRow,
280 
    bookmark: &str,
281 
    state: &str,
282 
    heads: &[ChangeHash],
283 
    actor: &str,
284 
    trigger: &str,
285 
    touched: Option<&[String]>,
286 
    at: Option<flow_core::pipeline::When>,
287 
) -> Result<Option<i64>, String> {
288 
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
289 
    let h = heads.to_vec();
290 
    let text = app
291 
        .repos
292 
        .read(&repo, move |r, cache| {
293 
            let loaded = cache.load(r, &h)?;
294 
            Ok(match loaded.snapshot.files.get(PIPELINE_PATH).map(|f| &f.content) {
295 
                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),
296 
                _ => None,
297 
            })
298 
        })
299 
        .await
300 
        .map_err(|e| format!("reading the pipeline: {e}"))?;
301 
    let Some(text) = text else { return Ok(None) };
302 
    let mut hex: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();
303 
    hex.sort();
304 
    let heads_s = hex.join("+");
305 
    let base = db::flow::NewRun { repo_id: row.id, bookmark, state, heads: &heads_s, pipeline_hash: "", actor, trigger, error: None };
306 
    let id = match Pipeline::parse(&text) {
307 
        Err(problems) => {
308 
            let msg = problems.to_string();
309 
            db::flow::create_run(&app.db, db::flow::NewRun { error: Some(&msg), ..base }, &[]).await.map_err(|e| e.to_string())?
310 
        }
311 
        Ok(p) => {
312 
            let depths = p.depths();
313 
            // Jobs out of the push's scope stay out, unless a kept job
314 
            // needs them.
315 
            // The jobs this run's event starts (`on`), out of those its
316 
            // push touched (`scope`); then what they depend on.
317 
            let event = match at {
318 
                Some(w) => flow_core::pipeline::Event::ScheduleAt(w),
319 
                None => event_of(trigger, bookmark),
320 
            };
321 
            let mut keep: std::collections::HashSet<String> = p
322 
                .order()
323 
                .into_iter()
324 
                .filter(|j| p.on_of(j).matches(event) && in_scope(&j.scope, touched))
325 
                .map(|j| j.name.clone())
326 
                .collect();
327 
            loop {
328 
                let needed: Vec<String> = p
329 
                    .order()
330 
                    .into_iter()
331 
                    .filter(|j| keep.contains(&j.name))
332 
                    .flat_map(|j| j.depends_on.iter().cloned())
333 
                    .filter(|d| !keep.contains(d))
334 
                    .collect();
335 
                if needed.is_empty() {
336 
                    break;
337 
                }
338 
                keep.extend(needed);
339 
            }
340 
            let jobs: Vec<fdb::NewJob> = p
341 
                .order()
342 
                .into_iter()
343 
                .filter(|j| keep.contains(&j.name))
344 
                .map(|j| {
345 
                    let sb = &p.sandboxes[&j.sandbox];
346 
                    fdb::NewJob {
347 
                        name: j.name.clone(),
348 
                        depth: depths[j.name.as_str()] as i64,
349 
                        depends: j.depends_on.clone(),
350 
                        spec: serde_json::to_string(j).unwrap_or_default(),
351 
                        sandbox: serde_json::to_string(sb).unwrap_or_default(),
352 
                        runtime: sb.runtime.as_str().into(),
353 
                    }
354 
                })
355 
                .collect();
356 
            if jobs.is_empty() {
357 
                // Nothing this event starts (or this push touched).
358 
                return Ok(None);
359 
            }
360 
            let hash = hex::encode(p.hash());
361 
            let id =
362 
                db::flow::create_run(&app.db, db::flow::NewRun { pipeline_hash: &hash, ..base }, &jobs).await.map_err(|e| e.to_string())?;
363 
            if let Ok(Some(run)) = fdb::run_by_id(&app.db, id).await {
364 
                for j in &jobs {
365 
                    report_check(app, row, state, run.number, &j.name, "queued", "waiting for a runner").await;
366 
                }
367 
            }
368 
            id
369 
        }
370 
    };
371 
    let run = fdb::run_by_id(&app.db, id).await.map_err(|e| e.to_string())?.ok_or("the run vanished")?;
372 
    Ok(Some(run.number))
373 
}
374
375 
/// Record a job's check; it links to the run's page by the run's number.
376 
async fn report_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_number: i64, job: &str, status: &str, summary: &str) {
377 
    let url = format!("{}/{}/{}/flow/{run_number}", app.config.base_url(), row.owner, row.name);
378 
    let name = format!("flow/{job}");
379 
    let c = db::hooks::upsert_check(
380 
        &app.db,
381 
        row.id,
382 
        db::hooks::CheckReport { state, name: &name, status, summary, url: &url, reporter: None, seal: "" },
383 
    )
384 
    .await;
385 
    if let Ok(c) = c {
386 
        let api = crate::api::check_of(&c);
387 
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
388 
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
389 
    }
390 
}
391
392 
/// The check status a job's status shows as.
393 
fn check_status(job_status: &str) -> &'static str {
394 
    match job_status {
395 
        "queued" => "queued",
396 
        "running" => "running",
397 
        "passed" => "passed",
398 
        "skipped" => "skipped",
399 
        "cancelled" => "cancelled",
400 
        _ => "failed",
401 
    }
402 
}
403
404 
/// The runner a bearer token belongs to: a registered runner, or one
405 
/// configured with a hub.toml token (tracked from its first request). A
406 
/// disabled runner is refused.
407 
async fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<db::runners::Runner, ApiError> {
408 
    let Some(t) = oxy_server::bearer_token(headers) else {
409 
        return Err(ApiError(StatusCode::UNAUTHORIZED, "a runner token is required".into()));
410 
    };
411 
    let hash = blake3::hash(t.as_bytes()).to_hex().to_string();
412 
    let runner = match db::runners::by_token(&app.db, &hash).await? {
413 
        Some(r) => r,
414 
        None if app.config.flow.runner_tokens.iter().any(|x| crate::auth::mac_equal(x, t)) => {
415 
            db::runners::configured(&app.db, &hash, "runner").await?
416 
        }
417 
        None => return Err(ApiError(StatusCode::UNAUTHORIZED, "not a runner token".into())),
418 
    };
419 
    if runner.state == "disabled" {
420 
        return Err(ApiError(StatusCode::FORBIDDEN, format!("runner {} is disabled on this hub", runner.name)));
421 
    }
422 
    Ok(runner)
423 
}
424
425 
/// `POST /api/v1/flow/runners/register`: a one-time registration token
426 
/// from Admin → Runners buys a runner credential, binding the runner's
427 
/// signing identity (hub spec §13.2).
428 
pub async fn register(
429 
    State(app): State<Shared>,
430 
    Json(req): Json<flow_core::api::Register>,
431 
) -> Result<Json<flow_core::api::Registered>, ApiError> {
432 
    let bytes = hex::decode(req.identity.trim()).map_err(|_| ApiError(StatusCode::BAD_REQUEST, "identity: not hex".into()))?;
433 
    let identity = oxy_core::identity::Identity::decode(&bytes).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("identity: {e}")))?;
434 
    let name: String = req.name.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).take(64).collect();
435 
    if name.is_empty() {
436 
        return Err(ApiError(StatusCode::BAD_REQUEST, "give the runner a name of letters, digits, - _ or .".into()));
437 
    }
438 
    let credential = format!("flowrunner_{}", crate::auth::random_hex(32));
439 
    let hash = blake3::hash(credential.as_bytes()).to_hex().to_string();
440 
    let reg = blake3::hash(req.token.trim().as_bytes()).to_hex().to_string();
441 
    let runner =
442 
        db::runners::register(&app.db, &reg, &name, &hash, &identity.hash().to_hex(), &identity.name, "").await?.ok_or_else(|| {
443 
            ApiError(StatusCode::FORBIDDEN, "that registration token is unknown, used or expired; create one on Admin → Runners".into())
444 
        })?;
445 
    db::runners::seen(&app.db, runner.id, None, Some(&req.info)).await?;
446 
    tracing::info!("flow: runner {} registered (key {})", runner.name, &runner.key[..12]);
447 
    Ok(Json(flow_core::api::Registered { id: runner.id, name: runner.name, credential }))
448 
}
449
450 
/// `POST /api/v1/flow/claim`.
451 
pub async fn claim(State(app): State<Shared>, headers: HeaderMap, Json(req): Json<ClaimRequest>) -> Result<Response, ApiError> {
452 
    let me = runner_auth(&app, &headers).await?;
453 
    let claimed: String = req.runner.chars().filter(|c| !c.is_control()).take(64).collect();
454 
    db::runners::seen(&app.db, me.id, Some(&claimed), req.info.as_ref()).await?;
455 
    if let Some(holding) = &req.holding {
456 
        end_jobs_not_held(&app, &me, holding).await?;
457 
    }
458 
    // A draining runner finishes what it has and takes nothing new.
459 
    if me.state == "draining" {
460 
        return Ok(StatusCode::NO_CONTENT.into_response());
461 
    }
462 
    let runner = if me.source == "registered" { me.name.clone() } else { claimed };
463 
    let secret = format!("{JOB_TOKEN_PREFIX}{}", crate::auth::random_hex(32));
464 
    // What it holds and what it is (OxygenFlow spec §11): the images'
465 
    // architectures it reports are remembered for every runner's claims.
466 
    if let Some(archs) = &req.image_archs {
467 
        fdb::record_image_archs(&app.db, archs).await?;
468 
    }
469 
    let arch = req.info.as_ref().map(|i| i.arch.clone()).filter(|a| !a.is_empty()).unwrap_or_else(|| me.arch.clone());
470 
    // What it offers, kept for the placement pass; a job it declined is
471 
    // not offered to it again while this stays the same (spec §11.5).
472 
    let profile = profile_of(&req, &arch);
473 
    db::runners::set_offer(&app.db, me.id, &req.runtimes, req.images.as_deref(), &profile).await?;
474 
    let room = fdb::Room {
475 
        declined: fdb::declined_by(&app.db, me.id, &profile).await?,
476 
        memory_mib: req.free_memory_mib,
477 
        cpus: req.free_cpus,
478 
        images: req.images.as_ref().map(|i| i.iter().map(|h| h.to_ascii_lowercase()).collect()),
479 
        arch,
480 
        image_archs: fdb::image_archs(&app.db).await?,
481 
    };
482 
    // The labels it reports now, or those it registered with, and those
483 
    // an administrator gave it (#98).
484 
    let mut labels: Vec<String> = match &req.info {
485 
        Some(i) => i.labels.clone(),
486 
        None => me.reported_label_list().into_iter().map(String::from).collect(),
487 
    };
488 
    for l in me.admin_label_list() {
489 
        if !labels.iter().any(|x| x == l) {
490 
            labels.push(l.to_string());
491 
        }
492 
    }
493 
    let Some(job) =
494 
        fdb::claim_within(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope(), &labels).await?
495 
    else {
496 
        return Ok(StatusCode::NO_CONTENT.into_response());
497 
    };
498 
    db::runners::set_job_runner(&app.db, job.id, me.id).await?;
499 
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
500 
    let row =
501 
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
502 
    report_check(&app, &row, &run.state, run.number, &job.name, "running", &format!("running on {runner}")).await;
503 
    let bad = |e: serde_json::Error| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}"));
504 
    // The job's limit: its own timeout, else its sandbox's, else the site
505 
    // default, never past the site maximum; recorded for the reaper.
506 
    let limits = limits(&app).await;
507 
    let spec: flow_core::pipeline::Job = serde_json::from_str(&job.spec).map_err(bad)?;
508 
    let sandbox: flow_core::pipeline::Sandbox = serde_json::from_str(&job.sandbox).map_err(bad)?;
509 
    let mut timeout_s = spec.timeout_s.or(sandbox.timeout_s).unwrap_or(limits.default_s).min(limits.max_s).max(60);
510 
    // ...and never past the runner's own limit, when it has one (#98).
511 
    if let Some(m) = me.timeout_minutes.filter(|m| *m > 0) {
512 
        timeout_s = timeout_s.min(m as u64 * 60).max(60);
513 
    }
514 
    fdb::set_job_timeout(&app.db, job.id, timeout_s).await?;
515 
    let claim = Claim {
516 
        job_id: job.id,
517 
        run_id: run.id,
518 
        repo: format!("{}/{}", row.owner, row.name),
519 
        clone_url: format!("{}/{}/{}", app.config.base_url(), row.owner, row.name),
520 
        token: secret,
521 
        heads: run.head_list().into_iter().map(String::from).collect(),
522 
        pipeline_hash: run.pipeline_hash.clone(),
523 
        job: spec,
524 
        sandbox,
525 
        timeout_s,
526 
        subrepos: vec![],
527 
        state: run.state.clone(),
528 
        bookmark: run.bookmark.clone(),
529 
        run_number: run.number,
530 
        // The site's, the owner's and the repository's variables (#102).
531 
        env: crate::variables::env_for_repo(&app.db, &row).await?,
532 
        secrets: vec![],
533 
    };
534 
    let mut claim = claim;
535 
    let (inputs, ids, mut left_out) = subrepo_inputs(&app, &row, &run).await;
536 
    // The secrets the job declares, as files (#107); one not set is said
537 
    // at the top of the log and left out, so the job fails on the file.
538 
    let names: Vec<String> = claim.job.secrets.iter().map(|s| s.name.clone()).collect();
539 
    let found = crate::secrets::for_job(&app, &row, &names).await?;
540 
    // The identities it asks for (#103): a token each, as a file.
541 
    if !claim.job.identities.is_empty() {
542 
        let issuer = app.oidc().await.map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("oidc: {e}")))?;
543 
        for i in &claim.job.identities {
544 
            let token = issuer.mint(&crate::oidc::JobClaims {
545 
                issuer: app.config.base_url(),
546 
                audience: &i.audience,
547 
                owner: &row.owner,
548 
                name: &row.name,
549 
                visibility: &row.visibility,
550 
                bookmark: &run.bookmark,
551 
                run_number: run.number,
552 
                job: &job.name,
553 
                state: &run.state,
554 
                runner: &runner,
555 
                actor: &run.actor,
556 
                now_ms: db::now_ms(),
557 
                ttl_s: timeout_s,
558 
            });
559 
            claim.secrets.push(flow_core::api::SecretInput {
560 
                name: format!("identity:{}", i.audience),
561 
                path: i.mount_path.clone(),
562 
                value: token,
563 
            });
564 
        }
565 
    }
566 
    for s in &claim.job.secrets {
567 
        match found.iter().find(|(n, _)| *n == s.name) {
568 
            Some((name, value)) => {
569 
                claim.secrets.push(flow_core::api::SecretInput { name: name.clone(), path: s.mount_path.clone(), value: value.clone() })
570 
            }
571 
            None => left_out.push(format!(
572 
                "secret {:?} is not set for this repository, its owner or the site; {} will be missing",
573 
                s.name, s.mount_path
574 
            )),
575 
        }
576 
    }
577 
    if !ids.is_empty() {
578 
        fdb::set_also_read(&app.db, job.id, &ids).await?;
579 
    }
580 
    // Say at the top of the job's log what it will not have (oxygenhub#56),
581 
    // rather than let it fail later on missing files.
582 
    if !left_out.is_empty() {
583 
        let records: Vec<flow_core::api::Record> =
584 
            left_out.into_iter().map(|text| flow_core::api::Record { t: 0, stream: "flow".into(), text }).collect();
585 
        fdb::append_log(&app.db, job.id, &records, MAX_RECORDS as i64).await?;
586 
    }
587 
    claim.subrepos = inputs;
588 
    Ok(Json(claim).into_response())
589 
}
590
591 
/// BLAKE3 over what a claiming runner offers: tiers, images, architecture,
592 
/// labels and capacity, each list sorted.
593 
fn profile_of(req: &ClaimRequest, arch: &str) -> String {
594 
    let sorted = |v: &[String]| {
595 
        let mut v = v.to_vec();
596 
        v.sort();
597 
        v
598 
    };
599 
    let info = req.info.as_ref();
600 
    let v = serde_json::json!({
601 
        "arch": arch,
602 
        "tiers": sorted(&req.runtimes),
603 
        "images": req.images.as_deref().map(sorted),
604 
        "labels": info.map(|i| sorted(&i.labels)),
605 
        "memory_mib": info.and_then(|i| i.memory_mib),
606 
        "cpus": info.and_then(|i| i.cpus),
607 
    });
608 
    blake3::hash(v.to_string().as_bytes()).to_hex().to_string()
609 
}
610
611 
/// A job handed back by this many different runners is unplaceable.
612 
pub const UNPLACEABLE_AFTER: usize = 3;
613
614 
/// `POST /api/v1/flow/jobs/{id}/decline` (OxygenFlow spec §11.5): the
615 
/// runner will not run a job it was handed. The job is queued again and
616 
/// not offered to that runner while its profile stays the same (and for
617 
/// [`fdb::DECLINE_HOLDS_MS`] at most); once [`UNPLACEABLE_AFTER`]
618 
/// different runners have declined it, it ends as unplaceable with their
619 
/// reasons. A decline is never a failed check.
620 
pub async fn decline(
621 
    State(app): State<Shared>,
622 
    headers: HeaderMap,
623 
    Path(id): Path<i64>,
624 
    Json(d): Json<flow_core::api::Decline>,
625 
) -> Result<StatusCode, ApiError> {
626 
    let me = runner_auth(&app, &headers).await?;
627 
    db::runners::seen(&app.db, me.id, None, None).await?;
628 
    let reason: String = d.reason.chars().map(|c| if c.is_control() { ' ' } else { c }).take(500).collect();
629 
    let reason = match reason.trim() {
630 
        "" => "no reason given".to_string(),
631 
        r => r.to_string(),
632 
    };
633 
    let Some(all) = fdb::decline(&app.db, id, me.id, &me.name, &me.profile, &reason).await? else {
634 
        return Err(ApiError(StatusCode::CONFLICT, "that job is not running on this runner".into()));
635 
    };
636 
    tracing::info!("flow: runner {} declined job {id}: {reason}", me.name);
637 
    let job = fdb::job(&app.db, id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such job".into()))?;
638 
    let runners: std::collections::BTreeSet<i64> = all.iter().map(|d| d.runner_id).collect();
639 
    if runners.len() >= UNPLACEABLE_AFTER {
640 
        let error = format!(
641 
            "unplaceable: declined by {} runners: {}",
642 
            runners.len(),
643 
            all.iter().map(|d| format!("{}: {}", d.runner, d.reason)).collect::<Vec<_>>().join("; ")
644 
        );
645 
        let changed =
646 
            fdb::finish(&app.db, id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
647 
                .await?;
648 
        report_changed(&app, job.run_id, changed).await?;
649 
    } else if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
650 
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
651 
            let summary = format!("declined by {}: {reason}; waiting for another runner", me.name);
652 
            report_check(&app, &row, &run.state, run.number, &job.name, "queued", &summary).await;
653 
            fdb::set_waiting(&app.db, id, &summary).await?;
654 
        }
655 
    }
656 
    Ok(StatusCode::NO_CONTENT)
657 
}
658
659 
/// Why no runner can take `job` of a repository of `owner` (id `repo_id`)
660 
/// now (spec §11.4), from what each runner offered at its last claim:
661 
/// empty when one could (it waits its turn or a free slot), else what
662 
/// each one lacks. `declines` are the job's.
663 
pub fn why_waiting(
664 
    job: &fdb::Job,
665 
    owner: &str,
666 
    repo_id: i64,
667 
    runners: &[db::runners::Runner],
668 
    image_archs: &std::collections::HashMap<String, String>,
669 
    declines: &[fdb::DeclineRow],
670 
    now: i64,
671 
) -> String {
672 
    let sandbox: serde_json::Value = serde_json::from_str(&job.sandbox).unwrap_or_default();
673 
    let rootfs = sandbox["rootfs"].as_str().unwrap_or("");
674 
    let wanted_arch = job.arch().or_else(|| image_archs.get(rootfs).cloned());
675 
    let need = |k: &str| sandbox[k].as_u64().unwrap_or(0);
676 
    let needs = job.needs();
677 
    let lacks = |r: &db::runners::Runner| -> Option<String> {
678 
        if r.state == "draining" {
679 
            return Some("is draining".into());
680 
        }
681 
        let scope = r.scope();
682 
        if (!scope.owner.is_empty() && scope.owner != owner) || scope.repo_id.is_some_and(|id| id != repo_id) {
683 
            return Some("serves other repositories".into());
684 
        }
685 
        let tiers = r.tier_list();
686 
        if !tiers.is_empty() && !tiers.contains(&job.runtime.as_str()) {
687 
            return Some(format!("has no {} tier (it offers {})", job.runtime, tiers.join(", ")));
688 
        }
689 
        if let Some(w) = wanted_arch.as_deref().filter(|w| !r.arch.is_empty() && *w != r.arch) {
690 
            return Some(format!("is {}; the job needs {w}", r.arch));
691 
        }
692 
        if r.image_list().is_some_and(|have| !have.iter().any(|h| h == rootfs)) {
693 
            return Some(format!("does not hold image {}", &rootfs[..12.min(rootfs.len())]));
694 
        }
695 
        let labels = r.label_list();
696 
        if let Some(l) = needs.iter().find(|l| !labels.contains(&l.as_str())) {
697 
            return Some(format!("lacks label {l}"));
698 
        }
699 
        if let Some(m) = r.memory_mib.filter(|m| (*m as u64) < need("memory_mib")) {
700 
            return Some(format!("has {m} MiB; the job needs {} MiB", need("memory_mib")));
701 
        }
702 
        if let Some(c) = r.cpus.filter(|c| (*c as u64) < need("cpus")) {
703 
            return Some(format!("has {c} CPUs; the job needs {}", need("cpus")));
704 
        }
705 
        if let Some(d) = declines.iter().find(|d| d.runner_id == r.id && now - d.ms < fdb::DECLINE_HOLDS_MS) {
706 
            return Some(format!("declined it: {}", d.reason));
707 
        }
708 
        None
709 
    };
710 
    let online: Vec<&db::runners::Runner> = runners.iter().filter(|r| r.state != "disabled" && r.online()).collect();
711 
    if online.is_empty() {
712 
        return "no runner is online".into();
713 
    }
714 
    let mut why = Vec::new();
715 
    for r in &online {
716 
        match lacks(r) {
717 
            None => return String::new(),
718 
            Some(w) => why.push(format!("{} {w}", r.name)),
719 
        }
720 
    }
721 
    let more = why.len().saturating_sub(4);
722 
    why.truncate(4);
723 
    let mut text = format!("no runner can take it: {}", why.join("; "));
724 
    if more > 0 {
725 
        text.push_str(&format!("; and {more} more"));
726 
    }
727 
    text
728 
}
729
730 
/// One placement pass: say why each ready queued job waits.
731 
pub async fn explain_waits(app: &Shared) -> sqlx::Result<()> {
732 
    let ready = fdb::ready_queued(&app.db).await?;
733 
    if ready.is_empty() {
734 
        return Ok(());
735 
    }
736 
    let runners = db::runners::list(&app.db).await?;
737 
    let archs = fdb::image_archs(&app.db).await?;
738 
    let now = db::now_ms();
739 
    for (job, owner, repo_id) in ready {
740 
        let declines = fdb::declines_of_job(&app.db, job.id).await?;
741 
        let why = why_waiting(&job, &owner, repo_id, &runners, &archs, &declines, now);
742 
        fdb::set_waiting(&app.db, job.id, &why).await?;
743 
    }
744 
    Ok(())
745 
}
746
747 
/// Run a placement pass every 15 seconds.
748 
pub async fn placement(app: Shared) {
749 
    loop {
750 
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
751 
        if let Err(e) = explain_waits(&app).await {
752 
            tracing::warn!("flow: placement pass: {e}");
753 
        }
754 
    }
755 
}
756
757 
/// The Oxygen subrepo pins in `name`'s state `heads`: (path, repository id, pinned heads).
758 
async fn pins_in(
759 
    app: &Shared,
760 
    name: crate::repos::RepoName,
761 
    heads: Vec<oxy_core::hash::ChangeHash>,
762 
) -> Vec<(String, [u8; 16], Vec<oxy_core::hash::ChangeHash>)> {
763 
    app.repos
764 
        .read(&name, move |repo, cache| {
765 
            let loaded = cache.load(repo, &heads)?;
766 
            let mut out = Vec::new();
767 
            for (path, f) in &loaded.snapshot.files {
768 
                if f.kind != oxy_core::objects::NodeKind::Subrepo {
769 
                    continue;
770 
                }
771 
                if let oxy_core::state::FileContent::Opaque { blobs } = &f.content {
772 
                    if let Some(b) = blobs.first() {
773 
                        if let Ok(oxy_core::subrepo::Pin::Oxy { repo: id, heads, .. }) = repo
774 
                            .read_blob(b)
775 
                            .map_err(crate::repos::RepoError::Oxy)
776 
                            .and_then(|bytes| oxy_core::subrepo::Pin::decode(&bytes).map_err(|e| crate::repos::RepoError::Oxy(e.into())))
777 
                        {
778 
                            out.push((path.clone(), id, heads));
779 
                        }
780 
                    }
781 
                }
782 
            }
783 
            Ok(out)
784 
        })
785 
        .await
786 
        .unwrap_or_default()
787 
}
788
789 
/// How deep subrepos of subrepos are followed.
790 
const MAX_SUBREPO_DEPTH: usize = 3;
791
792 
/// The Oxygen subrepos the run's state pins, and those their states pin in
793 
/// turn (parents first, paths joined), that live on this hub and that the
794 
/// run's pusher may read: what the job's workspace needs beside its own
795 
/// tree. A pin the hub cannot serve is left out, with what it pins, and
796 
/// the third list says why, for the job's log.
797 
async fn subrepo_inputs(
798 
    app: &Shared,
799 
    row: &db::repos::RepoRow,
800 
    run: &fdb::Run,
801 
) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>, Vec<String>) {
802 
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![], vec![]) };
803 
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
804 
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
805 
    let rows = db::repos::list(&app.db).await.unwrap_or_default();
806 
    let mut inputs = Vec::new();
807 
    let mut ids = Vec::new();
808 
    let mut left_out = Vec::new();
809 
    let actor = if run.actor.is_empty() { "the run's starter".to_string() } else { run.actor.clone() };
810 
    // (path prefix, repository, heads, depth), parents before children.
811 
    let mut queue = std::collections::VecDeque::from([(String::new(), name, heads, 0usize)]);
812 
    while let Some((prefix, name, heads, depth)) = queue.pop_front() {
813 
        if depth == MAX_SUBREPO_DEPTH {
814 
            continue;
815 
        }
816 
        for (path, id, pinned) in pins_in(app, name, heads).await {
817 
            let path = if prefix.is_empty() { path } else { format!("{prefix}/{path}") };
818 
            let mut hosted = false;
819 
            for other in &rows {
820 
                let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };
821 
                let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);
822 
                if !matches {
823 
                    continue;
824 
                }
825 
                hosted = true;
826 
                let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();
827 
                if crate::auth::roles::allows(role, crate::auth::Access::Read) {
828 
                    inputs.push(flow_core::api::SubrepoInput {
829 
                        path: path.clone(),
830 
                        repo: format!("{}/{}", other.owner, other.name),
831 
                        clone_url: format!("{}/{}/{}", app.config.base_url(), other.owner, other.name),
832 
                        heads: pinned.iter().map(|h| h.to_hex()).collect(),
833 
                    });
834 
                    if !ids.contains(&other.id) {
835 
                        ids.push(other.id);
836 
                    }
837 
                    queue.push_back((path.clone(), other_name, pinned.clone(), depth + 1));
838 
                } else {
839 
                    left_out.push(format!(
840 
                        "subrepo {path} ({}/{}) is left out: {actor} cannot read it, so the job has no files there",
841 
                        other.owner, other.name
842 
                    ));
843 
                }
844 
                break;
845 
            }
846 
            if !hosted {
847 
                left_out.push(format!("subrepo {path} is left out: the repository it pins is not on this hub"));
848 
            }
849 
        }
850 
    }
851 
    (inputs, ids, left_out)
852 
}
853
854 
async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {
855 
    match fdb::job(&app.db, id).await? {
856 
        Some(j) if j.status == "running" => Ok(j),
857 
        Some(_) => Err(ApiError(StatusCode::CONFLICT, "this job is not running".into())),
858 
        None => Err(ApiError(StatusCode::NOT_FOUND, "no such job".into())),
859 
    }
860 
}
861
862 
/// `POST /api/v1/flow/jobs/{id}/log`.
863 
pub async fn log(
864 
    State(app): State<Shared>,
865 
    headers: HeaderMap,
866 
    Path(id): Path<i64>,
867 
    Json(mut batch): Json<LogBatch>,
868 
) -> Result<StatusCode, ApiError> {
869 
    runner_auth(&app, &headers).await?;
870 
    let job = running_job(&app, id).await?;
871 
    // An empty batch asks only whether the job still runs: a cancelled
872 
    // one answers 409 and its runner stops it (oxygenflow#32).
873 
    if batch.records.is_empty() {
874 
        return Ok(StatusCode::NO_CONTENT);
875 
    }
876 
    // A secret's value never reaches the stored log (#107).
877 
    let values: Vec<String> = match serde_json::from_str::<flow_core::Job>(&job.spec) {
878 
        Ok(spec) if !spec.secrets.is_empty() => {
879 
            let run = fdb::run_by_id(&app.db, job.run_id).await?;
880 
            let row = match run {
881 
                Some(run) => db::repos::get_by_id(&app.db, run.repo_id).await?,
882 
                None => None,
883 
            };
884 
            match row {
885 
                Some(row) => {
886 
                    let names: Vec<String> = spec.secrets.iter().map(|s| s.name.clone()).collect();
887 
                    crate::secrets::for_job(&app, &row, &names).await?.into_iter().map(|(_, v)| v).collect()
888 
                }
889 
                None => vec![],
890 
            }
891 
        }
892 
        _ => vec![],
893 
    };
894 
    let kid = app.oidc.get().map(|i| i.kid.clone());
895 
    for r in &mut batch.records {
896 
        if !values.is_empty() {
897 
            r.text = crate::secrets::mask(&r.text, &values);
898 
        }
899 
        if let Some(kid) = &kid {
900 
            r.text = crate::oidc::mask(&r.text, kid);
901 
        }
902 
        if r.text.len() > MAX_RECORD_TEXT {
903 
            let mut cut = MAX_RECORD_TEXT;
904 
            while !r.text.is_char_boundary(cut) {
905 
                cut -= 1;
906 
            }
907 
            r.text.truncate(cut);
908 
        }
909 
        if !matches!(r.stream.as_str(), "out" | "err" | "console" | "flow") {
910 
            r.stream = "console".into();
911 
        }
912 
    }
913 
    let kept = fdb::append_log(&app.db, job.id, &batch.records, MAX_RECORDS as i64).await?;
914 
    if kept < batch.records.len() && job.log_count < MAX_RECORDS as i64 {
915 
        tracing::info!("flow: job {} reached the log limit", job.id);
916 
    }
917 
    Ok(StatusCode::NO_CONTENT)
918 
}
919
920 
/// Largest artifact the hub keeps; the runner refuses larger ones first.
921 
pub const MAX_ARTIFACT_BYTES: usize = 512 << 20;
922
923 
/// `POST /api/v1/flow/jobs/{id}/artifacts/{name}`: the bytes of an artifact
924 
/// the job declares (OxygenFlow spec §5.2), kept as a blob of the run's
925 
/// repository. Answers `{"blob"}`.
926 
pub async fn artifact(
927 
    State(app): State<Shared>,
928 
    headers: HeaderMap,
929 
    Path((id, name)): Path<(i64, String)>,
930 
    body: axum::body::Bytes,
931 
) -> Result<Response, ApiError> {
932 
    runner_auth(&app, &headers).await?;
933 
    let job = running_job(&app, id).await?;
934 
    let spec: flow_core::Job =
935 
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
936 
    let Some(out) = spec.outputs.iter().find(|o| o.name == name) else {
937 
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no artifact {name:?}", job.name)));
938 
    };
939 
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
940 
    let row =
941 
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
942 
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
943 
    let content = blake3::hash(&body).to_hex().to_string();
944 
    let size = body.len() as i64;
945 
    let report_bytes = (name == crate::coverage::ARTIFACT).then(|| body.clone());
946 
    let blob = app
947 
        .repos
948 
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
949 
        .await
950 
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the artifact: {e}")))?;
951 
    let a = fdb::Artifact { job_id: job.id, name, path: out.path.clone(), blob: blob.to_hex(), size, content, created_ms: db::now_ms() };
952 
    fdb::add_artifact(&app.db, &a).await?;
953 
    // A coverage report (#113) is read now; what it says, or why it could
954 
    // not be read, goes in the job's log. It never fails the upload.
955 
    if let Some(bytes) = report_bytes {
956 
        let said = match read_coverage(&app, &repo, &row, &run, &job, &a.blob, bytes).await {
957 
            Ok(text) | Err(text) => text,
958 
        };
959 
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said };
960 
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
961 
    }
962 
    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())
963 
}
964
965 
/// Parse and keep a job's coverage report: what it says, or why not.
966 
async fn read_coverage(
967 
    app: &Shared,
968 
    repo: &RepoName,
969 
    row: &db::repos::RepoRow,
970 
    run: &fdb::Run,
971 
    job: &fdb::Job,
972 
    blob: &str,
973 
    bytes: axum::body::Bytes,
974 
) -> Result<String, String> {
975 
    if bytes.len() > crate::coverage::MAX_REPORT {
976 
        return Err(format!(
977 
            "coverage: the report is {} MiB; at most {} MiB is read",
978 
            bytes.len() >> 20,
979 
            crate::coverage::MAX_REPORT >> 20
980 
        ));
981 
    }
982 
    let text = String::from_utf8(bytes.to_vec()).map_err(|_| "coverage: the report is not text (LCOV is expected)".to_string())?;
983 
    let (files, outside) = tokio::task::spawn_blocking(move || crate::coverage::parse_lcov(&text))
984 
        .await
985 
        .map_err(|e| format!("coverage: {e}"))?
986 
        .map_err(|e| format!("coverage: {e}"))?;
987 
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
988 
    let in_state: std::collections::HashSet<String> = app
989 
        .repos
990 
        .read(repo, move |r, cache| Ok(cache.load(r, &heads)?.snapshot.files.keys().cloned().collect()))
991 
        .await
992 
        .map_err(|e| format!("coverage: the run's files could not be read: {e}"))?;
993 
    let heads: Vec<String> = run.head_list().into_iter().map(String::from).collect();
994 
    let origin = crate::coverage::Origin {
995 
        repo_id: row.id,
996 
        run_id: run.id,
997 
        run_number: run.number,
998 
        job_id: job.id,
999 
        job: &job.name,
1000 
        bookmark: &run.bookmark,
1001 
        state: &run.state,
1002 
        heads: &heads,
1003 
        blob,
1004 
    };
1005 
    let id = crate::coverage::keep(app, &origin, files, outside, &in_state).await.map_err(|e| format!("coverage: {e}"))?;
1006 
    let Some(r) = crate::coverage::report(&app.db, row.id, id).await.map_err(|e| format!("coverage: {e}"))? else {
1007 
        return Err("coverage: the report was not kept".into());
1008 
    };
1009 
    let pct = r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into());
1010 
    let dropped =
1011 
        if r.dropped > 0 { format!(" ({} named files are not in this state and were left out)", r.dropped) } else { String::new() };
1012 
    Ok(format!("coverage: {pct} of lines ({} of {}) in {} files{dropped}", r.lines_hit, r.lines_found, r.files))
1013 
}
1014
1015 
/// `POST /api/v1/flow/jobs/{id}/finish`.
1016 
pub async fn finish(
1017 
    State(app): State<Shared>,
1018 
    headers: HeaderMap,
1019 
    Path(id): Path<i64>,
1020 
    Json(f): Json<Finish>,
1021 
) -> Result<StatusCode, ApiError> {
1022 
    let me = runner_auth(&app, &headers).await?;
1023 
    let job = running_job(&app, id).await?;
1024 
    let status = match (f.exit_code, &f.error) {
1025 
        (Some(0), None) => "passed",
1026 
        (Some(_), _) => "failed",
1027 
        (None, _) => "error",
1028 
    };
1029 
    let (receipt, receipt_hash) = match &f.receipt {
1030 
        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {
1031 
            // The receipt names exactly the artifacts uploaded, in name order.
1032 
            let kept: Vec<String> = fdb::artifacts(&app.db, job.id).await?.into_iter().map(|a| a.blob).collect();
1033 
            if r.outputs.iter().map(hex::encode).collect::<Vec<_>>() != kept {
1034 
                return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt names other outputs than the artifacts uploaded".into()));
1035 
            }
1036 
            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))
1037 
        }
1038 
        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),
1039 
        None => (String::new(), String::new()),
1040 
    };
1041 
    let error = f.error.clone().unwrap_or_default();
1042 
    let changed = fdb::finish(
1043 
        &app.db,
1044 
        job.id,
1045 
        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },
1046 
    )
1047 
    .await?;
1048 
    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
1049 
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
1050 
            if let (Some(r), Some(signed)) = (&f.receipt, &f.signed_receipt) {
1051 
                let (attest, signer, note) = match keep_receipt(&app, &me, &row, &run, r, signed).await {
1052 
                    Ok((attest, signer)) => (attest, signer, String::new()),
1053 
                    Err(e) => {
1054 
                        tracing::warn!("flow: job {} receipt not kept: {e}", job.id);
1055 
                        (String::new(), String::new(), e)
1056 
                    }
1057 
                };
1058 
                fdb::set_receipt(&app.db, job.id, &attest, &signer, &note).await?;
1059 
            }
1060 
        }
1061 
    }
1062 
    report_changed(&app, job.run_id, changed).await?;
1063 
    Ok(StatusCode::NO_CONTENT)
1064 
}
1065
1066 
/// Report the checks of jobs whose status changed.
1067 
async fn report_changed(app: &Shared, run_id: i64, changed: Vec<fdb::Job>) -> sqlx::Result<()> {
1068 
    let Some(run) = fdb::run_by_id(&app.db, run_id).await? else {
1069 
        return Ok(());
1070 
    };
1071 
    let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? else {
1072 
        return Ok(());
1073 
    };
1074 
    for j in changed {
1075 
        let summary = match j.status.as_str() {
1076 
            "passed" => format!("passed in {}", duration(&j)),
1077 
            "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),
1078 
            "skipped" => "skipped: a dependency did not pass".into(),
1079 
            "cancelled" => j.error.clone(),
1080 
            _ => format!("could not run: {}", j.error),
1081 
        };
1082 
        report_check(app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;
1083 
    }
1084 
    Ok(())
1085 
}
1086
1087 
/// The site's job time limits (oxygenhub#89): set on Admin → Runners →
1088 
/// Settings, else from `hub.toml`.
1089 
pub struct Limits {
1090 
    /// For a job whose pipeline names none.
1091 
    pub default_s: u64,
1092 
    /// The most a pipeline may ask for.
1093 
    pub max_s: u64,
1094 
    /// Grace past a job's limit before the hub gives it up: a runner stops
1095 
    /// a job at its limit and reports it within moments, so one still
1096 
    /// running this long after is one the runner no longer reports (it was
1097 
    /// restarted, or lost the hub while the job ended, oxygenhub#71).
1098 
    pub grace_ms: i64,
1099 
}
1100
1101 
pub const SETTING_DEFAULT: &str = "flow.job_timeout_minutes";
1102 
pub const SETTING_MAX: &str = "flow.job_timeout_max_minutes";
1103 
pub const SETTING_GRACE: &str = "flow.lost_grace_minutes";
1104
1105 
pub async fn limits(app: &Shared) -> Limits {
1106 
    let minutes = |key: &'static str| async move {
1107 
        db::accounts::setting(&app.db, key).await.ok().flatten().and_then(|v| v.parse::<u64>().ok()).filter(|m| *m > 0)
1108 
    };
1109 
    let default_m = minutes(SETTING_DEFAULT).await.unwrap_or(app.config.flow.job_timeout_minutes.max(1));
1110 
    let max_m = minutes(SETTING_MAX).await.unwrap_or(default_m.max(6 * 60)).max(default_m);
1111 
    let grace_m = minutes(SETTING_GRACE).await.unwrap_or(5);
1112 
    Limits { default_s: default_m * 60, max_s: max_m * 60, grace_ms: grace_m as i64 * 60_000 }
1113 
}
1114
1115 
/// End every job still running past its time limit plus the grace:
1116 
/// it is marked an error with the reason, the jobs that depended on it are
1117 
/// skipped and its run settled, as when a runner reports an error.
1118 
pub async fn reap_lost_jobs(app: &Shared) -> sqlx::Result<usize> {
1119 
    let limits = limits(app).await;
1120 
    let lost = fdb::running_past_limit(&app.db, db::now_ms(), limits.default_s, limits.grace_ms).await?;
1121 
    let n = lost.len();
1122 
    for job in lost {
1123 
        let error = format!(
1124 
            "lost: still running {} minutes after it started, past the time limit; the runner{} stopped reporting it",
1125 
            (db::now_ms() - job.started_ms.unwrap_or(0)) / 60_000,
1126 
            if job.runner.is_empty() { String::new() } else { format!(" {}", job.runner) },
1127 
        );
1128 
        let changed =
1129 
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
1130 
                .await?;
1131 
        tracing::warn!("flow: job {} {error}", job.id);
1132 
        report_changed(app, job.run_id, changed).await?;
1133 
    }
1134 
    Ok(n)
1135 
}
1136
1137 
/// End the jobs running on `runner` that it does not hold (oxygenhub#71):
1138 
/// it restarted since claiming them, or never received the claim. A
1139 
/// runner claims one job at a time, so a job it holds is always listed.
1140 
async fn end_jobs_not_held(app: &Shared, runner: &db::runners::Runner, holding: &[i64]) -> sqlx::Result<()> {
1141 
    for job in fdb::running_on(&app.db, runner.id).await? {
1142 
        if holding.contains(&job.id) {
1143 
            continue;
1144 
        }
1145 
        let error = format!("lost: the runner {} no longer holds this job (it restarted, or never received it)", runner.name);
1146 
        let changed =
1147 
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
1148 
                .await?;
1149 
        tracing::warn!("flow: job {} {error}", job.id);
1150 
        report_changed(app, job.run_id, changed).await?;
1151 
    }
1152 
    Ok(())
1153 
}
1154
1155 
/// Start the scheduled runs due at `when` (oxygenflow#24): on each
1156 
/// repository that runs Flow, the jobs of its default bookmark's pipeline
1157 
/// with a `schedule` line due then. Returns how many runs started.
1158 
pub async fn run_due(app: &Shared, when: flow_core::pipeline::When) -> usize {
1159 
    let Ok(rows) = db::repos::list(&app.db).await else { return 0 };
1160 
    let mut started = 0;
1161 
    for row in rows.into_iter().filter(|r| enabled(app, r)) {
1162 
        let Some(repo) = RepoName::new(&row.owner, &row.name) else { continue };
1163 
        let b = row.default_bookmark.clone();
1164 
        let found = app
1165 
            .repos
1166 
            .read(&repo, move |r, _| {
1167 
                let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
1168 
                Ok(Some((st, r.store().get_state(&st)?.heads)))
1169 
            })
1170 
            .await;
1171 
        let Ok(Some((state, heads))) = found else { continue };
1172 
        if heads.is_empty() {
1173 
            continue;
1174 
        }
1175 
        match start_run_at(app, &row, &row.default_bookmark, &state.to_hex(), &heads, &row.owner, "schedule", None, Some(when)).await {
1176 
            Ok(Some(_)) => started += 1,
1177 
            Ok(None) => {}
1178 
            Err(e) => tracing::warn!("flow: scheduled run of {}/{}: {e}", row.owner, row.name),
1179 
        }
1180 
    }
1181 
    started
1182 
}
1183
1184 
/// The hub's clock as a [`When`](flow_core::pipeline::When).
1185 
pub fn when_now() -> flow_core::pipeline::When {
1186 
    let now = time::OffsetDateTime::now_utc();
1187 
    flow_core::pipeline::When {
1188 
        minute: now.minute() as u32,
1189 
        hour: now.hour() as u32,
1190 
        day: now.day() as u32,
1191 
        month: u8::from(now.month()) as u32,
1192 
        weekday: now.weekday().number_days_from_sunday() as u32,
1193 
    }
1194 
}
1195
1196 
/// Start the scheduled runs due each minute, once a minute.
1197 
pub async fn scheduler(app: Shared) {
1198 
    let mut last = None;
1199 
    loop {
1200 
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
1201 
        let minute = time::OffsetDateTime::now_utc().unix_timestamp() / 60;
1202 
        if last == Some(minute) {
1203 
            continue;
1204 
        }
1205 
        last = Some(minute);
1206 
        let n = run_due(&app, when_now()).await;
1207 
        if n > 0 {
1208 
            tracing::info!("flow: {n} scheduled run(s) started");
1209 
        }
1210 
    }
1211 
}
1212
1213 
/// Look for lost jobs once a minute.
1214 
pub async fn reaper(app: Shared) {
1215 
    loop {
1216 
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
1217 
        if let Err(e) = reap_lost_jobs(&app).await {
1218 
            tracing::warn!("flow: looking for lost jobs: {e}");
1219 
        }
1220 
    }
1221 
}
1222
1223 
/// Check a runner's signed receipt for a job of `run` and keep it in the
1224 
/// repository (Oxygen spec §14.6): the attestation's hex and the runner's
1225 
/// name, or why it was not kept. It must be the job's receipt, for the
1226 
/// run's state and repository, signed by the key the runner's token first
1227 
/// signed with.
1228 
pub async fn keep_receipt(
1229 
    app: &Shared,
1230 
    runner: &db::runners::Runner,
1231 
    row: &db::repos::RepoRow,
1232 
    run: &fdb::Run,
1233 
    receipt: &flow_core::Receipt,
1234 
    signed_hex: &str,
1235 
) -> Result<(String, String), String> {
1236 
    use oxy_core::receipt::SignedReceipt;
1237 
    let bytes = hex::decode(signed_hex).map_err(|_| "the signed receipt is not hex".to_string())?;
1238 
    let sr = SignedReceipt::decode(&bytes).map_err(|e| format!("not a signed receipt: {e}"))?;
1239 
    sr.verify().map_err(|e| format!("it does not verify: {e}"))?;
1240 
    if sr.receipt.hash().0 != receipt.hash() {
1241 
        return Err("it signs another receipt than the job's".into());
1242 
    }
1243 
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
1244 
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(heads).encode());
1245 
    if sr.state() != state {
1246 
        return Err("it is for another state than the run's".into());
1247 
    }
1248 
    let key = sr.runner.hash().to_hex();
1249 
    let bound = db::runners::bind_key(&app.db, runner.id, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;
1250 
    if bound != key {
1251 
        return Err(format!("signed by {}, not the key this runner signs with ({})", &key[..12], &bound[..12.min(bound.len())]));
1252 
    }
1253 
    let name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
1254 
    let signer = sr.runner.name.clone();
1255 
    let attest = app.repos.write(&name, move |repo, _| Ok(repo.add_receipt(&sr)?)).await.map_err(|e| e.to_string())?;
1256 
    Ok((attest.to_hex(), signer))
1257 
}
1258
1259 
/// A job's wall time, as the pages show it.
1260 
pub fn duration(j: &fdb::Job) -> String {
1261 
    match (j.started_ms, j.finished_ms) {
1262 
        (Some(a), Some(b)) => human_ms(b - a),
1263 
        (Some(a), None) => human_ms(db::now_ms() - a),
1264 
        _ => "–".into(),
1265 
    }
1266 
}
1267
1268 
pub fn human_ms(ms: i64) -> String {
1269 
    let ms = ms.max(0);
1270 
    if ms < 1000 {
1271 
        format!("{ms} ms")
1272 
    } else if ms < 60_000 {
1273 
        format!("{:.1} s", ms as f64 / 1000.0)
1274 
    } else {
1275 
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
1276 
    }
1277 
}
1278
1279 
/// Where a live run stands among the runners' work: what its viewers may
1280 
/// know of it. Jobs of other repositories are named only when those are
1281 
/// public; the rest are counted.
1282 
#[derive(Clone, Debug, Default, PartialEq, Eq)]
1283 
pub struct QueueInfo {
1284 
    /// 1-based place of the run's first queued job among all queued jobs.
1285 
    pub position: Option<usize>,
1286 
    /// Jobs running now, and queued jobs ahead of the run's first one.
1287 
    pub running: Vec<String>,
1288 
    pub ahead: Vec<String>,
1289 
    /// Of those, jobs in repositories the page may not name.
1290 
    pub hidden: usize,
1291 
}
1292
1293 
impl QueueInfo {
1294 
    /// "3rd in line" and the like.
1295 
    pub fn place(&self) -> Option<String> {
1296 
        self.position.map(|n| {
1297 
            let suffix = match (n % 10, n % 100) {
1298 
                (1, 11) | (2, 12) | (3, 13) => "th",
1299 
                (1, _) => "st",
1300 
                (2, _) => "nd",
1301 
                (3, _) => "rd",
1302 
                _ => "th",
1303 
            };
1304 
            format!("{n}{suffix} in line")
1305 
        })
1306 
    }
1307 
}
1308
1309 
/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.
1310 
pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {
1311 
    let mut info = QueueInfo::default();
1312 
    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);
1313 
    info.position = first.map(|i| i + 1);
1314 
    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);
1315 
    let mut queued_seen = 0;
1316 
    for e in queue {
1317 
        if e.run_id == run_id {
1318 
            continue;
1319 
        }
1320 
        let ahead = match e.status.as_str() {
1321 
            "running" => true,
1322 
            _ => {
1323 
                queued_seen += 1;
1324 
                first.is_some_and(|f| queued_seen <= f)
1325 
            }
1326 
        };
1327 
        if !ahead {
1328 
            continue;
1329 
        }
1330 
        if e.repo_id != repo_id && e.visibility != "public" {
1331 
            info.hidden += 1;
1332 
        } else if e.status == "running" {
1333 
            info.running.push(label(e));
1334 
        } else {
1335 
            info.ahead.push(label(e));
1336 
        }
1337 
    }
1338 
    info
1339 
}
1340
1341 
#[cfg(test)]
1342 
mod queue_tests {
1343 
    use super::*;
1344
1345 
    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {
1346 
        fdb::QueueEntry {
1347 
            job_id,
1348 
            run_id,
1349 
            repo_id,
1350 
            owner: "o".into(),
1351 
            name: format!("r{repo_id}"),
1352 
            visibility: vis.into(),
1353 
            run_number: run_id,
1354 
            job: job.into(),
1355 
            status: status.into(),
1356 
            runner: String::new(),
1357 
            spec: String::new(),
1358 
            sandbox: String::new(),
1359 
        }
1360 
    }
1361
1362 
    #[test]
1363 
    fn a_run_learns_its_place_and_what_is_ahead_of_it() {
1364 
        // Running first, then queued in claim order (as `fdb::queue` returns them).
1365 
        let q = vec![
1366 
            e(1, 10, 1, "public", "build", "running"),
1367 
            e(2, 11, 2, "private", "test", "running"),
1368 
            e(3, 10, 1, "public", "test", "queued"),
1369 
            e(4, 12, 3, "private", "lint", "queued"),
1370 
            e(5, 13, 4, "public", "build", "queued"),
1371 
            e(6, 13, 4, "public", "test", "queued"),
1372 
            e(7, 14, 5, "public", "late", "queued"),
1373 
        ];
1374 
        let info = queue_info(&q, 4, 13);
1375 
        assert_eq!(info.position, Some(3));
1376 
        assert_eq!(info.place().as_deref(), Some("3rd in line"));
1377 
        assert_eq!(info.running, ["o/r1 #10 build"]);
1378 
        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");
1379 
        assert_eq!(info.hidden, 2, "private repositories are counted, not named");
1380 
        // A private repository's own jobs are named to its viewers.
1381 
        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);
1382 
        assert_eq!(queue_info(&q, 3, 12).position, Some(2));
1383 
        // A run with nothing queued has no place.
1384 
        assert_eq!(queue_info(&q, 1, 99).position, None);
1385 
        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();
1386 
        assert_eq!(
1387 
            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],
1388 
            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]
1389 
        );
1390 
    }
1391 
}
1392
1393 
#[cfg(test)]
1394 
mod placement_tests {
1395 
    use super::*;
1396
1397 
    #[test]
1398 
    fn a_waiting_job_says_what_each_runner_lacks() {
1399 
        use crate::db::runners::Runner;
1400 
        let rootfs = "ab".repeat(32);
1401 
        let job = fdb::Job {
1402 
            id: 7,
1403 
            runtime: "syscall".into(),
1404 
            sandbox: serde_json::json!({ "rootfs": rootfs, "memory_mib": 4096, "cpus": 2 }).to_string(),
1405 
            spec: serde_json::json!({ "name": "build" }).to_string(),
1406 
            ..Default::default()
1407 
        };
1408 
        let now = db::now_ms();
1409 
        let runner = |id: i64, name: &str| Runner {
1410 
            id,
1411 
            name: name.into(),
1412 
            state: "active".into(),
1413 
            last_seen_ms: Some(now),
1414 
            arch: "x86_64".into(),
1415 
            tiers: "microvm,syscall".into(),
1416 
            images: serde_json::to_string(&[&rootfs]).unwrap(),
1417 
            memory_mib: Some(16384),
1418 
            cpus: Some(8),
1419 
            ..Default::default()
1420 
        };
1421 
        let archs = std::collections::HashMap::new();
1422 
        let why = |runners: &[Runner], declines: &[fdb::DeclineRow]| why_waiting(&job, "andy", 1, runners, &archs, declines, now);
1423 
        assert_eq!(why(&[], &[]), "no runner is online");
1424 
        assert_eq!(why(&[runner(1, "earth-1")], &[]), "", "one can take it: it only waits its turn");
1425 
        let microvm_only = Runner { tiers: "microvm".into(), ..runner(1, "earth-1") };
1426 
        let riscv = Runner { arch: "riscv64".into(), ..runner(2, "narvi") };
1427 
        let small = Runner { memory_mib: Some(2048), ..runner(3, "pi") };
1428 
        let other = Runner { scope_owner: "bob".into(), ..runner(4, "bobs") };
1429 
        let offline = Runner { last_seen_ms: Some(now - 3_600_000), ..runner(5, "gone") };
1430 
        assert_eq!(
1431 
            why(&[microvm_only, small, other, offline], &[]),
1432 
            "no runner can take it: earth-1 has no syscall tier (it offers microvm); pi has 2048 MiB; the job needs 4096 MiB; bobs serves other repositories"
1433 
        );
1434 
        let archs_known: std::collections::HashMap<String, String> = [(rootfs.clone(), "x86_64".to_string())].into();
1435 
        assert_eq!(
1436 
            why_waiting(&job, "andy", 1, &[riscv], &archs_known, &[], now),
1437 
            "no runner can take it: narvi is riscv64; the job needs x86_64"
1438 
        );
1439 
        let without = Runner { images: "[]".into(), ..runner(6, "new") };
1440 
        assert_eq!(why(&[without], &[]), "no runner can take it: new does not hold image abababababab");
1441 
        let d = fdb::DeclineRow {
1442 
            job_id: 7,
1443 
            runner_id: 1,
1444 
            runner: "earth-1".into(),
1445 
            reason: "only 512 MiB of memory is free here".into(),
1446 
            ms: now,
1447 
        };
1448 
        assert_eq!(
1449 
            why(&[runner(1, "earth-1")], std::slice::from_ref(&d)),
1450 
            "no runner can take it: earth-1 declined it: only 512 MiB of memory is free here"
1451 
        );
1452 
        let old = fdb::DeclineRow { ms: now - fdb::DECLINE_HOLDS_MS - 1, ..d };
1453 
        assert_eq!(why(&[runner(1, "earth-1")], &[old]), "", "an old decline no longer holds");
1454 
    }
1455 
}