From 13bdec6ab07bd00b3276464af025a520486955fb Mon Sep 17 00:00:00 2001 From: HJ <16863475+hjcud@users.noreply.github.com> Date: Sun, 9 Aug 2026 15:57:25 +0900 Subject: [PATCH] fix: retry transient result publication --- src/backtest_engine/production.py | 50 ++++++++++--- tests/test_production.py | 115 ++++++++++++++++++++++++++++++ 2 files changed, 156 insertions(+), 9 deletions(-) diff --git a/src/backtest_engine/production.py b/src/backtest_engine/production.py index cd34e80..b3745db 100644 --- a/src/backtest_engine/production.py +++ b/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: diff --git a/tests/test_production.py b/tests/test_production.py index d060dbc..8a4cc9a 100644 --- a/tests/test_production.py +++ b/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",