Runner updates: a build that rolls back on one runner is held faf5246c900e

-- Builds held back from runners (oxygenhub#326): one rolled back on a

-- runner (it did not reach the hub after updating), so no other runner is

-- offered it until an administrator lifts the hold.

CREATE TABLE runner_update_holds (

    version TEXT PRIMARY KEY,

    runner  TEXT NOT NULL,

    why     TEXT NOT NULL,

    at_ms   BIGINT NOT NULL

);

-- Builds held back from runners (oxygenhub#326): one rolled back on a

-- runner (it did not reach the hub after updating), so no other runner is

-- offered it until an administrator lifts the hold.

CREATE TABLE runner_update_holds (

    version TEXT PRIMARY KEY,

    runner  TEXT NOT NULL,

    why     TEXT NOT NULL,

    at_ms   INTEGER NOT NULL

);

- **A build that rolls back on one runner is held for the rest (#326):** when a runner reports it rolled back from flow X (it did not reach the hub after updating, or would not start), the hub offers X to no other runner, saying why, until an administrator lifts the hold on Admin › Runners › Updates, which lists each held build with the runner and the reason. One failed canary halts the rollout. Migration 0097.

/// A build held back from every runner (#326).

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

pub struct Hold {

    pub version: String,

    /// The runner it rolled back on.

    pub runner: String,

    pub why: String,

    pub at_ms: i64,

}

/// Hold `version` back, the first time it rolls back.

pub async fn hold(pool: &Pool, version: &str, runner: &str, why: &str) -> sqlx::Result<()> {

    sqlx::query("INSERT INTO runner_update_holds (version, runner, why, at_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (version) DO NOTHING")

        .bind(version)

        .bind(runner)

        .bind(why)

        .bind(now_ms())

        .execute(pool)

        .await?;

    Ok(())

}

pub async fn holds(pool: &Pool) -> sqlx::Result<Vec<Hold>> {

    sqlx::query_as("SELECT * FROM runner_update_holds ORDER BY at_ms DESC").fetch_all(pool).await

}

pub async fn held(pool: &Pool, version: &str) -> sqlx::Result<Option<Hold>> {

    sqlx::query_as("SELECT * FROM runner_update_holds WHERE version = $1").bind(version).fetch_optional(pool).await

}

/// Lift the hold on `version`; whether there was one.

pub async fn lift(pool: &Pool, version: &str) -> sqlx::Result<bool> {

    Ok(sqlx::query("DELETE FROM runner_update_holds WHERE version = $1").bind(version).execute(pool).await?.rows_affected() > 0)

}

    // A build that rolled back here is held back from every runner (#326).

    if let Some(v) = crate::runner_updates::rolled_back_version(note) {

        let why = note.split_once(": ").map(|(_, w)| w).unwrap_or("rolled back");

        db::runner_updates::hold(&app.db, &v, &me.name, why).await?;

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

    }

/// The version a runner's report says rolled back: `rolled back from flow

/// 0.4.0 to 0.3.1: …` (the runner's words, oxygenflow spec §11.5.3) → `0.4.0`.

pub fn rolled_back_version(note: &str) -> Option<String> {

    let v = note.strip_prefix("rolled back from flow ")?.split_whitespace().next()?;

    v.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())).then(|| v.to_string())

}

    }

    if let Some(h) = rdb::held(pool, &want).await? {

        return Ok((

            None,

            format!("flow {want} is held: it rolled back on {} ({}); lift the hold on Admin › Runners › Updates", h.runner, h.why),

        ));

        assert_eq!(

            rolled_back_version("rolled back from flow 0.4.0 to 0.3.1: 3 starts without reaching the hub").as_deref(),

            Some("0.4.0")

        );

        assert_eq!(rolled_back_version("runs flow 0.4.0 (was 0.3.1); proven"), None);

        .route("/admin/runners/updates/holds/lift", post(runner_updates::lift_hold))

    /// Builds held back after rolling back on a runner (#326).

    holds: Vec<rdb::Hold>,

    let holds = rdb::holds(&app.db).await.map_err(e)?;

    Ok(Page(UpdatesPage { shell, repo, site: v.site, owners: v.owners, groups: v.groups, runners: v.runners, holds, notice, error }))

#[derive(Deserialize)]

pub struct LiftForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    version: String,

}

/// `POST /admin/runners/updates/holds/lift`: offer a held build again (#326).

pub async fn lift_hold(State(app): State<Shared>, viewer: Viewer, Form(f): Form<LiftForm>) -> Result<Page<UpdatesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/updates", &viewer);

    let me = require_admin(&viewer, &shell)?;

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

    let v = f.version.trim();

    if !rdb::lift(&app.db, v).await.map_err(|e| Error::from_db(shell.clone(), e))? {

        return page(&app, shell, None, Some(format!("flow {v} is not held."))).await;

    }

    accounts::audit(&app.db, Some(&me), "runner.update.lifted", v, "offered again").await.ok();

    page(&app, shell, Some(format!("flow {v} is offered again where the settings say so.")), None).await

}

    // #326: a build that rolls back on one runner is held for every other.

    let report = serde_json::json!({ "note": "rolled back from flow 0.2.0 to 0.1.0: 3 starts without reaching the hub" }).to_string();

    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runner/update", Some(&creds[0]), Some(report)).await.0, StatusCode::NO_CONTENT);

    let a = ask(0).await;

    assert!(a.offer.is_none() && a.note.contains("flow 0.2.0 is held: it rolled back on plex (3 starts"), "{a:?}");

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

    assert!(page.contains("Held builds") && page.contains("<strong>flow 0.2.0</strong>"), "{page}");

    let (_, page) = web("POST", "/admin/runners/updates/holds/lift", Some(format!("_csrf={csrf}&version=0.2.0"))).await;

    assert!(page.contains("flow 0.2.0 is offered again"), "{page}");

    assert!(ask(0).await.note.contains("waits for its window"), "lifted: back to what its settings say");

    let (_, page) = web("POST", "/admin/runners/updates/holds/lift", Some(format!("_csrf={csrf}&version=0.2.0"))).await;

    assert!(page.contains("flow 0.2.0 is not held."), "{page}");

{% if !holds.is_empty() %}

<section class="panel strip warn" id="holds">

  <h2>Held builds</h2>

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

  {% for h in holds %}

  <form class="pol-form" method="post" action="/admin/runners/updates/holds/lift"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="hidden" name="version" value="{{ h.version }}">

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

  </form>

  {% endfor %}

</section>

{% endif %}