OXYGENHUB

Flow: say why a queued job waits, in oh and its check 67040b158ae2

The placement pass already worked out why no runner can take a ready
queued job (scope, tier, architecture, image held, labels, capacity, a
recent decline) and kept it on the job. oh run view and oh run watch now
print it under the job, the run page opens that job's panel to show it,
and the job's check carries it whenever it changes, so a proposal says
why its check is stuck instead of only 'waiting for a runner'.

closes #28
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-10-11 11:36 UTC
Signature
✔ signed by Andy McCall
Hash
67040b158ae233ccec6fa90ca649cce9ebb3761ee64a23f5a1d9265c54b46bd3
Depends on
0b203bbf7cc3 2a83f5303458 36e5a9c15a84 42fa95540468 4d2e44d79b03 5c6e92fd8d71 93be9ce4653e cc19697f8871 efd29a0033e8
Depended on by
fac6853c4e46 794ff94f194a 13b27cb2200c 58cbf42447bc 501d8b268dd4 92a797b20912
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 wiki-says-so
In tags
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
11 atoms, 0 tree operations, 5 files

Browse the tree at this change

modified crates/hub/src/flow.rs +9 −0
…
738 unchanged lines
739 739
    for (job, owner, repo_id) in ready {
740 740
        let declines = fdb::declines_of_job(&app.db, job.id).await?;
741 741
        let why = why_waiting(&job, &owner, repo_id, &runners, &archs, &declines, now);
742 +
        if why == job.waiting {
743 +
            continue;
744 +
        }
742 745
        fdb::set_waiting(&app.db, job.id, &why).await?;
746 +
        // The job's check says it too, so a proposal shows why it is stuck
747 +
        // (oxygenhub#28).
748 +
        if let (Some(run), Some(row)) = (fdb::run_by_id(&app.db, job.run_id).await?, db::repos::get_by_id(&app.db, repo_id).await?) {
749 +
            let summary = if why.is_empty() { "waiting for a runner" } else { why.as_str() };
750 +
            report_check(app, &row, &run.state, run.number, &job.name, "queued", summary).await;
751 +
        }
743 752
    }
744 753
    Ok(())
745 754
}
…
759 unchanged lines
modified crates/hub/tests/flow.rs +34 −0
…
2314 unchanged lines
2315 2315
    assert!(page.contains("1 failing") && page.contains("Newly failing") && page.contains("b broke"), "{page}");
2316 2316
    assert!(page.contains("1 new tests"), "d is new");
2317 2317
}
2318 +
2319 +
/// oxygenhub#28: a queued job no runner can take says why, in the API, on
2320 +
/// the run page (its panel open) and in its check.
2321 +
#[tokio::test(flavor = "multi_thread")]
2322 +
async fn a_job_no_runner_can_take_says_why() {
2323 +
    let h = hub().await;
2324 +
    create(&h.app, "andy", "demo").await;
2325 +
    let url = format!("{}/andy/demo", h.base);
2326 +
    let work = tempfile::tempdir().unwrap();
2327 +
    let c = client(&url, work.path());
2328 +
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
2329 +
    write(work.path(), "test.sh", "echo ok\n");
2330 +
    c.commit("Add a pipeline", &[]).unwrap();
2331 +
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
2332 +
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
2333 +
2334 +
    // One runner, online, holding another image only.
2335 +
    let r = db::runners::configured(&h.app.db, &blake3::hash(b"elsewhere").to_hex(), "earth-9").await.unwrap();
2336 +
    db::runners::seen(&h.app.db, r.id, None, None).await.unwrap();
2337 +
    let other = "ab".repeat(32);
2338 +
    db::runners::set_offer(&h.app.db, r.id, &["microvm".to_string()], Some(&[other]), "p").await.unwrap();
2339 +
    oxygenhub::flow::explain_waits(&h.app).await.unwrap();
2340 +
2341 +
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", Some("site"), None).await;
2342 +
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
2343 +
    let build = v["jobs"].as_array().unwrap().iter().find(|j| j["name"] == "build").unwrap().clone();
2344 +
    let why = build["waiting"].as_str().unwrap_or_default().to_string();
2345 +
    assert!(why.contains("earth-9 does not hold image") && why.contains(&ROOTFS[..12]), "{body}");
2346 +
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/1", None, None).await;
2347 +
    assert!(page.contains("data-waiting") && page.contains("does not hold image"), "the run page says it");
2348 +
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
2349 +
    let check = checks.iter().find(|c| c.name == "flow/build").unwrap();
2350 +
    assert!(check.summary.contains("does not hold image"), "{check:?}");
2351 +
}
modified crates/oh/src/main.rs +10 −2
…
892 unchanged lines
893 893
        };
