Images by hash on the hub: chunks, verified manifests, and placement for runners that fetch 92a797b20912

-- Images by hash (oxygenflow#39, Jetstream): rootfs images this hub keeps

-- for runners to fetch, as chunks under <data>/images and a manifest each.

CREATE TABLE jet_images (

    hash        TEXT PRIMARY KEY,

    size        INTEGER NOT NULL,

    arch        TEXT NOT NULL,

    chunks      INTEGER NOT NULL,

    uploader    TEXT NOT NULL,

    created_ms  INTEGER NOT NULL

);

-- Whether a runner fetches a job's image it lacks from the hub, as its

-- claims say; the placement pass counts the hub's images as its own then.

ALTER TABLE flow_runners ADD COLUMN fetches_images INTEGER NOT NULL DEFAULT 0;

//! Images by hash over the API (oxygenflow#39): `flow rootfs push` sends

//! an image's missing chunks and then its manifest; runners, and `flow

//! rootfs pull`, fetch a manifest and the chunks they lack.

use super::{caller, not_found, ApiError};

use crate::Shared;

use axum::body::Bytes;

use axum::extract::{Path, State};

use axum::http::{header, HeaderMap, StatusCode};

use axum::response::{IntoResponse, Response};

use axum::Json;

use jet::images::{is_hash, Manifest};

use serde::{Deserialize, Serialize};

/// Most chunk names one `missing` question may ask about.

const MAX_ASK: usize = 200_000;

/// Who sends images: an account whose token may write repositories.

async fn sender(app: &Shared, headers: &HeaderMap) -> Result<String, ApiError> {

    let c = caller(app, headers).await?;

    Ok(c.require("repo:write")?.username.clone())

}

/// Who fetches images: a runner, or any account (for `flow rootfs pull`).

async fn fetcher(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {

    if crate::flow::runner_auth(app, headers).await.is_ok() {

        return Ok(());

    }

    let c = caller(app, headers).await?;

    c.account.as_ref().map(|_| ()).ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "a runner or personal access token is required".into()))

}

#[derive(Deserialize)]

pub struct Ask {

    chunks: Vec<String>,

}

#[derive(Serialize)]

pub struct Missing {

    missing: Vec<String>,

}

/// `POST /flow/images/missing {"chunks": [..]}`: which of these chunks the

/// hub lacks.

pub async fn missing(State(app): State<Shared>, headers: HeaderMap, Json(ask): Json<Ask>) -> Result<Json<Missing>, ApiError> {

    sender(&app, &headers).await?;

    if ask.chunks.len() > MAX_ASK {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("ask about at most {MAX_ASK} chunks at once")));

    }

    let store = app.images();

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

        let mut seen = std::collections::HashSet::new();

        ask.chunks.into_iter().filter(|h| is_hash(h) && seen.insert(h.clone()) && !store.has_chunk(h)).collect()

    })

    .await

    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(Missing { missing }))

}

/// `PUT /flow/images/chunks/{hash}`: one chunk, checked against its name.

pub async fn put_chunk(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(hash): Path<String>,

    body: Bytes,

) -> Result<StatusCode, ApiError> {

    sender(&app, &headers).await?;

    if !is_hash(&hash) || body.is_empty() || body.len() > oxy_core::chunk::MAX_SIZE {

        return Err(ApiError(StatusCode::BAD_REQUEST, "not a chunk: a name is 64 hex digits, a chunk 1 byte to 256 KiB".into()));

    }

    if blake3::hash(&body).to_hex().as_str() != hash {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("those bytes are not chunk {}", &hash[..12])));

    }

    let store = app.images();

    tokio::task::spawn_blocking(move || store.put_chunk(&hash, &body))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the chunk: {e}")))?;

    Ok(StatusCode::NO_CONTENT)

}

/// `PUT /flow/images/{hash}`: the manifest, accepted once every chunk is

/// here and the image they make has this hash.

pub async fn put_manifest(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(hash): Path<String>,

    Json(m): Json<Manifest>,

) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {

    let who = sender(&app, &headers).await?;

    if m.hash != hash {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));

    }

    let store = app.images();

    let kept = m.clone();

    tokio::task::spawn_blocking(move || store.verify_and_keep(&kept))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .map_err(|e| ApiError(StatusCode::CONFLICT, e))?;

    crate::images::record(&app.db, &m, &who).await?;

    crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok();

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

}

/// `GET /flow/images/{hash}`: an image's manifest.

