fix: retry transient result publication

src/backtest_engine/production.py

@@ -13,11 +13,12 @@
 import json
 import os
 import uuid
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from decimal import Decimal
 from pathlib import Path
+from time import sleep
 from typing import Any, BinaryIO
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
@@ -612,10 +613,28 @@ def dead_letter(self, message: DeadLetteredMessage) -> None:
 
 
 class HttpResultSink:
-    def __init__(self, base_url: str, token: str, *, timeout_seconds: float = 10.0) -> None:
+    def __init__(
+        self,
+        base_url: str,
+        token: str,
+        *,
+        timeout_seconds: float = 10.0,
+        max_transport_attempts: int = 3,
+        retry_backoff_seconds: float = 0.25,
+        sleeper: Callable[[float], None] = sleep,
+    ) -> None:
+        if timeout_seconds <= 0:
+            raise ValueError("timeout_seconds must be positive")
+        if not 1 <= max_transport_attempts <= 5:
+            raise ValueError("max_transport_attempts must be between 1 and 5")
+        if not 0 <= retry_backoff_seconds <= 5:
+            raise ValueError("retry_backoff_seconds must be between 0 and 5")
         self._base_url = base_url.rstrip("/")
         self._token = token
         self._timeout = timeout_seconds
+        self._max_transport_attempts = max_transport_attempts
+        self._retry_backoff = retry_backoff_seconds
+        self._sleep = sleeper
 
     def publish(self, event: Mapping[str, Any], *, delivery_attempt: int) -> None:
         run_id = str(event["backtestRunId"])
@@ -629,13 +648,26 @@ def publish(self, event: Mapping[str, Any], *, delivery_attempt: int) -> None:
                 "X-Delivery-Attempt": str(delivery_attempt),
             },
         )
-        try:
-            with urlopen(request, timeout=self._timeout) as response:
-                status = response.status
-        except (HTTPError, URLError, TimeoutError) as exc:
-            raise RuntimeError(f"backtest result ingestion failed: {exc}") from exc
-        if status != 200:
-            raise RuntimeError(f"backtest result ingestion returned HTTP {status}")
+        for transport_attempt in range(1, self._max_transport_attempts + 1):
+            try:
+                with urlopen(request, timeout=self._timeout) as response:
+                    status = response.status
+            except HTTPError as exc:
+                # The endpoint answered. Retrying a deterministic HTTP response can
+                # only duplicate load and must not disguise a rejected event as a
+                # transient network outage.
+                raise RuntimeError(f"backtest result ingestion failed: {exc}") from exc
+            except (URLError, TimeoutError) as exc:
+                if transport_attempt == self._max_transport_attempts:
+                    raise RuntimeError(
+                        "backtest result ingestion failed after "
+                        f"{transport_attempt} transport attempts: {exc}"
+                    ) from exc
+                self._sleep(self._retry_backoff * (2 ** (transport_attempt - 1)))
+                continue
+            if status != 200:
+                raise RuntimeError(f"backtest result ingestion returned HTTP {status}")
+            return
 
 
 class S3ParquetMarketDataReader:

tests/test_production.py

@@ -16,6 +16,7 @@
 from backtest_engine.backtest_request_intake import RequestLane
 from backtest_engine.production import (
     ConfigurationError,
+    HttpResultSink,
     JwtAuthenticator,
     PostgresCompiledPlanSource,
     PostgresDatasetManifestSource,
@@ -49,6 +50,120 @@ def test_worker_correlation_id_is_normalized_to_the_result_event_uuid_format() -
     ) == "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
 
 
+class _HttpResultResponse:
+    status = 200
+
+    def __enter__(self) -> _HttpResultResponse:
+        return self
+
+    def __exit__(self, *_args: object) -> None:
+        return None
+
+
+def test_result_sink_retries_a_transient_timeout_with_the_exact_same_event(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    requests: list[object] = []
+    sleeps: list[float] = []
+
+    def urlopen(request: object, *, timeout: float) -> _HttpResultResponse:
+        assert timeout == 2.0
+        requests.append(request)
+        if len(requests) == 1:
+            raise TimeoutError("temporary result endpoint timeout")
+        return _HttpResultResponse()
+
+    monkeypatch.setattr(production, "urlopen", urlopen)
+    sink = HttpResultSink(
+        "https://api.example.com",
+        "worker-token",
+        timeout_seconds=2.0,
+        max_transport_attempts=3,
+        retry_backoff_seconds=0.25,
+        sleeper=sleeps.append,
+    )
+
+    sink.publish(
+        {"backtestRunId": str(BOT_ID), "status": "FAILED"},
+        delivery_attempt=1,
+    )
+
+    assert len(requests) == 2
+    assert requests[0].data == requests[1].data  # type: ignore[attr-defined]
+    assert requests[0].headers == requests[1].headers  # type: ignore[attr-defined]
+    assert sleeps == [0.25]
+
+
+def test_result_sink_bounds_transient_transport_retries(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls = 0
+    sleeps: list[float] = []
+
+    def urlopen(_request: object, *, timeout: float) -> _HttpResultResponse:
+        nonlocal calls
+        assert timeout == 1.0
+        calls += 1
+        raise TimeoutError("still unavailable")
+
+    monkeypatch.setattr(production, "urlopen", urlopen)
+    sink = HttpResultSink(
+        "https://api.example.com",
+        "worker-token",
+        timeout_seconds=1.0,
+        max_transport_attempts=3,
+        retry_backoff_seconds=0.1,
+        sleeper=sleeps.append,
+    )
+
+    with pytest.raises(RuntimeError, match="after 3 transport attempts"):
+        sink.publish(
+            {"backtestRunId": str(BOT_ID), "status": "FAILED"},
+            delivery_attempt=1,
+        )
+
+    assert calls == 3
+    assert sleeps == [0.1, 0.2]
+
+
+def test_result_sink_does_not_retry_an_http_response_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls = 0
+    sleeps: list[float] = []
+
+    def urlopen(_request: object, *, timeout: float) -> _HttpResultResponse:
+        nonlocal calls
+        assert timeout == 1.0
+        calls += 1
+        raise production.HTTPError(
+            "https://api.example.com",
+            503,
+            "Service Unavailable",
+            {},
+            None,
+        )
+
+    monkeypatch.setattr(production, "urlopen", urlopen)
+    sink = HttpResultSink(
+        "https://api.example.com",
+        "worker-token",
+        timeout_seconds=1.0,
+        max_transport_attempts=3,
+        retry_backoff_seconds=0.1,
+        sleeper=sleeps.append,
+    )
+
+    with pytest.raises(RuntimeError, match="HTTP Error 503"):
+        sink.publish(
+            {"backtestRunId": str(BOT_ID), "status": "FAILED"},
+            delivery_attempt=1,
+        )
+
+    assert calls == 1
+    assert sleeps == []
+
+
 def test_service_specific_aws_endpoint_overrides_the_legacy_shared_endpoint() -> None:
     environment = {
         "AWS_ENDPOINT_URL": "http://legacy:4566",