chore: sync borg from Mono

.repository-projection.json

@@ -3,11 +3,11 @@
   "projection": "borg",
   "projectionSchemaVersion": 1,
   "sourceRepository": "dx-corp/mono",
-  "sourceSha": "7c1c4ccabe912b319ffbeda00acc89d238a25604",
+  "sourceSha": "6aa9d4a0331c653a7ce5b5335624c334de979977",
   "destinationRepository": "dx-corp/borg",
-  "priorProjectedBase": "5ea97abffad434077e597a41f8a183118e41124b",
+  "priorProjectedBase": "ddb6362e141a28af70e6641c3ae09d60f70ac2b1",
   "definitionDigest": "78ba8d1b5794c261ff0afba6d99abc10b9297ea0dd94be4397fb1cbf63cf0c6e",
   "toolDigest": "61f64178999b764105e6e5c5bc7c5971c164909e",
-  "contentDigest": "e0f71fcb7302dfac768903a8b223e61f5611416db2b8919bd101c3956975cbf6",
+  "contentDigest": "9e0e5f7acad9e39a50a5662518b27e7e7a292f38a693d24a3ab410efde4fa0ae",
   "publicationEligible": true
 }

docs/operations.md

@@ -710,6 +710,9 @@ a newer push leaves behind:
   `push` run on one of them is cancelled when a newer queued or running run of
   the same workflow on the same branch is on another commit. Running push runs
   are never cancelled, so the branch always finishes its newest started run.
+  GitHub reports a run as `queued` while any of its jobs waits for a runner, so
+  the janitor also reads the run's jobs and keeps the run when one is running
+  (`started_push`).
 
 When the janitor is enabled, `GET /status` has a `janitor` object:
 
@@ -735,6 +738,7 @@ When the janitor is enabled, `GET /status` has a `janitor` object:
 | `dedup_cancelled` | Queued duplicates cancelled. |
 | `would_cancel` | Runs `dry_run` would have cancelled, of any kind. |
 | `deferred` | Runs left for a later pass by `max_cancels_per_pass`. |
+| `started_push` | Superseded push runs kept because one of their jobs was running. |
 | `raced` | Runs whose status, head or pull request changed between the listing and the re-read just before the cancel; left alone. |
 | `uncancellable` | Runs whose cancel GitHub answered with 409. Runs stuck in `queued` for days (for example `workflow_run` runs GitHub never started) answer 409 on every pass and are counted each time. |
 | `errors` | Failed GitHub calls: a run listing for a repository, a pulls API read, a run re-read or a cancel. Each is logged as a `janitor:` warning in the queen journal. |

src/api.rs

@@ -2098,6 +2098,11 @@ pub fn render_metrics(status: &Status) -> String {
                 "Runs left for a later pass by the per-pass cap",
                 t.deferred,
             ),
