diff --git a/test/pr-e2e-dispatch-reconciliation.test.ts b/test/pr-e2e-dispatch-reconciliation.test.ts index d903fa43caa..929018f8aad 100644 --- a/test/pr-e2e-dispatch-reconciliation.test.ts +++ b/test/pr-e2e-dispatch-reconciliation.test.ts @@ -10,6 +10,7 @@ import { DispatchReconciliationError, dispatchWorkflowWithReconciliation, } from "../tools/e2e/pr-e2e-dispatch-reconciliation.mts"; +import type { DispatchNotObservedReceipt } from "../tools/e2e/pr-e2e-retry-receipt.mts"; const REPOSITORY = "NVIDIA/NemoClaw"; const WORKFLOW_SHA = "d".repeat(40); @@ -62,10 +63,50 @@ function workflowRun(runId = 23, overrides: Record = {}) { }; } +function unrelatedWorkflowRun(runId: number) { + return workflowRun(runId, { + name: `E2E unrelated run ${runId}`, + display_title: `E2E unrelated run ${runId}`, + }); +} + function inventory(runs: unknown[]) { return { total_count: runs.length, workflow_runs: runs }; } +function oldReceipt( + overrides: Partial = {}, +): DispatchNotObservedReceipt { + return { + correlationId: CORRELATION_ID, + workflowSha: WORKFLOW_SHA, + sentAtMs: SENT_AT_MS, + deadlineAtMs: SENT_AT_MS + WINDOW_MS, + result: "not-observed", + failureKind: "transport", + ...overrides, + }; +} + +function oldReceiptOptions(overrides: Partial = {}) { + return { + repository: REPOSITORY, + token: "token", + prNumber: 42, + receipt: oldReceipt(overrides), + }; +} + +function paginatedInventoryApi(runs: unknown[]) { + return vi.fn(async (apiPath: string) => { + const page = Number(new URL(`https://api.github.com/${apiPath}`).searchParams.get("page")); + return { + total_count: runs.length, + workflow_runs: runs.slice((page - 1) * 100, page * 100), + }; + }); +} + function reconciliationDeps( list: () => unknown | Promise, getRun: (runId: number) => unknown | Promise = (runId) => workflowRun(runId), @@ -366,24 +407,98 @@ describe("PR E2E workflow dispatch reconciliation", () => { const { deps, api } = reconciliationDeps(() => inventory([])); await expect( - assertDispatchStillNotObserved( - { - repository: REPOSITORY, - token: "token", - prNumber: 42, - receipt: { - correlationId: CORRELATION_ID, - workflowSha: WORKFLOW_SHA, - sentAtMs: SENT_AT_MS, - deadlineAtMs: SENT_AT_MS + WINDOW_MS, - result: "not-observed", - failureKind: "http", - status: 500, - }, - }, - deps, - ), + assertDispatchStillNotObserved(oldReceiptOptions({ failureKind: "http", status: 500 }), deps), + ).resolves.toBeUndefined(); + expect(api).toHaveBeenCalledOnce(); + }); + + it("paginates a stale zero-match receipt before allowing replacement", async () => { + const runs = Array.from({ length: 101 }, (_value, index) => unrelatedWorkflowRun(index + 100)); + const api = paginatedInventoryApi(runs); + + await expect( + assertDispatchStillNotObserved(oldReceiptOptions(), { + api, + now: () => SENT_AT_MS + WINDOW_MS + 2_000, + clockSkewMs: 10, + }), ).resolves.toBeUndefined(); + expect(api).toHaveBeenCalledTimes(2); + }); + + it("blocks replacement when the old correlation appears after the first inventory page", async () => { + const recheckTime = SENT_AT_MS + WINDOW_MS + 2_000; + const runs = [ + ...Array.from({ length: 100 }, (_value, index) => unrelatedWorkflowRun(index + 100)), + workflowRun(501, { + created_at: new Date(SENT_AT_MS + WINDOW_MS + 1_000).toISOString(), + }), + ]; + const api = paginatedInventoryApi(runs); + + await expect( + assertDispatchStillNotObserved(oldReceiptOptions(), { + api, + now: () => recheckTime, + clockSkewMs: 10, + }), + ).rejects.toMatchObject({ + name: "DispatchReconciliationError", + candidateRunIds: [501], + }); + expect(api).toHaveBeenCalledTimes(2); + }); + + it.each([ + { + label: "the total count changes", + secondPage: { + total_count: 102, + workflow_runs: [unrelatedWorkflowRun(500)], + }, + }, + { + label: "a run ID is duplicated", + secondPage: { + total_count: 101, + workflow_runs: [unrelatedWorkflowRun(100)], + }, + }, + ])("fails closed when paginated receipt inventory $label", async ({ secondPage }) => { + const firstPageRuns = Array.from({ length: 100 }, (_value, index) => + unrelatedWorkflowRun(index + 100), + ); + const pages = [{ total_count: 101, workflow_runs: firstPageRuns }, secondPage]; + const api = vi.fn(async (apiPath: string) => { + const page = Number(new URL(`https://api.github.com/${apiPath}`).searchParams.get("page")); + return pages[page - 1]; + }); + + await expect( + assertDispatchStillNotObserved(oldReceiptOptions(), { + api, + now: () => SENT_AT_MS + WINDOW_MS + 2_000, + clockSkewMs: 10, + }), + ).rejects.toThrow(/could not be rechecked/u); + expect(api).toHaveBeenCalledTimes(2); + }); + + it("fails closed when a stale receipt inventory exceeds GitHub's filtered-result cap", async () => { + const api = vi.fn().mockResolvedValue({ + total_count: 1_001, + workflow_runs: Array.from({ length: 100 }, (_value, index) => + unrelatedWorkflowRun(index + 100), + ), + }); + + await expect( + assertDispatchStillNotObserved(oldReceiptOptions(), { + api, + now: () => SENT_AT_MS + WINDOW_MS + 2_000, + clockSkewMs: 10, + }), + ).rejects.toThrow(/could not be rechecked/u); expect(api).toHaveBeenCalledOnce(); }); @@ -395,22 +510,7 @@ describe("PR E2E workflow dispatch reconciliation", () => { const { deps, api } = reconciliationDeps(() => inventory([lateRun])); await expect( - assertDispatchStillNotObserved( - { - repository: REPOSITORY, - token: "token", - prNumber: 42, - receipt: { - correlationId: CORRELATION_ID, - workflowSha: WORKFLOW_SHA, - sentAtMs: SENT_AT_MS, - deadlineAtMs: SENT_AT_MS + WINDOW_MS, - result: "not-observed", - failureKind: "transport", - }, - }, - { ...deps, now: () => recheckTime }, - ), + assertDispatchStillNotObserved(oldReceiptOptions(), { ...deps, now: () => recheckTime }), ).rejects.toMatchObject({ name: "DispatchReconciliationError", candidateRunIds: [23], diff --git a/tools/advisors/github.mts b/tools/advisors/github.mts index 77c04e066de..fec789eabca 100644 --- a/tools/advisors/github.mts +++ b/tools/advisors/github.mts @@ -25,6 +25,10 @@ export type GitHubApiFailureKind = "http" | "decode"; const MAX_GITHUB_RESPONSE_EXCERPT_CHARS = 512; const GITHUB_REQUEST_ID_PATTERN = /^[A-Za-z0-9:-]{1,128}$/u; +export function isValidGithubRequestId(value: unknown): value is string { + return typeof value === "string" && GITHUB_REQUEST_ID_PATTERN.test(value); +} + function responseExcerpt(text: string): string { const singleLine = text .replace(/[\r\n\t]+/gu, " ") @@ -41,7 +45,7 @@ function responseRequestId(response: Response): string | undefined { headers && typeof headers.get === "function" ? headers.get("x-github-request-id")?.trim() : undefined; - return requestId && GITHUB_REQUEST_ID_PATTERN.test(requestId) ? requestId : undefined; + return isValidGithubRequestId(requestId) ? requestId : undefined; } export class GitHubApiError extends Error { diff --git a/tools/e2e/pr-e2e-dispatch-reconciliation.mts b/tools/e2e/pr-e2e-dispatch-reconciliation.mts index 0a7884cdf3b..63cc0c76866 100644 --- a/tools/e2e/pr-e2e-dispatch-reconciliation.mts +++ b/tools/e2e/pr-e2e-dispatch-reconciliation.mts @@ -11,6 +11,7 @@ import { type DispatchFailureKind, type DispatchNotObservedReceipt, dispatchNotObservedReceiptMarker, + MAX_DISPATCH_RECONCILIATION_WINDOW_MS, } from "./pr-e2e-retry-receipt.mts"; const E2E_WORKFLOW = "e2e.yaml"; @@ -21,11 +22,11 @@ const DEFAULT_POLL_INTERVAL_MS = 2_000; const DEFAULT_CLOCK_SKEW_MS = 10_000; const DEFAULT_DISPATCH_TIMEOUT_MS = 10_000; const DEFAULT_API_TIMEOUT_MS = 5_000; -const MAX_RECONCILIATION_WINDOW_MS = 120_000; const MAX_DISPATCH_TIMEOUT_MS = 30_000; const MAX_API_TIMEOUT_MS = 10_000; const MAX_DISPATCH_AND_RECONCILIATION_BUDGET_MS = 65_000; const MAX_WORKFLOW_RUNS = 100; +const MAX_WORKFLOW_RUN_PAGES = 10; const SHA_PATTERN = /^[a-f0-9]{40}$/u; const CORRELATION_PATTERN = /^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/u; @@ -184,7 +185,7 @@ function assertTimingOptions(options: { if ( !positiveSafeInteger(options.sentAtMs) || !positiveSafeInteger(options.reconciliationWindowMs) || - options.reconciliationWindowMs > MAX_RECONCILIATION_WINDOW_MS || + options.reconciliationWindowMs > MAX_DISPATCH_RECONCILIATION_WINDOW_MS || !positiveSafeInteger(options.pollIntervalMs) || options.pollIntervalMs > options.reconciliationWindowMs || !Number.isSafeInteger(options.clockSkewMs) || @@ -265,14 +266,14 @@ function validateWorkflowRun(value: unknown): WorkflowRun { }; } -function validateWorkflowRunInventory(value: unknown): WorkflowRunInventory { +function validateWorkflowRunInventoryPage(value: unknown): WorkflowRunInventory { if ( !isObjectRecord(value) || !Number.isSafeInteger(value.total_count) || (value.total_count as number) < 0 || !Array.isArray(value.workflow_runs) || value.workflow_runs.length > MAX_WORKFLOW_RUNS || - value.workflow_runs.length !== value.total_count + value.workflow_runs.length > (value.total_count as number) ) { throw new Error("GitHub returned an invalid or incomplete workflow run listing"); } @@ -283,11 +284,20 @@ function validateWorkflowRunInventory(value: unknown): WorkflowRunInventory { return { totalCount: value.total_count as number, runs }; } +function validateWorkflowRunInventory(value: unknown): WorkflowRunInventory { + const inventory = validateWorkflowRunInventoryPage(value); + if (inventory.runs.length !== inventory.totalCount) { + throw new Error("GitHub returned an invalid or incomplete workflow run listing"); + } + return inventory; +} + function inventoryApiPath(options: { repository: string; workflowSha: string; lowerBoundMs: number; upperBoundMs: number; + page?: number; }): string { const created = `${new Date(options.lowerBoundMs).toISOString()}..${new Date(options.upperBoundMs).toISOString()}`; const query = new URLSearchParams({ @@ -296,10 +306,56 @@ function inventoryApiPath(options: { head_sha: options.workflowSha, created, per_page: String(MAX_WORKFLOW_RUNS), + ...(options.page === undefined ? {} : { page: String(options.page) }), }); return `repos/${options.repository}/actions/workflows/${E2E_WORKFLOW}/runs?${query}`; } +async function readPaginatedInventory( + options: { + repository: string; + token: string; + workflowSha: string; + lowerBoundMs: number; + upperBoundMs: number; + }, + api: GithubApi, + timeoutMs: number, +): Promise { + return boundedOperation("Paginated workflow run inventory read", timeoutMs, async (signal) => { + const runs: WorkflowRun[] = []; + const runIds = new Set(); + let totalCount: number | undefined; + for (let page = 1; page <= MAX_WORKFLOW_RUN_PAGES; page += 1) { + const inventory = validateWorkflowRunInventoryPage( + await api(inventoryApiPath({ ...options, page }), options.token, { + userAgent: USER_AGENT, + signal, + }), + ); + totalCount ??= inventory.totalCount; + if ( + inventory.totalCount !== totalCount || + totalCount > MAX_WORKFLOW_RUNS * MAX_WORKFLOW_RUN_PAGES + ) { + throw new Error("GitHub returned an unstable or oversized workflow run listing"); + } + for (const run of inventory.runs) { + if (runIds.has(run.id)) { + throw new Error("GitHub returned duplicate workflow run IDs across pages"); + } + runIds.add(run.id); + runs.push(run); + } + if (runs.length === totalCount) return { totalCount, runs }; + if (runs.length > totalCount || inventory.runs.length < MAX_WORKFLOW_RUNS) { + throw new Error("GitHub returned an invalid or incomplete workflow run listing"); + } + } + throw new Error("GitHub returned an incomplete workflow run listing after pagination"); + }); +} + async function readInventory( options: { repository: string; @@ -704,7 +760,7 @@ export async function assertDispatchStillNotObserved( const correlatedRunIds = new Set(); let inventory: WorkflowRunInventory; try { - inventory = await readInventory( + inventory = await readPaginatedInventory( { repository: options.repository, token: options.token, diff --git a/tools/e2e/pr-e2e-gate.mts b/tools/e2e/pr-e2e-gate.mts index 7b8de9eb2b0..68acd2f1683 100755 --- a/tools/e2e/pr-e2e-gate.mts +++ b/tools/e2e/pr-e2e-gate.mts @@ -74,6 +74,9 @@ const RESERVED_CHECK_SUMMARY = "This PR SHA and base SHA are reserved for deterministic E2E planning after CI completes."; const CONTROL_PLANE_AUTHORIZATION_TITLE = "E2E reviewer authorization required to run E2E"; const FORK_E2E_AUTHORIZATION_TITLE = "E2E reviewer authorization required to run fork E2E"; +const EVALUATING_PR_COMMIT_TITLE = "Evaluating PR commit"; +const RUNNER_LOSS_RETRY_PREPARATION_TITLE = "Preparing one-time hosted-runner-loss retry"; +const AUTHORIZED_EXECUTION_TITLE_PREFIX = "E2E execution authorized by @"; const PRE_DISPATCH_CHECK_READ_TIMEOUT_MS = 5_000; const RECONCILED_CHILD_VALIDATION_TIMEOUT_MS = 10_000; const CHILD_AUTHORIZATION_PUBLISH_TIMEOUT_MS = 5_000; @@ -2187,15 +2190,18 @@ async function requireUnchangedCompletedWorkflowRun( } function isValidExpectedPreDispatchTitle(title: string): boolean { - const authorizationPrefix = "E2E execution authorized by @"; return ( - title === "Evaluating PR commit" || - title === "Preparing one-time hosted-runner-loss retry" || - (title.startsWith(authorizationPrefix) && - MAINTAINER_PATTERN.test(title.slice(authorizationPrefix.length))) + title === EVALUATING_PR_COMMIT_TITLE || + title === RUNNER_LOSS_RETRY_PREPARATION_TITLE || + (title.startsWith(AUTHORIZED_EXECUTION_TITLE_PREFIX) && + MAINTAINER_PATTERN.test(title.slice(AUTHORIZED_EXECUTION_TITLE_PREFIX.length))) ); } +function authorizedExecutionTitle(maintainer: string): string { + return `${AUTHORIZED_EXECUTION_TITLE_PREFIX}${maintainer}`; +} + function assertCurrentPreDispatchCheck( history: readonly CheckRun[], options: { repository: string; controllerCheckId: number; expectedCheckTitle: string }, @@ -2745,7 +2751,7 @@ async function dispatchRunnerLossRetry(options: { workflowSha: options.state.workflowSha, planHash: options.state.planHash, correlationId, - expectedCheckTitle: "Preparing one-time hosted-runner-loss retry", + expectedCheckTitle: RUNNER_LOSS_RETRY_PREPARATION_TITLE, }); const childRunId = dispatch.runId; try { @@ -2852,7 +2858,7 @@ export async function retryRunnerLossPrGate( baseSha: state.baseSha, }, token, - "Preparing one-time hosted-runner-loss retry", + RUNNER_LOSS_RETRY_PREPARATION_TITLE, `Revalidating the exact PR/base SHA and risk plan after [attempt 1](${originalRunUrl}) lost its GitHub-hosted runner.`, ); @@ -3116,7 +3122,7 @@ export async function startPrGate( baseSha: ciIdentity.baseSha, }, token, - "Evaluating PR commit", + EVALUATING_PR_COMMIT_TITLE, "Validating the PR SHA and selecting deterministic E2E jobs and typed targets.", ); @@ -3286,7 +3292,7 @@ export async function startPrGate( workflowSha: command.workflowSha, plan, checkRunId, - expectedCheckTitle: "Evaluating PR commit", + expectedCheckTitle: EVALUATING_PR_COMMIT_TITLE, paths: command, }); } catch (error) { @@ -3322,6 +3328,7 @@ async function startAuthorizedPrGate( throw new Error("E2E authorization must use the first workflow run attempt"); } const reason = normalizedWaiverReason(command.reason); + const executionTitle = authorizedExecutionTitle(command.maintainer); const pendingTitle = authorizationKind === "fork" ? FORK_E2E_AUTHORIZATION_TITLE : CONTROL_PLANE_AUTHORIZATION_TITLE; @@ -3410,7 +3417,7 @@ async function startAuthorizedPrGate( baseSha: command.baseSha, }, token, - `E2E execution authorized by @${command.maintainer}`, + executionTitle, `Running the exact reviewed head and base revision. Review reason: ${reason.replace(/`/gu, "'")}`, ); await dispatchSelectedPrGate({ @@ -3421,7 +3428,7 @@ async function startAuthorizedPrGate( workflowSha: command.workflowSha, plan, checkRunId, - expectedCheckTitle: `E2E execution authorized by @${command.maintainer}`, + expectedCheckTitle: executionTitle, paths: command, }); } catch (error) { diff --git a/tools/e2e/pr-e2e-retry-receipt.mts b/tools/e2e/pr-e2e-retry-receipt.mts index b993e00d894..97242da84b8 100644 --- a/tools/e2e/pr-e2e-retry-receipt.mts +++ b/tools/e2e/pr-e2e-retry-receipt.mts @@ -1,16 +1,17 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +import { isValidGithubRequestId } from "../advisors/github.mts"; + const RETRYABLE_FAILURE_MARKER_PREFIX = ""; const DISPATCH_RECEIPT_MARKER_PREFIX = ""; const MAX_DISPATCH_RECEIPT_BYTES = 1024; -const MAX_DISPATCH_RECONCILIATION_WINDOW_MS = 120_000; +export const MAX_DISPATCH_RECONCILIATION_WINDOW_MS = 120_000; const SHA_PATTERN = /^[a-f0-9]{40}$/u; const CORRELATION_PATTERN = /^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/u; -const GITHUB_REQUEST_ID_PATTERN = /^[A-Za-z0-9:-]{1,128}$/u; const RETRYABLE_FAILURE_REASONS = new Set([ "prerequisite-ci", "child-cancelled", @@ -101,8 +102,7 @@ function validateDispatchReceipt(value: unknown): DispatchNotObservedReceipt { (!Number.isSafeInteger(receipt.status) || (receipt.status as number) < 100 || (receipt.status as number) > 599)) || - (receipt.requestId !== undefined && - (typeof receipt.requestId !== "string" || !GITHUB_REQUEST_ID_PATTERN.test(receipt.requestId))) + (receipt.requestId !== undefined && !isValidGithubRequestId(receipt.requestId)) ) { throw new Error("dispatch receipt fields are invalid"); }