OXYGENHUB
# 4ac01a732552 oxygenhub / crates / hub / src / flow.rs
48.1 KiB · Rust · last changed Secrets for pipelines, mounted as files 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
    let room = fdb::Room {
465
        memory_mib: req.free_memory_mib,
466
        cpus: req.free_cpus,
467
        images: req.images.as_ref().map(|i| i.iter().map(|h| h.to_ascii_lowercase()).collect()),
468
    };
469
    // The labels it reports now, or those it registered with.
470
    let labels: Vec<String> = match &req.info {
471
        Some(i) => i.labels.clone(),
472
        None => me.label_list().into_iter().map(String::from).collect(),
473
    };
474
    let Some(job) =
475
        fdb::claim_within(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope(), &labels).await?
476
    else {
477
        return Ok(StatusCode::NO_CONTENT.into_response());
478
    };
479
    db::runners::set_job_runner(&app.db, job.id, me.id).await?;
480
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
481
    let row =
482
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
483
    report_check(&app, &row, &run.state, run.number, &job.name, "running", &format!("running on {runner}")).await;
484
    let bad = |e: serde_json::Error| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}"));
485
    // The job's limit: its own timeout, else its sandbox's, else the site
486
    // default, never past the site maximum; recorded for the reaper.
487
    let limits = limits(&app).await;
488
    let spec: flow_core::pipeline::Job = serde_json::from_str(&job.spec).map_err(bad)?;
489
    let sandbox: flow_core::pipeline::Sandbox = serde_json::from_str(&job.sandbox).map_err(bad)?;
490
    let timeout_s = spec.timeout_s.or(sandbox.timeout_s).unwrap_or(limits.default_s).min(limits.max_s).max(60);
491
    fdb::set_job_timeout(&app.db, job.id, timeout_s).await?;
492
    let claim = Claim {
493
        job_id: job.id,
494
        run_id: run.id,
495
        repo: format!("{}/{}", row.owner, row.name),
496
        clone_url: format!("{}/{}/{}", app.config.base_url(), row.owner, row.name),
497
        token: secret,
498
        heads: run.head_list().into_iter().map(String::from).collect(),
499
        pipeline_hash: run.pipeline_hash.clone(),
500
        job: spec,
501
        sandbox,
502
        timeout_s,
503
        subrepos: vec![],
504
        state: run.state.clone(),
505
        bookmark: run.bookmark.clone(),
506
        run_number: run.number,
507
        // The site's, the owner's and the repository's variables (#102).
508
        env: crate::variables::env_for_repo(&app.db, &row).await?,
509
        secrets: vec![],
510
    };
511
    let mut claim = claim;
512
    let (inputs, ids, mut left_out) = subrepo_inputs(&app, &row, &run).await;
513
    // The secrets the job declares, as files (#107); one not set is said
514
    // at the top of the log and left out, so the job fails on the file.
515
    let names: Vec<String> = claim.job.secrets.iter().map(|s| s.name.clone()).collect();
516
    let found = crate::secrets::for_job(&app, &row, &names).await?;
517
    for s in &claim.job.secrets {
518
        match found.iter().find(|(n, _)| *n == s.name) {
519
            Some((name, value)) => {
520
                claim.secrets.push(flow_core::api::SecretInput { name: name.clone(), path: s.mount_path.clone(), value: value.clone() })
521
            }
522
            None => left_out.push(format!(
523
                "secret {:?} is not set for this repository, its owner or the site; {} will be missing",
524
                s.name, s.mount_path
525
            )),
526
        }
527
    }
528
    if !ids.is_empty() {
529
        fdb::set_also_read(&app.db, job.id, &ids).await?;
530
    }
531
    // Say at the top of the job's log what it will not have (oxygenhub#56),
532
    // rather than let it fail later on missing files.
533
    if !left_out.is_empty() {
534
        let records: Vec<flow_core::api::Record> =
535
            left_out.into_iter().map(|text| flow_core::api::Record { t: 0, stream: "flow".into(), text }).collect();
536
        fdb::append_log(&app.db, job.id, &records, MAX_RECORDS as i64).await?;
537
    }
