A queued run says where it stands bf3d07541368

    /// For a run waiting for a runner, its first queued job's 1-based

    /// place among all queued jobs.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub queue_position: Option<i64>,

fn flow_run_of(

    app: &Shared,

    row: &db::repos::RepoRow,

    r: &db::flow::Run,

    jobs: Vec<db::flow::Job>,

    queue: &[db::flow::QueueEntry],

) -> FlowRun {

        queue_position: crate::flow::queue_info(queue, row.id, r.id).position.map(|p| p as i64),

    let queue = if runs.iter().any(|r| r.live()) { db::flow::queue(&app.db).await? } else { vec![] };

    Ok(Json(runs.iter().map(|r| flow_run_of(&app, &row, r, vec![], &queue)).collect()))

    let queue = if run.live() { db::flow::queue(&app.db).await? } else { vec![] };

    Ok(Json(flow_run_of(&app, &row, &run, jobs, &queue)))

/// A job waiting for or held by a runner, with where it comes from.

#[derive(Clone, Debug, FromRow)]

pub struct QueueEntry {

    pub job_id: i64,

    pub run_id: i64,

    pub repo_id: i64,

    pub owner: String,

    pub name: String,

    pub visibility: String,

    pub run_number: i64,

    pub job: String,

    pub status: String,

}

/// Every job of a live run that is running or queued: running ones first,

/// then queued ones in the order runners claim them (hub spec §13.2).

pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {

    sqlx::query_as(

        "SELECT j.id AS job_id, j.run_id, r.repo_id, p.owner, p.name, p.visibility, r.number AS run_number, j.name AS job, j.status \

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

         WHERE j.status IN ('queued', 'running') AND r.status IN ('queued', 'running') \

         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, j.run_id, j.depth, j.id",

    )

    .fetch_all(pool)

    .await

}

/// Where a live run stands among the runners' work: what its viewers may

/// know of it. Jobs of other repositories are named only when those are

/// public; the rest are counted.

#[derive(Clone, Debug, Default, PartialEq, Eq)]

pub struct QueueInfo {

    /// 1-based place of the run's first queued job among all queued jobs.

    pub position: Option<usize>,

    /// Jobs running now, and queued jobs ahead of the run's first one.

    pub running: Vec<String>,

    pub ahead: Vec<String>,

    /// Of those, jobs in repositories the page may not name.

    pub hidden: usize,

}

impl QueueInfo {

    /// "3rd in line" and the like.

    pub fn place(&self) -> Option<String> {

        self.position.map(|n| {

            let suffix = match (n % 10, n % 100) {

                (1, 11) | (2, 12) | (3, 13) => "th",

                (1, _) => "st",

                (2, _) => "nd",

                (3, _) => "rd",

                _ => "th",

            };

            format!("{n}{suffix} in line")

        })

    }

}

/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.

pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {

    let mut info = QueueInfo::default();

    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);

    info.position = first.map(|i| i + 1);

    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);

    let mut queued_seen = 0;

    for e in queue {

        if e.run_id == run_id {

            continue;

        }

        let ahead = match e.status.as_str() {

            "running" => true,

            _ => {

                queued_seen += 1;

                first.is_some_and(|f| queued_seen <= f)

            }

        };

        if !ahead {

            continue;

        }

        if e.repo_id != repo_id && e.visibility != "public" {

            info.hidden += 1;

        } else if e.status == "running" {

            info.running.push(label(e));

        } else {

            info.ahead.push(label(e));

        }

    }

    info

}

#[cfg(test)]

mod queue_tests {

    use super::*;

    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {

        fdb::QueueEntry {

            job_id,

            run_id,

            repo_id,

            owner: "o".into(),

            name: format!("r{repo_id}"),

            visibility: vis.into(),

            run_number: run_id,

            job: job.into(),

            status: status.into(),

        }

    }

    #[test]

