Runners hand jobs back, and the queue takes them first (closes #312, refs andymccall/oxygenflow#76) ab4df6aae179

-- Jobs a runner handed back (oxygenflow#76, oxygenhub#312): drained before

-- it left, or its lease lapsed. Queued again ahead of new work; NULL if it

-- never was.

ALTER TABLE flow_jobs ADD COLUMN handed_back_ms BIGINT;

-- How many times; past a few it ends as an error rather than bounce on.

ALTER TABLE flow_jobs ADD COLUMN handbacks BIGINT NOT NULL DEFAULT 0;

-- Jobs a runner handed back (oxygenflow#76, oxygenhub#312): drained before

-- it left, or its lease lapsed. Queued again ahead of new work; NULL if it

-- never was.

ALTER TABLE flow_jobs ADD COLUMN handed_back_ms INTEGER;

-- How many times; past a few it ends as an error rather than bounce on.

ALTER TABLE flow_jobs ADD COLUMN handbacks INTEGER NOT NULL DEFAULT 0;

- **Runners hand jobs back, and the queue takes them first (#312, oxygenflow#76):** `POST /api/v1/flow/jobs/{id}/release` {reason} lets a draining runner hand back a job it holds as it leaves (a phone off Wi-Fi, a restart, a spot reclaim). The job is queued again **ahead of new work**, whatever its priority; its log says who handed it back and why, and its next attempt's log follows on; it is not a decline. A job whose runner the hub has not heard from for 3 minutes (claims, 30-second heartbeats, logs) is handed back the same way, rather than running on until its time limit and ending as lost. A job handed back more than 5 times ends as an error saying so. Migration 0094 adds `flow_jobs.handed_back_ms` and `handbacks`.

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

    /// When a runner last handed it back, drained or with its lease lapsed

    /// (oxygenflow#76, migration 0094), and how many times: such a job is

    /// queued ahead of new work.

    #[sqlx(default)]

    pub handed_back_ms: Option<i64>,

    #[sqlx(default)]

    pub handbacks: i64,

         ORDER BY CASE WHEN j.handed_back_ms IS NULL THEN 1 ELSE 0 END, {PRIORITY_RANK}, \

           ROW_NUMBER() OVER (PARTITION BY r.repo_id ORDER BY j.run_id, j.depth, j.id) \

}

/// Job `job_id`, running on runner `runner_id` (any runner when `None`),

/// goes back to the queue ahead of new work (oxygenflow#76): drained, or

/// its runner's lease lapsed. Its log is kept and its next attempt's

/// follows on. Returns how many times it has now been handed back, or

/// `None` when it was not running there.

pub async fn hand_back(pool: &Pool, job_id: i64, runner_id: Option<i64>) -> sqlx::Result<Option<i64>> {

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

    let now = now_ms();

    let back = sqlx::query(

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

         handed_back_ms = $1, handbacks = handbacks + 1, queued_ms = $2, ready_ms = NULL, placed = '' \

         WHERE id = $3 AND status = 'running' AND ($4 IS NULL OR runner_id = $5)",

    )

    .bind(now)

    .bind(now)

    .bind(job_id)

    .bind(runner_id)

    .bind(runner_id)

    .execute(&mut *tx)

    .await?;

    if back.rows_affected() != 1 {

        tx.rollback().await?;

        return Ok(None);

    }

    let n: i64 = sqlx::query_scalar("SELECT handbacks FROM flow_jobs WHERE id = $1").bind(job_id).fetch_one(&mut *tx).await?;

    tx.commit().await?;

    Ok(Some(n))

}

/// Jobs running on a runner the hub has not heard from since `before`

/// (claims, heartbeats, logs): their leases lapsed (oxygenflow#76).

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

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

        "SELECT j.id, r.name FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id \

         WHERE j.status = 'running' AND r.last_seen_ms < $1 ORDER BY j.id",

    )

    .bind(before)

    .fetch_all(pool)

    .await?;

    let mut out = Vec::new();

    for (id, runner) in rows {

        if let Some(j) = job(pool, id).await? {

            out.push((j, runner));

        }

    }

    Ok(out)

             approved_ms = NULL, approval_seal = '', approved_via = '', queued_ms = $2, ready_ms = NULL, handed_back_ms = NULL,

             handbacks = 0 WHERE id = $1",

/// How long a runner holds a job without the hub hearing from it (claims,

/// heartbeats every 30 s, logs) before the job is handed back

/// (oxygenflow#76).

pub const LEASE_MS: i64 = 3 * 60_000;

/// A job handed back more times than this ends as an error rather than

/// bounce from runner to runner.

pub const MAX_HANDBACKS: i64 = 5;

/// Hand job `id` back to the queue, ahead of new work (oxygenflow#76): its

/// runner drained before leaving (`runner` is it), or its lease lapsed

/// (`None`). The log says why, and the next attempt's follows on. False

/// when the job was not running there.

pub async fn hand_back_job(app: &Shared, id: i64, runner: Option<i64>, why: &str) -> sqlx::Result<bool> {

    let Some(before) = fdb::job(&app.db, id).await? else { return Ok(false) };

    let Some(n) = fdb::hand_back(&app.db, id, runner).await? else { return Ok(false) };

    let t = before.started_ms.map(|s| (db::now_ms() - s).max(0) as u64).unwrap_or(0);

    let record = flow_core::api::Record { t, stream: "flow".into(), text: format!("── {why}; queued again ahead of new work") };

    fdb::append_log(&app.db, id, &[record], MAX_RECORDS as i64).await.ok();

    tracing::info!("flow: job {id} handed back ({n}): {why}");

    if n > MAX_HANDBACKS {

        let error = format!("handed back {n} times; the last: {why}");

        let changed =

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

                .await?;

        report_changed(app, before.run_id, changed).await?;

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

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

            let summary = format!("{why}; queued again ahead of new work");

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

        }

    }

    Ok(true)

}

/// Hand back the jobs of runners not heard from for [`LEASE_MS`]: a phone

/// that lost every network, a host that died (oxygenflow#76).

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

    let lapsed = fdb::running_on_silent_runners(&app.db, db::now_ms() - LEASE_MS).await?;

    let mut n = 0;

    for (job, runner) in lapsed {

        let why = format!("the runner {runner} has not been heard from for {} minutes", LEASE_MS / 60_000);

        if hand_back_job(app, job.id, None, &why).await? {

            n += 1;

        }

    }

    Ok(n)

}

