From f74de141b4dbc3ebfa451a5c6e3cfd536dc14751 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:06:43 +0100 Subject: [PATCH] Reconcile convergence with current controls GitHub reruns preserve the failed workflow SHA, so a repaired convergence control cannot resolve an already committed release. A missed workflow_run event can also leave mutable aliases stranded without another attempt. Change-source: pulse-maintainer --- .../workflows/retry-release-convergence.yml | 105 ++-- .../subsystems/deployment-installability.md | 17 +- .../reconcile_release_convergence.py | 510 ++++++++++++++++++ .../reconcile_release_convergence_test.py | 295 ++++++++++ .../release_promotion_policy_test.py | 30 +- 5 files changed, 883 insertions(+), 74 deletions(-) create mode 100644 scripts/release_control/reconcile_release_convergence.py create mode 100644 scripts/release_control/reconcile_release_convergence_test.py diff --git a/.github/workflows/retry-release-convergence.yml b/.github/workflows/retry-release-convergence.yml index 62bd4abd3..fae0813c4 100644 --- a/.github/workflows/retry-release-convergence.yml +++ b/.github/workflows/retry-release-convergence.yml @@ -1,8 +1,9 @@ name: Retry Release Convergence -run-name: Retry release convergence run ${{ github.event.workflow_run.id }} attempt ${{ github.event.workflow_run.run_attempt }} +run-name: Reconcile release convergence debt # A committed release is never returned to draft because a mutable surface is -# temporarily unavailable. Re-run the complete convergence workflow so it +# temporarily unavailable. Re-run current controls, or dispatch them afresh +# when the failed run predates a repair, so the complete convergence workflow # reacquires the global lease and replays every idempotent surface. Successful # surfaces are safe to repeat; retrying only failed jobs would bypass the lease. on: @@ -13,80 +14,54 @@ on: - completed branches: - main + schedule: + # workflow_run delivery is not durable. Reconcile the current stable and + # preview channel heads hourly so a dropped event cannot strand aliases. + - cron: '37 * * * *' + workflow_dispatch: permissions: actions: write contents: read +concurrency: + group: retry-release-convergence + cancel-in-progress: false + jobs: retry_committed_convergence: - if: ${{ github.event.workflow_run.conclusion != 'success' }} + if: ${{ github.event_name == 'workflow_run' && github.event.workflow_run.conclusion != 'success' }} runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - name: Re-run the complete committed convergence + - name: Checkout current reconciliation controls + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Reconcile the failed convergence env: GH_TOKEN: ${{ github.token }} RUN_ID: ${{ github.event.workflow_run.id }} - RUN_ATTEMPT: ${{ github.event.workflow_run.run_attempt }} - DISPLAY_TITLE: ${{ github.event.workflow_run.display_title }} - MAX_ATTEMPTS: "5" - run: | - set -euo pipefail - if [[ ! "${DISPLAY_TITLE}" =~ ^Release\ convergence\ (v[^[:space:]]+)\ source\ ([0-9]+)$ ]]; then - echo "::error::Cannot derive the convergence tag from display title ${DISPLAY_TITLE}." - exit 1 - fi - tag="${BASH_REMATCH[1]}" - source_release_run_id="${BASH_REMATCH[2]}" - marker_url="https://github.com/${{ github.repository }}/releases/download/${tag}/release-activation.json" - release_state="$( - gh api "repos/${{ github.repository }}/releases/tags/${tag}" \ - --jq '[(.draft | tostring), (.published_at // "")] | @tsv' \ - 2>/dev/null || true - )" - is_draft="$(awk -F '\t' '{print $1}' <<<"${release_state}")" - published_at="$(awk -F '\t' '{print $2}' <<<"${release_state}")" - marker="$(mktemp)" - committed=false - if [ "${is_draft}" = "false" ] && [ -n "${published_at}" ] && \ - curl -fsSL --retry 2 --retry-delay 2 --retry-all-errors \ - -o "${marker}" "${marker_url}" && \ - jq -e \ - --arg tag "${tag}" \ - --arg source_release_run_id "${source_release_run_id}" \ - --arg convergence_run_id "${RUN_ID}" \ - '.schema_version == 1 and .tag == $tag and .source_release_run_id == $source_release_run_id and .convergence_run_id == $convergence_run_id' \ - "${marker}" >/dev/null; then - committed=true - fi - if [ "${committed}" != "true" ]; then - rm -f "${marker}" - source_status="$( - gh api "repos/${{ github.repository }}/actions/runs/${source_release_run_id}" \ - --jq '.status' 2>/dev/null || true - )" - if [ "${source_status}" != "completed" ] && (( RUN_ATTEMPT < 50 )); then - echo "Activation source run ${source_release_run_id} remains ${source_status:-unknown}; renewing the pre-commit convergence owner." - gh api \ - --method POST \ - -H "Accept: application/vnd.github+json" \ - -H "X-GitHub-Api-Version: 2026-03-10" \ - "repos/${{ github.repository }}/actions/runs/${RUN_ID}/rerun" - exit 0 - fi - echo "${tag} never crossed the irreversible activation commit and source run ${source_release_run_id} is ${source_status:-unavailable}; convergence will not retry." - exit 0 - fi - rm -f "${marker}" + run: >- + python3 scripts/release_control/reconcile_release_convergence.py + --repository "${GITHUB_REPOSITORY}" + --run-id "${RUN_ID}" - if (( RUN_ATTEMPT >= MAX_ATTEMPTS )); then - echo "::error::Committed release ${tag} still has convergence debt after ${RUN_ATTEMPT} attempts. Re-run the complete convergence workflow after repairing the failing surface." - exit 1 - fi - echo "Re-running complete convergence run ${RUN_ID} for committed ${tag}; next attempt is $((RUN_ATTEMPT + 1))/${MAX_ATTEMPTS}." - gh api \ - --method POST \ - -H "Accept: application/vnd.github+json" \ - -H "X-GitHub-Api-Version: 2026-03-10" \ - "repos/${{ github.repository }}/actions/runs/${RUN_ID}/rerun" + reconcile_missed_event: + if: ${{ github.event_name != 'workflow_run' }} + runs-on: ubuntu-24.04 + timeout-minutes: 10 + steps: + - name: Checkout current reconciliation controls + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - name: Reconcile current channel heads + env: + GH_TOKEN: ${{ github.token }} + run: >- + python3 scripts/release_control/reconcile_release_convergence.py + --repository "${GITHUB_REPOSITORY}" + --latest diff --git a/docs/release-control/v6/internal/subsystems/deployment-installability.md b/docs/release-control/v6/internal/subsystems/deployment-installability.md index 2da852a2a..82c3bc554 100644 --- a/docs/release-control/v6/internal/subsystems/deployment-installability.md +++ b/docs/release-control/v6/internal/subsystems/deployment-installability.md @@ -1847,7 +1847,7 @@ artifact-selection behaviour. the full workflow is retried so it reacquires the lease and safely replays all idempotent surfaces. The activation marker preserves its original convergence run ID. Once that - run is completed, a fresh manual convergence dispatch from fixed workflow + run is completed, a fresh successor convergence dispatch from fixed workflow code may adopt the same immutable tag, source run, target commit, release ID, R2 prefix, and activation-marker digest. Adoption happens only after lease acquisition and writes a unique owner record into the exact lease commit. @@ -1858,6 +1858,21 @@ artifact-selection behaviour. publication, while every successor gets immutable, run-scoped Git evidence. Reading a floating ref or a clobbered constant record is forbidden because cached prior bytes could authorize stale ownership. + Convergence retry is a current-control reconciliation, not unconditional + replay of the failed run's checkout. A terminal failure delivered through + `workflow_run` must be reconciled, and an hourly or manual pass must inspect + the current immutable stable and preview channel heads so a missed event + cannot strand mutable aliases. Discovery must not fall back behind a mutable + channel head. Before mutation, reconciliation must bind the exact mainline + workflow-dispatch identity, release, source run, activation marker and + original convergence owner; reject malformed marker digests; suppress a + newer matching run; and enforce one aggregate attempt budget across reruns + and successor runs. If the failed run used the current default-branch commit, + the complete run is rerun. If its checkout predates current controls, a fresh + default-branch convergence run is dispatched using only immutable identity + inputs recovered from the activation marker. Pre-commit owner renewal remains + limited to the original run while its exact source release run is active and + does not consume the post-commit convergence-debt budget. A support-only private Pro prerelease image is a narrower exception for customer verification of an already-fixed defect. It may dispatch the private `Build Pro Release` workflow with `publish_docker_image=true`, diff --git a/scripts/release_control/reconcile_release_convergence.py b/scripts/release_control/reconcile_release_convergence.py new file mode 100644 index 000000000..bc04b7183 --- /dev/null +++ b/scripts/release_control/reconcile_release_convergence.py @@ -0,0 +1,510 @@ +#!/usr/bin/env python3 +"""Reconcile a failed release convergence run without replaying stale controls.""" + +from __future__ import annotations + +import argparse +from datetime import datetime +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile +from typing import Any, Iterable + + +CONVERGENCE_PATH = ".github/workflows/release-convergence.yml" +CREATE_RELEASE_PATH = ".github/workflows/create-release.yml" +DISPLAY_TITLE = re.compile(r"^Release convergence (v[^\s]+) source ([1-9][0-9]*)$") +RELEASE_TAG = re.compile( + r"^v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)" + r"(?:-(?:alpha|beta|rc)\.[1-9][0-9]*)?$" +) +DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") +EXACT_SHA = re.compile(r"^[0-9a-f]{40}$") +TERMINAL_FAILURES = { + "action_required", + "cancelled", + "failure", + "stale", + "startup_failure", + "timed_out", +} + + +class ReconciliationError(ValueError): + """The remote evidence is incomplete or inconsistent.""" + + +class GitHubNotFound(ReconciliationError): + """GitHub returned a definite 404 for an object that may not exist yet.""" + + +def positive_int(value: object, subject: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ReconciliationError(f"{subject} is not a positive integer") + return value + + +def timestamp(value: object, subject: str) -> datetime: + if not isinstance(value, str): + raise ReconciliationError(f"{subject} has no timestamp") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ReconciliationError(f"{subject} has an invalid timestamp") from exc + if parsed.tzinfo is None: + raise ReconciliationError(f"{subject} timestamp has no timezone") + return parsed + + +def require_run_identity( + run: dict[str, Any], + *, + repository: str, + workflow_id: int, + path: str, + display_title: str | None = None, +) -> None: + checks = { + "repository": run.get("repository", {}).get("full_name") == repository, + "workflow": run.get("workflow_id") == workflow_id, + "workflow path": run.get("path") == path, + "event": run.get("event") == "workflow_dispatch", + } + if display_title is not None: + checks["display title"] = run.get("display_title") == display_title + failed = [name for name, valid in checks.items() if not valid] + if failed: + raise ReconciliationError("run has mismatched " + ", ".join(failed)) + + +def validate_marker( + marker: object, + *, + release: dict[str, Any], + tag: str, + source_run_id: int, +) -> int: + if not isinstance(marker, dict): + raise ReconciliationError("activation marker is not an object") + release_id = positive_int(release.get("id"), "release ID") + source_sha = release.get("target_commitish") + if not isinstance(source_sha, str) or EXACT_SHA.fullmatch(source_sha) is None: + raise ReconciliationError("release target is not an exact commit") + checks = { + "schema": marker.get("schema_version") == 1, + "tag": marker.get("tag") == tag, + "release ID": marker.get("release_id") == str(release_id), + "source commit": marker.get("target_commitish") == source_sha, + "source release run": marker.get("source_release_run_id") == str(source_run_id), + "R2 prefix": isinstance(marker.get("r2_prefix"), str), + "server digest": bool(DIGEST.fullmatch(str(marker.get("server_image_digest", "")))), + "control-plane digest": bool( + DIGEST.fullmatch(str(marker.get("control_plane_image_digest", ""))) + ), + "Helm digest": bool(DIGEST.fullmatch(str(marker.get("helm_chart_digest", "")))), + } + failed = [name for name, valid in checks.items() if not valid] + if failed: + raise ReconciliationError("activation marker has mismatched " + ", ".join(failed)) + owner = marker.get("convergence_run_id") + if not isinstance(owner, str) or not owner.isdigit() or int(owner) <= 0: + raise ReconciliationError("activation marker has an invalid convergence owner") + return int(owner) + + +def latest_failed_runs( + releases: Iterable[object], + runs: Iterable[object], + *, + workflow_id: int, + default_branch: str, +) -> list[int]: + """Return the latest failed convergence for each current immutable channel.""" + latest_release: dict[bool, tuple[datetime, dict[str, Any]]] = {} + for index, value in enumerate(releases): + if not isinstance(value, dict) or value.get("draft") is not False: + continue + prerelease = value.get("prerelease") + tag = value.get("tag_name") + if ( + not isinstance(prerelease, bool) + or not isinstance(tag, str) + or RELEASE_TAG.fullmatch(tag) is None + ): + continue + published = timestamp(value.get("published_at"), f"release {index}") + if prerelease not in latest_release or published > latest_release[prerelease][0]: + latest_release[prerelease] = (published, value) + + # Never fall back to an older release when the advertised channel head is + # mutable. That is continuity debt requiring a replacement, not a target + # whose aliases should be promoted again. + current_tags = { + str(release["tag_name"]) + for _, release in latest_release.values() + if release.get("immutable") is True + } + newest: dict[str, tuple[datetime, dict[str, Any]]] = {} + for index, value in enumerate(runs): + if not isinstance(value, dict): + continue + title = value.get("display_title") + match = DISPLAY_TITLE.fullmatch(title) if isinstance(title, str) else None + if match is None or match.group(1) not in current_tags: + continue + if ( + value.get("workflow_id") != workflow_id + or value.get("path") != CONVERGENCE_PATH + or value.get("event") != "workflow_dispatch" + or value.get("head_branch") != default_branch + ): + continue + created = timestamp(value.get("created_at"), f"convergence run {index}") + tag = match.group(1) + if tag not in newest or created > newest[tag][0]: + newest[tag] = (created, value) + + result: list[int] = [] + for _, run in newest.values(): + if run.get("status") == "completed" and run.get("conclusion") in TERMINAL_FAILURES: + result.append(positive_int(run.get("id"), "convergence run ID")) + return sorted(result) + + +class GitHub: + def __init__(self, repository: str, gh: str, *, mutate: bool = True) -> None: + self.repository = repository + self.gh = gh + self.mutate = mutate + + def _run(self, arguments: list[str], *, output: bool = True) -> str: + result = subprocess.run( + [self.gh, *arguments], + check=False, + capture_output=output, + text=True, + ) + if result.returncode != 0: + detail = result.stderr.strip().splitlines() if output else [] + suffix = f": {detail[-1]}" if detail else "" + if "HTTP 404" in result.stderr: + raise GitHubNotFound( + f"GitHub object was not found ({' '.join(arguments[:3])}){suffix}" + ) + raise ReconciliationError(f"GitHub command failed ({' '.join(arguments[:3])}){suffix}") + return result.stdout if output else "" + + def api(self, endpoint: str) -> dict[str, Any]: + try: + value = json.loads( + self._run( + [ + "api", + "-H", + "Accept: application/vnd.github+json", + "-H", + "X-GitHub-Api-Version: 2026-03-10", + endpoint, + ] + ) + ) + except json.JSONDecodeError as exc: + raise ReconciliationError(f"GitHub returned invalid JSON for {endpoint}") from exc + if not isinstance(value, dict): + raise ReconciliationError(f"GitHub returned a non-object for {endpoint}") + return value + + def pages(self, endpoint: str) -> list[object]: + output = self._run(["api", "--paginate", endpoint]) + decoder = json.JSONDecoder() + value: list[object] = [] + offset = 0 + while offset < len(output): + while offset < len(output) and output[offset].isspace(): + offset += 1 + if offset == len(output): + break + try: + page, offset = decoder.raw_decode(output, offset) + except json.JSONDecodeError as exc: + raise ReconciliationError( + f"GitHub returned invalid pages for {endpoint}" + ) from exc + value.append(page) + if not value: + raise ReconciliationError(f"GitHub returned no pages for {endpoint}") + return value + + def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> None: + if not self.mutate: + print(f"DRY RUN: POST {endpoint}") + return + arguments = [ + "api", + "--method", + "POST", + "-H", + "Accept: application/vnd.github+json", + "-H", + "X-GitHub-Api-Version: 2026-03-10", + endpoint, + ] + if payload is not None: + arguments.extend(["--input", "-"]) + result = subprocess.run( + [self.gh, *arguments], + input=json.dumps(payload, separators=(",", ":")), + check=False, + capture_output=True, + text=True, + ) + if result.returncode != 0: + detail = result.stderr.strip().splitlines() + suffix = f": {detail[-1]}" if detail else "" + raise ReconciliationError(f"GitHub mutation failed for {endpoint}{suffix}") + return + self._run(arguments) + + def download_marker(self, tag: str, directory: Path) -> dict[str, Any]: + self._run( + [ + "release", + "download", + tag, + "--repo", + self.repository, + "--pattern", + "release-activation.json", + "--dir", + str(directory), + ] + ) + try: + value = json.loads((directory / "release-activation.json").read_text()) + except (OSError, json.JSONDecodeError) as exc: + raise ReconciliationError("downloaded activation marker is unreadable") from exc + if not isinstance(value, dict): + raise ReconciliationError("downloaded activation marker is not an object") + return value + + +def flatten_pages(pages: Iterable[object], key: str | None = None) -> list[object]: + values: list[object] = [] + for index, page in enumerate(pages): + if key is None: + if not isinstance(page, list): + raise ReconciliationError(f"GitHub page {index} is not a list") + values.extend(page) + else: + if not isinstance(page, dict) or not isinstance(page.get(key), list): + raise ReconciliationError(f"GitHub page {index} has no {key} list") + values.extend(page[key]) + return values + + +def reconcile(github: GitHub, run_id: int, max_attempts: int) -> None: + repository = github.repository + repository_state = github.api(f"repos/{repository}") + default_branch = repository_state.get("default_branch") + if not isinstance(default_branch, str) or not default_branch: + raise ReconciliationError("repository has no default branch") + convergence_workflow = github.api( + f"repos/{repository}/actions/workflows/release-convergence.yml" + ) + create_workflow = github.api(f"repos/{repository}/actions/workflows/create-release.yml") + convergence_workflow_id = positive_int(convergence_workflow.get("id"), "workflow ID") + create_workflow_id = positive_int(create_workflow.get("id"), "create-release workflow ID") + + run = github.api(f"repos/{repository}/actions/runs/{run_id}") + require_run_identity( + run, + repository=repository, + workflow_id=convergence_workflow_id, + path=CONVERGENCE_PATH, + ) + if run.get("status") != "completed" or run.get("conclusion") not in TERMINAL_FAILURES: + print(f"Convergence run {run_id} no longer has terminal convergence debt; no action.") + return + title = run.get("display_title") + match = DISPLAY_TITLE.fullmatch(title) if isinstance(title, str) else None + if match is None: + raise ReconciliationError("convergence display title has an invalid identity") + tag, source_text = match.groups() + if RELEASE_TAG.fullmatch(tag) is None: + raise ReconciliationError("convergence display title has an invalid release tag") + source_run_id = int(source_text) + + all_runs = flatten_pages( + github.pages( + f"repos/{repository}/actions/workflows/release-convergence.yml/runs" + "?event=workflow_dispatch&per_page=100" + ), + "workflow_runs", + ) + matching_runs = [ + value + for value in all_runs + if isinstance(value, dict) + and value.get("workflow_id") == convergence_workflow_id + and value.get("display_title") == title + ] + if not matching_runs: + raise ReconciliationError("convergence history omitted the requested run") + newest = max(matching_runs, key=lambda item: timestamp(item.get("created_at"), "run")) + if newest.get("id") != run_id: + print(f"A newer convergence run already owns {title}; no action.") + return + try: + release = github.api(f"repos/{repository}/releases/tags/{tag}") + except GitHubNotFound: + release = {} + committed = ( + release.get("tag_name") == tag + and release.get("draft") is False + and release.get("immutable") is True + and isinstance(release.get("published_at"), str) + and bool(release.get("published_at")) + ) + if not committed: + source = github.api(f"repos/{repository}/actions/runs/{source_run_id}") + require_run_identity( + source, + repository=repository, + workflow_id=create_workflow_id, + path=CREATE_RELEASE_PATH, + ) + if source.get("status") != "completed" and positive_int( + run.get("run_attempt"), "run attempt" + ) < 50: + github.post(f"repos/{repository}/actions/runs/{run_id}/rerun") + print(f"Renewed pre-commit convergence owner {run_id} for active source {source_run_id}.") + return + print(f"{tag} has no immutable activation commit; no convergence retry was dispatched.") + return + + attempts = sum( + positive_int(item.get("run_attempt"), "run attempt") for item in matching_runs + ) + if attempts >= max_attempts: + raise ReconciliationError( + f"{title} still has convergence debt after {attempts} attempts" + ) + + with tempfile.TemporaryDirectory(prefix="release-convergence-") as raw: + marker = github.download_marker(tag, Path(raw)) + owner_run_id = validate_marker( + marker, release=release, tag=tag, source_run_id=source_run_id + ) + owner_run = github.api(f"repos/{repository}/actions/runs/{owner_run_id}") + require_run_identity( + owner_run, + repository=repository, + workflow_id=convergence_workflow_id, + path=CONVERGENCE_PATH, + display_title=title, + ) + if owner_run.get("status") != "completed": + raise ReconciliationError("activation marker owner is not terminal") + + default_commit = github.api(f"repos/{repository}/commits/{default_branch}").get("sha") + if not isinstance(default_commit, str) or EXACT_SHA.fullmatch(default_commit) is None: + raise ReconciliationError("default branch did not resolve to an exact commit") + if run.get("head_sha") == default_commit: + github.post(f"repos/{repository}/actions/runs/{run_id}/rerun") + print(f"Re-ran current-control convergence {run_id} for committed {tag}.") + return + + prerelease = release.get("prerelease") + if not isinstance(prerelease, bool): + raise ReconciliationError("release has no prerelease classification") + payload = { + "ref": default_branch, + "inputs": { + "tag": tag, + "version": tag.removeprefix("v"), + # The workflow-dispatch API models input values as strings even + # when the receiving workflow declares a boolean input. + "prerelease": "true" if prerelease else "false", + "target_commitish": marker["target_commitish"], + "release_id": marker["release_id"], + "r2_prefix": marker["r2_prefix"], + "source_release_run_id": marker["source_release_run_id"], + }, + } + github.post( + f"repos/{repository}/actions/workflows/release-convergence.yml/dispatches", + payload, + ) + print( + f"Dispatched fresh convergence controls after observing {default_commit} for committed {tag}; " + f"the failed run used {run.get('head_sha')}." + ) + + +def discover(github: GitHub) -> list[int]: + repository = github.repository + repository_state = github.api(f"repos/{repository}") + default_branch = repository_state.get("default_branch") + if not isinstance(default_branch, str) or not default_branch: + raise ReconciliationError("repository has no default branch") + workflow = github.api(f"repos/{repository}/actions/workflows/release-convergence.yml") + workflow_id = positive_int(workflow.get("id"), "workflow ID") + releases = flatten_pages( + github.pages(f"repos/{repository}/releases?per_page=100") + ) + runs = flatten_pages( + github.pages( + f"repos/{repository}/actions/workflows/release-convergence.yml/runs" + "?event=workflow_dispatch&per_page=100" + ), + "workflow_runs", + ) + return latest_failed_runs( + releases, + runs, + workflow_id=workflow_id, + default_branch=default_branch, + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--repository", required=True) + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument("--run-id", type=int) + group.add_argument("--latest", action="store_true") + parser.add_argument("--max-attempts", type=int, default=5) + parser.add_argument("--dry-run", action="store_true") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if args.run_id is not None and args.run_id <= 0: + print("run ID must be positive", file=sys.stderr) + return 2 + if not 1 <= args.max_attempts <= 20: + print("max attempts must be between 1 and 20", file=sys.stderr) + return 2 + gh = os.environ.get("GH_BIN", "gh") + github = GitHub(args.repository, gh, mutate=not args.dry_run) + try: + run_ids = discover(github) if args.latest else [args.run_id] + if not run_ids: + print("No current immutable release has unattended convergence debt.") + return 0 + for run_id in run_ids: + assert run_id is not None + reconcile(github, run_id, args.max_attempts) + except (OSError, ReconciliationError) as exc: + print(f"release convergence reconciliation failed: {exc}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/release_control/reconcile_release_convergence_test.py b/scripts/release_control/reconcile_release_convergence_test.py new file mode 100644 index 000000000..75bb3df9e --- /dev/null +++ b/scripts/release_control/reconcile_release_convergence_test.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import unittest + +import reconcile_release_convergence as subject + + +WORKFLOW_ID = 42 + + +def release(tag: str, published_at: str, *, prerelease: bool, immutable: bool = True): + return { + "id": abs(hash(tag)), + "tag_name": tag, + "draft": False, + "prerelease": prerelease, + "immutable": immutable, + "published_at": published_at, + } + + +def run( + run_id: int, + tag: str, + created_at: str, + *, + status: str = "completed", + conclusion: str = "failure", +): + return { + "id": run_id, + "workflow_id": WORKFLOW_ID, + "path": subject.CONVERGENCE_PATH, + "event": "workflow_dispatch", + "head_branch": "main", + "display_title": f"Release convergence {tag} source 1234", + "created_at": created_at, + "status": status, + "conclusion": conclusion, + } + + +class LatestFailedRunsTests(unittest.TestCase): + def select(self, releases, runs): + return subject.latest_failed_runs( + releases, runs, workflow_id=WORKFLOW_ID, default_branch="main" + ) + + def test_selects_failed_heads_of_stable_and_preview_channels(self): + releases = [ + release("v6.4.1", "2026-09-01T00:00:00Z", prerelease=False), + release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True), + ] + runs = [ + run(10, "v6.4.1", "2026-09-01T00:01:00Z"), + run(11, "v6.5.0-rc.1", "2026-09-02T00:01:00Z"), + ] + self.assertEqual([10, 11], self.select(releases, runs)) + + def test_does_not_fall_back_behind_a_mutable_channel_head(self): + releases = [ + release("v6.4.1", "2026-09-01T00:00:00Z", prerelease=False), + release( + "v6.4.2", "2026-09-02T00:00:00Z", prerelease=False, immutable=False + ), + ] + self.assertEqual([], self.select(releases, [run(10, "v6.4.1", "2026-09-01T00:01:00Z")])) + + def test_ignores_newer_companion_chart_releases(self): + releases = [ + release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True), + release( + "helm-chart-6.5.0-rc.1", + "2026-09-02T00:01:00Z", + prerelease=True, + ), + ] + self.assertEqual( + [10], + self.select( + releases, [run(10, "v6.5.0-rc.1", "2026-09-02T00:02:00Z")] + ), + ) + + def test_newer_success_or_active_run_clears_old_debt(self): + releases = [release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True)] + failed = run(10, "v6.5.0-rc.1", "2026-09-02T00:01:00Z") + success = run( + 11, + "v6.5.0-rc.1", + "2026-09-02T00:02:00Z", + conclusion="success", + ) + self.assertEqual([], self.select(releases, [failed, success])) + active = run( + 12, + "v6.5.0-rc.1", + "2026-09-02T00:03:00Z", + status="in_progress", + conclusion="", + ) + self.assertEqual([], self.select(releases, [failed, active])) + + def test_ignores_wrong_workflow_branch_and_title(self): + releases = [release("v6.5.0-rc.1", "2026-09-02T00:00:00Z", prerelease=True)] + wrong_workflow = run(10, "v6.5.0-rc.1", "2026-09-02T00:01:00Z") + wrong_workflow["workflow_id"] = 99 + wrong_branch = run(11, "v6.5.0-rc.1", "2026-09-02T00:02:00Z") + wrong_branch["head_branch"] = "release/v6.5" + wrong_title = run(12, "v6.5.0-rc.1", "2026-09-02T00:03:00Z") + wrong_title["display_title"] += " injected" + self.assertEqual( + [], self.select(releases, [wrong_workflow, wrong_branch, wrong_title]) + ) + + +class MarkerTests(unittest.TestCase): + def test_marker_binds_release_source_and_all_customer_digests(self): + current_release = { + "id": 88, + "target_commitish": "a" * 40, + } + marker = { + "schema_version": 1, + "tag": "v6.5.0-rc.1", + "release_id": "88", + "target_commitish": "a" * 40, + "source_release_run_id": "1234", + "convergence_run_id": "5678", + "r2_prefix": "packet", + "server_image_digest": "sha256:" + "b" * 64, + "control_plane_image_digest": "sha256:" + "c" * 64, + "helm_chart_digest": "sha256:" + "d" * 64, + } + self.assertEqual( + 5678, + subject.validate_marker( + marker, + release=current_release, + tag="v6.5.0-rc.1", + source_run_id=1234, + ), + ) + marker["server_image_digest"] = "sha256:bad" + with self.assertRaisesRegex(subject.ReconciliationError, "server digest"): + subject.validate_marker( + marker, + release=current_release, + tag="v6.5.0-rc.1", + source_run_id=1234, + ) + + +class FakeGitHub: + repository = "rcourtman/Pulse" + + def __init__(self, *, current_controls: bool = False, committed: bool = True): + self.run_id = 100 + self.source_run_id = 200 + self.owner_run_id = 90 + self.title = "Release convergence v6.5.0-rc.1 source 200" + self.old_sha = "a" * 40 + self.main_sha = self.old_sha if current_controls else "b" * 40 + self.committed = committed + self.posts = [] + self.runs = [self.run(self.run_id, self.old_sha)] + self.marker = { + "schema_version": 1, + "tag": "v6.5.0-rc.1", + "release_id": "300", + "target_commitish": "c" * 40, + "source_release_run_id": "200", + "convergence_run_id": "90", + "r2_prefix": "packet", + "server_image_digest": "sha256:" + "d" * 64, + "control_plane_image_digest": "sha256:" + "e" * 64, + "helm_chart_digest": "sha256:" + "f" * 64, + } + + def run(self, run_id, head_sha, *, attempt=1, status="completed", conclusion="failure"): + return { + "id": run_id, + "workflow_id": 42, + "path": subject.CONVERGENCE_PATH, + "event": "workflow_dispatch", + "repository": {"full_name": self.repository}, + "display_title": self.title, + "head_branch": "main", + "head_sha": head_sha, + "status": status, + "conclusion": conclusion, + "run_attempt": attempt, + "created_at": f"2026-09-02T00:{run_id % 60:02d}:00Z", + } + + def api(self, endpoint): + if endpoint == f"repos/{self.repository}": + return {"default_branch": "main"} + suffix = endpoint.removeprefix(f"repos/{self.repository}/") + if suffix == "actions/workflows/release-convergence.yml": + return {"id": 42} + if suffix == "actions/workflows/create-release.yml": + return {"id": 43} + if suffix == "commits/main": + return {"sha": self.main_sha} + if suffix == "releases/tags/v6.5.0-rc.1": + if not self.committed: + return {"tag_name": "v6.5.0-rc.1", "draft": True} + return { + "id": 300, + "tag_name": "v6.5.0-rc.1", + "target_commitish": "c" * 40, + "draft": False, + "immutable": True, + "published_at": "2026-09-02T00:00:00Z", + "prerelease": True, + } + if suffix.startswith("actions/runs/"): + run_id = int(suffix.rsplit("/", 1)[1]) + if run_id == self.owner_run_id: + owner = self.run(self.owner_run_id, self.old_sha, conclusion="success") + owner["status"] = "completed" + return owner + if run_id == self.source_run_id: + return { + "id": self.source_run_id, + "workflow_id": 43, + "path": subject.CREATE_RELEASE_PATH, + "event": "workflow_dispatch", + "repository": {"full_name": self.repository}, + "status": "in_progress", + } + return next(value for value in self.runs if value["id"] == run_id) + raise AssertionError(endpoint) + + def pages(self, endpoint): + self.assert_endpoint = endpoint + return [{"workflow_runs": self.runs}] + + def download_marker(self, tag, directory): + self.download = (tag, directory) + return self.marker + + def post(self, endpoint, payload=None): + self.posts.append((endpoint, payload)) + + +class ReconciliationTests(unittest.TestCase): + def test_stale_failed_run_dispatches_current_controls_with_bound_inputs(self): + github = FakeGitHub() + subject.reconcile(github, github.run_id, 5) + self.assertEqual(1, len(github.posts)) + endpoint, payload = github.posts[0] + self.assertTrue(endpoint.endswith("release-convergence.yml/dispatches")) + self.assertEqual("main", payload["ref"]) + self.assertEqual("v6.5.0-rc.1", payload["inputs"]["tag"]) + self.assertEqual("200", payload["inputs"]["source_release_run_id"]) + self.assertEqual("true", payload["inputs"]["prerelease"]) + + def test_current_failed_run_uses_exact_rerun(self): + github = FakeGitHub(current_controls=True) + subject.reconcile(github, github.run_id, 5) + self.assertEqual( + [(f"repos/{github.repository}/actions/runs/{github.run_id}/rerun", None)], + github.posts, + ) + + def test_newer_sibling_prevents_duplicate_mutation(self): + github = FakeGitHub() + github.runs.append(github.run(101, github.main_sha, status="in_progress", conclusion="")) + subject.reconcile(github, github.run_id, 5) + self.assertEqual([], github.posts) + + def test_attempt_budget_counts_distinct_runs_and_reruns(self): + github = FakeGitHub() + github.runs[0]["run_attempt"] = 3 + github.runs.append(github.run(101, github.old_sha, attempt=2)) + with self.assertRaisesRegex(subject.ReconciliationError, "after 5 attempts"): + subject.reconcile(github, 101, 5) + self.assertEqual([], github.posts) + + def test_precommit_owner_is_renewed_without_using_committed_budget(self): + github = FakeGitHub(committed=False) + github.runs[0]["run_attempt"] = 6 + subject.reconcile(github, github.run_id, 5) + self.assertEqual( + [(f"repos/{github.repository}/actions/runs/{github.run_id}/rerun", None)], + github.posts, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/release_control/release_promotion_policy_test.py b/scripts/release_control/release_promotion_policy_test.py index 697c3e86d..6a2541ad0 100644 --- a/scripts/release_control/release_promotion_policy_test.py +++ b/scripts/release_control/release_promotion_policy_test.py @@ -585,6 +585,9 @@ class ReleasePromotionPolicyTest(unittest.TestCase): release_workflow = read(".github/workflows/create-release.yml") convergence = read(".github/workflows/release-convergence.yml") retry = read(".github/workflows/retry-release-convergence.yml") + reconciler = read( + "scripts/release_control/reconcile_release_convergence.py" + ) commit_verdict = workflow_job_block(release_workflow, "release_commit_verdict") convergence_verdict = workflow_job_block(convergence, "convergence_verdict") @@ -612,10 +615,16 @@ class ReleasePromotionPolicyTest(unittest.TestCase): self.assertIn("the committed release remains public", convergence_verdict) self.assertIn("github.event.workflow_run.conclusion != 'success'", retry) - self.assertIn("actions/runs/${RUN_ID}/rerun", retry) + self.assertIn("cron: '37 * * * *'", retry) + self.assertIn("workflow_dispatch:", retry) + self.assertIn("reconcile_release_convergence.py", retry) + self.assertIn('--run-id "${RUN_ID}"', retry) + self.assertIn("--latest", retry) self.assertNotIn("rerun-failed-jobs", retry) - self.assertIn("RUN_ATTEMPT >= MAX_ATTEMPTS", retry) - self.assertIn("release-activation.json", retry) + self.assertIn('actions/runs/{run_id}/rerun', reconciler) + self.assertIn('release-convergence.yml/dispatches', reconciler) + self.assertIn("attempts >= max_attempts", reconciler) + self.assertIn("validate_marker(", reconciler) def test_mutating_reusable_workflows_have_no_direct_dispatch_lock_bypass(self) -> None: convergence = read(".github/workflows/release-convergence.yml") @@ -805,6 +814,9 @@ class ReleasePromotionPolicyTest(unittest.TestCase): release_workflow = read(".github/workflows/create-release.yml") convergence = read(".github/workflows/release-convergence.yml") retry = read(".github/workflows/retry-release-convergence.yml") + reconciler = read( + "scripts/release_control/reconcile_release_convergence.py" + ) activation = workflow_job_block(release_workflow, "activate_release") self.assertIn( @@ -830,11 +842,13 @@ class ReleasePromotionPolicyTest(unittest.TestCase): self.assertIn("successor adoption is not yet allowed", await_commit) self.assertIn("activation_marker_sha256", await_commit) - self.assertIn("source_release_run_id=\"${BASH_REMATCH[2]}\"", retry) - self.assertIn("actions/runs/${source_release_run_id}", retry) - self.assertIn('source_status}" != "completed"', retry) - self.assertIn("RUN_ATTEMPT < 50", retry) - self.assertIn("renewing the pre-commit convergence owner", retry) + self.assertIn("--run-id", retry) + self.assertIn("DISPLAY_TITLE = re.compile", reconciler) + self.assertIn("source_run_id = int(source_text)", reconciler) + self.assertIn('actions/runs/{source_run_id}', reconciler) + self.assertIn('source.get("status") != "completed"', reconciler) + self.assertIn('< 50', reconciler) + self.assertIn("Renewed pre-commit convergence owner", reconciler) self.assertIn("require_viable_convergence_owner()", activation) self.assertEqual(activation.count("require_viable_convergence_owner"), 3)