A pipeline may lower its jobs' priority, never raise it 53a0fb2e4ac6

-- A job's own priority from its pipeline (oxygenhub#244): `low` or

-- `normal` to stand below its repository's level, or '' for none. A

-- pipeline may only lower its jobs, never raise them.

ALTER TABLE flow_jobs ADD COLUMN priority TEXT NOT NULL DEFAULT '';

-- A job's own priority from its pipeline (oxygenhub#244): `low` or

-- `normal` to stand below its repository's level, or '' for none. A

-- pipeline may only lower its jobs, never raise them.

ALTER TABLE flow_jobs ADD COLUMN priority TEXT NOT NULL DEFAULT '';

/// SQL for the level a run's jobs get from where they are, as a rank

/// (0 urgent … 3 low): the run's own (expedited), else its repository's,

/// its organisation's, the site's (`settings.flow_priority`), else normal.

macro_rules! level_rank {

    () => {

        "CASE COALESCE(NULLIF(r.priority, ''), NULLIF(rep.flow_priority, ''), \

           NULLIF((SELECT o.flow_priority FROM organisations o WHERE o.name = rep.owner), ''), \

           NULLIF((SELECT s.value FROM settings s WHERE s.key = 'flow_priority'), ''), 'normal') \

         WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 WHEN 'low' THEN 3 ELSE 2 END"

    };

}

/// SQL for a queued job's level as a rank (0 urgent … 3 low): the level

/// its run gets from where it is, lowered by the job's own priority from

/// its pipeline (`low`, `normal`), never raised by it; an expedited run is

/// urgent whatever its jobs say (oxygenhub#244). Needs `j` (the job), `r`

/// (its run) and `rep` (its repository) in the query.

pub const PRIORITY_RANK: &str = concat!(

    "CASE WHEN r.priority <> '' THEN ",

    level_rank!(),

    " WHEN j.priority = 'low' THEN 3 WHEN j.priority = 'normal' AND (",

    level_rank!(),

    ") < 2 THEN 2 ELSE ",

    level_rank!(),

    " END"

);

/// SQL for why a job stands where it does, beside [`PRIORITY_RANK`]: an

/// expedited run's note, else the pipeline's word for a job it lowered.

pub const PRIORITY_NOTE: &str =

    "CASE WHEN r.priority <> '' THEN r.priority_note WHEN j.priority <> '' THEN 'pipeline: ' || j.priority ELSE r.priority_note END";

            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, need_disk_mib, approval, environment, priority) \

             VALUES ($1, $2, $3, $4, $5, $6, $7, 'queued', $8, $9, $10, $11)",

        .bind(priority_of(&j.spec))

}

/// A compiled job's own priority (`low`, `normal`), or empty; anything

/// else is not one a pipeline may give (oxygenhub#244).

fn priority_of(spec: &str) -> String {

    serde_json::from_str::<serde_json::Value>(spec)

        .ok()

        .and_then(|v| v["priority"].as_str().map(String::from))

        .filter(|p| p == "low" || p == "normal")

        .unwrap_or_default()

           CAST({PRIORITY_RANK} AS BIGINT) AS priority_rank, {PRIORITY_NOTE} AS priority_note \

/// A pipeline lowers its jobs, never raises them (oxygenhub#244): a

/// repository whose pipeline says `priority "low"` waits for another's

/// everyday jobs, even set high itself; the queue says why; an expedited

/// run is still urgent.

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

async fn a_pipeline_lowers_its_jobs_and_expediting_still_wins() {

    let h = hub().await;

    for (name, extra) in [("nightly", "priority \"low\"\n"), ("everyday", "")] {

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

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

    }

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

    db::repos::set_flow_priority(&h.app.db, nightly.id, "high").await.unwrap();

    let q = db::flow::queue(&h.app.db).await.unwrap();

    assert!(

        q.iter().filter(|e| e.name == "nightly").all(|e| e.priority() == "low" && e.priority_note == "pipeline: low"),

        "{:?}",

        q.iter().map(|e| (&e.name, e.priority(), &e.priority_note)).collect::<Vec<_>>()

    );

    let repo_of = |c: serde_json::Value| c["repo"].as_str().unwrap().to_string();

    // nightly pushed first and its repository is high, but its pipeline

    // says low: everyday's jobs go first.

    assert_eq!(repo_of(claim(&h.app).await.unwrap()), "andy/everyday");

    assert_eq!(repo_of(claim(&h.app).await.unwrap()), "andy/everyday");

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

    assert!(db::flow::expedite(&h.app.db, run.id, "expedited by andy: needed now").await.unwrap());

    assert_eq!(repo_of(claim(&h.app).await.unwrap()), "andy/nightly", "an expedited run is urgent whatever its jobs say");

}

- **Priority and turns** (oxygenhub#244). Runners take queued jobs by level first (`urgent`, `high`, `normal`, `low`), then by turn: a job's place in its own repository's queue plus that repository's running jobs, oldest first among equals, so repositories take turns and one repository's burst can't hold up the rest. A run's level is its own if it was expedited, else its repository's (Settings › OxygenFlow priority; maintainers choose low, normal or the default, site administrators also high), else its organisation's (its settings page, by its owners), else the site's (Admin › Runners › Settings), else normal. A maintainer expedites a queued or running run to urgent on its page with a reason; it is audited (`flow.expedite`), shown on the run and in the queue, and lapses when the run ends. A pipeline may lower its jobs, never raise them: `priority "low"` (or `"normal"`) on a job or for the whole file (OxygenFlow spec) makes a job's level the lower of its own and its run's, except that an expedited run stays urgent; the queue says `pipeline: low` (migration 0074). Nothing running is stopped for a more urgent job (migration 0073).