OXYGENHUB
# 67040b158ae2 oxygenhub / crates / hub / src / flow.rs
65.8 KiB · Rust · last changed Flow: say why a queued job waits, in oh and its check 25 days 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 
        if why == job.waiting {
743 
            continue;
744 
        }
745 
        fdb::set_waiting(&app.db, job.id, &why).await?;
746 
        // The job's check says it too, so a proposal shows why it is stuck
747 
        // (oxygenhub#28).
748 
        if let (Some(run), Some(row)) = (fdb::run_by_id(&app.db, job.run_id).await?, db::repos::get_by_id(&app.db, repo_id).await?) {
749 
            let summary = if why.is_empty() { "waiting for a runner" } else { why.as_str() };
750 
            report_check(app, &row, &run.state, run.number, &job.name, "queued", summary).await;
751 
        }
752 
    }
753 
    Ok(())
754 
}
755
756 
/// Run a placement pass every 15 seconds.
757 
pub async fn placement(app: Shared) {
758 
    loop {
759 
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
760 
        if let Err(e) = explain_waits(&app).await {
761 
            tracing::warn!("flow: placement pass: {e}");
762 
        }
763 
    }
764 
}
765
766 
/// The Oxygen subrepo pins in `name`'s state `heads`: (path, repository id, pinned heads).
767 
async fn pins_in(
768 
    app: &Shared,
769 
    name: crate::repos::RepoName,
770 
    heads: Vec<oxy_core::hash::ChangeHash>,
771 
) -> Vec<(String, [u8; 16], Vec<oxy_core::hash::ChangeHash>)> {
772 
    app.repos
773 
        .read(&name, move |repo, cache| {
774 
            let loaded = cache.load(repo, &heads)?;
775 
            let mut out = Vec::new();
776 
            for (path, f) in &loaded.snapshot.files {
777 
                if f.kind != oxy_core::objects::NodeKind::Subrepo {
778 
                    continue;
779 
                }
780 
                if let oxy_core::state::FileContent::Opaque { blobs } = &f.content {
781 
                    if let Some(b) = blobs.first() {
782 
                        if let Ok(oxy_core::subrepo::Pin::Oxy { repo: id, heads, .. }) = repo
783 
                            .read_blob(b)
784 
                            .map_err(crate::repos::RepoError::Oxy)
785 
                            .and_then(|bytes| oxy_core::subrepo::Pin::decode(&bytes).map_err(|e| crate::repos::RepoError::Oxy(e.into())))
786 
                        {
787 
                            out.push((path.clone(), id, heads));
788 
                        }
789 
                    }
790 
                }
791 
            }
792 
            Ok(out)
793 
        })
794 
        .await
795 
        .unwrap_or_default()
796 
}
797
798 
/// How deep subrepos of subrepos are followed.
799 
const MAX_SUBREPO_DEPTH: usize = 3;
800
801 
/// The Oxygen subrepos the run's state pins, and those their states pin in
802 
/// turn (parents first, paths joined), that live on this hub and that the
803 
/// run's pusher may read: what the job's workspace needs beside its own
804 
/// tree. A pin the hub cannot serve is left out, with what it pins, and
805 
/// the third list says why, for the job's log.
806 
async fn subrepo_inputs(
807 
    app: &Shared,
808 
    row: &db::repos::RepoRow,
809 
    run: &fdb::Run,
810 
) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>, Vec<String>) {
811 
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![], vec![]) };
812 
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
813 
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
814 
    let rows = db::repos::list(&app.db).await.unwrap_or_default();
815 
    let mut inputs = Vec::new();
816 
    let mut ids = Vec::new();
817 
    let mut left_out = Vec::new();
818 
    let actor = if run.actor.is_empty() { "the run's starter".to_string() } else { run.actor.clone() };
819 
    // (path prefix, repository, heads, depth), parents before children.
820 
    let mut queue = std::collections::VecDeque::from([(String::new(), name, heads, 0usize)]);
