The hub ends jobs no runner reports, past the time limit (refs #71) 070a7b64b368

- **Lost jobs end by themselves (#71).** A job still running five minutes past the job time limit is one its runner no longer reports (it restarted, or lost the hub as the job ended). The hub now looks for those once a minute and ends them as errors with the reason, skipping the jobs that depended on them and settling the run, so a runner is not shown busy with jobs it is not running. Earth's reset stranded five such jobs, closed by hand.

/// Jobs running since before `ms` (the oldest first).

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

    sqlx::query_as("SELECT * FROM flow_jobs WHERE status = 'running' AND COALESCE(started_ms, 0) < ? ORDER BY started_ms")

        .bind(ms)

        .fetch_all(pool)

        .await

}

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

}

/// Report the checks of jobs whose status changed.

async fn report_changed(app: &Shared, run_id: i64, changed: Vec<fdb::Job>) -> sqlx::Result<()> {

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

        return Ok(());

    };

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

        return Ok(());

    };

    for j in changed {

        let summary = match j.status.as_str() {

            "passed" => format!("passed in {}", duration(&j)),

            "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),

            "skipped" => "skipped: a dependency did not pass".into(),

            _ => format!("could not run: {}", j.error),

        };

        report_check(app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;

    }

    Ok(())

}

/// Grace past the job time limit before the hub gives a job up: a runner

/// stops a job at the limit and reports it within moments, so one still

/// running this long after is one the runner no longer reports (it was

/// restarted, or lost the hub while the job ended, oxygenhub#71).

const LOST_GRACE_MS: i64 = 5 * 60 * 1000;

/// End every job still running past the job time limit plus the grace:

/// it is marked an error with the reason, the jobs that depended on it are

/// skipped and its run settled, as when a runner reports an error.

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

    let limit_ms = app.config.flow.job_timeout_minutes.max(1) as i64 * 60_000 + LOST_GRACE_MS;

    let lost = fdb::running_since_before(&app.db, db::now_ms() - limit_ms).await?;

    let n = lost.len();

    for job in lost {

        let error = format!(

            "lost: still running {} minutes after it started, past the time limit; the runner{} stopped reporting it",

            (db::now_ms() - job.started_ms.unwrap_or(0)) / 60_000,

            if job.runner.is_empty() { String::new() } else { format!(" {}", job.runner) },

        );

        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(n)

}

/// Look for lost jobs once a minute.

pub async fn reaper(app: Shared) {

    loop {

        tokio::time::sleep(std::time::Duration::from_secs(60)).await;

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

            tracing::warn!("flow: looking for lost jobs: {e}");

        }

    }

    tokio::spawn(flow::reaper(app.clone()));

/// oxygenhub#71: a job its runner stopped reporting (restarted, or lost

/// the hub as the job ended) is ended once it has run past the time limit

/// and a grace: an error with the reason, its dependents skipped and its

/// run settled, as if the runner had reported the error.

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

async fn a_job_no_runner_reports_is_ended_after_the_time_limit() {

    let h = hub().await;

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

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

    let url = format!("{}/andy/lost", 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", "lost").await.len(), 1);

    let job = claim(&h.app).await.expect("a job to claim");

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

    // Running, but well within the limit: left alone.

    assert_eq!(oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap(), 0);

    // Started long ago and never reported: ended.

    let old = db::now_ms() - 24 * 3600 * 1000;

    sqlx::query("UPDATE flow_jobs SET started_ms = ? WHERE id = ?").bind(old).bind(id).execute(&h.app.db).await.unwrap();

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

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

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

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

    assert!(jobs.iter().all(|j| j.status != "running"), "nothing left running");

    // A late report, should one come, does not overwrite the ending.

    assert_ne!(finish(&h.app, id, serde_json::json!({ "exit_code": 0 })).await, StatusCode::NO_CONTENT);

    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "error");

}