OXYGENHUB

A queued run says where it stands bf3d07541368

The Flow tab shows a waiting run's place in line ("3rd in line"), its
page also the jobs running now and those queued ahead of it, and the API
queue_position. Other repositories' jobs are named only when public and
otherwise counted, so a private repository's name never shows.

closes #28
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-09-29 02:38 UTC
Signature
✔ signed by Andy McCall
Hash
bf3d07541368f292b04d0d7532dfd263174396eaad73bb62cd9c76252c558e1e
Depends on
1edac4b68c6d 42fa95540468 9008bf6c2b8c 90326c9601f9 93be9ce4653e a0d4bbfbedd1 d23464c0ac60
Depended on by
367f34b8e590 107eaba52c9d 55673c650986 e9b9619cd857 417184c73f4e 4d2e44d79b03 efd29a0033e8 745d225de344 c4c87f481d05 b66e70b7b6c7 4f6925337848 10708ab8f54d
In bookmarks
api-unknown-fields cancel-landed-runs clear-waiting close-proposal-140 diffusion-pin-bookmark draft-ready image-check job-waits main oh-comment-body-file proposal-list-fast run-one-job wiki-says-so
In tags
v0.1.0 v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.4.1 v0.4.2 v0.5.0 v0.5.1 v0.5.2 v0.6.0 v0.6.1 v0.7.0 v0.8.0 v0.9.0 v0.9.1 v0.9.11 v0.9.15 v0.9.16 v0.9.19 v0.9.2 v0.9.20 v0.9.21 v0.9.22 v0.9.23 v0.9.24 v0.9.3 v0.9.4 v0.9.5 v0.9.6 v0.9.7 v0.9.8 v0.9.9
Edits
21 atoms, 0 tree operations, 9 files

Browse the tree at this change

modified crates/hub-api/src/lib.rs +4 −0
…
448 unchanged lines
449 449
    pub created_ms: i64,
450 450
    pub finished_ms: Option<i64>,
451 451
    pub url: String,
452 +
    /// For a run waiting for a runner, its first queued job's 1-based
453 +
    /// place among all queued jobs.
454 +
    #[serde(default, skip_serializing_if = "Option::is_none")]
455 +
    pub queue_position: Option<i64>,
452 456
    /// Filled by the single-run endpoint.
453 457
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
454 458
    pub jobs: Vec<FlowJob>,
…
428 unchanged lines
modified crates/hub/src/api/mod.rs +12 −3
…
1434 unchanged lines
1435 1435
    Ok(Json(RequiredChecks { names }))