821 
    while let Some((prefix, name, heads, depth)) = queue.pop_front() {
822 
        if depth == MAX_SUBREPO_DEPTH {
823 
            continue;
824 
        }
825 
        for (path, id, pinned) in pins_in(app, name, heads).await {
826 
            let path = if prefix.is_empty() { path } else { format!("{prefix}/{path}") };
827 
            let mut hosted = false;
828 
            for other in &rows {
829 
                let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };
830 
                let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);
831 
                if !matches {
832 
                    continue;
833 
                }
834 
                hosted = true;
835 
                let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();
836 
                if crate::auth::roles::allows(role, crate::auth::Access::Read) {
837 
                    inputs.push(flow_core::api::SubrepoInput {
838 
                        path: path.clone(),
839 
                        repo: format!("{}/{}", other.owner, other.name),
840 
                        clone_url: format!("{}/{}/{}", app.config.base_url(), other.owner, other.name),
841 
                        heads: pinned.iter().map(|h| h.to_hex()).collect(),
842 
                    });
843 
                    if !ids.contains(&other.id) {
844 
                        ids.push(other.id);
845 
                    }
846 
                    queue.push_back((path.clone(), other_name, pinned.clone(), depth + 1));
847 
                } else {
848 
                    left_out.push(format!(
849 
                        "subrepo {path} ({}/{}) is left out: {actor} cannot read it, so the job has no files there",
850 
                        other.owner, other.name
851 
                    ));
852 
                }
853 
                break;
854 
            }
855 
            if !hosted {
856 
                left_out.push(format!("subrepo {path} is left out: the repository it pins is not on this hub"));
857 
            }
858 
        }
859 
    }
860 
    (inputs, ids, left_out)
861 
}
862
863 
async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {
864 
    match fdb::job(&app.db, id).await? {
865 
        Some(j) if j.status == "running" => Ok(j),
866 
        Some(_) => Err(ApiError(StatusCode::CONFLICT, "this job is not running".into())),
867 
        None => Err(ApiError(StatusCode::NOT_FOUND, "no such job".into())),
868 
    }
869 
}
870
871 
/// `POST /api/v1/flow/jobs/{id}/log`.
872 
pub async fn log(
873 
    State(app): State<Shared>,
874 
    headers: HeaderMap,
875 
    Path(id): Path<i64>,
876 
    Json(mut batch): Json<LogBatch>,
877 
) -> Result<StatusCode, ApiError> {
878 
    runner_auth(&app, &headers).await?;
879 
    let job = running_job(&app, id).await?;
880 
    // An empty batch asks only whether the job still runs: a cancelled
881 
    // one answers 409 and its runner stops it (oxygenflow#32).
882 
    if batch.records.is_empty() {
883 
        return Ok(StatusCode::NO_CONTENT);
884 
    }
885 
    // A secret's value never reaches the stored log (#107).
886 
    let values: Vec<String> = match serde_json::from_str::<flow_core::Job>(&job.spec) {
887 
        Ok(spec) if !spec.secrets.is_empty() => {
888 
            let run = fdb::run_by_id(&app.db, job.run_id).await?;
889 
            let row = match run {
890 
                Some(run) => db::repos::get_by_id(&app.db, run.repo_id).await?,
891 
                None => None,
892 
            };
893 
            match row {
894 
                Some(row) => {
895 
                    let names: Vec<String> = spec.secrets.iter().map(|s| s.name.clone()).collect();
896 
                    crate::secrets::for_job(&app, &row, &names).await?.into_iter().map(|(_, v)| v).collect()
897 
                }
898 
                None => vec![],
899 
            }
900 
        }
901 
        _ => vec![],
902 
    };
903 
    let kid = app.oidc.get().map(|i| i.kid.clone());
904 
    for r in &mut batch.records {
905 
        if !values.is_empty() {
906 
            r.text = crate::secrets::mask(&r.text, &values);
907 
        }
908 
        if let Some(kid) = &kid {
909 
            r.text = crate::oidc::mask(&r.text, kid);
910 
        }
911 
        if r.text.len() > MAX_RECORD_TEXT {
912 
            let mut cut = MAX_RECORD_TEXT;
913 
            while !r.text.is_char_boundary(cut) {
914 
                cut -= 1;
915 
            }
916 
            r.text.truncate(cut);
917 
        }
918 
        if !matches!(r.stream.as_str(), "out" | "err" | "console" | "flow") {
919 
            r.stream = "console".into();
920 
        }
921 
    }
922 
    let kept = fdb::append_log(&app.db, job.id, &batch.records, MAX_RECORDS as i64).await?;
923 
    if kept < batch.records.len() && job.log_count < MAX_RECORDS as i64 {
924 
        tracing::info!("flow: job {} reached the log limit", job.id);
925 
    }
926 
    Ok(StatusCode::NO_CONTENT)
927 
}
928
929 
/// Largest artifact the hub keeps; the runner refuses larger ones first.
930 
pub const MAX_ARTIFACT_BYTES: usize = 512 << 20;
931
932 
/// `POST /api/v1/flow/jobs/{id}/artifacts/{name}`: the bytes of an artifact
933 
/// the job declares (OxygenFlow spec §5.2), kept as a blob of the run's
934 
/// repository. Answers `{"blob"}`.
935 
pub async fn artifact(
936 
    State(app): State<Shared>,
937 
    headers: HeaderMap,
938 
    Path((id, name)): Path<(i64, String)>,
939 
    body: axum::body::Bytes,
940 
) -> Result<Response, ApiError> {
941 
    runner_auth(&app, &headers).await?;
942 
    let job = running_job(&app, id).await?;
943 
    let spec: flow_core::Job =
944 
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
945 
    let Some(out) = spec.outputs.iter().find(|o| o.name == name) else {
946 
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no artifact {name:?}", job.name)));
947 
    };
948 
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
949 
    let row =
950 
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
951 
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
952 
    let content = blake3::hash(&body).to_hex().to_string();
953 
    let size = body.len() as i64;
954 
    let report_bytes = (name == crate::coverage::ARTIFACT).then(|| body.clone());
955 
    let junit_bytes = (name == crate::junit::ARTIFACT).then(|| body.clone());
956 
    let blob = app
957 
        .repos
958 
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
959 
        .await
960 
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the artifact: {e}")))?;
961 
    let a = fdb::Artifact { job_id: job.id, name, path: out.path.clone(), blob: blob.to_hex(), size, content, created_ms: db::now_ms() };
962 
    fdb::add_artifact(&app.db, &a).await?;
963 
    // A coverage report (#113) is read now; what it says, or why it could
964 
    // not be read, goes in the job's log. It never fails the upload.
965 
    if let Some(bytes) = report_bytes {
966 
        let said = match read_coverage(&app, &repo, &row, &run, &job, &a.blob, bytes).await {
967 
            Ok(text) | Err(text) => text,
968 
        };
969 
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said };
970 
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
971 
    }
