The hub reads what each image is for, and learns the rest from passes 93501da2a162

    // Kept with the architecture its own files name (oxygenhub#173).

    let m = tokio::task::spawn_blocking(move || store.keep_reading_arch(&m))

    Ok((StatusCode::CREATED, Json(serde_json::json!({ "hash": m.hash, "size": m.size, "chunks": m.chunks.len(), "arch": m.arch }))))

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

        // or learned from its passes (oxygenhub#173); one nobody has read

        // or run yet matches (OxygenFlow spec §11.3).

/// Give every image nobody has read the architecture of the runners its

/// jobs passed on (oxygenhub#173), for the jobs before the hub learned at

/// each pass. Never overrides a known architecture.

pub async fn learn_archs_from_runs(pool: &Pool) -> sqlx::Result<u64> {

    let r = sqlx::query(

        "INSERT INTO flow_images (hash, arch, seen_ms)

         SELECT lower(json_extract(j.sandbox, '$.rootfs')), MIN(r.arch), ? FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id

         WHERE j.status = 'passed' AND r.arch != '' AND length(json_extract(j.sandbox, '$.rootfs')) = 64

         GROUP BY lower(json_extract(j.sandbox, '$.rootfs'))

         ON CONFLICT (hash) DO NOTHING",

    )

    .bind(now_ms())

    .execute(pool)

    .await?;

    Ok(r.rows_affected())

}

    // A pass teaches what its image is for, when nobody has read it

    // (oxygenhub#173): the architecture of the runner it passed on.

    if o.status == "passed" {

        let rootfs = serde_json::from_str::<serde_json::Value>(&j.sandbox)

            .ok()

            .and_then(|v| v["rootfs"].as_str().map(str::to_ascii_lowercase))

            .unwrap_or_default();

        if rootfs.len() == 64 {

            sqlx::query(

                "INSERT INTO flow_images (hash, arch, seen_ms) SELECT ?, r.arch, ? FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id \

                 WHERE j.id = ? AND r.arch != '' ON CONFLICT (hash) DO NOTHING",

            )

            .bind(&rootfs)

            .bind(now)

            .bind(job_id)

            .execute(&mut *tx)

            .await?;

        }

    }

        self.keep_manifest(m)

    }

    /// Write manifest `m`, whose image is here.

    fn keep_manifest(&self, m: &Manifest) -> Result<(), String> {

    /// What image `m` is for, read from its own files (OxygenFlow spec

    /// §11.2, oxygenhub#173): its shell's ELF header, found through its ext4

    /// out of the kept chunks, a few of them, not the whole image. `None`

    /// when it is not ext4 or has no shell to read.

    pub fn read_arch(&self, m: &Manifest) -> Option<&'static str> {

        let mut starts = Vec::with_capacity(m.chunks.len());

        let mut end = 0u64;

        for c in &m.chunks {

            starts.push(end);

            end += c.len() as u64;

        }

        let mut cache: std::collections::HashMap<usize, Vec<u8>> = std::collections::HashMap::new();

        let short = || std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "past the image's end");

        jet::rootfs::arch(&mut |at: u64, buf: &mut [u8]| {

            let mut done = 0usize;

            while done < buf.len() {

                let pos = at + done as u64;

                let i = starts.partition_point(|&s| s <= pos).checked_sub(1).ok_or_else(short)?;

                let c = &m.chunks[i];

                let within = (pos - starts[i]) as usize;

                if within >= c.len() {

                    return Err(short());

                }

                let n = (c.len() - within).min(buf.len() - done);

                if c.is_zeros() {

                    buf[done..done + n].fill(0);

                } else {

                    if !cache.contains_key(&i) {

                        if cache.len() >= 32 {

                            cache.clear();

                        }

                        cache.insert(i, self.chunk(c.hash())?);

                    }

                    let bytes = &cache[&i];

                    if bytes.len() != c.len() {

                        return Err(short());

                    }

                    buf[done..done + n].copy_from_slice(&bytes[within..within + n]);

                }

                done += n;

            }

            Ok(())

        })

    }

    /// Keep image `m`, checked as [`Store::verify_and_keep`], with the

    /// architecture its own files name: the manifest's word is taken only

    /// when the image cannot be read. The manifest as kept.

    pub fn keep_reading_arch(&self, m: &Manifest) -> Result<Manifest, String> {

        self.verify_and_keep(m)?;

        match self.read_arch(m) {

            Some(a) if m.arch.as_deref() != Some(a) => {

                let mut kept = m.clone();

                kept.arch = Some(a.to_string());

                self.keep_manifest(&kept)?;

                Ok(kept)

            }

            _ => Ok(m.clone()),

        }

    }

    /// Record `arch` in kept image `hash`'s manifest.

    fn set_arch(&self, hash: &str, arch: &str) -> Result<(), String> {

        let mut m = self.manifest(hash).ok_or("no such image")?;

        m.arch = Some(arch.to_string());

        self.keep_manifest(&m)

    }

         ON CONFLICT (hash) DO UPDATE SET arch = excluded.arch WHERE excluded.arch != ''",

}

