Reuse failed merge queue workflows

.github/workflows/ci.yml

@@ -23,10 +23,15 @@ permissions:
 
 jobs:
   merge_queue_noop:
-    name: Detect merge queue no-op
+    name: Merge queue no-op/reuse
     runs-on: ubuntu-latest
+    permissions:
+      actions: write
+      contents: read
+      pull-requests: read
     outputs:
-      skip: ${{ steps.compare.outputs.skip }}
+      # A reused failure also skips fresh jobs; the final step reports that failure on this commit.
+      skip: ${{ steps.compare.outputs.skip == 'true' || steps.reuse.outputs.reused == 'true' }}
     steps:
       - name: Checkout merge queue commit
         uses: actions/checkout@v4
@@ -70,6 +75,21 @@ jobs:
             echo "Merge queue commit ${GITHUB_SHA} differs from PR #${pr_number} head ${pr_head_sha}; running CI normally."
           fi
 
+      # The merge queue coordinator is part of cargo ci and needs the repository's pinned Rust toolchain.
+      - uses: dsherret/rust-toolchain-file@v1
+        if: ${{ github.event_name == 'merge_group' && steps.compare.outputs.skip != 'true' }}
+
+      - name: Reuse merge queue run
+        id: reuse
+        if: ${{ github.event_name == 'merge_group' && steps.compare.outputs.skip != 'true' }}
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: cargo ci other-workflows coordinate merge-queue-reuse
+
+      - name: Report merge queue workflow reuse failure on this commit
+        if: ${{ steps.reuse.outputs.failed == 'true' }}
+        run: exit 1
+
   upload-build-artifacts-linux:
     needs: [merge_queue_noop, lints]
     if: ${{ needs.merge_queue_noop.outputs.skip != 'true' }}

.github/workflows/internal-tests.yml

@@ -63,7 +63,7 @@ jobs:
           fi
           cargo run --quiet \
             --manifest-path tools/ci/Cargo.toml \
-            -- other-workflows coordinate-internal-tests "${args[@]}"
+            -- other-workflows coordinate internal-tests "${args[@]}"
 
       - name: Wait for Internal Tests to complete
         if: steps.filter.outputs.non_docs == 'true'

Cargo.toml