1436 1436
}
1437 1437
1438 -
fn flow_run_of(app: &Shared, row: &db::repos::RepoRow, r: &db::flow::Run, jobs: Vec<db::flow::Job>) -> FlowRun {
1438 +
fn flow_run_of(
1439 +
    app: &Shared,
1440 +
    row: &db::repos::RepoRow,
1441 +
    r: &db::flow::Run,
1442 +
    jobs: Vec<db::flow::Job>,
1443 +
    queue: &[db::flow::QueueEntry],
1444 +
) -> FlowRun {
1439 1445
    FlowRun {
1446 +
        queue_position: crate::flow::queue_info(queue, row.id, r.id).position.map(|p| p as i64),
1440 1447
        number: r.number,
1441 1448
        bookmark: r.bookmark.clone(),
1442 1449
        state: r.state.clone(),
…
41 unchanged lines
1484 1491
    let (_, row, _) = reach(&app, &c, &owner, &name, Access::Read).await?;
1485 1492
    let limit = q.limit.unwrap_or(30).clamp(1, 100);
1486 1493
    let runs = db::flow::runs_by_number(&app.db, row.id, limit, q.before).await?;
1487 -
    Ok(Json(runs.iter().map(|r| flow_run_of(&app, &row, r, vec![])).collect()))
1494 +
    let queue = if runs.iter().any(|r| r.live()) { db::flow::queue(&app.db).await? } else { vec![] };
1495 +
    Ok(Json(runs.iter().map(|r| flow_run_of(&app, &row, r, vec![], &queue)).collect()))
1488 1496
}
1489 1497
1490 1498
/// `GET /repos/{o}/{n}/flow/runs/{number}`: a run with its jobs.
…
6 unchanged lines
1497 1505
    let (_, row, _) = reach(&app, &c, &owner, &name, Access::Read).await?;
1498 1506
    let run = db::flow::run(&app.db, row.id, number).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
1499 1507
    let jobs = db::flow::jobs(&app.db, run.id).await?;
1500 -
    Ok(Json(flow_run_of(&app, &row, &run, jobs)))
1508 +
    let queue = if run.live() { db::flow::queue(&app.db).await? } else { vec![] };
1509 +
    Ok(Json(flow_run_of(&app, &row, &run, jobs, &queue)))
1501 1510
}
1502 1511
1503 1512
#[derive(Deserialize)]
…
135 unchanged lines
modified crates/hub/src/db/flow.rs +27 −0
…
147 unchanged lines
148 148
    Ok(id)
149 149
}
150 150
151 +
/// A job waiting for or held by a runner, with where it comes from.
152 +
#[derive(Clone, Debug, FromRow)]
153 +
pub struct QueueEntry {
154 +
    pub job_id: i64,
155 +
    pub run_id: i64,
156 +
    pub repo_id: i64,
157 +
    pub owner: String,
158 +
    pub name: String,
159 +
    pub visibility: String,
160 +
    pub run_number: i64,
161 +
    pub job: String,
162 +
    pub status: String,
163 +
}
164 +
165 +
/// Every job of a live run that is running or queued: running ones first,
166 +
/// then queued ones in the order runners claim them (hub spec §13.2).
167 +
pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {
168 +
    sqlx::query_as(
169 +
        "SELECT j.id AS job_id, j.run_id, r.repo_id, p.owner, p.name, p.visibility, r.number AS run_number, j.name AS job, j.status \
170 +
         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id \
171 +
         WHERE j.status IN ('queued', 'running') AND r.status IN ('queued', 'running') \
172 +
         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, j.run_id, j.depth, j.id",
173 +
    )
174 +
    .fetch_all(pool)
175 +
    .await
176 +
}
177 +
151 178
/// A repository's runs, newest first, numbered below `before`.
152 179
pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {
153 180
    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number < ? ORDER BY number DESC LIMIT ?")
…
241 unchanged lines
modified crates/hub/src/flow.rs +111 −0
…
407 unchanged lines
408 408
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
409 409
    }
410 410
}
411 +
412 +
/// Where a live run stands among the runners' work: what its viewers may
413 +
/// know of it. Jobs of other repositories are named only when those are
414 +
/// public; the rest are counted.
415 +
#[derive(Clone, Debug, Default, PartialEq, Eq)]
416 +
pub struct QueueInfo {
417 +
    /// 1-based place of the run's first queued job among all queued jobs.
418 +
    pub position: Option<usize>,
419 +
    /// Jobs running now, and queued jobs ahead of the run's first one.
420 +
    pub running: Vec<String>,
421 +
    pub ahead: Vec<String>,
422 +
    /// Of those, jobs in repositories the page may not name.
423 +
    pub hidden: usize,
424 +
}
425 +
426 +
impl QueueInfo {
427 +
    /// "3rd in line" and the like.
428 +
    pub fn place(&self) -> Option<String> {
429 +
        self.position.map(|n| {
430 +
            let suffix = match (n % 10, n % 100) {
431 +
                (1, 11) | (2, 12) | (3, 13) => "th",
432 +
                (1, _) => "st",
433 +
                (2, _) => "nd",
434 +
                (3, _) => "rd",
435 +
                _ => "th",
436 +
            };
437 +
            format!("{n}{suffix} in line")
438 +
        })
439 +
    }
440 +
}
441 +
442 +
/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.
443 +
pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {
444 +
    let mut info = QueueInfo::default();
445 +
    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);
446 +
    info.position = first.map(|i| i + 1);
447 +
    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);
448 +
    let mut queued_seen = 0;
