From 6ec32592412e772f6b514e0b9954928a15226670 Mon Sep 17 00:00:00 2001
From: Kumar Ujjawal <ujjawalpathak6@gmail.com>
Date: Mon, 26 Oct 2026 10:52:51 +0530
Subject: [PATCH 1/4] ci: cancel obsolete merge queue runs

---
 .github/workflows/dev.yml                   |   2 +-
 .github/workflows/merge_queue_cleanup.yml   |  72 ++++
 ci/scripts/cancel_merge_queue_runs.cjs      | 160 +++++++++
 ci/scripts/test_cancel_merge_queue_runs.cjs | 375 ++++++++++++++++++++
 dev/rust_lint.sh                            |   1 +
 docs/source/contributor-guide/index.md      |  19 +
 6 files changed, 628 insertions(+), 1 deletion(-)
 create mode 100644 .github/workflows/merge_queue_cleanup.yml
 create mode 100644 ci/scripts/cancel_merge_queue_runs.cjs
 create mode 100755 ci/scripts/test_cancel_merge_queue_runs.cjs

diff --git a/.github/workflows/dev.yml b/.github/workflows/dev.yml
index cd6d52094468e..56bcabeb9006d 100644
--- a/.github/workflows/dev.yml
+++ b/.github/workflows/dev.yml
@@ -88,7 +88,7 @@ jobs:
       - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1  # v7.1.0
         with:
           node-version: "24"
-      - run: node --test ci/scripts/test_fork_ci_status.cjs
+      - run: node --test ci/scripts/test_fork_ci_status.cjs ci/scripts/test_cancel_merge_queue_runs.cjs
 
   typos:
     name: Spell Check with Typos
