fix(sdk): retry transient QDC 5xx during job status/log polling

sdk/benchmark/qdc/_qdc.py

@@ -11,8 +11,10 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import random
+import re
 import time
 import zipfile
 from pathlib import Path
@@ -32,9 +34,12 @@
 
 POLL_INTERVAL = 30
 LOG_UPLOAD_TIMEOUT = 600
-SUBMIT_RETRY_BUDGET = 3600
+SUBMIT_RETRY_BUDGET = 7200
 SUBMIT_BACKOFF_BASE = 30
 SUBMIT_BACKOFF_CAP = 300
+TRANSIENT_RETRY_ATTEMPTS = 5
+TRANSIENT_BACKOFF_BASE = 10
+TRANSIENT_BACKOFF_CAP = 60
 
 FRAMEWORK = {
     "linux": TestFramework.BASH,
@@ -53,6 +58,43 @@ def _is_quota_error(exc: Exception) -> bool:
     return any(h in msg for h in _QUOTA_HINTS)
 
 
+# QDC's upload/status/log endpoints occasionally blip with a 5xx (mostly 504
+# Gateway Time-out, some 500) with no fault on our side; retrying the same
+# call a few seconds later almost always succeeds. Unlike the pending-job
+# quota, this isn't capacity we need to wait out, so the retry is short.
+# A 504's body is an empty/HTML gateway page rather than JSON, and the SDK's
+# try_call() does a bare json.loads() on it without checking the status code
+# first, so the observable symptom is often a JSONDecodeError rather than an
+# exception that mentions "status code 5xx".
+_TRANSIENT_STATUS_RE = re.compile(r"status code 5\d\d\b")
+
+
+def _is_transient_error(exc: Exception) -> bool:
+    if isinstance(exc, json.JSONDecodeError):
+        return True
+    return bool(_TRANSIENT_STATUS_RE.search(str(exc)))
+
+
+def _call_with_retry(fn, *args, what: str, **kwargs):
+    attempt = 0
+    while True:
+        try:
+            return fn(*args, **kwargs)
+        except Exception as exc:
+            if not _is_transient_error(exc) or attempt >= TRANSIENT_RETRY_ATTEMPTS:
+                raise
+            sleep = min(TRANSIENT_BACKOFF_CAP, TRANSIENT_BACKOFF_BASE * 2**attempt)
+            log.warning(
+                "%s hit a transient error (attempt %d): %s; retrying in %ds",
+                what,
+                attempt + 1,
+                exc,
+                sleep,
+            )
+            time.sleep(sleep)
+            attempt += 1
+
+
 def make_client(api_key: str):
     return qdc_api.get_public_api_client_using_api_key(
         api_key_header=api_key,
@@ -73,7 +115,9 @@ def _wait_for_job(client, job_id: str, timeout: int) -> str:
     terminal = {JobState.COMPLETED, JobState.CANCELED}
     elapsed = 0
     while elapsed < timeout:
-        raw = qdc_api.get_job_status(client, job_id)
+        raw = _call_with_retry(
+            qdc_api.get_job_status, client, job_id, what="job status poll"
+        )
         try:
             state = JobState(raw)
         except ValueError:
@@ -90,7 +134,11 @@ def _wait_for_job(client, job_id: str, timeout: int) -> str:
 def _submit_with_retry(client, **submit_kwargs) -> str:
     # All max-parallel runners share the key's pending-job quota; instead of
     # crashing when it's full, back off (with jitter so the runners don't retry
-    # in lockstep) until a slot frees up or the budget runs out.
+    # in lockstep) until a slot frees up or the budget runs out. Under a full
+    # matrix run the quota realistically stays contested for closer to two
+    # hours than one (observed elapsed-before-giveup times up to ~54min on a
+    # 60min budget, with the "N pending" count still fluctuating rather than
+    # stuck), so the budget leaves comfortable room under the job's timeout.
     elapsed = 0
     attempt = 0
     while True:
@@ -127,7 +175,13 @@ def submit_and_wait(
 ) -> str:
     """Upload the artifact, submit the job (retrying on quota), and block until terminal."""
     log.info("uploading artifact (%d MB)", zip_path.stat().st_size // 1_000_000)
-    artifact_id = qdc_api.upload_file(client, str(zip_path), ArtifactType.TESTSCRIPT)
+    artifact_id = _call_with_retry(
+        qdc_api.upload_file,
+        client,
+        str(zip_path),
+        ArtifactType.TESTSCRIPT,
+        what="upload artifact",
+    )
     job_id = _submit_with_retry(
         client,
         target_id=target_id,
@@ -165,19 +219,36 @@ def download_log_members(
     """
     elapsed = 0
     while elapsed < LOG_UPLOAD_TIMEOUT:
-        status = (qdc_api.get_job_log_upload_status(client, job_id) or "").lower()
+        status = (
+            _call_with_retry(
+                qdc_api.get_job_log_upload_status,
+                client,
+                job_id,
+                what="log upload status poll",
+            )
+            or ""
+        ).lower()
         if status in {"completed", "failed"}:
             break
         log.info("waiting for log upload (status=%s)", status)
         time.sleep(POLL_INTERVAL)
         elapsed += POLL_INTERVAL
 
     out: list[tuple[str, bytes]] = []
-    for lf in qdc_api.get_job_log_files(client, job_id) or []:
+    log_files = _call_with_retry(
+        qdc_api.get_job_log_files, client, job_id, what="list job log files"
+    )
+    for lf in log_files or []:
         if not want(_basename(lf.filename)):
             continue
         dl = tmp / "log.bin"
-        qdc_api.download_job_log_files(client, lf.filename, str(dl))
+        _call_with_retry(
+            qdc_api.download_job_log_files,
+            client,
+            lf.filename,
+            str(dl),
+            what="download job log file",
+        )
         if zipfile.is_zipfile(dl):
             with zipfile.ZipFile(dl) as z:
                 for name in z.namelist():