Jetstream: the swarm's tracker and peer tickets a46d15e72101

//! The swarm's tracker (oxygenhub#144, oxygenflow docs/jetstream.md §8.1):

//! runners that share images tell the hub where peers reach them with each

//! claim; a runner fetching an image announces it and is given the peers

//! in its site that hold or are fetching that image, and a ticket, signed

//! with the hub key, that those peers accept.

//!

//! What the tracker knows lives in memory: runners repeat their address

//! with every claim and fetchers announce every `INTERVAL_S`, so a

//! restarted hub knows the swarm again within seconds.

use crate::api::ApiError;

use crate::db;

use crate::Shared;

use axum::extract::State;

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

use axum::Json;

use flow_core::api::{Announce, Announced, SwarmAddr, SwarmPeer};

use jet::swarm::{Ticket, TICKET_LIFE_MS};

use std::collections::HashMap;

use std::sync::Mutex;

/// How often a fetcher announces again.

pub const INTERVAL_S: u64 = 15;

/// A runner whose address is older than this is not offered as a peer.

const ADDR_FRESH_MS: i64 = 2 * 60 * 1000;

/// A fetcher that has not announced for this long is no longer fetching.

const FETCHING_FRESH_MS: i64 = 3 * INTERVAL_S as i64 * 1000;

/// A ticket is made again once it has less than this to run.

const TICKET_RENEW_MS: u64 = 5 * 60 * 1000;

#[derive(Default)]

pub struct Tracker {

    /// Runner id to its swarm address and when it last said so.

    addrs: Mutex<HashMap<i64, (SwarmAddr, i64)>>,

    /// Image to the runners fetching it, with when each last announced.

    fetching: Mutex<HashMap<String, HashMap<i64, i64>>>,

    /// Runner id to its current ticket and when that expires.

    tickets: Mutex<HashMap<i64, (String, u64)>>,

}

impl Tracker {

    /// What a runner's claim said: where peers reach it, or that it is not

    /// in a swarm.

    pub fn claimed(&self, runner: i64, swarm: Option<&SwarmAddr>) {

        let mut addrs = self.addrs.lock().unwrap_or_else(|p| p.into_inner());

        match swarm.filter(|s| plausible(&s.addr)) {

            Some(s) => {

                addrs.insert(runner, (s.clone(), db::now_ms()));

            }

            None => {

                addrs.remove(&runner);

            }

        }

    }

    /// Where peers reach `runner`, if it is in a swarm now.

    pub fn addr(&self, runner: i64) -> Option<SwarmAddr> {

        let addrs = self.addrs.lock().unwrap_or_else(|p| p.into_inner());

        addrs.get(&runner).filter(|(_, seen)| db::now_ms() - seen < ADDR_FRESH_MS).map(|(a, _)| a.clone())

    }

    fn fetch(&self, image: &str, runner: i64, state: &str) {

        let mut fetching = self.fetching.lock().unwrap_or_else(|p| p.into_inner());

        let now = db::now_ms();

        for who in fetching.values_mut() {

            who.retain(|_, seen| now - *seen < FETCHING_FRESH_MS);

        }

        fetching.retain(|_, who| !who.is_empty());

        if state == "fetching" {

            fetching.entry(image.to_string()).or_default().insert(runner, now);

        } else if let Some(who) = fetching.get_mut(image) {

            who.remove(&runner);

        }

    }

    fn fetchers(&self, image: &str) -> Vec<i64> {

        let fetching = self.fetching.lock().unwrap_or_else(|p| p.into_inner());

        fetching.get(image).map(|who| who.keys().copied().collect()).unwrap_or_default()

    }

}

/// An address a peer could dial: `host:port`, nothing else.

fn plausible(addr: &str) -> bool {

    addr.len() <= 255 && addr.rsplit_once(':').is_some_and(|(host, port)| !host.is_empty() && port.parse::<u16>().is_ok_and(|p| p > 0))

}

/// `POST /flow/swarm/announce`: a runner starting, continuing or ending a

