Requirements learned from each job's past runs 44a3d2ebaddd

-- Requirements learned from past runs (oxygenflow#41): what each job took,

-- as its runner reported, and the disk a new job needs from the most its

-- recent runs of the same repository and name took, plus headroom.

ALTER TABLE flow_jobs ADD COLUMN peak_disk_mib INTEGER;

ALTER TABLE flow_jobs ADD COLUMN peak_memory_mib INTEGER;

ALTER TABLE flow_jobs ADD COLUMN need_disk_mib INTEGER;

    /// What it took, as its runner reported (oxygenflow#41), in MiB.

    pub peak_disk_mib: Option<i64>,

    pub peak_memory_mib: Option<i64>,

    /// The free disk it needs, learned from past runs of the same job.

    pub need_disk_mib: Option<i64>,

    /// The free disk it needs, from [`learned_disk_need`].

    pub need_disk_mib: Option<i64>,

/// Runs of a job whose peaks set its next need.

const NEED_FROM_RUNS: i64 = 5;

/// The free disk a new run of job `name` in repository `repo_id` needs

/// (oxygenflow#41): the most its last runs took, plus a quarter; `None`

/// for a job that has not run here yet.

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

    let peak: Option<i64> = sqlx::query_scalar(

        "SELECT MAX(peak_disk_mib) FROM (SELECT j.peak_disk_mib FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \

         WHERE r.repo_id = ? AND j.name = ? AND j.peak_disk_mib IS NOT NULL ORDER BY j.id DESC LIMIT ?)",

    )

    .bind(repo_id)

    .bind(name)

    .bind(NEED_FROM_RUNS)

    .fetch_one(pool)

    .await?;

    Ok(peak.map(|p| p + p / 4 + 64))

}

/// Keep what a job took, as its runner reported.

pub async fn set_peaks(pool: &Pool, job_id: i64, disk_mib: Option<u64>, memory_mib: Option<u64>) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET peak_disk_mib = ?, peak_memory_mib = ? WHERE id = ?")

        .bind(disk_mib.map(|v| v.min(i64::MAX as u64) as i64))

        .bind(memory_mib.map(|v| v.min(i64::MAX as u64) as i64))

        .bind(job_id)

        .execute(pool)

        .await?;

    Ok(())

}

            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, need_disk_mib) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', ?)",

        .bind(j.need_disk_mib)

    /// Its free disk now, from its claim's facts (oxygenflow#40); `None`

    /// when it does not say.

    pub free_disk_mib: Option<u64>,

}

impl Room {

    /// Whether a job needing `need` MiB of disk fits in its free disk.

    pub fn disk_fits(&self, need: Option<i64>) -> bool {

        match (need, self.free_disk_mib) {

            (Some(n), Some(free)) => n.max(0) as u64 <= free,

            _ => true,

        }

    }

        if !runtimes.iter().any(|r| r == &j.runtime)

            || !room.fits(&j.sandbox)

            || !room.disk_fits(j.need_disk_mib)

            || room.declined.contains(&j.id)

        {

            let mut jobs: Vec<fdb::NewJob> = p

                        need_disk_mib: None,

            // What each job needs, learned from its past runs (oxygenflow#41).

            for j in &mut jobs {

                j.need_disk_mib = fdb::learned_disk_need(&app.db, row.id, &j.name).await.map_err(|e| e.to_string())?;

            }

        free_disk_mib: req.facts.as_ref().and_then(|f| f.free_disk_mib),

        }

        // The disk it learned it needs (oxygenflow#41).

