comphead commented on code in PR #26760: URL: https://github.com/apache/datafusion/pull/26760#discussion_r4390655627
########## .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 Review Comment: The `setFailed` part is fixed, thanks. The cancelled part remains. A queue rebuild creates several groups at once, so one scan runs, the next waits as pending and a third replaces it. The replaced run shows as cancelled on a group commit that can land on `main`. From Oct 19 to 27, three Rust merge-group runs started within 60 s of each other 15 times. Could `merge_group` scans use a per-SHA group, or drop that trigger if the observation week shows it rarely finds anything? ########## ci/scripts/cancel_merge_queue_runs.cjs: ########## @@ -0,0 +1,245 @@ +// 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" && + run.run_attempt === 1 && + 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; + } +} + +// 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) { Review Comment: Thanks for handling the cap. Active merge-group runs stay far below 1,000 (167 Rust merge-group runs in all of Oct 19 to 27). Would failing the scan when `total_count >= 1000` be enough? It drops the recursive split and its two tests. ########## ci/scripts/test_cancel_merge_queue_runs.cjs: ########## @@ -0,0 +1,493 @@ +#!/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 { 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, + created_at: "2026-10-20T00:00:00Z", + ...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, + listResponse, + comparisons = {}, + eventName = "schedule", +} = {}) { + const calls = []; + const cancellations = []; + const failures = []; + const warnings = []; + const logs = []; + const refReads = new Map(); + const runReads = new Map(); + const actions = { + 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; + 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]; + calls.push(`cancel:${run_id}`); Review Comment: This pushes `cancel:${run_id}` a second time. Was it meant to record something else? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
