Placement ranks runners and says why a job ran where it did 501d8b268dd4

-- Why a job ran where it did (oxygenflow#42, Jetstream §7): "holds the

-- image, idle, speed 100 (class 3)".

ALTER TABLE flow_jobs ADD COLUMN placed TEXT NOT NULL DEFAULT '';

    /// Why it ran on its runner (oxygenflow#42): "holds the image, idle,

    /// speed 100 (class 3)".

    #[serde(default, skip_serializing_if = "String::is_empty")]

    pub placed: String,

                placed: j.placed,

    /// Why it ran on its runner (oxygenflow#42).

    pub placed: String,

/// Record why job `job_id` ran where it did.

pub async fn set_placed(pool: &Pool, job_id: i64, placed: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET placed = ? WHERE id = ?").bind(placed).bind(job_id).execute(pool).await?;

    Ok(())

}

    claim_ranked(pool, runtimes, runner, token_hash, room, scope, labels, &|_| false).await

}

/// [`claim_within`], passing over a job `elsewhere` says a better runner

/// should take (oxygenflow#42).

#[allow(clippy::too_many_arguments)]

pub async fn claim_ranked(

    pool: &Pool,

    runtimes: &[String],

    runner: &str,

    token_hash: &str,

    room: Room,

    scope: &super::runners::Scope,

    labels: &[String],

    elsewhere: &(dyn Fn(&Job) -> bool + Sync),

) -> sqlx::Result<Option<Job>> {

        }

        if elsewhere(&j) {

            continue;

    // Ranking (oxygenflow#42): a job another free runner would do better

    // (it holds the image, or is faster) is held for it briefly.

    let runners = db::runners::list(&app.db).await?;

    let me_now = runners.iter().find(|r| r.id == me.id).cloned().unwrap_or_else(|| me.clone());

    let running: std::collections::HashMap<i64, i64> = db::runners::running(&app.db).await?.into_iter().collect();

    let owners: std::collections::HashMap<i64, (String, i64)> =

        fdb::queue(&app.db).await?.into_iter().map(|q| (q.job_id, (q.owner, q.repo_id))).collect();

    let (archs, held) = (fdb::image_archs(&app.db).await?, crate::images::held(&app.db).await?);

    let now = db::now_ms();

    let fleet = Fleet { image_archs: &archs, hub_images: &held, now };

    let better = |j: &fdb::Job| -> bool {

        let Some((owner, repo_id)) = owners.get(&j.id) else { return false };

        let rootfs = rootfs_of(j);

        let mut holds = HELD.lock().unwrap_or_else(|p| p.into_inner());

        let holds = holds.get_or_insert_with(Default::default);

        holds.retain(|_, t| now - *t < 600_000);

        if holds.get(&j.id).is_some_and(|t| now - *t >= BETTER_HOLD_MS) {

            return false;

        }

        let mine = rank(&me_now, &rootfs);

        let other = runners.iter().any(|r| {

            r.id != me.id

                && r.state == "active"

                && r.last_seen_ms.is_some_and(|t| now - t < POLLING_MS)

                && running.get(&r.id).copied().unwrap_or(0) < r.max_jobs.max(1)

                && rank(r, &rootfs) > mine

                && runner_lacks(j, owner, *repo_id, r, &fleet, &[]).is_none()

        });

        if other {

            holds.entry(j.id).or_insert(now);

        }

        other

    };

        fdb::claim_ranked(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope(), &labels, &better).await?

    // Why here (oxygenflow#42).

    let waited = HELD.lock().unwrap_or_else(|p| p.into_inner()).as_mut().and_then(|h| h.remove(&job.id)).map(|t| now - t);

    let rootfs = rootfs_of(&job);

    let mut why = vec![if me_now.image_list().is_some_and(|have| have.contains(&rootfs)) {

        "holds the image".to_string()

    } else {

        "fetches the image from the hub".to_string()

    }];

    match running.get(&me.id).copied().unwrap_or(0) {

        0 => why.push("idle".into()),

        n => why.push(format!("running {n} other job{}", if n == 1 { "" } else { "s" })),

    }

    if let Some(sp) = me_now.speed {

        why.push(format!("speed {sp} (class {})", me_now.speed_class.unwrap_or(0)));

    }

    if let Some(w) = waited {

        why.push(format!("after {} s held for a better runner", w / 1000));

    }

    fdb::set_placed(&app.db, job.id, &why.join(", ")).await?;

}

/// Why no runner can take `job` of a repository of `owner` (id `repo_id`)

/// now (spec §11.4), from what each runner offered at its last claim:

