Runners are offered only jobs whose image they hold (closes #73) b0612e7dce30

- **Runners get only jobs they can run (#73).** A runner that lists the rootfs images it holds (OxygenFlow does, at every claim) is offered only jobs whose image is among them, rather than jobs it would fail for want of one.

/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no

/// limit, and the rootfs images it holds, `None` when it does not say.

#[derive(Clone, Debug, Default)]

    pub images: Option<std::collections::HashSet<String>>,

    /// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,

    /// and its rootfs image among the runner's (oxygenhub#73).

        let rootfs = v["rootfs"].as_str().unwrap_or("");

        self.memory_mib.is_none_or(|m| need("memory_mib") <= m as u64)

            && self.cpus.is_none_or(|c| need("cpus") <= c as u64)

            && (rootfs.is_empty() || self.images.as_ref().is_none_or(|have| have.contains(rootfs)))

    let room = fdb::Room {

        memory_mib: req.free_memory_mib,

        cpus: req.free_cpus,

        images: req.images.as_ref().map(|i| i.iter().map(|h| h.to_ascii_lowercase()).collect()),

    };

/// oxygenhub#73: a runner that lists the images it holds is offered only

/// the jobs whose rootfs is among them; one that does not say gets any.

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

async fn a_runner_is_offered_only_jobs_whose_image_it_holds() {

    let h = hub().await;

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

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

    let url = format!("{}/andy/images", 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));

    assert_eq!(runs_of(&h.app, "andy", "images").await.len(), 1);

    let claim_with = |images: Vec<String>| {

        let body = serde_json::json!({ "runner": "test-1", "runtimes": ["microvm"], "images": images }).to_string();

        let app = h.app.clone();

        async move {

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

            (st, body)

        }

    };

    let (st, _) = claim_with(vec!["ab".repeat(32)]).await;

    assert_eq!(st, StatusCode::NO_CONTENT, "no job for a runner without the image");

    let jobs = db::flow::jobs(&h.app.db, runs_of(&h.app, "andy", "images").await[0].id).await.unwrap();

    let rootfs: serde_json::Value = serde_json::from_str(&jobs[0].sandbox).unwrap();

    let rootfs = rootfs["rootfs"].as_str().unwrap().to_string();

    let (st, body) = claim_with(vec!["ab".repeat(32), rootfs]).await;

    assert_eq!(st, StatusCode::OK, "{body}");

}