Diffusion: the proposal and its trigger a7e6e8e6094c

## Unreleased

- **Diffusion (#148, docs/diffusion.md).** When a repository that others pin as a subrepo moves on, each consumer that follows it gets a proposal moving the pin, signed by the hub, tested by the consumer's own pipeline, with the upstream's changes listed. One proposal per consumer and subrepo, kept up to date; never landed by itself. Turned on per repository (Settings › Diffusion, `PUT /repos/{owner}/{name}/diffusion`, or `follow` in `.oxymodules`), off by default; `green` waits for the upstream's run to pass. A repository's page shows who uses it and how far behind each is (Used by, `GET /repos/{owner}/{name}/consumers`). Consumers are bumped on the bookmark their work lands on, learned from where their proposals land.

- **Admin › Runners › Images (#146):** every runner image, where it is and what uses it.

- **Faster pages after a restart (#145):** bookmark summaries are kept on disk; slow requests are logged.

    Moved { bookmark: String, change: ChangeHash, heads: Vec<ChangeHash>, from: Vec<ChangeHash> },

    let old = old.clone();

            Ok(Bumped::Moved { bookmark, change: hash, heads, from: old })

/// The account Diffusion's proposals are opened as.

pub const ACCOUNT: &str = "diffusion";

/// The `diffusion` account, made on first use with no way to sign in.

pub async fn account(pool: &Pool) -> Option<Account> {

    if let Ok(Some(a)) = db::accounts::account_by_name(pool, ACCOUNT).await {

        return Some(a);

    }

    db::accounts::create_account(pool, ACCOUNT, "diffusion@localhost", "!", "", false).await.ok()

}

/// The upstream changes a bump brings: in `new` and not in `old`, newest

/// first, as Markdown lines linking to the upstream's change pages.

pub async fn changelog(app: &Shared, upstream: &RepoRow, old: Vec<ChangeHash>, new: Vec<ChangeHash>) -> Vec<String> {

    let Some(name) = RepoName::new(&upstream.owner, &upstream.name) else { return Vec::new() };

    let linked: std::collections::HashMap<String, String> =

        db::accounts::identity_accounts(&app.db).await.unwrap_or_default().into_iter().collect();

    let base = format!("/{}/{}", upstream.owner, upstream.name);

    app.repos

        .read(&name, move |r, cache| {

            let (o, n) = (cache.summary(r, &old)?, cache.summary(r, &new)?);

            let mut added: Vec<(u64, ChangeHash, String, String)> = Vec::new();

            for h in n.changes.iter().filter(|c| !o.changes.contains(*c)) {

                let Ok(c) = r.store().get_change(h) else { continue };

                let key = c.author.to_hex();

                let who = linked.get(&key).cloned().unwrap_or_else(|| key[..12].to_string());

                let first = c.message.lines().next().unwrap_or("").trim().to_string();

                added.push((c.hlc.wall_ms, *h, first, who));

            }

            added.sort_by_key(|a| std::cmp::Reverse(a.0));

            Ok(added

                .into_iter()

                .map(|(_, h, msg, who)| {

                    format!("- [`{}`]({base}/change/{}) {} ({who})", &h.to_hex()[..12], h.to_hex(), msg.replace('\n', " "))

                })

                .collect())

        })

        .await

        .unwrap_or_default()

}

/// Open the bump proposal for `bookmark`, or bring the open one up to

/// date: from `bookmark` into `target`, by the `diffusion` account,

/// labelled `dependencies`, its body the changelog. Its number.

pub async fn propose(

    app: &Shared,

    consumer: &RepoRow,

    target: &str,

    path: &str,

    upstream: &RepoRow,

    from: Vec<ChangeHash>,

    to: Vec<ChangeHash>,

) -> Option<i64> {

    use crate::db::proposals as pdb;

    let bookmark = bump_bookmark(path);

    let bookmark = bookmark.as_str();

    let me = account(&app.db).await?;

    let lines = changelog(app, upstream, from, to.clone()).await;

    let slug = format!("{}/{}", upstream.owner, upstream.name);

    let short: Vec<String> = to.iter().map(|h| h.to_hex()[..12].to_string()).collect();

    let title = format!(

        "Diffusion: {path} to {slug} {} ({})",

        short.join("+"),

        if lines.len() == 1 { "1 change".into() } else { format!("{} changes", lines.len()) }

    );

    let body = format!(

        "Moves the subrepo `{path}` to the newest state of [{slug}](/{slug}) that {}. Only the pin changes; this repository's own checks test it.\n\n**What it brings** ({} change{}):\n\n{}\n\nOpened and kept up to date by Diffusion (docs/diffusion.md). It never lands by itself.",

        if Follow::parse(&consumer.diffusion) == Some(Follow::Any) { "moved" } else { "passed its checks" },

        lines.len(),

        if lines.len() == 1 { "" } else { "s" },

        lines.join("\n")

    );

    let open = pdb::list(&app.db, consumer.id, &["open", "draft"]).await.ok()?.into_iter().find(|p| p.source_bookmark == bookmark);

    let number = match open {

        Some(p) => {

            pdb::edit(&app.db, p.id, &title, &body, target).await.ok()?;

            pdb::add_event(&app.db, p.id, ACCOUNT, "updated", &short.join("+")).await.ok();

            p.number

        }

        None => {

            let p = pdb::create(

                &app.db,

                pdb::NewProposal {

                    repo_id: consumer.id,

                    title: &title,

                    body: &body,

                    author_id: Some(me.id),

                    author: ACCOUNT,

                    draft: false,

                    target,

                    source_kind: "bookmark",

                    source_repo_id: None,

                    source_bookmark: bookmark,

                    source_changes: "",

                },

            )

            .await

            .ok()?;

            pdb::add_event(&app.db, p.id, ACCOUNT, "opened", "").await.ok();

            db::notify::index(&app.db, "proposal", consumer.id, p.id, p.number, &p.title, &p.body).await.ok();

            if let Ok(label) =

                db::issues::upsert_label(&app.db, consumer.id, "dependencies", "#0072ff", "A subrepo moves (Diffusion)").await

            {

                db::issues::set_proposal_label(&app.db, p.id, label.id, true).await.ok();

            }

            p.number

        }

    };

    Some(number)

}

/// The text of `.oxymodules` on `bookmark` of `name`, if any.

async fn oxymodules(app: &Shared, name: &RepoName, bookmark: &str) -> Option<String> {

    let b = bookmark.to_string();

    app.repos

        .read(name, move |r, cache| {

            let heads = heads_of(r, &b)?;

            if heads.is_empty() {

                return Ok(None);

            }

            let loaded = cache.load(r, &heads)?;

            Ok(match loaded.snapshot.files.get(".oxymodules").map(|f| &f.content) {

                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),

                _ => None,

            })

        })

        .await

        .ok()

        .flatten()

}

/// The upstream `upstream` reached `heads` on its default bookmark: by a

/// push (`green` false) or a passing run of that state (`green` true).

/// Bump and propose in every consumer that follows it that way (§4.3).

pub async fn propagate(app: &Shared, upstream: &RepoRow, heads: Vec<ChangeHash>, green: bool) {

    let Some(uname) = RepoName::new(&upstream.owner, &upstream.name) else { return };

    let Ok(id) = app.repos.read(&uname, |r, _| Ok(r.id()?)).await else { return };

    let rows: Vec<(i64, String, String)> = sqlx::query_as("SELECT consumer_id, bookmark, path FROM diffusion_pins WHERE upstream = ?")

        .bind(hex::encode(id))

        .fetch_all(&app.db)

        .await

        .unwrap_or_default();

    for (consumer_id, indexed, path) in rows {

        let Ok(Some(consumer)) = db::repos::get_by_id(&app.db, consumer_id).await else { continue };

        let Some(cname) = RepoName::new(&consumer.owner, &consumer.name) else { continue };

        let learned = target(&app.db, &consumer).await;

        let modules = oxymodules(app, &cname, &learned).await;

        let r = rule(&app.config.diffusion.follow, &consumer, modules.as_deref(), &path);

        let target = r.target.clone().unwrap_or(learned);

        if indexed != target {

            continue;

        }

        let go = match r.follow {

            Follow::Off => false,

            Follow::Green => green,

            Follow::Any => !green,

        };

        if !go {

            continue;

        }

        match bump(app, &consumer, &target, &path, upstream, heads.clone()).await {

            Ok(Bumped::Moved { from, .. }) => {

                let n = propose(app, &consumer, &target, &path, upstream, from, heads.clone()).await;

                tracing::info!(

                    "diffusion: {}/{} {path} bumped to {}/{} (proposal {n:?})",

                    consumer.owner,

                    consumer.name,

                    upstream.owner,

                    upstream.name

                );

            }

            Ok(_) => {}

            Err(e) => tracing::warn!("diffusion: bump {}/{} {path}: {e}", consumer.owner, consumer.name),

        }

    }

}

    // A passing run of an upstream's default bookmark: consumers following

    // it when green are bumped (Diffusion, #148).

    if let Some(done) = fdb::run_by_id(&app.db, run_id).await? {

        if done.status == "passed" && done.bookmark == row.default_bookmark {

            let heads: Vec<ChangeHash> = done.head_list().iter().filter_map(|h| ChangeHash::from_hex(h)).collect();

            let (app, row) = (app.clone(), row.clone());

            tokio::spawn(async move { crate::diffusion::propagate(&app, &row, heads, true).await });

        }

    }

                // An upstream's default bookmark moved: consumers following

                // every move are bumped now, those following green when

                // its run passes (#148).

                if let Some((_, _, heads)) = moved.iter().find(|(b, _, _)| *b == row.default_bookmark) {

                    let (app, row, heads) = (app.clone(), row.clone(), heads.clone());

                    tokio::spawn(async move { crate::diffusion::propagate(&app, &row, heads, false).await });

                }

/// Diffusion D1 end to end (#148): a consumer that follows every move gets

/// a bump proposal on a push; one that follows green gets it when the

/// upstream's run passes. One proposal, kept up to date.

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

async fn diffusion_proposes_a_bump_when_the_upstream_moves() {

    let h = hub().await;

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

    create(&h.app, "andy", "api").await;

    create(&h.app, "andy", "game").await;

    // Run every job of the upstream's newest run to a pass.

    let pass_all = || async {

        for _ in 0..20 {

            let Some(a) = claim(&h.app).await else {

                tokio::time::sleep(std::time::Duration::from_millis(30)).await;

                if !jobs_left(&h.app).await {

                    break;

                }

                continue;

            };

            let id = a["job_id"].as_i64().unwrap();

            let (name, ph) = (a["job"]["name"].as_str().unwrap().to_string(), a["pipeline_hash"].as_str().unwrap().to_string());

            assert_eq!(

                finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 10, "receipt": receipt(&name, &ph, 0) })).await,

                StatusCode::NO_CONTENT

            );

        }

    };

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

    let api = client(&format!("{}/andy/api", h.base), api_work.path());

    write(api_work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(api_work.path(), "test.sh", "echo ok\n");

    write(api_work.path(), "api.h", "v1\n");

    api.snapshot("t").unwrap();

    api.commit("api v1", &[]).unwrap();

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

    pass_all().await;

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

    let game = client(&format!("{}/andy/game", h.base), game_work.path());

    game.submodule_add(&format!("{}/andy/api", h.base), "deps/api", Some("main")).unwrap();

    write(game_work.path(), "game.c", "int main;\n");

    game.snapshot("t").unwrap();

    game.commit("game on api v1", &[]).unwrap();

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

    let game_row = db::repos::get(&h.app.db, "andy", "game").await.unwrap().unwrap();

    let bump = || async {

        for _ in 0..100 {

            let open = db::proposals::list(&h.app.db, game_row.id, &["open"]).await.unwrap();

            if let Some(p) = open.into_iter().find(|p| p.source_bookmark == "diffusion/deps/api") {

                return Some(p);

            }

            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        }

        None

    };

    // Off by default: an upstream move proposes nothing.

    write(api_work.path(), "a2.c", "v2\n");

    api.snapshot("t").unwrap();

    api.commit("api v2: faster sprites", &[]).unwrap();

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

    pass_all().await;

    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    assert!(db::proposals::list(&h.app.db, game_row.id, &[]).await.unwrap().is_empty(), "off proposes nothing");

    // Following every move: the next push proposes, by the diffusion account.

    let put = |body: &'static str| call(&h.app, "PUT", "/api/v1/repos/andy/game/diffusion", Some("site"), Some(body.into()));

    assert_eq!(put(r#"{"follow":"any"}"#).await.0, StatusCode::OK);

    write(api_work.path(), "a3.c", "v3\n");

    api.snapshot("t").unwrap();

    api.commit("api v3: save slots past ten", &[]).unwrap();

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

    let p = bump().await.expect("a bump proposal");

    assert_eq!((p.author.as_str(), p.target.as_str()), ("diffusion", "main"));

    assert!(p.title.contains("deps/api") && p.title.contains("andy/api") && p.title.contains("2 changes"), "{}", p.title);

    assert!(p.body.contains("api v2: faster sprites") && p.body.contains("api v3: save slots past ten"), "{}", p.body);

    let labels = db::issues::proposal_labels(&h.app.db, p.id).await.unwrap();

    assert!(labels.iter().any(|l| l.name == "dependencies"));

    pass_all().await;

    // Following green: a push alone does not move it; the passing run does,

    // and the same proposal is brought up to date.

    assert_eq!(put(r#"{"follow":"green"}"#).await.0, StatusCode::OK);

    write(api_work.path(), "a4.c", "v4\n");

    api.snapshot("t").unwrap();

    api.commit("api v4: rsm_set cheaper", &[]).unwrap();

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

    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    assert!(!bump().await.unwrap().body.contains("api v4"), "not before its run passes");

    pass_all().await;

    let mut updated = None;

    for _ in 0..100 {

        let p2 = bump().await.unwrap();

        if p2.body.contains("api v4") {

            updated = Some(p2);

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    let p2 = updated.expect("brought up to date once the run passed");

    assert_eq!(p2.number, p.number, "the same proposal");

    assert!(p2.title.contains("3 changes"), "{}", p2.title);

    assert_eq!(db::proposals::list(&h.app.db, game_row.id, &[]).await.unwrap().len(), 1, "never a pile of bumps");

}

/// Whether any job is queued or running.

async fn jobs_left(app: &Shared) -> bool {

    sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM flow_jobs WHERE status IN ('queued', 'running')")

        .fetch_one(&app.db)

        .await

        .unwrap_or(0)

        > 0

}

- **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. **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).