[v3-4-test] Keep the spot interruption notice green on transient GitHub errors (#75681)

dev/breeze/doc/ci/05_workflows.md

@@ -525,7 +525,9 @@ its last. An interrupted job ends with exit code 137 and no test failure in its
   each pull request every 10 minutes, including runs still in progress, and keeps one
   comment on the pull request: which jobs were interrupted and that RunsOn re-runs them,
   and, once the re-run finished, whether it passed. It runs on a schedule, not on the
-  runs' events, so no pull request run gets write access.
+  runs' events, so no pull request run gets write access. GitHub server errors (HTTP 5xx)
+  are retried, and when they persist the affected run is skipped with a warning and left
+  to the next check rather than failing the workflow; other errors still fail it.
 - **Slack.** The `notify-slack` job of a canary does not post when every failure is a Spot
   interruption that RunsOn re-runs, and leaves the notification state alone; it posts for
   RunsOn's re-run instead, including when the re-run passed. It still posts for attempt

scripts/ci/spot_interruptions.py

@@ -51,6 +51,7 @@
 import re
 import subprocess
 import sys
+import time
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from urllib.parse import quote
@@ -78,6 +79,10 @@
 COMMENT_AUTHOR = "github-actions[bot]"
 # A job cancelled while its run was not, e.g. by its timeout, is a failure too.
 FAILED_CONCLUSIONS = ("failure", "cancelled", "timed_out")
+# GitHub answers with a server error while overloaded, and the same request often succeeds later.
+TRANSIENT_ERROR = re.compile(r"\(HTTP 5\d\d\)")
+GH_API_ATTEMPTS = 3
+GH_API_RETRY_DELAY_SECONDS = 5
 
 
 @dataclass(frozen=True)
@@ -89,8 +94,22 @@ class FailedJob:
     conclusion: str = "failure"
 
 
+def is_transient(error: subprocess.CalledProcessError) -> bool:
+    return bool(TRANSIENT_ERROR.search(error.stderr or ""))
+
+
 def gh_api(*args: str) -> str:
-    return subprocess.run(["gh", "api", *args], check=True, capture_output=True, text=True).stdout
+    command = ["gh", "api", *args]
+    # A POST that failed with a server error may still have created the comment.
+    attempts = 1 if "POST" in args else GH_API_ATTEMPTS
+    for attempt in range(1, attempts):
+        try:
+            return subprocess.run(command, check=True, capture_output=True, text=True).stdout
+        except subprocess.CalledProcessError as e:
+            if not is_transient(e):
+                raise
+            time.sleep(GH_API_RETRY_DELAY_SECONDS * attempt)
+    return subprocess.run(command, check=True, capture_output=True, text=True).stdout
 
 
 def is_spot_interrupted(repo: str, job_id: int) -> bool:
@@ -362,15 +381,34 @@ def build(failed: list[FailedJob], for_attempt: int) -> str | None:
     write_comment(repo, pr_number, existing, body)
 
 
+def describe_error(error: subprocess.CalledProcessError) -> str:
+    return " ".join((error.stderr or str(error)).split())
+
+
 def run_pr_comments(repo: str, now: datetime) -> bool:
-    """Update the notices; return whether every run could be handled."""
+    """
+    Update the notices; return False when a run could not be handled for a lasting reason.
+
+    Server errors that outlast the retries only skip the work they hit: the next scheduled run does
+    it again.
+    """
+    try:
+        runs = list_pull_request_runs(repo, now)
+    except subprocess.CalledProcessError as e:
+        if not is_transient(e):
+            raise
+        print(f"::warning::Could not list the CI runs, GitHub keeps failing: {describe_error(e)}")
+        return True
     ok = True
-    for run in list_pull_request_runs(repo, now):
+    for run in runs:
         try:
             update_notice(repo, run)
         except subprocess.CalledProcessError as e:
-            print(f"Could not handle run {run['html_url']}: {e.stderr or e}", file=sys.stderr)
-            ok = False
+            if is_transient(e):
+                print(f"::warning::Skipped run {run['html_url']}, GitHub keeps failing: {describe_error(e)}")
+            else:
+                print(f"Could not handle run {run['html_url']}: {describe_error(e)}", file=sys.stderr)
+                ok = False
     return ok
 
 

scripts/tests/ci/test_spot_interruptions.py

@@ -295,18 +295,59 @@ def test_find_comment_ignores_comments_of_others(spot):
         assert spot.find_comment("apache/airflow", 42) == spot.Comment(1, f"{marker}\nnotice")
 
 
-def test_run_pr_comments_carries_on_past_a_failing_run(spot):
+SERVER_ERROR = subprocess.CalledProcessError(1, ["gh"], stderr="gh: Server Error (HTTP 502)\n")
+NOT_FOUND_ERROR = subprocess.CalledProcessError(1, ["gh"], stderr="gh: Not Found (HTTP 404)\n")
+
+
+@pytest.mark.parametrize(("error", "expected_ok"), [(SERVER_ERROR, True), (NOT_FOUND_ERROR, False)])
+def test_run_pr_comments_carries_on_past_a_failing_run(spot, capsys, error, expected_ok):
     runs = [_run(branch="a"), _run(branch="b")]
-    error = subprocess.CalledProcessError(1, ["gh"], stderr="HTTP 502")
 
     with (
         mock.patch.object(spot, "list_pull_request_runs", autospec=True, return_value=runs),
         mock.patch.object(spot, "update_notice", autospec=True, side_effect=[error, None]) as mock_update,
     ):
         ok = spot.run_pr_comments("apache/airflow", datetime(2026, 10, 25, tzinfo=UTC))
 
-    assert ok is False
+    assert ok is expected_ok
     assert mock_update.call_count == 2
+    assert capsys.readouterr().out.startswith("::warning::") is expected_ok
+
+
+@pytest.mark.parametrize(("error", "expected_ok"), [(SERVER_ERROR, True), (NOT_FOUND_ERROR, None)])
+def test_run_pr_comments_when_listing_the_runs_fails(spot, error, expected_ok):
+    with mock.patch.object(spot, "list_pull_request_runs", autospec=True, side_effect=error):
+        if expected_ok is None:
+            with pytest.raises(subprocess.CalledProcessError):
+                spot.run_pr_comments("apache/airflow", datetime(2026, 10, 25, tzinfo=UTC))
+        else:
+            assert spot.run_pr_comments("apache/airflow", datetime(2026, 10, 25, tzinfo=UTC)) is expected_ok
+
+
+@pytest.mark.parametrize(
+    ("args", "errors", "expected_calls", "expected_error"),
+    [
+        pytest.param(("repos/x",), [SERVER_ERROR, SERVER_ERROR], 3, None, id="retries-server-errors"),
+        pytest.param(("repos/x",), [SERVER_ERROR] * 3, 3, SERVER_ERROR, id="gives-up"),
+        pytest.param(("repos/x",), [NOT_FOUND_ERROR], 1, NOT_FOUND_ERROR, id="no-retry-on-client-error"),
+        pytest.param(("-X", "POST", "repos/x"), [SERVER_ERROR], 1, SERVER_ERROR, id="no-retry-on-post"),
+        pytest.param(("-X", "PATCH", "repos/x"), [SERVER_ERROR], 2, None, id="retries-patch"),
+    ],
+)
+@mock.patch("time.sleep", autospec=True)
+@mock.patch("subprocess.run", autospec=True)
+def test_gh_api_retries_server_errors(
+    mock_run, mock_sleep, spot, args, errors, expected_calls, expected_error
+):
+    mock_run.side_effect = [*errors, subprocess.CompletedProcess([], 0, stdout="out")]
+
+    if expected_error is None:
+        assert spot.gh_api(*args) == "out"
+    else:
+        with pytest.raises(subprocess.CalledProcessError) as raised:
+            spot.gh_api(*args)
+        assert raised.value is expected_error
+    assert mock_run.call_count == expected_calls
 
 
 def test_run_slack_writes_outputs(spot, monkeypatch, tmp_path):