538
    claim.subrepos = inputs;
539
    Ok(Json(claim).into_response())
540
}
541
542
/// The Oxygen subrepo pins in `name`'s state `heads`: (path, repository id, pinned heads).
543
async fn pins_in(
544
    app: &Shared,
545
    name: crate::repos::RepoName,
546
    heads: Vec<oxy_core::hash::ChangeHash>,
547
) -> Vec<(String, [u8; 16], Vec<oxy_core::hash::ChangeHash>)> {
548
    app.repos
549
        .read(&name, move |repo, cache| {
550
            let loaded = cache.load(repo, &heads)?;
551
            let mut out = Vec::new();
552
            for (path, f) in &loaded.snapshot.files {
553
                if f.kind != oxy_core::objects::NodeKind::Subrepo {
554
                    continue;
555
                }
556
                if let oxy_core::state::FileContent::Opaque { blobs } = &f.content {
557
                    if let Some(b) = blobs.first() {
558
                        if let Ok(oxy_core::subrepo::Pin::Oxy { repo: id, heads, .. }) = repo
559
                            .read_blob(b)
560
                            .map_err(crate::repos::RepoError::Oxy)
561
                            .and_then(|bytes| oxy_core::subrepo::Pin::decode(&bytes).map_err(|e| crate::repos::RepoError::Oxy(e.into())))
562
                        {
563
                            out.push((path.clone(), id, heads));
564
                        }
565
                    }
566
                }
567
            }
568
            Ok(out)
569
        })
570
        .await
571
        .unwrap_or_default()
572
}
573
574
/// How deep subrepos of subrepos are followed.
575
const MAX_SUBREPO_DEPTH: usize = 3;
576
577
/// The Oxygen subrepos the run's state pins, and those their states pin in
578
/// turn (parents first, paths joined), that live on this hub and that the
579
/// run's pusher may read: what the job's workspace needs beside its own
580
/// tree. A pin the hub cannot serve is left out, with what it pins, and
581
/// the third list says why, for the job's log.
582
async fn subrepo_inputs(
583
    app: &Shared,
584
    row: &db::repos::RepoRow,
585
    run: &fdb::Run,
586
) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>, Vec<String>) {
587
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![], vec![]) };
588
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
589
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
590
    let rows = db::repos::list(&app.db).await.unwrap_or_default();
591
    let mut inputs = Vec::new();
592
    let mut ids = Vec::new();
593
    let mut left_out = Vec::new();
594
    let actor = if run.actor.is_empty() { "the run's starter".to_string() } else { run.actor.clone() };
595
    // (path prefix, repository, heads, depth), parents before children.
596
    let mut queue = std::collections::VecDeque::from([(String::new(), name, heads, 0usize)]);
597
    while let Some((prefix, name, heads, depth)) = queue.pop_front() {
598
        if depth == MAX_SUBREPO_DEPTH {
599
            continue;
600
        }
601
        for (path, id, pinned) in pins_in(app, name, heads).await {
602
            let path = if prefix.is_empty() { path } else { format!("{prefix}/{path}") };
603
            let mut hosted = false;
604
            for other in &rows {
605
                let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };
606
                let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);
607
                if !matches {
608
                    continue;
609
                }
610
                hosted = true;
611
                let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();
612
                if crate::auth::roles::allows(role, crate::auth::Access::Read) {
613
                    inputs.push(flow_core::api::SubrepoInput {
614
                        path: path.clone(),
615
                        repo: format!("{}/{}", other.owner, other.name),
616
                        clone_url: format!("{}/{}/{}", app.config.base_url(), other.owner, other.name),
617
                        heads: pinned.iter().map(|h| h.to_hex()).collect(),
618
                    });
619
                    if !ids.contains(&other.id) {
620
                        ids.push(other.id);
621
                    }
622
                    queue.push_back((path.clone(), other_name, pinned.clone(), depth + 1));
623
                } else {
624
                    left_out.push(format!(
625
                        "subrepo {path} ({}/{}) is left out: {actor} cannot read it, so the job has no files there",
626
                        other.owner, other.name
627
                    ));
628
                }
629
                break;
630
            }
