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 "always": 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 |