Runners decline, and the queue says why a job waits 4d2e44d79b03

-- Placement you can see (OxygenFlow spec §11.4–11.5, oxygenflow#36): what

-- each runner offers at its last claim, the jobs runners handed back with

-- a reason, and why a queued job waits.

ALTER TABLE flow_runners ADD COLUMN tiers TEXT NOT NULL DEFAULT '';

-- JSON array of image hashes; empty when the runner did not say.

ALTER TABLE flow_runners ADD COLUMN images TEXT NOT NULL DEFAULT '';

-- BLAKE3 of what it offers: a decline holds until this changes.

ALTER TABLE flow_runners ADD COLUMN profile TEXT NOT NULL DEFAULT '';

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

CREATE TABLE flow_declines (

    job_id     INTEGER NOT NULL REFERENCES flow_jobs(id) ON DELETE CASCADE,

    runner_id  INTEGER NOT NULL,

    runner     TEXT NOT NULL,

    profile    TEXT NOT NULL,

    reason     TEXT NOT NULL,

    ms         INTEGER NOT NULL,

    PRIMARY KEY (job_id, runner_id)

);

CREATE INDEX flow_declines_runner ON flow_declines (runner_id, ms);

- **Runners decline, and the queue says why a job waits (oxygenflow#36, OxygenFlow spec §11.4–11.5).** A runner can hand back a job it cannot run with a reason. The job is queued again and not offered to that runner while what it offers stays the same, for ten minutes at most. After three different runners decline it, the job ends as unplaceable with their reasons. Every 15 seconds the hub works out why each ready job no runner can take is waiting, from each runner's last claim: no runner online, a missing tier, the wrong architecture, an image it does not hold, a label, memory or CPUs, a scope, or a decline. The run page and the API show the reason, and a runner's page shows its tiers and what it declined.

    /// Why a queued job no runner can take waits (OxygenFlow spec §11.4);

    /// empty when it only waits its turn.

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

    pub waiting: String,

export type Via = {

  expires_ms?: number | null;

  key: string;

  name: string;

  revoked: boolean;

  scope: string;

};

  via?: Via | null;

export type Label = {

  colour: string;

  description: string;

  name: string;

};

  labels?: Label[];

export type NewMilestone = {

  description?: string;

  due?: string | null;

  title: string;

};

export type MilestoneEdit = {

  description?: string | null;

  due?: string | null;

  state?: string | null;

  title?: string | null;

};

  via?: string;

  title?: string | null;

};

export type ReleaseEdit = {

  notes?: string | null;

  prerelease?: boolean | null;

  title?: string | null;

};

export type ProposalEdit = {

  body?: string | null;

  labels?: string[] | null;

  waiting?: string;

        .route("/flow/jobs/{id}/decline", post(crate::flow::decline))

                waiting: j.waiting,

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

    /// Why a queued job no runner can take waits (spec §11.4), from the

    /// last placement pass; empty when it is only waiting its turn.

    pub waiting: String,

    }

    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).

    pub fn arch(&self) -> Option<String> {

        arch_in(&self.sandbox)

    /// Jobs this runner declined, not to be offered to it again (spec §11.5).

    pub declined: std::collections::HashSet<i64>,

        if !runtimes.iter().any(|r| r == &j.runtime) || !room.fits(&j.sandbox) || room.declined.contains(&j.id) {

}

/// How long a decline keeps a job from its runner when the runner's

/// profile stays the same: a reason such as memory in use passes.

pub const DECLINE_HOLDS_MS: i64 = 10 * 60_000;

/// The jobs runner `runner_id` declined with this profile, recently.

pub async fn declined_by(pool: &Pool, runner_id: i64, profile: &str) -> sqlx::Result<std::collections::HashSet<i64>> {

    let rows: Vec<(i64,)> = sqlx::query_as("SELECT job_id FROM flow_declines WHERE runner_id = ? AND profile = ? AND ms > ?")

        .bind(runner_id)

        .bind(profile)

        .bind(now_ms() - DECLINE_HOLDS_MS)

        .fetch_all(pool)

        .await?;

    Ok(rows.into_iter().map(|(j,)| j).collect())

}

/// A decline: one runner's reason for handing a job back.

#[derive(Clone, Debug, sqlx::FromRow)]

pub struct DeclineRow {

    pub job_id: i64,

    pub runner_id: i64,

    pub runner: String,

    pub reason: String,

    pub ms: i64,

}

/// Runner `runner_id` hands back job `job_id`, running on it: the decline

/// is kept (one per job and runner, the latest), and the job is queued

/// again. Returns every decline the job has had, or `None` when the job

/// was not running on that runner.

pub async fn decline(

    pool: &Pool,

    job_id: i64,

    runner_id: i64,

    runner: &str,

    profile: &str,

    reason: &str,

) -> sqlx::Result<Option<Vec<DeclineRow>>> {

    let mut tx = super::begin_write(pool).await?;

    let back = sqlx::query(

        "UPDATE flow_jobs SET status = 'queued', runner = '', token_hash = '', started_ms = NULL, runner_id = NULL \

         WHERE id = ? AND status = 'running' AND runner_id = ?",

    )

    .bind(job_id)

    .bind(runner_id)

    .execute(&mut *tx)

    .await?;

    if back.rows_affected() != 1 {

        tx.rollback().await?;

        return Ok(None);

    }

    sqlx::query(

        "INSERT INTO flow_declines (job_id, runner_id, runner, profile, reason, ms) VALUES (?, ?, ?, ?, ?, ?) \

         ON CONFLICT (job_id, runner_id) DO UPDATE SET runner = excluded.runner, profile = excluded.profile, reason = excluded.reason, ms = excluded.ms",

    )

    .bind(job_id)

    .bind(runner_id)

    .bind(runner)

    .bind(profile)

    .bind(reason)

    .bind(now_ms())

    .execute(&mut *tx)

    .await?;

    let all: Vec<DeclineRow> =

        sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = ? ORDER BY ms")

            .bind(job_id)

            .fetch_all(&mut *tx)

            .await?;

    tx.commit().await?;

    Ok(Some(all))

}

/// The declines of job `job_id`, oldest first.

pub async fn declines_of_job(pool: &Pool, job_id: i64) -> sqlx::Result<Vec<DeclineRow>> {

    sqlx::query_as("SELECT job_id, runner_id, runner, reason, ms FROM flow_declines WHERE job_id = ? ORDER BY ms")

        .bind(job_id)

        .fetch_all(pool)

        .await

}

/// A runner's latest declines, newest first, with where the job is.

#[derive(Clone, Debug, sqlx::FromRow)]

pub struct RunnerDecline {

    pub job_id: i64,

    pub job: String,

    pub run_number: i64,

    pub owner: String,

    pub name: String,

    pub reason: String,

    pub ms: i64,

}

pub async fn declines_of_runner(pool: &Pool, runner_id: i64, limit: i64) -> sqlx::Result<Vec<RunnerDecline>> {

    sqlx::query_as(

        "SELECT d.job_id, j.name AS job, r.number AS run_number, rep.owner, rep.name, d.reason, d.ms \

         FROM flow_declines d JOIN flow_jobs j ON j.id = d.job_id JOIN flow_runs r ON r.id = j.run_id \

         JOIN repositories rep ON rep.id = r.repo_id WHERE d.runner_id = ? ORDER BY d.ms DESC LIMIT ?",

    )

    .bind(runner_id)

    .bind(limit)

    .fetch_all(pool)

    .await

}

/// Queued jobs whose dependencies have passed, in their live runs, with

/// their repositories (owner, id): the ones a runner could take now.

pub async fn ready_queued(pool: &Pool) -> sqlx::Result<Vec<(Job, String, i64)>> {

    let queued: Vec<Job> = sqlx::query_as(

        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \

         WHERE j.status = 'queued' AND r.status IN ('queued', 'running') ORDER BY j.run_id, j.depth, j.id LIMIT 500",

    )

    .fetch_all(pool)

    .await?;

    let mut out = Vec::new();

    for j in queued {

        let mut ready = true;

        for d in j.depends_on() {

            let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")

                .bind(j.run_id)

                .bind(&d)

                .fetch_optional(pool)

                .await?;

            if st.as_deref() != Some("passed") {

                ready = false;

                break;

            }

        }

        if !ready {

            continue;

        }

        let repo: Option<(String, i64)> =

            sqlx::query_as("SELECT rep.owner, rep.id FROM flow_runs r JOIN repositories rep ON rep.id = r.repo_id WHERE r.id = ?")

                .bind(j.run_id)

                .fetch_optional(pool)

                .await?;

        if let Some((owner, id)) = repo {

            out.push((j, owner, id));

        }

    }

    Ok(out)

}

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

    sqlx::query("UPDATE flow_jobs SET waiting = ? WHERE id = ? AND waiting <> ?")

        .bind(waiting)

        .bind(job_id)

        .bind(waiting)

        .execute(pool)

        .await?;

    Ok(())

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

    /// The tiers it offered at its last claim, comma-separated.

    pub tiers: String,

    /// The images it held then (a JSON array); empty when it did not say.

    pub images: String,

    /// BLAKE3 of what it offers; its declines hold until this changes.

    pub profile: String,

    Ok(())

}

