OXYGENHUB
# 7fff30c4f5e2 oxygenhub / crates / hub / src / flow.rs
64.3 KiB · Rust · last changed Coverage: read Cobertura XML reports too 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())
983 
        .map_err(|_| "coverage: the report is not text (LCOV or Cobertura XML is expected)".to_string())?;
984 
    let (files, outside, format) = tokio::task::spawn_blocking(move || crate::coverage::parse(&text))
985 
        .await
986 
        .map_err(|e| format!("coverage: {e}"))?
987 
        .map_err(|e| format!("coverage: {e}"))?;
988 
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
989 
    let in_state: std::collections::HashSet<String> = app
990 
        .repos
991 
        .read(repo, move |r, cache| Ok(cache.load(r, &heads)?.snapshot.files.keys().cloned().collect()))
992 
        .await
993 
        .map_err(|e| format!("coverage: the run's files could not be read: {e}"))?;
994 
    let heads: Vec<String> = run.head_list().into_iter().map(String::from).collect();
995 
    let origin = crate::coverage::Origin {
996 
        repo_id: row.id,
997 
        run_id: run.id,
998 
        run_number: run.number,
999 
        job_id: job.id,
1000 
        job: &job.name,
1001 
        bookmark: &run.bookmark,
1002 
        state: &run.state,
1003 
        heads: &heads,
1004 
        blob,
1005 
        format,
1006 
    };
1007 
    let id = crate::coverage::keep(app, &origin, files, outside, &in_state).await.map_err(|e| format!("coverage: {e}"))?;
1008 
    let Some(r) = crate::coverage::report(&app.db, row.id, id).await.map_err(|e| format!("coverage: {e}"))? else {
1009 
        return Err("coverage: the report was not kept".into());
1010 
    };
1011 
    // The `coverage` check on the run's state (#113): against the default
1012 
    // bookmark, under the repository's rules.
1013 
    let base = crate::coverage::baseline(&app.db, &r, &row.default_bookmark).await.map_err(|e| format!("coverage: {e}"))?;
1014 
    let rules = crate::coverage::rules(&app.db, row.id).await.map_err(|e| format!("coverage: {e}"))?;
1015 
    let (status, summary) = crate::coverage::verdict(&r, base.as_ref(), &row.default_bookmark, rules);
1016 
    let url = format!("{}/{}/{}/coverage?report={}", app.config.base_url(), row.owner, row.name, r.id);
1017 
    if let Ok(c) = db::hooks::upsert_check(
1018 
        &app.db,
1019 
        row.id,
1020 
        db::hooks::CheckReport { state: &run.state, name: "coverage", status, summary: &summary, url: &url, reporter: None, seal: "" },
1021 
    )
1022 
    .await
1023 
    {
1024 
        let api = crate::api::check_of(&c);
1025 
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
1026 
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
1027 
    }
1028 
    let pct = r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into());
1029 
    let dropped =
1030 
        if r.dropped > 0 { format!(" ({} named files are not in this state and were left out)", r.dropped) } else { String::new() };
1031 
    Ok(format!("coverage: {pct} of lines ({} of {}) in {} files{dropped}", r.lines_hit, r.lines_found, r.files))
1032 
}
1033
1034 
/// `POST /api/v1/flow/jobs/{id}/finish`.
1035 
pub async fn finish(
1036 
    State(app): State<Shared>,
1037 
    headers: HeaderMap,
1038 
    Path(id): Path<i64>,
1039 
    Json(f): Json<Finish>,
1040 
) -> Result<StatusCode, ApiError> {
1041 
    let me = runner_auth(&app, &headers).await?;
1042 
    let job = running_job(&app, id).await?;
1043 
    let status = match (f.exit_code, &f.error) {
1044 
        (Some(0), None) => "passed",
1045 
        (Some(_), _) => "failed",
1046 
        (None, _) => "error",
1047 
    };
1048 
    let (receipt, receipt_hash) = match &f.receipt {
1049 
        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {
1050 
            // The receipt names exactly the artifacts uploaded, in name order.
1051 
            let kept: Vec<String> = fdb::artifacts(&app.db, job.id).await?.into_iter().map(|a| a.blob).collect();
1052 
            if r.outputs.iter().map(hex::encode).collect::<Vec<_>>() != kept {
1053 
                return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt names other outputs than the artifacts uploaded".into()));
1054 
            }
1055 
            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))
1056 
        }
