Diffusion: turning it on e71583501b1d

-- Diffusion (#148, docs/diffusion.md §4.2): whether this repository follows

-- the upstreams it pins. Empty follows hub.toml's [diffusion] follow;

-- otherwise off, green (when the upstream's checks pass) or any (on

-- every move). A repository's .oxymodules keys win over this.

ALTER TABLE repositories ADD COLUMN diffusion TEXT NOT NULL DEFAULT '';

        .route("/repos/{owner}/{name}/diffusion", axum::routing::put(set_diffusion))

#[derive(serde::Deserialize)]

struct DiffusionSetting {

    /// `off`, `green`, `any`, or `default` (the site's).

    follow: String,

}

/// `PUT /repos/{owner}/{name}/diffusion {"follow": ...}`: whether this

/// repository follows the upstreams it pins (#148). Owners and admins.

async fn set_diffusion(

    State(app): State<Shared>,

    headers: HeaderMap,

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

    Json(d): Json<DiffusionSetting>,

) -> Result<Json<serde_json::Value>, ApiError> {

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

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

    let setting = match d.follow.as_str() {

        "off" | "green" | "any" => d.follow.as_str(),

        "default" => "",

        _ => return Err(ApiError(StatusCode::BAD_REQUEST, "follow is off, green, any or default".into())),

    };

    db::repos::set_diffusion(&app.db, row.id, setting).await?;

    db::accounts::audit(

        &app.db,

        c.account.as_ref(),

        "repo.diffusion",

        &format!("{owner}/{name}"),

        if setting.is_empty() { "default" } else { setting },

    )

    .await

    .ok();

    let row = db::repos::get_by_id(&app.db, row.id).await?.unwrap_or(row);

    let rule = crate::diffusion::rule(&app.config.diffusion.follow, &row, None, "");

    Ok(Json(serde_json::json!({ "follow": rule.follow, "from": rule.from })))

}

    pub diffusion: Diffusion,

/// Diffusion (#148, docs/diffusion.md §4.2): the site's default for

/// repositories that say nothing.

#[derive(Clone, Debug, Deserialize)]

#[serde(default, deny_unknown_fields)]

pub struct Diffusion {

    /// `off`, `green` or `any`.

    pub follow: String,

}

impl Default for Diffusion {

    fn default() -> Self {

        Diffusion { follow: "off".into() }

    }

}

            diffusion: Diffusion::default(),

#[derive(Clone, Debug, Default, FromRow)]

    /// Diffusion: `off`, `green`, `any`, or empty to follow hub.toml

    /// (migration 0058, #148).

    #[sqlx(default)]

    pub diffusion: String,

    Ok(())

}

pub async fn set_diffusion(pool: &Pool, id: i64, setting: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE repositories SET diffusion = ? WHERE id = ?").bind(setting).bind(id).execute(pool).await?;

/// When a consumer takes an upstream's moves (§4.2).

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]

#[serde(rename_all = "lowercase")]

pub enum Follow {

    Off,

    /// When the upstream state's checks pass.

    Green,

    /// On every move.

    Any,

}

impl Follow {

    pub fn parse(s: &str) -> Option<Follow> {

        match s.trim() {

            "off" => Some(Follow::Off),

            "green" => Some(Follow::Green),

            "any" => Some(Follow::Any),

            _ => None,

        }

    }

}

/// How one consumer follows one subrepo, from its `.oxymodules` entry, its

/// repository setting and the site's default, in that order.

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]

pub struct Rule {

    pub follow: Follow,

    /// Land a green bump itself (D5); D1 only proposes.

    pub land_auto: bool,

    /// The bookmark to bump, when `.oxymodules` names one (else learned).

    pub target: Option<String>,

    /// Paths the consumer watches: upstream changes touching them are

    /// flagged in a bump's changelog.

    pub watch: Vec<String>,

    /// Where `follow` came from: `.oxymodules`, `setting` or `site`.

    pub from: &'static str,

}

/// The Diffusion keys of `path`'s entry in `.oxymodules` text: `follow`,

/// `land`, `target` and `watch` (comma-separated, or a TOML list).