972 
    // A test report (#114) likewise.
973 
    if let Some(bytes) = junit_bytes {
974 
        let said = match read_junit(&app, &row, &run, &job, &a.blob, bytes).await {
975 
            Ok(text) | Err(text) => text,
976 
        };
977 
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said };
978 
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
979 
    }
980 
    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())
981 
}
982
983 
/// Parse and keep a job's JUnit report: what it says, or why not.
984 
async fn read_junit(
985 
    app: &Shared,
986 
    row: &db::repos::RepoRow,
987 
    run: &fdb::Run,
988 
    job: &fdb::Job,
989 
    blob: &str,
990 
    bytes: axum::body::Bytes,
991 
) -> Result<String, String> {
992 
    if bytes.len() > crate::junit::MAX_REPORT {
993 
        return Err(format!("tests: the report is {} MiB; at most {} MiB is read", bytes.len() >> 20, crate::junit::MAX_REPORT >> 20));
994 
    }
995 
    let text = String::from_utf8(bytes.to_vec()).map_err(|_| "tests: the report is not UTF-8 text (JUnit XML is expected)".to_string())?;
996 
    let cases = tokio::task::spawn_blocking(move || crate::junit::parse(&text))
997 
        .await
998 
        .map_err(|e| format!("tests: {e}"))?
999 
        .map_err(|e| format!("tests: {e}"))?;
1000 
    let origin = crate::junit::Origin {
1001 
        repo_id: row.id,
1002 
        run_id: run.id,
1003 
        run_number: run.number,
1004 
        job_id: job.id,
1005 
        job: &job.name,
1006 
        bookmark: &run.bookmark,
1007 
        state: &run.state,
1008 
        blob,
1009 
    };
1010 
    crate::junit::keep(app, &origin, &cases).await.map_err(|e| format!("tests: {e}"))?;
1011 
    let r = crate::junit::for_run(&app.db, run.id)
1012 
        .await
1013 
        .map_err(|e| format!("tests: {e}"))?
1014 
        .into_iter()
1015 
        .find(|r| r.job_id == job.id)
1016 
        .ok_or_else(|| "tests: the report was not kept".to_string())?;
1017 
    Ok(format!("tests: {} in {}", r.summary(), human_ms(r.time_ms)))
1018 
}
1019
1020 
/// Parse and keep a job's coverage report: what it says, or why not.
1021 
async fn read_coverage(
1022 
    app: &Shared,
1023 
    repo: &RepoName,
1024 
    row: &db::repos::RepoRow,
1025 
    run: &fdb::Run,
1026 
    job: &fdb::Job,
1027 
    blob: &str,
1028 
    bytes: axum::body::Bytes,
1029 
) -> Result<String, String> {
1030 
    if bytes.len() > crate::coverage::MAX_REPORT {
1031 
        return Err(format!(
1032 
            "coverage: the report is {} MiB; at most {} MiB is read",
1033 
            bytes.len() >> 20,
1034 
            crate::coverage::MAX_REPORT >> 20
1035 
        ));
1036 
    }
1037 
    let text = String::from_utf8(bytes.to_vec()).map_err(|_| "coverage: the report is not text (LCOV is expected)".to_string())?;
1038 
    let (files, outside) = tokio::task::spawn_blocking(move || crate::coverage::parse_lcov(&text))
1039 
        .await
1040 
        .map_err(|e| format!("coverage: {e}"))?
1041 
        .map_err(|e| format!("coverage: {e}"))?;
1042 
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
1043 
    let in_state: std::collections::HashSet<String> = app
1044 
        .repos
1045 
        .read(repo, move |r, cache| Ok(cache.load(r, &heads)?.snapshot.files.keys().cloned().collect()))
1046 
        .await
1047 
        .map_err(|e| format!("coverage: the run's files could not be read: {e}"))?;
1048 
    let heads: Vec<String> = run.head_list().into_iter().map(String::from).collect();
1049 
    let origin = crate::coverage::Origin {
1050 
        repo_id: row.id,
1051 
        run_id: run.id,
1052 
        run_number: run.number,
1053 
        job_id: job.id,
1054 
        job: &job.name,
1055 
        bookmark: &run.bookmark,
1056 
        state: &run.state,
1057 
        heads: &heads,
1058 
        blob,
1059 
    };
1060 
    let id = crate::coverage::keep(app, &origin, files, outside, &in_state).await.map_err(|e| format!("coverage: {e}"))?;
1061 
    let Some(r) = crate::coverage::report(&app.db, row.id, id).await.map_err(|e| format!("coverage: {e}"))? else {
1062 
        return Err("coverage: the report was not kept".into());
1063 
    };
1064 
    let pct = r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into());
