OXYGENHUB

Diffusion: keep a pin's own changes when moving it 7a14bdaeb857

A consumer may pin upstream changes its main lacks, such as an upstream
proposal not landed yet. A bump now moves the pin to the upstream's state
and those changes together, rather than refusing, so nothing it relies on
is dropped; it is skipped, with the reason logged, when the combination
would conflict or the pin names changes the upstream does not hold.

refs #148
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-10-13 01:43 UTC
Signature
✔ signed by Andy McCall
Hash
7a14bdaeb857c2c83237194d212adeb59a19ce96dab5517fde6c89b05f786516
Depends on
1d634f698ecd 3776b70d9b17 42fa95540468 93be9ce4653e a7e6e8e6094c cc19697f8871 f677d6270298
Depended on by
3e868a1f94e1 9aaa88b2d9c8
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, 3 files

Browse the tree at this change

modified crates/hub/src/diffusion.rs +49 −25
…
180 unchanged lines
181 181
    UpToDate,
182 182
    /// The bump bookmark already moves the pin to that state.
183 183
    Already,
184 -
    /// The upstream state does not hold everything the pin does (upstream
185 -
    /// moved backwards or was rewritten): consumers only move forward.
186 -
    Backwards,
184 +
    /// The pin names changes the upstream does not hold: not followed.
185 +
    Unknown,
186 +
    /// Taking the upstream's state with the pin's own extra changes would
187 +
    /// conflict.
188 +
    Conflicts,
187 189
    /// The consumer has no Oxygen pin of that upstream at `path` on its
188 190
    /// target bookmark.
189 191
    NoPin,
190 192
}
191 193
192 -
/// How the upstream's state `new` relates to a consumer's pin `old`, read
193 -
/// in the upstream: `Some(n)` new changes when `new` holds all of `old`
194 -
/// and more, `Some(0)` when it adds nothing, `None` when it lacks some.
195 -
pub async fn ahead_of(
196 -
    app: &Shared,
197 -
    upstream: &RepoName,
198 -
    old: Vec<ChangeHash>,
199 -
    new: Vec<ChangeHash>,
200 -
) -> Result<Option<usize>, crate::repos::RepoError> {
194 +
/// Where a consumer's pin `old` moves when the upstream reaches `new`,
195 +
/// worked out in the upstream (§4.3).
196 +
#[derive(Clone, Debug, PartialEq, Eq)]
197 +
pub enum Plan {
198 +
    /// To these heads, which add this many changes.
199 +
    To(Vec<ChangeHash>, usize),
200 +
    /// Nothing to add.
201 +
    UpToDate,
202 +
    /// The pin names changes the upstream does not hold (rewritten, or
203 +
    /// from elsewhere): not followed.
204 +
    Unknown,
205 +
    /// Taking `new` with the pin's own extra changes would conflict.
206 +
    Conflicts,
207 +
}
208 +
209 +
/// The pin `old` moved to the upstream's `new`. When the pin has changes
210 +
/// `new` lacks (say, from an upstream proposal not landed yet), they are
211 +
/// kept: changes that do not touch each other combine, so the move is to
212 +
/// both, never dropping what the consumer already relies on.
213 +
pub async fn plan(app: &Shared, upstream: &RepoName, old: Vec<ChangeHash>, new: Vec<ChangeHash>) -> Result<Plan, crate::repos::RepoError> {
201 214
    app.repos
202 215
        .read(upstream, move |r, cache| {
203 -
            let (o, n) = (cache.summary(r, &old)?, cache.summary(r, &new)?);
204 -
            if !o.changes.iter().all(|c| n.changes.contains(c)) {
205 -
                return Ok(None);
216 +
            let Ok(o) = cache.summary(r, &old) else { return Ok(Plan::Unknown) };
217 +
            let n = cache.summary(r, &new)?;
218 +
            if o.changes.iter().all(|c| n.changes.contains(c)) {
219 +
                let added = n.changes.len() - o.changes.len();
220 +
                return Ok(if added == 0 { Plan::UpToDate } else { Plan::To(new, added) });
221 +
            }
222 +
            let mut both = new.clone();
223 +
            both.extend(old.iter().copied());
224 +
            let heads = r.load_state(&both)?.heads();
225 +
            let b = cache.summary(r, &heads)?;
226 +
            if b.conflicted && !o.conflicted {
227 +
                return Ok(Plan::Conflicts);
206 228
            }
207 -
            Ok(Some(n.changes.len() - o.changes.len()))
229 +
            let added = b.changes.len() - o.changes.len();
230 +
            Ok(if added == 0 { Plan::UpToDate } else { Plan::To(heads, added) })
208 231
        })
209 232
        .await
210 233
}
…
37 unchanged lines
248 271
        })