1057 
        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),
1058 
        None => (String::new(), String::new()),
1059 
    };
1060 
    let error = f.error.clone().unwrap_or_default();
1061 
    let changed = fdb::finish(
1062 
        &app.db,
1063 
        job.id,
1064 
        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },
1065 
    )
1066 
    .await?;
1067 
    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
1068 
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
1069 
            if let (Some(r), Some(signed)) = (&f.receipt, &f.signed_receipt) {
1070 
                let (attest, signer, note) = match keep_receipt(&app, &me, &row, &run, r, signed).await {
1071 
                    Ok((attest, signer)) => (attest, signer, String::new()),
1072 
                    Err(e) => {
1073 
                        tracing::warn!("flow: job {} receipt not kept: {e}", job.id);
1074 
                        (String::new(), String::new(), e)
1075 
                    }
1076 
                };
1077 
                fdb::set_receipt(&app.db, job.id, &attest, &signer, &note).await?;
1078 
            }
1079 
        }
1080 
    }
1081 
    report_changed(&app, job.run_id, changed).await?;
1082 
    Ok(StatusCode::NO_CONTENT)
1083 
}
1084
1085 
/// Report the checks of jobs whose status changed.
1086 
async fn report_changed(app: &Shared, run_id: i64, changed: Vec<fdb::Job>) -> sqlx::Result<()> {
1087 
    let Some(run) = fdb::run_by_id(&app.db, run_id).await? else {
1088 
        return Ok(());
1089 
    };
1090 
    let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? else {
1091 
        return Ok(());
1092 
    };
1093 
    for j in changed {
1094 
        let summary = match j.status.as_str() {
1095 
            "passed" => format!("passed in {}", duration(&j)),
1096 
            "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),
1097 
            "skipped" => "skipped: a dependency did not pass".into(),
1098 
            "cancelled" => j.error.clone(),
1099 
            _ => format!("could not run: {}", j.error),
1100 
        };
1101 
        report_check(app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;
1102 
    }
1103 
    Ok(())
1104 
}
1105
1106 
/// The site's job time limits (oxygenhub#89): set on Admin → Runners →
1107 
/// Settings, else from `hub.toml`.
1108 
pub struct Limits {
1109 
    /// For a job whose pipeline names none.
1110 
    pub default_s: u64,
1111 
    /// The most a pipeline may ask for.
1112 
    pub max_s: u64,
1113 
    /// Grace past a job's limit before the hub gives it up: a runner stops
1114 
    /// a job at its limit and reports it within moments, so one still
1115 
    /// running this long after is one the runner no longer reports (it was
1116 
    /// restarted, or lost the hub while the job ended, oxygenhub#71).
1117 
    pub grace_ms: i64,
1118 
}
1119
1120 
pub const SETTING_DEFAULT: &str = "flow.job_timeout_minutes";
1121 
pub const SETTING_MAX: &str = "flow.job_timeout_max_minutes";
1122 
pub const SETTING_GRACE: &str = "flow.lost_grace_minutes";
1123
1124 
pub async fn limits(app: &Shared) -> Limits {
1125 
    let minutes = |key: &'static str| async move {
1126 
        db::accounts::setting(&app.db, key).await.ok().flatten().and_then(|v| v.parse::<u64>().ok()).filter(|m| *m > 0)
1127 
    };
1128 
    let default_m = minutes(SETTING_DEFAULT).await.unwrap_or(app.config.flow.job_timeout_minutes.max(1));
1129 
    let max_m = minutes(SETTING_MAX).await.unwrap_or(default_m.max(6 * 60)).max(default_m);
1130 
    let grace_m = minutes(SETTING_GRACE).await.unwrap_or(5);
1131 
    Limits { default_s: default_m * 60, max_s: max_m * 60, grace_ms: grace_m as i64 * 60_000 }
1132 
}
1133
1134 
/// End every job still running past its time limit plus the grace:
1135 
/// it is marked an error with the reason, the jobs that depended on it are
1136 
/// skipped and its run settled, as when a runner reports an error.
1137 
pub async fn reap_lost_jobs(app: &Shared) -> sqlx::Result<usize> {
1138 
    let limits = limits(app).await;
1139 
    let lost = fdb::running_past_limit(&app.db, db::now_ms(), limits.default_s, limits.grace_ms).await?;
1140 
    let n = lost.len();
1141 
    for job in lost {
1142 
        let error = format!(
1143 
            "lost: still running {} minutes after it started, past the time limit; the runner{} stopped reporting it",
1144 
            (db::now_ms() - job.started_ms.unwrap_or(0)) / 60_000,
1145 
            if job.runner.is_empty() { String::new() } else { format!(" {}", job.runner) },
1146 
        );
1147 
        let changed =
1148 
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
1149 
                .await?;
1150 
        tracing::warn!("flow: job {} {error}", job.id);
1151 
        report_changed(app, job.run_id, changed).await?;
1152 
    }
1153 
    Ok(n)
1154 
}
1155
1156 
/// End the jobs running on `runner` that it does not hold (oxygenhub#71):
1157 
/// it restarted since claiming them, or never received the claim. A
1158 
/// runner claims one job at a time, so a job it holds is always listed.
1159 
async fn end_jobs_not_held(app: &Shared, runner: &db::runners::Runner, holding: &[i64]) -> sqlx::Result<()> {
1160 
    for job in fdb::running_on(&app.db, runner.id).await? {
1161 
        if holding.contains(&job.id) {
1162 
            continue;
1163 
        }
1164 
        let error = format!("lost: the runner {} no longer holds this job (it restarted, or never received it)", runner.name);
1165 
        let changed =
1166 
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
1167 
                .await?;
1168 
        tracing::warn!("flow: job {} {error}", job.id);
1169 
        report_changed(app, job.run_id, changed).await?;
1170 
    }
1171 
    Ok(())
1172 
}
1173
1174 
/// Start the scheduled runs due at `when` (oxygenflow#24): on each
1175 
/// repository that runs Flow, the jobs of its default bookmark's pipeline
1176 
/// with a `schedule` line due then. Returns how many runs started.
1177 
pub async fn run_due(app: &Shared, when: flow_core::pipeline::When) -> usize {
1178 
    let Ok(rows) = db::repos::list(&app.db).await else { return 0 };
1179 
    let mut started = 0;
1180 
    for row in rows.into_iter().filter(|r| enabled(app, r)) {
1181 
        let Some(repo) = RepoName::new(&row.owner, &row.name) else { continue };
1182 
        let b = row.default_bookmark.clone();
1183 
        let found = app
1184 
            .repos
1185 
            .read(&repo, move |r, _| {
1186 
                let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
1187 
                Ok(Some((st, r.store().get_state(&st)?.heads)))
1188 
            })
1189 
            .await;
1190 
        let Ok(Some((state, heads))) = found else { continue };
1191 
        if heads.is_empty() {
1192 
            continue;
1193 
        }
1194 
        match start_run_at(app, &row, &row.default_bookmark, &state.to_hex(), &heads, &row.owner, "schedule", None, Some(when)).await {
1195 
            Ok(Some(_)) => started += 1,
1196 
            Ok(None) => {}
1197 
            Err(e) => tracing::warn!("flow: scheduled run of {}/{}: {e}", row.owner, row.name),
1198 
        }
1199 
    }
1200 
    started
1201 
}
1202
1203 
/// The hub's clock as a [`When`](flow_core::pipeline::When).
1204 
pub fn when_now() -> flow_core::pipeline::When {
1205 
    let now = time::OffsetDateTime::now_utc();
1206 
    flow_core::pipeline::When {
1207 
        minute: now.minute() as u32,
1208 
        hour: now.hour() as u32,
1209 
        day: now.day() as u32,
1210 
        month: u8::from(now.month()) as u32,
1211 
        weekday: now.weekday().number_days_from_sunday() as u32,
1212 
    }
1213 
}
1214
1215 
/// Start the scheduled runs due each minute, once a minute.
1216 
pub async fn scheduler(app: Shared) {
1217 
    let mut last = None;
1218 
    loop {
1219 
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
1220 
        let minute = time::OffsetDateTime::now_utc().unix_timestamp() / 60;
1221 
        if last == Some(minute) {
1222 
            continue;
1223 
        }
1224 
        last = Some(minute);
1225 
        let n = run_due(&app, when_now()).await;
1226 
        if n > 0 {
1227 
            tracing::info!("flow: {n} scheduled run(s) started");
1228 
        }
1229 
    }
1230 
}
1231
1232 
/// Look for lost jobs once a minute.
1233 
pub async fn reaper(app: Shared) {
1234 
    loop {
1235 
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
1236 
        if let Err(e) = reap_lost_jobs(&app).await {
1237 
            tracing::warn!("flow: looking for lost jobs: {e}");
1238 
        }
1239 
    }
1240 
}
1241
1242 
/// Check a runner's signed receipt for a job of `run` and keep it in the
1243 
/// repository (Oxygen spec §14.6): the attestation's hex and the runner's
1244 
/// name, or why it was not kept. It must be the job's receipt, for the
1245 
/// run's state and repository, signed by the key the runner's token first
1246 
/// signed with.
1247 
pub async fn keep_receipt(
1248 
    app: &Shared,
1249 
    runner: &db::runners::Runner,
1250 
    row: &db::repos::RepoRow,
1251 
    run: &fdb::Run,
1252 
    receipt: &flow_core::Receipt,
1253 
    signed_hex: &str,
1254 
) -> Result<(String, String), String> {
1255 
    use oxy_core::receipt::SignedReceipt;
1256 
    let bytes = hex::decode(signed_hex).map_err(|_| "the signed receipt is not hex".to_string())?;
1257 
    let sr = SignedReceipt::decode(&bytes).map_err(|e| format!("not a signed receipt: {e}"))?;
1258 
    sr.verify().map_err(|e| format!("it does not verify: {e}"))?;
1259 
    if sr.receipt.hash().0 != receipt.hash() {
1260 
        return Err("it signs another receipt than the job's".into());
1261 
    }
1262 
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
1263 
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(heads).encode());
1264 
    if sr.state() != state {
1265 
        return Err("it is for another state than the run's".into());
1266 
    }
