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