Diffusion: the reverse pin index, Used by, and how far behind f677d6270298

-- Diffusion (#148, docs/diffusion.md §4.1): the reverse pin index. Each

-- Oxygen subrepo pin on a repository's default bookmark, and on the

-- bookmark its work lands on, recorded when a push moves that bookmark:

-- who consumes which upstream (by the upstream's Oxygen repository id,

-- hex), at which heads.

CREATE TABLE diffusion_pins (

    consumer_id  INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    bookmark     TEXT NOT NULL,

    path         TEXT NOT NULL,

    upstream     TEXT NOT NULL,

    heads        TEXT NOT NULL,

    updated_ms   INTEGER NOT NULL,

    PRIMARY KEY (consumer_id, bookmark, path)

);

CREATE INDEX diffusion_pins_by_upstream ON diffusion_pins (upstream);

//! Diffusion (#148, docs/diffusion.md): an upstream's changes reach the

//! repositories that pin it as a subrepo.

//!

//! D1 starts with the reverse pin index (§4.1): when a push moves a

//! repository's default bookmark, or the bookmark its work lands on, each

//! Oxygen subrepo pin in the new state is recorded. That answers "who

//! consumes this repository, at which state, and how far behind", which

//! an upstream's page shows as Used by.

use crate::auth::roles;

use crate::db::{self, accounts::Account, repos::RepoRow, Pool};

use crate::repos::RepoName;

use crate::Shared;

use oxy_core::hash::ChangeHash;

use serde::Serialize;

/// How far back landed proposals teach where a repository's work lands.

pub const LEARN_DAYS: i64 = 90;

/// The bookmark a repository's work lands on (§4.3): the most common

/// target of its proposals landed in the last `LEARN_DAYS`, else its

/// default bookmark.

pub async fn target(pool: &Pool, row: &RepoRow) -> String {

    let since = db::now_ms() - LEARN_DAYS * 24 * 3600 * 1000;

    let learned: Option<String> = sqlx::query_scalar(

        "SELECT target FROM proposals WHERE repo_id = ? AND state = 'landed' AND landed_ms >= ?

         GROUP BY target ORDER BY COUNT(*) DESC, MAX(landed_ms) DESC LIMIT 1",

    )

    .bind(row.id)

    .bind(since)

    .fetch_optional(pool)

    .await

    .ok()

    .flatten();

    learned.filter(|t| !t.is_empty()).unwrap_or_else(|| row.default_bookmark.clone())

}

/// The bookmarks whose pins are indexed: the default and the learned

/// target (one when they are the same).

pub async fn followed_bookmarks(pool: &Pool, row: &RepoRow) -> Vec<String> {

    let mut out = vec![row.default_bookmark.clone()];

    let t = target(pool, row).await;

    if !out.contains(&t) {

        out.push(t);

    }

    out

}

/// Record `row`'s pins on `bookmark` at `heads`, replacing what was there.

pub async fn record(app: &Shared, row: &RepoRow, bookmark: &str, heads: &[ChangeHash]) {

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

    let pins = if heads.is_empty() { Vec::new() } else { crate::flow::pins_in(app, name, heads.to_vec()).await };

    let now = db::now_ms();

    let Ok(mut tx) = app.db.begin().await else { return };

    let mut ok = sqlx::query("DELETE FROM diffusion_pins WHERE consumer_id = ? AND bookmark = ?")

        .bind(row.id)

        .bind(bookmark)

        .execute(&mut *tx)

        .await

        .is_ok();

    for (path, id, pinned) in &pins {

        let heads: Vec<String> = pinned.iter().map(|h| h.to_hex()).collect();

        ok &=

            sqlx::query("INSERT INTO diffusion_pins (consumer_id, bookmark, path, upstream, heads, updated_ms) VALUES (?, ?, ?, ?, ?, ?)")

                .bind(row.id)

                .bind(bookmark)

                .bind(path)

                .bind(hex::encode(id))

                .bind(heads.join("+"))

                .bind(now)

                .execute(&mut *tx)

                .await

                .is_ok();

    }

    if ok {

        tx.commit().await.ok();

    }

}

/// After a push: index the pins of each moved bookmark that is followed.

pub async fn after_push(app: &Shared, row: &RepoRow, moved: &[(String, String, Vec<ChangeHash>)]) {

    let followed = followed_bookmarks(&app.db, row).await;

    for (bookmark, _, heads) in moved {

        if followed.contains(bookmark) {

            record(app, row, bookmark, heads).await;

        }

    }

}

/// Index every repository's followed bookmarks, for a hub that had no

/// index before (run once at start, in the background).

pub async fn backfill(app: &Shared) {

    let Ok(rows) = db::repos::list(&app.db).await else { return };

    for row in rows {

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

        for bookmark in followed_bookmarks(&app.db, &row).await {

            let b = bookmark.clone();

            let heads = app.repos.read(&name, move |r, _| heads_of(r, &b)).await;

            if let Ok(heads) = heads {

                record(app, &row, &bookmark, &heads).await;

            }

        }

    }

}

/// A bookmark's heads as the repository has them now (not a cached

/// list, which a push refreshes a moment later); empty when it has none.

fn heads_of(r: &oxy_core::repo::Repo, bookmark: &str) -> Result<Vec<ChangeHash>, crate::repos::RepoError> {

    Ok(r.bookmarks()?.into_iter().find(|b| b.name == bookmark).map(|b| b.heads).unwrap_or_default())

}

/// One consumer of an upstream, as an upstream's page and API show it.

#[derive(Clone, Debug, Serialize)]

pub struct Consumer {

    pub owner: String,

    pub name: String,

    pub bookmark: String,

    pub path: String,

    /// The upstream heads it pins, hex.

    pub pinned: Vec<String>,

    /// Upstream changes on its default bookmark the pin lacks; `None` when

    /// the hub cannot tell (the pinned changes are not in the upstream).

    pub behind: Option<usize>,

}

/// The consumers of `row` that `account` may read, with how far behind

/// the upstream's default bookmark each pin is.

pub async fn consumers(app: &Shared, row: &RepoRow, account: Option<&Account>) -> Result<Vec<Consumer>, crate::repos::RepoError> {

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

    let id = app.repos.read(&name, |r, _| Ok(r.id()?)).await?;

    let found: Vec<(i64, String, String, String)> = sqlx::query_as(

        "SELECT consumer_id, bookmark, path, heads FROM diffusion_pins WHERE upstream = ? ORDER BY consumer_id, bookmark, path",

    )

    .bind(hex::encode(id))

    .fetch_all(&app.db)

    .await

    .map_err(|e| crate::repos::RepoError::Io(std::io::Error::other(e)))?;

    let mut out = Vec::new();

    let mut pinned_sets = Vec::new();

    for (consumer_id, bookmark, path, heads) in found {

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

        if !roles::allows(roles::role_for(&app.db, account, &c).await.ok().flatten(), crate::auth::Access::Read) {

            continue;

        }

        let pinned: Vec<String> = heads.split('+').filter(|h| !h.is_empty()).map(String::from).collect();

        pinned_sets.push(pinned.iter().filter_map(|h| ChangeHash::from_hex(h)).collect::<Vec<_>>());

        out.push(Consumer { owner: c.owner, name: c.name, bookmark, path, pinned, behind: None });

    }

    if out.is_empty() {

        return Ok(out);

    }

    let default = row.default_bookmark.clone();

    let behind = app

        .repos

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

            let tip = heads_of(r, &default)?;

            let tip = cache.summary(r, &tip)?;

            Ok(pinned_sets

                .iter()

                .map(|p| cache.summary(r, p).ok().map(|s| tip.changes.iter().filter(|c| !s.changes.contains(*c)).count()))

                .collect::<Vec<_>>())

        })

        .await?;

    for (c, b) in out.iter_mut().zip(behind) {

        c.behind = b;

    }

    Ok(out)

}

        .route("/repos/{owner}/{name}/consumers", get(consumers))