/// What a runner offers at a claim (OxygenFlow spec §11.1): its tiers and

/// images, and the profile hash over those, its architecture, labels and

/// capacity.

pub async fn set_offer(pool: &Pool, id: i64, tiers: &[String], images: Option<&[String]>, profile: &str) -> sqlx::Result<()> {

    let images = images.map(|i| serde_json::to_string(i).unwrap_or_default()).unwrap_or_default();

    sqlx::query("UPDATE flow_runners SET tiers = ?, images = ?, profile = ? WHERE id = ?")

        .bind(tiers.join(","))

        .bind(images)

        .bind(profile)

        .bind(id)

        .execute(pool)

        .await?;

}

impl Runner {

    pub fn tier_list(&self) -> Vec<&str> {

        self.tiers.split(',').filter(|t| !t.is_empty()).collect()

    }

    /// The images it holds, `None` when it does not say.

    pub fn image_list(&self) -> Option<Vec<String>> {

        if self.images.is_empty() {

            None

        } else {

            serde_json::from_str(&self.images).ok()

        }

    }

    // What it offers, kept for the placement pass; a job it declined is

    // not offered to it again while this stays the same (spec §11.5).

    let profile = profile_of(&req, &arch);

    db::runners::set_offer(&app.db, me.id, &req.runtimes, req.images.as_deref(), &profile).await?;