1267 
    let key = sr.runner.hash().to_hex();
1268 
    let bound = db::runners::bind_key(&app.db, runner.id, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;
1269 
    if bound != key {
1270 
        return Err(format!("signed by {}, not the key this runner signs with ({})", &key[..12], &bound[..12.min(bound.len())]));
1271 
    }
1272 
    let name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
1273 
    let signer = sr.runner.name.clone();
1274 
    let attest = app.repos.write(&name, move |repo, _| Ok(repo.add_receipt(&sr)?)).await.map_err(|e| e.to_string())?;
1275 
    Ok((attest.to_hex(), signer))
1276 
}
1277
1278 
/// A job's wall time, as the pages show it.
1279 
pub fn duration(j: &fdb::Job) -> String {
1280 
    match (j.started_ms, j.finished_ms) {
1281 
        (Some(a), Some(b)) => human_ms(b - a),
1282 
        (Some(a), None) => human_ms(db::now_ms() - a),
1283 
        _ => "–".into(),
1284 
    }
1285 
}
1286
1287 
pub fn human_ms(ms: i64) -> String {
1288 
    let ms = ms.max(0);
1289 
    if ms < 1000 {
1290 
        format!("{ms} ms")
1291 
    } else if ms < 60_000 {
1292 
        format!("{:.1} s", ms as f64 / 1000.0)
1293 
    } else {
1294 
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
1295 
    }
1296 
}
1297
1298 
/// Where a live run stands among the runners' work: what its viewers may
1299 
/// know of it. Jobs of other repositories are named only when those are
1300 
/// public; the rest are counted.
1301 
#[derive(Clone, Debug, Default, PartialEq, Eq)]
1302 
pub struct QueueInfo {
1303 
    /// 1-based place of the run's first queued job among all queued jobs.
1304 
    pub position: Option<usize>,
1305 
    /// Jobs running now, and queued jobs ahead of the run's first one.
1306 
    pub running: Vec<String>,
1307 
    pub ahead: Vec<String>,
1308 
    /// Of those, jobs in repositories the page may not name.
1309 
    pub hidden: usize,
1310 
}
1311
1312 
impl QueueInfo {
1313 
    /// "3rd in line" and the like.
1314 
    pub fn place(&self) -> Option<String> {
1315 
        self.position.map(|n| {
1316 
            let suffix = match (n % 10, n % 100) {
1317 
                (1, 11) | (2, 12) | (3, 13) => "th",
1318 
                (1, _) => "st",
1319 
                (2, _) => "nd",
1320 
                (3, _) => "rd",
1321 
                _ => "th",
1322 
            };
1323 
            format!("{n}{suffix} in line")
1324 
        })
1325 
    }
1326 
}
1327
1328 
/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.
1329 
pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {
1330 
    let mut info = QueueInfo::default();
1331 
    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);
