Alerts: a failed run tells its pusher and watchers; an erroring runner is drained 4a4400c5812d

-- Alerts (oxygenflow#30): when a runner's state last changed, so the

-- infrastructure errors that drain it are counted only since an

-- administrator last made it active.

ALTER TABLE flow_runners ADD COLUMN state_ms INTEGER NOT NULL DEFAULT 0;

    /// A runner whose last this many jobs all ended in an infrastructure

    /// error (status `error`, not `failed`) is drained and the

    /// administrators told (oxygenflow#30); 0 never drains one.

    pub drain_after_errors: u32,

        Flow { runner_tokens: vec![], repos: vec![], job_timeout_minutes: 60, drain_after_errors: 5 }

    sqlx::query("UPDATE flow_runners SET state = ?, state_ms = ? WHERE id = ?")

        .bind(state)

        .bind(super::now_ms())

        .bind(id)

        .execute(pool)

        .await?;

}

/// Whether runner `id`'s last `n` jobs since its state last changed all

/// ended in an infrastructure error (oxygenflow#30), with the last one's.

pub async fn erroring(pool: &Pool, id: i64, n: u32) -> sqlx::Result<Option<String>> {

    if n == 0 {

        return Ok(None);

    }

    let last: Vec<(String, String)> = sqlx::query_as(

        "SELECT j.status, j.error FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id

         WHERE j.runner_id = ? AND j.finished_ms IS NOT NULL AND j.finished_ms >= r.state_ms

         ORDER BY j.finished_ms DESC, j.id DESC LIMIT ?",

    )

    .bind(id)

    .bind(n as i64)

    .fetch_all(pool)

    .await?;

    Ok((last.len() == n as usize && last.iter().all(|(s, _)| s == "error")).then(|| last[0].1.clone()))

    let mut errored_on: Vec<i64> = Vec::new();

        if j.status == "error" {

            if let Some(r) = runner_of(app, j.id).await {

                errored_on.push(r);

            }

        }

    }

    errored_on.dedup();

    for r in errored_on {

        drain_if_erroring(app, r).await;

    }

    if let Some(done) = fdb::run_by_id(&app.db, run_id).await? {

        if done.status == "failed" && !done.live() {

            tell_failed(app, &row, &done).await;

        }

}

/// The account OxygenFlow's notices come from, made on first use with no

/// way to sign in.

pub const ACCOUNT: &str = "oxygenflow";

async fn flow_account(pool: &db::Pool) -> Option<db::accounts::Account> {

    if let Ok(Some(a)) = db::accounts::account_by_name(pool, ACCOUNT).await {

        return Some(a);

    }

    db::accounts::create_account(pool, ACCOUNT, "oxygenflow@localhost", "!", "", false).await.ok()

}

async fn runner_of(app: &Shared, job_id: i64) -> Option<i64> {

    sqlx::query_scalar("SELECT runner_id FROM flow_jobs WHERE id = ?").bind(job_id).fetch_optional(&app.db).await.ok().flatten().flatten()

}

/// A failed run (oxygenflow#30): its pusher and the repository's watchers

/// are told, in their inboxes and by email as they chose, naming the jobs

/// that did not pass.

async fn tell_failed(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) {

    let Some(me) = flow_account(&app.db).await else { return };

    let jobs = fdb::jobs(&app.db, run.id).await.unwrap_or_default();

    let broken: Vec<String> = jobs.iter().filter(|j| matches!(j.status.as_str(), "failed" | "error")).map(|j| j.name.clone()).collect();

    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();

    let on = if run.bookmark.is_empty() { String::new() } else { format!(" on {}", run.bookmark) };

    crate::notify::deliver(

        app,

        crate::notify::Event {

            kind: "flow.failed",

            title: format!("{}/{}: run #{}{on} failed ({})", row.owner, row.name, run.number, broken.join(", ")),

            url: format!("/{}/{}/flow/{}", row.owner, row.name, run.number),

            actor: &me,

            repo_id: row.id,

            thread: None,

            mention_text: "",

            direct: pusher.into_iter().map(|a| a.id).collect(),

        },

    )

    .await;

}

/// A runner whose last jobs all ended in infrastructure errors is drained,

/// so it stops taking jobs it fails before running them, and the

/// administrators are told (oxygenflow#30).