        declined: fdb::declined_by(&app.db, me.id, &profile).await?,

/// BLAKE3 over what a claiming runner offers: tiers, images, architecture,

/// labels and capacity, each list sorted.

fn profile_of(req: &ClaimRequest, arch: &str) -> String {

    let sorted = |v: &[String]| {

        let mut v = v.to_vec();

        v.sort();

        v

    };

    let info = req.info.as_ref();

    let v = serde_json::json!({

        "arch": arch,

        "tiers": sorted(&req.runtimes),

        "images": req.images.as_deref().map(sorted),

        "labels": info.map(|i| sorted(&i.labels)),

        "memory_mib": info.and_then(|i| i.memory_mib),

        "cpus": info.and_then(|i| i.cpus),

    });

    blake3::hash(v.to_string().as_bytes()).to_hex().to_string()

}

/// A job handed back by this many different runners is unplaceable.

pub const UNPLACEABLE_AFTER: usize = 3;

/// `POST /api/v1/flow/jobs/{id}/decline` (OxygenFlow spec §11.5): the

/// runner will not run a job it was handed. The job is queued again and

/// not offered to that runner while its profile stays the same (and for

/// [`fdb::DECLINE_HOLDS_MS`] at most); once [`UNPLACEABLE_AFTER`]

/// different runners have declined it, it ends as unplaceable with their

/// reasons. A decline is never a failed check.

pub async fn decline(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(id): Path<i64>,

    Json(d): Json<flow_core::api::Decline>,

) -> Result<StatusCode, ApiError> {

    let me = runner_auth(&app, &headers).await?;

    db::runners::seen(&app.db, me.id, None, None).await?;

    let reason: String = d.reason.chars().map(|c| if c.is_control() { ' ' } else { c }).take(500).collect();

    let reason = match reason.trim() {

        "" => "no reason given".to_string(),

        r => r.to_string(),

    };

    let Some(all) = fdb::decline(&app.db, id, me.id, &me.name, &me.profile, &reason).await? else {

        return Err(ApiError(StatusCode::CONFLICT, "that job is not running on this runner".into()));

    };

    tracing::info!("flow: runner {} declined job {id}: {reason}", me.name);

    let job = fdb::job(&app.db, id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such job".into()))?;

    let runners: std::collections::BTreeSet<i64> = all.iter().map(|d| d.runner_id).collect();

    if runners.len() >= UNPLACEABLE_AFTER {

        let error = format!(

            "unplaceable: declined by {} runners: {}",

            runners.len(),

            all.iter().map(|d| format!("{}: {}", d.runner, d.reason)).collect::<Vec<_>>().join("; ")

        );

        let changed =

            fdb::finish(&app.db, id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })

                .await?;

        report_changed(&app, job.run_id, changed).await?;

    } else if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {

        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {

            let summary = format!("declined by {}: {reason}; waiting for another runner", me.name);

            report_check(&app, &row, &run.state, run.number, &job.name, "queued", &summary).await;

            fdb::set_waiting(&app.db, id, &summary).await?;

        }

    }

    Ok(StatusCode::NO_CONTENT)

}

/// 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.

pub fn why_waiting(

    job: &fdb::Job,

    owner: &str,

    repo_id: i64,

    runners: &[db::runners::Runner],

    image_archs: &std::collections::HashMap<String, String>,

    declines: &[fdb::DeclineRow],

    now: i64,

) -> 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(|| image_archs.get(rootfs).cloned());

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