894 894
        let on = if j.runner.is_empty() { String::new() } else { format!(" on {}", j.runner) };
895 895
        println!("  {} {}{}", ui::status(&j.status), j.name, ui::dim(&format!("{dur}{on}")));
896 +
        // Why a queued job no runner can take waits (oxygenhub#28).
897 +
        if j.status == "queued" && !j.waiting.is_empty() {
898 +
            println!("    {}", ui::warn(&format!("waiting: {}", j.waiting)));
899 +
        }
896 900
    }
897 901
    if !r.error.is_empty() {
898 902
        println!("{}", ui::bad(&r.error));
…
12 unchanged lines
911 915
        let r: FlowRun = c.get(&path)?;
912 916
        if !json {
913 917
            for j in &r.jobs {
914 -
                if seen.get(&j.name) != Some(&j.status) {
918 +
                let shown = format!("{}\u{0}{}", j.status, j.waiting);
919 +
                if seen.get(&j.name) != Some(&shown) {
915 920
                    println!("{} {}", ui::status(&j.status), j.name);
916 -
                    seen.insert(j.name.clone(), j.status.clone());
921 +
                    if j.status == "queued" && !j.waiting.is_empty() {
922 +
                        println!("  {}", ui::warn(&format!("waiting: {}", j.waiting)));
923 +
                    }
924 +
                    seen.insert(j.name.clone(), shown);
917 925
                }
918 926
            }
919 927
        }
…
352 unchanged lines
modified docs/spec.md +1 −1
…
514 unchanged lines
515 515
  - its key is the one the runner's token first signed with. The first signed receipt under a token binds that token to its key in `flow_runner_keys`.
516 516
517 517
  Otherwise the job's result still stands, and the run page says why the receipt was not kept. The run page links each kept receipt, with its signer. The change page lists the receipts for the state whose only head is that change. `/{owner}/{name}/receipts/{attest}` downloads a signed receipt for `oxy receipt verify` or `import`.
518 -
- **Architecture** (OxygenFlow spec §11.2–§11.3, oxygenflow#36): a runner reports each image's architecture on its claims (`image_archs`), kept in `flow_images`; a job is offered only to a runner whose architecture matches its sandbox's `arch`, else its image's as known; unknown matches, as before. The queue says which architecture a waiting job asks for.
518 +
- **Architecture** (OxygenFlow spec §11.2–§11.3, oxygenflow#36): a runner reports each image's architecture on its claims (`image_archs`), kept in `flow_images`; a job is offered only to a runner whose architecture matches its sandbox's `arch`, else its image's as known; unknown matches, as before. The queue says which architecture a waiting job asks for. A placement pass every 15 seconds works out why each ready queued job no runner can take waits, from what each online runner offered at its last claim (scope, tier, architecture, image held, labels, memory and CPUs, a recent decline), and shows it on the job (API `waiting`, `oh run view` and `oh run watch`, its panel open on the run page) and in its check's summary, so a proposal says why its check is stuck (#28).
519 519
- **Variables** (oxygenhub#102): Administration, an organisation's settings and a repository's settings each hold `NAME=value` pairs (a name is a letter or `_` then letters, digits and `_`, up to 64, never `FLOW_*`; a value up to 4096 characters; up to 100 per level) under **Variables**, each page showing what it inherits and which inherited names it overrides. A claim carries the repository's effective set as `env`, inner levels overriding outer; Flow puts them in the job's environment under `FLOW_*` and the pipeline's own `env`, so they are part of the receipt. They are not secrets and are shown in full to whoever sees the page; secrets are mounted by Flow (OxygenFlow spec §9), never put in the environment. Changes are audited (`flow.variable.set`, `flow.variable.deleted`); `flow run --var NAME=VALUE` gives a local run the same.
520 520
- **Secrets** (oxygenhub#107; OxygenFlow spec §9): the same three levels hold **Secrets**, named as a pipeline's `secret "name" { mount-path "/run/secrets/…" }` names them (1 to 64 of `A-Z a-z 0-9 . _ -`, a value up to 64 KiB, up to 100 per level), typed once and sealed under `keys/secrets.key`; pages show names, when each was set and where it comes from, never a value. A claim carries only the secrets its job declares, each with its path and value, from the innermost level that sets the name; one no level sets is noted at the top of the job's log and left out, so the job fails on the missing file rather than running with a stand-in. Flow writes each as a file (0400, the job's user's, on an in-memory mount) before the job starts and never puts it in the environment, the receipt or an artifact. Log intake replaces each value, and its base64, with `***` before a record is stored. Changes are audited (`flow.secret.set`, `flow.secret.deleted`), never with the value.
521 521
- **Artifacts** (OxygenFlow spec §5.2): a job may declare artifacts with `output`.
…
227 unchanged lines
modified templates/flow_run.html +1 −1
…
30 unchanged lines
31 31
  </svg>
32 32
</section>
33 33
{% for l in logs %}
34 -
<details class="panel flow-job st-{{ l.j.status }}" id="job-{{ l.j.id }}" data-job="{{ l.j.id }}"{% if l.j.status == "failed" || l.j.status == "error" || l.j.status == "running" %} open{% endif %}>
34 +
<details class="panel flow-job st-{{ l.j.status }}" id="job-{{ l.j.id }}" data-job="{{ l.j.id }}"{% if l.j.status == "failed" || l.j.status == "error" || l.j.status == "running" || (l.j.status == "queued" && !l.j.waiting.is_empty()) %} open{% endif %}>
35 35
  <summary><span class="flow-glyph st-{{ l.j.status }}" data-glyph>{{ crate::web::flow::glyph(l.j.status) }}</span> <strong>{{ l.j.name }}</strong> <span class="small mute" data-meta>{{ l.j.status }}{% if let Some(c) = l.j.exit_code %} · exit {{ c }}{% endif %} · {{ l.duration }}{% if !l.j.runner.is_empty() %} · on {{ l.j.runner }}{% endif %}</span>{% if !l.j.receipt_attest.is_empty() %} <span class="small mute">· <a href="{{ repo.url }}/receipts/{{ l.j.receipt_attest }}" title="Receipt {{ l.j.receipt_hash }}, kept in the repository; check it with oxy receipt verify">signed receipt</a> by {{ l.j.receipt_signer }}</span>{% else if !l.j.receipt_note.is_empty() %} <span class="small warn" title="{{ l.j.receipt_note }}">· receipt not kept: {{ l.j.receipt_note }}</span>{% else if !l.j.receipt_hash.is_empty() %} <span class="small mute">· unsigned receipt <code title="{{ l.j.receipt_hash }}">{{ l.j.receipt_hash|short }}</code></span>{% endif %}{% if !l.artifacts.is_empty() %} <span class="small mute">· {{ l.artifacts.len()|plural("artifact", "artifacts") }}</span>{% endif %}{% if let Some(t) = l.tests %} <span class="small{% if t.broken() > 0 %} warn{% endif %}">· <a href="#tests-{{ l.j.id }}">{{ t.summary() }}</a></span>{% endif %}{% if let Some((cid, pct)) = l.coverage %} <span class="small">· <a href="{{ repo.url }}/coverage?report={{ cid }}">coverage {{ pct }}</a></span>{% endif %}</summary>
36 36
  {% if !l.j.error.is_empty() %}<div class="strip bad">{{ l.j.error }}</div>{% endif %}
37 37
  {% if !l.artifacts.is_empty() %}<ul class="flow-artifacts small">{% for a in l.artifacts %}<li><a href="{{ repo.url }}/flow/{{ run.number }}/artifacts/{{ l.j.name }}/{{ a.name }}">{{ a.name }}</a> <span class="mute">{{ a.path }} · {{ a.size|bytes }} · blob <code title="{{ a.blob }}">{{ a.blob|short }}</code></span>{% if l.j.status != "passed" %} <span class="warn" title="Kept with when &quot;always&quot;: the job did not pass, so its receipt does not name this file">· from a job that did not pass, not in its receipt</span>{% endif %}{% if !releases.is_empty() && l.j.status == "passed" %} <form class="tiny inline" method="post" action="{{ repo.url }}/flow/{{ run.number }}/artifacts/{{ l.j.name }}/{{ a.name }}/attach"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><select name="release" aria-label="Release to attach {{ a.name }} to">{% for r in releases %}<option value="{{ r.id }}">{{ r.tag }}</option>{% endfor %}</select> <button>Attach to release</button></form>{% endif %}</li>{% endfor %}</ul>{% endif %}
…
16 unchanged lines