631
            if !hosted {
632
                left_out.push(format!("subrepo {path} is left out: the repository it pins is not on this hub"));
633
            }
634
        }
635
    }
636
    (inputs, ids, left_out)
637
}
638
639
async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {
640
    match fdb::job(&app.db, id).await? {
641
        Some(j) if j.status == "running" => Ok(j),
642
        Some(_) => Err(ApiError(StatusCode::CONFLICT, "this job is not running".into())),
643
        None => Err(ApiError(StatusCode::NOT_FOUND, "no such job".into())),
644
    }
645
}
646
647
/// `POST /api/v1/flow/jobs/{id}/log`.
648
pub async fn log(
649
    State(app): State<Shared>,
650
    headers: HeaderMap,
651
    Path(id): Path<i64>,
652
    Json(mut batch): Json<LogBatch>,
653
) -> Result<StatusCode, ApiError> {
654
    runner_auth(&app, &headers).await?;
655
    let job = running_job(&app, id).await?;
656
    // An empty batch asks only whether the job still runs: a cancelled
657
    // one answers 409 and its runner stops it (oxygenflow#32).
658
    if batch.records.is_empty() {
659
        return Ok(StatusCode::NO_CONTENT);
660
    }
661
    // A secret's value never reaches the stored log (#107).
662
    let values: Vec<String> = match serde_json::from_str::<flow_core::Job>(&job.spec) {
663
        Ok(spec) if !spec.secrets.is_empty() => {
664
            let run = fdb::run_by_id(&app.db, job.run_id).await?;
665
            let row = match run {
666
                Some(run) => db::repos::get_by_id(&app.db, run.repo_id).await?,
667
                None => None,
668
            };
669
            match row {
670
                Some(row) => {
671
                    let names: Vec<String> = spec.secrets.iter().map(|s| s.name.clone()).collect();
672
                    crate::secrets::for_job(&app, &row, &names).await?.into_iter().map(|(_, v)| v).collect()
673
                }
674
                None => vec![],
675
            }
676
        }
677
        _ => vec![],
678
    };
679
    for r in &mut batch.records {
680
        if !values.is_empty() {
681
            r.text = crate::secrets::mask(&r.text, &values);
682
        }
683
        if r.text.len() > MAX_RECORD_TEXT {
684
            let mut cut = MAX_RECORD_TEXT;
685
            while !r.text.is_char_boundary(cut) {
686
                cut -= 1;
687
            }
688
            r.text.truncate(cut);
689
        }
690
        if !matches!(r.stream.as_str(), "out" | "err" | "console" | "flow") {
691
            r.stream = "console".into();
692
        }
693
    }
694
    let kept = fdb::append_log(&app.db, job.id, &batch.records, MAX_RECORDS as i64).await?;
695
    if kept < batch.records.len() && job.log_count < MAX_RECORDS as i64 {
696
        tracing::info!("flow: job {} reached the log limit", job.id);
697
    }
698
    Ok(StatusCode::NO_CONTENT)
699
}
700
701
/// Largest artifact the hub keeps; the runner refuses larger ones first.
702
pub const MAX_ARTIFACT_BYTES: usize = 512 << 20;
703
704
/// `POST /api/v1/flow/jobs/{id}/artifacts/{name}`: the bytes of an artifact
705
/// the job declares (OxygenFlow spec §5.2), kept as a blob of the run's
706
/// repository. Answers `{"blob"}`.
707
pub async fn artifact(
708
    State(app): State<Shared>,
709
    headers: HeaderMap,
710
    Path((id, name)): Path<(i64, String)>,
711
    body: axum::body::Bytes,
712
) -> Result<Response, ApiError> {
713
    runner_auth(&app, &headers).await?;
714
    let job = running_job(&app, id).await?;
715
    let spec: flow_core::Job =
716
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
717
    let Some(out) = spec.outputs.iter().find(|o| o.name == name) else {
718
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no artifact {name:?}", job.name)));
719
    };
720
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
721
    let row =
722
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
723
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
724
    let content = blake3::hash(&body).to_hex().to_string();
725
    let size = body.len() as i64;
726
    let blob = app
727
        .repos
728
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
729
        .await
730
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the artifact: {e}")))?;
731
    let a = fdb::Artifact { job_id: job.id, name, path: out.path.clone(), blob: blob.to_hex(), size, content, created_ms: db::now_ms() };
732
    fdb::add_artifact(&app.db, &a).await?;
733
    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())