1065 
    let dropped =
1066 
        if r.dropped > 0 { format!(" ({} named files are not in this state and were left out)", r.dropped) } else { String::new() };
1067 
    Ok(format!("coverage: {pct} of lines ({} of {}) in {} files{dropped}", r.lines_hit, r.lines_found, r.files))
1068 
}
1069
1070 
/// `POST /api/v1/flow/jobs/{id}/finish`.
1071 
pub async fn finish(
1072 
    State(app): State<Shared>,
1073 
    headers: HeaderMap,
1074 
    Path(id): Path<i64>,
1075 
    Json(f): Json<Finish>,
1076 
) -> Result<StatusCode, ApiError> {
1077 
    let me = runner_auth(&app, &headers).await?;
1078 
    let job = running_job(&app, id).await?;
1079 
    let status = match (f.exit_code, &f.error) {
1080 
        (Some(0), None) => "passed",
1081 
        (Some(_), _) => "failed",
1082 
        (None, _) => "error",
1083 
    };
1084 
    let (receipt, receipt_hash) = match &f.receipt {
1085 
        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {
1086 
            // A passing job's receipt names exactly the artifacts uploaded,
1087 
            // in name order; a failing one's names none, though it may have
1088 
            // kept `when "always"` artifacts (oxygenflow#37).
1089 
            let kept: Vec<String> =
1090 
                if r.exit_code == 0 { fdb::artifacts(&app.db, job.id).await?.into_iter().map(|a| a.blob).collect() } else { vec![] };
1091 
            if r.outputs.iter().map(hex::encode).collect::<Vec<_>>() != kept {
1092 
                return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt names other outputs than the artifacts uploaded".into()));
1093 
            }
1094 
            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))
1095 
        }
1096 
        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),
1097 
        None => (String::new(), String::new()),
1098 
    };
1099 
    let error = f.error.clone().unwrap_or_default();
1100 
    let changed = fdb::finish(
1101 
        &app.db,
1102 
        job.id,
1103 
        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },
1104 
    )
1105 
    .await?;
1106 
    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
1107 
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
1108 
            if let (Some(r), Some(signed)) = (&f.receipt, &f.signed_receipt) {
1109 
                let (attest, signer, note) = match keep_receipt(&app, &me, &row, &run, r, signed).await {
1110 
                    Ok((attest, signer)) => (attest, signer, String::new()),
1111 
                    Err(e) => {
1112 
                        tracing::warn!("flow: job {} receipt not kept: {e}", job.id);
1113 
                        (String::new(), String::new(), e)
1114 
                    }
1115 
                };
1116 
                fdb::set_receipt(&app.db, job.id, &attest, &signer, &note).await?;
1117 
            }
1118 
        }
1119 
    }
1120 
    report_changed(&app, job.run_id, changed).await?;
1121 
    Ok(StatusCode::NO_CONTENT)