/// Learn what every image is for (oxygenhub#173), once at start: read

/// each kept image that has no architecture from its own files, then give

/// any image still unknown the architecture its jobs passed on. A runner's

/// own report, read from the image it holds, wins over what was learned.

/// How many images were read.

pub async fn learn_archs(app: &crate::Shared) -> sqlx::Result<usize> {

    let mut read = 0;

    for k in kept(&app.db).await?.into_iter().filter(|k| k.arch.is_empty()) {

        let store = app.images();

        let hash = k.hash.clone();

        let arch = tokio::task::spawn_blocking(move || {

            let a = store.read_arch(&store.manifest(&hash)?)?;

            store.set_arch(&hash, a).ok()?;

            Some(a)

        })

        .await

        .ok()

        .flatten();

        if let Some(a) = arch {

            sqlx::query("UPDATE jet_images SET arch = ? WHERE hash = ?").bind(a).bind(&k.hash).execute(&app.db).await?;

            crate::db::flow::record_image_archs(&app.db, &[(k.hash.clone(), a.to_string())].into_iter().collect()).await?;

            read += 1;

        }

    }

    crate::db::flow::learn_archs_from_runs(&app.db).await?;

    Ok(read)

    // What each image is for, read or learned once (oxygenhub#173).

    tokio::spawn({

        let app = app.clone();

        async move {

            match images::learn_archs(&app).await {

                Ok(n) if n > 0 => tracing::info!(images = n, "read the architecture of kept images"),

                Ok(_) => {}

                Err(e) => tracing::warn!(error = %e, "learning image architectures"),

            }

        }

    });

/// Send image bytes to the hub as their chunks and then a manifest that

/// names no architecture; the manifest as the hub answers it.

async fn send_image(app: &Shared, image: &[u8]) -> (String, serde_json::Value) {

    let mut chunks = std::collections::HashMap::new();

    let m = jet::images::scan(image, None, |c, b| {

        chunks.insert(c.hash().to_string(), b.to_vec());

        Ok(())

    })

    .unwrap();

    for c in m.needed() {

        let (st, body) = put_bytes(app, &format!("/api/v1/flow/images/chunks/{c}"), "site", chunks[c].clone()).await;

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

    }

    let (st, body) =

        call(app, "PUT", &format!("/api/v1/flow/images/{}", m.hash), Some("site"), Some(serde_json::to_string(&m).unwrap())).await;

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

    (m.hash.clone(), serde_json::from_str(&body).unwrap())

}

/// oxygenhub#173 (OxygenFlow spec §11.2, Andy's "detect and learn"): an

/// image sent with no architecture is read from its own files, so a

/// riscv64 runner is never offered an x86_64 image's jobs; an image

/// nobody can read takes the architecture its first pass ran on.

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

async fn an_image_names_its_own_architecture_and_a_pass_teaches_the_rest() {

    use oxy_core::identity::SecretKey;

    let h = hub().await;

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

    // An ext4 image whose shell is x86_64, as `flow rootfs import` makes one.

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

    let tree = dir.path().join("tree");

    std::fs::create_dir_all(tree.join("bin")).unwrap();

    let mut elf = vec![0u8; 64];

    elf[..7].copy_from_slice(b"\x7fELF\x02\x01\x01");

    elf[18..20].copy_from_slice(&0x3Eu16.to_le_bytes());

    std::fs::write(tree.join("bin/sh"), &elf).unwrap();

    let img = dir.path().join("rootfs.ext4");

    let made = std::process::Command::new("mkfs.ext4").args(["-q", "-F", "-d"]).arg(&tree).arg(&img).arg("8M").status();

    if !made.is_ok_and(|s| s.success()) {

        eprintln!("mkfs.ext4 not found: skipped");

        return;

    }

    let (x86, answer) = send_image(&h.app, &std::fs::read(&img).unwrap()).await;

    assert_eq!(answer["arch"], "x86_64", "read from its shell: {answer}");

    let (_, body) = call(&h.app, "GET", &format!("/api/v1/flow/images/{x86}"), Some("runner-secret"), None).await;

    assert_eq!(serde_json::from_str::<jet::images::Manifest>(&body).unwrap().arch.as_deref(), Some("x86_64"), "runners fetch it named");

    // One nobody can read: no ext4, no shell.

    let blob: Vec<u8> = (0..300_000u32).map(|i| (i % 241) as u8).collect();

    let (unread, answer) = send_image(&h.app, &blob).await;

    assert!(answer["arch"].is_null(), "{answer}");

    let archs = db::flow::image_archs(&h.app.db).await.unwrap();

    assert_eq!((archs.get(&x86).map(String::as_str), archs.get(&unread)), (Some("x86_64"), None));

    // Kept images with none are read once at start.

    sqlx::query("UPDATE jet_images SET arch = ''").execute(&h.app.db).await.unwrap();

    sqlx::query("DELETE FROM flow_images").execute(&h.app.db).await.unwrap();

    assert_eq!(oxygenhub::images::learn_archs(&h.app).await.unwrap(), 1);

    assert_eq!(db::flow::image_archs(&h.app.db).await.unwrap().get(&x86).map(String::as_str), Some("x86_64"));

    // A job on each image.

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

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

    let c = client(&format!("{}/andy/arch", h.base), work.path());

    let kdl = format!(

        "schema-version 1\nsandbox \"unread\" {{\n    rootfs \"cas://blob/{unread}\"\n}}\nsandbox \"x86\" {{\n    rootfs \"cas://blob/{x86}\"\n}}\n\

         job \"first\" {{\n    sandbox \"unread\"\n    exec \"make\"\n}}\njob \"second\" {{\n    sandbox \"x86\"\n    exec \"make\"\n    depends-on \"first\"\n}}\n\

         job \"third\" {{\n    sandbox \"unread\"\n    exec \"make\"\n    depends-on \"first\"\n}}\n"

    );

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

    c.commit("Jobs on two images", &[]).unwrap();

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

    runs_of(&h.app, "andy", "arch").await;

    // The x86_64 runner passes the first job: its image is x86_64 now.

    let x86_info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });

    let x86_claim =

        serde_json::json!({ "runner": "earth", "runtimes": ["microvm"], "images": [], "fetches_images": true, "info": x86_info });

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

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

    let first: serde_json::Value = serde_json::from_str(&body).unwrap();

    assert_eq!(first["job"]["name"], "first");

    let ph = first["pipeline_hash"].as_str().unwrap();

    let id = first["job_id"].as_i64().unwrap();

    assert_eq!(

        finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("first", ph, 0) })).await,

        StatusCode::NO_CONTENT

    );

    assert_eq!(db::flow::image_archs(&h.app.db).await.unwrap().get(&unread).map(String::as_str), Some("x86_64"), "learned");

    // A riscv64 runner is offered neither: one read, one learned.

    let token = "flowreg_narvi";

    db::runners::add_registration(&h.app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &Default::default()).await.unwrap();

    let identity = SecretKey::generate().identity("narvi", "", 1);

    let info = serde_json::json!({ "arch": "riscv64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });

    let reg = serde_json::json!({ "token": token, "name": "narvi", "identity": hex::encode(identity.encode()), "info": info });

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;

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

    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();

    let narvi =

        serde_json::json!({ "runner": "narvi", "runtimes": ["microvm"], "images": [], "fetches_images": true, "info": info }).to_string();

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(narvi)).await;

    assert_eq!(st, StatusCode::NO_CONTENT, "the second job's image is x86_64 and the third's learned so: {body}");

    for _ in 0..2 {

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

        assert_eq!(st, StatusCode::OK, "both go to x86_64: {body}");

    }

}

