Runner updates in waves: a later wave takes a version once the waves before it proved it; an error stops it (refs andymccall/oxygenflow#84) 1ec29691b67a

-- Runner updates in waves (oxygenflow#84): the runner build each job ran

-- on, from its claim, to count the jobs that prove a version; and the wave

-- a test group's runners take a version in (none: after every wave).

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

ALTER TABLE runner_groups ADD COLUMN wave BIGINT;

CREATE INDEX flow_jobs_runner_version ON flow_jobs (runner_version) WHERE runner_version <> '';

-- Runner updates in waves (oxygenflow#84): the runner build each job ran

-- on, from its claim, to count the jobs that prove a version; and the wave

-- a test group's runners take a version in (none: after every wave).

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

ALTER TABLE runner_groups ADD COLUMN wave INTEGER;

CREATE INDEX flow_jobs_runner_version ON flow_jobs (runner_version) WHERE runner_version <> '';

## Unreleased

- **Runner updates in waves (oxygenflow#84).** A test group can be given a wave (1 to 9) on Admin › Runners › Updates or an organisation's settings: a new version reaches wave 2's runners only once every runner of wave 1 that took it has finished a number of jobs on it (5 by default) with none ending in error, or a number of hours (24) has passed since its first job there; runners in no wave come last. A job that ends in `error` on a wave's runner (the runner's or the machine's fault; failed tests do not count) stops the version: the next runner to ask holds it for every runner, and the page names the job, until an administrator lifts the hold. Each job now keeps the runner build it was claimed by (migration 0102). A runner waiting says so: "flow 0.5.1 waits for wave 1 to prove it: 1 of 2 runners on it, 3 jobs". First slice: errors are not yet weighed against the version before, and a stopped wave's runners are not rolled back by the hub.

    /// The wave it takes a version in (oxygenflow#84); `None`: after every wave.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub wave: Option<i64>,

                wave: g.g.wave,

/// Keep the runner build job `id` was claimed by (oxygenflow#84), for

/// counting the jobs that prove a version in a wave.

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

    sqlx::query("UPDATE flow_jobs SET runner_version = $1 WHERE id = $2").bind(version).bind(id).execute(pool).await?;

    Ok(())

}

    /// The wave its runners take a version in (oxygenflow#84): wave 1

    /// first, each later one once those before it have proved the version;

    /// `None`: after every wave.

    pub wave: Option<i64>,

pub async fn set_wave(pool: &Pool, id: i64, wave: Option<i64>) -> sqlx::Result<()> {

    sqlx::query("UPDATE runner_groups SET wave = $1 WHERE id = $2").bind(wave).bind(id).execute(pool).await?;

    Ok(())

}

}

/// How a runner has done on a version (oxygenflow#84): its finished jobs

/// on that build, how many ended in `error` (the runner's or the

/// infrastructure's fault, not a test's), and when the first started.

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

pub struct Soak {

    pub runner_id: i64,

    pub jobs: i64,

    pub errors: i64,

    pub first_ms: Option<i64>,

}

/// Every runner's [`Soak`] on `version`.

pub async fn soaks(pool: &Pool, version: &str) -> sqlx::Result<Vec<Soak>> {

    sqlx::query_as(

        "SELECT runner_id, COUNT(*) AS jobs, SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS errors, MIN(started_ms) AS first_ms

         FROM flow_jobs WHERE runner_version = $1 AND runner_id IS NOT NULL AND status IN ('passed', 'failed', 'error')

         GROUP BY runner_id",

    )

    .bind(version)

    .fetch_all(pool)

    .await

}

/// The first job on `version` that ended in error on runner `runner`, in

/// words: `job build of OWNER/NAME run #N: why`.

pub async fn first_error(pool: &Pool, version: &str, runner: i64) -> sqlx::Result<Option<String>> {

    let row: Option<(String, String, String, i64, String)> = sqlx::query_as(

        "SELECT j.name, p.owner, p.name, r.number, j.error FROM flow_jobs j

         JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id

         WHERE j.runner_version = $1 AND j.runner_id = $2 AND j.status = 'error'

         ORDER BY j.finished_ms LIMIT 1",

    )

    .bind(version)

    .bind(runner)

    .fetch_optional(pool)

    .await?;

    Ok(row.map(|(job, owner, name, n, error)| {

        let why: String = error.chars().take(120).collect();

        format!("job {job} of {owner}/{name} run #{n}{}", if why.is_empty() { String::new() } else { format!(": {why}") })

    }))

    // The build it runs on, for waves of runner updates (oxygenflow#84).

    fdb::set_runner_version(&app.db, job.id, &crate::runner_updates::version_of(&me_now.version)).await?;

    let levels = crate::runner_updates::Levels::load(&app.db).await?;

    let policy = levels.resolve(&me);

    let (offer, note) = crate::runner_updates::offer(&app.db, &levels, &me, &policy, hour, db::now_ms(), true).await?;

        db::runner_updates::hold(&app.db, &v, &me.name, &format!("it rolled back on {} ({why})", me.name)).await?;

        db::accounts::audit(&app.db, None, "runner.update.held", &v, &format!("it rolled back on {}: {why}", me.name)).await.ok();

use crate::db::runner_updates::{self as rdb, Group, Policy, Soak};

/// The site settings for proving a version in a wave (oxygenflow#84): the

/// jobs each runner of the wave that took it must finish, or the hours

/// since its first job there after which one job is enough.

pub const SETTING_SOAK_JOBS: &str = "runner_updates_soak_jobs";

pub const SETTING_SOAK_HOURS: &str = "runner_updates_soak_hours";

pub const SOAK_JOBS: i64 = 5;

pub const SOAK_HOURS: i64 = 24;

    /// The wave `r` takes a version in: the first of its groups' waves;

    /// `None` after every wave.

    pub fn wave_of(&self, r: &Runner) -> Option<i64> {

        self.groups_of(r).iter().filter_map(|g| g.wave).min()

    }

    /// The waves before `r`'s, in order, each with its runners among `all`:

    /// the waves of the groups that may hold `r` (the site's, and its

    /// owner's), a runner counted in its own first wave.

    pub fn waves_before<'a>(&self, r: &Runner, all: &'a [Runner]) -> Vec<(i64, Vec<&'a Runner>)> {

        let mine = self.wave_of(r);

        let mut waves: Vec<i64> =

            self.groups.iter().filter(|g| in_reach(g, r)).filter_map(|g| g.wave).filter(|w| mine.is_none_or(|m| *w < m)).collect();

        waves.sort();

        waves.dedup();

        waves

            .into_iter()

            .map(|w| {

                let members = all

                    .iter()

                    .filter(|o| self.wave_of(o) == Some(w) && self.groups_of(o).iter().any(|g| g.wave == Some(w) && in_reach(g, r)))

                    .collect();

                (w, members)

            })

            .collect()

    }

}

/// Where a wave stands with a version (oxygenflow#84).

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

pub enum WaveState {

    /// Every runner of the wave that took it has finished its soak jobs on

    /// it, or the soak time has passed since its first job there; or the

    /// wave has no runners.

    Proved,

    /// Not yet: `on` of `runners` run it, with `jobs` jobs between them.

    Soaking { runners: usize, on: usize, jobs: i64 },

    /// A job on it ended in error on this runner: the version stops here.

    Stopped { runner_id: i64 },

}

/// Where a wave of `members` stands with a version, from each runner's

/// [`Soak`] on it, at `now` (ms).

pub fn wave_state(members: &[i64], soaks: &[Soak], soak_jobs: i64, soak_ms: i64, now: i64) -> WaveState {

    let mine: Vec<&Soak> = soaks.iter().filter(|s| members.contains(&s.runner_id) && s.jobs > 0).collect();

    if let Some(s) = mine.iter().find(|s| s.errors > 0) {

        return WaveState::Stopped { runner_id: s.runner_id };

    }

    if members.is_empty() {

        return WaveState::Proved;

    }

    let jobs = mine.iter().map(|s| s.jobs).sum();

    let soaked = mine.iter().filter_map(|s| s.first_ms).min().is_some_and(|first| now - first >= soak_ms);

    if !mine.is_empty() && (soaked || mine.iter().all(|s| s.jobs >= soak_jobs)) {

        WaveState::Proved

    } else {

        WaveState::Soaking { runners: members.len(), on: mine.len(), jobs }

    }

}

/// What the waves before `r`'s say about `version`: `Ok(None)` go ahead;

/// `Ok(Some(why))` wait; `Err((runner, why))` it stopped on that runner.

async fn wave_gate(

    pool: &Pool,

    levels: &Levels,

    r: &Runner,

    version: &str,

    now: i64,

) -> sqlx::Result<Result<Option<String>, (String, String)>> {

    let all = db::runners::list(pool).await?;

    let waves = levels.waves_before(r, &all);

    if waves.is_empty() {

        return Ok(Ok(None));

    }

    let setting = |v: Option<String>, d: i64| v.and_then(|s| s.trim().parse::<i64>().ok()).filter(|n| *n > 0).unwrap_or(d);

    let soak_jobs = setting(db::accounts::setting(pool, SETTING_SOAK_JOBS).await?, SOAK_JOBS);

    let soak_hours = setting(db::accounts::setting(pool, SETTING_SOAK_HOURS).await?, SOAK_HOURS);

    let soaks = rdb::soaks(pool, version).await?;

    for (w, members) in waves {

        let ids: Vec<i64> = members.iter().map(|m| m.id).collect();

        match wave_state(&ids, &soaks, soak_jobs, soak_hours * 3_600_000, now) {

            WaveState::Proved => continue,

            WaveState::Stopped { runner_id } => {

                let name = members.iter().find(|m| m.id == runner_id).map(|m| m.name.clone()).unwrap_or_default();

                let job = rdb::first_error(pool, version, runner_id).await?.unwrap_or_else(|| "a job".into());

                return Ok(Err((name.clone(), format!("wave {w} stopped it on {name}: {job} ended in error"))));

            }

            WaveState::Soaking { runners, on, jobs } => {

                return Ok(Ok(Some(format!(

                    "flow {version} waits for wave {w} to prove it: {on} of {runners} runner{} on it, {jobs} job{} (each needs {soak_jobs} without an error, or {soak_hours} h)",

                    if runners == 1 { "" } else { "s" },

                    if jobs == 1 { "" } else { "s" },

                ))));

            }

        }

    }

    Ok(Ok(None))

/// why there is none. A version a wave stopped is held back from every

/// runner when `act` (a runner asking; a page only says so).

pub async fn offer(

    pool: &Pool,

    levels: &Levels,

    r: &Runner,

    policy: &UpdatePolicy,

    hour: u8,

    now: i64,

    act: bool,

) -> sqlx::Result<(Option<UpdateOffer>, String)> {

        return Ok((None, format!("flow {want} is held: {}; lift the hold on Admin › Runners › Updates", h.why)));

    }

    // Waves (oxygenflow#84): a later wave takes a version once those before it proved it.

    if !going_back {

        match wave_gate(pool, levels, r, &want, now).await? {

            Ok(None) => {}

            Ok(Some(why)) => return Ok((None, why)),

            Err((runner, why)) => {

                if act {

                    rdb::hold(pool, &want, &runner, &why).await?;

                    db::accounts::audit(pool, None, "runner.update.held", &want, &why).await.ok();

                }

                return Ok((None, format!("flow {want} is held: {why}; lift the hold on Admin › Runners › Updates")));

            }

        }

        l.groups.push(Group { id: 1, owner: "acme".into(), name: "canary".into(), percent: None, created_ms: 0, wave: None });

        l.groups.push(Group { id: 5, owner: String::new(), name: "farm 20%".into(), percent: Some(20), created_ms: 0, wave: None });

    }

    #[test]

    fn a_wave_proves_a_version_by_its_jobs_or_its_time_and_an_error_stops_it() {

        let soak = |runner_id, jobs, errors, first_ms| Soak { runner_id, jobs, errors, first_ms: Some(first_ms) };

        let hour = 3_600_000;

        // Nobody has taken it yet: not proved.

        assert_eq!(wave_state(&[1, 2], &[], 5, 24 * hour, 0), WaveState::Soaking { runners: 2, on: 0, jobs: 0 });

        // One runner took it and ran 3 of its 5; the other has not updated.

        let s = [soak(1, 3, 0, 0), soak(9, 50, 0, 0)];

        assert_eq!(wave_state(&[1, 2], &s, 5, 24 * hour, hour), WaveState::Soaking { runners: 2, on: 1, jobs: 3 });

        // Five each on every runner that took it proves it.

        assert_eq!(wave_state(&[1, 2], &[soak(1, 5, 0, 0)], 5, 24 * hour, hour), WaveState::Proved);

        // Or a day since the first job there, with no error.

        assert_eq!(wave_state(&[1, 2], &[soak(1, 1, 0, 0)], 5, 24 * hour, 24 * hour), WaveState::Proved);

        // An error anywhere in the wave stops it, however many passed.

        assert_eq!(

            wave_state(&[1, 2], &[soak(1, 40, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, 48 * hour),

            WaveState::Stopped { runner_id: 2 }

        );

        // Another wave's errors are not this one's.

        assert_eq!(wave_state(&[1], &[soak(1, 5, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, hour), WaveState::Proved);

        // An empty wave holds nothing up.

        assert_eq!(wave_state(&[], &[], 5, 24 * hour, 0), WaveState::Proved);

    }

    #[test]

    fn a_runner_waits_for_the_waves_before_its_own() {

        let mut l = Levels::default();

        let group = |id, name: &str, wave| Group { id, owner: String::new(), name: name.into(), percent: None, created_ms: 0, wave };

        l.groups.extend([group(1, "first", Some(1)), group(2, "second", Some(2)), group(3, "watchers", None)]);

        l.chosen.extend([(1, 1), (2, 2), (2, 1), (3, 3)]);

        let all: Vec<Runner> = (1..=4).map(|id| runner(id, "")).collect();

        let ids = |ws: Vec<(i64, Vec<&Runner>)>| {

            ws.into_iter().map(|(w, rs)| (w, rs.iter().map(|r| r.id).collect::<Vec<_>>())).collect::<Vec<_>>()

        };

        assert_eq!(l.wave_of(&all[0]), Some(1), "its first wave");

        assert!(l.waves_before(&all[0], &all).is_empty(), "wave 1 goes first");

        assert_eq!(ids(l.waves_before(&all[1], &all)), vec![(1, vec![1])]);

        assert_eq!(l.wave_of(&all[2]), None, "a group with no wave");

        assert_eq!(ids(l.waves_before(&all[2], &all)), vec![(1, vec![1]), (2, vec![2])], "in no wave: after every wave");

        assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])]);

        // An owner's waves gate only that owner's runners.

        l.groups.push(Group { id: 4, owner: "acme".into(), name: "acme first".into(), percent: None, created_ms: 0, wave: Some(1) });

        l.chosen.insert((4, 5));

        let mut all = all;

        all.push(runner(5, "acme"));

        all.push(runner(6, "acme"));

        assert_eq!(ids(l.waves_before(&all[5], &all)), vec![(1, vec![1, 5]), (2, vec![2])]);

        assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])], "not another owner's");

    /// Proving a version in a wave (oxygenflow#84).

    soak_jobs: i64,

    soak_hours: i64,

        let (_, offer) = ru::offer(&app.db, &levels, &r, &effective, hour, crate::db::now_ms(), false).await?;

    let number = |v: Option<String>, d: i64| v.and_then(|s| s.trim().parse::<i64>().ok()).filter(|n| *n > 0).unwrap_or(d);

    let soak_jobs = number(accounts::setting(&app.db, ru::SETTING_SOAK_JOBS).await.map_err(e)?, ru::SOAK_JOBS);

    let soak_hours = number(accounts::setting(&app.db, ru::SETTING_SOAK_HOURS).await.map_err(e)?, ru::SOAK_HOURS);

    Ok(Page(UpdatesPage {

        shell,

        repo,

        site: v.site,

        owners: v.owners,

        groups: v.groups,

        runners: v.runners,

        holds,

        soak_jobs,

        soak_hours,

        notice,

        error,

    }))

    #[serde(default)]

    soak_jobs: String,

    #[serde(default)]

    soak_hours: String,

    // Waves (oxygenflow#84): blank keeps the default.

    for (key, value, max, what) in [

        (ru::SETTING_SOAK_JOBS, &f.soak_jobs, 1000, "jobs each runner of a wave finishes"),

        (ru::SETTING_SOAK_HOURS, &f.soak_hours, 720, "hours a wave soaks"),

    ] {

        let v = value.trim();

        if !v.is_empty() && !v.parse::<i64>().is_ok_and(|n| (1..=max).contains(&n)) {

            return page(&app, shell, None, Some(format!("The {what} is a whole number from 1 to {max}."))).await;

        }

        accounts::set_setting(&app.db, key, v).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    }

    /// Its wave (oxygenflow#84), 1 to 9; blank for none.

    #[serde(default)]

    wave: String,

}