/// fetch. The peers it may ask, in its site, and its ticket.

pub async fn announce(State(app): State<Shared>, headers: HeaderMap, Json(a): Json<Announce>) -> Result<Json<Announced>, ApiError> {

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

    if !jet::images::is_hash(&a.image) || !matches!(a.state.as_str(), "fetching" | "done" | "stopped") {

        return Err(ApiError(StatusCode::BAD_REQUEST, "announce an image's hash with state fetching, done or stopped".into()));

    }

    let image = a.image.to_ascii_lowercase();

    let Some(mine) = app.swarm.addr(me.id) else {

        return Err(ApiError(

            StatusCode::CONFLICT,

            "this runner has not said where peers reach it: give it a [runner.swarm] section and let it claim".into(),

        ));

    };

    app.swarm.fetch(&image, me.id, &a.state);

    let fetchers = app.swarm.fetchers(&image);

    let mut peers = Vec::new();

    if a.state == "fetching" {

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

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

                continue;

            }

            let Some(addr) = app.swarm.addr(r.id).filter(|s| s.site == mine.site) else { continue };

            let complete = r.image_list().is_some_and(|held| held.iter().any(|h| h.eq_ignore_ascii_case(&image)));

            if complete || fetchers.contains(&r.id) {

                peers.push(SwarmPeer { runner: r.id, name: r.name.clone(), addr: addr.addr, complete });

            }

        }

    }

    let ticket = ticket(&app, &me, &mine.site)?;

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

}

/// The runner's ticket, made again when it nears its end or its site

/// changed.

fn ticket(app: &Shared, me: &db::runners::Runner, site: &str) -> Result<String, ApiError> {

    let now = db::now_ms() as u64;

    let mut tickets = app.swarm.tickets.lock().unwrap_or_else(|p| p.into_inner());

    if let Some((t, until)) = tickets.get(&me.id) {

        if *until > now + TICKET_RENEW_MS && Ticket::open(t, &app.hub_key.identity, now).is_ok_and(|t| t.site == site) {

            return Ok(t.clone());

        }

    }

    let (key, secret) = crate::keys::load_secret(&app.config.keys_dir())

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

    let expires_ms = now + TICKET_LIFE_MS;

    let t = Ticket { hub: app.hub_key.key.to_hex(), runner: me.id, name: me.name.clone(), site: site.to_string(), expires_ms }

        .seal(&key, &secret);

    tickets.insert(me.id, (t.clone(), expires_ms));

    Ok(t)

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn only_host_and_port_addresses_are_kept() {

        assert!(plausible("10.0.0.5:7401"));

        assert!(plausible("[::1]:7401"));

        assert!(plausible("runner.lan:7401"));

        assert!(!plausible("10.0.0.5"));

        assert!(!plausible(":7401"));

        assert!(!plausible("10.0.0.5:0"));

        assert!(!plausible("10.0.0.5:99999"));

    }

}

        .route("/flow/swarm/announce", post(crate::swarm::announce))

    app.swarm.claimed(me.id, req.swarm.as_ref());

pub mod swarm;

    /// The image swarm's tracker (oxygenhub#144).

    pub swarm: swarm::Tracker,

        swarm: Default::default(),

    /// Where its peers reach it in the image swarm, and its site

    /// (oxygenhub#144); empty when it is not in one.

    pub swarm: String,

        let swarm = app

            .swarm

            .addr(r.id)

            .map(|s| if s.site.is_empty() { s.addr } else { format!("{} · site {}", s.addr, s.site) })

            .unwrap_or_default();

        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, swarm, r });

/// The swarm's tracker (#144): a fetching runner is given the peers in its

/// site that hold the image or are fetching it, and a ticket signed with

/// the hub key; a runner not in a swarm is told how to join one.

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