734
}
735
736
/// `POST /api/v1/flow/jobs/{id}/finish`.
737
pub async fn finish(
738
    State(app): State<Shared>,
739
    headers: HeaderMap,
740
    Path(id): Path<i64>,
741
    Json(f): Json<Finish>,
742
) -> Result<StatusCode, ApiError> {
743
    let me = runner_auth(&app, &headers).await?;
744
    let job = running_job(&app, id).await?;
745
    let status = match (f.exit_code, &f.error) {
746
        (Some(0), None) => "passed",
747
        (Some(_), _) => "failed",
748
        (None, _) => "error",
749
    };
750
    let (receipt, receipt_hash) = match &f.receipt {
751
        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {
752
            // The receipt names exactly the artifacts uploaded, in name order.
753
            let kept: Vec<String> = fdb::artifacts(&app.db, job.id).await?.into_iter().map(|a| a.blob).collect();
754
            if r.outputs.iter().map(hex::encode).collect::<Vec<_>>() != kept {
755
                return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt names other outputs than the artifacts uploaded".into()));
756
            }
757
            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))
758
        }
759
        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),
760
        None => (String::new(), String::new()),
761
    };
762
    let error = f.error.clone().unwrap_or_default();
763
    let changed = fdb::finish(
764
        &app.db,
765
        job.id,
766
        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },
767
    )
768
    .await?;
769
    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
770
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
771
            if let (Some(r), Some(signed)) = (&f.receipt, &f.signed_receipt) {
772
                let (attest, signer, note) = match keep_receipt(&app, &me, &row, &run, r, signed).await {
773
                    Ok((attest, signer)) => (attest, signer, String::new()),
774
                    Err(e) => {
775
                        tracing::warn!("flow: job {} receipt not kept: {e}", job.id);
776
                        (String::new(), String::new(), e)
777
                    }
778
                };
779
                fdb::set_receipt(&app.db, job.id, &attest, &signer, &note).await?;
780
            }
781
        }
782
    }
783
    report_changed(&app, job.run_id, changed).await?;
784
    Ok(StatusCode::NO_CONTENT)