        if let (Some(n), Some(free)) = (job.need_disk_mib, r.free_disk_mib) {

            if free < n {

                return Some(format!(

                    "has {} free; the job needs ~{} (its recent runs took up to {})",

                    mib(free),

                    mib(n),

                    mib((n - 64) * 4 / 5)

                ));

            }

}

/// A size in MiB as people say it: 9.4 GiB, 740 MiB.

fn mib(m: i64) -> String {

    if m >= 1024 {

        format!("{:.1} GiB", m as f64 / 1024.0)

    } else {

        format!("{m} MiB")

    }

    fdb::set_peaks(&app.db, job.id, f.peak_disk_mib, f.peak_memory_mib).await?;

/// oxygenflow#41: a job's peaks set what its next run needs; a runner

/// without that much free disk is not offered it, and the reason says so.

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

async fn a_jobs_next_run_needs_the_disk_its_last_took() {

    let h = hub().await;

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

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

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

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

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

    write(

        work.path(),

        ".flow/pipeline.kdl",

        &format!("schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\n"),

    );

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

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

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

    let claim_with = |free: u64| format!(r#"{{"runner":"test-1","runtimes":["microvm"],"facts":{{"free_disk_mib":{free}}}}}"#);

    // The first run has nothing to learn from: any runner takes it.

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(claim_with(500))).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let id = serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap();

    let done = serde_json::json!({ "exit_code": 0, "duration_ms": 10, "peak_disk_mib": 8000, "peak_memory_mib": 900 });

    assert_eq!(finish(&h.app, id, done).await, StatusCode::NO_CONTENT);

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

    assert_eq!((job.peak_disk_mib, job.peak_memory_mib), (Some(8000), Some(900)));

    // The next run needs 8000 MiB and a quarter more.

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

    assert_eq!(st, StatusCode::CREATED);

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

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

    let next = db::flow::jobs(&h.app.db, run2.id).await.unwrap().remove(0);

    assert_eq!(next.need_disk_mib, Some(8000 + 2000 + 64));

    let (st, _) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(claim_with(9000))).await;

    assert_eq!(st, StatusCode::NO_CONTENT, "9000 MiB free is not enough");

    oxygenhub::flow::explain_waits(&h.app).await.unwrap();

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

    assert!(waiting.contains("8.8 GiB free; the job needs ~9.8 GiB"), "{waiting}");

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(claim_with(20_000))).await;

    assert_eq!(st, StatusCode::OK, "a roomier runner takes it: {body}");

}

- **Architecture** (OxygenFlow spec §11.2–§11.3, oxygenflow#36): a runner reports each image's architecture on its claims (`image_archs`), kept in `flow_images`; a job is offered only to a runner whose architecture matches its sandbox's `arch`, else its image's as known; unknown matches, as before. The queue says which architecture a waiting job asks for. **Images by hash** (oxygenflow#39, Jetstream): `flow rootfs push HASH` sends a rootfs image as the FastCDC chunks the hub lacks (`POST /flow/images/missing`, `PUT /flow/images/chunks/{hash}`, by a token that may write repositories) and then its manifest (`PUT /flow/images/{hash}`), which the hub accepts only when every chunk is held and they make an image with that hash; chunks are kept zstd-compressed under `<data>/images`, shared by every image, and `jet_images` lists what is held. Runners and accounts fetch with `GET /flow/images/{hash}` and `GET /flow/images/chunks/{hash}`. A runner whose claims say `fetches_images` is offered jobs whose image only the hub holds, and fetches it before its preflight. Each claim also carries the runner's `facts` (OxygenFlow spec §11.1, oxygenflow#40): free disk, load, speed score and class, kernel; the hub keeps the latest on the runner, shows them on Admin › Runners and in `GET /runners`. A job's finish carries the most disk and memory it took (`peak_disk_mib`, `peak_memory_mib`, oxygenflow#41); a new job's `need_disk_mib` is the most its last five runs of the same repository and name took, plus a quarter and 64 MiB, and it is offered only to a runner whose claim reports at least that much free disk; the waiting reason says so. A placement pass every 15 seconds works out why each ready queued job no runner can take waits, from what each online runner offered at its last claim (scope, tier, architecture, image held, labels, memory and CPUs, a recent decline), and shows it on the job (API `waiting`, `oh run view` and `oh run watch`, its panel open on the run page) and in its check's summary, so a proposal says why its check is stuck (#28).