Claims say how long a job is expected to take (refs andymccall/oxygenflow#76) 3b65065298b7

- **Claims say how long a job is expected to take (oxygenflow#76):** `expected_ms`, the median of the job's last five passing runs in the repository (reused passes left out), so a runner leaving with notice finishes a job that fits it and hands back the rest. Pins OxygenFlow's main.

/// How long job `name` of repository `repo_id` is expected to take: the

/// median of its last five passing runs (claimed to finished), passes

/// reused from another run left out (oxygenflow#76). `None` before it has

/// passed: a runner leaving then hands it back rather than guess.

pub async fn expected_ms(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Option<i64>> {

    let mut times: Vec<i64> = sqlx::query_scalar(

        "SELECT j.finished_ms - j.started_ms FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \

         WHERE r.repo_id = $1 AND j.name = $2 AND j.status = 'passed' AND j.error = '' \

         AND j.started_ms IS NOT NULL AND j.finished_ms IS NOT NULL ORDER BY j.id DESC LIMIT 5",

    )

    .bind(repo_id)

    .bind(name)

    .fetch_all(pool)

    .await?;

    times.sort_unstable();

    Ok(times.get(times.len() / 2).copied().filter(|ms| *ms > 0))

}

        // For a runner that must leave: finish it, or hand it back

        // (oxygenflow#76).

        expected_ms: fdb::expected_ms(&app.db, row.id, &job.name).await?.map(|ms| ms as u64),

/// oxygenflow#76: a claim says how long the job is expected to take, once

/// it has passed: the median of its recent passing runs in the repository,

/// so a runner leaving with notice can tell whether to finish it.

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

async fn a_claim_carries_the_job_s_expected_time() {

    let h = hub().await;

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

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

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

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

    let one = "schema-version 1\nsandbox \"alpine\" {\n    rootfs \"cas://blob/ROOTFS\"\n}\njob \"build\" {\n    sandbox \"alpine\"\n    exec \"make\"\n}\n".replace("ROOTFS", ROOTFS);

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

    c.commit("One job", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

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

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

    assert!(first.get("expected_ms").is_none(), "never passed: no estimate, {first}");

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

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

    // It took 42 s from its claim.

    sqlx::query("UPDATE flow_jobs SET started_ms = finished_ms - 42000 WHERE id = $1").bind(id).execute(&h.app.db).await.unwrap();

    write(work.path(), "README", "again\n");

    c.commit("Again", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    let mut next = None;

    for _ in 0..100 {

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

            next = Some(j);

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    let next = next.expect("the second run's job");

    assert_eq!(next["expected_ms"], 42_000, "{next}");

}