/// `GET /repos/{owner}/{name}/consumers`: the repositories that pin this

/// one as a subrepo and that the caller may read, each with its pin and

/// how far behind this repository's default bookmark it is (#148).

async fn consumers(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path((owner, name)): Path<(String, String)>,

) -> Result<Json<Vec<crate::diffusion::Consumer>>, ApiError> {

    let c = caller(&app, &headers).await?;

    let (_, row, _) = reach(&app, &c, &owner, &name, Access::Read).await?;

    Ok(Json(crate::diffusion::consumers(&app, &row, c.account.as_ref()).await?))

}

pub(crate) async fn pins_in(

pub mod diffusion;

    // The reverse pin index for repositories pushed before it existed (#148).

    tokio::spawn({

        let app = app.clone();

        async move { diffusion::backfill(&app).await }

    });

                crate::diffusion::after_push(&app, &row, &moved).await;

    /// Repositories that pin this one as a subrepo, and how far behind

    /// each is (Diffusion, #148).

    used_by: Vec<crate::diffusion::Consumer>,

    let used_by = crate::diffusion::consumers(&app, &row, viewer.account()).await.unwrap_or_default();

    Ok(Page(OverviewPage {

        shell,

        repo: rs,

        o,

        readme_html,

        watching,

        latest_release,

        quickstart,

        fork,

        starred,

        stars,

        watchers,

        topics,

        used_by,

    }))

/// Diffusion D1 (#148): the reverse pin index. A repository that pins

/// another as a subrepo is a consumer of it; the upstream's API and page

/// say who, at which state, and how far behind its default bookmark.

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

async fn consumers_of_a_subrepo_are_indexed_with_how_far_behind_they_are() {

    let h = hub().await;

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

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

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

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

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

    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 consumers = || async {

        let (st, body) = call(&h.app, "GET", "/api/v1/repos/andy/api/consumers", Some("site"), None).await;

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

        serde_json::from_str::<Vec<serde_json::Value>>(&body).unwrap()

    };

    // The index is written by the push's after-work, a moment after the

    // client hears the push succeeded.

    let settle = |want: &'static (dyn Fn(&[serde_json::Value]) -> bool + Sync)| {

        let consumers = &consumers;

        async move {

            for _ in 0..100 {

                let list = consumers().await;

                if want(&list) {

                    return list;

                }

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

            }

            consumers().await

        }

    };

    let list = settle(&|l| !l.is_empty()).await;

    assert_eq!(list.len(), 1, "{list:?}");

    assert_eq!(

        (list[0]["name"].as_str(), list[0]["path"].as_str(), list[0]["bookmark"].as_str()),

        (Some("game"), Some("deps/api"), Some("main"))

    );

    assert_eq!(list[0]["behind"], 0);

    // The upstream moves on: the consumer is one change behind.

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

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

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

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

    assert_eq!(settle(&|l| l.first().is_some_and(|c| c["behind"] == 1)).await[0]["behind"], 1);

    let (_, page) = call(&h.app, "GET", "/andy/api", None, None).await;

    assert!(page.contains("Used by") && page.contains("andy/game") && page.contains("1 behind"), "the page shows it");

    // A private consumer is not shown to someone who cannot read it.

    sqlx::query("UPDATE repositories SET visibility = 'private' WHERE name = 'game'").execute(&h.app.db).await.unwrap();

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/api/consumers", None, None).await;

    assert_eq!(body.trim(), "[]");

    let (_, page) = call(&h.app, "GET", "/andy/api", None, None).await;

    assert!(!page.contains("andy/game"));

    sqlx::query("UPDATE repositories SET visibility = 'public' WHERE name = 'game'").execute(&h.app.db).await.unwrap();

    // Dropping the subrepo drops the consumer.

    game.submodule_remove("deps/api").unwrap();

    std::fs::remove_dir_all(game_work.path().join("deps/api")).unwrap();

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

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

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

    let left = settle(&|l| l.is_empty()).await;

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

}

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

    {% if !used_by.is_empty() %}<h3>Used by <span class="count-chip">{{ used_by.len() }}</span></h3>

    <ul class="plain used-by">{% for c in used_by %}<li><a href="/{{ c.owner }}/{{ c.name }}">{{ c.owner }}/{{ c.name }}</a> <span class="small mute">{{ c.bookmark }} · {{ c.path }}</span><br><span class="small">{% match c.behind %}{% when Some(0) %}<span class="tag">up to date</span>{% when Some(n) %}<span class="tag warn" title="Changes on {{ repo.default_bookmark }} its pin lacks">{{ n }} behind</span>{% when None %}<span class="mute" title="Its pin names changes this repository does not hold">pin not here</span>{% endmatch %}</span></li>{% endfor %}</ul>{% endif %}