    let needs = job.needs();

    let lacks = |r: &db::runners::Runner| -> Option<String> {

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

        }

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

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

        }

        let labels = r.label_list();

        if let Some(l) = 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")));

        }

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

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

        }

        None

    };

    let online: Vec<&db::runners::Runner> = runners.iter().filter(|r| r.state != "disabled" && r.online()).collect();

    if online.is_empty() {

        return "no runner is online".into();

    }

    let mut why = Vec::new();

    for r in &online {

        match lacks(r) {

            None => return String::new(),

            Some(w) => why.push(format!("{} {w}", r.name)),

        }

    }

    let more = why.len().saturating_sub(4);

    why.truncate(4);

    let mut text = format!("no runner can take it: {}", why.join("; "));

    if more > 0 {

        text.push_str(&format!("; and {more} more"));

    }

    text

}

/// One placement pass: say why each ready queued job waits.

pub async fn explain_waits(app: &Shared) -> sqlx::Result<()> {

    let ready = fdb::ready_queued(&app.db).await?;

    if ready.is_empty() {

        return Ok(());

    }

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

    let archs = fdb::image_archs(&app.db).await?;

    let now = db::now_ms();

    for (job, owner, repo_id) in ready {

        let declines = fdb::declines_of_job(&app.db, job.id).await?;

        let why = why_waiting(&job, &owner, repo_id, &runners, &archs, &declines, now);

        fdb::set_waiting(&app.db, job.id, &why).await?;

    }

    Ok(())

}

/// Run a placement pass every 15 seconds.

pub async fn placement(app: Shared) {

    loop {

        tokio::time::sleep(std::time::Duration::from_secs(15)).await;

        if let Err(e) = explain_waits(&app).await {

            tracing::warn!("flow: placement pass: {e}");

        }

    }

}

        );

    }

}

#[cfg(test)]

mod placement_tests {

    use super::*;

    #[test]