1122 
}
1123
1124 
/// Report the checks of jobs whose status changed.
1125 
async fn report_changed(app: &Shared, run_id: i64, changed: Vec<fdb::Job>) -> sqlx::Result<()> {
1126 
    let Some(run) = fdb::run_by_id(&app.db, run_id).await? else {
1127 
        return Ok(());
1128 
    };
1129 
    let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? else {
1130 
        return Ok(());
1131 
    };
1132 
    for j in changed {
1133 
        let summary = match j.status.as_str() {
1134 
            "passed" => format!("passed in {}", duration(&j)),
1135 
            "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),
1136 
            "skipped" => "skipped: a dependency did not pass".into(),
1137 
            "cancelled" => j.error.clone(),
1138 
            _ => format!("could not run: {}", j.error),
1139 
        };
1140 
        report_check(app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;
1141 
    }
1142 
    Ok(())
1143 
}
1144
1145 
/// The site's job time limits (oxygenhub#89): set on Admin → Runners →
1146 
/// Settings, else from `hub.toml`.
1147 
pub struct Limits {
1148 
    /// For a job whose pipeline names none.
1149 
    pub default_s: u64,
1150 
    /// The most a pipeline may ask for.
1151 
    pub max_s: u64,
1152 
    /// Grace past a job's limit before the hub gives it up: a runner stops
1153 
    /// a job at its limit and reports it within moments, so one still
1154 
    /// running this long after is one the runner no longer reports (it was
1155 
    /// restarted, or lost the hub while the job ended, oxygenhub#71).
1156 
    pub grace_ms: i64,
1157 
}
1158
1159 
pub const SETTING_DEFAULT: &str = "flow.job_timeout_minutes";
1160 
pub const SETTING_MAX: &str = "flow.job_timeout_max_minutes";
1161 
pub const SETTING_GRACE: &str = "flow.lost_grace_minutes";
1162
1163 
pub async fn limits(app: &Shared) -> Limits {
1164 
    let minutes = |key: &'static str| async move {
1165 
        db::accounts::setting(&app.db, key).await.ok().flatten().and_then(|v| v.parse::<u64>().ok()).filter(|m| *m > 0)
1166 
    };
1167 
    let default_m = minutes(SETTING_DEFAULT).await.unwrap_or(app.config.flow.job_timeout_minutes.max(1));
1168 
    let max_m = minutes(SETTING_MAX).await.unwrap_or(default_m.max(6 * 60)).max(default_m);
1169 
    let grace_m = minutes(SETTING_GRACE).await.unwrap_or(5);
1170 
    Limits { default_s: default_m * 60, max_s: max_m * 60, grace_ms: grace_m as i64 * 60_000 }
1171 
}
1172
1173 
/// End every job still running past its time limit plus the grace:
1174 
/// it is marked an error with the reason, the jobs that depended on it are
1175 
/// skipped and its run settled, as when a runner reports an error.
1176 
pub async fn reap_lost_jobs(app: &Shared) -> sqlx::Result<usize> {
1177 
    let limits = limits(app).await;
1178 
    let lost = fdb::running_past_limit(&app.db, db::now_ms(), limits.default_s, limits.grace_ms).await?;
1179 
    let n = lost.len();
1180 
    for job in lost {
1181 
        let error = format!(
1182 
            "lost: still running {} minutes after it started, past the time limit; the runner{} stopped reporting it",
1183 
            (db::now_ms() - job.started_ms.unwrap_or(0)) / 60_000,
1184 
            if job.runner.is_empty() { String::new() } else { format!(" {}", job.runner) },
1185 
        );
1186 
        let changed =
1187 
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
1188 
                .await?;
1189 
        tracing::warn!("flow: job {} {error}", job.id);
1190 
        report_changed(app, job.run_id, changed).await?;
1191 
    }
1192 
    Ok(n)
1193 
}
1194
1195 
/// End the jobs running on `runner` that it does not hold (oxygenhub#71):
1196 
/// it restarted since claiming them, or never received the claim. A
1197 
/// runner claims one job at a time, so a job it holds is always listed.
1198 
async fn end_jobs_not_held(app: &Shared, runner: &db::runners::Runner, holding: &[i64]) -> sqlx::Result<()> {
1199 
    for job in fdb::running_on(&app.db, runner.id).await? {
1200 
        if holding.contains(&job.id) {
1201 
            continue;
1202 
        }
1203 
        let error = format!("lost: the runner {} no longer holds this job (it restarted, or never received it)", runner.name);
1204 
        let changed =
1205 
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
1206 
                .await?;
1207 
        tracing::warn!("flow: job {} {error}", job.id);
1208 
        report_changed(app, job.run_id, changed).await?;
1209 
    }
1210 
    Ok(())
1211 
}
1212
1213 
/// Start the scheduled runs due at `when` (oxygenflow#24): on each
1214 
/// repository that runs Flow, the jobs of its default bookmark's pipeline
1215 
/// with a `schedule` line due then. Returns how many runs started.
1216 
pub async fn run_due(app: &Shared, when: flow_core::pipeline::When) -> usize {
1217 
    let Ok(rows) = db::repos::list(&app.db).await else { return 0 };
1218 
    let mut started = 0;
1219 
    for row in rows.into_iter().filter(|r| enabled(app, r)) {
1220 
        let Some(repo) = RepoName::new(&row.owner, &row.name) else { continue };
1221 
        let b = row.default_bookmark.clone();
1222 
        let found = app
1223 
            .repos
1224 
            .read(&repo, move |r, _| {
1225 
                let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
1226 
                Ok(Some((st, r.store().get_state(&st)?.heads)))
1227 
            })
1228 
            .await;
1229 
        let Ok(Some((state, heads))) = found else { continue };
1230 
        if heads.is_empty() {
1231 
            continue;
1232 
        }
1233 
        match start_run_at(app, &row, &row.default_bookmark, &state.to_hex(), &heads, &row.owner, "schedule", None, Some(when)).await {
1234 
            Ok(Some(_)) => started += 1,
1235 
            Ok(None) => {}
1236 
            Err(e) => tracing::warn!("flow: scheduled run of {}/{}: {e}", row.owner, row.name),
1237 
        }
1238 
    }
1239 
    started
1240 
}
1241
1242 
/// The hub's clock as a [`When`](flow_core::pipeline::When).
1243 
pub fn when_now() -> flow_core::pipeline::When {
1244 
    let now = time::OffsetDateTime::now_utc();
1245 
    flow_core::pipeline::When {
1246 
        minute: now.minute() as u32,
1247 
        hour: now.hour() as u32,
1248 
        day: now.day() as u32,
1249 
        month: u8::from(now.month()) as u32,
1250 
        weekday: now.weekday().number_days_from_sunday() as u32,
1251 
    }
1252 
}
1253
1254 
/// Start the scheduled runs due each minute, once a minute.
1255 
pub async fn scheduler(app: Shared) {
1256 
    let mut last = None;
1257 
    loop {
1258 
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
1259 
        let minute = time::OffsetDateTime::now_utc().unix_timestamp() / 60;
1260 
        if last == Some(minute) {
1261 
            continue;
1262 
        }
1263 
        last = Some(minute);
1264 
        let n = run_due(&app, when_now()).await;
1265 
        if n > 0 {
1266 
            tracing::info!("flow: {n} scheduled run(s) started");
1267 
        }
1268 
    }
1269 
}
1270
1271 
/// Look for lost jobs once a minute.
1272 
pub async fn reaper(app: Shared) {
1273 
    loop {
1274 
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
1275 
        if let Err(e) = reap_lost_jobs(&app).await {
1276 
            tracing::warn!("flow: looking for lost jobs: {e}");
1277 
        }
1278 
    }
1279 
}
1280
1281 
/// Check a runner's signed receipt for a job of `run` and keep it in the
1282 
/// repository (Oxygen spec §14.6): the attestation's hex and the runner's
1283 
/// name, or why it was not kept. It must be the job's receipt, for the
1284 
/// run's state and repository, signed by the key the runner's token first
1285 
/// signed with.
1286 
pub async fn keep_receipt(
1287 
    app: &Shared,
1288 
    runner: &db::runners::Runner,
1289 
    row: &db::repos::RepoRow,
1290 
    run: &fdb::Run,
1291 
    receipt: &flow_core::Receipt,
1292 
    signed_hex: &str,
1293 
) -> Result<(String, String), String> {
1294 
    use oxy_core::receipt::SignedReceipt;
1295 
    let bytes = hex::decode(signed_hex).map_err(|_| "the signed receipt is not hex".to_string())?;
1296 
    let sr = SignedReceipt::decode(&bytes).map_err(|e| format!("not a signed receipt: {e}"))?;
1297 
    sr.verify().map_err(|e| format!("it does not verify: {e}"))?;
1298 
    if sr.receipt.hash().0 != receipt.hash() {
1299 
        return Err("it signs another receipt than the job's".into());
1300 
    }
1301 
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
1302 
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(heads).encode());
1303 
    if sr.state() != state {
1304 
        return Err("it is for another state than the run's".into());
1305 
    }