785
}
786
787
/// Report the checks of jobs whose status changed.
788
async fn report_changed(app: &Shared, run_id: i64, changed: Vec<fdb::Job>) -> sqlx::Result<()> {
789
    let Some(run) = fdb::run_by_id(&app.db, run_id).await? else {
790
        return Ok(());
791
    };
792
    let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? else {
793
        return Ok(());
794
    };
795
    for j in changed {
796
        let summary = match j.status.as_str() {
797
            "passed" => format!("passed in {}", duration(&j)),
798
            "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),
799
            "skipped" => "skipped: a dependency did not pass".into(),
800
            "cancelled" => j.error.clone(),
801
            _ => format!("could not run: {}", j.error),
802
        };
803
        report_check(app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;
804
    }
805
    Ok(())
806
}
807
808
/// The site's job time limits (oxygenhub#89): set on Admin → Runners →
809
/// Settings, else from `hub.toml`.
810
pub struct Limits {
811
    /// For a job whose pipeline names none.
812
    pub default_s: u64,
813
    /// The most a pipeline may ask for.
814
    pub max_s: u64,
815
    /// Grace past a job's limit before the hub gives it up: a runner stops
816
    /// a job at its limit and reports it within moments, so one still
817
    /// running this long after is one the runner no longer reports (it was
818
    /// restarted, or lost the hub while the job ended, oxygenhub#71).
819
    pub grace_ms: i64,
820
}
821
822
pub const SETTING_DEFAULT: &str = "flow.job_timeout_minutes";
823
pub const SETTING_MAX: &str = "flow.job_timeout_max_minutes";
824
pub const SETTING_GRACE: &str = "flow.lost_grace_minutes";
825
826
pub async fn limits(app: &Shared) -> Limits {
827
    let minutes = |key: &'static str| async move {
828
        db::accounts::setting(&app.db, key).await.ok().flatten().and_then(|v| v.parse::<u64>().ok()).filter(|m| *m > 0)
829
    };
830
    let default_m = minutes(SETTING_DEFAULT).await.unwrap_or(app.config.flow.job_timeout_minutes.max(1));
831
    let max_m = minutes(SETTING_MAX).await.unwrap_or(default_m.max(6 * 60)).max(default_m);
832
    let grace_m = minutes(SETTING_GRACE).await.unwrap_or(5);
833
    Limits { default_s: default_m * 60, max_s: max_m * 60, grace_ms: grace_m as i64 * 60_000 }
834
}
835
836
/// End every job still running past its time limit plus the grace:
837
/// it is marked an error with the reason, the jobs that depended on it are
838
/// skipped and its run settled, as when a runner reports an error.
839
pub async fn reap_lost_jobs(app: &Shared) -> sqlx::Result<usize> {
840
    let limits = limits(app).await;
841
    let lost = fdb::running_past_limit(&app.db, db::now_ms(), limits.default_s, limits.grace_ms).await?;
842
    let n = lost.len();
843
    for job in lost {
844
        let error = format!(
845
            "lost: still running {} minutes after it started, past the time limit; the runner{} stopped reporting it",
846
            (db::now_ms() - job.started_ms.unwrap_or(0)) / 60_000,
847
            if job.runner.is_empty() { String::new() } else { format!(" {}", job.runner) },
848
        );
849
        let changed =
850
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
851
                .await?;
852
        tracing::warn!("flow: job {} {error}", job.id);
853
        report_changed(app, job.run_id, changed).await?;
854
    }
855
    Ok(n)
856
}
857
858
/// End the jobs running on `runner` that it does not hold (oxygenhub#71):
859
/// it restarted since claiming them, or never received the claim. A
860
/// runner claims one job at a time, so a job it holds is always listed.
861
async fn end_jobs_not_held(app: &Shared, runner: &db::runners::Runner, holding: &[i64]) -> sqlx::Result<()> {
862
    for job in fdb::running_on(&app.db, runner.id).await? {
863
        if holding.contains(&job.id) {
864
            continue;
865
        }
866
        let error = format!("lost: the runner {} no longer holds this job (it restarted, or never received it)", runner.name);
867
        let changed =
868
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
869
                .await?;
870
        tracing::warn!("flow: job {} {error}", job.id);
871
        report_changed(app, job.run_id, changed).await?;
872
    }
873
    Ok(())
874
}
875
876
/// Start the scheduled runs due at `when` (oxygenflow#24): on each
877
/// repository that runs Flow, the jobs of its default bookmark's pipeline
878
/// with a `schedule` line due then. Returns how many runs started.
879
pub async fn run_due(app: &Shared, when: flow_core::pipeline::When) -> usize {
880
    let Ok(rows) = db::repos::list(&app.db).await else { return 0 };
881
    let mut started = 0;
882
    for row in rows.into_iter().filter(|r| enabled(app, r)) {
883
        let Some(repo) = RepoName::new(&row.owner, &row.name) else { continue };
884
        let b = row.default_bookmark.clone();
885
        let found = app
886
            .repos
887
            .read(&repo, move |r, _| {
888
                let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
889
                Ok(Some((st, r.store().get_state(&st)?.heads)))
890
            })
891
            .await;
892
        let Ok(Some((state, heads))) = found else { continue };
893
        if heads.is_empty() {
894
            continue;
895
        }
896
        match start_run_at(app, &row, &row.default_bookmark, &state.to_hex(), &heads, &row.owner, "schedule", None, Some(when)).await {
897
            Ok(Some(_)) => started += 1,
898
            Ok(None) => {}
899
            Err(e) => tracing::warn!("flow: scheduled run of {}/{}: {e}", row.owner, row.name),
900
        }
901
    }
902
    started
903
}
904
905
/// The hub's clock as a [`When`](flow_core::pipeline::When).
906
pub fn when_now() -> flow_core::pipeline::When {
907
    let now = time::OffsetDateTime::now_utc();
908
    flow_core::pipeline::When {
909
        minute: now.minute() as u32,
910
        hour: now.hour() as u32,
911
        day: now.day() as u32,
912
        month: u8::from(now.month()) as u32,
913
        weekday: now.weekday().number_days_from_sunday() as u32,
914
    }
915
}
916
917
/// Start the scheduled runs due each minute, once a minute.
918
pub async fn scheduler(app: Shared) {
919
    let mut last = None;
920
    loop {
921
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
922
        let minute = time::OffsetDateTime::now_utc().unix_timestamp() / 60;
923
        if last == Some(minute) {
924
            continue;
925
        }
926
        last = Some(minute);
927
        let n = run_due(&app, when_now()).await;
928
        if n > 0 {
929
            tracing::info!("flow: {n} scheduled run(s) started");
930
        }
931
    }
932
}
933
934
/// Look for lost jobs once a minute.
935
pub async fn reaper(app: Shared) {
936
    loop {
937
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
938
        if let Err(e) = reap_lost_jobs(&app).await {
939
            tracing::warn!("flow: looking for lost jobs: {e}");
940
        }
941
    }
942
}
943
944
/// Check a runner's signed receipt for a job of `run` and keep it in the
945
/// repository (Oxygen spec §14.6): the attestation's hex and the runner's
946
/// name, or why it was not kept. It must be the job's receipt, for the
947
/// run's state and repository, signed by the key the runner's token first
948
/// signed with.
949
pub async fn keep_receipt(
950
    app: &Shared,
951
    runner: &db::runners::Runner,
952
    row: &db::repos::RepoRow,
953
    run: &fdb::Run,
954
    receipt: &flow_core::Receipt,
955
    signed_hex: &str,
956
) -> Result<(String, String), String> {
957
    use oxy_core::receipt::SignedReceipt;
958
    let bytes = hex::decode(signed_hex).map_err(|_| "the signed receipt is not hex".to_string())?;
959
    let sr = SignedReceipt::decode(&bytes).map_err(|e| format!("not a signed receipt: {e}"))?;
960
    sr.verify().map_err(|e| format!("it does not verify: {e}"))?;
961
    if sr.receipt.hash().0 != receipt.hash() {
962
        return Err("it signs another receipt than the job's".into());
963
    }
964
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
965
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(heads).encode());
966
    if sr.state() != state {
967
        return Err("it is for another state than the run's".into());
968
    }