    fn a_waiting_job_says_what_each_runner_lacks() {

        use crate::db::runners::Runner;

        let rootfs = "ab".repeat(32);

        let job = fdb::Job {

            id: 7,

            runtime: "syscall".into(),

            sandbox: serde_json::json!({ "rootfs": rootfs, "memory_mib": 4096, "cpus": 2 }).to_string(),

            spec: serde_json::json!({ "name": "build" }).to_string(),

            ..Default::default()

        };

        let now = db::now_ms();

        let runner = |id: i64, name: &str| Runner {

            id,

            name: name.into(),

            state: "active".into(),

            last_seen_ms: Some(now),

            arch: "x86_64".into(),

            tiers: "microvm,syscall".into(),

            images: serde_json::to_string(&[&rootfs]).unwrap(),

            memory_mib: Some(16384),

            cpus: Some(8),

            ..Default::default()

        };

        let archs = std::collections::HashMap::new();

        let why = |runners: &[Runner], declines: &[fdb::DeclineRow]| why_waiting(&job, "andy", 1, runners, &archs, declines, now);

        assert_eq!(why(&[], &[]), "no runner is online");

        assert_eq!(why(&[runner(1, "earth-1")], &[]), "", "one can take it: it only waits its turn");

        let microvm_only = Runner { tiers: "microvm".into(), ..runner(1, "earth-1") };

        let riscv = Runner { arch: "riscv64".into(), ..runner(2, "narvi") };

        let small = Runner { memory_mib: Some(2048), ..runner(3, "pi") };

        let other = Runner { scope_owner: "bob".into(), ..runner(4, "bobs") };

        let offline = Runner { last_seen_ms: Some(now - 3_600_000), ..runner(5, "gone") };

        assert_eq!(

            why(&[microvm_only, small, other, offline], &[]),

            "no runner can take it: earth-1 has no syscall tier (it offers microvm); pi has 2048 MiB; the job needs 4096 MiB; bobs serves other repositories"

        );

        let archs_known: std::collections::HashMap<String, String> = [(rootfs.clone(), "x86_64".to_string())].into();

        assert_eq!(

            why_waiting(&job, "andy", 1, &[riscv], &archs_known, &[], now),

            "no runner can take it: narvi is riscv64; the job needs x86_64"

        );

        let without = Runner { images: "[]".into(), ..runner(6, "new") };

        assert_eq!(why(&[without], &[]), "no runner can take it: new does not hold image abababababab");

        let d = fdb::DeclineRow {

            job_id: 7,

            runner_id: 1,

            runner: "earth-1".into(),

            reason: "only 512 MiB of memory is free here".into(),

            ms: now,

        };

        assert_eq!(

            why(&[runner(1, "earth-1")], std::slice::from_ref(&d)),

            "no runner can take it: earth-1 declined it: only 512 MiB of memory is free here"

        let old = fdb::DeclineRow { ms: now - fdb::DECLINE_HOLDS_MS - 1, ..d };

        assert_eq!(why(&[runner(1, "earth-1")], &[old]), "", "an old decline no longer holds");

    tokio::spawn(flow::placement(app.clone()));

    declines: Vec<crate::db::flow::RunnerDecline>,

    let declines = crate::db::flow::declines_of_runner(&app.db, id, 20).await.map_err(e)?;

    Ok(Page(RunnerPage { shell, r, running, history, declines, older, jobs, passed, busy_ms }))

    config.flow.runner_tokens = vec!["runner-secret".into(), "runner-two".into(), "runner-three".into()];

/// OxygenFlow spec §11.5: a runner hands back a job it cannot run. The

/// job is queued again and not offered to that runner again, the decline

/// shows on the run's page and is kept for the runner's, and once three

/// runners have declined it the job ends as unplaceable with their reasons.

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

async fn a_declined_job_goes_to_another_runner_or_ends_unplaceable() {

    let h = hub().await;

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

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

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

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

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

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

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

    c.commit("Add a pipeline", &[]).unwrap();

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

    let run = runs_of(&h.app, "andy", "declines").await[0].clone();

    let claim_as = |token: &'static str, name: &'static str| {

        let app = h.app.clone();

        async move {

            let body = serde_json::json!({ "runner": name, "runtimes": ["microvm"] }).to_string();

            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some(token), Some(body)).await;

            (st == StatusCode::OK).then(|| serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap())

        }

    };

    let decline = |token: &'static str, id: i64, reason: &'static str| {

        let app = h.app.clone();

        async move {

            let body = serde_json::json!({ "reason": reason }).to_string();

            call(&app, "POST", &format!("/api/v1/flow/jobs/{id}/decline"), Some(token), Some(body)).await.0

        }

    };

    let job = claim_as("runner-secret", "earth-1").await.expect("a job");

    // The other runners are known to the hub before they decline anything.

    assert_eq!(decline("runner-two", job, "not mine").await, StatusCode::CONFLICT, "only the runner it runs on may decline it");

    assert_eq!(decline("runner-secret", job, "the MicroVM tier needs /dev/kvm").await, StatusCode::NO_CONTENT);

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

    assert_eq!((j.status.as_str(), j.runner.as_str()), ("queued", ""));

    assert_eq!(j.waiting, "declined by earth-1: the MicroVM tier needs /dev/kvm; waiting for another runner");

    // earth-1 is offered the other ready job, not this one, while it stays the same.

    let next = claim_as("runner-secret", "earth-1").await.expect("the other ready job");

    assert_ne!(next, job);

    let page = call(&h.app, "GET", &format!("/andy/declines/flow/{}", run.number), None, None).await.1;

    assert!(page.contains("Waiting: declined by earth-1: the MicroVM tier needs /dev/kvm"), "{page}");

    assert_eq!(claim_as("runner-two", "plex-1").await, Some(job), "another runner takes it");

    assert_eq!(decline("runner-two", job, "rootfs 36fc4e323804 is not on this runner").await, StatusCode::NO_CONTENT);

    assert_eq!(claim_as("runner-three", "narvi").await, Some(job));

    assert_eq!(decline("runner-three", job, "rootfs 36fc4e323804 is for x86_64; this runner is riscv64").await, StatusCode::NO_CONTENT);

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

    assert_eq!(j.status, "error");

    assert_eq!(

        j.error,

        "unplaceable: declined by 3 runners: earth-1: the MicroVM tier needs /dev/kvm; plex-1: rootfs 36fc4e323804 is not on this runner; narvi: rootfs 36fc4e323804 is for x86_64; this runner is riscv64"

    );

    let r = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex-1").unwrap();

    let declined = db::flow::declines_of_runner(&h.app.db, r.id, 10).await.unwrap();

    assert_eq!(declined.len(), 1);

    assert_eq!((declined[0].job_id, declined[0].reason.as_str()), (job, "rootfs 36fc4e323804 is not on this runner"));

    assert_eq!(r.tier_list(), vec!["microvm"]);

}

    <dt>Tiers</dt><dd>{% if r.tier_list().is_empty() %}<span class="mute">not reported yet</span>{% else %}{{ r.tier_list().join(", ") }}{% endif %}{% if let Some(i) = r.image_list() %} · {{ i.len()|plural("image", "images") }} held{% endif %}</dd>

{% if !declines.is_empty() %}<section class="panel">

  <h2>Declined</h2>

  <p class="small mute">Jobs this runner handed back before running them, and why (OxygenFlow spec §11.5).</p>

  <div class="table"><table>

  <tr><th>Repository</th><th>Run</th><th>Job</th><th>Reason</th><th>When</th></tr>

  {% for d in declines %}<tr>

    <td><a href="/{{ d.owner }}/{{ d.name }}">{{ d.owner }}/{{ d.name }}</a></td>

    <td><a href="/{{ d.owner }}/{{ d.name }}/flow/{{ d.run_number }}">#{{ d.run_number }}</a></td>

    <td>{{ d.job }}</td>

    <td>{{ d.reason }}</td>

    <td class="small"><time title="{{ d.ms|when_i }}">{{ d.ms|ago_i }}</time></td>

  </tr>{% endfor %}

  </table></div>

</section>{% endif %}

  {% if l.j.status == "queued" && !l.j.waiting.is_empty() %}<p class="small warn pad" data-waiting>Waiting: {{ l.j.waiting }}.</p>{% else if l.lines.is_empty() && l.j.status == "queued" %}<p class="small mute pad">Waiting for a runner{% if !l.j.depends_on().is_empty() %} and for {{ l.j.depends_on().join(", ") }}{% endif %}.</p>{% endif %}