1332 
    info.position = first.map(|i| i + 1);
1333 
    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);
1334 
    let mut queued_seen = 0;
1335 
    for e in queue {
1336 
        if e.run_id == run_id {
1337 
            continue;
1338 
        }
1339 
        let ahead = match e.status.as_str() {
1340 
            "running" => true,
1341 
            _ => {
1342 
                queued_seen += 1;
1343 
                first.is_some_and(|f| queued_seen <= f)
1344 
            }
1345 
        };
1346 
        if !ahead {
1347 
            continue;
1348 
        }
1349 
        if e.repo_id != repo_id && e.visibility != "public" {
1350 
            info.hidden += 1;
1351 
        } else if e.status == "running" {
1352 
            info.running.push(label(e));
1353 
        } else {
1354 
            info.ahead.push(label(e));
1355 
        }
1356 
    }
1357 
    info
1358 
}
1359
1360 
#[cfg(test)]
1361 
mod queue_tests {
1362 
    use super::*;
1363
1364 
    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {
1365 
        fdb::QueueEntry {
1366 
            job_id,
1367 
            run_id,
1368 
            repo_id,
1369 
            owner: "o".into(),
1370 
            name: format!("r{repo_id}"),
1371 
            visibility: vis.into(),
1372 
            run_number: run_id,
1373 
            job: job.into(),
1374 
            status: status.into(),
1375 
            runner: String::new(),
1376 
            spec: String::new(),
1377 
            sandbox: String::new(),
1378 
        }
1379 
    }
1380
1381 
    #[test]
1382 
    fn a_run_learns_its_place_and_what_is_ahead_of_it() {
1383 
        // Running first, then queued in claim order (as `fdb::queue` returns them).
1384 
        let q = vec![
1385 
            e(1, 10, 1, "public", "build", "running"),
1386 
            e(2, 11, 2, "private", "test", "running"),
1387 
            e(3, 10, 1, "public", "test", "queued"),
1388 
            e(4, 12, 3, "private", "lint", "queued"),
1389 
            e(5, 13, 4, "public", "build", "queued"),
1390 
            e(6, 13, 4, "public", "test", "queued"),
1391 
            e(7, 14, 5, "public", "late", "queued"),
1392 
        ];
1393 
        let info = queue_info(&q, 4, 13);
1394 
        assert_eq!(info.position, Some(3));
1395 
        assert_eq!(info.place().as_deref(), Some("3rd in line"));
1396 
        assert_eq!(info.running, ["o/r1 #10 build"]);
1397 
        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");
1398 
        assert_eq!(info.hidden, 2, "private repositories are counted, not named");
1399 
        // A private repository's own jobs are named to its viewers.
1400 
        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);
1401 
        assert_eq!(queue_info(&q, 3, 12).position, Some(2));
1402 
        // A run with nothing queued has no place.
1403 
        assert_eq!(queue_info(&q, 1, 99).position, None);
1404 
        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();
1405 
        assert_eq!(
1406 
            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],
1407 
            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]