diff --git a/.github/workflows/merge_queue_cleanup.yml b/.github/workflows/merge_queue_cleanup.yml
new file mode 100644
index 0000000000000..e1c11b0048c28
--- /dev/null
+++ b/.github/workflows/merge_queue_cleanup.yml
@@ -0,0 +1,72 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Merge queue cleanup
+
+on:
+  # A replacement group starts cleanup promptly; the sweep also catches groups
+  # removed without a replacement and runs that appear after an earlier scan.
+  merge_group:
+    types: [checks_requested]
+  schedule:
+    - cron: "*/15 * * * *"
+  workflow_dispatch:
+    inputs:
+      dry_run:
+        description: "Report obsolete runs without cancelling them"
+        type: boolean
+        default: true
+
+permissions:
+  contents: read
+  actions: write
+
+# Every invocation scans all active merge-group runs, so a newer pending
+# invocation can safely replace an older pending one. Do not interrupt a scan.
+concurrency:
+  group: merge-queue-cleanup
+  cancel-in-progress: false
+
+jobs:
+  cleanup:
+    if: github.repository == 'apache/datafusion'
+    runs-on: ubuntu-slim
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          ref: ${{ github.event.repository.default_branch }}
+          persist-credentials: false
+      - name: Cancel obsolete merge-group runs
+        # During the PR that introduces this workflow, main has no script yet.
+        # Always execute main's script, never code from the merge group.
+        if: hashFiles('ci/scripts/cancel_merge_queue_runs.cjs') != ''
+        uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
+        env:
+          DRY_RUN: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run }}
+        with:
+          script: |
+            const { run } = require('./ci/scripts/cancel_merge_queue_runs.cjs');
+            const dryRun = process.env.DRY_RUN !== 'false';
+            const results = await run({ github, context, core, dryRun });
+            await core.summary
+              .addHeading(dryRun ? 'Would cancel merge-group runs' : 'Requested merge-group cancellations')
+              .addTable([
+                [{ data: 'Run', header: true }, { data: 'Reason', header: true }],
+                ...results.map(({ id, reason }) => [String(id), reason]),
+              ])
+              .write();
diff --git a/ci/scripts/cancel_merge_queue_runs.cjs b/ci/scripts/cancel_merge_queue_runs.cjs
new file mode 100644
index 0000000000000..53fe5e8f4d406
--- /dev/null
+++ b/ci/scripts/cancel_merge_queue_runs.cjs
@@ -0,0 +1,160 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Called on each new merge group and on a schedule. The queue's exact Git
+// refs, rather than PR state or failing checks, determine which runs are stale.
+const ACTIVE = new Set([
+  "queued",
+  "in_progress",
+  "waiting",
+  "pending",
+  "requested",
+]);
+const CLEANUP_PATH = ".github/workflows/merge_queue_cleanup.yml";
+const QUEUE_REF = /^gh-readonly-queue\/.+\/pr-[1-9][0-9]*-[0-9a-f]{40}$/;
+
+function eligible(run, repo, currentRunId) {
+  return (
+    run.id !== currentRunId &&
+    run.event === "merge_group" &&
+    ACTIVE.has(run.status) &&
+    run.path?.split("@")[0] !== CLEANUP_PATH &&
+    run.repository?.full_name?.toLowerCase() === repo.toLowerCase() &&
+    run.head_repository?.full_name?.toLowerCase() === repo.toLowerCase() &&
+    QUEUE_REF.test(run.head_branch) &&
+    /^[0-9a-f]{40}$/.test(run.head_sha)
+  );
+}
+
+async function refSha(github, repo, branch) {
+  try {
+    const { data } = await github.rest.git.getRef({
+      ...repo,
+      ref: `heads/${branch}`,
+    });
+    // Only a missing ref or a known, different commit proves obsolescence.
+    // Unexpected responses and permission/server errors must not cancel CI.
+    if (
+      data.object?.type !== "commit" ||
+      !/^[0-9a-f]{40}$/.test(data.object.sha)
+    ) {
+      throw new Error(`Invalid Git ref response for ${branch}`);
+    }
+    return data.object.sha;
+  } catch (error) {
+    if (error.status === 404) return null;
+    throw error;
+  }
+}
+
+async function run({ github, context, core, dryRun = true }) {
+  const repo = { owner: context.repo.owner, repo: context.repo.repo };
+  const fullName = `${repo.owner}/${repo.repo}`;
+  if (fullName !== "apache/datafusion") return [];
+
+  // Finish pagination before cancelling: otherwise mutations shrink the active
+  // result sets and can cause later pages to skip runs.
+  const listed = new Map();
+  for (const status of ACTIVE) {
+    const runs = await github.paginate(
+      github.rest.actions.listWorkflowRunsForRepo,
+      {
+        ...repo,
+        event: "merge_group",
+        status,
+        per_page: 100,
+      }
+    );
+    for (const candidate of runs) listed.set(candidate.id, candidate);
+  }
+
+  const refs = new Map();
+  const results = [];
+  const errors = [];
+  for (const candidate of listed.values()) {
+    if (!eligible(candidate, fullName, context.runId)) continue;
+    try {
+      // Share the initial lookup among workflows for the same merge group.
+      // Cache failures too: an outage must not multiply requests for this ref.
+      if (!refs.has(candidate.head_branch)) {
+        refs.set(
+          candidate.head_branch,
+          refSha(github, repo, candidate.head_branch)
+        );
+      }
+      if ((await refs.get(candidate.head_branch)) === candidate.head_sha)
+        continue;
+
+      const { data: current } = await github.rest.actions.getWorkflowRun({
+        ...repo,
+        run_id: candidate.id,
+      });
+      // A completed run or a newly requested rerun is no longer this candidate.
+      if (
+        !eligible(current, fullName, context.runId) ||
+        current.head_branch !== candidate.head_branch ||
+        current.head_sha !== candidate.head_sha ||
+        current.run_attempt !== candidate.run_attempt
+      )
+        continue;
+
+      const head = await refSha(github, repo, current.head_branch);
+      if (head === current.head_sha) continue;
+      const reason = head === null ? "queue ref removed" : "queue ref replaced";
+      const description = `${current.id}: ${reason} (${current.head_branch}, ${current.head_sha})`;
+      if (dryRun) {
+        core.info(`Would cancel ${description}`);
+      } else {
+        try {
+          await github.rest.actions.cancelWorkflowRun({
+            ...repo,
+            run_id: current.id,
+          });
+        } catch (error) {
+          // A 409 can mean a completion race, but also an active run that
+          // GitHub cannot cancel. Only suppress it after confirming the race.
+          if (error.status === 409) {
+            const { data: latest } = await github.rest.actions.getWorkflowRun({
+              ...repo,
+              run_id: current.id,
+            });
+            if (
+              latest.status === "completed" ||
+              latest.run_attempt !== current.run_attempt
+            )
+              continue;
+          }
+          throw error;
+        }
+        core.info(`Requested cancellation of ${description}`);
+      }
+      results.push({ id: current.id, reason });
+    } catch (error) {
+      // One failed read/cancellation must not leave other obsolete groups running.
+      const message = `Run ${candidate.id}: ${error.message}`;
+      core.warning(message);
+      errors.push(message);
+    }
+  }
+  if (errors.length)
+    core.setFailed(
+      `${errors.length} merge queue cleanup errors; see warnings.`
+    );
+  return results;
+}
+
+module.exports = { run };
diff --git a/ci/scripts/test_cancel_merge_queue_runs.cjs b/ci/scripts/test_cancel_merge_queue_runs.cjs
new file mode 100755
index 0000000000000..82e04d68f9bcd
--- /dev/null
+++ b/ci/scripts/test_cancel_merge_queue_runs.cjs
@@ -0,0 +1,375 @@
+#!/usr/bin/env node
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+const assert = require("node:assert/strict");
+const { test } = require("node:test");
+const { readFileSync } = require("node:fs");
+const { execFileSync } = require("node:child_process");
+const { run } = require("./cancel_merge_queue_runs.cjs");
+
+const sha = "a".repeat(40);
+const otherSha = "b".repeat(40);
+const branch = `gh-readonly-queue/main/pr-123-${"c".repeat(40)}`;
+const fullName = "apache/datafusion";
+function workflow(id = 1, overrides = {}) {
+  return {
+    id,
+    event: "merge_group",
+    status: "in_progress",
+    path: ".github/workflows/rust.yml",
+    repository: { full_name: fullName },
+    head_repository: { full_name: fullName },
+    head_branch: branch,
+    head_sha: sha,
+    run_attempt: 1,
+    ...overrides,
+  };
+}
+const apiError = (status) =>
+  Object.assign(new Error(`HTTP ${status}`), { status });
+
+async function execute({
+  runs = [workflow()],
+  refs = {},
+  current = {},
+  cancelErrors = {},
+  dryRun = false,
+  omitDryRun = false,
+  owner = "apache",
+  listError,
+} = {}) {
+  const calls = [];
+  const cancellations = [];
+  const failures = [];
+  const warnings = [];
+  const logs = [];
+  const refReads = new Map();
+  const runReads = new Map();
+  const actions = {
+    listWorkflowRunsForRepo: () => {},
+    getWorkflowRun: async ({ run_id }) => {
+      calls.push(`get:${run_id}`);
+      const index = runReads.get(run_id) ?? 0;
+      runReads.set(run_id, index + 1);
+      let value = current[run_id] ?? runs.find((r) => r.id === run_id);
+      if (Array.isArray(value))
+        value = value[Math.min(index, value.length - 1)];
+      if (value instanceof Error) throw value;
+      return { data: value };
+    },
+    cancelWorkflowRun: async ({ owner, repo, run_id }) => {
+      assert.equal(`${owner}/${repo}`, fullName);
+      calls.push(`cancel:${run_id}`);
+      if (cancelErrors[run_id]) throw cancelErrors[run_id];
+      cancellations.push(run_id);
+    },
+  };
+  const github = {
+    rest: {
+      actions,
+      git: {
+        getRef: async ({ ref, owner, repo }) => {
+          assert.equal(`${owner}/${repo}`, fullName);
+          calls.push(`ref:${ref}`);
+          const index = refReads.get(ref) ?? 0;
+          refReads.set(ref, index + 1);
+          let value = refs[ref];
+          if (Array.isArray(value))
+            value = value[Math.min(index, value.length - 1)];
+          if (value instanceof Error) throw value;
+          if (value === undefined || value === null) throw apiError(404);
+          return {
+            data:
+              typeof value === "string"
+                ? { object: { type: "commit", sha: value } }
+                : value,
+          };
+        },
+      },
+    },
+    paginate: async (method, params) => {
+      assert.equal(method, actions.listWorkflowRunsForRepo);
+      assert.equal(params.event, "merge_group");
+      assert.equal(params.per_page, 100);
+      assert.equal(`${params.owner}/${params.repo}`, fullName);
+      calls.push(`list:${params.status}`);
+      if (params.status === listError) throw apiError(503);
+      return runs.filter((r) => r.status === params.status);
+    },
+  };
+  const results = await run({
+    github,
+    context: { repo: { owner, repo: "datafusion" }, runId: 999 },
+    core: {
+      info: (s) => logs.push(s),
+      warning: (s) => warnings.push(s),
+      setFailed: (s) => failures.push(s),
+    },
+    ...(omitDryRun ? {} : { dryRun }),
+  });
+  return { results, cancellations, calls, failures, warnings, logs };
+}
+
+test("keeps every workflow of an active group, including failed-check siblings", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, { path: ".github/workflows/extended.yml" }),
+    ],
+    refs: { [`heads/${branch}`]: sha },
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.calls.filter((c) => c.startsWith("ref:")).length, 1);
+});
+
+test("cancels all workflows of a removed group after collecting every active status", async () => {
+  const statuses = ["queued", "in_progress", "waiting", "pending", "requested"];
+  const result = await execute({
+    runs: statuses.map((status, i) => workflow(i + 1, { status })),
+  });
+  assert.deepEqual(result.cancellations.sort(), [1, 2, 3, 4, 5]);
+  assert.deepEqual(
+    result.calls.slice(0, 5),
+    statuses.map((s) => `list:${s}`)
+  );
+  assert.ok(result.results.every((r) => r.reason === "queue ref removed"));
+});
+
+test("cancels an old SHA when its queue ref points at a replacement", async () => {
+  const result = await execute({
+    runs: [workflow(1), workflow(2, { head_sha: otherSha })],
+    refs: { [`heads/${branch}`]: otherSha },
+  });
+  assert.deepEqual(result.cancellations, [1]);
+  assert.equal(result.results[0].reason, "queue ref replaced");
+});
+
+test("new base ref for one PR does not cancel its replacement or another PR", async () => {
+  const replacement = `gh-readonly-queue/main/pr-123-${"d".repeat(40)}`;
+  const unrelated = `gh-readonly-queue/main/pr-456-${"e".repeat(40)}`;
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, { head_branch: replacement }),
+      workflow(3, { head_branch: unrelated }),
+    ],
+    refs: { [`heads/${replacement}`]: sha, [`heads/${unrelated}`]: sha },
+  });
+  assert.deepEqual(result.cancellations, [1]);
+});
+
+test("checks the full queue ref, including a base branch containing slashes", async () => {
+  const release = `gh-readonly-queue/release/56/pr-123-${"c".repeat(40)}`;
+  const result = await execute({
+    runs: [workflow(1), workflow(2, { head_branch: release })],
+    refs: { [`heads/${release}`]: sha },
+  });
+  assert.deepEqual(result.cancellations, [1]);
+});
+
+test("ignores non-queue events, foreign repos, malformed refs, and cleanup runs", async () => {
+  const overrides = [
+    { event: "push" },
+    { event: "pull_request" },
+    { event: "workflow_dispatch" },
+    { head_repository: { full_name: "contributor/datafusion" } },
+    { repository: { full_name: "contributor/datafusion" } },
+    { head_repository: null },
+    { head_branch: "main" },
+    { head_branch: "gh-readonly-queue/main/pr-123-invalid" },
+    { head_sha: "invalid" },
+    { path: ".github/workflows/merge_queue_cleanup.yml" },
+    { path: ".github/workflows/merge_queue_cleanup.yml@refs/heads/main" },
+    { id: 999 },
+    { status: "completed" },
+  ];
+  const result = await execute({
+    runs: overrides.map((value, i) => workflow(i + 1, value)),
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.calls.length, 5);
+});
+
+test("does nothing on a fork", async () => {
+  const result = await execute({ owner: "contributor" });
+  assert.deepEqual(result.calls, []);
+});
+
+test("dry run performs the same checks without mutations", async () => {
+  const result = await execute({ dryRun: true });
+  assert.deepEqual(result.cancellations, []);
+  assert.deepEqual(result.results, [{ id: 1, reason: "queue ref removed" }]);
+  assert.match(result.logs[0], /Would cancel/);
+});
+
+test("defaults to a dry run when the caller does not specify a mode", async () => {
+  const result = await execute({ omitDryRun: true });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.results.length, 1);
+  assert.match(result.logs[0], /Would cancel/);
+});
+
+for (const status of [401, 403, 429, 500]) {
+  test(`a ${status} ref lookup cannot be mistaken for a removed ref`, async () => {
+    const result = await execute({
+      refs: { [`heads/${branch}`]: apiError(status) },
+    });
+    assert.deepEqual(result.cancellations, []);
+    assert.equal(result.failures.length, 1);
+  });
+}
+
+test("malformed ref responses cannot cancel a run", async () => {
+  for (const response of [
+    {},
+    { object: { type: "tag", sha: otherSha } },
+    { object: { type: "commit", sha: "" } },
+  ]) {
+    const result = await execute({ refs: { [`heads/${branch}`]: response } });
+    assert.deepEqual(result.cancellations, []);
+    assert.equal(result.failures.length, 1);
+  }
+});
+
+test("rechecks refs and keeps a group that is active again", async () => {
+  const result = await execute({ refs: { [`heads/${branch}`]: [null, sha] } });
+  assert.deepEqual(result.cancellations, []);
+});
+
+test("a failed second ref read cannot cancel a run", async () => {
+  const result = await execute({
+    refs: { [`heads/${branch}`]: [null, apiError(503)] },
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+});
+
+test("does not cancel completed runs, newer attempts, or changed run identities", async () => {
+  for (const update of [
+    { status: "completed" },
+    { run_attempt: 2 },
+    { head_sha: otherSha },
+    { head_branch: `gh-readonly-queue/main/pr-456-${"c".repeat(40)}` },
+    { event: "push" },
+  ]) {
+    const result = await execute({ current: { 1: workflow(1, update) } });
+    assert.deepEqual(result.cancellations, []);
+  }
+});
+
+test("409 on cancellation is a harmless completion race", async () => {
+  const result = await execute({
+    current: { 1: [workflow(), workflow(1, { status: "completed" })] },
+    cancelErrors: { 1: apiError(409) },
+  });
+  assert.deepEqual(result.failures, []);
+  assert.deepEqual(result.results, []);
+});
+
+test("409 for an unchanged active run reports a failed cancellation", async () => {
+  const result = await execute({ cancelErrors: { 1: apiError(409) } });
+  assert.equal(result.failures.length, 1);
+  assert.match(result.warnings[0], /HTTP 409/);
+});
+
+test("409 after a new rerun starts leaves that attempt alone", async () => {
+  const result = await execute({
+    current: { 1: [workflow(), workflow(1, { run_attempt: 2 })] },
+    cancelErrors: { 1: apiError(409) },
+  });
+  assert.deepEqual(result.failures, []);
+  assert.deepEqual(result.results, []);
+});
+
+test("409 on a read is still an error", async () => {
+  const result = await execute({ current: { 1: apiError(409) } });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+});
+
+test("one cancellation failure does not stop cleanup of other runs", async () => {
+  const result = await execute({
+    runs: [workflow(1), workflow(2)],
+    cancelErrors: { 1: apiError(403) },
+  });
+  assert.deepEqual(result.cancellations, [2]);
+  assert.equal(result.failures.length, 1);
+});
+
+test("one failed group lookup does not stop cleanup of another group", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, {
+        head_branch: `gh-readonly-queue/main/pr-456-${"c".repeat(40)}`,
+      }),
+    ],
+    refs: { [`heads/${branch}`]: apiError(503) },
+  });
+  assert.deepEqual(result.cancellations, [2]);
+  assert.equal(result.failures.length, 1);
+});
+
+test("a listing failure aborts before any cancellation", async () => {
+  await assert.rejects(execute({ listError: "requested" }), /HTTP 503/);
+});
+
+test("uses all paginated results and deduplicates repeated run IDs", async () => {
+  const runs = Array.from({ length: 205 }, (_, i) => workflow(i + 1));
+  runs.push(runs[0]);
+  const result = await execute({ runs });
+  assert.equal(result.cancellations.length, 205);
+  assert.equal(new Set(result.cancellations).size, 205);
+});
+
+test("workflow loads default-branch code and supports prompt cleanup, sweeping, and dry runs", () => {
+  const workflowPath = `${__dirname}/../../.github/workflows/merge_queue_cleanup.yml`;
+  const yaml = JSON.parse(
+    execFileSync(
+      "python3",
+      [
+        "-c",
+        "import json,sys,yaml; print(json.dumps(yaml.safe_load(sys.stdin)))",
+      ],
+      {
+        input: readFileSync(workflowPath),
+        encoding: "utf8",
+      }
+    )
+  );
+  const triggers = yaml.on ?? yaml.true;
+  assert.deepEqual(triggers.merge_group.types, ["checks_requested"]);
+  assert.equal(triggers.schedule[0].cron, "*/15 * * * *");
+  assert.equal(triggers.workflow_dispatch.inputs.dry_run.default, true);
+  assert.deepEqual(yaml.permissions, { contents: "read", actions: "write" });
+  assert.equal(yaml.concurrency["cancel-in-progress"], false);
+  assert.equal(
+    yaml.jobs.cleanup.if,
+    "github.repository == 'apache/datafusion'"
+  );
+  assert.equal(
+    yaml.jobs.cleanup.steps[0].with.ref,
+    "${{ github.event.repository.default_branch }}"
+  );
+  assert.equal(yaml.jobs.cleanup.steps[0].with["persist-credentials"], false);
+  assert.match(yaml.jobs.cleanup.steps[1].if, /hashFiles/);
+  assert.match(
+    yaml.jobs.cleanup.steps[1].with.script,
+    /cancel_merge_queue_runs\.cjs/
+  );
+});
diff --git a/dev/rust_lint.sh b/dev/rust_lint.sh
index b1dddaac3b65e..116bbf0b756f9 100755
--- a/dev/rust_lint.sh
+++ b/dev/rust_lint.sh
@@ -139,6 +139,7 @@ declare -a READONLY_STEPS=(
   "ci/scripts/check_rust_workflow_skips.py|false"
   "ci/scripts/test_rust_workflow_skips.py|false"
   "ci/scripts/test_fork_ci_status.cjs|false"
+  "ci/scripts/test_cancel_merge_queue_runs.cjs|false"
   "ci/scripts/check_large_files.sh|false"
   "ci/scripts/markdown_link_check.sh|false"
   "ci/scripts/security_audit.sh|false"
diff --git a/docs/source/contributor-guide/index.md b/docs/source/contributor-guide/index.md
index a9fc48b4bac46..149aa56022aca 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -284,3 +284,22 @@ To make this work:
    synced fork builds much faster.
 3. Rebase your branch before you push. The fork run tests your branch as it
    is, while the merge queue tests it merged with `main`.
+
+#### Obsolete Merge Queue Runs
+
+The **Merge queue cleanup** workflow cancels unfinished CI runs when their
+exact merge queue branch disappears or points to a different commit. It runs
+when a new merge group is created and every 15 minutes. This frees runners
+when the queue removes a PR or rebuilds groups after a change ahead of them.
+
+Cleanup checks the queue ref again before requesting cancellation. It keeps
+runs for active groups and does not cancel PR, push, or manually dispatched
+workflows. A failed check alone does not trigger cancellation. Each scan also
+skips runs whose attempt changed while it was checking them; a later scan can
+cancel the rerun if its group is still obsolete. To investigate an obsolete
+commit, create a branch at that commit and manually dispatch CI on the branch.
+
+Maintainers can preview cancellations by running **Merge queue cleanup**
+manually with its default `dry_run` input enabled. Disable this workflow in
+GitHub Actions to stop both automatic triggers. The cleanup executes the
+script from the default branch, rather than code from the merge group.

From 2ef0b6108aaf93a6441e4c47a0ea048af2108b04 Mon Sep 17 00:00:00 2001
From: Kumar Ujjawal <ujjawalpathak6@gmail.com>
Date: Tue, 27 Oct 2026 09:23:56 +0530
Subject: [PATCH 2/4] ci: preserve merged runs and stage queue cleanup rollout

---
 .github/workflows/merge_queue_cleanup.yml   |  19 +-
 ci/scripts/cancel_merge_queue_runs.cjs      | 115 +++++++++--
 ci/scripts/test_cancel_merge_queue_runs.cjs | 214 +++++++++++++++-----
 docs/source/contributor-guide/index.md      |  45 ++--
 4 files changed, 305 insertions(+), 88 deletions(-)

diff --git a/.github/workflows/merge_queue_cleanup.yml b/.github/workflows/merge_queue_cleanup.yml
index e1c11b0048c28..377a32c48851a 100644
--- a/.github/workflows/merge_queue_cleanup.yml
+++ b/.github/workflows/merge_queue_cleanup.yml
@@ -18,8 +18,8 @@
 name: Merge queue cleanup
 
 on:
-  # A replacement group starts cleanup promptly; the sweep also catches groups
-  # removed without a replacement and runs that appear after an earlier scan.
+  # New groups trigger an early scan. Ref deletion can lag this event; the
+  # sweep also catches groups removed without a replacement or missed earlier.
   merge_group:
     types: [checks_requested]
   schedule:
@@ -35,10 +35,10 @@ permissions:
   contents: read
   actions: write
 
-# Every invocation scans all active merge-group runs, so a newer pending
-# invocation can safely replace an older pending one. Do not interrupt a scan.
+# A dry run must not replace a pending cancellation scan. Within each mode,
+# a newer pending scan can replace an older one. Do not interrupt running scans.
 concurrency:
-  group: merge-queue-cleanup
+  group: merge-queue-cleanup-${{ vars.MERGE_QUEUE_CLEANUP_ENABLED != 'true' || (github.event_name == 'workflow_dispatch' && inputs.dry_run) }}
   cancel-in-progress: false
 
 jobs:
@@ -53,11 +53,12 @@ jobs:
           persist-credentials: false
       - name: Cancel obsolete merge-group runs
         # During the PR that introduces this workflow, main has no script yet.
-        # Always execute main's script, never code from the merge group.
+        # Execute main's copy of the script.
         if: hashFiles('ci/scripts/cancel_merge_queue_runs.cjs') != ''
         uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
         env:
-          DRY_RUN: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run }}
+          # Observe candidates for at least a week before enabling cancellation.
+          DRY_RUN: ${{ vars.MERGE_QUEUE_CLEANUP_ENABLED != 'true' || (github.event_name == 'workflow_dispatch' && inputs.dry_run) }}
         with:
           script: |
             const { run } = require('./ci/scripts/cancel_merge_queue_runs.cjs');