    fn a_run_learns_its_place_and_what_is_ahead_of_it() {

        // Running first, then queued in claim order (as `fdb::queue` returns them).

        let q = vec![

            e(1, 10, 1, "public", "build", "running"),

            e(2, 11, 2, "private", "test", "running"),

            e(3, 10, 1, "public", "test", "queued"),

            e(4, 12, 3, "private", "lint", "queued"),

            e(5, 13, 4, "public", "build", "queued"),

            e(6, 13, 4, "public", "test", "queued"),

            e(7, 14, 5, "public", "late", "queued"),

        ];

        let info = queue_info(&q, 4, 13);

        assert_eq!(info.position, Some(3));

        assert_eq!(info.place().as_deref(), Some("3rd in line"));

        assert_eq!(info.running, ["o/r1 #10 build"]);

        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");

        assert_eq!(info.hidden, 2, "private repositories are counted, not named");

        // A private repository's own jobs are named to its viewers.

        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);

        assert_eq!(queue_info(&q, 3, 12).position, Some(2));

        // A run with nothing queued has no place.

        assert_eq!(queue_info(&q, 1, 99).position, None);

        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();

        assert_eq!(

            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],

            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]

        );

    }

}

    /// "3rd in line" for a run still waiting for a runner.

    pub place: Option<String>,

    let queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] };

        let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None };

        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r, place });

    /// Where a live run stands among the runners' work.

    queue: Option<crate::flow::QueueInfo>,

    let queue = if run.live() {

        let q = fdb::queue(&app.db).await.unwrap_or_default();

        Some(crate::flow::queue_info(&q, row.id, run.id)).filter(|i| i.position.is_some())

    } else {

        None

    };

    Ok(Page(RunPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), run, graph, logs, duration, can_rerun, queue }))

    // With andy/first's jobs ahead, plain's run says where it stands.

    let a = claim(&h.app).await.unwrap();

    assert_eq!(a["repo"], "andy/first");

    let (_, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;

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

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/other/plain/flow/runs/1", None, None).await;

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

    let (_, list) = call(&h.app, "GET", "/other/plain/flow", None, None).await;

    assert!(list.contains("queued, 3rd in line"), "{list}");

- **Triggers.** A push that moves a bookmark of an allowed repository starts a **run** on the bookmark's new state, numbered within its repository from 1 (the run's page is `/{owner}/{name}/flow/{number}`): the hub reads `.flow/pipeline.kdl` from that state and validates it with `flow-core` (the same code `flow check` runs). An invalid pipeline records a run in `error` with the problems and their lines; a state without a pipeline records nothing. Tags and proposals start no runs yet, so only code from people with write access ever runs. A run waiting for a runner says where it stands: on the Flow tab its place in line, on its page also the jobs running now and those queued ahead of it (named when their repository is this one or public, otherwise counted), and in the API `queue_position`. Writers also start runs without moving a bookmark: **Run pipeline** on the Flow tab runs a bookmark's current state, and **Re-run** on a finished run's page runs its state again. The new run's checks replace the old ones for that state, and the run page says what started it (pushed by, started by, re-run of #N by).

{% if let Some(q) = queue %}

<section class="panel"><div class="strip">Waiting for a runner: {{ q.place().unwrap_or_default() }}.{% if !q.running.is_empty() %} Running now: {{ q.running.join(", ") }}.{% endif %}{% if !q.ahead.is_empty() %} Queued ahead: {{ q.ahead.join(", ") }}.{% endif %}{% if q.hidden > 0 %} Also ahead or running: {{ q.hidden }} job{% if q.hidden != 1 %}s{% endif %} in private repositories.{% endif %}</div></section>

{% endif %}

    <td><a href="{{ repo.url }}/flow/{{ row.r.number }}"><strong>#{{ row.r.number }}</strong></a> <span class="small mute">{{ row.r.status }}{% if let Some(p) = row.place %}, {{ p }}{% endif %}</span></td>