1408 
        );
1409 
    }
1410 
}
1411
1412 
#[cfg(test)]
1413 
mod placement_tests {
1414 
    use super::*;
1415
1416 
    #[test]
1417 
    fn a_waiting_job_says_what_each_runner_lacks() {
1418 
        use crate::db::runners::Runner;
1419 
        let rootfs = "ab".repeat(32);
1420 
        let job = fdb::Job {
1421 
            id: 7,
1422 
            runtime: "syscall".into(),
1423 
            sandbox: serde_json::json!({ "rootfs": rootfs, "memory_mib": 4096, "cpus": 2 }).to_string(),
1424 
            spec: serde_json::json!({ "name": "build" }).to_string(),
1425 
            ..Default::default()
1426 
        };
1427 
        let now = db::now_ms();
1428 
        let runner = |id: i64, name: &str| Runner {
1429 
            id,
1430 
            name: name.into(),
1431 
            state: "active".into(),
1432 
            last_seen_ms: Some(now),
1433 
            arch: "x86_64".into(),
1434 
            tiers: "microvm,syscall".into(),
1435 
            images: serde_json::to_string(&[&rootfs]).unwrap(),
1436 
            memory_mib: Some(16384),
1437 
            cpus: Some(8),
1438 
            ..Default::default()
1439 
        };
1440 
        let archs = std::collections::HashMap::new();
1441 
        let why = |runners: &[Runner], declines: &[fdb::DeclineRow]| why_waiting(&job, "andy", 1, runners, &archs, declines, now);
1442 
        assert_eq!(why(&[], &[]), "no runner is online");
1443 
        assert_eq!(why(&[runner(1, "earth-1")], &[]), "", "one can take it: it only waits its turn");
1444 
        let microvm_only = Runner { tiers: "microvm".into(), ..runner(1, "earth-1") };
1445 
        let riscv = Runner { arch: "riscv64".into(), ..runner(2, "narvi") };
1446 
        let small = Runner { memory_mib: Some(2048), ..runner(3, "pi") };
1447 
        let other = Runner { scope_owner: "bob".into(), ..runner(4, "bobs") };
1448 
        let offline = Runner { last_seen_ms: Some(now - 3_600_000), ..runner(5, "gone") };
1449 
        assert_eq!(
1450 
            why(&[microvm_only, small, other, offline], &[]),
1451 
            "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"
1452 
        );
1453 
        let archs_known: std::collections::HashMap<String, String> = [(rootfs.clone(), "x86_64".to_string())].into();
1454 
        assert_eq!(
1455 
            why_waiting(&job, "andy", 1, &[riscv], &archs_known, &[], now),
1456 
            "no runner can take it: narvi is riscv64; the job needs x86_64"
1457 
        );
1458 
        let without = Runner { images: "[]".into(), ..runner(6, "new") };
1459 
        assert_eq!(why(&[without], &[]), "no runner can take it: new does not hold image abababababab");
1460 
        let d = fdb::DeclineRow {
1461 
            job_id: 7,
1462 
            runner_id: 1,
1463 
            runner: "earth-1".into(),
1464 
            reason: "only 512 MiB of memory is free here".into(),
1465 
            ms: now,
1466 
        };
1467 
        assert_eq!(
1468 
            why(&[runner(1, "earth-1")], std::slice::from_ref(&d)),
1469 
            "no runner can take it: earth-1 declined it: only 512 MiB of memory is free here"
1470 
        );
1471 
        let old = fdb::DeclineRow { ms: now - fdb::DECLINE_HOLDS_MS - 1, ..d };
1472 
        assert_eq!(why(&[runner(1, "earth-1")], &[old]), "", "an old decline no longer holds");
1473 
    }
1474 
}