diff --git a/.github/workflows/reconcile-merge-queue.yml b/.github/workflows/reconcile-merge-queue.yml index 9cb90c5..ee16c79 100644 --- a/.github/workflows/reconcile-merge-queue.yml +++ b/.github/workflows/reconcile-merge-queue.yml @@ -23,7 +23,7 @@ jobs: github.event_name != 'workflow_run' || contains(fromJSON('["pull_request", "merge_group"]'), github.event.workflow_run.event) runs-on: ubuntu-24.04 - timeout-minutes: 5 + timeout-minutes: 15 permissions: checks: read # Read validation evidence for the queued commit. actions: write # Dispatch trusted queue validation. diff --git a/actions/queue-completion/action.yml b/actions/queue-completion/action.yml index c4efb65..2d62eba 100644 --- a/actions/queue-completion/action.yml +++ b/actions/queue-completion/action.yml @@ -15,15 +15,15 @@ runs: DEFAULT_BRANCH: ${{ github.event.repository.default_branch }} SOURCE_ACTIVITY: ${{ inputs.source-activity }} run: | - args=(-f source_run_id="$GITHUB_RUN_ID") + args=(-f ref="$DEFAULT_BRANCH" -f "inputs[source_run_id]=$GITHUB_RUN_ID") case "$SOURCE_ACTIVITY" in '') ;; in_progress|completed) - args+=(-f source_run_attempt="$GITHUB_RUN_ATTEMPT" -f source_activity="$SOURCE_ACTIVITY") ;; + args+=(-f "inputs[source_run_attempt]=$GITHUB_RUN_ATTEMPT" -f "inputs[source_activity]=$SOURCE_ACTIVITY") ;; *) echo '::error::Invalid package-source notification phase'; exit 1 ;; esac for attempt in 1 2 3; do - if gh workflow run reconcile-merge-queue.yml --ref "$DEFAULT_BRANCH" "${args[@]}"; then + if gh api "repos/$GH_REPO/actions/workflows/reconcile-merge-queue.yml/dispatches" --method POST "${args[@]}"; then exit 0 fi echo "::warning::Queue completion notification attempt $attempt failed" diff --git a/docs/releases/v2.13.0.md b/docs/releases/v2.13.0.md new file mode 100644 index 0000000..cd57e15 --- /dev/null +++ b/docs/releases/v2.13.0.md @@ -0,0 +1,53 @@ +# Nix Forge CI v2.13.0 + +## Changelog + +- Reconcile all eligible package-source heads on a completion notification. + This recovers ready heads when workflow concurrency replaces a pending callback. +- Read current PR identity, source refs and latest statuses in a GraphQL batch. + Use a bounded native check-suite projection to defer healthy partial producers + and recognize unchanged already-admitted results without writes. +- Preserve full immutable source, workflow, current-run and actual-job proofs + for every new result or admission. Real failures retain their actual job URLs + when another producer is running or absent. Changed attempts invalidate old + source results. Truncated, missing and approval-only projections take the + complete validation path. +- Replace callback workflow-name lookup with one direct dispatch request. + Preserve legacy queue callback inputs and bounded visible notification retries. +- Pace opted-in package reconciliation content writes four seconds apart, + including status, enqueue and validation dispatch requests. Package-source + callers must allow 15 minutes in their own reconciler jobs for a paced + six-head sweep. + +A finite six-package fixture executes the actual reconciler for source +notifications and all 30 queue completion callbacks. It preserves all 19 queue +results and includes notification and candidate-dispatch requests. That fixture +uses 831 REST requests. A separate 99-request source-inspection allowance brings +the normal automation-layer accounting model to 930 requests. Live read-only +GraphQL queries for the PR snapshot and six distinct native source heads each +cost one point. + +GitHub's standard repository token allows +[1,000 REST requests per hour](https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api). +Updater publication, native validation tools, repeated reruns and delayed +producer completion consume additional requests. Completion callbacks retain +their bounded wait when the producer is still running. The six native queue +candidates execute serially, so these measured calls need not share one hour. +These counts do not guarantee capacity for arbitrary bursts. The production +rollout must verify the current single-package update through native validation +and the protected queue. + +Existing source start notifications remain asynchronous. They do not acknowledge +that the reconciler has completed its status writes. Source completion +notifications remain blocking after all actual validation jobs. + +No token permissions or required native job gates change. The independently +reviewed release-builder and repository-checks pins remain separate. Consumers +without package-source reconciliation retain their existing callback contract. + +The 192 repository regression tests passed locally. Required hosted validation +and the protected normal merge queue must pass before publication. The signed +tag and verified release manifest identify the protected source commit. +Publication verifies the archive checksum and all release provenance subjects +against the separately pinned SLSA builder. Restore the previous reviewed v2.x +pin to roll back. Compatible v2.x consumers have no scheduled support end date. diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py index e289874..4b66c52 100644 --- a/scripts/reconcile-queue.py +++ b/scripts/reconcile-queue.py @@ -34,6 +34,13 @@ SOURCE_ACTIVITY = os.environ.get("SOURCE_ACTIVITY", "") SOURCE_CACHE: dict[str, Any] = {} SOURCE_EVIDENCE: dict[str, dict[str, Any]] = {} +SOURCE_CALLBACK = False +SOURCE_SWEEP = False +SOURCE_VIEW: dict[str, Any] = {} +SOURCE_RUN_LISTS: dict[str, Any] = {} +SOURCE_PREFLIGHT: dict[str, Any] = {} +SOURCE_LAST_WRITE: float | None = None +SOURCE_PRODUCER_COMPLETE = False PAGE_SIZE = 100 GH = shutil.which("gh") or "/usr/bin/gh" @@ -45,6 +52,20 @@ def api(path: str, method: str = "GET", data: object = None) -> Any: # ruff: ig The decoded response, or None for an empty response. """ + global SOURCE_LAST_WRITE + writes = method != "GET" and ( + path != "graphql" + or ( + isinstance(data, dict) and str(data.get("query", "")).startswith("mutation") + ) + ) + if (SOURCE_CALLBACK or SOURCE_CHECKS) and writes: + # A six-source sweep publishes 102 statuses. Pace all content writes, + # including enqueue/dispatch, leaving room for 60 notifier requests. + now = time.monotonic() + if SOURCE_LAST_WRITE is not None: + time.sleep(max(0.0, 4.0 - (now - SOURCE_LAST_WRITE))) + SOURCE_LAST_WRITE = time.monotonic() command = [GH, "api", path, "--method", method] if data is not None: command += ["--input", "-"] @@ -58,7 +79,7 @@ def api(path: str, method: str = "GET", data: object = None) -> Any: # ruff: ig return json.loads(result.stdout) if result.stdout.strip() else None -def graphql(query: str, **variables: str) -> dict[str, Any]: +def graphql(query: str, **variables: str | int) -> dict[str, Any]: """Send a parameterized GraphQL operation and reject API errors. Returns: @@ -86,11 +107,17 @@ def admit(pr: dict[str, Any]) -> bool: source_statuses = None if source: sha = pr["head"]["sha"] - if SOURCE_HEAD and SOURCE_HEAD != sha: + if SOURCE_HEAD and not SOURCE_SWEEP and SOURCE_HEAD != sha: return False - source_statuses = read_statuses(sha) - if not SOURCE_HEAD and any( - status["context"] == ADMISSION for status in source_statuses.values() + source_statuses = SOURCE_VIEW.get(sha) + if source_statuses is None: + source_statuses = read_statuses(sha) + if ( + not SOURCE_HEAD + and not SOURCE_SWEEP + and any( + status["context"] == ADMISSION for status in source_statuses.values() + ) ): state = graphql( "query($id:ID!) { node(id:$id) { ... on PullRequest { mergeQueueEntry { id } } } }", @@ -98,28 +125,30 @@ def admit(pr: dict[str, Any]) -> bool: )["node"] return bool(state["mergeQueueEntry"]) if not reconcile_source(pr, source_statuses): - if SOURCE_HEAD == sha and any( - status["context"] == ADMISSION and status.get("state") == "success" - for status in source_statuses.values() + if ( + (SOURCE_HEAD == sha or SOURCE_SWEEP) + and any( + status["context"] == ADMISSION and status.get("state") == "success" + for status in source_statuses.values() + ) + and source_lease(pr) ): - fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}") - current = api( - f"repos/{REPOSITORY}/git/ref/heads/{quote(pr['head']['ref'], safe='/')}" + api( + f"repos/{REPOSITORY}/statuses/{sha}", + "POST", + { + "state": "pending", + "context": ADMISSION, + "description": "Source validation changed; current successful jobs required", + }, ) - if ( - source_identity(fresh) - and fresh["head"]["sha"] == sha - and current["object"]["sha"] == sha - ): - api( - f"repos/{REPOSITORY}/statuses/{sha}", - "POST", - { - "state": "pending", - "context": ADMISSION, - "description": "Source validation changed; current successful jobs required", - }, - ) + return False + if ( + SOURCE_CALLBACK + and source_statuses.get(ADMISSION, {}).get("state") == "success" + ): + # A current already-admitted source needs no queue scan or write. + # Its queue validation completion callbacks own queue progression. return False trusted = ( source @@ -162,13 +191,17 @@ def admit(pr: dict[str, Any]) -> bool: if state["mergeQueueEntry"]: if ( source - and SOURCE_HEAD == sha + and (SOURCE_HEAD == sha or SOURCE_SWEEP) and any( status["context"] == ADMISSION and status.get("state") != "success" for status in (source_statuses or {}).values() ) ): - if not source_runs_current(pr) or source_contract(pr) is None: + if ( + not source_runs_current(pr) + or source_contract(pr) is None + or not source_lease(pr) + ): reconcile_source(pr, allow_success=False) return False api( @@ -449,11 +482,14 @@ def source_notification( ObsoleteSourceNotification: A verified attempt or branch has advanced or vanished. """ + global SOURCE_PRODUCER_COMPLETE + SOURCE_PRODUCER_COMPLETE = False if not run_id or not SOURCE_CHECKS: return None if not run_id.isascii() or not run_id.isdecimal() or int(run_id) <= 0: raise ValueError("Source run ID must be a positive ASCII integer") run = api(f"repos/{REPOSITORY}/actions/runs/{run_id}") + SOURCE_PRODUCER_COMPLETE = run.get("status") == "completed" branch = run.get("head_branch", "") if not isinstance(branch, str) or not branch.startswith( "automation/package-updates-" @@ -518,7 +554,7 @@ def valid(producer: dict[str, Any]) -> bool: "Older source callback does not match its immutable producer" ) raise ObsoleteSourceNotification("Producer has a newer validated attempt") - if activity == "completed": + if activity == "completed" and not SOURCE_PRODUCER_COMPLETE: wait_for_source_run(run_id) run = api(f"repos/{REPOSITORY}/actions/runs/{run_id}") if not valid(run) or run["head_sha"] != sha: @@ -531,15 +567,22 @@ def valid(producer: dict[str, Any]) -> bool: ) if run["run_attempt"] != int(attempt): raise ValueError("Source callback producer attempt moved backwards") - try: - ref = api(f"repos/{REPOSITORY}/git/ref/heads/{quote(branch, safe='/')}") - except subprocess.CalledProcessError as error: - if "HTTP 404" not in (error.stderr or ""): - raise - raise ObsoleteSourceNotification( - "Verified source producer branch no longer exists" - ) from error - current_sha = ref.get("object", {}).get("sha") + if SOURCE_CALLBACK: + current_sha = source_ref(branch) + if current_sha is None: + raise ObsoleteSourceNotification( + "Verified source producer branch no longer exists" + ) + else: + try: + ref = api(f"repos/{REPOSITORY}/git/ref/heads/{quote(branch, safe='/')}") + except subprocess.CalledProcessError as error: + if "HTTP 404" not in (error.stderr or ""): + raise + raise ObsoleteSourceNotification( + "Verified source producer branch no longer exists" + ) from error + current_sha = ref.get("object", {}).get("sha") if ( not isinstance(current_sha, str) or re.fullmatch(r"[0-9a-f]{40}", current_sha) is None @@ -571,6 +614,15 @@ def front_queue_entry() -> dict[str, Any] | None: return entries[0] if entries else None +def source_bot(actor: dict[str, Any]) -> bool: + """Require the canonical GitHub Actions bot identity, never login alone.""" + return ( + actor.get("id") == 41898282 + and actor.get("type") == "Bot" + and actor.get("login") == "github-actions[bot]" + ) + + def source_identity(pr: dict[str, Any]) -> bool: """Limit opt-in source reconciliation to the package updater's bot PRs. @@ -603,11 +655,278 @@ def source_read(path: str) -> Any: # ruff: ignore[any-type] The authoritative response; current refs and latest runs are never cached. """ + if SOURCE_CHECKS and "/actions/workflows/" in path and path not in SOURCE_CACHE: + inventory_path = f"repos/{REPOSITORY}/actions/workflows?per_page={PAGE_SIZE}" + if inventory_path not in SOURCE_CACHE: + inventory = api(inventory_path) + workflows = inventory.get("workflows") + count = inventory.get("total_count") + if ( + not isinstance(count, int) + or not isinstance(workflows, list) + or not 0 <= count < PAGE_SIZE + or count != len(workflows) + ): + raise ValueError("Workflow inventory is incomplete") + SOURCE_CACHE[inventory_path] = inventory + for item in workflows: + filename = item.get("path", "").removeprefix(".github/workflows/") + if filename in WORKFLOWS: + key = f"repos/{REPOSITORY}/actions/workflows/{filename}" + if key in SOURCE_CACHE and any( + SOURCE_CACHE[key].get(field) != item.get(field) + for field in ("id", "path", "state") + ): + raise ValueError("Workflow identity changed within callback") + SOURCE_CACHE[key] = item + if path not in SOURCE_CACHE: + raise ValueError("Required source workflow is absent") if path not in SOURCE_CACHE: SOURCE_CACHE[path] = api(path) return SOURCE_CACHE[path] +def source_snapshot( + number: int | None = None, *, sources_only: bool = True +) -> list[dict[str, Any]]: + """Read source PR identity, live ref and latest statuses in one fresh view.""" + owner, name = REPOSITORY.split("/", 1) + fields = ( + "number id isDraft baseRefName headRefName headRefOid " + "author { __typename login ... on Bot { databaseId } ... on User { databaseId } } " + "headRepository { nameWithOwner } " + "headRef { name repository { nameWithOwner } target { oid ... on Commit { " + "status { contexts { context state targetUrl description " + "creator { __typename login ... on Bot { databaseId } ... on User { databaseId } } } } } } }" + ) + if number is None: + selection = ( + "pullRequests(first:100,states:OPEN,baseRefName:$branch) { " + "pageInfo { hasNextPage } nodes { " + fields + " } }" + ) + result = graphql( + "query($owner:String!,$name:String!,$branch:String!) { " + "repository(owner:$owner,name:$name) { " + selection + " } }", + owner=owner, + name=name, + branch=DEFAULT_BRANCH, + )["repository"]["pullRequests"] + if result["pageInfo"]["hasNextPage"]: + raise ValueError("Source snapshot exceeds bounded PR inventory") + nodes = result["nodes"] + else: + result = graphql( + "query($owner:String!,$name:String!,$number:Int!) { " + "repository(owner:$owner,name:$name) { pullRequest(number:$number) { " + "state " + fields + " } } }", + owner=owner, + name=name, + number=number, + )["repository"]["pullRequest"] + nodes = [result] if result and result["state"] == "OPEN" else [] + pulls = [] + for node in nodes: + actor = node.get("author") or {} + eligible_source = ( + actor.get("__typename") == "Bot" + and actor.get("databaseId") == 41898282 + and actor.get("login") == "github-actions" + and (node.get("headRepository") or {}).get("nameWithOwner") == REPOSITORY + and node.get("baseRefName") == DEFAULT_BRANCH + and re.fullmatch( + r"automation/package-updates-[a-z0-9][a-z0-9+._-]*", + node.get("headRefName", ""), + ) + is not None + and not node.get("isDraft", True) + ) + if sources_only and not eligible_source: + continue + ref = node.get("headRef") + if not ref: + continue + target = ref["target"] + if ( + eligible_source + or (node.get("headRepository") or {}).get("nameWithOwner") == REPOSITORY + ) and ( + ref["name"] != node["headRefName"] + or ref["repository"]["nameWithOwner"] != REPOSITORY + or target["oid"] != node["headRefOid"] + ): + raise ValueError("Inconsistent live source ref snapshot") + + def bot(value: dict[str, Any]) -> dict[str, Any]: + login = value.get("login") + if value.get("__typename") == "Bot" and ( + (value.get("databaseId"), login) + in {(41898282, "github-actions"), (49699333, "dependabot")} + ): + login += "[bot]" + return { + "type": value.get("__typename"), + "id": value.get("databaseId"), + "login": login, + } + + statuses = {} + for status in (target.get("status") or {}).get("contexts", []): + context = status["context"] + if context in statuses: + raise ValueError("Duplicate source status context") + statuses[context] = { + "context": context, + "state": status["state"].lower(), + "target_url": status["targetUrl"], + "description": status["description"], + "creator": bot(status.get("creator") or {}), + } + pr = { + "number": node["number"], + "node_id": node["id"], + "state": "open", + "draft": node["isDraft"], + "user": bot(actor), + "head": { + "sha": node["headRefOid"], + "ref": node["headRefName"], + "repo": { + "full_name": (node.get("headRepository") or {}).get("nameWithOwner") + }, + }, + "base": {"ref": node["baseRefName"], "repo": {"full_name": REPOSITORY}}, + } + SOURCE_VIEW[target["oid"]] = statuses + pulls.append(pr) + return pulls + + +def source_ref(branch: str) -> str | None: + """Read the producer's exact current branch without a REST request.""" + owner, name = REPOSITORY.split("/", 1) + ref = graphql( + "query($owner:String!,$name:String!,$branch:String!) { " + "repository(owner:$owner,name:$name) { ref(qualifiedName:$branch) { " + "name repository { nameWithOwner } target { oid } } } }", + owner=owner, + name=name, + branch=f"refs/heads/{branch}", + )["repository"]["ref"] + if ref is None: + return None + if ref["name"] != branch or ref["repository"]["nameWithOwner"] != REPOSITORY: + raise ValueError("Producer ref identity does not match") + return ref["target"]["oid"] + + +def source_lease(pr: dict[str, Any]) -> bool: + """Recheck the exact live identity before an immutable-SHA write batch.""" + if SOURCE_CALLBACK: + fresh = source_snapshot(pr["number"]) + return ( + len(fresh) == 1 + and source_identity(fresh[0]) + and fresh[0]["head"] == pr["head"] + ) + fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}") + if not source_identity(fresh) or fresh["head"]["sha"] != pr["head"]["sha"]: + return False + ref = quote(pr["head"]["ref"], safe="/") + return ( + api(f"repos/{REPOSITORY}/git/ref/heads/{ref}")["object"]["sha"] + == pr["head"]["sha"] + ) + + +def source_run_inventory(pr: dict[str, Any]) -> None: + """Refresh the bounded exact-head run inventory; never persist latest state.""" + sha = pr["head"]["sha"] + result = api(f"repos/{REPOSITORY}/actions/runs?head_sha={sha}&per_page={PAGE_SIZE}") + runs = result.get("workflow_runs") + count = result.get("total_count") + if ( + not isinstance(count, int) + or not isinstance(runs, list) + or not 0 <= count < PAGE_SIZE + or count != len(runs) + ): + raise ValueError("Source run inventory is incomplete") + SOURCE_RUN_LISTS[sha] = result + + +def source_readiness(pulls: list[dict[str, Any]]) -> None: + """Batch fresh bounded check-suite projections for read-only readiness. + + This projection never authorizes result publication or a new admission. + Missing, truncated or approval-required evidence takes the full REST path. + """ + sources = [pr for pr in pulls if source_identity(pr)] + if not sources: + return + owner, name = REPOSITORY.split("/", 1) + fields = ( + "oid checkSuites(first:100) { pageInfo { hasNextPage } nodes { " + "status conclusion repository { nameWithOwner } commit { oid } " + "branch { name repository { nameWithOwner } } workflowRun { " + "databaseId runAttempt event file { path } workflow { databaseId state } } } }" + ) + selections = " ".join( + f'head{index}:object(expression:"{pr["head"]["sha"]}") {{ ... on Commit {{ {fields} }} }}' + for index, pr in enumerate(sources) + ) + result = graphql( + "query($owner:String!,$name:String!) { repository(owner:$owner,name:$name) { " + + selections + + " } }", + owner=owner, + name=name, + )["repository"] + for index, pr in enumerate(sources): + sha = pr["head"]["sha"] + item = result.get(f"head{index}") + if not item or item.get("oid") != sha: + continue + suites = item.get("checkSuites") or {} + if suites.get("pageInfo", {}).get("hasNextPage") is not False: + continue + selected = {} + for suite in suites.get("nodes", []): + run = suite.get("workflowRun") or {} + workflow = run.get("file") or {} + filename = workflow.get("path", "").removeprefix(".github/workflows/") + if filename not in WORKFLOWS: + continue + metadata = source_read(f"repos/{REPOSITORY}/actions/workflows/{filename}") + branch = suite.get("branch") or {} + if not ( + suite.get("repository", {}).get("nameWithOwner") == REPOSITORY + and suite.get("commit", {}).get("oid") == sha + and branch.get("name") == pr["head"]["ref"] + and branch.get("repository", {}).get("nameWithOwner") == REPOSITORY + and run.get("workflow", {}).get("databaseId") == metadata["id"] + and run["workflow"].get("state") == "ACTIVE" + and workflow.get("path") == f".github/workflows/{filename}" + and run.get("event") in {"pull_request", "workflow_dispatch"} + and isinstance(run.get("databaseId"), int) + and run["databaseId"] > 0 + and isinstance(run.get("runAttempt"), int) + and run["runAttempt"] > 0 + ): + continue + record = { + "id": run["databaseId"], + "run_attempt": run["runAttempt"], + "status": suite["status"].lower(), + "conclusion": (suite.get("conclusion") or "").lower(), + } + if filename not in selected or record["id"] > selected[filename]["id"]: + selected[filename] = record + if set(selected) == set(WORKFLOWS) and not any( + run["conclusion"] == "action_required" for run in selected.values() + ): + SOURCE_PREFLIGHT[sha] = selected + + def source_contract(pr: dict[str, Any]) -> str | None: """Prove the real source parent, owned paths and current workflow identities. @@ -623,9 +942,18 @@ def source_contract(pr: dict[str, Any]) -> str | None: base = parents[0]["sha"] if re.fullmatch(r"[0-9a-f]{40}", base) is None: return None - protected = api( - f"repos/{REPOSITORY}/git/ref/heads/{quote(DEFAULT_BRANCH, safe='')}" - )["object"]["sha"] + protected = ( + source_ref(DEFAULT_BRANCH) + if SOURCE_CALLBACK + else api(f"repos/{REPOSITORY}/git/ref/heads/{quote(DEFAULT_BRANCH, safe='')}")[ + "object" + ]["sha"] + ) + if ( + not isinstance(protected, str) + or re.fullmatch(r"[0-9a-f]{40}", protected) is None + ): + return None comparison = source_read(f"repos/{REPOSITORY}/compare/{base}...{protected}") if comparison.get("status") not in {"ahead", "identical"}: return None @@ -684,8 +1012,12 @@ def source_run( path = f".github/workflows/{workflow}" if metadata.get("path") != path or metadata.get("state") != "active": return None - response = api( - f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page={PAGE_SIZE}" + response = ( + SOURCE_RUN_LISTS[sha] + if SOURCE_CALLBACK + else api( + f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page={PAGE_SIZE}" + ) ) if ( not isinstance(response.get("total_count"), int) @@ -729,6 +1061,8 @@ def source_runs_current(pr: dict[str, Any]) -> bool: Whether every collected run still has its exact inspected identity/state. """ + if SOURCE_CALLBACK: + source_run_inventory(pr) for workflow, expected in SOURCE_EVIDENCE[pr["head"]["sha"]]["runs"].items(): selected = source_run(pr, workflow) if selected is None or any( @@ -753,14 +1087,17 @@ def source_jobs( SOURCE_EVIDENCE[sha] = evidence jobs: list[dict[str, Any]] = [] failures = {"failure", "timed_out", "cancelled"} + complete = True for workflow in WORKFLOWS: selected = source_run(pr, workflow) if selected is None: - return None + complete = False + continue run, batch = selected evidence["runs"][workflow] = run if run.get("status") != "completed": - return None + complete = False + continue previous = ( [statuses.get(context, {}) for context in SOURCE_CHECKS[workflow]] if statuses @@ -774,7 +1111,7 @@ def source_jobs( and all( status.get("state") == "success" and status.get("description") == description - and status.get("creator", {}).get("id") == 41898282 + and source_bot(status.get("creator", {})) and re.fullmatch( re.escape(prefix) + r"[0-9]+", status.get("target_url", "") ) @@ -829,9 +1166,10 @@ def source_jobs( if run.get("conclusion") != "success" or any( job.get("conclusion") != "success" for job in current_jobs ): - return None + complete = False + continue jobs.extend(current_jobs) - return jobs + return jobs if complete else None def reconcile_source( @@ -852,7 +1190,7 @@ def reconcile_source( or not source_identity(pr) ): return False - sha, ref = pr["head"]["sha"], f"refs/heads/{pr['head']['ref']}" + sha = pr["head"]["sha"] statuses = read_statuses(sha) if statuses is None else statuses SOURCE_EVIDENCE[sha] = {"runs": {}, "failures": []} initial_event = SOURCE_HEAD == sha and SOURCE_ACTIVITY in { @@ -864,6 +1202,73 @@ def reconcile_source( for status in statuses.values() ): return False + projected = SOURCE_PREFLIGHT.get(sha, {}) if SOURCE_CALLBACK else {} + if projected and not initial_event: + ready = all(run["status"] == "completed" for run in projected.values()) + failed_run = any( + run["conclusion"] in {"failure", "timed_out", "cancelled"} + for run in projected.values() + ) + if not ready and not failed_run: + initial_event = True + elif ( + ready + and statuses.get(ADMISSION, {}).get("state") == "success" + and allow_success + and all( + run["conclusion"] == "success" + and status.get("state") == "success" + and source_bot(status.get("creator", {})) + and status.get("description") + == f"{SOURCE_REPORT} result; attempt {run['run_attempt']}" + and re.fullmatch( + re.escape( + f"https://github.com/{REPOSITORY}/actions/runs/{run['id']}/job/" + ) + + r"[0-9]+", + status.get("target_url", ""), + ) + for workflow, run in projected.items() + for context in SOURCE_CHECKS[workflow] + for status in [statuses.get(context, {})] + ) + ): + return True + ready = True + if SOURCE_CALLBACK and not initial_event: + source_run_inventory(pr) + selected = [source_run(pr, workflow) for workflow in WORKFLOWS] + ready = all(item and item[0].get("status") == "completed" for item in selected) + failed_run = any( + item + and item[0].get("status") == "completed" + and item[0].get("conclusion") in {"failure", "timed_out", "cancelled"} + for item in selected + ) + if ready and statuses.get(ADMISSION, {}).get("state") == "success": + # A write-free duplicate callback can reuse only exact current-run + # attestations. Changed attempts always take the full contract path. + attested = all( + item[0].get("conclusion") == "success" + and status.get("state") == "success" + and source_bot(status.get("creator", {})) + and status.get("description") + == f"{SOURCE_REPORT} result; attempt {item[0]['run_attempt']}" + and re.fullmatch( + re.escape( + f"https://github.com/{REPOSITORY}/actions/runs/{item[0]['id']}/job/" + ) + + r"[0-9]+", + status.get("target_url", ""), + ) + for workflow, item in zip(WORKFLOWS, selected, strict=True) + for context in SOURCE_CHECKS[workflow] + for status in [statuses.get(context, {})] + ) + if attested and allow_success: + return True + if not ready and not failed_run: + initial_event = True jobs = ( source_jobs(pr, statuses) if not initial_event and source_contract(pr) is not None @@ -912,22 +1317,32 @@ def reconcile_source( "target_url": status["target_url"], "attempt": 0, } + changed = [ + result + for result in results.values() + if not source_bot(statuses.get(result["context"], {}).get("creator", {})) + or any( + statuses.get(result["context"], {}).get(key) != value + for key, value in { + "state": result["state"], + "target_url": result["target_url"], + "description": f"{SOURCE_REPORT} result; attempt {result['attempt']}", + }.items() + ) + ] + if changed and not source_lease(pr): + return False wrote = False for result in results.values(): previous = statuses.get(result["context"], {}) description = f"{SOURCE_REPORT} result; attempt {result['attempt']}" if ( - previous.get("state") == result["state"] + source_bot(previous.get("creator", {})) + and previous.get("state") == result["state"] and previous.get("target_url") == result["target_url"] and previous.get("description") == description ): continue - fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}") - if not source_identity(fresh) or fresh["head"]["sha"] != sha: - return False - current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}") - if current["object"]["sha"] != sha: - return False wrote = True api( f"repos/{REPOSITORY}/statuses/{sha}", @@ -939,23 +1354,43 @@ def reconcile_source( "description": description, }, ) - if jobs and wrote and (not source_runs_current(pr) or source_contract(pr) is None): + if ( + jobs + and wrote + and ( + not source_runs_current(pr) + or source_contract(pr) is None + or not source_lease(pr) + ) + ): reconcile_source(pr, allow_success=False) return False return bool(jobs) -def main() -> None: +def reconcile_main() -> None: """Reconcile open automation PRs, then start missing queue validation. Raises: subprocess.CalledProcessError: An API error other than a deleted ref occurred. """ - global SOURCE_HEAD, SOURCE_ACTIVITY + global \ + SOURCE_HEAD, \ + SOURCE_ACTIVITY, \ + SOURCE_CALLBACK, \ + SOURCE_SWEEP, \ + SOURCE_LAST_WRITE SOURCE_CACHE.clear() SOURCE_EVIDENCE.clear() + SOURCE_VIEW.clear() + SOURCE_RUN_LISTS.clear() + SOURCE_PREFLIGHT.clear() + SOURCE_LAST_WRITE = time.monotonic() + SOURCE_CALLBACK = False + SOURCE_SWEEP = False run_id = os.environ.get("SOURCE_RUN_ID", "") + SOURCE_CALLBACK = bool(run_id and SOURCE_CHECKS) try: notification = source_notification( run_id, @@ -965,24 +1400,44 @@ def main() -> None: except ObsoleteSourceNotification as error: print(f"::warning::Ignoring obsolete verified source callback: {error}") return + callback_pulls = None if notification is None: - wait_for_source_run(run_id) + SOURCE_CALLBACK = False + if not SOURCE_PRODUCER_COMPLETE: + wait_for_source_run(run_id) + if run_id and SOURCE_CHECKS: + # Queue callbacks also scan admitted source PRs. Share one fresh + # exact-head status view instead of six independent REST reads. + callback_pulls = source_snapshot(sources_only=False) else: SOURCE_HEAD, SOURCE_ACTIVITY = notification + SOURCE_CALLBACK = True + SOURCE_SWEEP = SOURCE_ACTIVITY == "completed" # Pagination is bounded by GitHub's repository PR limit, not a waiting loop. queue_expected = False page = 1 while True: - pulls = api( - f"repos/{REPOSITORY}/pulls?state=open&base={quote(DEFAULT_BRANCH, safe='')}" - f"&per_page={PAGE_SIZE}&page={page}" - ) + if SOURCE_CALLBACK: + pulls = source_snapshot() + if SOURCE_SWEEP: + source_readiness(pulls) + elif callback_pulls is not None: + pulls = callback_pulls + else: + pulls = api( + f"repos/{REPOSITORY}/pulls?state=open&base={quote(DEFAULT_BRANCH, safe='')}" + f"&per_page={PAGE_SIZE}&page={page}" + ) for pr in pulls: + if SOURCE_CALLBACK and not source_identity(pr): + continue queue_expected = admit(pr) or queue_expected - if len(pulls) < PAGE_SIZE: + if SOURCE_CALLBACK or callback_pulls is not None or len(pulls) < PAGE_SIZE: break page += 1 + if SOURCE_CALLBACK and (not SOURCE_SWEEP or not queue_expected): + return # GITHUB_TOKEN enqueue events do not start merge_group workflows. Explicit # dispatch is supported. Discover queue refs and validate their exact SHA. @@ -1025,6 +1480,19 @@ def main() -> None: advance_after_publication(published_passes) +def main() -> None: + """Run one reconciliation with no mutable callback state leaking to another pass.""" + global SOURCE_CALLBACK, SOURCE_SWEEP + try: + reconcile_main() + finally: + SOURCE_CALLBACK = False + SOURCE_SWEEP = False + SOURCE_VIEW.clear() + SOURCE_PREFLIGHT.clear() + SOURCE_RUN_LISTS.clear() + + def advance_after_publication(published: set[str]) -> None: """Wait once for GitHub's asynchronous merge to expose its next live front.""" for attempt in range(12): @@ -1068,10 +1536,18 @@ def dispatch_entry(entry: dict[str, Any]) -> bool: if not ref.startswith(QUEUE_PREFIX): return False base = api(f"repos/{REPOSITORY}/git/commits/{sha}")["parents"][0]["sha"] + inventory = queue_run_inventory(ref, sha) if SOURCE_CHECKS else None validation_runs: list[dict[str, Any]] = [] for workflow in WORKFLOWS: - runs = api( - f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page=1" + runs = ( + { + "total_count": int(workflow in inventory), + "workflow_runs": [inventory[workflow]] if workflow in inventory else [], + } + if inventory is not None + else api( + f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page=1" + ) ) if runs["total_count"]: latest_run = runs["workflow_runs"][0] @@ -1097,6 +1573,45 @@ def dispatch_entry(entry: dict[str, Any]) -> bool: return publish_dispatch_results(ref, sha, validation_runs) +def queue_run_inventory(ref: str, sha: str) -> dict[str, Any]: + """Read complete exact-head queue producers without five repeated lists.""" + source_run_inventory({"head": {"sha": sha}}) + result = {} + for workflow in WORKFLOWS: + metadata = source_read(f"repos/{REPOSITORY}/actions/workflows/{workflow}") + if ( + metadata.get("path") != f".github/workflows/{workflow}" + or metadata.get("state") != "active" + ): + raise ValueError("Queue workflow identity is not active") + for run in SOURCE_RUN_LISTS[sha]["workflow_runs"]: + if run.get("workflow_id") != metadata["id"]: + continue + # After a normal merge, the same commit can have a main-push run + # while its old queue ref still exists. That run is not evidence + # for this candidate; keep inspecting its own queue producers. + if run.get("event") not in {"merge_group", "workflow_dispatch"} or run.get( + "head_branch" + ) != ref.removeprefix("refs/heads/"): + continue + if not ( + run.get("path") == metadata["path"] + and run.get("head_sha") == sha + and run.get("head_branch") == ref.removeprefix("refs/heads/") + and run.get("repository", {}).get("full_name") == REPOSITORY + and run.get("head_repository", {}).get("full_name") == REPOSITORY + and run.get("event") in {"merge_group", "workflow_dispatch"} + and type(run.get("id")) is int + and run["id"] > 0 + and type(run.get("run_attempt")) is int + and run["run_attempt"] > 0 + ): + raise ValueError("Queue producer identity does not match its exact ref") + result[workflow] = {**run, "queue_workflow": workflow} + break + return result + + def read_statuses(sha: str) -> dict[str, Any]: """Read the latest commit status for every context. @@ -1143,6 +1658,7 @@ def latest_dispatch_jobs( """ jobs: list[dict[str, Any]] = [] for run in runs: + run_jobs: list[dict[str, Any]] = [] page = 1 while True: batch = api( @@ -1150,9 +1666,18 @@ def latest_dispatch_jobs( f"?filter=latest&per_page={PAGE_SIZE}&page={page}" )["jobs"] jobs.extend(batch) + run_jobs.extend(batch) if len(batch) < PAGE_SIZE: break page += 1 + if SOURCE_CHECKS: + if any( + job.get("run_id") != run["id"] + or job.get("run_attempt") != run["run_attempt"] + for job in run_jobs + ): + return None + continue # A separate dispatch or rerun can replace the evidence being inspected. fresh = api( f"repos/{REPOSITORY}/actions/workflows/{run['queue_workflow']}/runs" @@ -1165,13 +1690,35 @@ def latest_dispatch_jobs( } if any(current.get(key) != value for key, value in expected.items()): return None + if SOURCE_CHECKS and not queue_runs_current(sha, runs): + return None if any(job["head_sha"] != sha or job["status"] != "completed" for job in jobs): return None return [job for job in jobs if job["name"] != "Queue completion callback"] +def queue_runs_current(sha: str, runs: list[dict[str, Any]]) -> bool: + """Recheck every collected producer after collection or publication.""" + if not runs: + return False + current = queue_run_inventory(f"refs/heads/{runs[0]['head_branch']}", sha) + return all( + expected["queue_workflow"] in current + and all( + current[expected["queue_workflow"]].get(key) == expected.get(key) + for key in ("id", "run_attempt", "head_sha", "status", "conclusion") + ) + for expected in runs + ) + + def invalidate_results( - ref: str, sha: str, latest: dict[str, Any], reported: set[str] | None = None + ref: str, + sha: str, + latest: dict[str, Any], + reported: set[str] | None = None, + *, + leased: bool = False, ) -> None: """Invalidate old adapter results absent from the latest verified jobs.""" for status in latest.values(): @@ -1187,6 +1734,7 @@ def invalidate_results( "state": "pending", "target_url": status["target_url"], }, + leased=leased, ) @@ -1218,7 +1766,13 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> # success still requires the complete configured workflow set to pass. jobs = latest_dispatch_jobs(sha, evidence) if evidence else None if not jobs: - invalidate_results(ref, sha, latest) + leased = bool(SOURCE_CHECKS) and any( + status.get("description", "").startswith("Mirrored queue job") + for status in latest.values() + ) + if leased and not queue_ref_current(ref, sha): + return False + invalidate_results(ref, sha, latest, leased=leased) return False results = { job["name"]: "success" if job["conclusion"] == "success" else "failure" @@ -1226,8 +1780,12 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> if job["conclusion"] in failures or (all_passed and job["conclusion"] == "success") } - invalidate_results(ref, sha, latest, set(results)) + leased = bool(SOURCE_CHECKS) + if leased and not queue_ref_current(ref, sha): + return False + invalidate_results(ref, sha, latest, set(results), leased=leased) published_success = False + published_any = False for job in jobs: # Skipped and neutral jobs provide no passing or failing evidence. if job["name"] in results: @@ -1240,8 +1798,16 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> "state": results[job["name"]], "target_url": job["html_url"], }, + leased=leased, ) published_success |= changed and results[job["name"]] == "success" + published_any |= changed + if leased and published_any: + if not queue_ref_current(ref, sha): + return False + if not queue_runs_current(sha, runs): + invalidate_results(ref, sha, latest, leased=True) + return False source_required = {name for names in SOURCE_CHECKS.values() for name in names} return ( published_success @@ -1256,6 +1822,8 @@ def publish_status( sha: str, latest: dict[str, Any], result: dict[str, str], + *, + leased: bool = False, ) -> bool: """Publish one actual job result once, while its queue ref still matches. @@ -1267,8 +1835,7 @@ def publish_status( previous = latest.get(name, {}) if previous.get("state") == state and previous.get("target_url") == target_url: return False - current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}") - if current["object"]["sha"] != sha: + if not leased and not queue_ref_current(ref, sha): return False api( f"repos/{REPOSITORY}/statuses/{sha}", @@ -1283,8 +1850,15 @@ def publish_status( }, ) print(f"Reported {name}: {state} for {sha}") + latest[name] = {**result, "description": "Mirrored queue job result"} return True +def queue_ref_current(ref: str, sha: str) -> bool: + """Lease only this exact immutable queue commit, never a replacement head.""" + current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}") + return current["object"]["sha"] == sha + + if __name__ == "__main__": main() diff --git a/site/mkdocs.yml b/site/mkdocs.yml index 6cdcf29..e04fc53 100644 --- a/site/mkdocs.yml +++ b/site/mkdocs.yml @@ -36,6 +36,7 @@ nav: - OpenSSF baseline: openssf-baseline.md - Releases: releases.md - Release runbook: releases/README.md + - v2.13.0 release notes: releases/v2.13.0.md - v2.12.0 release notes: releases/v2.12.0.md - v2.11.0 release notes: releases/v2.11.0.md - v2.10.0 release notes: releases/v2.10.0.md diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py index 5923add..9084785 100644 --- a/tests/test_reconcile_queue.py +++ b/tests/test_reconcile_queue.py @@ -4,6 +4,7 @@ import base64 import contextlib +import copy import importlib.util import io import os @@ -25,6 +26,14 @@ spec.loader.exec_module(queue) +def snapshot_fixture(fixture: Any) -> list[dict[str, Any]]: + """Expose the same immutable-head statuses as the GraphQL projection.""" + queue.SOURCE_VIEW[fixture.sha] = { + status["context"]: status for status in fixture.statuses + } + return [fixture.pr] + + class ReconcileTests(unittest.TestCase): """Keep CI and queue failures observable without executing PR code.""" @@ -1008,6 +1017,19 @@ def _api(self, path: str, method: str = "GET", data: object = None) -> Any: if method != "GET": self.writes.append((path, data)) return None + if "/actions/workflows?" in path: + workflows = [ + { + "id": index, + "path": f".github/workflows/{workflow}", + "state": "active", + } + for index, workflow in enumerate(self.checks, 1) + ] + return {"total_count": len(workflows), "workflows": workflows} + if "/actions/runs?head_sha=" in path: + runs = [run for group in self.runs.values() for run in group] + return {"total_count": len(runs), "workflow_runs": runs} if "/git/commits/" in path: return {"parents": self.parents} if f"compare/{self.base}...{self.base}" in path: @@ -1239,7 +1261,11 @@ def _attest(self) -> None: "state": "success", "target_url": job["html_url"], "description": f"{queue.SOURCE_REPORT} result; attempt 1", - "creator": {"id": 41898282}, + "creator": { + "id": 41898282, + "type": "Bot", + "login": "github-actions[bot]", + }, } for jobs in self.jobs.values() for job in jobs @@ -1289,7 +1315,16 @@ def test_rerun_during_status_publication_is_bounded_and_invalidates_new_greens( def api(path: str, method: str = "GET", data: object = None) -> Any: result = self._api(path, method, data) if method == "POST": - self.statuses.append({**data, "creator": {"id": 41898282}}) + self.statuses.append( + { + **data, + "creator": { + "id": 41898282, + "type": "Bot", + "login": "github-actions[bot]", + }, + } + ) if len(self.writes) == 1: self.runs["ci.yml"][0].update( run_attempt=2, status="in_progress", conclusion=None @@ -1309,6 +1344,61 @@ def api(path: str, method: str = "GET", data: object = None) -> Any: ["success"] * 3 + ["pending"] * 3, ) + def test_noncanonical_creator_is_replaced_once_after_actual_jobs_prove_success( + self, + ) -> None: + for creator in ( + {"id": 9, "type": "User", "login": "developer"}, + {"id": 9, "type": "Bot", "login": "other[bot]"}, + {"id": 41898282, "type": "User", "login": "github-actions[bot]"}, + {"id": 41898282, "type": "Bot", "login": "other[bot]"}, + ): + self.setUp() + self._attest() + self.statuses[0]["creator"] = creator + calls = [] + + def api( + path: str, + method: str = "GET", + data: object = None, + *, + records: list[tuple[str, str]] = calls, + ) -> Any: + records.append((method, path)) + result = self._api(path, method, data) + if method == "POST": + previous = next( + value + for value in self.statuses + if value["context"] == data["context"] + ) + previous.update(data) + previous["creator"] = { + "id": 41898282, + "type": "Bot", + "login": "github-actions[bot]", + } + return result + + with ( + self.subTest(creator=creator), + patch.object(queue, "api", side_effect=api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", self.checks), + patch.object(queue, "WORKFLOWS", list(self.checks)), + patch.object(queue, "SOURCE_CALLBACK", False), + ): + self.assertTrue(queue.reconcile_source(self.pr)) + self.assertEqual(len(self.writes), 1) + self.assertTrue(any("/jobs?" in path for _method, path in calls)) + self.writes.clear() + calls.clear() + self.assertTrue(queue.reconcile_source(self.pr)) + self.assertEqual(self.writes, []) + self.assertFalse(any("/jobs?" in path for _method, path in calls)) + def test_untrusted_cached_status_cannot_replace_actual_job_evidence(self) -> None: self._attest() self.statuses[-1]["creator"]["id"] = 9 @@ -1478,7 +1568,7 @@ def test_source_event_inspects_only_its_exact_head(self) -> None: def test_rerun_event_invalidates_even_an_admitted_source(self) -> None: counts = self._case("admitted_event") - self.assertEqual(counts, {"get": 35, "write": 17, "jobs": 0}) + self.assertEqual(counts, {"get": 5, "write": 17, "jobs": 0}) def test_initial_requested_event_with_no_attestation_stops_immediately( self, @@ -1529,7 +1619,35 @@ def test_delayed_start_revalidates_completed_attempt(self) -> None: self.assertEqual(result, (self.fixture.sha, "completed")) wait.assert_not_called() + def test_already_completed_producer_reuses_actual_metadata_without_wait( + self, + ) -> None: + reads = [] + + def api(path: str, method: str = "GET", data: object = None) -> Any: + reads.append(path) + return self.fixture._api(path, method, data) + + with ( + patch.object(queue, "api", side_effect=api), + patch.object(queue, "wait_for_source_run") as wait, + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_CALLBACK", False), + ): + self.assertEqual( + queue.source_notification(self.run_id, "1", "completed"), + (self.fixture.sha, "completed"), + ) + wait.assert_not_called() + self.assertEqual( + sum(path.endswith("/actions/runs/" + self.run_id) for path in reads), 1 + ) + def test_new_attempt_during_completion_cannot_reuse_old_identity(self) -> None: + self.run.update(status="in_progress", conclusion=None) + def rerun(_run_id: str) -> None: self.run.update(status="in_progress", conclusion=None, run_attempt=2) @@ -1579,6 +1697,8 @@ def test_future_attempt_is_a_real_metadata_error(self) -> None: self.notification("in_progress", "2") def test_head_mutation_during_wait_is_a_real_error(self) -> None: + self.run.update(status="in_progress", conclusion=None) + def mutate(_run_id: str) -> None: self.run["head_sha"] = "e" * 40 @@ -1636,6 +1756,11 @@ def api( patch.object(queue, "SOURCE_EVIDENCE", {}), patch.object(queue, "admit") as admit, patch.object(queue, "dispatch_entry") as dispatch, + patch.object( + queue, + "source_ref", + return_value=None if reason == "deleted" else self.fixture.head, + ), patch.object(queue, "graphql") as graphql, contextlib.redirect_stdout(output), ): @@ -1759,6 +1884,12 @@ def api(path: str, method: str = "GET", data: object = None) -> Any: patch.object(queue, "SOURCE_HEAD", ""), patch.object(queue, "SOURCE_ACTIVITY", ""), patch.object(queue, "wait_for_source_run") as wait, + patch.object(queue, "source_ref", return_value=self.fixture.sha), + patch.object( + queue, + "source_snapshot", + side_effect=lambda *args: snapshot_fixture(self.fixture), + ), patch.object(queue, "graphql") as graphql, ): queue.main() @@ -1879,6 +2010,9 @@ def test_actual_sixteen_status_publication_reveals_next_front_later(self) -> Non fixture.statuses = [] old = {"ref": queue.QUEUE_PREFIX + "old", "object": {"sha": fixture.sha}} new = {"ref": queue.QUEUE_PREFIX + "next", "object": {"sha": "d" * 40}} + for runs in fixture.runs.values(): + for run in runs: + run["head_branch"] = old["ref"].removeprefix("refs/heads/") statuses = [] dispatches = [] ticks = 0 @@ -1902,7 +2036,7 @@ def api(path: str, method: str = "GET", data: object = None) -> Any: } if "/git/commits/" in path: return {"parents": [{"sha": fixture.base}]} - if "/actions/workflows/" in path and "head_sha=" + "d" * 40 in path: + if "/actions/" in path and "head_sha=" + "d" * 40 in path: return {"total_count": 0, "workflow_runs": []} return fixture._api(path, method, data) @@ -1934,3 +2068,735 @@ def front() -> dict[str, Any]: if __name__ == "__main__": unittest.main() + + +class SourceSnapshotTests(unittest.TestCase): + """Check live-ref projection and batch leases independently of native jobs.""" + + def node(self) -> dict[str, Any]: + return { + "number": 7, + "id": "PR7", + "isDraft": False, + "baseRefName": "main", + "headRefName": "automation/package-updates-example", + "headRefOid": "a" * 40, + "author": { + "__typename": "Bot", + "databaseId": 41898282, + "login": "github-actions", + }, + "headRepository": {"nameWithOwner": queue.REPOSITORY}, + "headRef": { + "name": "automation/package-updates-example", + "repository": {"nameWithOwner": queue.REPOSITORY}, + "target": {"oid": "a" * 40, "status": {"contexts": []}}, + }, + } + + def snapshot( + self, + node: dict[str, Any], + truncated: bool = False, + *, + sources_only: bool = True, + ) -> list[dict[str, Any]]: + response = { + "repository": { + "pullRequests": { + "pageInfo": {"hasNextPage": truncated}, + "nodes": [node], + } + } + } + with ( + patch.object(queue, "graphql", return_value=response), + patch.object(queue, "SOURCE_VIEW", {}), + ): + return queue.source_snapshot(sources_only=sources_only) + + def test_generic_snapshot_preserves_rest_bot_human_fork_and_draft_fields( + self, + ) -> None: + for kind in ("dependabot", "actions", "human", "fork", "draft"): + node = self.node() + if kind in {"dependabot", "draft"}: + node["author"].update(databaseId=49699333, login="dependabot") + node["headRefName"] = "dependabot/example" + node["isDraft"] = kind == "draft" + expected_login = "dependabot[bot]" + elif kind == "actions": + node["headRefName"] = "automation/update" + expected_login = "github-actions[bot]" + else: + node["author"] = { + "__typename": "User", + "databaseId": 1, + "login": "developer", + } + node["headRefName"] = "feature/change" + expected_login = "developer" + if kind == "fork": + node["headRepository"]["nameWithOwner"] = "developer/fork" + node["headRef"]["repository"]["nameWithOwner"] = "developer/fork" + node["headRef"]["name"] = node["headRefName"] + with self.subTest(kind=kind): + pr = self.snapshot(node, sources_only=False)[0] + self.assertEqual(pr["user"]["login"], expected_login) + self.assertEqual(pr["user"]["id"], node["author"]["databaseId"]) + self.assertEqual(pr["draft"], node["isDraft"]) + self.assertEqual( + pr["head"]["repo"]["full_name"], + node["headRepository"]["nameWithOwner"], + ) + self.assertEqual(pr["head"]["sha"], node["headRefOid"]) + self.assertEqual(pr["head"]["ref"], node["headRefName"]) + self.assertEqual(pr["base"]["ref"], node["baseRefName"]) + + def test_generic_dependabot_snapshot_still_admits_with_real_required_policy( + self, + ) -> None: + node = self.node() + node["author"].update(databaseId=49699333, login="dependabot") + node["headRefName"] = "dependabot/example" + node["headRef"]["name"] = node["headRefName"] + pr = self.snapshot(node, sources_only=False)[0] + + def api(path: str, method: str = "GET", data: object = None) -> Any: + if "/files?" in path: + return [] + if path.endswith("/status"): + return {"statuses": []} + if method == "POST": + return None + raise AssertionError(path) + + with ( + patch.object(queue, "SOURCE_CHECKS", {}), + patch.object(queue, "api", side_effect=api), + patch.object( + queue, "graphql", return_value={"node": {"mergeQueueEntry": None}} + ) as gql, + patch.object( + queue.subprocess, + "run", + return_value=subprocess.CompletedProcess([], 0, '[{"bucket":"pass"}]'), + ), + ): + self.assertTrue(queue.admit(pr)) + self.assertIn("enqueuePullRequest", gql.call_args.args[0]) + + def test_exact_bot_projection_preserves_source_identity(self) -> None: + self.assertTrue(queue.source_identity(self.snapshot(self.node())[0])) + + def test_login_alone_does_not_prove_bot_identity(self) -> None: + node = self.node() + node["author"]["__typename"] = "User" + self.assertEqual(self.snapshot(node), []) + + def test_inconsistent_live_head_and_cross_repository_ref_rejected(self) -> None: + for mutation in ("oid", "repository", "name"): + node = self.node() + if mutation == "oid": + node["headRef"]["target"]["oid"] = "b" * 40 + elif mutation == "repository": + node["headRef"]["repository"]["nameWithOwner"] = "other/repository" + else: + node["headRef"]["name"] = "other-branch" + with self.subTest(mutation=mutation), self.assertRaises(ValueError): + self.snapshot(node) + + def test_truncated_pull_inventory_is_not_silently_accepted(self) -> None: + with self.assertRaises(ValueError): + self.snapshot(self.node(), truncated=True) + + def test_duplicate_status_context_rejected(self) -> None: + node = self.node() + value = { + "context": "DCO sign-off", + "state": "SUCCESS", + "targetUrl": "url", + "description": "real", + "creator": node["author"], + } + node["headRef"]["target"]["status"]["contexts"] = [value, value] + with self.assertRaises(ValueError): + self.snapshot(node) + + def test_human_fork_does_not_abort_two_owned_source_heads(self) -> None: + first = self.node() + second = self.node() + second["number"] = 8 + second["id"] = "PR8" + second["headRefOid"] = "b" * 40 + second["headRef"]["target"]["oid"] = "b" * 40 + fork = self.node() + fork["author"] = {"__typename": "User", "login": "contributor"} + fork["headRepository"]["nameWithOwner"] = "contributor/fork" + fork["headRef"]["repository"]["nameWithOwner"] = "contributor/fork" + response = { + "repository": { + "pullRequests": { + "pageInfo": {"hasNextPage": False}, + "nodes": [fork, first, second], + } + } + } + with ( + patch.dict(os.environ, {"SOURCE_RUN_ID": "123"}), + patch.object(queue, "SOURCE_CHECKS", {"ci.yml": ["Build"]}), + patch.object( + queue, "source_notification", return_value=("a" * 40, "completed") + ), + patch.object(queue, "source_readiness"), + patch.object(queue, "graphql", return_value=response), + patch.object(queue, "admit", return_value=False) as admit, + ): + queue.main() + self.assertEqual( + [call.args[0]["number"] for call in admit.call_args_list], [7, 8] + ) + + def test_missing_deleted_ref_is_not_an_admission_lease(self) -> None: + node = self.node() + node["headRef"] = None + self.assertEqual(self.snapshot(node), []) + + def test_aggregate_run_inventory_rejects_truncation_and_count_mismatch( + self, + ) -> None: + for response in ( + {"total_count": 100, "workflow_runs": []}, + {"total_count": 1, "workflow_runs": []}, + ): + with ( + patch.object(queue, "api", return_value=response), + self.assertRaises(ValueError), + ): + queue.source_run_inventory({"head": {"sha": "a" * 40}}) + + def test_partial_ci_does_not_hide_real_terminal_dco_failure(self) -> None: + fixture = PackageSourceTests() + fixture.setUp() + fixture.runs["ci.yml"][0].update(status="in_progress", conclusion=None) + run = fixture.runs["dco.yml"][0] + run["conclusion"] = "failure" + fixture.jobs[run["id"]][0]["conclusion"] = "failure" + with ( + patch.object(queue, "api", side_effect=fixture._api), + patch.object(queue, "SOURCE_CHECKS", fixture.checks), + patch.object(queue, "WORKFLOWS", list(fixture.checks)), + patch.object(queue, "SOURCE_CALLBACK", False), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + ): + self.assertIsNone(queue.source_jobs(fixture.pr)) + self.assertEqual( + queue.SOURCE_EVIDENCE[fixture.sha]["failures"][0]["html_url"], + fixture.jobs[run["id"]][0]["html_url"], + ) + + +class SourceReadinessTests(unittest.TestCase): + """Keep the cheap readiness path read-only and fail closed on partial evidence.""" + + def setUp(self) -> None: + self.fixture = PackageSourceTests() + self.fixture.setUp() + self.fixture._attest() + self.fixture.statuses.append({"context": queue.ADMISSION, "state": "success"}) + self.suites = [] + for runs in self.fixture.runs.values(): + run = runs[0] + self.suites.append( + { + "status": "COMPLETED", + "conclusion": "SUCCESS", + "repository": {"nameWithOwner": queue.REPOSITORY}, + "commit": {"oid": self.fixture.sha}, + "branch": { + "name": self.fixture.branch, + "repository": {"nameWithOwner": queue.REPOSITORY}, + }, + "workflowRun": { + "databaseId": run["id"], + "runAttempt": run["run_attempt"], + "event": run["event"], + "file": {"path": run["path"]}, + "workflow": { + "databaseId": run["workflow_id"], + "state": "ACTIVE", + }, + }, + } + ) + + def response(self, truncated: bool = False) -> dict[str, Any]: + return { + "repository": { + "head0": { + "oid": self.fixture.sha, + "checkSuites": { + "pageInfo": {"hasNextPage": truncated}, + "nodes": self.suites, + }, + } + } + } + + def test_readiness_rejects_truncated_missing_and_tampered_suite_identity( + self, + ) -> None: + for mode in ( + "truncated", + "missing", + "branch", + "repository", + "workflow", + "head", + "approval", + ): + self.setUp() + if mode == "missing": + self.suites.pop() + elif mode == "branch": + self.suites[0]["branch"]["name"] = "wrong" + elif mode == "repository": + self.suites[0]["repository"]["nameWithOwner"] = "other/repo" + elif mode == "workflow": + self.suites[0]["workflowRun"]["workflow"]["databaseId"] = 99 + elif mode == "head": + self.suites[0]["commit"]["oid"] = "b" * 40 + elif mode == "approval": + self.suites[0]["conclusion"] = "ACTION_REQUIRED" + with ( + self.subTest(mode=mode), + patch.object( + queue, "graphql", return_value=self.response(mode == "truncated") + ), + patch.object(queue, "api", side_effect=self.fixture._api), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_PREFLIGHT", {}), + ): + queue.source_readiness([self.fixture.pr]) + self.assertNotIn(self.fixture.sha, queue.SOURCE_PREFLIGHT) + + def report(self, projected: bool = True) -> bool: + statuses = {value["context"]: value for value in self.fixture.statuses} + with ( + patch.object(queue, "api", side_effect=self.fixture._api), + patch.object(queue, "graphql", return_value=self.response()), + patch.object(queue, "source_lease", return_value=True), + patch.object(queue, "source_ref", return_value=self.fixture.base), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_PREFLIGHT", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_RUN_LISTS", {}), + patch.object(queue, "SOURCE_CALLBACK", True), + patch.object(queue, "SOURCE_HEAD", ""), + ): + if projected: + queue.source_readiness([self.fixture.pr]) + return queue.reconcile_source(self.fixture.pr, statuses) + + def test_failed_current_run_with_old_greens_cannot_take_either_duplicate_shortcut( + self, + ) -> None: + for projected in (False, True): + self.setUp() + run = self.fixture.runs["dco.yml"][0] + run["conclusion"] = "failure" + self.suites[-1]["conclusion"] = "FAILURE" + self.fixture.jobs[run["id"]][0]["conclusion"] = "failure" + with self.subTest(projected=projected): + self.assertFalse(self.report(projected)) + failures = [ + data + for _path, data in self.fixture.writes + if data["state"] == "failure" + ] + self.assertEqual(len(failures), 1) + self.assertEqual( + failures[0]["target_url"], + self.fixture.jobs[run["id"]][0]["html_url"], + ) + + def test_missing_ci_producer_does_not_hide_later_failed_dco(self) -> None: + self.fixture.runs["ci.yml"] = [] + self.suites.pop(0) + run = self.fixture.runs["dco.yml"][0] + run["conclusion"] = "failure" + self.fixture.jobs[run["id"]][0]["conclusion"] = "failure" + self.assertFalse(self.report()) + self.assertEqual( + [ + data["target_url"] + for _path, data in self.fixture.writes + if data["state"] == "failure" + ], + [self.fixture.jobs[run["id"]][0]["html_url"]], + ) + + def test_known_pending_ci_and_failed_dco_preserve_actual_failure_url(self) -> None: + run = self.fixture.runs["ci.yml"][0] + run.update(status="in_progress", conclusion=None) + self.suites[0].update(status="IN_PROGRESS", conclusion=None) + run = self.fixture.runs["dco.yml"][0] + run["conclusion"] = "failure" + self.suites[-1]["conclusion"] = "FAILURE" + self.fixture.jobs[run["id"]][0]["conclusion"] = "failure" + self.assertFalse(self.report()) + self.assertEqual( + [ + data["target_url"] + for _path, data in self.fixture.writes + if data["state"] == "failure" + ], + [self.fixture.jobs[run["id"]][0]["html_url"]], + ) + + def test_new_attempt_invalidates_old_managed_greens(self) -> None: + self.fixture.runs["ci.yml"][0].update( + run_attempt=2, status="in_progress", conclusion=None + ) + self.suites[0].update(status="IN_PROGRESS", conclusion=None) + self.suites[0]["workflowRun"]["runAttempt"] = 2 + self.assertFalse(self.report()) + self.assertTrue(self.fixture.writes) + self.assertEqual( + {data["state"] for _path, data in self.fixture.writes}, {"pending"} + ) + + def test_duplicate_completed_callback_does_not_fetch_jobs_or_publish(self) -> None: + self.fixture.fail_jobs_api = True + self.assertTrue(self.report()) + self.assertEqual(self.fixture.writes, []) + + +class SourcePacingTests(unittest.TestCase): + """Read operations do not consume the paced source content-write slots.""" + + def test_actual_transport_paces_rest_and_enqueue_but_not_graphql_reads( + self, + ) -> None: + now = [0.0] + sleeps = [] + + def sleep(seconds: float) -> None: + sleeps.append(seconds) + now[0] += seconds + + response = subprocess.CompletedProcess([], 0, "{}") + with ( + patch.object(queue, "SOURCE_CALLBACK", True), + patch.object(queue, "SOURCE_LAST_WRITE", None), + patch.object(queue.time, "monotonic", side_effect=lambda: now[0]), + patch.object(queue.time, "sleep", side_effect=sleep), + patch.object(queue.subprocess, "run", return_value=response), + ): + queue.api("repos/example/statuses/sha", "POST", {}) + queue.api("graphql", "POST", {"query": "query {viewer {login}}"}) + queue.api("graphql", "POST", {"query": "mutation { enqueuePullRequest }"}) + queue.api("repos/example/dispatches", "POST", {}) + self.assertEqual(sleeps, [4.0, 4.0]) + + +class SourceQueueBatchTests(unittest.TestCase): + """Keep all nineteen actual queue results bound to fresh producers and refs.""" + + def setUp(self) -> None: + self.fixture = SourceBudgetTests()._fixtures()[0] + self.fixture.branch = "gh-readonly-queue/main/pr-7-base" + self.fixture.statuses = [] + self.ref = "refs/heads/" + self.fixture.branch + for runs in self.fixture.runs.values(): + runs[0]["head_branch"] = self.fixture.branch + ci = self.fixture.runs["ci.yml"][0] + for position, name in enumerate( + ( + "Discover CI inputs", + "Flake lock (flake.lock) / check", + "Flake lock (flake/dev/flake.lock) / check", + ), + 50, + ): + self.fixture.jobs[ci["id"]].append( + { + "id": ci["id"] * 100 + position, + "run_id": ci["id"], + "run_attempt": 1, + "name": name, + "head_sha": self.fixture.sha, + "status": "completed", + "conclusion": "success", + "html_url": f"https://github.com/{queue.REPOSITORY}/actions/runs/{ci['id']}/job/{ci['id'] * 100 + position}", + } + ) + self.runs = [ + {**runs[0], "queue_workflow": workflow} + for workflow, runs in self.fixture.runs.items() + ] + self.reads = [] + + def report(self, mutation: str = "") -> bool: + posts = 0 + + def api(path: str, method: str = "GET", data: object = None) -> Any: + nonlocal posts + if method == "GET": + self.reads.append(path) + if mutation == "changed_before" and "/git/ref/" in path: + self.fixture.head = "b" * 40 + result = self.fixture._api(path, method, data) + if method == "POST": + posts += 1 + if posts == 1: + if mutation == "changed_after": + self.fixture.head = "b" * 40 + elif mutation == "new_attempt": + self.fixture.runs["ci.yml"][0].update( + run_attempt=2, status="in_progress", conclusion=None + ) + elif mutation == "new_run": + self.fixture.runs["ci.yml"][0].update( + id=9999, status="in_progress", conclusion=None + ) + return result + + with ( + patch.object(queue, "api", side_effect=api), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_RUN_LISTS", {}), + ): + return queue.publish_dispatch_results(self.ref, self.fixture.sha, self.runs) + + def test_all_nineteen_results_keep_exact_job_links_with_two_ref_leases( + self, + ) -> None: + self.assertTrue(self.report()) + results = [data for _path, data in self.fixture.writes] + expected = { + job["name"]: job["html_url"] + for jobs in self.fixture.jobs.values() + for job in jobs + } + self.assertEqual(len(results), 19) + self.assertEqual( + {data["context"]: data["target_url"] for data in results}, expected + ) + self.assertEqual({data["state"] for data in results}, {"success"}) + self.assertEqual(sum("/git/ref/" in path for path in self.reads), 2) + self.assertEqual( + sum("/actions/runs?head_sha=" in path for path in self.reads), 2 + ) + + def test_changed_queue_ref_before_batch_writes_nothing(self) -> None: + self.assertFalse(self.report("changed_before")) + self.assertEqual(self.fixture.writes, []) + + def test_changed_ref_after_batch_only_writes_old_immutable_sha_and_cannot_advance( + self, + ) -> None: + self.assertFalse(self.report("changed_after")) + self.assertEqual(len(self.fixture.writes), 19) + self.assertTrue( + all( + path.endswith("/" + self.fixture.sha) + for path, _data in self.fixture.writes + ) + ) + + def test_new_attempt_or_producer_after_batch_invalidates_every_published_success( + self, + ) -> None: + for mutation in ("new_attempt", "new_run"): + self.setUp() + with self.subTest(mutation=mutation): + self.assertFalse(self.report(mutation)) + results = [data for _path, data in self.fixture.writes] + self.assertEqual( + [data["state"] for data in results], + ["success"] * 19 + ["pending"] * 19, + ) + + def test_actual_failed_job_preserves_failure_link_and_no_success(self) -> None: + run = self.fixture.runs["ci.yml"][0] + run["conclusion"] = "failure" + self.runs[0]["conclusion"] = "failure" + job = self.fixture.jobs[run["id"]][0] + job["conclusion"] = "failure" + self.assertFalse(self.report()) + self.assertEqual( + [data["state"] for _path, data in self.fixture.writes], ["failure"] + ) + self.assertEqual(self.fixture.writes[0][1]["target_url"], job["html_url"]) + + def test_wrong_job_run_or_attempt_never_produces_success(self) -> None: + for field in ("run_id", "run_attempt"): + self.setUp() + self.fixture.jobs[self.runs[0]["id"]][0][field] = 999 + with self.subTest(field=field): + self.assertFalse(self.report()) + self.assertEqual(self.fixture.writes, []) + + def test_same_sha_main_push_does_not_replace_real_queue_codeql_evidence( + self, + ) -> None: + run = self.fixture.runs["codeql.yml"][0] + main_push = { + **run, + "id": run["id"] + 1000, + "event": "push", + "head_branch": "main", + } + self.fixture.runs["codeql.yml"].insert(0, main_push) + self.assertTrue(self.report()) + links = [data["target_url"] for _path, data in self.fixture.writes] + self.assertEqual(len(links), 19) + self.assertTrue(all(f"/runs/{main_push['id']}/" not in link for link in links)) + + def test_lingering_refs_with_main_pushes_do_not_block_new_live_front_dispatch( + self, + ) -> None: + old = self.fixture + older = copy.deepcopy(old) + older.sha = "c" * 40 + older.head = older.sha + older.branch = "gh-readonly-queue/main/older" + new_jobs = {} + for runs in older.runs.values(): + run = runs[0] + jobs = older.jobs[run["id"]] + run.update( + id=run["id"] + 1000, head_sha=older.sha, head_branch=older.branch + ) + for job in jobs: + job.update(id=job["id"] + 100000, run_id=run["id"], head_sha=older.sha) + job["html_url"] = ( + f"https://github.com/{queue.REPOSITORY}/actions/runs/{run['id']}/job/{job['id']}" + ) + new_jobs[run["id"]] = jobs + older.jobs = new_jobs + for fixture in (old, older): + run = fixture.runs["codeql.yml"][0] + fixture.runs["codeql.yml"].insert( + 0, + { + **run, + "id": run["id"] + 10000, + "event": "push", + "head_branch": "main", + }, + ) + fixture.statuses = [ + { + "context": job["name"], + "state": "success", + "target_url": job["html_url"], + "description": "Mirrored queue job result", + } + for jobs in fixture.jobs.values() + for job in jobs + ] + new_sha = "d" * 40 + new_ref = queue.QUEUE_PREFIX + "next" + refs = [ + {"ref": "refs/heads/" + fixture.branch, "object": {"sha": fixture.sha}} + for fixture in (old, older) + ] + [{"ref": new_ref, "object": {"sha": new_sha}}] + dispatched = [] + + def api(path: str, method: str = "GET", data: object = None) -> Any: + if method == "POST": + self.assertTrue(path.endswith("/dispatches")) + dispatched.append(data) + return None + if "/git/matching-refs/" in path: + return refs + if "head_sha=" + new_sha in path: + return {"total_count": 0, "workflow_runs": []} + if "/git/commits/" in path: + return {"parents": [{"sha": old.base}]} + if path.endswith(new_ref.removeprefix("refs/")): + return {"object": {"sha": new_sha}} + if ( + older.sha in path + or older.branch in path + or any(f"/runs/{run_id}/" in path for run_id in older.jobs) + ): + return older._api(path, method, data) + return old._api(path, method, data) + + response = { + "repository": { + "pullRequests": {"pageInfo": {"hasNextPage": False}, "nodes": []} + } + } + with ( + patch.dict( + os.environ, + { + "SOURCE_RUN_ID": str(self.runs[1]["id"]), + "SOURCE_RUN_ATTEMPT": "", + "SOURCE_NOTIFICATION_ACTIVITY": "", + }, + ), + patch.object(queue, "api", side_effect=api), + patch.object(queue, "graphql", return_value=response), + patch.object(queue, "SOURCE_CHECKS", old.checks), + patch.object(queue, "WORKFLOWS", list(old.checks)), + patch.object(queue, "SOURCE_HEAD", ""), + patch.object(queue, "SOURCE_ACTIVITY", ""), + ): + queue.main() + self.assertEqual(len(dispatched), 5) + self.assertEqual( + {data["ref"] for data in dispatched}, {new_ref.removeprefix("refs/heads/")} + ) + self.assertEqual(old.writes + older.writes, []) + + def test_nonqueue_branch_producer_is_not_candidate_evidence(self) -> None: + self.fixture.runs["ci.yml"][0]["head_branch"] = "main" + with ( + patch.object(queue, "api", side_effect=self.fixture._api), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_CACHE", {}), + ): + self.assertNotIn( + "ci.yml", queue.queue_run_inventory(self.ref, self.fixture.sha) + ) + + def test_aggregate_rejects_tampered_path_branch_repository_and_truncated_inventory( + self, + ) -> None: + for field, value in ( + ("path", ".github/workflows/wrong.yml"), + ("repository", {"full_name": "other/repo"}), + ("head_repository", {"full_name": "other/repo"}), + ("run_attempt", 0), + ): + self.setUp() + self.fixture.runs["ci.yml"][0][field] = value + with ( + self.subTest(field=field), + patch.object(queue, "api", side_effect=self.fixture._api), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_CACHE", {}), + self.assertRaises(ValueError), + ): + queue.queue_run_inventory(self.ref, self.fixture.sha) + with ( + patch.object( + queue, "api", return_value={"total_count": 100, "workflow_runs": []} + ), + self.assertRaises(ValueError), + ): + queue.queue_run_inventory(self.ref, self.fixture.sha) diff --git a/tests/test_source_callback_action.py b/tests/test_source_callback_action.py index 78c2137..ae0373d 100644 --- a/tests/test_source_callback_action.py +++ b/tests/test_source_callback_action.py @@ -14,7 +14,7 @@ class SourceCallbackActionTests(unittest.TestCase): """Keep existing queue callers compatible and bind source attempt notifications.""" - def invoke(self, activity: str) -> tuple[int, list[str], str]: + def invoke(self, activity: str, exit_code: int = 0) -> tuple[int, list[str], str]: action = yaml.load( ( Path(__file__).parents[1] / "actions/queue-completion/action.yml" @@ -29,8 +29,12 @@ def invoke(self, activity: str) -> tuple[int, list[str], str]: f"#!{sys.executable}\nimport json, os, sys\n" "from pathlib import Path\n" "Path(os.environ['CALLBACK_RECEIPT']).write_text(json.dumps(sys.argv[1:]))\n" + "sys.exit(int(os.environ['CALLBACK_EXIT_CODE']))\n" ) gh.chmod(0o755) + sleeper = root / "sleep" + sleeper.write_text(f"#!{sys.executable}\n") + sleeper.chmod(0o755) result = subprocess.run( ["bash", "-e", "-c", action["runs"]["steps"][0]["run"]], env={ @@ -38,6 +42,8 @@ def invoke(self, activity: str) -> tuple[int, list[str], str]: "PATH": f"{root}:{os.environ['PATH']}", "CALLBACK_RECEIPT": str(receipt), "DEFAULT_BRANCH": "main", + "GH_REPO": "nix-forge/example", + "CALLBACK_EXIT_CODE": str(exit_code), "GITHUB_RUN_ID": "123", "GITHUB_RUN_ATTEMPT": "2", "SOURCE_ACTIVITY": activity, @@ -49,7 +55,7 @@ def invoke(self, activity: str) -> tuple[int, list[str], str]: return ( result.returncode, json.loads(receipt.read_text()) if receipt.exists() else [], - result.stderr, + result.stdout + result.stderr, ) def test_legacy_queue_payload_is_unchanged(self) -> None: @@ -58,13 +64,14 @@ def test_legacy_queue_payload_is_unchanged(self) -> None: self.assertEqual( args, [ - "workflow", - "run", - "reconcile-merge-queue.yml", - "--ref", - "main", + "api", + "repos/nix-forge/example/actions/workflows/reconcile-merge-queue.yml/dispatches", + "--method", + "POST", "-f", - "source_run_id=123", + "ref=main", + "-f", + "inputs[source_run_id]=123", ], ) @@ -73,11 +80,20 @@ def test_source_start_and_completion_bind_actual_attempt(self) -> None: with self.subTest(activity=activity): status, args, stderr = self.invoke(activity) self.assertEqual(status, 0, stderr) - self.assertIn("source_run_id=123", args) - self.assertIn("source_run_attempt=2", args) - self.assertIn(f"source_activity={activity}", args) + self.assertIn("inputs[source_run_id]=123", args) + self.assertIn("inputs[source_run_attempt]=2", args) + self.assertIn(f"inputs[source_activity]={activity}", args) def test_invalid_phase_cannot_dispatch(self) -> None: status, args, stderr = self.invoke("completed --ref unreviewed") self.assertNotEqual(status, 0, stderr) self.assertEqual(args, []) + + def test_failed_notification_keeps_bounded_retries_and_visible_failure( + self, + ) -> None: + status, args, output = self.invoke("completed", exit_code=1) + self.assertEqual(status, 1) + self.assertEqual(output.count("notification attempt"), 3) + self.assertEqual(args[0], "api") + self.assertIn("inputs[source_activity]=completed", args)