1306 
    let key = sr.runner.hash().to_hex();
1307 
    let bound = db::runners::bind_key(&app.db, runner.id, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;
1308 
    if bound != key {
1309 
        return Err(format!("signed by {}, not the key this runner signs with ({})", &key[..12], &bound[..12.min(bound.len())]));
1310 
    }
1311 
    let name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
1312 
    let signer = sr.runner.name.clone();
1313 
    let attest = app.repos.write(&name, move |repo, _| Ok(repo.add_receipt(&sr)?)).await.map_err(|e| e.to_string())?;
1314 
    Ok((attest.to_hex(), signer))
1315 
}
1316
1317 
/// A job's wall time, as the pages show it.
1318 
pub fn duration(j: &fdb::Job) -> String {
1319 
    match (j.started_ms, j.finished_ms) {
1320 
        (Some(a), Some(b)) => human_ms(b - a),
1321 
        (Some(a), None) => human_ms(db::now_ms() - a),
1322 
        _ => "–".into(),
1323 
    }
1324 
}
1325
1326 
pub fn human_ms(ms: i64) -> String {
1327 
    let ms = ms.max(0);
1328 
    if ms < 1000 {
1329 
        format!("{ms} ms")
1330 
    } else if ms < 60_000 {
1331 
        format!("{:.1} s", ms as f64 / 1000.0)
1332 
    } else {
1333 
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
1334 
    }
1335 
}
1336
1337 
/// Where a live run stands among the runners' work: what its viewers may
1338 
/// know of it. Jobs of other repositories are named only when those are
1339 
/// public; the rest are counted.
1340 
#[derive(Clone, Debug, Default, PartialEq, Eq)]
1341 
pub struct QueueInfo {
1342 
    /// 1-based place of the run's first queued job among all queued jobs.
1343 
    pub position: Option<usize>,
1344 
    /// Jobs running now, and queued jobs ahead of the run's first one.
1345 
    pub running: Vec<String>,
1346 
    pub ahead: Vec<String>,
1347 
    /// Of those, jobs in repositories the page may not name.
1348 
    pub hidden: usize,
1349 
}
1350
1351 
impl QueueInfo {
1352 
    /// "3rd in line" and the like.
1353 
    pub fn place(&self) -> Option<String> {
1354 
        self.position.map(|n| {
1355 
            let suffix = match (n % 10, n % 100) {
1356 
                (1, 11) | (2, 12) | (3, 13) => "th",
1357 
                (1, _) => "st",
1358 
                (2, _) => "nd",
1359 
                (3, _) => "rd",
1360 
                _ => "th",
1361 
            };
1362 
            format!("{n}{suffix} in line")
1363 
        })
1364 
    }
1365 
}
1366
1367 
/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.
1368 
pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {
1369 
    let mut info = QueueInfo::default();
1370 
    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);