pub async fn get_manifest(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Json<Manifest>, ApiError> {

    fetcher(&app, &headers).await?;

    app.images().manifest(&hash).map(Json).ok_or_else(|| not_found("this hub does not hold that image; send it with flow rootfs push"))

}

/// `GET /flow/images/chunks/{hash}`: one chunk's bytes.

pub async fn get_chunk(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Response, ApiError> {

    fetcher(&app, &headers).await?;

    let store = app.images();

    let bytes = tokio::task::spawn_blocking(move || store.chunk(&hash))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .map_err(|_| not_found("no such chunk"))?;

    Ok(([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, "private, max-age=31536000, immutable")], bytes)

        .into_response())

}

//! Images by hash (oxygenflow#39, OxygenFlow docs/jetstream.md §8): rootfs

//! images kept site-wide for runners to fetch, so nobody copies an image

//! between machines. An image arrives as chunks (Oxygen's FastCDC, sent

//! only when this hub lacks them) and then its manifest, which is accepted

//! only when every chunk is here and the image they make has the hash it

//! is named by. Chunks are kept zstd-compressed under

//! `<data>/images/chunks`, shared by every image that has them.

use crate::db::{now_ms, Pool};

use jet::images::{is_hash, Manifest};

use std::io::Write;

use std::path::{Path, PathBuf};

/// Where this hub keeps images.

#[derive(Clone, Debug)]

pub struct Store {

    root: PathBuf,

}

impl Store {

    pub fn new(data: &Path) -> Store {

        Store { root: data.join("images") }

    }

    fn chunk_path(&self, hash: &str) -> PathBuf {

        self.root.join("chunks").join(&hash[..2]).join(hash)

    }

    fn manifest_path(&self, hash: &str) -> PathBuf {

        self.root.join("manifests").join(format!("{hash}.json"))

    }

    pub fn has_chunk(&self, hash: &str) -> bool {

        is_hash(hash) && self.chunk_path(hash).exists()

    }

    /// Keep chunk `hash`, whose bytes the caller has checked.

    pub fn put_chunk(&self, hash: &str, bytes: &[u8]) -> std::io::Result<()> {

        let path = self.chunk_path(hash);

        if path.exists() {

            return Ok(());

        }

        std::fs::create_dir_all(path.parent().expect("a chunk's directory"))?;

        let tmp = path.with_extension(format!("tmp{}", std::process::id()));

        let mut f = std::fs::File::create(&tmp)?;

        f.write_all(&zstd::encode_all(bytes, 3)?)?;

        f.sync_all()?;

        std::fs::rename(&tmp, &path)

    }

    pub fn chunk(&self, hash: &str) -> std::io::Result<Vec<u8>> {

        if !is_hash(hash) {

            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "not a chunk name"));

        }

        zstd::decode_all(std::fs::File::open(self.chunk_path(hash))?)

    }

    pub fn manifest(&self, hash: &str) -> Option<Manifest> {

        if !is_hash(hash) {

            return None;

        }

        serde_json::from_slice(&std::fs::read(self.manifest_path(hash)).ok()?).ok()

    }

    /// Accept manifest `m`: well formed, every chunk here, and the image

    /// they make hashing to its name. Reads the whole image once.

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

        m.check().map_err(|e| e.to_string())?;

        let mut whole = blake3::Hasher::new();

        for c in &m.chunks {

            if c.is_zeros() {

                whole.update(&vec![0u8; c.len()]);

                continue;

            }

            let bytes = self.chunk(c.hash()).map_err(|_| format!("chunk {} was not sent; send it first", &c.hash()[..12]))?;

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

                return Err(format!("chunk {} is {} bytes, not {}", &c.hash()[..12], bytes.len(), c.len()));

            }

            whole.update(&bytes);

        }

        let got = whole.finalize().to_hex().to_string();

        if got != m.hash {

            return Err(format!("the chunks make an image named {}, not {}", &got[..12], &m.hash[..12]));

        }

        let path = self.manifest_path(&m.hash);

        std::fs::create_dir_all(path.parent().expect("the manifests directory")).map_err(|e| e.to_string())?;

        let tmp = path.with_extension("json.tmp");

        std::fs::write(&tmp, serde_json::to_vec(m).map_err(|e| e.to_string())?).map_err(|e| e.to_string())?;

        std::fs::rename(&tmp, &path).map_err(|e| e.to_string())

    }

}

/// Record an image this hub now keeps.

pub async fn record(pool: &Pool, m: &Manifest, uploader: &str) -> sqlx::Result<()> {

    sqlx::query(

        "INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES (?, ?, ?, ?, ?, ?) \

         ON CONFLICT (hash) DO NOTHING",

    )

    .bind(&m.hash)

    .bind(m.size as i64)

    .bind(m.arch.clone().unwrap_or_default())

    .bind(m.chunks.len() as i64)

    .bind(uploader)

    .bind(now_ms())

    .execute(pool)

    .await?;

    if let Some(a) = m.arch.as_ref().filter(|a| !a.is_empty()) {

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

    }

    Ok(())

}

