OXYGENHUB

Swarm: the tracker passes on each peer's certificate fingerprint 6adcd68b90b1

A runner's claim now carries the BLAKE3 of its swarm TLS certificate; the
tracker hands it on with the peer so the fetcher pins its connection to
it, and a runner without one is not in the swarm. The OxygenFlow pin moves
to the TLS change.

refs andymccall/oxygenflow#48
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-10-12 16:09 UTC
Signature
✔ signed by Andy McCall
Hash
6adcd68b90b10957250797a6db8e73adbefd6d9a60d7f2bdead5bc5e74eb0983
Depends on
380996db84e5 42fa95540468 93be9ce4653e a46d15e72101 cc19697f8871
Depended on by
1d634f698ecd 6ca0b971ff07
In bookmarks
api-unknown-fields cancel-landed-runs close-proposal-140 diffusion-pin-bookmark draft-ready main oh-comment-body-file proposal-list-fast
In tags
v0.5.0 v0.5.1 v0.5.2 v0.6.0 v0.6.1 v0.7.0 v0.8.0 v0.9.0 v0.9.1 v0.9.11 v0.9.15 v0.9.16 v0.9.19 v0.9.2 v0.9.20 v0.9.21 v0.9.22 v0.9.23 v0.9.24 v0.9.3 v0.9.4 v0.9.5 v0.9.6 v0.9.7 v0.9.8 v0.9.9
Edits
12 atoms, 0 tree operations, 4 files

Browse the tree at this change

modified crates/hub/src/swarm.rs +7 −4
1 1
//! The swarm's tracker (oxygenhub#144, oxygenflow docs/jetstream.md §8.1):
2 2
//! runners that share images tell the hub where peers reach them with each
3 3
//! claim; a runner fetching an image announces it and is given the peers
4 -
//! in its site that hold or are fetching that image, and a ticket, signed
5 -
//! with the hub key, that those peers accept.
4 +
//! in its site that hold or are fetching that image, each with the
5 +
//! fingerprint of its TLS certificate, and a ticket, signed with the hub
6 +
//! key, that those peers accept.
6 7
//!
7 8
//! What the tracker knows lives in memory: runners repeat their address
8 9
//! with every claim and fetchers announce every `INTERVAL_S`, so a
…
34 unchanged lines
43 44
    /// in a swarm.
44 45
    pub fn claimed(&self, runner: i64, swarm: Option<&SwarmAddr>) {
45 46
        let mut addrs = self.addrs.lock().unwrap_or_else(|p| p.into_inner());
46 -
        match swarm.filter(|s| plausible(&s.addr)) {
47 +
        // A peer is trusted by its certificate, so one without is not in
48 +
        // the swarm (an older flow).
49 +
        match swarm.filter(|s| plausible(&s.addr) && jet::images::is_hash(&s.cert)) {
47 50
            Some(s) => {
48 51
                addrs.insert(runner, (s.clone(), db::now_ms()));
49 52
            }
…
59 unchanged lines
109 112
            let Some(addr) = app.swarm.addr(r.id).filter(|s| s.site == mine.site) else { continue };
110 113
            let complete = r.image_list().is_some_and(|held| held.iter().any(|h| h.eq_ignore_ascii_case(&image)));
111 114
            if complete || fetchers.contains(&r.id) {
112 -
                peers.push(SwarmPeer { runner: r.id, name: r.name.clone(), addr: addr.addr, complete });
115 +
                peers.push(SwarmPeer { runner: r.id, name: r.name.clone(), addr: addr.addr, cert: addr.cert, complete });
113 116
            }
114 117
        }
115 118
    }
…
38 unchanged lines
modified crates/hub/tests/flow.rs +31 −4
…
2977 unchanged lines
2978 2978
        async move { call(app, "POST", "/api/v1/flow/swarm/announce", Some(token), Some(body)).await }
2979 2979
    };
2980 2980
    // A holds the image; B and C hold nothing; all three are in site "lab".
2981 -
    claim_as("runner-secret", "a", serde_json::json!({ "images": [image], "swarm": { "addr": "10.0.0.1:7401", "site": "lab" } })).await;
2982 -
    claim_as("runner-two", "b", serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.2:7401", "site": "lab" } })).await;
2983 -
    claim_as("runner-three", "c", serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "lab" } })).await;
2981 +
    claim_as(
2982 +
        "runner-secret",
2983 +
        "a",
2984 +
        serde_json::json!({ "images": [image], "swarm": { "addr": "10.0.0.1:7401", "site": "lab", "cert": "11".repeat(32) } }),
2985 +
    )
2986 +
    .await;
2987 +
    claim_as(
2988 +
        "runner-two",
2989 +
        "b",
2990 +
        serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.2:7401", "site": "lab", "cert": "22".repeat(32) } }),
2991 +
    )
2992 +
    .await;