/// The wave a form gives, or why it cannot be one.

fn wave_of(f: &GroupForm) -> Result<Option<i64>, String> {

    let w = f.wave.trim();

    if w.is_empty() {

        return Ok(None);

    }

    match w.parse::<i64>() {

        Ok(n) if (1..=9).contains(&n) => Ok(Some(n)),

        _ => Err("A wave is a whole number from 1 to 9, or blank for after every wave.".into()),

    }

    let wave = match wave_of(f) {

        Ok(w) => w,

        Err(why) => return Ok(Err(why)),

    };

    rdb::set_wave(&app.db, g.id, wave).await?;

    let mut what = match percent {

    if let Some(w) = wave {

        what.push_str(&format!(", wave {w}"));

    }

    let wave = match wave_of(f) {

        Ok(w) => w,

        Err(why) => return Ok(Err(why)),

    };

    rdb::set_wave(&app.db, id, wave).await?;

    let mut what = match percent {

    if let Some(w) = wave {

        what.push_str(&format!(" in wave {w}"));

    }

    // The build it ran on is kept with the job, for waves of updates (oxygenflow#84).

    let kept: String =

        sqlx::query_scalar("SELECT runner_version FROM flow_jobs WHERE runner_id = $1").bind(plex.id).fetch_one(&h.app.db).await.unwrap();

    assert_eq!(kept, "0.1.0");

/// oxygenflow#84: a version reaches a later wave once every runner of the

/// waves before it that took it has run its soak jobs on it; a job ending

/// in error there stops it for every runner.

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

async fn a_version_reaches_wave_two_once_wave_one_proves_it() {

    use oxy_core::identity::SecretKey;

    let h = hub().await;

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

    let res = oxygenhub::router(h.app.clone())

        .oneshot(

            Request::post("/register")

                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery&setup=test-setup"))

                .unwrap(),

        )

        .await

        .unwrap();

    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();

    let admin = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();

    db::accounts::set_admin(&h.app.db, admin.id, true).await.unwrap();

    let web = |method: &'static str, path: &str, body: Option<String>| {

        let (app, cookie, path) = (h.app.clone(), cookie.clone(), path.to_string());

        async move {

            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);

            if body.is_some() {

                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");

            }

            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();

            (res.status(), String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

        }

    };

    let (_, page) = web("GET", "/admin/runners/updates", None).await;

    let csrf = {

        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();

        page[i..].split('"').next().unwrap().to_string()

    };

    let mut creds = vec![];

    for name in ["plex", "narvi"] {

        let (_, page) = web("POST", "/admin/runners/token", Some(format!("_csrf={csrf}"))).await;

        let i = page.find("--token ").unwrap() + "--token ".len();

        let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();

        let identity = SecretKey::generate().identity(&format!("{name} (OxygenFlow runner)"), "", 1);

        let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0 (build)", "max_jobs": 1 });

        let reg = serde_json::json!({ "token": token, "name": name, "identity": hex::encode(identity.encode()), "info": info });

        let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;

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

        creds.push(serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string());

    }

    let ask = |who: usize| {

        let (app, cred) = (h.app.clone(), creds[who].clone());

        async move {

            let (st, body) = call(&app, "GET", "/api/v1/flow/runner/update", Some(&cred), None).await;

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

            serde_json::from_str::<flow_core::api::UpdateAnswer>(&body).unwrap()

        }

    };

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

    let rel = db::releases::create(&h.app.db, repo.id, "v0.2.0", "0.2.0", "", false, None).await.unwrap();

    db::releases::add_asset(&h.app.db, rel.id, "flow-linux-x86_64", 42, &"ab".repeat(32), "application/octet-stream").await.unwrap();

    let (_, page) =

        web("POST", "/admin/runners/updates/source", Some(format!("_csrf={csrf}&repo=andy%2Foxygenflow&soak_jobs=2&soak_hours=48"))).await;

    assert!(page.contains("runner builds come from andy/oxygenflow") && page.contains("value=\"2\""), "{page}");

    let (_, page) = web("POST", "/admin/runners/updates/source", Some(format!("_csrf={csrf}&repo=andy%2Foxygenflow&soak_jobs=0"))).await;

    assert!(page.contains("is a whole number from 1 to 1000"), "{page}");

    web("POST", "/admin/runners/updates/policy", Some(format!("_csrf={csrf}&scope=site&updates=on"))).await;

    // plex in wave 1, narvi in wave 2.

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

    let id = |name: &str| runners.iter().find(|r| r.name == name).unwrap().id;

    let (_, page) =

        web("POST", "/admin/runners/updates/groups", Some(format!("_csrf={csrf}&name=first&mode=chosen&runner={}&wave=1", id("plex"))))

            .await;

    assert!(page.contains("Test group first made: 1 chosen runners, wave 1") && page.contains(" · wave 1</small>"), "{page}");

    let (_, page) =

        web("POST", "/admin/runners/updates/groups", Some(format!("_csrf={csrf}&name=rest&mode=chosen&runner={}&wave=0", id("narvi"))))

            .await;

    assert!(page.contains("A wave is a whole number from 1 to 9"), "{page}");

    web("POST", "/admin/runners/updates/groups", Some(format!("_csrf={csrf}&name=rest&mode=chosen&runner={}&wave=2", id("narvi")))).await;

    assert_eq!(ask(0).await.offer.map(|o| o.version).as_deref(), Some("0.2.0"), "wave 1 goes first");

    let b = ask(1).await;

    assert!(b.offer.is_none() && b.note.contains("flow 0.2.0 waits for wave 1 to prove it: 0 of 1 runner on it, 0 jobs"), "{b:?}");

    // plex runs 0.2.0: one job is not enough, two are.

    let run: i64 = sqlx::query_scalar(

        "INSERT INTO flow_runs (repo_id, bookmark, state, heads, status, created_ms, number) VALUES ($1, 'main', 's', 'h', 'passed', 1, 1) RETURNING id",

    )

    .bind(repo.id)

    .fetch_one(&h.app.db)

    .await

    .unwrap();

    let job = |name: &'static str, status: &'static str| {

        let (app, plex) = (h.app.clone(), id("plex"));

        async move {

            sqlx::query(

                "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, runner, runner_id, started_ms, finished_ms, error, runner_version)

                 VALUES ($1, $2, 0, '', '{}', 'microvm', 'microvm', $3, 'plex', $4, $5, $5, $6, '0.2.0')",

            )

            .bind(run)

            .bind(name)

            .bind(status)

            .bind(plex)

            .bind(db::now_ms())

            .bind(if status == "error" { "the runner lost its disk" } else { "" })

            .execute(&app.db)

            .await

            .unwrap();

        }

    };

    job("build", "passed").await;

    assert!(ask(1).await.note.contains("1 of 1 runner on it, 1 job (each needs 2"), "still soaking");

    job("test", "failed").await;

    assert_eq!(ask(1).await.offer.map(|o| o.version).as_deref(), Some("0.2.0"), "a failed test is not the runner's fault: proved");

    let (_, page) = web("GET", "/admin/runners/updates", None).await;

    assert!(!page.contains("waits for wave"), "{page}");

    // An error on wave 1 stops it: the page only says so; the next runner to ask holds it for all.

    job("lint", "error").await;

    let (_, page) = web("GET", "/admin/runners/updates", None).await;

    assert!(page.contains("wave 1 stopped it on plex: job lint of andy/oxygenflow run #1: the runner lost its disk"), "{page}");

    assert!(db::runner_updates::held(&h.app.db, "0.2.0").await.unwrap().is_none(), "a page view holds nothing");

    let b = ask(1).await;

    assert!(b.offer.is_none() && b.note.contains("flow 0.2.0 is held: wave 1 stopped it on plex"), "{b:?}");

    assert!(ask(0).await.note.contains("is held"), "wave 1 too");

    let (_, page) = web("GET", "/admin/runners/updates", None).await;

    assert!(page.contains("Held builds") && page.contains("wave 1 stopped it on plex"), "{page}");

}

                let mut what = g.percent.map(|p| format!("{p}% of {of}")).unwrap_or_else(|| format!("chosen, of {of}"));

                if let Some(w) = g.wave {

                    what.push_str(&format!(", wave {w}"));

                }

            groups: vec![hub_api::RunnerGroup {

                id: 4,

                owner: "acme".into(),

                name: "canary".into(),

                percent: None,

                members: vec![],

                wave: None,

            }],

| `oh runner-updates [show]` | Site administrators: runners that update themselves (oxygenhub#322): where builds come from, every level's settings, test groups and their waves (oxygenflow#84), each runner's policy and what it is offered (or which wave it waits for), held builds |

  <h3>Waves</h3>

  <p class="small mute">Give test groups a wave (1, 2, …) and a new version reaches them in order: wave 2's runners are offered it once every runner of wave 1 that took it has finished this many jobs on it with none ending in error, or this many hours have passed since its first job there. A job that ends in error (the runner's or the machine's fault, not a test's) on a wave stops the version for everyone until you lift it. Runners in no wave come last.</p>

  <div class="pol-row">

    <label class="pol">Jobs each <input type="number" name="soak_jobs" min="1" max="1000" size="4" value="{{ soak_jobs }}"></label>

    <label class="pol">or hours <input type="number" name="soak_hours" min="1" max="720" size="4" value="{{ soak_hours }}"></label>

  </div>

  <p class="small">A build that rolled back on a runner, or that a wave stopped, is offered to no other until you lift the hold.</p>

    <strong>flow {{ h.version }}</strong><span class="small">{{ h.why }} · <time title="{{ h.at_ms|when_i }}">{{ h.at_ms|ago_i }}</time></span><button type="submit" class="small">LIFT</button>

    <h3>{{ g.name }} <small class="mute">{{ gv.reach }} · {% if let Some(p) = g.percent %}{{ p }}%{% else %}chosen{% endif %} · {{ gv.members.len()|plural("runner", "runners") }} now{% if let Some(w) = g.wave %} · wave {{ w }}{% endif %}</small></h3>

      <label class="pol">Wave <input type="number" name="wave" min="1" max="9" size="3" value="{% if let Some(w) = g.wave %}{{ w }}{% endif %}" title="Blank: after every wave"></label>

    <label class="pol">Wave <input type="number" name="wave" min="1" max="9" size="3" title="Blank: after every wave"></label>

    <h3>{{ g.name }} <small class="mute">{% if let Some(p) = g.percent %}{{ p }}%{% else %}chosen{% endif %} · {{ gv.members.len()|plural("runner", "runners") }} now{% if let Some(w) = g.wave %} · wave {{ w }}{% endif %}</small></h3>

      <label class="pol">Wave <input type="number" name="wave" min="1" max="9" size="3" value="{% if let Some(w) = g.wave %}{{ w }}{% endif %}" title="Blank: after every wave"></label>

    <label class="pol">Wave <input type="number" name="wave" min="1" max="9" size="3" title="Blank: after every wave"></label>