diff --git a/CHANGELOG.md b/CHANGELOG.md index 14828bb..224de25 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,11 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); ver ## Unreleased +### Fixed + +- Served OneJev uses one HTTP decision deadline across its permitted retry. A retry receives only the remaining + `ONEJEV_TIMEOUT_S` budget, and an exhausted budget fails without sending another request. + ### Changed - The direct TypeSafe path pins `jev-1.13.0` instead of the moving `jev-latest` alias (which moved to Jev 1.14.0 diff --git a/s1a/decision_models/onejev.py b/s1a/decision_models/onejev.py index 978a85e..53d6edb 100644 --- a/s1a/decision_models/onejev.py +++ b/s1a/decision_models/onejev.py @@ -135,12 +135,16 @@ async def warm(self) -> Json: async def decide(self, body: Json, request_id: str) -> tuple[Json, int, dict[str, str]]: """One decision: the payload, its round trip in ms and the response headers.""" headers = {"X-Request-Id": request_id} + deadline = self._clock() + self._timeout_s retried = False while True: + remaining = deadline - self._clock() + if remaining <= 0: + raise self._no_answer(None) started = time.perf_counter() - post = self._client.post(f"{self.url}/v1/systemone", json=body, headers=headers, timeout=self._timeout_s) + post = self._client.post(f"{self.url}/v1/systemone", json=body, headers=headers, timeout=remaining) try: - response = await asyncio.wait_for(post, self._timeout_s) + response = await asyncio.wait_for(post, remaining) except TimeoutError as exc: raise self._no_answer(exc) from exc except _RETRIED_TRANSPORT as exc: @@ -183,7 +187,7 @@ async def decide(self, body: Json, request_id: str) -> tuple[Json, int, dict[str raise build_error(StatusCode.MODEL_CALL_FAILED, error_msg="served OneJev returned no answers object") return payload, ms, dict(response.headers) - def _no_answer(self, cause: Exception) -> Exception: + def _no_answer(self, cause: Exception | None) -> Exception: return build_error( StatusCode.MODEL_CALL_FAILED, cause=cause, diff --git a/tests/test_decision_models_onejev.py b/tests/test_decision_models_onejev.py index d474f23..86b4d18 100644 --- a/tests/test_decision_models_onejev.py +++ b/tests/test_decision_models_onejev.py @@ -3,6 +3,7 @@ from __future__ import annotations +import asyncio import base64 import json import os @@ -176,6 +177,83 @@ async def test_http_error_is_a_call_failure(self) -> None: self.assertIn("500", str(caught.exception)) +class TestDecisionDeadline(IsolatedAsyncioTestCase): + async def test_retry_uses_remaining_budget_for_httpx_and_wait_for(self) -> None: + for failure in (httpx.ConnectError("refused"), httpx.Response(502), httpx.Response(504)): + with self.subTest(failure=failure): + now = 0.0 + requests = [] + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal now + requests.append(request) + if len(requests) == 1: + now += 1.25 + if isinstance(failure, Exception): + raise failure + return failure + return httpx.Response(200, json={"answers": {}}) + + client = OneJevClient(url=URL, timeout_s=2.0, transport=httpx.MockTransport(handler), clock=lambda: now) + self.addAsyncCleanup(client.close) + with patch("s1a.decision_models.onejev.asyncio.wait_for", wraps=asyncio.wait_for) as wait_for: + await client.decide({"questions": {}}, "same-request") + self.assertEqual([call.args[1] for call in wait_for.call_args_list], [2.0, 0.75]) + for request, timeout in zip(requests, (2.0, 0.75), strict=True): + self.assertEqual(set(request.extensions["timeout"].values()), {timeout}) + self.assertEqual(request.headers["X-Request-Id"], "same-request") + + async def test_exhausted_budget_prevents_retry(self) -> None: + for failure in (httpx.ConnectError("refused"), httpx.Response(502), httpx.Response(504)): + with self.subTest(failure=failure): + now = 0.0 + requests = [] + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal now + requests.append(request) + now += 2.0 + if isinstance(failure, Exception): + raise failure + return failure + + client = OneJevClient(url=URL, timeout_s=2.0, transport=httpx.MockTransport(handler), clock=lambda: now) + self.addAsyncCleanup(client.close) + with self.assertRaises(BaseError) as caught: + await client.decide({"questions": {}}, "exhausted") + self.assertEqual(caught.exception.status, StatusCode.MODEL_CALL_FAILED) + self.assertIn("within 2 s", str(caught.exception)) + self.assertEqual(len(requests), 1) + + async def test_stalled_retry_is_cancelled_at_remaining_deadline(self) -> None: + now = 0.0 + attempts = 0 + stopped = asyncio.Event() + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal now, attempts + attempts += 1 + if attempts == 1: + now = 0.99 + raise httpx.ConnectError("refused", request=request) + try: + await asyncio.Event().wait() + finally: + stopped.set() + raise AssertionError("stalled retry was not cancelled") + + client = OneJevClient(url=URL, timeout_s=1.0, transport=httpx.MockTransport(handler), clock=lambda: now) + self.addAsyncCleanup(client.close) + with patch("s1a.decision_models.onejev.asyncio.wait_for", wraps=asyncio.wait_for) as wait_for: + with self.assertRaises(BaseError) as caught: + await client.decide({"questions": {}}, "stalled") + self.assertEqual(caught.exception.status, StatusCode.MODEL_CALL_FAILED) + self.assertIsInstance(caught.exception.__cause__, TimeoutError) + self.assertAlmostEqual(wait_for.call_args_list[1].args[1], 0.01) + self.assertEqual(attempts, 2) + self.assertTrue(stopped.is_set()) + + class TestMappingHelpers(TestCase): def test_state_wraps_text_and_marks_one_image(self) -> None: self.assertEqual(