969
    let key = sr.runner.hash().to_hex();
970
    let bound = db::runners::bind_key(&app.db, runner.id, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;
971
    if bound != key {
972
        return Err(format!("signed by {}, not the key this runner signs with ({})", &key[..12], &bound[..12.min(bound.len())]));
973
    }
974
    let name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
975
    let signer = sr.runner.name.clone();
976
    let attest = app.repos.write(&name, move |repo, _| Ok(repo.add_receipt(&sr)?)).await.map_err(|e| e.to_string())?;
977
    Ok((attest.to_hex(), signer))
978
}
979
980
/// A job's wall time, as the pages show it.
981
pub fn duration(j: &fdb::Job) -> String {
982
    match (j.started_ms, j.finished_ms) {
983
        (Some(a), Some(b)) => human_ms(b - a),
984
        (Some(a), None) => human_ms(db::now_ms() - a),
985
        _ => "–".into(),
986
    }
987
}
988
989
pub fn human_ms(ms: i64) -> String {
990
    let ms = ms.max(0);
991
    if ms < 1000 {
992
        format!("{ms} ms")
993
    } else if ms < 60_000 {
994
        format!("{:.1} s", ms as f64 / 1000.0)
995
    } else {
996
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
997
    }
998
}
999
1000
/// Where a live run stands among the runners' work: what its viewers may
1001
/// know of it. Jobs of other repositories are named only when those are
1002
/// public; the rest are counted.
1003
#[derive(Clone, Debug, Default, PartialEq, Eq)]
1004
pub struct QueueInfo {
1005
    /// 1-based place of the run's first queued job among all queued jobs.
1006
    pub position: Option<usize>,
1007
    /// Jobs running now, and queued jobs ahead of the run's first one.
1008
    pub running: Vec<String>,
1009
    pub ahead: Vec<String>,
1010
    /// Of those, jobs in repositories the page may not name.
1011
    pub hidden: usize,
1012
}
1013
1014
impl QueueInfo {
1015
    /// "3rd in line" and the like.
1016
    pub fn place(&self) -> Option<String> {
1017
        self.position.map(|n| {
1018
            let suffix = match (n % 10, n % 100) {
1019
                (1, 11) | (2, 12) | (3, 13) => "th",
1020
                (1, _) => "st",
1021
                (2, _) => "nd",
1022
                (3, _) => "rd",
1023
                _ => "th",
1024
            };
1025
            format!("{n}{suffix} in line")
1026
        })
1027
    }
1028
}
1029
1030
/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.
1031
pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {
1032
    let mut info = QueueInfo::default();
1033
    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);
