Peer-held images: the hub keeps runners' manifests of images it lacks and places their jobs where a peer in the site holds them (closes #342; pins OxygenFlow #94) 951aed50c2f5

-- Images only runners hold (oxygenflow#94, oxygenhub#342): a runner sends the

-- manifest of an image it holds that the hub keeps none of, and the swarm

-- moves the chunks between runners. `source` is 'hub' for an image pushed

-- here and verified whole, 'peer' for one only runners hold; `holder` names

-- the runner that sent a peer manifest.

ALTER TABLE jet_images ADD COLUMN source TEXT NOT NULL DEFAULT 'hub';

ALTER TABLE jet_images ADD COLUMN holder TEXT NOT NULL DEFAULT '';

-- Images only runners hold (oxygenflow#94, oxygenhub#342): a runner sends the

-- manifest of an image it holds that the hub keeps none of, and the swarm

-- moves the chunks between runners. `source` is 'hub' for an image pushed

-- here and verified whole, 'peer' for one only runners hold; `holder` names

-- the runner that sent a peer manifest.

ALTER TABLE jet_images ADD COLUMN source TEXT NOT NULL DEFAULT 'hub';

ALTER TABLE jet_images ADD COLUMN holder TEXT NOT NULL DEFAULT '';

- **Images only runners hold (#342, oxygenflow#94).** A runner that serves its peers sends the manifest of each image it holds that the hub lacks (`POST /flow/images/held`, `PUT /flow/images/{hash}/held`, runner tokens only); the hub keeps it as held by runners and offers the image's jobs to runners of the same swarm site, which fetch it from their peers (the announce says the hub is no seed for it). A pushed image replaces such a record. On 2026-10-25 a re-pinned image that was never pushed kept every Rust job on one runner for a day. Migration 0098 (`jet_images.source`, `holder`).

/// Most images one `held` question may name.

const MAX_HELD: usize = 1000;

/// `POST /flow/images/held` (a runner's token): the images it holds; the

/// answer names those the hub keeps no manifest of, for the runner to send

/// with `PUT /flow/images/{hash}/held` (oxygenflow#94).

pub async fn held(

    State(app): State<Shared>,

    headers: HeaderMap,

    Json(ask): Json<flow_core::api::HeldImages>,

) -> Result<Json<flow_core::api::WantedManifests>, ApiError> {

    crate::flow::runner_auth(&app, &headers).await?;

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

    let mut wanted: Vec<String> =

        ask.images.into_iter().take(MAX_HELD).map(|h| h.to_ascii_lowercase()).filter(|h| is_hash(h) && !kept.contains(h)).collect();

    wanted.sort();

    wanted.dedup();

    Ok(Json(flow_core::api::WantedManifests { wanted }))

}

/// `PUT /flow/images/{hash}/held` (a runner's token): the manifest of an

/// image the runner holds and the hub does not. Kept as held by runners

/// only: the hub has none of its chunks to check it against, and a fetcher

/// checks what it assembles against the pinned hash (jetstream.md §8.1.1).

pub async fn put_held(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(hash): Path<String>,

    Json(m): Json<Manifest>,

) -> Result<StatusCode, ApiError> {

    let runner = crate::flow::runner_auth(&app, &headers).await?;

    if !is_hash(&hash) || m.hash != hash {

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

    }

    if crate::images::held(&app.db).await?.contains(&hash) {

        return Ok(StatusCode::NO_CONTENT);

    }

    let store = app.images();

    let kept = m.clone();

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

        .await

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

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

    crate::images::record_peer(&app.db, &m, &runner.name).await?;

    Ok(StatusCode::CREATED)

}

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

        // A runner's manifest of an image only runners hold (oxygenflow#94);

        // a 14 GB image's is about 11 MB.

        .route("/flow/images/{hash}/held", axum::routing::put(images::put_held).layer(axum::extract::DefaultBodyLimit::max(64 << 20)))

        // The hub's own images, and those a runner of its site holds (oxygenflow#94).

        fetchable: if req.fetches_images { crate::images::reach(&app).await?.for_runner(me.id) } else { Default::default() },

    let (archs, reach) = (fdb::image_archs(&app.db).await?, crate::images::reach(&app).await?);

    let fleet = Fleet { image_archs: &archs, reach: &reach, now };

    /// Where images can come from: the hub, or a peer in a runner's site.

    pub reach: &'a crate::images::Reach,

    let fetches = r.fetches_images && fleet.reach.fetchable(r.id, rootfs);

        let why = if !r.fetches_images {

            ""

        } else if fleet.reach.peer_only(rootfs) {

            " (only runners hold it, none in this runner's swarm site: send it to the hub with flow rootfs push)"

        } else {

            " (nor does the hub: send it with flow rootfs push)"

        };

    reach: &crate::images::Reach,

    let fleet = Fleet { image_archs, reach, now };

    fleet: (&std::collections::HashMap<String, String>, &crate::images::Reach, i64),

    let fleet = Fleet { image_archs: fleet.0, reach: fleet.1, now: fleet.2 };

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

    /// Keep manifest `m` of an image only runners hold (oxygenflow#94): well

    /// formed, its chunks not here, so not checked against its hash; a

    /// fetcher checks the image it assembles against the pinned hash.

    /// Never over a manifest already kept.

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

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

        if self.manifest_path(&m.hash).exists() {

            return Ok(());

        }

        self.keep_manifest(m)

    }

/// Record an image this hub now keeps, its chunks and all: one only

/// runners held before becomes the hub's.

         ON CONFLICT (hash) DO UPDATE SET source = 'hub', holder = '', \

         arch = CASE WHEN excluded.arch != '' THEN excluded.arch ELSE jet_images.arch END",

}

/// Record an image only runners hold, whose manifest `holder` sent

/// (oxygenflow#94). Never over one the hub keeps.

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

    sqlx::query(

        "INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms, source, holder) VALUES ($1, $2, $3, $4, $5, $6, 'peer', $7) \

         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(holder)

    .bind(now_ms())

    .bind(holder)

    .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 a manifest of, by hash: its own and those

/// only runners hold.

/// Where the images jobs need can come from (oxygenflow#94): the hub's

/// own, pushed and verified, and those only runners hold, each with the

/// swarm sites of the online runners holding it now.

#[derive(Debug, Default)]

pub struct Reach {

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

    pub peers: std::collections::HashMap<String, Vec<String>>,

    /// Each runner in a swarm now, and its site.

    pub sites: std::collections::HashMap<i64, String>,

}

impl Reach {

    /// Whether a runner that fetches images can get `image`: from the hub,

    /// or from a runner of its own site.

    pub fn fetchable(&self, runner: i64, image: &str) -> bool {

        self.hub.contains(image) || self.sites.get(&runner).is_some_and(|s| self.peers.get(image).is_some_and(|v| v.contains(s)))

    }

    /// Every image `runner` can fetch.

    pub fn for_runner(&self, runner: i64) -> std::collections::HashSet<String> {

        let mut out = self.hub.clone();

        out.extend(self.peers.keys().filter(|i| self.fetchable(runner, i)).cloned());

        out

    }

    /// Whether only runners hold `image`.

    pub fn peer_only(&self, image: &str) -> bool {

        !self.hub.contains(image) && self.peers.contains_key(image)

    }

}

/// [`Reach`] now.

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

    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, source FROM jet_images").fetch_all(&app.db).await?;

    let mut reach = Reach::default();

    for (hash, source) in rows {

        if source == "peer" {

            reach.peers.insert(hash, Vec::new());

        } else {

            reach.hub.insert(hash);

        }

    }

    for r in crate::db::runners::list(&app.db).await? {

        if r.state == "disabled" || !r.online() {

            continue;

        }

        let Some(addr) = app.swarm.addr(r.id) else { continue };

        for image in r.image_list().unwrap_or_default() {

            if let Some(sites) = reach.peers.get_mut(&image.to_ascii_lowercase()) {

                if !sites.contains(&addr.site) {

                    sites.push(addr.site.clone());

                }

            }

        }

        reach.sites.insert(r.id, addr.site);

    }

    Ok(reach)

}

    /// 'hub', or 'peer' for one only runners hold (oxygenflow#94).

    pub source: String,

    pub holder: String,

    sqlx::query_as("SELECT hash, size, arch, chunks, uploader, created_ms, source, holder FROM jet_images").fetch_all(pool).await

}

