Flow priorities: site, organisation and repository levels, and expediting a run 107eaba52c9d

-- Flow priorities (oxygenhub#244): a repository's and an organisation's

-- level ('' to inherit: low, normal, high), and a run's own, set only by

-- expediting it (urgent), with who did it and why.

ALTER TABLE repositories ADD COLUMN flow_priority TEXT NOT NULL DEFAULT '';

ALTER TABLE organisations ADD COLUMN flow_priority TEXT NOT NULL DEFAULT '';

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

ALTER TABLE flow_runs ADD COLUMN priority_note TEXT NOT NULL DEFAULT '';

-- Flow priorities (oxygenhub#244): a repository's and an organisation's

-- level ('' to inherit: low, normal, high), and a run's own, set only by

-- expediting it (urgent), with who did it and why.

ALTER TABLE repositories ADD COLUMN flow_priority TEXT NOT NULL DEFAULT '';

ALTER TABLE organisations ADD COLUMN flow_priority TEXT NOT NULL DEFAULT '';

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

ALTER TABLE flow_runs ADD COLUMN priority_note TEXT NOT NULL DEFAULT '';

    /// `urgent` when expedited (oxygenhub#244), else empty: the run follows

    /// its repository's, organisation's or the site's level.

    #[sqlx(default)]

    pub priority: String,

    /// Who expedited it and why.

    #[sqlx(default)]

    pub priority_note: String,

}

/// The priority levels, most urgent first (oxygenhub#244).

pub const PRIORITIES: [&str; 4] = ["urgent", "high", "normal", "low"];

/// SQL for a queued job's level 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. Needs `r` (the run) and `rep`

/// (its repository) in the query.

pub const PRIORITY_RANK: &str = "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";

/// Raise run `run_id` to urgent, saying who and why; a run already ended

/// is left alone. Whether it changed.

pub async fn expedite(pool: &Pool, run_id: i64, note: &str) -> sqlx::Result<bool> {

    let r = sqlx::query("UPDATE flow_runs SET priority = 'urgent', priority_note = $1 WHERE id = $2 AND status IN ('queued', 'running')")

        .bind(note)

        .bind(run_id)

        .execute(pool)

        .await?;

    Ok(r.rows_affected() > 0)

    /// Its level (0 urgent … 3 low, oxygenhub#244) and, when expedited, who

    /// did it and why.

    pub priority_rank: i64,

    pub priority_note: String,

    pub fn priority(&self) -> &'static str {

        PRIORITIES.get(self.priority_rank as usize).copied().unwrap_or("normal")

    }

}

impl QueueEntry {

    sqlx::query_as(&format!(

        "SELECT j.id AS job_id, j.run_id, r.repo_id, rep.owner, rep.name, rep.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox, j.started_ms, \

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

         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \

           CASE j.status WHEN 'running' THEN 0 ELSE {PRIORITY_RANK} END, \

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

    ))

    let queued: Vec<Job> = sqlx::query_as(&format!(

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

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

    ))

    /// Its jobs' priority: `low`, `normal`, `high`, or empty to follow its

    /// organisation's and the site's (oxygenhub#244, migration 0073).

    #[sqlx(default)]

    pub flow_priority: String,

    Ok(())

}

pub async fn set_flow_priority(pool: &Pool, id: i64, priority: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE repositories SET flow_priority = $1 WHERE id = $2").bind(priority).bind(id).execute(pool).await?;

            priority_rank: 2,

            priority_note: String::new(),

#[derive(Deserialize)]

pub struct ExpediteForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    #[serde(default)]

    reason: String,

}

/// `POST /{owner}/{repo}/flow/{n}/expedite` (oxygenhub#244): a maintainer

/// raises a running or queued run to urgent, saying why (an incident's

/// fix, say). Audited and shown on the run; it lapses when the run ends.