1371 
    info.position = first.map(|i| i + 1);
1372 
    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);
1373 
    let mut queued_seen = 0;
1374 
    for e in queue {
1375 
        if e.run_id == run_id {
1376 
            continue;
1377 
        }
1378 
        let ahead = match e.status.as_str() {
1379 
            "running" => true,
1380 
            _ => {
1381 
                queued_seen += 1;
1382 
                first.is_some_and(|f| queued_seen <= f)
1383 
            }
1384 
        };
1385 
        if !ahead {
1386 
            continue;
1387 
        }
1388 
        if e.repo_id != repo_id && e.visibility != "public" {
1389 
            info.hidden += 1;
1390 
        } else if e.status == "running" {
1391 
            info.running.push(label(e));
1392 
        } else {
1393 
            info.ahead.push(label(e));
1394 
        }
1395 
    }
1396 
    info
1397 
}
1398
1399 
#[cfg(test)]
1400 
mod queue_tests {
1401 
    use super::*;
1402
1403 
    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {
1404 
        fdb::QueueEntry {
1405 
            job_id,
1406 
            run_id,
1407 
            repo_id,
1408 
            owner: "o".into(),
1409 
            name: format!("r{repo_id}"),
1410 
            visibility: vis.into(),
1411 
            run_number: run_id,
1412 
            job: job.into(),
1413 
            status: status.into(),
1414 
            runner: String::new(),
1415 
            spec: String::new(),
1416 
            sandbox: String::new(),
1417 
        }
1418 
    }
1419
1420 
    #[test]
1421 
    fn a_run_learns_its_place_and_what_is_ahead_of_it() {
1422 
        // Running first, then queued in claim order (as `fdb::queue` returns them).
1423 
        let q = vec![
1424 
            e(1, 10, 1, "public", "build", "running"),
1425 
            e(2, 11, 2, "private", "test", "running"),
1426 
            e(3, 10, 1, "public", "test", "queued"),
1427 
            e(4, 12, 3, "private", "lint", "queued"),
1428 
            e(5, 13, 4, "public", "build", "queued"),
1429 
            e(6, 13, 4, "public", "test", "queued"),
1430 
            e(7, 14, 5, "public", "late", "queued"),
1431 
        ];
1432 
        let info = queue_info(&q, 4, 13);
1433 
        assert_eq!(info.position, Some(3));
1434 
        assert_eq!(info.place().as_deref(), Some("3rd in line"));
1435 
        assert_eq!(info.running, ["o/r1 #10 build"]);
1436 
        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");
1437 
        assert_eq!(info.hidden, 2, "private repositories are counted, not named");
1438 
        // A private repository's own jobs are named to its viewers.
1439 
        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);
1440 
        assert_eq!(queue_info(&q, 3, 12).position, Some(2));
1441 
        // A run with nothing queued has no place.
1442 
        assert_eq!(queue_info(&q, 1, 99).position, None);
1443 
        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();
1444 
        assert_eq!(
1445 
            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],
1446 
            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]