#[cfg(test)]

mod reach_tests {

    use super::Reach;

    #[test]

    fn an_image_only_runners_hold_is_fetchable_in_their_site_alone() {

        let mut r = Reach::default();

        r.hub.insert("hub-image".into());

        r.peers.insert("peer-image".into(), vec!["home".into()]);

        r.sites.insert(1, "home".into());

        r.sites.insert(2, "cloud".into());

        assert!(r.fetchable(1, "hub-image") && r.fetchable(2, "hub-image") && r.fetchable(3, "hub-image"));

        assert!(r.fetchable(1, "peer-image"), "a runner of the holder's site");

        assert!(!r.fetchable(2, "peer-image"), "another site");

        assert!(!r.fetchable(3, "peer-image"), "a runner in no swarm");

        assert!(r.peer_only("peer-image") && !r.peer_only("hub-image"));

        assert_eq!(r.for_runner(1).len(), 2);

        assert_eq!(r.for_runner(2).len(), 1);

    }

    // An image only runners hold (oxygenflow#94): the hub has its manifest

    // and none of its chunks, so the fetcher asks peers alone.

    let seed = Some(

        sqlx::query_scalar::<_, String>("SELECT source FROM jet_images WHERE hash = $1")

            .bind(&image)

            .fetch_optional(&app.db)

            .await?

            .is_none_or(|s| s != "peer"),

    );