@@ -66,7 +67,7 @@ jobs:
             await core.summary
               .addHeading(dryRun ? 'Would cancel merge-group runs' : 'Requested merge-group cancellations')
               .addTable([
-                [{ data: 'Run', header: true }, { data: 'Reason', header: true }],
-                ...results.map(({ id, reason }) => [String(id), reason]),
+                ['Run', 'SHA', 'Attempt', 'Observed at (UTC)', 'Reason'].map(data => ({ data, header: true })),
+                ...results.map(({ id, sha, attempt, observed_at, reason }) => [String(id), sha, String(attempt), observed_at, reason]),
               ])
               .write();
diff --git a/ci/scripts/cancel_merge_queue_runs.cjs b/ci/scripts/cancel_merge_queue_runs.cjs
index 53fe5e8f4d406..4d40c2949245d 100644
--- a/ci/scripts/cancel_merge_queue_runs.cjs
+++ b/ci/scripts/cancel_merge_queue_runs.cjs
@@ -25,12 +25,13 @@ const ACTIVE = new Set([
   "requested",
 ]);
 const CLEANUP_PATH = ".github/workflows/merge_queue_cleanup.yml";
-const QUEUE_REF = /^gh-readonly-queue\/.+\/pr-[1-9][0-9]*-[0-9a-f]{40}$/;
+const QUEUE_REF = /^gh-readonly-queue\/(.+)\/pr-[1-9][0-9]*-[0-9a-f]{40}$/;
 
 function eligible(run, repo, currentRunId) {
   return (
     run.id !== currentRunId &&
     run.event === "merge_group" &&
+    run.run_attempt === 1 &&
     ACTIVE.has(run.status) &&
     run.path?.split("@")[0] !== CLEANUP_PATH &&
     run.repository?.full_name?.toLowerCase() === repo.toLowerCase() &&
@@ -61,6 +62,54 @@ async function refSha(github, repo, branch) {
   }
 }
 
+// Filtered Actions listings are capped at 1,000 results even with pagination.
+// Split full windows by creation time before trusting the result set.
+async function listRuns(github, repo, status, start, end) {
+  const params = {
+    ...repo,
+    event: "merge_group",
+    status,
+    per_page: 100,
+    created: `${new Date(start * 1000).toISOString()}..${new Date(
+      end * 1000
+    ).toISOString()}`,
+  };
+  const runs = [];
+  for (let page = 1; ; page++) {
+    const { data } = await github.rest.actions.listWorkflowRunsForRepo({
+      ...params,
+      page,
+    });
+    if (
+      !Number.isInteger(data.total_count) ||
+      data.total_count < 0 ||
+      !Array.isArray(data.workflow_runs)
+    ) {
+      throw new Error("Invalid workflow run listing");
+    }
+    if (
+      data.total_count >= 1000 ||
+      runs.length + data.workflow_runs.length >= 1000
+    ) {
+      if (start === end)
+        throw new Error("Cannot fully list merge-group runs within one second");
+      const middle = Math.floor((start + end) / 2);
+      const older = await listRuns(github, repo, status, start, middle);
+      const newer = await listRuns(github, repo, status, middle + 1, end);
+      return [...older, ...newer];
+    }
+    runs.push(...data.workflow_runs);
+    if (data.workflow_runs.length < 100) return runs;
+  }
+}
+
+function reportError(core, context, message) {
+  core.warning(message);
+  // Maintenance errors must not mark a merge-group commit as failing CI.
+  // Scheduled/manual scans still expose broken cleanup as workflow failures.
+  if (context.eventName !== "merge_group") core.setFailed(message);
+}
+
 async function run({ github, context, core, dryRun = true }) {
   const repo = { owner: context.repo.owner, repo: context.repo.repo };
   const fullName = `${repo.owner}/${repo.repo}`;
@@ -69,17 +118,19 @@ async function run({ github, context, core, dryRun = true }) {
   // Finish pagination before cancelling: otherwise mutations shrink the active
   // result sets and can cause later pages to skip runs.
   const listed = new Map();
-  for (const status of ACTIVE) {
-    const runs = await github.paginate(
-      github.rest.actions.listWorkflowRunsForRepo,
-      {
-        ...repo,
-        event: "merge_group",
-        status,
-        per_page: 100,
-      }
+  const end = Math.floor(Date.now() / 1000);
+  try {
+    for (const status of ACTIVE) {
+      const runs = await listRuns(github, repo, status, 0, end);
+      for (const candidate of runs) listed.set(candidate.id, candidate);
+    }
+  } catch (error) {
+    reportError(
+      core,
+      context,
+      `Cannot list merge-group runs: ${error.message}`
     );
-    for (const candidate of runs) listed.set(candidate.id, candidate);
+    return [];
   }
 
   const refs = new Map();
@@ -96,8 +147,18 @@ async function run({ github, context, core, dryRun = true }) {
           refSha(github, repo, candidate.head_branch)
         );
       }
-      if ((await refs.get(candidate.head_branch)) === candidate.head_sha)
+      if ((await refs.get(candidate.head_branch)) === candidate.head_sha) {
+        core.info(
+          `Kept active group ${JSON.stringify({
+            id: candidate.id,
+            sha: candidate.head_sha,
+            branch: candidate.head_branch,
+            observed_at: new Date().toISOString(),
+            trigger: context.eventName,
+          })}`
+        );
         continue;
+      }
 
       const { data: current } = await github.rest.actions.getWorkflowRun({
         ...repo,
@@ -114,8 +175,30 @@ async function run({ github, context, core, dryRun = true }) {
 
       const head = await refSha(github, repo, current.head_branch);
       if (head === current.head_sha) continue;
+      // A successful merge removes the queue ref too. Preserve checks on
+      // commits already in the target branch, including optional checks.
+      const base = QUEUE_REF.exec(current.head_branch)[1];
+      const { data: comparison } =
+        await github.rest.repos.compareCommitsWithBasehead({
+          ...repo,
+          basehead: `${base}...${current.head_sha}`,
+          per_page: 1,
+        });
+      if (["identical", "behind"].includes(comparison.status)) continue;
+      if (!["ahead", "diverged"].includes(comparison.status)) {
+        throw new Error(`Invalid comparison status for ${current.head_sha}`);
+      }
       const reason = head === null ? "queue ref removed" : "queue ref replaced";
-      const description = `${current.id}: ${reason} (${current.head_branch}, ${current.head_sha})`;
+      const result = {
+        id: current.id,
+        sha: current.head_sha,
+        branch: current.head_branch,
+        attempt: current.run_attempt,
+        reason,
+        observed_at: new Date().toISOString(),
+        trigger: context.eventName,
+      };
+      const description = JSON.stringify(result);
       if (dryRun) {
         core.info(`Would cancel ${description}`);
       } else {
@@ -142,7 +225,7 @@ async function run({ github, context, core, dryRun = true }) {
         }
         core.info(`Requested cancellation of ${description}`);
       }
-      results.push({ id: current.id, reason });
+      results.push(result);
     } catch (error) {
       // One failed read/cancellation must not leave other obsolete groups running.
       const message = `Run ${candidate.id}: ${error.message}`;
@@ -151,7 +234,9 @@ async function run({ github, context, core, dryRun = true }) {
     }
   }
   if (errors.length)
-    core.setFailed(
+    reportError(
+      core,
+      context,
       `${errors.length} merge queue cleanup errors; see warnings.`
     );
   return results;
diff --git a/ci/scripts/test_cancel_merge_queue_runs.cjs b/ci/scripts/test_cancel_merge_queue_runs.cjs
index 82e04d68f9bcd..a2c62cef77d0b 100755
--- a/ci/scripts/test_cancel_merge_queue_runs.cjs
+++ b/ci/scripts/test_cancel_merge_queue_runs.cjs
@@ -18,8 +18,6 @@
 
 const assert = require("node:assert/strict");
 const { test } = require("node:test");
-const { readFileSync } = require("node:fs");
-const { execFileSync } = require("node:child_process");
 const { run } = require("./cancel_merge_queue_runs.cjs");
 
 const sha = "a".repeat(40);
@@ -37,6 +35,7 @@ function workflow(id = 1, overrides = {}) {
     head_branch: branch,
     head_sha: sha,
     run_attempt: 1,
+    created_at: "2026-10-20T00:00:00Z",
     ...overrides,
   };
 }
@@ -52,6 +51,9 @@ async function execute({
   omitDryRun = false,
   owner = "apache",
   listError,
+  listResponse,
+  comparisons = {},
+  eventName = "schedule",
 } = {}) {
   const calls = [];
   const cancellations = [];
@@ -61,7 +63,30 @@ async function execute({
   const refReads = new Map();
   const runReads = new Map();
   const actions = {
-    listWorkflowRunsForRepo: () => {},
+    listWorkflowRunsForRepo: async (params) => {
+      assert.equal(params.event, "merge_group");
+      assert.equal(params.per_page, 100);
+      assert.equal(`${params.owner}/${params.repo}`, fullName);
+      calls.push(`list:${params.status}`);
+      if (params.status === listError) throw apiError(503);
+      if (listResponse) return { data: await listResponse(params) };
+      const [start, end] = params.created.split("..").map(Date.parse);
+      const filtered = runs.filter(
+        (r) =>
+          r.status === params.status &&
+          Date.parse(r.created_at) >= start &&
+          Date.parse(r.created_at) <= end
+      );
+      return {
+        data: {
+          total_count: filtered.length,
+          workflow_runs: filtered.slice(
+            (params.page - 1) * 100,
+            params.page * 100
+          ),
+        },
+      };
+    },
     getWorkflowRun: async ({ run_id }) => {
       calls.push(`get:${run_id}`);
       const index = runReads.get(run_id) ?? 0;
@@ -76,12 +101,22 @@ async function execute({
       assert.equal(`${owner}/${repo}`, fullName);
       calls.push(`cancel:${run_id}`);
       if (cancelErrors[run_id]) throw cancelErrors[run_id];
+      calls.push(`cancel:${run_id}`);
       cancellations.push(run_id);
     },
   };
   const github = {
     rest: {
       actions,
+      repos: {
+        compareCommitsWithBasehead: async ({ basehead, owner, repo }) => {
+          assert.equal(`${owner}/${repo}`, fullName);
+          calls.push(`compare:${basehead}`);
+          const response = comparisons[basehead] ?? "diverged";
+          if (response instanceof Error) throw response;
+          return { data: { status: response } };
+        },
+      },
       git: {
         getRef: async ({ ref, owner, repo }) => {
           assert.equal(`${owner}/${repo}`, fullName);
@@ -102,19 +137,10 @@ async function execute({
         },
       },
     },
-    paginate: async (method, params) => {
-      assert.equal(method, actions.listWorkflowRunsForRepo);
-      assert.equal(params.event, "merge_group");
-      assert.equal(params.per_page, 100);
-      assert.equal(`${params.owner}/${params.repo}`, fullName);
-      calls.push(`list:${params.status}`);
-      if (params.status === listError) throw apiError(503);
-      return runs.filter((r) => r.status === params.status);
-    },
   };
   const results = await run({
     github,
-    context: { repo: { owner, repo: "datafusion" }, runId: 999 },
+    context: { repo: { owner, repo: "datafusion" }, runId: 999, eventName },
     core: {
       info: (s) => logs.push(s),
       warning: (s) => warnings.push(s),
@@ -213,7 +239,11 @@ test("does nothing on a fork", async () => {
 test("dry run performs the same checks without mutations", async () => {
   const result = await execute({ dryRun: true });
   assert.deepEqual(result.cancellations, []);
-  assert.deepEqual(result.results, [{ id: 1, reason: "queue ref removed" }]);
+  assert.equal(result.results[0].id, 1);
+  assert.equal(result.results[0].sha, sha);
+  assert.equal(result.results[0].attempt, 1);
+  assert.equal(result.results[0].reason, "queue ref removed");
+  assert.ok(Number.isFinite(Date.parse(result.results[0].observed_at)));
   assert.match(result.logs[0], /Would cancel/);
 });
 
@@ -326,7 +356,10 @@ test("one failed group lookup does not stop cleanup of another group", async ()
 });
 
 test("a listing failure aborts before any cancellation", async () => {
-  await assert.rejects(execute({ listError: "requested" }), /HTTP 503/);
+  const result = await execute({ listError: "requested" });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+  assert.match(result.warnings[0], /HTTP 503/);
 });
 
 test("uses all paginated results and deduplicates repeated run IDs", async () => {
@@ -337,39 +370,124 @@ test("uses all paginated results and deduplicates repeated run IDs", async () =>
   assert.equal(new Set(result.cancellations).size, 205);
 });
 
-test("workflow loads default-branch code and supports prompt cleanup, sweeping, and dry runs", () => {
-  const workflowPath = `${__dirname}/../../.github/workflows/merge_queue_cleanup.yml`;
-  const yaml = JSON.parse(
-    execFileSync(
-      "python3",
-      [
-        "-c",
-        "import json,sys,yaml; print(json.dumps(yaml.safe_load(sys.stdin)))",
-      ],
-      {
-        input: readFileSync(workflowPath),
-        encoding: "utf8",
-      }
-    )
-  );
-  const triggers = yaml.on ?? yaml.true;
-  assert.deepEqual(triggers.merge_group.types, ["checks_requested"]);
-  assert.equal(triggers.schedule[0].cron, "*/15 * * * *");
-  assert.equal(triggers.workflow_dispatch.inputs.dry_run.default, true);
-  assert.deepEqual(yaml.permissions, { contents: "read", actions: "write" });
-  assert.equal(yaml.concurrency["cancel-in-progress"], false);
-  assert.equal(
-    yaml.jobs.cleanup.if,
-    "github.repository == 'apache/datafusion'"
-  );
-  assert.equal(
-    yaml.jobs.cleanup.steps[0].with.ref,
-    "${{ github.event.repository.default_branch }}"
+test("keeps merged commits even when their queue ref has disappeared or changed", async () => {
+  for (const status of ["identical", "behind"]) {
+    for (const ref of [null, otherSha]) {
+      const result = await execute({
+        comparisons: { [`main...${sha}`]: status },
+        refs: { [`heads/${branch}`]: ref },
+      });
+      assert.deepEqual(result.cancellations, []);
+      assert.deepEqual(result.failures, []);
+    }
+  }
+});
+
+test("compares against the actual target branch, including slashes", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1, {
+        head_branch: `gh-readonly-queue/release/56/pr-123-${sha}`,
+      }),
+    ],
+    comparisons: { [`release/56...${sha}`]: "behind" },
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.ok(result.calls.includes(`compare:release/56...${sha}`));
+});
+
+test("only ahead or diverged comparisons permit cancellation", async () => {
+  for (const status of ["ahead", "diverged"]) {
+    const result = await execute({
+      comparisons: { [`main...${sha}`]: status },
+    });
+    assert.deepEqual(result.cancellations, [1]);
+  }
+  for (const status of [
+    "unknown",
+    "",
+    apiError(404),
+    apiError(403),
+    apiError(503),
+  ]) {
+    const result = await execute({
+      comparisons: { [`main...${sha}`]: status },
+    });
+    assert.deepEqual(result.cancellations, []);
+    assert.equal(result.failures.length, 1);
+  }
+});
+
+test("reruns are exempt across successive scans, not just on attempt changes", async () => {
+  for (let scan = 0; scan < 2; scan++) {
+    const result = await execute({
+      runs: [workflow(1, { run_attempt: 2 }), workflow(2)],
+    });
+    assert.deepEqual(result.cancellations, [2]);
+    assert.ok(!result.calls.includes("get:1"));
+  }
+});
+
+test("merge-group maintenance errors warn without failing the commit", async () => {
+  for (const options of [
+    { listError: "requested" },
+    { refs: { [`heads/${branch}`]: apiError(403) } },
+    { comparisons: { [`main...${sha}`]: apiError(503) } },
+    { cancelErrors: { 1: apiError(409) } },
+  ]) {
+    const result = await execute({ eventName: "merge_group", ...options });
+    assert.deepEqual(result.failures, []);
+    assert.ok(result.warnings.length > 0);
+    assert.deepEqual(result.cancellations, []);
+  }
+});
+
+test("manual cleanup failures remain visible", async () => {
+  const result = await execute({
+    eventName: "workflow_dispatch",
+    cancelErrors: { 1: apiError(409) },
+  });
+  assert.equal(result.failures.length, 1);
+});
+
+test("an unresolved 409 is retried by later scans and is never reported as cancelled", async () => {
+  const first = await execute({ cancelErrors: { 1: apiError(409) } });
+  assert.deepEqual(first.results, []);
+  assert.equal(first.failures.length, 1);
+  const next = await execute();
+  assert.deepEqual(next.cancellations, [1]);
+});
+
+test("splits capped listings and collects all results before cancelling", async () => {
+  const runs = Array.from({ length: 1001 }, (_, i) =>
+    workflow(i + 10000, {
+      created_at: new Date(
+        Date.parse("2026-10-20T00:00:00Z") + i * 1000
+      ).toISOString(),
+    })
   );
-  assert.equal(yaml.jobs.cleanup.steps[0].with["persist-credentials"], false);
-  assert.match(yaml.jobs.cleanup.steps[1].if, /hashFiles/);
-  assert.match(
-    yaml.jobs.cleanup.steps[1].with.script,
-    /cancel_merge_queue_runs\.cjs/
+  const result = await execute({ runs });
+  assert.equal(result.cancellations.length, 1001);
+  assert.ok(
+    result.calls.findLastIndex((c) => c.startsWith("list:")) <
+      result.calls.findIndex((c) => c.startsWith("cancel:"))
   );
+  assert.ok(result.calls.filter((c) => c.startsWith("list:")).length > 10);
+});
+
+test("cannot cancel from a truncated single-second window", async () => {
+  const result = await execute({
+    runs: Array.from({ length: 1000 }, (_, i) => workflow(i + 1)),
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
+  assert.match(result.warnings[0], /within one second/);
+});
+
+test("malformed listing metadata aborts without cancelling", async () => {
+  const result = await execute({
+    listResponse: () => ({ workflow_runs: [workflow()] }),
+  });
+  assert.deepEqual(result.cancellations, []);
+  assert.equal(result.failures.length, 1);
 });
diff --git a/docs/source/contributor-guide/index.md b/docs/source/contributor-guide/index.md
index 149aa56022aca..06f68b087cbab 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -287,19 +287,32 @@ To make this work:
 
 #### Obsolete Merge Queue Runs
 
-The **Merge queue cleanup** workflow cancels unfinished CI runs when their
-exact merge queue branch disappears or points to a different commit. It runs
-when a new merge group is created and every 15 minutes. This frees runners
-when the queue removes a PR or rebuilds groups after a change ahead of them.
-
-Cleanup checks the queue ref again before requesting cancellation. It keeps
-runs for active groups and does not cancel PR, push, or manually dispatched
-workflows. A failed check alone does not trigger cancellation. Each scan also
-skips runs whose attempt changed while it was checking them; a later scan can
-cancel the rerun if its group is still obsolete. To investigate an obsolete
-commit, create a branch at that commit and manually dispatch CI on the branch.
-
-Maintainers can preview cancellations by running **Merge queue cleanup**
-manually with its default `dry_run` input enabled. Disable this workflow in
-GitHub Actions to stop both automatic triggers. The cleanup executes the
-script from the default branch, rather than code from the merge group.
+The **Merge queue cleanup** workflow identifies unfinished CI runs whose
+exact merge queue branch disappears or points to a different commit. It scans
+when a new merge group is created and every 15 minutes. Ref deletion can lag
+the new-group event, and scheduled scans can be delayed. These triggers do not
+guarantee a particular cancellation latency.
+
+Cleanup rechecks the queue ref and run before requesting cancellation. It
+preserves commits already in their target branch, active groups, and rerun
+attempts. Contributors can rerun obsolete jobs to investigate failures. PR,
+push, and manually dispatched workflows are excluded. A failed check alone
+does not trigger cancellation. API errors preserve affected runs; cleanup
+continues with other candidates. Errors produce warnings on merge-group scans
+and fail scheduled or manual scans. An unresolved cancellation is retried on
+a later scan; HTTP 409 alone is not treated as successful cancellation.
+
+Cancellation is disabled by default, including for manual runs. After at least
+one week of observation, maintainers can set the repository Actions variable
+`MERGE_QUEUE_CLEANUP_ENABLED` to `true`. Before enabling it, compare the logged
+candidate SHAs, attempts, reasons, and UTC observation times with subsequent
+queue and workflow outcomes. Check for false candidates and delays between
+replacement-group creation and detection. Dry runs measure detection, not
+actual cancellation latency or runner-instance shutdown costs.
+
+When cancellation is enabled, manual runs still default to `dry_run`. Dry runs
+use a separate concurrency group so they cannot displace cancellation scans.
+Remove the variable or set it to `false` to return to observation mode. Disable
+the workflow in GitHub Actions to stop both automatic triggers. The workflow
+checks out the default branch's script; the workflow definition itself comes
+from the triggering revision.

From aa88eb2a063f19f66cb3c84f9398cfdf13d8aba3 Mon Sep 17 00:00:00 2001
From: Kumar Ujjawal <ujjawalpathak6@gmail.com>
Date: Tue, 27 Oct 2026 22:33:37 +0530
Subject: [PATCH 3/4] ci: address merge queue cleanup review nits

---
 .github/workflows/merge_queue_cleanup.yml   |  8 ++--
 ci/scripts/cancel_merge_queue_runs.cjs      | 51 +++++++++------------
 ci/scripts/test_cancel_merge_queue_runs.cjs | 47 ++++++++-----------
 docs/source/contributor-guide/index.md      | 41 +++++------------
 4 files changed, 58 insertions(+), 89 deletions(-)

diff --git a/.github/workflows/merge_queue_cleanup.yml b/.github/workflows/merge_queue_cleanup.yml
index 377a32c48851a..1d6b2fc9115e6 100644
--- a/.github/workflows/merge_queue_cleanup.yml
+++ b/.github/workflows/merge_queue_cleanup.yml
@@ -35,10 +35,12 @@ permissions:
   contents: read
   actions: write
 
-# A dry run must not replace a pending cancellation scan. Within each mode,
-# a newer pending scan can replace an older one. Do not interrupt running scans.
+# Merge-group scans attach checks to commits that can land on main, so each
+# group SHA gets its own group: a replaced pending scan would show as cancelled.
+# Scheduled and manual scans share one group per mode, so a dry run cannot
+# replace a pending cancellation scan. Do not interrupt running scans.
 concurrency:
-  group: merge-queue-cleanup-${{ vars.MERGE_QUEUE_CLEANUP_ENABLED != 'true' || (github.event_name == 'workflow_dispatch' && inputs.dry_run) }}
+  group: merge-queue-cleanup-${{ github.event_name == 'merge_group' && github.event.merge_group.head_sha || (vars.MERGE_QUEUE_CLEANUP_ENABLED != 'true' || (github.event_name == 'workflow_dispatch' && inputs.dry_run)) }}
   cancel-in-progress: false
 
 jobs:
diff --git a/ci/scripts/cancel_merge_queue_runs.cjs b/ci/scripts/cancel_merge_queue_runs.cjs
index 4d40c2949245d..9ebec8942a496 100644
--- a/ci/scripts/cancel_merge_queue_runs.cjs
+++ b/ci/scripts/cancel_merge_queue_runs.cjs
@@ -63,17 +63,9 @@ async function refSha(github, repo, branch) {
 }
 
 // Filtered Actions listings are capped at 1,000 results even with pagination.
-// Split full windows by creation time before trusting the result set.
-async function listRuns(github, repo, status, start, end) {
-  const params = {
-    ...repo,
-    event: "merge_group",
-    status,
-    per_page: 100,
-    created: `${new Date(start * 1000).toISOString()}..${new Date(
-      end * 1000
-    ).toISOString()}`,
-  };
+// Active merge-group runs stay far below the cap, so a full listing is an error.
+async function listRuns(github, repo, status) {
+  const params = { ...repo, event: "merge_group", status, per_page: 100 };
   const runs = [];
   for (let page = 1; ; page++) {
     const { data } = await github.rest.actions.listWorkflowRunsForRepo({
@@ -87,16 +79,10 @@ async function listRuns(github, repo, status, start, end) {
     ) {
       throw new Error("Invalid workflow run listing");
     }
-    if (
-      data.total_count >= 1000 ||
-      runs.length + data.workflow_runs.length >= 1000
-    ) {
-      if (start === end)
-        throw new Error("Cannot fully list merge-group runs within one second");
-      const middle = Math.floor((start + end) / 2);
-      const older = await listRuns(github, repo, status, start, middle);
-      const newer = await listRuns(github, repo, status, middle + 1, end);
-      return [...older, ...newer];
+    if (data.total_count >= 1000) {
+      throw new Error(
+        `${data.total_count} ${status} runs exceed the listing cap`
+      );
     }
     runs.push(...data.workflow_runs);
     if (data.workflow_runs.length < 100) return runs;
@@ -118,10 +104,9 @@ async function run({ github, context, core, dryRun = true }) {
   // Finish pagination before cancelling: otherwise mutations shrink the active
   // result sets and can cause later pages to skip runs.
   const listed = new Map();
-  const end = Math.floor(Date.now() / 1000);
   try {
     for (const status of ACTIVE) {
-      const runs = await listRuns(github, repo, status, 0, end);
+      const runs = await listRuns(github, repo, status);
       for (const candidate of runs) listed.set(candidate.id, candidate);
     }
   } catch (error) {
@@ -134,6 +119,7 @@ async function run({ github, context, core, dryRun = true }) {
   }
 
   const refs = new Map();
+  const comparisons = new Map();
   const results = [];
   const errors = [];
   for (const candidate of listed.values()) {
@@ -177,13 +163,20 @@ async function run({ github, context, core, dryRun = true }) {
       if (head === current.head_sha) continue;
       // A successful merge removes the queue ref too. Preserve checks on
       // commits already in the target branch, including optional checks.
+      // Workflows of one group share the comparison, failures included.
       const base = QUEUE_REF.exec(current.head_branch)[1];
-      const { data: comparison } =
-        await github.rest.repos.compareCommitsWithBasehead({
-          ...repo,
-          basehead: `${base}...${current.head_sha}`,
-          per_page: 1,
-        });
+      const basehead = `${base}...${current.head_sha}`;
+      if (!comparisons.has(basehead)) {
+        comparisons.set(
+          basehead,
+          github.rest.repos.compareCommitsWithBasehead({
+            ...repo,
+            basehead,
+            per_page: 1,
+          })
+        );
+      }
+      const { data: comparison } = await comparisons.get(basehead);
       if (["identical", "behind"].includes(comparison.status)) continue;
       if (!["ahead", "diverged"].includes(comparison.status)) {
         throw new Error(`Invalid comparison status for ${current.head_sha}`);
diff --git a/ci/scripts/test_cancel_merge_queue_runs.cjs b/ci/scripts/test_cancel_merge_queue_runs.cjs
index a2c62cef77d0b..64d91ed26f266 100755
--- a/ci/scripts/test_cancel_merge_queue_runs.cjs
+++ b/ci/scripts/test_cancel_merge_queue_runs.cjs
@@ -35,7 +35,6 @@ function workflow(id = 1, overrides = {}) {
     head_branch: branch,
     head_sha: sha,
     run_attempt: 1,
-    created_at: "2026-10-20T00:00:00Z",
     ...overrides,
   };
 }
@@ -70,13 +69,8 @@ async function execute({
       calls.push(`list:${params.status}`);
       if (params.status === listError) throw apiError(503);
       if (listResponse) return { data: await listResponse(params) };
-      const [start, end] = params.created.split("..").map(Date.parse);
-      const filtered = runs.filter(
-        (r) =>
-          r.status === params.status &&
-          Date.parse(r.created_at) >= start &&
-          Date.parse(r.created_at) <= end
-      );
+      assert.equal(params.created, undefined);
+      const filtered = runs.filter((r) => r.status === params.status);
       return {
         data: {
           total_count: filtered.length,
@@ -101,7 +95,6 @@ async function execute({
       assert.equal(`${owner}/${repo}`, fullName);
       calls.push(`cancel:${run_id}`);
       if (cancelErrors[run_id]) throw cancelErrors[run_id];
-      calls.push(`cancel:${run_id}`);
       cancellations.push(run_id);
     },
   };
@@ -458,30 +451,28 @@ test("an unresolved 409 is retried by later scans and is never reported as cance
   assert.deepEqual(next.cancellations, [1]);
 });
 
-test("splits capped listings and collects all results before cancelling", async () => {
-  const runs = Array.from({ length: 1001 }, (_, i) =>
-    workflow(i + 10000, {
-      created_at: new Date(
-        Date.parse("2026-10-20T00:00:00Z") + i * 1000
-      ).toISOString(),
-    })
-  );
-  const result = await execute({ runs });
-  assert.equal(result.cancellations.length, 1001);
-  assert.ok(
-    result.calls.findLastIndex((c) => c.startsWith("list:")) <
-      result.calls.findIndex((c) => c.startsWith("cancel:"))
-  );
-  assert.ok(result.calls.filter((c) => c.startsWith("list:")).length > 10);
-});
-
-test("cannot cancel from a truncated single-second window", async () => {
+test("a capped listing aborts without cancelling", async () => {
   const result = await execute({
     runs: Array.from({ length: 1000 }, (_, i) => workflow(i + 1)),
   });
   assert.deepEqual(result.cancellations, []);
+  assert.deepEqual(
+    result.calls.filter((c) => !c.startsWith("list:")),
+    []
+  );
   assert.equal(result.failures.length, 1);
-  assert.match(result.warnings[0], /within one second/);
+  assert.match(result.warnings[0], /exceed the listing cap/);
+});
+
+test("workflows of one group share the target-branch comparison", async () => {
+  const result = await execute({
+    runs: [
+      workflow(1),
+      workflow(2, { path: ".github/workflows/extended.yml" }),
+    ],
+  });
+  assert.deepEqual(result.cancellations, [1, 2]);
+  assert.equal(result.calls.filter((c) => c.startsWith("compare:")).length, 1);
 });
 
 test("malformed listing metadata aborts without cancelling", async () => {
diff --git a/docs/source/contributor-guide/index.md b/docs/source/contributor-guide/index.md
index 06f68b087cbab..28c26205081a7 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -287,32 +287,15 @@ To make this work:
 
 #### Obsolete Merge Queue Runs
 
-The **Merge queue cleanup** workflow identifies unfinished CI runs whose
-exact merge queue branch disappears or points to a different commit. It scans
-when a new merge group is created and every 15 minutes. Ref deletion can lag
-the new-group event, and scheduled scans can be delayed. These triggers do not
-guarantee a particular cancellation latency.
-
-Cleanup rechecks the queue ref and run before requesting cancellation. It
-preserves commits already in their target branch, active groups, and rerun
-attempts. Contributors can rerun obsolete jobs to investigate failures. PR,
-push, and manually dispatched workflows are excluded. A failed check alone
-does not trigger cancellation. API errors preserve affected runs; cleanup
-continues with other candidates. Errors produce warnings on merge-group scans
-and fail scheduled or manual scans. An unresolved cancellation is retried on
-a later scan; HTTP 409 alone is not treated as successful cancellation.
-
-Cancellation is disabled by default, including for manual runs. After at least
-one week of observation, maintainers can set the repository Actions variable
-`MERGE_QUEUE_CLEANUP_ENABLED` to `true`. Before enabling it, compare the logged
-candidate SHAs, attempts, reasons, and UTC observation times with subsequent
-queue and workflow outcomes. Check for false candidates and delays between
-replacement-group creation and detection. Dry runs measure detection, not
-actual cancellation latency or runner-instance shutdown costs.
-
-When cancellation is enabled, manual runs still default to `dry_run`. Dry runs
-use a separate concurrency group so they cannot displace cancellation scans.
-Remove the variable or set it to `false` to return to observation mode. Disable
-the workflow in GitHub Actions to stop both automatic triggers. The workflow
-checks out the default branch's script; the workflow definition itself comes
-from the triggering revision.
+The **Merge queue cleanup** workflow finds unfinished merge queue CI runs whose
+queue branch was removed or now points to a different commit. It scans when a
+new merge group is created and every 15 minutes.
+
+Cleanup keeps runs for active groups, commits already in their target branch,
+and rerun attempts. Contributors can rerun an obsolete job to investigate a
+failure. A failed check alone does not cause cancellation.
+
+The workflow only reports candidates until a maintainer sets the repository
+Actions variable `MERGE_QUEUE_CLEANUP_ENABLED` to `true`. Manual runs default
+to `dry_run`. To stop cancellation, remove the variable or set it to `false`.
+The rollout plan is in [#25148](https://github.com/apache/datafusion/issues/25148).

From c39896e2ece1e78b10993532f815d673565b3051 Mon Sep 17 00:00:00 2001
From: Kumar Ujjawal <ujjawalpathak6@gmail.com>
Date: Tue, 27 Oct 2026 22:36:49 +0530
Subject: [PATCH 4/4] docs: remove rollout plan link from merge queue cleanup
 section

---
 docs/source/contributor-guide/index.md | 1 -
 1 file changed, 1 deletion(-)

diff --git a/docs/source/contributor-guide/index.md b/docs/source/contributor-guide/index.md
index 28c26205081a7..5df074618de08 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -298,4 +298,3 @@ failure. A failed check alone does not cause cancellation.
 The workflow only reports candidates until a maintainer sets the repository
 Actions variable `MERGE_QUEUE_CLEANUP_ENABLED` to `true`. Manual runs default
 to `dry_run`. To stop cancellation, remove the variable or set it to `false`.
-The rollout plan is in [#25148](https://github.com/apache/datafusion/issues/25148).