- **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. The hub knows an image's architecture from its own files, read through its ext4 to its shell's ELF header (`jet::rootfs`) out of the kept chunks, when it is sent (the manifest's word is taken only when the image cannot be read) and, once at start, for every kept image that had none; an image nobody can read takes the architecture of the runner its first job passed on, and a runner's own report, read from the image it holds, wins over that (oxygenhub#173, Andy's "detect and learn"). Only an image nobody has read or run matches any runner. 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 build cache, which counts most: claims list the caches a runner holds, `caches`, each `flow_core::api::cache_slot(repo, job)`, oxygenflow#51; or it holds the job's image while the claimant would fetch it; or it 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 its build cache, 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). **Diffusion** (#148, `docs/diffusion.md`): a push that moves a repository's default bookmark, or the bookmark its work lands on (the most common target of its proposals landed in the last 90 days), records each Oxygen subrepo pin in the new state (`diffusion_pins`: consumer, bookmark, path, the upstream's Oxygen repository id, the pinned heads); a hub that had no index fills it at start. `GET /repos/{owner}/{name}/consumers` lists the consumers the caller may read, each with its pin and how many changes of this repository's default bookmark it lacks (`behind`, null when the pin names changes this repository does not hold), and the overview shows them under Diffuses to. `GET /repos/{owner}/{name}/diffusion` (#153) gives both directions, `takes_from` (each upstream the repository pins) and `diffuses_to` (each consumer the caller may read), every entry naming its `upstream` and `consumer`, its `path`, `bookmark`, `behind`, `follow`, open `bump`, the bump's newest `run` and `run_status`, and the pill's halves (`upstream_state`, `consumer_state`: `tone` green, amber, red, grey or off, `word`, `title`); hub-api's `Diffusion`. A repository that pins, or is pinned by, another on the hub has a **Diffusion tab** (#150): Diffuses to (each consumer's pin bookmark, how far behind, its status, its open bump and that bump's newest run, its last landed bump, how it follows) and Takes from (each upstream it pins, with the same); its overview shows both in brief, and Admin › Diffusion lists every upstream with its consumers. Each consumer's status is a two-half pill: the upstream on the left (checked when its newest default-bookmark run passed, untested when it has no checks, failing, or testing) and the consumer on the right (passes or fails for its run on the bump, testing, waiting, or up to date), each half coloured green, amber, red or grey; the API keeps one overall `colour` as well. Landing a proposal on the hub refreshes the pin index and propagates as a push does. A bump's run, started by the `diffusion` account, is given the upstream it bumps as a subrepo whatever that account may read (the consumer's own pin of it was committed by someone who may write the consumer); its other subrepos need read access as usual. A consumer follows a subrepo's upstream as its `.oxymodules` entry says (`follow = off | green | any`, `land = propose | auto`, `target` naming the bookmark to bump, `watch` listing paths whose upstream changes are flagged), else as its repository setting says (Settings › Diffusion, or `PUT /repos/{owner}/{name}/diffusion {"follow": ...}` by its owners and admins; migration 0058), else as `[diffusion] follow` in hub.toml says (default `off`). A **bump** moves a consumer's pin at a path to a newer upstream state: one change on the target bookmark's heads that replaces only the pin (oxy-core `build_pin_change`), signed with the hub's key, on the bookmark `diffusion/<path>`, which the hub owns and rebuilds on the target each time (a forced update). When the pin has changes the upstream's state lacks (say, from an upstream proposal not landed yet), the bump moves to both, so a consumer never loses changes it relies on; it is skipped when that combination would conflict, when the pin names changes the upstream does not hold, when the pin already holds the state, or when the bump bookmark already moves it there, and the hub logs why. A bump is **proposed** once per bump bookmark, by the hub's `diffusion` account (made on first use, never able to sign in), into the target, labelled `dependencies`, its body the upstream changes the pin lacks (newest first, linking to the upstream's change pages); a later bump edits the same open proposal. It is triggered when a push moves the upstream's default bookmark (for consumers following `any`) or when a run of that bookmark passes (for `green`). Diffusion never lands a proposal. An upstream with no checks of its own (Flow off, or no run ever of its default bookmark) is **unverified**: it never turns green, so its moves also bump consumers following `green`, and those proposals are marked amber with the `upstream unverified` label and a note in the body; a later bump from a passing run removes the label. A bump already in place refreshes its proposal's text and labels. A consumer's `watch` patterns (`*` within a path segment, `**` across them, `?` one character; a plain path covers everything below it) flag the upstream changes touching a matching file: they are listed first under "Touches what you watch", with the files, and marked ⚑ in the changelog. Turning Diffusion on for a consumer (the setting or the API) catches it up at once: it is bumped to each upstream's newest state whose default-bookmark run passed, or to its newest state when it follows every move. **Images page** (#146): Admin › Runners › Images lists every rootfs image a runner holds, the hub keeps, or a job ran on in the last 90 days (from each job's sandbox), newest use first: its size and architecture, where it is (the hub, each runner holding it, each runner fetching it from the swarm now), the repositories whose jobs used it, and who sent it; an image held nowhere is marked. **The swarm** (#144, oxygenflow docs/jetstream.md §8.1): a runner in an image swarm reports its peer address, its site and the BLAKE3 of its TLS certificate with each claim (`swarm`; without a certificate it is not in the swarm), kept in memory and shown on Admin › Runners. `POST /flow/swarm/announce {image, state}` (by a runner; `fetching`, `done` or `stopped`) returns the online runners in its site that hold the image or announced fetching it in the last 45 seconds, each with its certificate's fingerprint, which the fetcher pins its TLS connection to, and a ticket signed with the hub key (JSON naming the runner, its site and an expiry 15 minutes on, hex, then the signature) that peers check against `/.well-known/oxyhub-key` before serving chunks; a runner that has not reported an address is told to add `[runner.swarm]` (409).