449 +
    for e in queue {
450 +
        if e.run_id == run_id {
451 +
            continue;
452 +
        }
453 +
        let ahead = match e.status.as_str() {
454 +
            "running" => true,
455 +
            _ => {
456 +
                queued_seen += 1;
457 +
                first.is_some_and(|f| queued_seen <= f)
458 +
            }
459 +
        };
460 +
        if !ahead {
461 +
            continue;
462 +
        }
463 +
        if e.repo_id != repo_id && e.visibility != "public" {
464 +
            info.hidden += 1;
465 +
        } else if e.status == "running" {
466 +
            info.running.push(label(e));
467 +
        } else {
468 +
            info.ahead.push(label(e));
469 +
        }
470 +
    }
471 +
    info
472 +
}
473 +
474 +
#[cfg(test)]
475 +
mod queue_tests {
476 +
    use super::*;
477 +
478 +
    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {
479 +
        fdb::QueueEntry {
480 +
            job_id,
481 +
            run_id,
482 +
            repo_id,
483 +
            owner: "o".into(),
484 +
            name: format!("r{repo_id}"),
485 +
            visibility: vis.into(),
486 +
            run_number: run_id,
487 +
            job: job.into(),
488 +
            status: status.into(),
489 +
        }
490 +
    }
491 +
492 +
    #[test]
493 +
    fn a_run_learns_its_place_and_what_is_ahead_of_it() {
494 +
        // Running first, then queued in claim order (as `fdb::queue` returns them).
495 +
        let q = vec![
496 +
            e(1, 10, 1, "public", "build", "running"),
497 +
            e(2, 11, 2, "private", "test", "running"),
498 +
            e(3, 10, 1, "public", "test", "queued"),
499 +
            e(4, 12, 3, "private", "lint", "queued"),
500 +
            e(5, 13, 4, "public", "build", "queued"),
501 +
            e(6, 13, 4, "public", "test", "queued"),
502 +
            e(7, 14, 5, "public", "late", "queued"),
503 +
        ];
504 +
        let info = queue_info(&q, 4, 13);
505 +
        assert_eq!(info.position, Some(3));
506 +
        assert_eq!(info.place().as_deref(), Some("3rd in line"));
507 +
        assert_eq!(info.running, ["o/r1 #10 build"]);
508 +
        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");
509 +
        assert_eq!(info.hidden, 2, "private repositories are counted, not named");
510 +
        // A private repository's own jobs are named to its viewers.
511 +
        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);
512 +
        assert_eq!(queue_info(&q, 3, 12).position, Some(2));
513 +
        // A run with nothing queued has no place.
514 +
        assert_eq!(queue_info(&q, 1, 99).position, None);
515 +
        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();
516 +
        assert_eq!(
517 +
            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],
518 +
            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]
519 +
        );
520 +
    }
521 +
}
modified crates/hub/src/web/flow.rs +14 −2
…
15 unchanged lines
16 16
    pub r: fdb::Run,
17 17
    pub duration: String,
18 18
    pub jobs: Vec<(String, String)>,
19 +
    /// "3rd in line" for a run still waiting for a runner.
20 +
    pub place: Option<String>,
19 21
}
20 22
21 23
#[derive(Template)]
…
29 unchanged lines
51 53
    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;
52 54
    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;
53 55
    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);
56 +
    let queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] };
54 57
    let mut runs = Vec::new();
55 58
    for r in list.into_iter().take(PER_PAGE as usize) {
59 +
        let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None };
56 60
        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();
57 61
        let duration = match r.finished_ms {
58 62
            Some(f) => crate::flow::human_ms(f - r.created_ms),
59 63
            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),
60 64
            None => "–".into(),
61 65
        };
62 -
        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r });
66 +
        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r, place });
63 67
    }
64 68
    let enabled = crate::flow::enabled(&app, &row);
65 69
    let can_run = enabled && viewer.account().is_some() && role >= Role::Write;
…
179 unchanged lines
245 249
    duration: String,
246 250
    /// Whether the viewer may re-run it.
247 251
    can_rerun: bool,
252 +
    /// Where a live run stands among the runners' work.
253 +
    queue: Option<crate::flow::QueueInfo>,
248 254
}
249 255
250 256
/// Lines of a job's log a page shows before "the rest is loaded as it comes".
…
23 unchanged lines
274 280
    };
275 281
    let graph = layout(jobs);
276 282
    let can_rerun = !run.live() && viewer.account().is_some() && role >= Role::Write && crate::flow::enabled(&app, &row);
277 -
    Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration, can_rerun }))