pub fn module_keys(text: &str, path: &str) -> (Option<Follow>, Option<bool>, Option<String>, Vec<String>) {

    let (mut follow, mut land, mut target, mut watch) = (None, None, None, Vec::new());

    let mut here = false;

    for line in text.lines() {

        let line = line.trim();

        if line.is_empty() || line.starts_with('#') {

            continue;

        }

        if let Some(inner) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {

            here = inner.trim().trim_matches('"').trim_end_matches('/') == path.trim_end_matches('/');

            continue;

        }

        let (true, Some((k, v))) = (here, line.split_once('=')) else { continue };

        let v = v.trim();

        let unquoted = v.trim_matches('"').trim();

        match k.trim() {

            "follow" => follow = Follow::parse(unquoted),

            "land" => land = Some(unquoted == "auto"),

            "target" if !unquoted.is_empty() => target = Some(unquoted.to_string()),

            "watch" => {

                watch = v

                    .trim_start_matches('[')

                    .trim_end_matches(']')

                    .split(',')

                    .map(|w| w.trim().trim_matches('"').trim().to_string())

                    .filter(|w| !w.is_empty())

                    .collect()

            }

            _ => {}

        }

    }

    (follow, land, target, watch)

}

/// How `row` follows its subrepo at `path`, given its `.oxymodules` text.

pub fn rule(site_default: &str, row: &RepoRow, oxymodules: Option<&str>, path: &str) -> Rule {

    let (follow, land, target, watch) = oxymodules.map(|t| module_keys(t, path)).unwrap_or_default();

    let (follow, from) = match follow {

        Some(f) => (f, ".oxymodules"),

        None => match Follow::parse(&row.diffusion) {

            Some(f) => (f, "setting"),

            None => (Follow::parse(site_default).unwrap_or(Follow::Off), "site"),

        },

    };

    Rule { follow, land_auto: land.unwrap_or(false), target, watch, from }

}

#[cfg(test)]

mod tests {

    use super::*;

    const MODULES: &str = r#"

[rstar-api]

url = https://oxygenhub.io/doomsdayone/rstar-api

follow = green

land = "propose"

target = develop

watch = ["src/x16/irq*", "src/x16/frame*"]

[other]

url = https://example/other

follow = any

"#;

    fn row(diffusion: &str) -> RepoRow {

        RepoRow {

            id: 1,

            owner: "o".into(),

            name: "n".into(),

            description: String::new(),

            default_bookmark: "main".into(),

            visibility: "public".into(),

            created_ms: 0,

            updated_ms: 0,

            git_mirror: false,

            git_mirror_state: String::new(),

            forked_from: None,

            topics: String::new(),

            flow: String::new(),

            diffusion: diffusion.into(),

            ..Default::default()

        }

    }

    #[test]

    fn oxymodules_keys_are_read_per_entry() {

        let (f, land, target, watch) = module_keys(MODULES, "rstar-api");

        assert_eq!((f, land, target.as_deref()), (Some(Follow::Green), Some(false), Some("develop")));

        assert_eq!(watch, vec!["src/x16/irq*", "src/x16/frame*"]);

        assert_eq!(module_keys(MODULES, "other").0, Some(Follow::Any));

        assert_eq!(module_keys(MODULES, "missing"), (None, None, None, vec![]));

        assert_eq!(module_keys("[a]\nwatch = x, y\n", "a").3, vec!["x", "y"]);

    }

    #[test]

    fn oxymodules_wins_then_the_setting_then_the_site() {

        let r = rule("off", &row("any"), Some(MODULES), "rstar-api");

        assert_eq!((r.follow, r.from), (Follow::Green, ".oxymodules"));

        let r = rule("off", &row("any"), Some("[rstar-api]\nurl = x\n"), "rstar-api");

        assert_eq!((r.follow, r.from), (Follow::Any, "setting"));

        let r = rule("green", &row(""), None, "rstar-api");

        assert_eq!((r.follow, r.from), (Follow::Green, "site"));

        assert_eq!(rule("nonsense", &row(""), None, "x").follow, Follow::Off);

    }

}

        .route("/{owner}/{name}/settings/diffusion", post(repo_settings::diffusion))

    /// hub.toml's `[diffusion] follow` (#148).

    diffusion_site: String,

    let diffusion_site = app.config.diffusion.follow.clone();

        diffusion_site,

