End the jobs a runner no longer holds at its next claim eef677c48cd1

- **A restarted runner's lost jobs end at once (#71).** A runner lists the jobs it holds at each claim (OxygenFlow does), and the hub ends any other job it has as running on that runner, with the reason, rather than an hour later at the time limit.

/// The jobs running on the runner with this id.

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

    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND runner_id = ? ORDER BY id").bind(runner_id).fetch_all(pool).await

}

    if let Some(holding) = &req.holding {

        end_jobs_not_held(&app, &me, holding).await?;

    }

}

/// End the jobs running on `runner` that it does not hold (oxygenhub#71):

/// it restarted since claiming them, or never received the claim. A

/// runner claims one job at a time, so a job it holds is always listed.

async fn end_jobs_not_held(app: &Shared, runner: &db::runners::Runner, holding: &[i64]) -> sqlx::Result<()> {

    for job in fdb::running_on(&app.db, runner.id).await? {

        if holding.contains(&job.id) {

            continue;

        }

        let error = format!("lost: the runner {} no longer holds this job (it restarted, or never received it)", runner.name);

        let changed =

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

                .await?;

        tracing::warn!("flow: job {} {error}", job.id);

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

    }

    Ok(())

/// oxygenhub#71: a runner that lists the jobs it holds has the others it

/// was given ended at once (it restarted, or never received them), not at

/// the time limit; a held job is left alone.

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

async fn a_job_its_runner_no_longer_holds_is_ended_at_its_next_claim() {

    let h = hub().await;

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

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

    let url = format!("{}/andy/held", 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));

    assert_eq!(runs_of(&h.app, "andy", "held").await.len(), 1);

    let id = claim(&h.app).await.expect("a job to claim")["job_id"].as_i64().unwrap();

    let claim_holding = |holding: Vec<i64>| {

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

        let app = h.app.clone();

        async move { call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body)).await.0 }

    };

    claim_holding(vec![id]).await;

    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "running", "a held job stays");

    claim_holding(vec![]).await;

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

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

    assert!(j.error.starts_with("lost:") && j.error.contains("no longer holds"), "{}", j.error);

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

    assert!(db::flow::jobs(&h.app.db, run.id).await.unwrap().iter().all(|j| j.status != "running"), "the run settles");

}