283 +
    let queue = if run.live() {
284 +
        let q = fdb::queue(&app.db).await.unwrap_or_default();
285 +
        Some(crate::flow::queue_info(&q, row.id, run.id)).filter(|i| i.position.is_some())
286 +
    } else {
287 +
        None
288 +
    };
289 +
    Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration, can_rerun, queue }))
278 290
}
279 291
280 292
#[derive(Deserialize)]
…
174 unchanged lines
modified crates/hub/tests/flow.rs +9 −0
…
540 unchanged lines
541 541
    }
542 542
    let plain = runs_of(&h.app, "other", "plain").await;
543 543
    assert_eq!(plain.len(), 1, "turned on");
544 +
    // With andy/first's jobs ahead, plain's run says where it stands.
545 +
    let a = claim(&h.app).await.unwrap();
546 +
    assert_eq!(a["repo"], "andy/first");
547 +
    let (_, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;
548 +
    assert!(page.contains("Waiting for a runner: 3rd in line") && page.contains("Running now: andy/first #1"), "{page}");
549 +
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/other/plain/flow/runs/1", None, None).await;
550 +
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["queue_position"], 3);
551 +
    let (_, list) = call(&h.app, "GET", "/other/plain/flow", None, None).await;
552 +
    assert!(list.contains("queued, 3rd in line"), "{list}");
544 553
    // Runs are numbered per repository: this is plain's first, whatever
545 554
    // other repositories ran before.
546 555
    assert_eq!(plain[0].number, 1);
…
146 unchanged lines
modified docs/spec.md +1 −1
…
419 unchanged lines
420 420
421 421
- **Configuration.** `[flow] runner_tokens` (bearer tokens for runners, compared in constant time), `repos` (patterns such as `owner/*` or `owner/name`: the repositories whose pushes start runs by default; empty means none) and `job_timeout_minutes` (60).
422 422
- **Per repository.** A site administrator turns OxygenFlow on or off for one repository under Settings → OxygenFlow, which overrides `repos` in either direction; "site default" follows it again. Runs use the site's runners, so repository administrators see the setting but cannot change it. Changes are audited (`repo.flow`) and apply from the next push, with no restart.
423 -
- **Triggers.** A push that moves a bookmark of an allowed repository starts a **run** on the bookmark's new state, numbered within its repository from 1 (the run's page is `/{owner}/{name}/flow/{number}`): the hub reads `.flow/pipeline.kdl` from that state and validates it with `flow-core` (the same code `flow check` runs). An invalid pipeline records a run in `error` with the problems and their lines; a state without a pipeline records nothing. Tags and proposals start no runs yet, so only code from people with write access ever runs. Writers also start runs without moving a bookmark: **Run pipeline** on the Flow tab runs a bookmark's current state, and **Re-run** on a finished run's page runs its state again. The new run's checks replace the old ones for that state, and the run page says what started it (pushed by, started by, re-run of #N by).
423 +
- **Triggers.** A push that moves a bookmark of an allowed repository starts a **run** on the bookmark's new state, numbered within its repository from 1 (the run's page is `/{owner}/{name}/flow/{number}`): the hub reads `.flow/pipeline.kdl` from that state and validates it with `flow-core` (the same code `flow check` runs). An invalid pipeline records a run in `error` with the problems and their lines; a state without a pipeline records nothing. Tags and proposals start no runs yet, so only code from people with write access ever runs. A run waiting for a runner says where it stands: on the Flow tab its place in line, on its page also the jobs running now and those queued ahead of it (named when their repository is this one or public, otherwise counted), and in the API `queue_position`. Writers also start runs without moving a bookmark: **Run pipeline** on the Flow tab runs a bookmark's current state, and **Re-run** on a finished run's page runs its state again. The new run's checks replace the old ones for that state, and the run page says what started it (pushed by, started by, re-run of #N by).
424 424
- **Runs and jobs.** A run holds the repository, bookmark, state hash and heads, pipeline hash and pusher; one job per pipeline job with its compiled form, sandbox and dependencies. Job states: `queued`, `running`, `passed`, `failed`, `skipped` (a dependency did not pass), `error` (the runner could not run it). A run is `running` while any job is, then `passed` when every job passed, else `failed`.
425 425
- **Runner API** (runner token): `POST /api/v1/flow/claim` hands out the oldest queued job whose dependencies passed and whose tier the runner offers, marking it `running`, with a **job token**; `204` when there is none. `POST /api/v1/flow/jobs/{id}/log` appends log records (text cut at 16 KiB, at most 50 000 per job); `POST /api/v1/flow/jobs/{id}/finish` records the exit code and receipt (the hub recomputes the receipt hash) and settles dependents. Only the runner holding the job may report on it.
426 426
- **Job tokens.** Random, stored as a hash, valid only for reading the job's repository over the wire protocol while the job is running and for at most two hours. They cannot push, call the API or read other repositories.
…
158 unchanged lines
modified templates/flow_run.html +3 −0
…
9 unchanged lines
10 10
  <p class="small mute"><span class="lbl lbl-bm">{{ run.bookmark }}</span> at <a href="{{ repo.url }}/tree/{{ run.heads }}" title="state {{ run.state }}"><code>{{ run.state|short }}</code></a> · {{ run.started_by() }} <time title="{{ run.created_ms|when_i }}">{{ run.created_ms|ago_i }}</time> · {{ duration }}{% if !run.pipeline_hash.is_empty() %} · pipeline <code title="{{ run.pipeline_hash }}">{{ run.pipeline_hash|short }}</code>{% endif %} · <a href="{{ repo.url }}/file/{{ run.heads }}/.flow/pipeline.kdl">.flow/pipeline.kdl</a> · <a href="{{ repo.url }}/flow">all runs</a></p>
