Flow: a pipeline pinning an image nobody holds fails a check 1c8276b90314

                images_check(app, row, state, run.number, &jobs).await;

/// The `pipeline/images` check (#137): set failing when an image the run's

/// jobs pin is held by neither the hub nor a runner, so a proposal pinning

/// one says so before it lands rather than its runs queueing for good;

/// passed again by a later run of the state once every image is held.

async fn images_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_number: i64, jobs: &[fdb::NewJob]) {

    let mut pinned: Vec<String> = jobs

        .iter()

        .filter_map(|j| serde_json::from_str::<serde_json::Value>(&j.sandbox).ok())

        .filter_map(|v| v["rootfs"].as_str().map(String::from))

        .filter(|r| !r.is_empty())

        .collect();

    pinned.sort();

    pinned.dedup();

    if pinned.is_empty() {

        return;

    }

    let mut held = crate::images::held(&app.db).await.unwrap_or_default();

    // Nobody can be said to lack an image unless some runner is known and

    // every one reports what it holds.

    let runners: Vec<db::runners::Runner> =

        db::runners::list(&app.db).await.unwrap_or_default().into_iter().filter(|r| r.state != "disabled").collect();

    if runners.is_empty() || runners.iter().any(|r| r.image_list().is_none()) {

        return;

    }

    for r in &runners {

        held.extend(r.image_list().unwrap_or_default());

    }

    let missing: Vec<&String> = pinned.iter().filter(|h| !held.contains(*h)).collect();

    // Only a missing image makes the check; one that was failing on this

    // state turns to passed once its images are held.

    let failing_before = db::hooks::checks_for_state(&app.db, row.id, state)

        .await

        .unwrap_or_default()

        .iter()

        .any(|c| c.name == "pipeline/images" && c.status != "passed");

    if missing.is_empty() && !failing_before {

        return;

    }

    let (status, summary) = if missing.is_empty() {

        (

            "passed",

            format!("{} now held", if pinned.len() == 1 { "its image is".to_string() } else { format!("all {} images are", pinned.len()) }),

        )

    } else {

        let names: Vec<String> = missing.iter().map(|h| h[..12.min(h.len())].to_string()).collect();

        ("failed", format!("neither the hub nor a runner holds image {}: send it with flow rootfs push HASH", names.join(", ")))

    };

    let url = format!("{}/{}/{}/flow/{run_number}", app.config.base_url(), row.owner, row.name);

    if let Ok(c) = db::hooks::upsert_check(

        &app.db,

        row.id,

        db::hooks::CheckReport { state, name: "pipeline/images", status, summary: &summary, url: &url, reporter: None, seal: "" },

    )

    .await

    {

        let api = crate::api::check_of(&c);

        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;

    }

}

    // #137: the run's images are held (by the hub), so no images check.

    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();

    let run1 = db::flow::run(&h.app.db, row.id, 1).await.unwrap().unwrap();

    let checks = checks_when(&h.app, row.id, &run1.state, 1).await;

    assert!(checks.iter().all(|c| c.name != "pipeline/images"), "{checks:?}");

}

/// #137: a run pinning an image neither the hub nor any runner holds sets a

/// failing `pipeline/images` check saying how to send it; once it is held,

/// a run of the same state turns the check to passed.

#[tokio::test(flavor = "multi_thread")]

async fn a_pipeline_pinning_an_image_nobody_holds_fails_a_check() {

    let h = hub().await;

    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    // The one runner reports holding no images.

    let none = r#"{"runner":"test-1","runtimes":["microvm"],"images":[]}"#;

    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(none.into())).await;

    assert_eq!(st, StatusCode::NO_CONTENT);

    create(&h.app, "andy", "demo").await;

    let url = format!("{}/andy/demo", h.base);

    let work = tempfile::tempdir().unwrap();

    let c = client(&url, work.path());

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "test.sh", "echo ok\n");

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

    let checks = checks_when(&h.app, run.repo_id, &run.state, 4).await;

    let check = checks.iter().find(|c| c.name == "pipeline/images").expect("an images check");

    assert!(check.status == "failed" && check.summary.contains(&ROOTFS[..12]) && check.summary.contains("flow rootfs push"), "{check:?}");

    // Once a runner holds it, a run of the state passes the check.

    let holds = format!(r#"{{"runner":"test-1","runtimes":["microvm"],"images":["{ROOTFS}"]}}"#);

    call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(holds)).await;

    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;

    assert_eq!(st, StatusCode::CREATED);

    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();

    let check = checks.iter().find(|c| c.name == "pipeline/images").unwrap();

    assert_eq!(check.status, "passed", "{check:?}");

- **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. **Images by hash** (oxygenflow#39, Jetstream): `flow rootfs push HASH` sends a rootfs image as the FastCDC chunks the hub lacks (`POST /flow/images/missing`, `PUT /flow/images/chunks/{hash}`, by a token that may write repositories) and then its manifest (`PUT /flow/images/{hash}`), which the hub accepts only when every chunk is held and they make an image with that hash; chunks are kept zstd-compressed under `<data>/images`, shared by every image, and `jet_images` lists what is held. Runners and accounts fetch with `GET /flow/images/{hash}` and `GET /flow/images/chunks/{hash}`. A runner whose claims say `fetches_images` is offered jobs whose image only the hub holds, and fetches it before its preflight. Each claim also carries the runner's `facts` (OxygenFlow spec §11.1, oxygenflow#40): free disk, load, speed score and class, kernel; the hub keeps the latest on the runner, shows them on Admin › Runners and in `GET /runners`. A job's finish carries the most disk and memory it took (`peak_disk_mib`, `peak_memory_mib`, oxygenflow#41); a new job's `need_disk_mib` is the most its last five runs of the same repository and name took, plus a quarter and 64 MiB, and it is offered only to a runner whose claim reports at least that much free disk; the waiting reason says so. **Ranking** (oxygenflow#42): a job that another free runner which polled in the last 10 s would do better (it holds the job's image while the claimant would fetch it, or has a higher speed class) is held from the claimant for up to 15 s, then any runner that fits takes it; each job records why it ran where it did (`placed`: "holds the image, idle, speed 100 (class 3)", with "after N s held for a better runner" when it was), shown on the run page, in the API and by `oh run view`. A run whose jobs pin an image held by neither the hub nor a runner sets a failing `pipeline/images` check on its state (#137), naming the images and how to send them (`flow rootfs push`), so a proposal pinning a new image shows it before it lands; a later run of the state turns it to passed once every image is held. 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).