249 272
        .await?;
250 273
    let Some(Pin::Oxy { repo, heads: old, bookmark }) = current else { return Ok(Bumped::NoPin) };
274 +
    if repo != upstream_id {
275 +
        return Ok(Bumped::NoPin);
276 +
    }
277 +
    let new_heads = match plan(app, &uname, old.clone(), new_heads.clone()).await? {
278 +
        Plan::To(heads, _) => heads,
279 +
        Plan::UpToDate => return Ok(Bumped::UpToDate),
280 +
        Plan::Unknown => return Ok(Bumped::Unknown),
281 +
        Plan::Conflicts => return Ok(Bumped::Conflicts),
282 +
    };
251 283
    if let Some(Pin::Oxy { heads: b, .. }) = &bumped {
252 284
        let (mut b, mut n) = (b.clone(), new_heads.clone());
253 285
        b.sort();
…
1 unchanged lines
255 287
        if b == n {
256 288
            return Ok(Bumped::Already);
257 289
        }
258 -
    }
259 -
    if repo != upstream_id {
260 -
        return Ok(Bumped::NoPin);
261 -
    }
262 -
    match ahead_of(app, &uname, old.clone(), new_heads.clone()).await? {
263 -
        None => return Ok(Bumped::Backwards),
264 -
        Some(0) => return Ok(Bumped::UpToDate),
265 -
        Some(_) => {}
266 290
    }
267 291
    let (key, secret) =
268 292
        crate::keys::load_secret(&app.config.keys_dir()).map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e.to_string())))?;
…
211 unchanged lines
480 504
                    upstream.name
481 505
                );
482 506
            }
483 -
            Ok(_) => {}
507 +
            Ok(other) => tracing::info!("diffusion: {}/{} {path}: no bump ({other:?})", consumer.owner, consumer.name),
484 508
            Err(e) => tracing::warn!("diffusion: bump {}/{} {path}: {e}", consumer.owner, consumer.name),
485 509
        }
486 510
    }
…
194 unchanged lines
modified crates/hub/tests/flow.rs +15 −5
…
3200 unchanged lines
3201 3201
        oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, api_heads.clone()).await.unwrap(),
3202 3202
        oxygenhub::diffusion::Bumped::Already
3203 3203
    );
3204 -
    // A state lacking what the pin has is not followed backwards.
3204 +
    // A state lacking changes the pin has (as when the consumer pinned an
3205 +
    // upstream proposal not landed yet) is combined with the pin, never
3206 +
    // dropping what it relies on.
3205 3207
    write(api_work.path(), "other.c", "v3\n");
3206 3208
    api.snapshot("t").unwrap();
3207 3209
    let v3 = api.commit("api v3, on its own", &[]).unwrap();
3208 3210
    assert!(api.push("origin", false).unwrap().iter().all(|r| r.ok));
3209 3211
    let only_v3 = api.load_state(&[v3]).unwrap().heads().to_vec();
3210 -
    assert_eq!(
3211 -
        oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, only_v3).await.unwrap(),
3212 -
        oxygenhub::diffusion::Bumped::Backwards
3213 -
    );
3212 +
    let combined = oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, only_v3).await.unwrap();
3213 +
    let oxygenhub::diffusion::Bumped::Moved { heads: ch, .. } = combined else { panic!("{combined:?}") };
3214 +
    game.fetch("origin").unwrap();
3215 +
    let st = game.load_state(&ch).unwrap().materialize();
3216 +
    let pin = match &st.files["deps/api"].content {
3217 +
        oxy_core::state::FileContent::Opaque { blobs } => oxy_core::subrepo::Pin::decode(&game.read_blob(&blobs[0]).unwrap()).unwrap(),
3218 +
        other => panic!("{other:?}"),
3219 +
    };
3220 +
    let oxy_core::subrepo::Pin::Oxy { heads: pinned, .. } = pin else { panic!() };
3221 +
    let all = api.load_state(&pinned).unwrap();
3222 +
    // The game's main still pins v1 alone; combined with v3, it is both.
3223 +
    assert!(all.contains(&v3) && all.len() == 2, "v1 and v3, {} changes", all.len());
3214 3224
    // The upstream moves again: the same bookmark moves.
3215 3225
    let newer = api.bookmark_heads().unwrap();
3216 3226
    let again = oxygenhub::diffusion::bump(&h.app, &game_row, "main", "deps/api", &api_row, newer).await.unwrap();
…
189 unchanged lines
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). **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 as Used by. 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). It is skipped when the pin already holds the state, when the bump bookmark already moves it there, or when the state lacks changes the pin has (consumers only move forward). 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. 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).
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). **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 as Used by. 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. 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).
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