fix(bench): retry the post-ingest database write across RDS restarts (#11018)

`post-ingest.py` connected to the benchmarks database only once. During
the two RDS restarts on 2026-10-28, four ingest jobs failed. The script
now repeats the write after 5, 10, 20, 40, 60, 60 and 60 seconds when
the connection is lost or refused. Each try connects again with a new
IAM token. A statement timeout or a validation error is not retried.

A repeated write is safe. The commit row is an upsert, and the fact rows
land in one transaction, so a lost connection leaves nothing
half-written. The worst case is about 5.5 minutes, which fits the
10-minute `commit-metadata` job. New tests cover the retry behavior.

Signed-off-by: Will Manning <will@willmanning.io>

Author: lwwmanning

scripts/post-ingest.py

@@ -30,10 +30,15 @@
 import os
 import subprocess
 import sys
+import time
 import urllib.request
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from types import ModuleType
+from typing import TypeVar
+
+T = TypeVar("T")
 
 # MUST equal `bench/web/lib/schema-version.ts::SCHEMA_VERSION` in vortex-data/vortex.dev.
 # Bumping this is a coordinated change across the website contract, v3.rs, and
@@ -987,16 +992,57 @@ def refresh_site_cache(base_url: str, token: str, timeout: float) -> None:
     _warm_default_windows(base, timeout)
 
 
+# A restart of the RDS instance refuses connections for 30 seconds to a few minutes: on 2026-10-28
+# an engine upgrade took 33 seconds and an OS update about 3 minutes, and four ingest jobs failed.
+# These waits, about 4 minutes in all, span that window.
+_RETRY_DELAYS_S = (5, 10, 20, 40, 60, 60, 60)
+
+
+def _with_retries(
+    attempt: Callable[[], T],
+    *,
+    should_retry: Callable[[Exception], bool],
+    delays: tuple[float, ...] = _RETRY_DELAYS_S,
+    sleep: Callable[[float], None] = time.sleep,
+) -> T:
+    """Run `attempt`, and run it again after each delay while it raises an error that
+    `should_retry` accepts. The last error is raised once the delays run out."""
+    for delay in delays:
+        try:
+            return attempt()
+        except Exception as exc:
+            if not should_retry(exc):
+                raise
+            print(f"warning: database unavailable ({exc}); retrying in {delay:g}s", file=sys.stderr)
+            sleep(delay)
+    return attempt()
+
+
+def _is_transient_db_error(exc: Exception) -> bool:
+    """True for a lost or refused connection. A statement timeout is not retried: it is a slow
+    query, not an outage."""
+    import psycopg
+
+    return isinstance(exc, psycopg.OperationalError) and not isinstance(exc, psycopg.errors.QueryCanceled)
+
+
 def _main_postgres(args: argparse.Namespace) -> int:
     records = read_records(args.jsonl_path)
     # `build_commit` runs `git show <commit_sha>`, so the SHA must be in the runner's local git
     # history. A shallow checkout missing the SHA fails before opening a database connection.
     commit = build_commit(args.commit_sha, args.branch, args.repo_url, args.git_dir)
-    conn = connect_postgres(args.postgres, args.region)
-    try:
-        inserted, updated = ingest_postgres(conn, commit, records)
-    finally:
-        conn.close()
+
+    # Each attempt connects again, with a fresh IAM token, and writes the whole file. Retrying the
+    # write is safe: the commit row is an idempotent upsert, and the fact rows land in one
+    # transaction, so a connection lost mid-write leaves nothing half-written.
+    def write() -> tuple[int, int]:
+        conn = connect_postgres(args.postgres, args.region)
+        try:
+            return ingest_postgres(conn, commit, records)
+        finally:
+            conn.close()
+
+    inserted, updated = _with_retries(write, should_retry=_is_transient_db_error)
     print(
         json.dumps(
             {"records": len(records), "inserted": inserted, "updated": updated},

scripts/tests/test_post_ingest.py

@@ -94,3 +94,54 @@ def apply(connection, mid, record) -> bool:
     ) == sorted(expected_records, key=lambda record: (record["kind"], record.get("query_idx", 0)))
     keys = [(kind, mid) for kind, mid, _ in applied]
     assert keys == sorted(keys)
+
+
+class _Transient(Exception):
+    pass
+
+
+def test_retries_transient_errors_until_success(writer):
+    calls: list[int] = []
+    sleeps: list[float] = []
+
+    def attempt() -> str:
+        calls.append(1)
+        if len(calls) < 3:
+            raise _Transient("connection refused")
+        return "written"
+
+    result = writer._with_retries(
+        attempt, should_retry=lambda e: isinstance(e, _Transient), delays=(5, 10, 20), sleep=sleeps.append
+    )
+    assert result == "written"
+    assert len(calls) == 3
+    assert sleeps == [5, 10]
+
+
+def test_raises_the_last_error_when_retries_run_out(writer):
+    sleeps: list[float] = []
+
+    def attempt() -> None:
+        raise _Transient("connection refused")
+
+    with pytest.raises(_Transient):
+        writer._with_retries(
+            attempt, should_retry=lambda e: isinstance(e, _Transient), delays=(5, 10), sleep=sleeps.append
+        )
+    assert sleeps == [5, 10]
+
+
+def test_does_not_retry_other_errors(writer):
+    sleeps: list[float] = []
+
+    def attempt() -> None:
+        raise ValueError("bad record")
+
+    with pytest.raises(ValueError):
+        writer._with_retries(attempt, should_retry=lambda e: isinstance(e, _Transient), sleep=sleeps.append)
+    assert sleeps == []
+
+
+def test_retry_schedule_spans_an_rds_restart(writer):
+    # An OS update on the instance refused connections for about 3 minutes.
+    assert sum(writer._RETRY_DELAYS_S) >= 180