    Ok(Json(Announced { ticket, peers, interval_s: INTERVAL_S, seed }))

/// oxygenflow#94, #342: a runner sends the manifest of an image it holds

/// that the hub lacks; the hub keeps it as held by runners only (no chunks,

/// announce says it is no seed) until the image is pushed whole.

async fn a_runner_offers_the_manifest_of_an_image_the_hub_lacks() {

    let h = hub().await;

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

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

    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();

    // Asked which of its images the hub lacks: this one.

    let held = serde_json::json!({ "images": [hash.clone(), "not-a-hash"] }).to_string();

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

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

    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["wanted"], serde_json::json!([hash.clone()]));

    // Only runners may say so.

    assert_ne!(call(&h.app, "POST", "/api/v1/flow/images/held", Some("site"), Some(held.clone())).await.0, StatusCode::OK);

    // A manifest filed under another hash is refused; its own is kept as peer-held.

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

    let wrong = format!("/api/v1/flow/images/{}/held", "ab".repeat(32));

    assert_eq!(call(&h.app, "PUT", &wrong, Some("runner-secret"), Some(manifest.clone())).await.0, StatusCode::BAD_REQUEST);

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

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

    let kept = oxygenhub::images::kept(&h.app.db).await.unwrap();

    let row = kept.iter().find(|k| k.hash == hash).expect("recorded");

    assert_eq!(row.source, "peer");

    let (_, body) = call(&h.app, "POST", "/api/v1/flow/images/held", Some("runner-secret"), Some(held)).await;

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

    // Runners fetch its manifest; the hub has none of its chunks.

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

    assert_eq!(st, StatusCode::OK);

    let first = m.needed()[0].to_string();

    assert_ne!(call(&h.app, "GET", &format!("/api/v1/flow/images/chunks/{first}"), Some("runner-secret"), None).await.0, StatusCode::OK);

    let reach = oxygenhub::images::reach(&h.app).await.unwrap();

    assert!(reach.peer_only(&hash));

    // Pushed whole, it becomes the hub's.

    for c in m.needed() {

        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 (st, body) = call(&h.app, "PUT", &format!("/api/v1/flow/images/{hash}"), Some("site"), Some(manifest)).await;

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

    let kept = oxygenhub::images::kept(&h.app.db).await.unwrap();

    assert_eq!(kept.iter().find(|k| k.hash == hash).unwrap().source, "hub");

    assert!(!oxygenhub::images::reach(&h.app).await.unwrap().peer_only(&hash));

}

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