pub async fn expedite(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, number)): Path<(String, String, i64)>,

    Form(f): Form<ExpediteForm>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);

    let me = require(&viewer, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Maintain).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let run = fdb::run(&app.db, row.id, number)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    let reason: String = f.reason.chars().filter(|c| !c.is_control()).take(200).collect();

    let reason = reason.trim();

    if reason.is_empty() {

        return Ok(Redirect::to(&format!("/{owner}/{name}/flow?error={}", urlencoding("Say why the run is urgent."))).into_response());

    }

    let note = format!("expedited by {}: {reason}", me.account.username);

    if fdb::expedite(&app.db, run.id, &note).await.map_err(|e| Error::from_db(shell.clone(), e))? {

        db::accounts::audit(&app.db, Some(&me.account), "flow.expedite", &format!("{owner}/{name}"), &format!("#{number}: {reason}"))

            .await

            .ok();

    }

    Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response())

}

    /// May raise the run to urgent (oxygenhub#244): maintainers, while it runs.

    can_expedite: bool,

    let can_expedite = run.live() && run.priority.is_empty() && viewer.account().is_some() && role >= Role::Maintain;

        can_expedite,

        .route("/admin/runners/priority", post(runners::save_priority))

        .route("/{owner}/{name}/flow/{id}/expedite", post(flow::expedite))

        .route("/{owner}/{name}/settings/flow-priority", post(repo_settings::flow_priority))

pub struct PriorityForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// `low`, `normal`, `high`, or `default`.

    priority: String,

}

/// `POST /{owner}/{repo}/settings/flow-priority` (oxygenhub#244): its jobs'

/// level. Maintainers choose low, normal or the default; only site

/// administrators raise a repository to high.

pub async fn flow_priority(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Form(f): Form<PriorityForm>,

) -> Result<Page<RepoSettingsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/settings"), &viewer);

    let (_, row, role) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Maintain).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let level = match f.priority.as_str() {

        "low" | "normal" | "high" => f.priority.as_str(),

        "default" => "",

        _ => return settings_page(&app, shell, row, role, None, Some("Pick low, normal, high or the default.".into())).await,

    };

    if level == "high" && !viewer.is_admin() {

        return settings_page(&app, shell, row, role, None, Some("Only site administrators raise a repository to high.".into())).await;

    }

    db::repos::set_flow_priority(&app.db, row.id, level).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(

        &app.db,

        viewer.account(),

        "repo.flow_priority",

        &format!("{owner}/{name}"),

        if level.is_empty() { "default" } else { level },

    )

    .await

    .ok();

    let row = db::repos::get_by_id(&app.db, row.id).await.map_err(|e| Error::from_db(shell.clone(), e))?.unwrap_or(row);

    let notice = if level.is_empty() {

        "Its jobs follow the organisation's and the site's priority.".to_string()

    } else {

        format!("Its jobs are {level} priority.")

    };

    settings_page(&app, shell, row, role, Some(notice), None).await

}

#[derive(Deserialize)]

    /// The site's priority for jobs (oxygenhub#244): low, normal or high.

    priority: String,

        priority: accounts::setting(&app.db, "flow_priority")

            .await

            .ok()

            .flatten()

            .filter(|p| !p.is_empty())

            .unwrap_or_else(|| "normal".into()),

#[derive(Deserialize)]

pub struct PriorityForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    priority: String,

}

/// `POST /admin/runners/priority` (oxygenhub#244): the level every job has

/// unless its organisation, repository or run says otherwise.

pub async fn save_priority(State(app): State<Shared>, viewer: Viewer, Form(f): Form<PriorityForm>) -> Result<Page<SettingsPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    if !["low", "normal", "high"].contains(&f.priority.as_str()) {

        return Ok(settings_page(&app, shell, None, Some("Pick low, normal or high (urgent is for expediting one run).".into())).await);

    }

    accounts::set_setting(&app.db, "flow_priority", &f.priority).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, Some(&me), "flow.priority", "site", &f.priority).await.ok();

    Ok(settings_page(

        &app,

        shell,

        Some(format!("Saved: jobs are {} unless their organisation, repository or run says otherwise.", f.priority)),

        None,

    )

    .await)

}

/// Priority comes before taking turns (oxygenhub#244): a low repository's

/// jobs wait for a normal one's, and an expedited run goes first of all.

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