pub struct DiffusionForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// `off`, `green`, `any`, or `default`.

    diffusion: String,

}

/// Whether this repository follows the upstreams it pins (Diffusion,

/// #148): its owners and admins decide, since it opens proposals here.

pub async fn diffusion(

    State(app): State<Shared>,

    viewer: Viewer,

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

    Form(f): Form<DiffusionForm>,

) -> Result<Page<RepoSettingsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/settings"), &viewer);

    let (_, row, role) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Maintain).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let setting = match f.diffusion.as_str() {

        "off" | "green" | "any" => f.diffusion.as_str(),

        "default" => "",

        _ => return settings_page(&app, shell, row, role, None, Some("Pick off, green, any or the site default.".into())).await,

    };

    db::repos::set_diffusion(&app.db, row.id, setting).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(

        &app.db,

        viewer.account(),

        "repo.diffusion",

        &format!("{owner}/{name}"),

        if setting.is_empty() { "default" } else { setting },

    )

    .await

    .ok();

    let row = db::repos::get_by_id(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?.unwrap_or(row);

    let notice = match crate::diffusion::rule(&app.config.diffusion.follow, &row, None, "").follow {

        crate::diffusion::Follow::Off => "Diffusion is off: upstream moves open no proposals here.",

        crate::diffusion::Follow::Green => {

            "Diffusion is on: when a subrepo's upstream moves and its checks pass, a proposal moves the pin here."

        }

        crate::diffusion::Follow::Any => "Diffusion is on: every move of a subrepo's upstream opens or updates a proposal here.",

    };

    settings_page(&app, shell, row, role, Some(notice.into()), None).await

}

#[derive(Deserialize)]

    // The site token acts as the first site administrator.

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

    // Following is off by default; owners turn it on (DF2).

    let put =

        |body: &'static str, who: Option<&'static str>| call(&h.app, "PUT", "/api/v1/repos/andy/game/diffusion", who, Some(body.into()));

    let (st, body) = put(r#"{"follow":"green"}"#, Some("site")).await;

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

    assert!(body.contains("\"follow\":\"green\"") && body.contains("\"from\":\"setting\""), "{body}");

    assert_eq!(put(r#"{"follow":"sometimes"}"#, Some("site")).await.0, StatusCode::BAD_REQUEST);

    assert_ne!(put(r#"{"follow":"any"}"#, None).await.0, StatusCode::OK, "not without a token");

    let (_, body) = put(r#"{"follow":"default"}"#, Some("site")).await;

    assert!(body.contains("\"follow\":\"off\"") && body.contains("\"from\":\"site\""), "{body}");

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

<form class="card wide" id="diffusion" method="post" action="{{ repo.url }}/settings/diffusion">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

  <h2>Diffusion</h2>

  <p>When a repository this one pins as a subrepo moves on, Diffusion opens a proposal here that moves the pin, tested by this repository's own pipeline, with the upstream's changes listed. It never lands anything by itself.</p>

  <div class="choice">

    <label><input type="radio" name="diffusion" value="green"{% if row.diffusion == "green" %} checked{% endif %}> when the upstream's checks pass</label>

    <label><input type="radio" name="diffusion" value="any"{% if row.diffusion == "any" %} checked{% endif %}> on every upstream move</label>

    <label><input type="radio" name="diffusion" value="off"{% if row.diffusion == "off" %} checked{% endif %}> off</label>

    <label><input type="radio" name="diffusion" value="default"{% if row.diffusion.is_empty() %} checked{% endif %}> site default ({{ diffusion_site }}, from <code>[diffusion] follow</code> in <code>hub.toml</code>)</label>

  </div>

  <p class="small mute">A subrepo's <code>.oxymodules</code> entry overrides this with <code>follow = green</code>, <code>any</code> or <code>off</code>; <code>target</code> names the bookmark to move (else the one this repository's proposals land on) and <code>watch</code> lists paths whose upstream changes are flagged.</p>

  <button type="submit">SAVE</button>

</form>