11 11
  {% if can_rerun %}<form class="inline" method="post" action="{{ repo.url }}/flow/{{ run.number }}/rerun"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" class="small">RE-RUN</button></form>{% endif %}
12 12
</div>
13 +
{% if let Some(q) = queue %}
14 +
<section class="panel"><div class="strip">Waiting for a runner: {{ q.place().unwrap_or_default() }}.{% if !q.running.is_empty() %} Running now: {{ q.running.join(", ") }}.{% endif %}{% if !q.ahead.is_empty() %} Queued ahead: {{ q.ahead.join(", ") }}.{% endif %}{% if q.hidden > 0 %} Also ahead or running: {{ q.hidden }} job{% if q.hidden != 1 %}s{% endif %} in private repositories.{% endif %}</div></section>
15 +
{% endif %}
13 16
{% if run.status == "error" %}
14 17
<section class="panel"><div class="strip bad">The pipeline is not valid, so nothing ran.</div><pre class="snippet">{{ run.error }}</pre><p class="small mute">Run <code>flow check</code> locally to see the same problems.</p></section>
15 18
{% else %}
…
25 unchanged lines
modified templates/flow_runs.html +1 −1
…
34 unchanged lines
35 35
  {% for row in runs %}
36 36
  <tr class="st-{{ row.r.status }}">
37 37
    <td class="st"><span class="flow-glyph st-{{ row.r.status }}" title="{{ row.r.status }}">{{ crate::web::flow::glyph(row.r.status) }}</span></td>
38 -
    <td><a href="{{ repo.url }}/flow/{{ row.r.number }}"><strong>#{{ row.r.number }}</strong></a> <span class="small mute">{{ row.r.status }}</span></td>
38 +
    <td><a href="{{ repo.url }}/flow/{{ row.r.number }}"><strong>#{{ row.r.number }}</strong></a> <span class="small mute">{{ row.r.status }}{% if let Some(p) = row.place %}, {{ p }}{% endif %}</span></td>
39 39
    <td class="flow-dots">{% for (n, s) in row.jobs %}<span class="flow-glyph st-{{ s }}" title="{{ n }}: {{ s }}">{{ crate::web::flow::glyph(s) }}</span>{% endfor %}{% if row.r.status == "error" %}<span class="small bad">pipeline invalid</span>{% endif %}</td>
40 40
    <td><span class="lbl lbl-bm">{{ row.r.bookmark }}</span></td>
41 41
    <td><a href="{{ repo.url }}/tree/{{ row.r.heads }}" title="state {{ row.r.state }}"><code>{{ row.r.state|short }}</code></a></td>
…
10 unchanged lines