2993 +
    claim_as(
2994 +
        "runner-three",
2995 +
        "c",
2996 +
        serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "lab", "cert": "33".repeat(32) } }),
2997 +
    )
2998 +
    .await;
2984 2999
2985 3000
    let (st, body) = announce("runner-two", "fetching").await;
2986 3001
    assert_eq!(st, StatusCode::OK, "{body}");
2987 3002
    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();
2988 3003
    let names: Vec<(&str, bool)> = a.peers.iter().map(|p| (p.addr.as_str(), p.complete)).collect();
2989 3004
    assert_eq!(names, vec![("10.0.0.1:7401", true)], "only A holds it so far");
3005 +
    assert_eq!(a.peers[0].cert, "11".repeat(32), "with the fingerprint peers check it by");
2990 3006
    let t = jet::swarm::Ticket::open(&a.ticket, &app.hub_key.identity, oxygenhub::db::now_ms() as u64).unwrap();
2991 3007
    assert_eq!((t.name.as_str(), t.site.as_str()), ("b", "lab"));
2992 3008
…
6 unchanged lines
2999 3015
3000 3016
    // B is done: no longer offered as fetching. C moves site: no peers.
3001 3017
    announce("runner-two", "done").await;
3002 -
    claim_as("runner-three", "c", serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "elsewhere" } })).await;
3018 +
    claim_as(
3019 +
        "runner-three",
3020 +
        "c",
3021 +
        serde_json::json!({ "images": [], "swarm": { "addr": "10.0.0.3:7401", "site": "elsewhere", "cert": "33".repeat(32) } }),
3022 +
    )
3023 +
    .await;
3003 3024
    let (_, body) = announce("runner-three", "fetching").await;
3004 3025
    let a: flow_core::api::Announced = serde_json::from_str(&body).unwrap();
3005 3026
    assert!(a.peers.is_empty(), "{:?}", a.peers);
3027 +
3028 +
    // A runner without a certificate (an older flow) is not in the swarm.
3029 +
    claim_as("runner-secret", "a", serde_json::json!({ "images": [image], "swarm": { "addr": "10.0.0.1:7401", "site": "elsewhere" } }))
3030 +
        .await;
3031 +
    let (st, _) = announce("runner-secret", "fetching").await;
3032 +
    assert_eq!(st, StatusCode::CONFLICT);
3006 3033
3007 3034
    // A runner that left the swarm is told how to join; nonsense is refused.
3008 3035
    claim_as("runner-two", "b", serde_json::json!({ "images": [] })).await;
…
17 unchanged lines
replaced deps/oxygenflow +0 −0

No line changes.

modified docs/spec.md +1 −1
…
514 unchanged lines
515 515
  - its key is the one the runner's token first signed with. The first signed receipt under a token binds that token to its key in `flow_runner_keys`.
516 516
517 517
  Otherwise the job's result still stands, and the run page says why the receipt was not kept. The run page links each kept receipt, with its signer. The change page lists the receipts for the state whose only head is that change. `/{owner}/{name}/receipts/{attest}` downloads a signed receipt for `oxy receipt verify` or `import`.
518 -
- **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).
518 +
- **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, 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).
519 519
- **Variables** (oxygenhub#102): Administration, an organisation's settings and a repository's settings each hold `NAME=value` pairs (a name is a letter or `_` then letters, digits and `_`, up to 64, never `FLOW_*`; a value up to 4096 characters; up to 100 per level) under **Variables**, each page showing what it inherits and which inherited names it overrides. A claim carries the repository's effective set as `env`, inner levels overriding outer; Flow puts them in the job's environment under `FLOW_*` and the pipeline's own `env`, so they are part of the receipt. They are not secrets and are shown in full to whoever sees the page; secrets are mounted by Flow (OxygenFlow spec §9), never put in the environment. Changes are audited (`flow.variable.set`, `flow.variable.deleted`); `flow run --var NAME=VALUE` gives a local run the same.
520 520
- **Secrets** (oxygenhub#107; OxygenFlow spec §9): the same three levels hold **Secrets**, named as a pipeline's `secret "name" { mount-path "/run/secrets/…" }` names them (1 to 64 of `A-Z a-z 0-9 . _ -`, a value up to 64 KiB, up to 100 per level), typed once and sealed under `keys/secrets.key`; pages show names, when each was set and where it comes from, never a value. A claim carries only the secrets its job declares, each with its path and value, from the innermost level that sets the name; one no level sets is noted at the top of the job's log and left out, so the job fails on the missing file rather than running with a stand-in. Flow writes each as a file (0400, the job's user's, on an in-memory mount) before the job starts and never puts it in the environment, the receipt or an artifact. Log intake replaces each value, and its base64, with `***` before a record is stored. Changes are audited (`flow.secret.set`, `flow.secret.deleted`), never with the value.
521 521
- **Artifacts** (OxygenFlow spec §5.2): a job may declare artifacts with `output`.
…
231 unchanged lines