/// `POST /api/v1/flow/jobs/{id}/release` (oxygenflow#76): a draining

/// runner hands back a job it holds, as it leaves (off the allowed network,

/// a restart, a spot reclaim). Not a decline: the runner is not held from

/// the job and it does not count towards unplaceable.

pub async fn release(

    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 why = match reason.trim() {

        "" => format!("handed back by {}", me.name),

        r => format!("handed back by {}: {r}", me.name),

    };

    if !hand_back_job(&app, id, Some(me.id), &why).await? {

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

    }

    Ok(StatusCode::NO_CONTENT)

}

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

            tracing::warn!("flow: looking for lapsed leases: {e}");

        }

/// oxygenflow#76, #312: a draining runner hands back a job it holds. It is

/// queued again ahead of new work (even a job of lower priority), its log

/// says why and goes on with the next attempt, and it is no decline. A

/// runner the hub stops hearing from has its jobs handed back the same way.

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

async fn a_drained_job_goes_back_ahead_of_new_work() {

    let h = hub().await;

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

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

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

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

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

    let bulk = "job \"bulk\" {\n    sandbox \"alpine\"\n    priority \"low\"\n    exec \"true\"\n}\n";

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

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

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

    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 release = |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}/release"), Some(token), Some(body)).await.0

        }

    };

    let name_of = |id: i64| {

        let app = h.app.clone();

        async move { db::flow::job(&app.db, id).await.unwrap().unwrap().name }

    };

    let build = claim_as("runner-secret", "earth-1").await.unwrap();

    let lint = claim_as("runner-two", "plex-1").await.unwrap();

    let low = claim_as("runner-three", "phone").await.unwrap();

    assert_eq!((name_of(build).await, name_of(lint).await, name_of(low).await), ("build".into(), "lint".into(), "bulk".into()));

    // The low-priority job logs a line, then its runner leaves.

    let line = serde_json::json!({ "records": [{ "t": 1200, "stream": "stdout", "text": "step 1 of 30" }] }).to_string();

    call(&h.app, "POST", &format!("/api/v1/flow/jobs/{low}/log"), Some("runner-three"), Some(line)).await;

    assert_eq!(finish(&h.app, build, serde_json::json!({ "exit_code": 0, "duration_ms": 700 })).await, StatusCode::NO_CONTENT);

    assert_eq!(release("runner-two", low, "off Wi-Fi").await, StatusCode::CONFLICT, "only the runner that holds it");

    assert_eq!(release("runner-three", low, "off Wi-Fi").await, StatusCode::NO_CONTENT);

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

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

    let log: Vec<String> = db::flow::log(&h.app.db, low, 0, 50).await.unwrap().into_iter().map(|l| l.text).collect();

    assert_eq!(log, ["step 1 of 30", "── handed back by phone: off Wi-Fi; queued again ahead of new work"]);

    assert!(db::flow::declines_of_job(&h.app.db, low).await.unwrap().is_empty(), "not a decline");

    // `test` became ready when `build` passed, and is of higher priority:

    // the handed-back job still goes first, to any runner, its own too.

    assert_eq!(claim_as("runner-three", "phone").await, Some(low));

    assert_eq!(claim_as("runner-secret", "earth-1").await.map(|id| id != low), Some(true), "then new work");

    // The phone goes quiet: once its lease lapses the job is handed back.

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

    sqlx::query("UPDATE flow_runners SET last_seen_ms = $1 WHERE id = $2")

        .bind(db::now_ms() - oxygenhub::flow::LEASE_MS - 1000)

        .bind(phone.id)

        .execute(&h.app.db)

        .await

        .unwrap();

    assert_eq!(oxygenhub::flow::requeue_lapsed(&h.app).await.unwrap(), 1);

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

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

    let log: Vec<String> = db::flow::log(&h.app.db, low, 0, 50).await.unwrap().into_iter().map(|l| l.text).collect();

    assert_eq!(log.last().unwrap(), "── the runner phone has not been heard from for 3 minutes; queued again ahead of new work");

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

}