async fn priority_then_turns_and_an_expedited_run_first() {

    let h = hub().await;

    for name in ["early", "later"] {

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

    }

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

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

    // early pushed first, but it is low: later's jobs go first.

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

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

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

    // Expedited, early's run goes before anything else.

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

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

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

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

    assert!(

        q.iter().any(|e| e.name == "early" && e.priority() == "urgent" && e.priority_note.contains("an incident")),

        "{:?}",

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

    );

    // The site's level is the default for everyone else.

    db::accounts::set_setting(&h.app.db, "flow_priority", "high").await.unwrap();

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

    assert!(q.iter().filter(|e| e.name == "later").all(|e| e.priority() == "high"));

}

- **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 (`organisations.flow_priority`, no form yet), 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. Nothing running is stopped for a more urgent job (migration 0073).

<form class="card wide" id="priority" method="post" action="/admin/runners/priority">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

  <h2>Priority</h2>

  <p>Runners take higher levels first; within a level, repositories take turns. An organisation or repository can set its own level, and a maintainer can expedite one run to urgent, with a reason.</p>

  <label for="priority-site">The site's level</label>

  <select id="priority-site" name="priority">{% for p in ["low", "normal", "high"] %}<option{% if *p == priority %} selected{% endif %}>{{ p }}</option>{% endfor %}</select>

  <button type="submit">Save</button>

</form>

  <tr><th>Repository</th><th>Run</th><th>Job</th><th>Status</th><th>Priority</th><th>Started</th><th>Runner</th></tr>

  {% for q in queue %}<tr><td><a href="/{{ q.owner }}/{{ q.name }}">{{ q.owner }}/{{ q.name }}</a></td><td><a href="/{{ q.owner }}/{{ q.name }}/flow/{{ q.run_number }}">#{{ q.run_number }}</a></td><td>{{ q.job }}</td><td>{{ q.status }}</td><td class="small"{% if !q.priority_note.is_empty() %} title="{{ q.priority_note }}"{% endif %}>{% if q.priority() == "normal" %}<span class="mute">normal</span>{% else %}<span class="tag">{{ q.priority() }}</span>{% endif %}</td><td class="small">{% if let Some(t) = q.started_ms %}<time title="{{ t|when_i }}">{{ t|ago_i }}</time>{% endif %}</td><td>{% if q.runner.is_empty() %}<span class="mute">waiting{% let needs = q.needs() %}{% if !needs.is_empty() %} for a runner with {% for l in needs %}<span class="lbl">{{ l }}</span> {% endfor %}{% endif %}{% if let Some(a) = q.arch() %} on <span class="lbl">{{ a }}</span>{% endif %}</span>{% else %}<a href="/admin/runners?runner={{ q.runner|urlencode }}#queue">{{ q.runner }}</a>{% endif %}</td></tr>{% endfor %}

  {% if !run.priority.is_empty() %}<p class="small"><span class="tag tag-urgent" title="Runners take it before any other">{{ run.priority }}</span> {{ run.priority_note }}</p>{% endif %}

  {% if can_expedite %}<form class="inline" method="post" action="{{ repo.url }}/flow/{{ run.number }}/expedite" title="Runners take an urgent run's jobs before any other's"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="text" name="reason" size="22" maxlength="200" placeholder="why it is urgent" required> <button type="submit" class="small">EXPEDITE</button></form>{% endif %}

<form class="card wide" id="flow-priority" method="post" action="{{ repo.url }}/settings/flow-priority">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

  <h2>OxygenFlow priority</h2>

  <p>Runners take higher levels first; within a level, repositories take turns. To rush one run (an incident's fix), expedite it on its page.</p>

  <select name="priority" aria-label="Priority">

    <option value="default"{% if row.flow_priority.is_empty() %} selected{% endif %}>the default (the organisation's or the site's)</option>

    {% if site_admin %}<option value="high"{% if row.flow_priority == "high" %} selected{% endif %}>high</option>{% endif %}

    <option value="normal"{% if row.flow_priority == "normal" %} selected{% endif %}>normal</option>

    <option value="low"{% if row.flow_priority == "low" %} selected{% endif %}>low</option>

  </select>

  <button type="submit">SAVE</button>

</form>