+            (
+                "started_push",
+                "Superseded push runs kept because a job was running",
+                t.started_push,
+            ),
             (
                 "raced",
                 "Runs that changed state before the janitor cancelled them",

src/janitor.rs

@@ -75,6 +75,9 @@ pub struct PassReport {
     pub would_cancel: u32,
     /// Runs left for a later pass by the per-pass cap.
     pub deferred: u32,
+    /// Superseded push runs left alone because one of their jobs is running.
+    /// GitHub shows a run as `queued` while any job waits for a runner.
+    pub started_push: u32,
     /// Runs that changed state between the listing and the cancel and were
     /// left alone.
     pub raced: u32,
@@ -97,6 +100,7 @@ impl PassReport {
         self.dedup_cancelled += o.dedup_cancelled;
         self.would_cancel += o.would_cancel;
         self.deferred += o.deferred;
+        self.started_push += o.started_push;
         self.raced += o.raced;
         self.uncancellable += o.uncancellable;
         self.errors += o.errors;
@@ -280,7 +284,7 @@ impl Janitor {
             }
         }
         tracing::debug!(
-            "janitor pass: considered {} superseded {} closed_pr {} superseded_push {} cancelled {} cancelled_running {} duplicates {} dedup_cancelled {} would_cancel {} deferred {} raced {} uncancellable {} errors {}",
+            "janitor pass: considered {} superseded {} closed_pr {} superseded_push {} cancelled {} cancelled_running {} duplicates {} dedup_cancelled {} would_cancel {} deferred {} started_push {} raced {} uncancellable {} errors {}",
             report.considered,
             report.superseded,
             report.closed_pr,
@@ -291,6 +295,7 @@ impl Janitor {
             report.dedup_cancelled,
             report.would_cancel,
             report.deferred,
+            report.started_push,
             report.raced,
             report.uncancellable,
             report.errors
@@ -376,6 +381,18 @@ impl Janitor {
                 report.deferred += 1;
                 continue;
             }
+            match self.has_running_job(repo, run.id).await {
+                Ok(false) => {}
+                Ok(true) => {
+                    report.started_push += 1;
+                    continue;
+                }
+                Err(e) => {
+                    tracing::warn!("janitor: {repo} run {}: {e}", run.id);
+                    report.errors += 1;
+                    continue;
+                }
+            }
             let why = format!(
                 "on {}: superseded by run {newer}",
                 run.head_branch.as_deref().unwrap_or("?")
@@ -447,6 +464,33 @@ impl Janitor {
     }
 
     /// Cancel `run` unless a fresh read shows its status or head changed.
+    /// Whether any job of `run` is running. A run reads `queued` while one job
+    /// waits for a runner, even when others have started.
+    async fn has_running_job(&self, repo: &str, run: u64) -> Result<bool, GithubError> {
+        #[derive(Deserialize)]
+        struct Jobs {
+            jobs: Vec<Job>,
+        }
+        #[derive(Deserialize)]
+        struct Job {
+            status: String,
+        }
+        for page in 1..=MAX_PAGES {
+            let jobs: Jobs = self
+                .get(&format!(
+                    "/repos/{repo}/actions/runs/{run}/jobs?filter=latest&per_page=100&page={page}"
+                ))
+                .await?;
+            if jobs.jobs.iter().any(|j| j.status == "in_progress") {
+                return Ok(true);
+            }
+            if jobs.jobs.len() < 100 {
+                break;
+            }
+        }
+        Ok(false)
+    }
+
     async fn cancel_if_unchanged(
         &self,
         repo: &str,

tests/janitor.rs

@@ -170,6 +170,33 @@ async fn older_queued_main_pushes_are_cancelled_and_running_ones_kept() {
     assert_eq!(cancelled, [1, 2]);
 }
 
+#[tokio::test]
+async fn a_superseded_main_push_with_a_running_job_is_kept() {
+    // GitHub shows a run as queued while any job waits for a runner, so a
+    // publish run building one image while its siblings wait reads `queued`.
+    let h = Harness::start(with(json!({"superseded_push_branches": ["main"]}))).await;
+    let push = |id: u64, sha: &str| {
+        json!({"id": id, "workflow_id": 1, "status": "queued", "event": "push",
+            "head_sha": sha, "head_branch": "main", "pull_requests": []})
+    };
+    {
+        let mut st = h.state();
+        st.pr_runs.push(push(1, "a"));
+        st.pr_runs.push(push(2, "b"));
+        st.jobs
+            .push(json!({"id": 10, "run_id": 1, "status": "in_progress"}));
+        st.jobs
+            .push(json!({"id": 11, "run_id": 1, "status": "queued"}));
+    }
+    let r = h.janitor_pass().await;
+    assert_eq!(
+        (r.superseded_push, r.started_push, r.cancelled),
+        (1, 1, 0),
+        "{r:?}"
+    );
+    assert!(h.state().cancelled.is_empty());
+}
+
 #[tokio::test]
 async fn the_per_pass_cap_defers_the_rest_to_later_passes() {
     let h = Harness::start(enabled(false, 2)).await;

tests/support/mod.rs

@@ -206,14 +206,19 @@ async fn runs(
 async fn jobs(
     State(s): State<Shared>,
     h: HeaderMap,
-    Path(_p): Path<(String, String, u64)>,
+    Path((_o, _r, run)): Path<(String, String, u64)>,
 ) -> Result<Json<Value>, StatusCode> {
-    check_token(&h)?;
+    // The janitor reads a run's jobs with its own token; queen jobs carry no run_id.
+    let janitor = check_janitor(&h).is_ok();
+    if !janitor {
+        check_token(&h)?;
+    }
     let state = lock(&s);
     let jobs: Vec<_> = state
         .jobs
         .iter()
         .filter(|job| !state.hide_completed_jobs || job["status"] != "completed")
+        .filter(|job| !janitor || job["run_id"] == run)
         .cloned()
         .collect();
     Ok(Json(json!({"jobs": jobs})))