/// empty when one could (it waits its turn or a free slot), else what

/// each one lacks. `declines` are the job's.

/// What placement knows of the fleet: images' architectures, the images

/// the hub holds, and the time.

pub struct Fleet<'a> {

    pub image_archs: &'a std::collections::HashMap<String, String>,

    pub hub_images: &'a std::collections::HashSet<String>,

    pub now: i64,

}

/// Why runner `r` cannot take `job` of a repository of `owner` (id

/// `repo_id`), from what it offered at its last claim (spec §11.4); `None`

/// when it can. `declines` are the job's.

pub fn runner_lacks(

    job: &fdb::Job,

    owner: &str,

    repo_id: i64,

    r: &db::runners::Runner,

    fleet: &Fleet,

    declines: &[fdb::DeclineRow],

) -> Option<String> {

    let sandbox: serde_json::Value = serde_json::from_str(&job.sandbox).unwrap_or_default();

    let rootfs = sandbox["rootfs"].as_str().unwrap_or("");

    let wanted_arch = job.arch().or_else(|| fleet.image_archs.get(rootfs).cloned());

    let need = |k: &str| sandbox[k].as_u64().unwrap_or(0);

    if r.state == "draining" {

        return Some("is draining".into());

    }

    let scope = r.scope();

    if (!scope.owner.is_empty() && scope.owner != owner) || scope.repo_id.is_some_and(|id| id != repo_id) {

        return Some("serves other repositories".into());

    }

    let tiers = r.tier_list();

    if !tiers.is_empty() && !tiers.contains(&job.runtime.as_str()) {

        return Some(format!("has no {} tier (it offers {})", job.runtime, tiers.join(", ")));

    }

    if let Some(w) = wanted_arch.as_deref().filter(|w| !r.arch.is_empty() && *w != r.arch) {

        return Some(format!("is {}; the job needs {w}", r.arch));

    }

    let fetches = r.fetches_images && fleet.hub_images.contains(rootfs);

    if !fetches && r.image_list().is_some_and(|have| !have.iter().any(|h| h == rootfs)) {

        let why = if r.fetches_images { " (nor does the hub: send it with flow rootfs push)" } else { "" };

        return Some(format!("does not hold image {}{why}", &rootfs[..12.min(rootfs.len())]));

    }

    let labels = r.label_list();

    if let Some(l) = job.needs().iter().find(|l| !labels.contains(&l.as_str())) {

        return Some(format!("lacks label {l}"));

    }

    if let Some(m) = r.memory_mib.filter(|m| (*m as u64) < need("memory_mib")) {

        return Some(format!("has {m} MiB; the job needs {} MiB", need("memory_mib")));

    }

    if let Some(c) = r.cpus.filter(|c| (*c as u64) < need("cpus")) {

        return Some(format!("has {c} CPUs; the job needs {}", need("cpus")));

    }

    // The disk it learned it needs (oxygenflow#41).

    if let (Some(n), Some(free)) = (job.need_disk_mib, r.free_disk_mib) {

        if free < n {

            return Some(format!(

                "has {} free; the job needs ~{} (its recent runs took up to {})",

                mib(free),

                mib(n),

                mib((n - 64) * 4 / 5)

            ));

        }

    }

    if let Some(d) = declines.iter().find(|d| d.runner_id == r.id && fleet.now - d.ms < fdb::DECLINE_HOLDS_MS) {

        return Some(format!("declined it: {}", d.reason));

    }

    None

    let fleet = Fleet { image_archs, hub_images, now };

        match runner_lacks(job, owner, repo_id, r, &fleet, declines) {

}

/// How long a job is held for a better runner before any runner that fits

/// takes it (oxygenflow#42).

const BETTER_HOLD_MS: i64 = 15_000;

/// A runner counts as there to take a job when it asked this recently.

const POLLING_MS: i64 = 10_000;

/// When each job was first passed over for a better runner.

static HELD: std::sync::Mutex<Option<std::collections::HashMap<i64, i64>>> = std::sync::Mutex::new(None);

/// How good runner `r` is for a job on image `rootfs`: holding the image

/// counts most, then the speed class.

fn rank(r: &db::runners::Runner, rootfs: &str) -> i64 {

    let holds = r.image_list().is_some_and(|have| have.iter().any(|h| h == rootfs));

    (holds as i64) * 10 + r.speed_class.unwrap_or(0)

}

fn rootfs_of(job: &fdb::Job) -> String {

    serde_json::from_str::<serde_json::Value>(&job.sandbox).ok().and_then(|v| v["rootfs"].as_str().map(String::from)).unwrap_or_default()

/// oxygenflow#42: a job another free runner would do better (it holds the

/// image) is held from a runner that would fetch it, the better runner

/// takes it, and the job says why it ran there.

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

async fn a_job_waits_briefly_for_a_runner_that_holds_its_image() {

    let h = hub().await;

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

    // A second runner, with a token of its own.

    db::runners::configured(&h.app.db, &blake3::hash(b"other-secret").to_hex(), "earth-9").await.unwrap();

    let ask = |images: &str| {

        format!(

            r#"{{"runner":"r","runtimes":["microvm"],"images":[{images}],"fetches_images":true,"facts":{{"speed":100,"speed_class":3,"free_disk_mib":50000}}}}"#

        )

    };

    let holder = ask(&format!("\"{ROOTFS}\""));

    // earth-9 holds the image and is polling; nothing is queued yet.

    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(holder.clone())).await;

    assert_eq!(st, StatusCode::NO_CONTENT);

    // The hub holds the image too, so test-1 could fetch it.

    sqlx::query("INSERT INTO jet_images (hash, size, arch, chunks, uploader, created_ms) VALUES (?, 1, 'x86_64', 1, 'andy', 0)")

        .bind(ROOTFS)

        .execute(&h.app.db)

        .await

        .unwrap();

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

    let url = format!("{}/andy/demo", h.base);

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

    let c = client(&url, work.path());

    write(

        work.path(),

        ".flow/pipeline.kdl",

        &format!("schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\n"),

    );

    c.commit("Build", &[]).unwrap();

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

    runs_of(&h.app, "andy", "demo").await;

    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(ask(""))).await;

    assert_eq!(st, StatusCode::NO_CONTENT, "held for the runner that holds the image");

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("other-secret"), Some(holder)).await;

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

    let id = serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap();

    let placed = db::flow::job(&h.app.db, id).await.unwrap().unwrap().placed;

    assert!(placed.starts_with("holds the image, idle, speed 100 (class 3)"), "{placed}");

    assert!(placed.contains("held for a better runner"), "{placed}");

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", Some("site"), None).await;

    assert!(body.contains("\"placed\":\"holds the image"), "{body}");

}

        // Why a queued job no runner can take waits (oxygenhub#28), and

        // why a job ran where it did (oxygenflow#42).

        } else if !j.placed.is_empty() {

            println!("    {}", ui::dim(&format!("placed: {}", j.placed)));

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

  <summary><span class="flow-glyph st-{{ l.j.status }}" data-glyph>{{ crate::web::flow::glyph(l.j.status) }}</span> <strong>{{ l.j.name }}</strong> <span class="small mute" data-meta>{{ l.j.status }}{% if let Some(c) = l.j.exit_code %} · exit {{ c }}{% endif %} · {{ l.duration }}{% if !l.j.runner.is_empty() %} · on <span{% if !l.j.placed.is_empty() %} title="Placed there: {{ l.j.placed }}"{% endif %}>{{ l.j.runner }}</span>{% endif %}</span>{% if !l.j.receipt_attest.is_empty() %} <span class="small mute">· <a href="{{ repo.url }}/receipts/{{ l.j.receipt_attest }}" title="Receipt {{ l.j.receipt_hash }}, kept in the repository; check it with oxy receipt verify">signed receipt</a> by {{ l.j.receipt_signer }}</span>{% else if !l.j.receipt_note.is_empty() %} <span class="small warn" title="{{ l.j.receipt_note }}">· receipt not kept: {{ l.j.receipt_note }}</span>{% else if !l.j.receipt_hash.is_empty() %} <span class="small mute">· unsigned receipt <code title="{{ l.j.receipt_hash }}">{{ l.j.receipt_hash|short }}</code></span>{% endif %}{% if !l.artifacts.is_empty() %} <span class="small mute">· {{ l.artifacts.len()|plural("artifact", "artifacts") }}</span>{% endif %}{% if let Some(t) = l.tests %} <span class="small{% if t.broken() > 0 %} warn{% endif %}">· <a href="#tests-{{ l.j.id }}">{{ t.summary() }}</a></span>{% endif %}{% if let Some((cid, pct)) = l.coverage %} <span class="small">· <a href="{{ repo.url }}/coverage?report={{ cid }}">coverage {{ pct }}</a></span>{% endif %}</summary>

  {% if !l.j.placed.is_empty() %}<p class="small mute pad">Ran on {{ l.j.runner }}: {{ l.j.placed }}.</p>{% endif %}