/// Every image this hub keeps, by hash.

pub async fn held(pool: &Pool) -> sqlx::Result<std::collections::HashSet<String>> {

    Ok(sqlx::query_scalar::<_, String>("SELECT hash FROM jet_images").fetch_all(pool).await?.into_iter().collect())

}

- **Images by hash (oxygenflow#39).** Rootfs images live on the hub: `flow rootfs push` sends only the chunks the hub lacks (a rebuilt image costs its changes, runs of zeros cost nothing), the hub checks the whole image against its hash, and a runner that lacks a job's image fetches it instead of anyone copying it between machines. The hub offers such a job to runners that fetch, and says when neither the runner nor the hub holds the image.

name = "jet"

version = "0.1.0"

dependencies = [

 "blake3",

 "hex",

 "oxy-core",

 "serde",

 "serde_json",

 "thiserror 1.0.69",

]

[[package]]

 "jet",

jet = { path = "deps/oxygenflow/crates/jet" }

jet.workspace = true

mod images;

        .route("/flow/images/missing", post(images::missing))

        .route("/flow/images/chunks/{hash}", axum::routing::put(images::put_chunk).get(images::get_chunk))

        .route(

            "/flow/images/{hash}",

            axum::routing::put(images::put_manifest).get(images::get_manifest).layer(axum::extract::DefaultBodyLimit::max(64 << 20)),

        )

    /// Images the hub holds that this runner would fetch when it lacks

    /// them (oxygenflow#39); empty for a runner that does not fetch.

    pub fetchable: std::collections::HashSet<String>,

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

    /// It fetches images it lacks from the hub (oxygenflow#39).

    pub fetches_images: bool,

pub async fn set_offer(

    pool: &Pool,

    id: i64,

    tiers: &[String],

    images: Option<&[String]>,

    fetches_images: bool,

    profile: &str,

) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET tiers = ?, images = ?, fetches_images = ?, profile = ? WHERE id = ?")

        .bind(fetches_images)

pub(crate) async fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<db::runners::Runner, ApiError> {

    db::runners::set_offer(&app.db, me.id, &req.runtimes, req.images.as_deref(), req.fetches_images, &profile).await?;

        fetchable: if req.fetches_images { crate::images::held(&app.db).await? } else { Default::default() },

#[allow(clippy::too_many_arguments)]

    hub_images: &std::collections::HashSet<String>,

        let fetches = r.fetches_images && hub_images.contains(rootfs);

        if !fetches && r.image_list().is_some_and(|have| !have.iter().any(|h| h == rootfs)) {

            let why = if r.fetches_images { " (nor does the hub: send it with flow rootfs push)" } else { "" };

            return Some(format!("does not hold image {}{why}", &rootfs[..12.min(rootfs.len())]));

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

        let why = why_waiting(&job, &owner, repo_id, &runners, &archs, &declines, now, &held);

        let why = |runners: &[Runner], declines: &[fdb::DeclineRow]| {

            why_waiting(&job, "andy", 1, runners, &archs, declines, now, &Default::default())

        };

            why_waiting(&job, "andy", 1, &[riscv], &archs_known, &[], now, &Default::default()),

pub mod images;

    /// The rootfs images this hub keeps for runners (oxygenflow#39).

    pub fn images(&self) -> images::Store {

        images::Store::new(&self.config.data)

    }

    db::runners::set_offer(&h.app.db, r.id, &["microvm".to_string()], Some(&[other]), false, "p").await.unwrap();

async fn put_bytes(app: &Shared, path: &str, bearer: &str, bytes: Vec<u8>) -> (StatusCode, String) {

    let req = Request::put(path)

        .header(header::AUTHORIZATION, format!("Bearer {bearer}"))

        .header(header::CONTENT_TYPE, "application/octet-stream")

        .body(Body::from(bytes))

        .unwrap();

    let res = oxygenhub::router(app.clone()).oneshot(req).await.unwrap();

    let st = res.status();

    (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

}

/// oxygenflow#39: an image is sent as the chunks the hub lacks and then its

/// manifest, checked against its hash; runners fetch it; and a job whose

/// image only the hub holds is offered to a runner that fetches, not to one

/// that does not.

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

async fn images_are_sent_by_chunk_and_offered_to_runners_that_fetch() {

    let h = hub().await;

    // The site token acts as the first site administrator.

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

    // An image: bytes, a run of zeros, more bytes.

    let mut image: Vec<u8> = (0..400_000u32).map(|i| (i % 251) as u8).collect();

    image.extend(std::iter::repeat_n(0u8, 2 << 20));

    image.extend((0..100_000u32).map(|i| (i % 239) as u8));

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

    let m = jet::images::scan(&image[..], Some("x86_64".into()), |c, b| {

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

        Ok(())

    })

    .unwrap();

    let hash = m.hash.clone();

    // Which chunks the hub lacks: every one but zeros; a manifest before them is refused.

    let ask = serde_json::json!({ "chunks": m.needed() }).to_string();

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/images/missing", Some("site"), Some(ask.clone())).await;

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

    let missing: Vec<String> = serde_json::from_str::<serde_json::Value>(&body).unwrap()["missing"]

        .as_array()

        .unwrap()

        .iter()

        .map(|v| v.as_str().unwrap().to_string())

        .collect();

    assert_eq!(missing.len(), m.needed().len());

    let manifest = serde_json::to_string(&m).unwrap();

    let (st, body) = call(&h.app, "PUT", &format!("/api/v1/flow/images/{hash}"), Some("site"), Some(manifest.clone())).await;

    assert!(st == StatusCode::CONFLICT && body.contains("was not sent"), "{st} {body}");

    // A chunk is checked against its name; a runner token may not send.

    let first = &missing[0];

    assert_eq!(put_bytes(&h.app, &format!("/api/v1/flow/images/chunks/{first}"), "site", vec![1, 2, 3]).await.0, StatusCode::BAD_REQUEST);

    assert_ne!(

        put_bytes(&h.app, &format!("/api/v1/flow/images/chunks/{first}"), "runner-secret", chunks[first].clone()).await.0,

        StatusCode::NO_CONTENT

    );

    for c in &missing {

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

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

    }

    let (_, body) = call(&h.app, "POST", "/api/v1/flow/images/missing", Some("site"), Some(ask)).await;

    assert!(body.contains("\"missing\":[]"), "{body}");

    // A manifest naming another hash is refused; the right one is kept.

    let mut wrong = m.clone();

    wrong.hash = "ab".repeat(32);

    let (st, _) =

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

            .await;

    assert_eq!(st, StatusCode::CONFLICT, "chunks that make another image");

    let (st, body) = call(&h.app, "PUT", &format!("/api/v1/flow/images/{hash}"), Some("site"), Some(manifest)).await;

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

    // A runner fetches the manifest and the chunks, and the image comes back whole.

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

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

    let got: jet::images::Manifest = serde_json::from_str(&body).unwrap();

    assert_eq!(got, m);

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

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

    let mut fetched = Vec::new();

    for c in got.needed() {

        let req = Request::get(format!("/api/v1/flow/images/chunks/{c}"))

            .header(header::AUTHORIZATION, "Bearer runner-secret")

            .body(Body::empty())

            .unwrap();

        let res = oxygenhub::router(h.app.clone()).oneshot(req).await.unwrap();

        assert_eq!(res.status(), StatusCode::OK);

        let bytes = res.into_body().collect().await.unwrap().to_bytes().to_vec();

        fetched.push((c.to_string(), bytes));

    }

    jet::images::assemble(&got, &out, |c| Ok(fetched.iter().find(|(h, _)| h == c.hash()).unwrap().1.clone())).unwrap();

    assert_eq!(std::fs::read(&out).unwrap(), image);

    assert_eq!(

        call(&h.app, "GET", &format!("/api/v1/flow/images/{}", "cd".repeat(32)), Some("runner-secret"), None).await.0,

        StatusCode::NOT_FOUND

    );

    assert_ne!(call(&h.app, "GET", &format!("/api/v1/flow/images/{hash}"), None, None).await.0, StatusCode::OK, "not anonymous");

    // A job on that image: a runner holding no images and not fetching

    // is offered nothing; one that fetches is offered it.

    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",

        &format!("schema-version 1\nsandbox \"img\" {{\n    rootfs \"cas://blob/{hash}\"\n}}\njob \"build\" {{\n    sandbox \"img\"\n    exec \"make\"\n}}\n"),

    );

    c.commit("Build on the image", &[]).unwrap();

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

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

    let ask = |fetches: bool| format!(r#"{{"runner":"test-1","runtimes":["microvm"],"images":[],"fetches_images":{fetches}}}"#);

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

    assert_eq!(st, StatusCode::NO_CONTENT, "it lacks the image and does not fetch");

    oxygenhub::flow::explain_waits(&h.app).await.unwrap();

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", Some("site"), None).await;

    assert!(body.contains("does not hold image"), "{body}");

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

    assert_eq!(st, StatusCode::OK, "it fetches, and the hub holds the image: {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; 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. 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).