1447 
        );
1448 
    }
1449 
}
1450
1451 
#[cfg(test)]
1452 
mod placement_tests {
1453 
    use super::*;
1454
1455 
    #[test]
1456 
    fn a_waiting_job_says_what_each_runner_lacks() {
1457 
        use crate::db::runners::Runner;
1458 
        let rootfs = "ab".repeat(32);
1459 
        let job = fdb::Job {
1460 
            id: 7,
1461 
            runtime: "syscall".into(),
1462 
            sandbox: serde_json::json!({ "rootfs": rootfs, "memory_mib": 4096, "cpus": 2 }).to_string(),
1463 
            spec: serde_json::json!({ "name": "build" }).to_string(),
1464 
            ..Default::default()
1465 
        };
1466 
        let now = db::now_ms();
1467 
        let runner = |id: i64, name: &str| Runner {
1468 
            id,
1469 
            name: name.into(),
1470 
            state: "active".into(),
1471 
            last_seen_ms: Some(now),
1472 
            arch: "x86_64".into(),
1473 
            tiers: "microvm,syscall".into(),
1474 
            images: serde_json::to_string(&[&rootfs]).unwrap(),
1475 
            memory_mib: Some(16384),
1476 
            cpus: Some(8),
1477 
            ..Default::default()
1478 
        };
1479 
        let archs = std::collections::HashMap::new();
1480 
        let why = |runners: &[Runner], declines: &[fdb::DeclineRow]| why_waiting(&job, "andy", 1, runners, &archs, declines, now);
1481 
        assert_eq!(why(&[], &[]), "no runner is online");
1482 
        assert_eq!(why(&[runner(1, "earth-1")], &[]), "", "one can take it: it only waits its turn");
1483 
        let microvm_only = Runner { tiers: "microvm".into(), ..runner(1, "earth-1") };
1484 
        let riscv = Runner { arch: "riscv64".into(), ..runner(2, "narvi") };
1485 
        let small = Runner { memory_mib: Some(2048), ..runner(3, "pi") };
1486 
        let other = Runner { scope_owner: "bob".into(), ..runner(4, "bobs") };
1487 
        let offline = Runner { last_seen_ms: Some(now - 3_600_000), ..runner(5, "gone") };
1488 
        assert_eq!(
1489 
            why(&[microvm_only, small, other, offline], &[]),
1490 
            "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"
1491 
        );
1492 
        let archs_known: std::collections::HashMap<String, String> = [(rootfs.clone(), "x86_64".to_string())].into();
1493 
        assert_eq!(
1494 
            why_waiting(&job, "andy", 1, &[riscv], &archs_known, &[], now),
1495 
            "no runner can take it: narvi is riscv64; the job needs x86_64"
1496 
        );
1497 
        let without = Runner { images: "[]".into(), ..runner(6, "new") };
1498 
        assert_eq!(why(&[without], &[]), "no runner can take it: new does not hold image abababababab");
1499 
        let d = fdb::DeclineRow {
1500 
            job_id: 7,
1501 
            runner_id: 1,
1502 
            runner: "earth-1".into(),
1503 
            reason: "only 512 MiB of memory is free here".into(),
1504 
            ms: now,
1505 
        };
1506 
        assert_eq!(
1507 
            why(&[runner(1, "earth-1")], std::slice::from_ref(&d)),
1508 
            "no runner can take it: earth-1 declined it: only 512 MiB of memory is free here"
1509 
        );
1510 
        let old = fdb::DeclineRow { ms: now - fdb::DECLINE_HOLDS_MS - 1, ..d };
1511 
        assert_eq!(why(&[runner(1, "earth-1")], &[old]), "", "an old decline no longer holds");
1512 
    }
1513 
}