Expediting a run moves the queue at once (closes #291) 794ff94f194a

- **Expediting a run moves the queue at once (#291):** a run's jobs kept the place they had ("9 of 9 jobs come before") until the next placement pass and some other change on the page. Expediting now places every queued job again straight away, every run whose jobs' notes changed updates live, and a job an expedited run moved down says so ("2 ready jobs come first (after expedited run andy/demo #12)").

    behind_urgent: &[String],

        let after = match behind_urgent {

            [] => String::new(),

            [one] => format!(" (after expedited run {one})"),

            many => format!(" (after expedited runs {})", many.join(", ")),

        };

        parts.push(format!("{ahead} ready {} first{after}", if ahead == 1 { "job comes" } else { "jobs come" }));

    let run_of: std::collections::HashMap<i64, i64> = ready.iter().map(|(j, _, _)| (j.id, j.run_id)).collect();

    // Expedited runs (rank 0) by name, for the jobs they moved down (#291).

    let mut urgent: std::collections::BTreeMap<i64, String> = std::collections::BTreeMap::new();

    for (j, _, _) in ready.iter().filter(|(j, _, _)| ranks.get(&j.id) == Some(&0)) {

        if urgent.contains_key(&j.run_id) {

            continue;

        }

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

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

                urgent.insert(j.run_id, format!("{}/{} #{}", row.owner, row.name, run.number));

            }

        }

    }

    // Runs whose waiting notes changed: their pages hear of it now.

    let mut changed = std::collections::BTreeSet::new();

            let mut behind: Vec<String> = Vec::new();

            if mine.0 > 0 {

                for o in order.iter().filter(|o| o.0 == 0 && **o < mine) {

                    let Some(r) = run_of.get(&o.1).filter(|r| **r != job.run_id) else { continue };

                    if let Some(label) = urgent.get(r).filter(|l| !behind.contains(l)) {

                        behind.push(label.clone());

                    }

                }

            }

            why = why_still_waiting(&job, &owner, repo_id, &runners, (&archs, &held, now), &declines, &running, ahead, &behind);

        changed.insert(job.run_id);

    }

    for run_id in changed {

        app.bus.live(&crate::events::topic::run(run_id), "jobs");

        // Every queued job's place changes now, not at the next pass

        // (#291): this run's page and the pages of the runs it moved down

        // hear of it.

        if let Err(e) = crate::flow::explain_waits(&app).await {

            tracing::warn!("flow: placing after an expedite: {e}");

        }

        app.bus.live(&crate::events::topic::run(run.id), "jobs");

/// #291: expediting a run places the queue again at once, and a job it

/// moved down says which expedited run went before it.

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

async fn an_expedited_run_moves_the_queue_at_once() {

    let h = hub().await;

    for name in ["first", "urgent"] {

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

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

        let c = client(&format!("{}/andy/{name}", h.base), 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));

        // In order: the second run is the newer.

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

    }

    // One runner holding the image, busy with something else.

    let r = db::runners::configured(&h.app.db, &blake3::hash(b"busy").to_hex(), "earth-9").await.unwrap();

    db::runners::seen(&h.app.db, r.id, None, None).await.unwrap();

    db::runners::set_offer(&h.app.db, r.id, &["microvm".to_string()], Some(&[ROOTFS.to_string()]), false, "p").await.unwrap();

    let first = runs_of(&h.app, "andy", "first").await.remove(0);

    sqlx::query(

        "UPDATE flow_jobs SET status = 'running', runner_id = $1, runner = 'earth-9', started_ms = $2 WHERE run_id = $3 AND name = 'lint'",

    )

    .bind(r.id)

    .bind(db::now_ms())

    .bind(first.id)

    .execute(&h.app.db)

    .await

    .unwrap();

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

    let db = h.app.db.clone();

    let build = |run: i64| {

        let db = db.clone();

        async move { db::flow::jobs(&db, run).await.unwrap().into_iter().find(|j| j.name == "build").unwrap() }

    };

    let urgent = runs_of(&h.app, "andy", "urgent").await.remove(0);

    assert!(build(urgent.id).await.waiting.contains("1 ready job comes first"), "{}", build(urgent.id).await.waiting);

    // Expedited from its page: the places change before any pass runs.

    let res = oxygenhub::router(h.app.clone())

        .oneshot(

            Request::post("/register")

                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

                .body(Body::from("username=andy&email=a%40example.com&password=correct-horse-battery&setup=test-setup"))

                .unwrap(),

        )

        .await

        .unwrap();

    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();

    let page = oxygenhub::router(h.app.clone())

        .oneshot(Request::get("/andy/urgent/flow/1").header(header::COOKIE, cookie.clone()).body(Body::empty()).unwrap())

        .await

        .unwrap();

    let page = String::from_utf8_lossy(&page.into_body().collect().await.unwrap().to_bytes()).into_owned();

    let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();

    let csrf = page[i..].split('"').next().unwrap().to_string();

    let res = oxygenhub::router(h.app.clone())

        .oneshot(

            Request::post("/andy/urgent/flow/1/expedite")

                .header(header::COOKIE, cookie.clone())

                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

                .body(Body::from(format!("_csrf={csrf}&reason=incident")))

                .unwrap(),

        )

        .await

        .unwrap();

    assert!(res.status().is_redirection(), "{}", res.status());

    assert!(!build(urgent.id).await.waiting.contains("come"), "first now: {}", build(urgent.id).await.waiting);

    let w = build(first.id).await.waiting;

    assert!(w.contains("2 ready jobs come first (after expedited run andy/urgent #1)"), "{w}");

}