async fn drain_if_erroring(app: &Shared, runner_id: i64) {

    let n = app.config.flow.drain_after_errors;

    let Ok(Some(last)) = db::runners::erroring(&app.db, runner_id, n).await else { return };

    let Ok(Some(r)) = db::runners::get(&app.db, runner_id).await else { return };

    if r.state != "active" || db::runners::set_state(&app.db, runner_id, "draining").await.is_err() {

        return;

    }

    tracing::warn!("flow: runner {} drained after {n} errors in a row (last: {last})", r.name);

    let Some(me) = flow_account(&app.db).await else { return };

    let admins: Vec<i64> = sqlx::query_scalar("SELECT id FROM accounts WHERE is_admin = 1").fetch_all(&app.db).await.unwrap_or_default();

    crate::notify::deliver(

        app,

        crate::notify::Event {

            kind: "flow.runner_drained",

            title: format!("Runner {} drained: its last {n} jobs ended in errors (last: {last})", r.name),

            url: format!("/admin/runners/{runner_id}"),

            actor: &me,

            repo_id: 0,

            thread: None,

            mention_text: "",

            direct: admins,

        },

    )

    .await;

/// oxygenflow#30: a failed run tells its pusher and the repository's

/// watchers; a runner whose last five jobs ended in infrastructure errors

/// is drained and the administrators told, counting only since an

/// administrator last made it active.

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

async fn failed_runs_are_told_and_an_erroring_runner_is_drained() {

    let h = hub().await;

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

    let bob = db::accounts::create_account(&h.app.db, "bob", "[email protected]", "x", "salt", false).await.unwrap();

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

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

    db::notify::set_watch(&h.app.db, bob.id, row.id, true).await.unwrap();

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

    let c = client(&format!("{}/andy/alerts", h.base), work.path());

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

    write(work.path(), "test.sh", "exit 1\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", "alerts").await[0].clone();

    sqlx::query("UPDATE flow_runs SET actor = 'andy' WHERE id = ?").bind(run.id).execute(&h.app.db).await.unwrap();

    while let Some(j) = claim(&h.app).await {

        let id = j["job_id"].as_i64().unwrap();

        let (name, ph) = (j["job"]["name"].as_str().unwrap().to_string(), j["pipeline_hash"].as_str().unwrap().to_string());

        let exit = if name == "test" { 1 } else { 0 };

        finish(&h.app, id, serde_json::json!({ "exit_code": exit, "duration_ms": 10, "receipt": receipt(&name, &ph, exit as u32) })).await;

    }

    let told = |id: i64| {

        let db = h.app.db.clone();

        async move { db::notify::inbox(&db, id, 20).await.unwrap().into_iter().filter(|n| n.kind == "flow.failed").collect::<Vec<_>>() }

    };

    let (for_andy, for_bob) = (told(andy.id).await, told(bob.id).await);

    assert_eq!((for_andy.len(), for_bob.len()), (1, 1), "the pusher and a watcher");

    assert!(for_andy[0].title.contains(&format!("run #{} on main failed (test)", run.number)), "{}", for_andy[0].title);

    assert_eq!(for_andy[0].url, format!("/andy/alerts/flow/{}", run.number));

    // Five infrastructure errors in a row drain the runner.

    let runner_id = || {

        let db = h.app.db.clone();

        async move { sqlx::query_scalar::<_, i64>("SELECT id FROM flow_runners WHERE name = 'test-1'").fetch_one(&db).await.unwrap() }

    };

    let mut errors = 0;

    while errors < 5 {

        let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/alerts/flow/runs", Some("site"), Some("{}".into())).await;

        assert_eq!(st, StatusCode::CREATED);

        while let Some(j) = claim(&h.app).await {

            let id = j["job_id"].as_i64().unwrap();

            let st = finish(

                &h.app,

                id,

                serde_json::json!({ "exit_code": null, "duration_ms": 10, "error": "the microVM did not start: /dev/kvm busy" }),

            )

            .await;

            assert_eq!(st, StatusCode::NO_CONTENT);

            errors += 1;

            if errors == 5 {

                break;

            }

        }

    }

    let state = |id: i64| {

        let db = h.app.db.clone();

        async move { db::runners::get(&db, id).await.unwrap().unwrap().state }

    };

    let id = runner_id().await;

    assert_eq!(state(id).await, "draining");

    let drained: Vec<_> =

        db::notify::inbox(&h.app.db, andy.id, 50).await.unwrap().into_iter().filter(|n| n.kind == "flow.runner_drained").collect();

    assert_eq!(drained.len(), 1, "the administrator is told");

    assert!(drained[0].title.contains("/dev/kvm busy"), "{}", drained[0].title);

    assert!(

        !told(bob.id).await.is_empty()

            && db::notify::inbox(&h.app.db, bob.id, 50).await.unwrap().iter().all(|n| n.kind != "flow.runner_drained")

    );

    // Made active again, past errors no longer count.

    db::runners::set_state(&h.app.db, id, "active").await.unwrap();

    assert_eq!(db::runners::erroring(&h.app.db, id, 5).await.unwrap(), None);

}

- **Configuration.** `[flow] runner_tokens` (bearer tokens for runners, compared in constant time), `repos` (patterns such as `owner/*` or `owner/name`: the repositories whose pushes start runs by default; empty means none) `job_timeout_minutes` (60) and `drain_after_errors` (5, 0 for never: see Alerts).

- **Alerts** (oxygenflow#30). A run that ends failed tells its pusher and the repository's watchers, in their inboxes and by email as each chose (`flow.failed`, from the `oxygenflow` account, which cannot sign in), naming the jobs that did not pass. A runner whose last `drain_after_errors` jobs since its state last changed all ended in an infrastructure error (status `error`: it could not run them, not a failing test) is set to draining and the site administrators are told (`flow.runner_drained`, with the last error); making it active again on Admin › Runners starts the count afresh (migration 0064, `state_ms`).