Flow: repositories take turns on the runners (fair share) 55673c650986

         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, \

           CASE j.status WHEN 'running' THEN 0 ELSE ROW_NUMBER() OVER (PARTITION BY r.repo_id, j.status ORDER BY j.run_id, j.depth, j.id) \

           + (SELECT COUNT(*) FROM flow_jobs x JOIN flow_runs xr ON xr.id = x.run_id WHERE x.status = 'running' AND xr.repo_id = r.repo_id) END, \

           j.run_id, j.depth, j.id",

/// scope may take, now running under `runner`. Repositories take turns

/// (fair share, oxygenhub#244): a job's turn is its place in its own

/// repository's queue plus that repository's running jobs, lowest first,

/// oldest first among equals, so one repository's burst can't hold up the

/// rest. A scoped runner never sees

         ORDER BY ROW_NUMBER() OVER (PARTITION BY r.repo_id ORDER BY j.run_id, j.depth, j.id) \

           + (SELECT COUNT(*) FROM flow_jobs x JOIN flow_runs xr ON xr.id = x.run_id WHERE x.status = 'running' AND xr.repo_id = r.repo_id), \

           j.run_id, j.depth, j.id LIMIT 200",

    // andy/first has a job running, so plain's goes next (repositories take

    // turns, #244), and its run says where it stands.

    assert!(page.contains("Waiting for a runner: 1st in line") && page.contains("Running now: andy/first #1"), "{page}");

    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["queue_position"], 1);

    assert!(list.contains("queued, 1st in line"), "{list}");

/// Repositories take turns (oxygenhub#244): a repository with a job running

/// waits behind one with none, so a burst can't hold up the rest.

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

async fn repositories_take_turns_on_the_runners() {

    let h = hub().await;

    for name in ["busy", "quiet"] {

        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.snapshot("test").unwrap();

        c.commit("Add a pipeline", &[]).unwrap();

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

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

    }

    // busy pushed first: its first job goes first; then quiet's turn, though

    // busy's other independent job is older; then busy's again.

    let order: Vec<String> = {

        let mut out = Vec::new();

        for _ in 0..3 {

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

            out.push(c["repo"].as_str().unwrap().to_string());

        }

        out

    };

    assert_eq!(order, ["andy/busy", "andy/quiet", "andy/busy"]);

}

  <p class="small mute">Queued jobs are in the order runners claim them: repositories take turns, a repository with jobs running waiting behind those without; a job waits for its dependencies and for a runner whose scope, tier and room fit it.</p>