async fn the_tracker_names_peers_in_the_same_site_and_signs_a_ticket() {

    let h = hub().await;

    let app = &h.app;

    let image = "ab".repeat(32);

    let claim_as = |token: &'static str, name: &'static str, body: serde_json::Value| async move {

        let mut b = serde_json::json!({ "runner": name, "runtimes": ["microvm"] });

        b.as_object_mut().unwrap().extend(body.as_object().unwrap().clone());

        let (st, out) = call(app, "POST", "/api/v1/flow/claim", Some(token), Some(b.to_string())).await;

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

    };

    let announce = |token: &'static str, state: &'static str| {

        let body = serde_json::json!({ "image": image, "state": state }).to_string();

        async move { call(app, "POST", "/api/v1/flow/swarm/announce", Some(token), Some(body)).await }

    };

    // A holds the image; B and C hold nothing; all three are in site "lab".

    claim_as("runner-secret", "a", serde_json::json!({ "images": [image], "swarm": { "addr": "10.0.0.1:7401", "site": "lab" } })).await;

    claim_as("runner-two", "b", serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.2:7401", "site": "lab" } })).await;

    claim_as("runner-three", "c", serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "lab" } })).await;

    let (st, body) = announce("runner-two", "fetching").await;

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

    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();

    let names: Vec<(&str, bool)> = a.peers.iter().map(|p| (p.addr.as_str(), p.complete)).collect();

    assert_eq!(names, vec![("10.0.0.1:7401", true)], "only A holds it so far");

    let t = jet::swarm::Ticket::open(&a.ticket, &app.hub_key.identity, oxygenhub::db::now_ms() as u64).unwrap();

    assert_eq!((t.name.as_str(), t.site.as_str()), ("b", "lab"));

    // C, fetching too, is told of A (whole) and B (fetching).

    let (_, body) = announce("runner-three", "fetching").await;

    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();

    let mut peers: Vec<(&str, bool)> = a.peers.iter().map(|p| (p.addr.as_str(), p.complete)).collect();

    peers.sort();

    assert_eq!(peers, vec![("10.0.0.1:7401", true), ("10.0.0.2:7401", false)]);

    // B is done: no longer offered as fetching. C moves site: no peers.

    announce("runner-two", "done").await;

    claim_as("runner-three", "c", serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "elsewhere" } })).await;

    let (_, body) = announce("runner-three", "fetching").await;

    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();

    assert!(a.peers.is_empty(), "{:?}", a.peers);

    // A runner that left the swarm is told how to join; nonsense is refused.

    claim_as("runner-two", "b", serde_json::json!({ "images": [] })).await;

    let (st, body) = announce("runner-two", "fetching").await;

    assert_eq!(st, StatusCode::CONFLICT);

    assert!(body.contains("[runner.swarm]"), "{body}");

    let (st, _) =

        call(app, "POST", "/api/v1/flow/swarm/announce", Some("runner-secret"), Some(r#"{"image":"nope","state":"fetching"}"#.into()))

            .await;

    assert_eq!(st, StatusCode::BAD_REQUEST);

    let (st, _) = call(

        app,

        "POST",

        "/api/v1/flow/swarm/announce",

        None,

        Some(serde_json::json!({ "image": image, "state": "fetching" }).to_string()),

    )

    .await;

    assert_eq!(st, StatusCode::UNAUTHORIZED);

}

- **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. 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 image while the claimant would fetch it, or 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 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). **The swarm** (#144, oxygenflow docs/jetstream.md §8.1): a runner in an image swarm reports its peer address and site with each claim (`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, 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).

    <td class="small">{% if r.arch.is_empty() %}<span class="mute">unknown</span>{% else %}{{ r.arch }} · {{ r.os }} · flow {{ r.version }}{% endif %}{% let f = r.facts_line() %}{% if !f.is_empty() %}<br><span class="mute" title="Measured by the runner at its last claim{% if !r.kernel.is_empty() %}; kernel {{ r.kernel }}{% endif %}">{{ f }}</span>{% endif %}{% if !row.swarm.is_empty() %}<br><span class="mute" title="Where other runners fetch image chunks from it">swarm {{ row.swarm }}</span>{% endif %}</td>