1034
    info.position = first.map(|i| i + 1);
1035
    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);
1036
    let mut queued_seen = 0;
1037
    for e in queue {
1038
        if e.run_id == run_id {
1039
            continue;
1040
        }
1041
        let ahead = match e.status.as_str() {
1042
            "running" => true,
1043
            _ => {
1044
                queued_seen += 1;
1045
                first.is_some_and(|f| queued_seen <= f)
1046
            }
1047
        };
1048
        if !ahead {
1049
            continue;
1050
        }
1051
        if e.repo_id != repo_id && e.visibility != "public" {
1052
            info.hidden += 1;
1053
        } else if e.status == "running" {
1054
            info.running.push(label(e));
1055
        } else {
1056
            info.ahead.push(label(e));
1057
        }
1058
    }
1059
    info
1060
}
1061
1062
#[cfg(test)]
1063
mod queue_tests {
1064
    use super::*;
1065
1066
    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {
1067
        fdb::QueueEntry {
1068
            job_id,
1069
            run_id,
1070
            repo_id,
1071
            owner: "o".into(),
1072
            name: format!("r{repo_id}"),
1073
            visibility: vis.into(),
1074
            run_number: run_id,
1075
            job: job.into(),
1076
            status: status.into(),
1077
            runner: String::new(),
1078
            spec: String::new(),
1079
            sandbox: String::new(),
1080
        }
1081
    }
1082
1083
    #[test]
1084
    fn a_run_learns_its_place_and_what_is_ahead_of_it() {
1085
        // Running first, then queued in claim order (as `fdb::queue` returns them).
1086
        let q = vec![
1087
            e(1, 10, 1, "public", "build", "running"),
1088
            e(2, 11, 2, "private", "test", "running"),
1089
            e(3, 10, 1, "public", "test", "queued"),
1090
            e(4, 12, 3, "private", "lint", "queued"),
1091
            e(5, 13, 4, "public", "build", "queued"),
1092
            e(6, 13, 4, "public", "test", "queued"),
1093
            e(7, 14, 5, "public", "late", "queued"),
1094
        ];
1095
        let info = queue_info(&q, 4, 13);
1096
        assert_eq!(info.position, Some(3));
1097
        assert_eq!(info.place().as_deref(), Some("3rd in line"));
1098
        assert_eq!(info.running, ["o/r1 #10 build"]);
1099
        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");
1100
        assert_eq!(info.hidden, 2, "private repositories are counted, not named");
1101
        // A private repository's own jobs are named to its viewers.
1102
        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);
1103
        assert_eq!(queue_info(&q, 3, 12).position, Some(2));
1104
        // A run with nothing queued has no place.
1105
        assert_eq!(queue_info(&q, 1, 99).position, None);
1106
        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();
1107
        assert_eq!(
1108
            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],
1109
            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]
1110
        );
1111
    }
1112
}