@@ -84,7 +84,7 @@ members = [
   "tools/ci/commands/docs",
   "tools/ci/commands/codeowners-check",
   "tools/ci/commands/cla-assistant",
-  "tools/ci/commands/coordinate-internal-tests",
+  "tools/ci/commands/workflow-coordinator",
   "tools/ci/commands/workflow-watch",
   "tools/ci/common",
   "tools/keynote-bench-harness",

tools/ci/commands/coordinate-internal-tests/Cargo.toml → tools/ci/commands/workflow-coordinator/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "ci-coordinate-internal-tests"
+name = "ci-workflow-coordinator"
 version = "0.1.0"
 edition.workspace = true
 

tools/ci/commands/coordinate-internal-tests/src/main.rs → tools/ci/commands/workflow-coordinator/src/main.rs

@@ -1,7 +1,7 @@
 #![allow(clippy::disallowed_macros)]
 
 use anyhow::{bail, ensure, Context, Result};
-use clap::Parser;
+use clap::{Parser, Subcommand};
 use duct::{cmd, Expression};
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
@@ -15,18 +15,31 @@ const PUBLIC_REPO: &str = "clockworklabs/SpacetimeDB";
 const PRIVATE_REPO: &str = "clockworklabs/SpacetimeDBPrivate";
 const PRIVATE_WORKFLOW: &str = "ci.yml";
 const PRIVATE_DEFAULT_BRANCH: &str = "master";
+const REUSE_JOB: &str = "Merge queue no-op/reuse"; // Must match ci.yml.
 
-/// Selects or starts the private workflow for a public Internal Tests run.
+/// Coordinates CI workflow runs.
 #[derive(Parser)]
-#[command(about = "Selects or starts the private workflow for a public Internal Tests run.")]
+#[command(about = "Coordinates CI workflow runs.")]
 struct Cli {
-    /// Immutable public commit to test.
-    #[arg(long)]
-    public_sha: String,
+    #[command(subcommand)]
+    command: CliCommand,
+}
+
+#[derive(Subcommand)]
+enum CliCommand {
+    /// Select or start internal tests for a public commit.
+    InternalTests {
+        /// Immutable public commit to test.
+        #[arg(long)]
+        public_sha: String,
 
-    /// Public pull request number, when coordinating a pull request run.
-    #[arg(long)]
-    public_pr_number: Option<u64>,
+        /// Public pull request number, when coordinating a pull request run.
+        #[arg(long)]
+        public_pr_number: Option<u64>,
+    },
+
+    /// Reuse an equivalent failed merge-queue workflow run.
+    MergeQueueReuse,
 }
 
 #[derive(Debug)]
@@ -141,23 +154,28 @@ fn workflow_runs(event: &str, head_sha: &str) -> Result<Vec<WorkflowRun>> {
     Ok(pages.into_iter().flat_map(|page| page.workflow_runs).collect())
 }
 
-fn workflow_run(run_id: u64) -> Result<WorkflowRunStatus> {
-    get(&format!("/repos/{PRIVATE_REPO}/actions/runs/{run_id}"))
+fn workflow_run(repo: &str, run_id: u64) -> Result<WorkflowRunStatus> {
+    get(&format!("/repos/{repo}/actions/runs/{run_id}"))
 }
 
-fn rerun_failed_jobs(run_id: u64) -> Result<()> {
-    cmd!(
-        "gh",
-        "run",
-        "rerun",
-        run_id.to_string(),
-        "--failed",
-        "--repo",
-        PRIVATE_REPO
-    )
-    .run()
-    .with_context(|| format!("failed to rerun unsuccessful jobs in private run {run_id}"))?;
-    Ok(())
+fn rerun_failed_jobs(repo: &str, run: &SelectedRun) -> Result<SelectedRun> {
+    cmd!("gh", "run", "rerun", run.id.to_string(), "--failed", "--repo", repo)
+        .run()
+        .with_context(|| format!("failed to rerun unsuccessful jobs in {repo} run {}", run.id))?;
+    for _ in 0..30 {
+        let current = workflow_run(repo, run.id)?;
+        if current.run_attempt > run.attempt {
+            return Ok(SelectedRun {
+                id: current.id,
+                status: current.status,
+                conclusion: current.conclusion,
+                attempt: current.run_attempt,
+                url: current.html_url,
+            });
+        }
+        std::thread::sleep(Duration::from_secs(2));
+    }
+    bail!("timed out waiting for {repo} run {} to start its rerun", run.id)
 }
 
 fn dispatch_workflow(public_sha: &str) -> Result<DispatchResponse> {
@@ -249,10 +267,16 @@ struct WorkflowRun {
     run_attempt: u64,
     html_url: String,
     created_at: String,
+    head_commit: Option<WorkflowRunCommit>,
     #[serde(default)]
     pull_requests: Vec<WorkflowRunPullRequest>,
 }
 
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
+struct WorkflowRunCommit {
+    tree_id: String,
+}
+
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
 struct WorkflowRunStatus {
     id: u64,
@@ -262,6 +286,17 @@ struct WorkflowRunStatus {
     html_url: String,
 }
 
+#[derive(Deserialize)]
+struct JobsPage {
+    jobs: Vec<Job>,
+}
+
+#[derive(Deserialize)]
+struct Job {
+    name: String,
+    conclusion: Option<String>,
+}
+
 #[derive(Serialize)]
 struct DispatchInputs<'a> {
     public_ref: &'a str,
@@ -394,29 +429,82 @@ fn prepare_existing_run(run: WorkflowRun) -> Result<CoordinatedRun> {
     }
 
     println!("Re-running unsuccessful jobs in the existing private run.");
-    rerun_failed_jobs(selected.id)?;
     Ok(CoordinatedRun {
-        selected: wait_for_rerun(&selected)?,
+        selected: rerun_failed_jobs(PRIVATE_REPO, &selected)?,
         did_start: true,
     })
 }
 
-/// Waits for GitHub to expose the new attempt so callers do not receive the stale completed result.
-fn wait_for_rerun(run: &SelectedRun) -> Result<SelectedRun> {
-    for _ in 0..30 {
-        let current = workflow_run(run.id)?;
-        if current.run_attempt > run.attempt {
-            return Ok(SelectedRun {
-                id: current.id,
-                status: current.status,
-                conclusion: current.conclusion,
-                attempt: current.run_attempt,
-                url: current.html_url,
-            });
+fn wait_for_completion(repo: &str, mut run: SelectedRun) -> Result<SelectedRun> {
+    while run.status != "completed" {
+        std::thread::sleep(Duration::from_secs(30));
+        let current = workflow_run(repo, run.id)?;
+        run.status = current.status;
+        run.conclusion = current.conclusion;
+        run.attempt = current.run_attempt;
+    }
+    Ok(run)
+}
+
+/// Whether this workflow failed itself, rather than only reporting a failure copied from an older run.
+fn has_original_failure(repo: &str, run_id: u64) -> Result<bool> {
+    let pages: Vec<JobsPage> = get_paginated(&format!(
+        "/repos/{repo}/actions/runs/{run_id}/jobs?filter=latest&per_page=100"
+    ))?;
+    let mut reuse_job_failed = false;
+    let mut other_job_failed = false;
+    for job in pages.into_iter().flat_map(|page| page.jobs) {
+        let failed = job
+            .conclusion
+            .as_deref()
+            .is_some_and(|conclusion| !matches!(conclusion, "success" | "skipped" | "neutral"));
+        if failed && job.name == REUSE_JOB {
+            reuse_job_failed = true;
+        } else if failed {
+            other_job_failed = true;
         }
-        std::thread::sleep(Duration::from_secs(2));
     }
-    bail!("timed out waiting for the private run to start its rerun")
+    Ok(!reuse_job_failed && other_job_failed)
+}
+
+fn previous_equivalent_run() -> Result<Option<WorkflowRun>> {
+    let current_tree = cmd!("git", "rev-parse", "HEAD^{tree}")
+        .read()
+        .context("failed to read the current Git tree")?;
+    let mut runs = get::<WorkflowRunsPage>(&format!(
+        "/repos/{PUBLIC_REPO}/actions/workflows/ci.yml/runs?event=merge_group&per_page=100"
+    ))?
+    .workflow_runs;
+    runs.sort_by(|left, right| right.created_at.cmp(&left.created_at));
+    for run in runs {
+        // Check the tree first: `has_original_failure` makes another API request.
+        if run
+            .head_commit
+            .as_ref()
+            // N.B. without checking this before the below condition, we can trigger a ton of API requests
+            .is_some_and(|commit| commit.tree_id == current_tree)
+            // N.B. This condition excludes us finding the _current_ workflow
+            && (run.conclusion.as_deref() == Some("success") || has_original_failure(PUBLIC_REPO, run.id)?)
+        {
+            return Ok(Some(run));
+        }
+    }
+    Ok(None)
+}
+
+fn coordinate_merge_queue_reuse() -> Result<()> {
+    let Some(run) = previous_equivalent_run()? else {
+        write_github_output("reused", false)?;
+        return Ok(());
+    };
+    println!("Found equivalent merge queue run: {}", run.html_url);
+    let mut selected = wait_for_completion(PUBLIC_REPO, run.into())?;
+    if should_rerun_failed_jobs(&selected) {
+        selected = wait_for_completion(PUBLIC_REPO, rerun_failed_jobs(PUBLIC_REPO, &selected)?)?;
+    }
+    write_github_output("failed", selected.conclusion.as_deref() != Some("success"))?;
+    write_github_output("reused", true)?;
+    Ok(())
 }
 
 /// Reuses or reruns `pull_request` CI for a public PR with a linked private PR.
@@ -453,14 +541,20 @@ fn write_github_output(name: &str, value: impl std::fmt::Display) -> Result<()>
     writeln!(output, "{name}={value}").context("failed to write GITHUB_OUTPUT")
 }
 
-/// Coordinates the public Internal Tests run without checking out or executing private code.
+/// Coordinates CI without checking out or executing private code.
 fn main() -> Result<()> {
-    let args = Cli::parse();
-    let private_source = resolve_private_source(args.public_pr_number)?;
+    let (public_sha, public_pr_number) = match Cli::parse().command {
+        CliCommand::InternalTests {
+            public_sha,
+            public_pr_number,
+        } => (public_sha, public_pr_number),
+        CliCommand::MergeQueueReuse => return coordinate_merge_queue_reuse(),
+    };
+    let private_source = resolve_private_source(public_pr_number)?;
 
     let coordinated = match private_source {
-        PrivateSource::LinkedPrivatePr { pull } => coordinate_linked_private_pr(&args.public_sha, &pull)?,
-        PrivateSource::PrivateMaster { sha } => coordinate_public_only(&args.public_sha, &sha)?,
+        PrivateSource::LinkedPrivatePr { pull } => coordinate_linked_private_pr(&public_sha, &pull)?,
+        PrivateSource::PrivateMaster { sha } => coordinate_public_only(&public_sha, &sha)?,
     };
 
     println!("View run: {}", coordinated.selected.url);
@@ -496,6 +590,9 @@ mod tests {
             run_attempt: 1,
             html_url: format!("https://example.test/{id}"),
             created_at: created_at.to_owned(),
+            head_commit: Some(WorkflowRunCommit {
+                tree_id: "tree".to_owned(),
+            }),
             pull_requests: Vec::new(),
         }
     }

tools/ci/src/main.rs

@@ -65,8 +65,8 @@ const COMMANDS: &[Command] = &[
         package: "ci-docs-build",
     },
     Command {
-        path: &["other-workflows", "coordinate-internal-tests"],
-        package: "ci-coordinate-internal-tests",
+        path: &["other-workflows", "coordinate"],
+        package: "ci-workflow-coordinator",
     },
     Command {
         path: &["other-workflows", "codeowners-check"],