Jobs go to runners of the right architecture efd29a0033e8

-- The architecture of each rootfs image, as runners report it from their

-- image manifests (OxygenFlow spec §11.2, oxygenflow#36): a job whose

-- image is for one architecture is never offered to a runner of another.

CREATE TABLE flow_images (

    hash     TEXT PRIMARY KEY,

    arch     TEXT NOT NULL,

    seen_ms  INTEGER NOT NULL

);

- **Jobs go to runners of the right architecture (oxygenflow#36).** Runners report each image's architecture; the hub matches a job's `arch` or its image's against the runner's, and the queue says what a waiting job asks for.

/// The architecture a sandbox's JSON asks for, if any.

fn arch_in(sandbox: &str) -> Option<String> {

    serde_json::from_str::<serde_json::Value>(sandbox).ok()?["arch"].as_str().map(String::from)

}

    }

    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).

    pub fn arch(&self) -> Option<String> {

        arch_in(&self.sandbox)

    /// The runner's architecture (`x86_64`, `riscv64`, …), empty when unknown.

    pub arch: String,

    /// Each image's architecture as runners have reported it (`flow_images`).

    pub image_archs: std::collections::HashMap<String, String>,

        // The architecture: the sandbox's own ask, else the image's as known;

        // an unknown one matches, as before (OxygenFlow spec §11.3).

        let wanted = v["arch"].as_str().map(String::from).or_else(|| self.image_archs.get(rootfs).cloned());

        let arch_ok = self.arch.is_empty() || wanted.as_deref().is_none_or(|w| w == self.arch);

        arch_ok

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

    }

}

/// Remember which architecture each image is for, as a runner reports it.

pub async fn record_image_archs(pool: &Pool, archs: &std::collections::BTreeMap<String, String>) -> sqlx::Result<()> {

    for (hash, arch) in archs.iter().take(500) {

        if hash.len() != 64 || !flow_core::arch_ok(arch) {

            continue;

        }

        sqlx::query("INSERT INTO flow_images (hash, arch, seen_ms) VALUES (?, ?, ?) ON CONFLICT (hash) DO UPDATE SET arch = excluded.arch, seen_ms = excluded.seen_ms")

        .bind(hash.to_ascii_lowercase())

        .bind(arch)

        .bind(now_ms())

        .execute(pool)

        .await?;

    Ok(())

}

/// Every image's architecture the hub knows.

pub async fn image_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {

    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, arch FROM flow_images").fetch_all(pool).await?;

    Ok(rows.into_iter().collect())

    // What it holds and what it is (OxygenFlow spec §11): the images'

    // architectures it reports are remembered for every runner's claims.

    if let Some(archs) = &req.image_archs {

        fdb::record_image_archs(&app.db, archs).await?;

    }

    let arch = req.info.as_ref().map(|i| i.arch.clone()).filter(|a| !a.is_empty()).unwrap_or_else(|| me.arch.clone());

        arch,

        image_archs: fdb::image_archs(&app.db).await?,

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

async fn jobs_are_offered_only_to_runners_of_their_architecture() {

    let h = hub().await;

    let app = &h.app;

    create(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("sandbox \"board\" {\n    rootfs \"cas://blob/{ROOTFS}\"\n    arch \"riscv64\"\n}\njob \"flash\" {\n    sandbox \"board\"\n    exec \"flash\"\n}\n").replace("{ROOTFS}", &"ab".repeat(32)));

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

    c.snapshot("test").unwrap();

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

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

    let claim_as = |arch: &'static str, archs: serde_json::Value| {

        let app = app.clone();

        async move {

            let body = serde_json::json!({

                "runner": format!("{arch}-1"), "runtimes": ["microvm"],

                "info": { "arch": arch, "os": "linux", "version": "t", "labels": [], "max_jobs": 1 },

                "image_archs": archs,

            });

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

            match st {

                StatusCode::NO_CONTENT => None,

                StatusCode::OK => {

                    Some(serde_json::from_str::<serde_json::Value>(&body).unwrap()["job"]["name"].as_str().unwrap().to_string())

                }

                other => panic!("claim: {other} {body}"),

            }

        }

    };

    // An x86_64 runner that says the alpine image is x86_64: it takes the

    // alpine jobs, never the one that asks for riscv64.

    let rootfs = pipeline("").split("cas://blob/").nth(1).unwrap()[..64].to_string();

    for _ in 0..50 {

        if !runs_of(app, "andy", "demo").await.is_empty() {

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    }

    let runs = runs_of(app, "andy", "demo").await;

    assert_eq!(runs.len(), 1, "{runs:?}");

    let mut taken = vec![];

    while let Some(j) = claim_as("x86_64", serde_json::json!({ rootfs.clone(): "x86_64" })).await {

        taken.push(j);

    }

    assert_eq!(taken, ["build", "lint"], "test waits for build; flash is for riscv64");

    // A riscv64 runner: the alpine image is known to be x86_64, so only flash.

    assert_eq!(claim_as("riscv64", serde_json::json!({})).await.as_deref(), Some("flash"));

    assert_eq!(claim_as("riscv64", serde_json::json!({})).await, None);

    let (_, queue) = call(app, "GET", "/admin/runners", None, None).await;

    let _ = queue;

}

- **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.

  {% for q in queue %}<tr><td><a href="/{{ q.owner }}/{{ q.name }}">{{ q.owner }}/{{ q.name }}</a></td><td><a href="/{{ q.owner }}/{{ q.name }}/flow/{{ q.run_number }}">#{{ q.run_number }}</a></td><td>{{ q.job }}</td><td>{{ q.status }}</td><td>{% if q.runner.is_empty() %}<span class="mute">waiting{% let needs = q.needs() %}{% if !needs.is_empty() %} for a runner with {% for l in needs %}<span class="lbl">{{ l }}</span> {% endfor %}{% endif %}{% if let Some(a) = q.arch() %} on <span class="lbl">{{ a }}</span>{% endif %}</span>{% else %}{{ q.runner }}{% endif %}</td></tr>{% endfor %}