From 594c190ca6fe3a8cecb29ec0baf96f97ff904688 Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Wed, 11 Nov 2026 00:05:42 +0100 Subject: [PATCH 1/4] ci: enforce production order for concurrent merge-group validation Part of #6035 --- .github/workflows/ci.yaml | 49 +++++++ .../production_outcome_test.go | 6 +- scripts/verify-merge-group-queued/main.go | 120 ++++++++++++++---- .../verify-merge-group-queued/main_test.go | 62 ++++++++- 4 files changed, 203 insertions(+), 34 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index c2583626c4..70a79233b1 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -4387,6 +4387,36 @@ jobs: kube_version="$(yq -er '.spec.cluster.kubernetesVersion | select(tag == "!!str" and . != "")' ksail.prod.yaml)" bash scripts/guard-pdb-selector-match.sh --kube-version "$kube_version" k8s + wait-for-production-turn: + name: ๐Ÿงพ Wait for the production deployment turn + needs: [changes] + if: github.event_name == 'merge_group' && needs.changes.outputs.k8s == 'true' + runs-on: ubuntu-latest + timeout-minutes: 90 + permissions: + contents: read + pull-requests: read + steps: + - name: ๐Ÿ“‘ Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + fetch-depth: 2 + + - name: โš™๏ธ Setup Go + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + cache: false + + # Wait outside the shared production lock so a later candidate cannot + # hold that lock while the first candidate still needs it to merge. + - name: ๐Ÿงพ Wait until the candidate is first on current main + env: + GH_TOKEN: ${{ github.token }} + QUEUE_RECEIPT: ${{ needs.changes.outputs.queue_receipt }} + run: go run ./scripts/verify-merge-group-queued wait + # Every merge-group validator that ci-required-checks enforces is listed here too. # A required validator the deploy does not wait for runs in parallel with it, so # production can receive a revision that validator then rejects (#4148). @@ -4412,6 +4442,7 @@ jobs: validate-root-verify-merge-group, validate-rgd-templates-merge-group, validate-ghcr-fanout-merge-group, + wait-for-production-turn, ] if: github.event_name == 'merge_group' && needs.changes.outputs.k8s == 'true' runs-on: ubuntu-latest @@ -4791,6 +4822,7 @@ jobs: validate-root-verify-merge-group, validate-rgd-templates-merge-group, validate-ghcr-fanout-merge-group, + wait-for-production-turn, deploy-prod, ] if: always() @@ -4921,8 +4953,25 @@ jobs: ${{ needs.validate-root-verify-merge-group.result }} ${{ needs.validate-rgd-templates-merge-group.result }} ${{ needs.validate-ghcr-fanout-merge-group.result }} + ${{ needs.wait-for-production-turn.result }} ${{ needs.deploy-prod.result }} + - name: ๐Ÿš€ Require the selected production deployment + env: + EVENT_NAME: ${{ github.event_name }} + K8S_CHANGED: ${{ needs.changes.outputs.k8s }} + DEPLOY_RESULT: ${{ needs.deploy-prod.result }} + run: | + case "$EVENT_NAME:$K8S_CHANGED" in + merge_group:true) expected=success ;; + merge_group:false|pull_request:true|pull_request:false) expected=skipped ;; + *) echo '::error::Cannot determine production deployment selection'; exit 1 ;; + esac + if [ "$DEPLOY_RESULT" != "$expected" ]; then + echo "::error::Production deployment result was $DEPLOY_RESULT; expected $expected" + exit 1 + fi + - name: ๐Ÿ›ก๏ธ Require successful repository guards env: REPOSITORY_GUARDS_RESULT: ${{ needs.validate-repository-guards.result }} diff --git a/scripts/validate-merge-group-heal/production_outcome_test.go b/scripts/validate-merge-group-heal/production_outcome_test.go index e0662ffdcc..f65b5e6a59 100644 --- a/scripts/validate-merge-group-heal/production_outcome_test.go +++ b/scripts/validate-merge-group-heal/production_outcome_test.go @@ -73,6 +73,8 @@ func TestProductionOutcomeMutations(t *testing.T) { jobAnchor := " # `true` only when the queue preflight refused" stepsAnchor := "stopped_before_production }}\n steps:\n - name: ๐Ÿ“‘ Checkout\n uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1\n with:\n" setupAnchor := stepsAnchor + " persist-credentials: false\n fetch-depth: 0\n\n - name: โš™๏ธ Setup Go\n uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0\n with:\n" + receiptLine := " QUEUE_RECEIPT: ${{ needs.changes.outputs.queue_receipt }}\n" + checkLine := " run: go run ./scripts/verify-merge-group-queued check\n" for _, tc := range []struct{ name, from, to string }{ {"output from another step", "${{ steps.production-outcome.outputs.stopped_before_production }}", "${{ steps.queue-preflight.outputs.stopped_before_production }}"}, {"extra output", " stopped_before_production: ${{ steps.production-outcome.outputs.stopped_before_production }}", " stopped_before_production: ${{ steps.production-outcome.outputs.stopped_before_production }}\n other: 'true'"}, @@ -89,8 +91,8 @@ func TestProductionOutcomeMutations(t *testing.T) { {"preflight is conditional", " id: queue-preflight\n", " id: queue-preflight\n if: success()\n"}, {"preflight is suppressed", " id: queue-preflight\n", " id: queue-preflight\n continue-on-error: true\n"}, {"preflight changes shell", " run: go run ./scripts/verify-merge-group-queued check\n", " shell: sh\n run: go run ./scripts/verify-merge-group-queued check\n"}, - {"preflight gains a variable", " QUEUE_RECEIPT: ${{ needs.changes.outputs.queue_receipt }}\n run: go run", " QUEUE_RECEIPT: ${{ needs.changes.outputs.queue_receipt }}\n GOFLAGS: -toolexec=./deploy\n run: go run"}, - {"preflight reads another receipt", " QUEUE_RECEIPT: ${{ needs.changes.outputs.queue_receipt }}\n run: go run", " QUEUE_RECEIPT: forged\n run: go run"}, + {"preflight gains a variable", receiptLine + checkLine, receiptLine + " GOFLAGS: -toolexec=./deploy\n" + checkLine}, + {"preflight reads another receipt", receiptLine + checkLine, " QUEUE_RECEIPT: forged\n" + checkLine}, {"preflight runs elsewhere", " run: go run ./scripts/verify-merge-group-queued check\n", " working-directory: other\n run: go run ./scripts/verify-merge-group-queued check\n"}, {"workflow environment", "\njobs:\n", "\nenv:\n BASH_ENV: ./deploy.sh\njobs:\n"}, {"workflow defaults", "\njobs:\n", "\ndefaults:\n run:\n working-directory: other\njobs:\n"}, diff --git a/scripts/verify-merge-group-queued/main.go b/scripts/verify-merge-group-queued/main.go index e17945151d..9f5f174532 100644 --- a/scripts/verify-merge-group-queued/main.go +++ b/scripts/verify-merge-group-queued/main.go @@ -44,12 +44,20 @@ type receipt struct { } type oid struct{ OID string } + +type witness struct { + receipt + main string + position int +} + type response struct { Errors json.RawMessage Data struct { Repository struct { NameWithOwner string Ref struct{ Target oid } + Main struct{ Target oid } MergeQueue struct{ ID string } PullRequest struct { Number int @@ -57,6 +65,7 @@ type response struct { IsInMergeQueue bool Entry struct { ID, EnqueuedAt string + Position int HeadCommit, BaseCommit oid MergeQueue struct{ ID string } PullRequest struct { @@ -104,23 +113,23 @@ func parseEvent(data []byte, repo, sha, name string) (event, int, error) { // evaluate requires a complete, error-free association between the current PR, // its queue entry, the base commit and the exact synthetic Git ref. -func evaluate(data []byte, e event, number int) (receipt, error) { +func evaluate(data []byte, e event, number int) (witness, error) { var r response if err := json.Unmarshal(data, &r); err != nil { - return receipt{}, errors.New("unreadable queue witness") + return witness{}, errors.New("unreadable queue witness") } if len(r.Errors) > 0 && string(r.Errors) != "null" && string(r.Errors) != "[]" { - return receipt{}, errors.New("queue read contains GraphQL errors") + return witness{}, errors.New("queue read contains GraphQL errors") } repo := r.Data.Repository pr, entry := repo.PullRequest, repo.PullRequest.Entry enqueued, err := utc(entry.EnqueuedAt) if err != nil { - return receipt{}, err + return witness{}, err } created, err := utc(e.Group.Commit.Timestamp) if err != nil { - return receipt{}, err + return witness{}, err } if repo.NameWithOwner != repository || pr.Number != number || pr.State != "OPEN" || pr.BaseRefName != "main" || !pr.IsInMergeQueue || @@ -129,9 +138,13 @@ func evaluate(data []byte, e event, number int) (receipt, error) { !shaPattern.MatchString(pr.HeadRefOID) || entry.PullRequest.HeadRefOID != pr.HeadRefOID || entry.HeadCommit.OID != e.Group.HeadSHA || entry.BaseCommit.OID != e.Group.BaseSHA || repo.Ref.Target.OID != e.Group.HeadSHA || enqueued.After(created) { - return receipt{}, errors.New("merge group is absent, replaced or inconsistent with the active queue") + return witness{}, errors.New("merge group is absent, replaced or inconsistent with the active queue") } - return receipt{repository, e.Group.HeadRef, e.Group.HeadSHA, e.Group.BaseSHA, entry.ID, pr.HeadRefOID, number}, nil + return witness{ + receipt: receipt{repository, e.Group.HeadRef, e.Group.HeadSHA, e.Group.BaseSHA, entry.ID, pr.HeadRefOID, number}, + main: repo.Main.Target.OID, + position: entry.Position, + }, nil } // command bounds each read, disables interactive authentication and never emits @@ -157,7 +170,7 @@ func command(ctx context.Context, workspace, name string, args ...string) ([]byt } // The same query ties the PR entry to its queue and the actual synthetic ref. -const query = `query($number:Int!,$ref:String!){repository(owner:"devantler-tech",name:"platform"){nameWithOwner ref(qualifiedName:$ref){target{oid}} mergeQueue(branch:"main"){id} pullRequest(number:$number){number state baseRefName headRefOid isInMergeQueue mergeQueueEntry{id enqueuedAt headCommit{oid} baseCommit{oid} mergeQueue{id} pullRequest{number headRefOid}}}}}` +const query = `query($number:Int!,$ref:String!){repository(owner:"devantler-tech",name:"platform"){nameWithOwner ref(qualifiedName:$ref){target{oid}} main:ref(qualifiedName:"refs/heads/main"){target{oid}} mergeQueue(branch:"main"){id} pullRequest(number:$number){number state baseRefName headRefOid isInMergeQueue mergeQueueEntry{id enqueuedAt position headCommit{oid} baseCommit{oid} mergeQueue{id} pullRequest{number headRefOid}}}}}` // readQueue retries failed transport at most three times within the existing // deadline. A returned witness is evaluated once; refusals are never retried. @@ -183,11 +196,46 @@ func readQueue(ctx context.Context, workspace string, number int, ref string) ([ return nil, err } -// verify captures the run's initial source identity or rechecks that receipt. -// It does not infer a historical source head from the event or commit ancestry. +// waitForTurn polls only while the same candidate is valid but is still behind +// another entry or main has not advanced to its parent. The production job +// calls this once more without polling, after acquiring the shared lock. +func waitForTurn(ctx context.Context, read func() (witness, error), initial receipt, parent string, interval time.Duration, stdout io.Writer) error { + waiting := false + for { + current, err := read() + if err != nil { + return err + } + if current.receipt != initial { + return errors.New("initial queue entry or source head was replaced") + } + if current.position < 1 || !shaPattern.MatchString(current.main) { + return errors.New("missing queue position or main revision") + } + if current.position == 1 && current.main == parent { + return nil + } + if interval == 0 { + return errors.New("merge group is not first on the current main revision") + } + if !waiting { + _, _ = fmt.Fprintln(stdout, "Waiting for the previous merge group to finish.") + waiting = true + } + timer := time.NewTimer(interval) + select { + case <-ctx.Done(): + timer.Stop() + return errors.New("timed out waiting for the production deployment turn") + case <-timer.C: + } + } +} + +// verify captures the run's initial source identity or proves its turn to deploy. func verify(mode string, stdout io.Writer) error { - if mode != "capture" && mode != "check" { - return errors.New("usage: verify-merge-group-queued ") + if mode != "capture" && mode != "wait" && mode != "check" { + return errors.New("usage: verify-merge-group-queued ") } file, err := os.Open(os.Getenv("GITHUB_EVENT_PATH")) //nolint:gosec // Trusted Actions event file. if err != nil { @@ -219,24 +267,44 @@ func verify(mode string, stdout io.Writer) error { if err != nil || len(dirty) != 0 { return errors.New("checkout is not clean") } - if mode == "check" && os.Getenv("QUEUE_RECEIPT") == "" { - return errors.New("initial queue receipt is missing") - } - data, err = readQueue(ctx, workspace, number, e.Group.HeadRef) - if err != nil { - return err + deadline := 30 * time.Second + if mode == "wait" { + deadline = 85 * time.Minute } - current, err := evaluate(data, e, number) - if err != nil { - return err + queueContext, cancelQueue := context.WithTimeout(context.Background(), deadline) + defer cancelQueue() + read := func() (witness, error) { + readContext, cancelRead := context.WithTimeout(queueContext, 30*time.Second) + defer cancelRead() + reply, err := readQueue(readContext, workspace, number, e.Group.HeadRef) + if err != nil { + return witness{}, err + } + return evaluate(reply, e, number) } - if mode == "check" { + if mode != "capture" { var initial receipt - if err := json.Unmarshal([]byte(os.Getenv("QUEUE_RECEIPT")), &initial); err != nil || initial != current { - return errors.New("initial queue entry or source head was replaced") + if err := json.Unmarshal([]byte(os.Getenv("QUEUE_RECEIPT")), &initial); err != nil { + return errors.New("initial queue receipt is missing or unreadable") + } + revision, err := command(ctx, workspace, "git", "--no-replace-objects", "-c", "core.fsmonitor=false", "rev-list", "--parents", "-n", "1", "HEAD") + parents := strings.Fields(string(revision)) + if err != nil || len(parents) != 2 || parents[0] != e.Group.HeadSHA || !shaPattern.MatchString(parents[1]) { + return errors.New("merge group does not have exactly one verifiable parent") + } + var interval time.Duration + if mode == "wait" { + interval = 20 * time.Second + } + if err := waitForTurn(queueContext, read, initial, parents[1], interval, stdout); err != nil { + return err } } else { - encoded, err := json.Marshal(current) + current, err := read() + if err != nil { + return err + } + encoded, err := json.Marshal(current.receipt) if err != nil { return err } @@ -257,7 +325,7 @@ func verify(mode string, stdout io.Writer) error { // main preserves refusal as a failing step so production access cannot follow it. func main() { if len(os.Args) != 2 { - fmt.Fprintln(os.Stderr, "usage: verify-merge-group-queued ") + fmt.Fprintln(os.Stderr, "usage: verify-merge-group-queued ") os.Exit(2) } if err := verify(os.Args[1], os.Stdout); err != nil { diff --git a/scripts/verify-merge-group-queued/main_test.go b/scripts/verify-merge-group-queued/main_test.go index 2a7b98c2f9..ef65ac1971 100644 --- a/scripts/verify-merge-group-queued/main_test.go +++ b/scripts/verify-merge-group-queued/main_test.go @@ -157,8 +157,11 @@ func TestNativeCaptureAndPreflight(t *testing.T) { return strings.TrimSpace(string(out)) } git("init", "--quiet", "--initial-branch=main", "--object-format=sha1", workspace) - git("-C", workspace, "-c", "user.name=Queue Fixture", "-c", "user.email=queue@example.invalid", "-c", "commit.gpgsign=false", "commit", "--quiet", "--allow-empty", "-m", "queue") + for _, message := range []string{"main", "queue"} { + git("-C", workspace, "-c", "user.name=Queue Fixture", "-c", "user.email=queue@example.invalid", "-c", "commit.gpgsign=false", "commit", "--quiet", "--allow-empty", "-m", message) + } head := git("-C", workspace, "rev-parse", "HEAD") + parent := git("-C", workspace, "rev-parse", "HEAD^") eventPath, responsePath, logPath, output := filepath.Join(dir, "event.json"), filepath.Join(dir, "response.json"), filepath.Join(dir, "calls"), filepath.Join(dir, "output") write := func(path, data string) { t.Helper() @@ -167,7 +170,10 @@ func TestNativeCaptureAndPreflight(t *testing.T) { } } write(eventPath, strings.ReplaceAll(validEvent, groupSHA, head)) - write(responsePath, strings.ReplaceAll(validResponse, groupSHA, head)) + // The synthetic ref's suffix and event base are independent of the live + // target branch. Only the candidate's actual parent controls deployment. + reply := strings.NewReplacer(groupSHA, head, `"mergeQueueEntry":{`, `"mergeQueueEntry":{"position":1,`, `"nameWithOwner":`, `"main":{"target":{"oid":"`+parent+`"}},"nameWithOwner":`).Replace(validResponse) + write(responsePath, reply) write(output, "") script := "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$QUEUE_TEST_LOG\"\n[ \"$QUEUE_TEST_FAIL\" != 1 ] || exit 1\nif [ -n \"$QUEUE_TEST_FAIL_ONCE\" ] && [ ! -f \"$QUEUE_TEST_FAIL_ONCE\" ]; then printf 'failed\\n' > \"$QUEUE_TEST_FAIL_ONCE\"; exit 1; fi\ncat \"$QUEUE_TEST_RESPONSE\"\n" if err := os.WriteFile(filepath.Join(bin, "gh"), []byte(script), 0o700); err != nil { @@ -198,6 +204,21 @@ func TestNativeCaptureAndPreflight(t *testing.T) { if err := verify("check", &stdout); err != nil { t.Fatal(err) } + if err := verify("wait", &stdout); err != nil { + t.Fatal(err) + } + for _, tc := range []struct{ name, from, to string }{ + {"position two", `"position":1`, `"position":2`}, + {"missing position", `"position":1,`, ""}, + {"stale parent", `"main":{"target":{"oid":"` + parent + `"}}`, `"main":{"target":{"oid":"` + baseSHA + `"}}`}, + } { + t.Run(tc.name, func(t *testing.T) { + write(responsePath, strings.Replace(reply, tc.from, tc.to, 1)) + if err := verify("check", &stdout); err == nil { + t.Fatal("out-of-order candidate passed the production check") + } + }) + } log, err := os.ReadFile(logPath) if err != nil || !strings.Contains(string(log), "api --hostname github.com graphql") || !strings.Contains(string(log), "-F number=4891") || !strings.Contains(string(log), "-f ref="+headRef) { t.Fatalf("wrong native read: %s, %v", log, err) @@ -209,7 +230,7 @@ func TestNativeCaptureAndPreflight(t *testing.T) { {"evicted", `"isInMergeQueue":true`, `"isInMergeQueue":false`}, } { t.Run(tc.name, func(t *testing.T) { - write(responsePath, strings.ReplaceAll(strings.ReplaceAll(validResponse, groupSHA, head), tc.from, tc.to)) + write(responsePath, strings.ReplaceAll(reply, tc.from, tc.to)) before, _ := os.ReadFile(logPath) if err := verify("check", &stdout); err == nil { t.Fatal("accepted changed receipt") @@ -220,7 +241,7 @@ func TestNativeCaptureAndPreflight(t *testing.T) { } }) } - write(responsePath, strings.ReplaceAll(validResponse, groupSHA, head)) + write(responsePath, reply) t.Setenv("QUEUE_TEST_FAIL_ONCE", filepath.Join(dir, "one-transport-failure")) beforeRetry, _ := os.ReadFile(logPath) if err := verify("check", &stdout); err != nil { @@ -290,12 +311,41 @@ func TestReceiptJSON(t *testing.T) { if err != nil { t.Fatal(err) } - data, err := json.Marshal(r) + data, err := json.Marshal(r.receipt) if err != nil { t.Fatal(err) } var decoded receipt - if err := json.Unmarshal(data, &decoded); err != nil || decoded != r { + if err := json.Unmarshal(data, &decoded); err != nil || decoded != r.receipt { t.Fatal("receipt is not lossless") } } + +func TestWaitForTurnUsesSameCandidateAndCurrentParent(t *testing.T) { + for _, replaced := range []bool{false, true} { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + initial := receipt{Entry: "original"} + turns := []witness{ + {receipt: initial, main: sourceSHA, position: 2}, + {receipt: initial, main: sourceSHA, position: 1}, + {receipt: initial, main: baseSHA, position: 1}, + } + if replaced { + turns[2].receipt.Entry = "replacement" + } + var stdout bytes.Buffer + read := func() (witness, error) { + if len(turns) == 0 { + t.Fatal("queue was read after reaching a terminal result") + } + current := turns[0] + turns = turns[1:] + return current, nil + } + err := waitForTurn(ctx, read, initial, baseSHA, time.Millisecond, &stdout) + cancel() + if len(turns) != 0 || (err != nil) != replaced { + t.Fatalf("unexpected turn result (replaced=%t): %v, %d unread observations", replaced, err, len(turns)) + } + } +} From 452f8e375aa73ec8b5a84cd1e75dfc152c0e0b48 Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Wed, 11 Nov 2026 00:19:09 +0100 Subject: [PATCH 2/4] test: target the production job in signing contract regressions --- scripts/validate-dr-signing/main_test.go | 62 ++++++++---------------- 1 file changed, 19 insertions(+), 43 deletions(-) diff --git a/scripts/validate-dr-signing/main_test.go b/scripts/validate-dr-signing/main_test.go index 2aae4d7f6a..adfde270a7 100644 --- a/scripts/validate-dr-signing/main_test.go +++ b/scripts/validate-dr-signing/main_test.go @@ -535,6 +535,22 @@ func TestCDWiringRejectsEachAblation(t *testing.T) { } } +// mutateMergeQueueDeploy keeps negative controls on the production job even +// when another job has the same condition or dependency. +func mutateMergeQueueDeploy(workflow, old, replacement string) string { + const deployHeader = "\n deploy-prod:\n" + const nextJob = "\n merge-group-queue-membership:\n" + before, fromDeploy, found := strings.Cut(workflow, deployHeader) + if !found { + return workflow + } + deploy, after, found := strings.Cut(fromDeploy, nextJob) + if !found { + return workflow + } + return before + deployHeader + strings.Replace(deploy, old, replacement, 1) + nextJob + after +} + func TestMergeQueueProductionRoutesUseTheCheckedAction(t *testing.T) { t.Parallel() @@ -579,11 +595,10 @@ func TestMergeQueueProductionRoutesUseTheCheckedAction(t *testing.T) { // signing validator runs inside `changes`, so the publishing job runs // behind a gate that already failed while this check reports success. "merge-queue deploy overrides a failed dependency with always()": func(s string) string { - return strings.Replace( + return mutateMergeQueueDeploy( s, " if: github.event_name == 'merge_group' && needs.changes.outputs.k8s == 'true'", " if: always() && github.event_name == 'merge_group' && needs.changes.outputs.k8s == 'true'", - 1, ) }, // #2942 on the route production normally takes. Requiring the job to @@ -1596,46 +1611,7 @@ func TestMergeQueueContractGateIsEnforced(t *testing.T) { validatorLine = " go run ./scripts/validate-dr-signing .github/workflows/dr-rebuild.yaml ksail.prod.yaml\n" testLine = " go test ./scripts/validate-dr-signing\n" gateJobHeader = " validate-publication-contract:\n" - deployNeeds = ` needs: - [ - changes, - validate-repository-guards, - validate-independent-guards, - test-publish-release-recovery, - validate-publication-generator, - validate-go-regressions, - validate-webapp-compiler, - validate-consumer-discovery, - validate-floating-image-tags, - validate-helm-post-renderers, - validate-eks-authorization, - validate-publication-contract, - validate-matcher-efficacy, - validate-talos, - validate-root-verify-merge-group, - validate-rgd-templates-merge-group, - validate-ghcr-fanout-merge-group, - ]` - deployNeedsWithoutGate = ` needs: - [ - changes, - validate-repository-guards, - validate-independent-guards, - test-publish-release-recovery, - validate-publication-generator, - validate-go-regressions, - validate-webapp-compiler, - validate-consumer-discovery, - validate-floating-image-tags, - validate-helm-post-renderers, - validate-eks-authorization, - validate-matcher-efficacy, - validate-talos, - validate-root-verify-merge-group, - validate-rgd-templates-merge-group, - validate-ghcr-fanout-merge-group, - ]` - healNeeds = ` needs: + healNeeds = ` needs: [ changes, deploy-prod, @@ -1713,7 +1689,7 @@ func TestMergeQueueContractGateIsEnforced(t *testing.T) { // are not redundant. "deploy-prod no longer requires the gate job": { func(s string) string { - return strings.Replace(s, deployNeeds, deployNeedsWithoutGate, 1) + return mutateMergeQueueDeploy(s, " validate-publication-contract,\n", "") }, "does not require validate-publication-contract", }, From d2b3affef6265daa424dc04fc27a9d8e5ac1aaf5 Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Wed, 11 Nov 2026 00:55:12 +0100 Subject: [PATCH 3/4] test: use the promoted receipt field in the queue regression --- scripts/verify-merge-group-queued/main_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/verify-merge-group-queued/main_test.go b/scripts/verify-merge-group-queued/main_test.go index ef65ac1971..d231007eb9 100644 --- a/scripts/verify-merge-group-queued/main_test.go +++ b/scripts/verify-merge-group-queued/main_test.go @@ -331,7 +331,7 @@ func TestWaitForTurnUsesSameCandidateAndCurrentParent(t *testing.T) { {receipt: initial, main: baseSHA, position: 1}, } if replaced { - turns[2].receipt.Entry = "replacement" + turns[2].Entry = "replacement" } var stdout bytes.Buffer read := func() (witness, error) { From 3faeed6289a54ef439b81dc10f61739c81a6c83a Mon Sep 17 00:00:00 2001 From: Nikolai Emil Damm Date: Wed, 11 Nov 2026 01:11:16 +0100 Subject: [PATCH 4/4] fix: retry transient merge queue reads while waiting for production --- scripts/verify-merge-group-queued/main.go | 27 +++++++++------ .../verify-merge-group-queued/main_test.go | 33 +++++++++++++++++++ 2 files changed, 50 insertions(+), 10 deletions(-) diff --git a/scripts/verify-merge-group-queued/main.go b/scripts/verify-merge-group-queued/main.go index 9f5f174532..50311837b2 100644 --- a/scripts/verify-merge-group-queued/main.go +++ b/scripts/verify-merge-group-queued/main.go @@ -23,6 +23,7 @@ const repository = "devantler-tech/platform" var shaPattern = regexp.MustCompile(`^[0-9a-f]{40}$`) var queueRefPattern = regexp.MustCompile(`^refs/heads/gh-readonly-queue/main/pr-([1-9][0-9]*)-[0-9a-f]{40}$`) var idPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{8,200}$`) +var errQueueTransport = errors.New("bounded queue or checkout read failed") type event struct { Action string @@ -161,7 +162,10 @@ func command(ctx context.Context, workspace, name string, args ...string) ([]byt var out bytes.Buffer cmd.Stdout = &out if err := cmd.Run(); err != nil { - return nil, errors.New("bounded queue or checkout read failed") + if out.Len() == 0 { + return nil, errQueueTransport + } + return nil, errors.New("queue or checkout command returned an unsuccessful response") } if out.Len() > 4<<20 { return nil, errors.New("queue read is too large") @@ -182,6 +186,9 @@ func readQueue(ctx context.Context, workspace string, number int, ref string) ([ if err == nil { return data, nil } + if !errors.Is(err, errQueueTransport) { + return nil, err + } if attempt == 2 || ctx.Err() != nil { break } @@ -189,7 +196,7 @@ func readQueue(ctx context.Context, workspace string, number int, ref string) ([ select { case <-ctx.Done(): timer.Stop() - return nil, errors.New("queue read deadline expired") + return nil, fmt.Errorf("%w: read deadline expired", errQueueTransport) case <-timer.C: } } @@ -203,23 +210,23 @@ func waitForTurn(ctx context.Context, read func() (witness, error), initial rece waiting := false for { current, err := read() - if err != nil { + switch { + case err != nil && (interval == 0 || !errors.Is(err, errQueueTransport)): return err - } - if current.receipt != initial { + case err != nil: + // No response to evaluate. Use the next poll within the same deadline. + case current.receipt != initial: return errors.New("initial queue entry or source head was replaced") - } - if current.position < 1 || !shaPattern.MatchString(current.main) { + case current.position < 1 || !shaPattern.MatchString(current.main): return errors.New("missing queue position or main revision") - } - if current.position == 1 && current.main == parent { + case current.position == 1 && current.main == parent: return nil } if interval == 0 { return errors.New("merge group is not first on the current main revision") } if !waiting { - _, _ = fmt.Fprintln(stdout, "Waiting for the previous merge group to finish.") + _, _ = fmt.Fprintln(stdout, "Waiting for queue admission on the current main revision.") waiting = true } timer := time.NewTimer(interval) diff --git a/scripts/verify-merge-group-queued/main_test.go b/scripts/verify-merge-group-queued/main_test.go index d231007eb9..6a965cc58f 100644 --- a/scripts/verify-merge-group-queued/main_test.go +++ b/scripts/verify-merge-group-queued/main_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "os" "os/exec" "path/filepath" @@ -349,3 +350,35 @@ func TestWaitForTurnUsesSameCandidateAndCurrentParent(t *testing.T) { } } } + +func TestWaitForTurnRecoversOnlyFromTransportFailure(t *testing.T) { + initial := receipt{Entry: "original"} + // The production check never polls, even for a retryable transport error. + for _, tc := range []struct { + name string + readErr error + interval time.Duration + reads int + }{ + {"poll retries transport", errQueueTransport, time.Millisecond, 2}, + {"poll refuses invalid witness", errors.New("queue entry was replaced"), time.Millisecond, 1}, + {"production refuses transport", errQueueTransport, 0, 1}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + reads := 0 + read := func() (witness, error) { + reads++ + if reads == 1 { + return witness{}, tc.readErr + } + return witness{receipt: initial, main: baseSHA, position: 1}, nil + } + err := waitForTurn(ctx, read, initial, baseSHA, tc.interval, &bytes.Buffer{}) + if reads != tc.reads || (err == nil) != (tc.reads == 2) { + t.Fatalf("read %d times, want %d, got %v", reads, tc.reads, err) + } + }) + } +}