diff --git a/.github/workflows/weekly_load_anomaly.yml b/.github/workflows/weekly_load_anomaly.yml index 6853ffa1cde..4c2103f026d 100644 --- a/.github/workflows/weekly_load_anomaly.yml +++ b/.github/workflows/weekly_load_anomaly.yml @@ -2,7 +2,7 @@ name: "Weekly Load Anomaly Check" on: schedule: - - cron: "0 6 * * 1" + - cron: "0 12 * * 6" workflow_dispatch: permissions: diff --git a/tests/e2e/e2e_config.py b/tests/e2e/e2e_config.py index 4fdd91aa3d9..30353b93dd6 100644 --- a/tests/e2e/e2e_config.py +++ b/tests/e2e/e2e_config.py @@ -81,6 +81,7 @@ LOAD_MAX_FAILURE_RATIO = float(os.environ.get("E2E_LOAD_MAX_FAILURE_RATIO", "0.0 WEEKLY_ANOMALY_OPT_IN_ENV = "E2E_WEEKLY_ANOMALY" ANOMALY_SESSIONS = int(os.environ.get("E2E_ANOMALY_SESSIONS", "6")) ANOMALY_TURNS_PER_SESSION = int(os.environ.get("E2E_ANOMALY_TURNS_PER_SESSION", "6")) +ANOMALY_TURN_ATTEMPTS = int(os.environ.get("E2E_ANOMALY_TURN_ATTEMPTS", "3")) ANOMALY_MAX_ERROR_RATIO = float(os.environ.get("E2E_ANOMALY_MAX_ERROR_RATIO", "0.05")) ANOMALY_MIN_WARM_CACHE_READ_SHARE = float( os.environ.get("E2E_ANOMALY_MIN_WARM_CACHE_READ_SHARE", "0.65") diff --git a/tests/e2e/load/session_anomaly.py b/tests/e2e/load/session_anomaly.py index b2e23d24ab2..c29b833635d 100644 --- a/tests/e2e/load/session_anomaly.py +++ b/tests/e2e/load/session_anomaly.py @@ -110,6 +110,22 @@ def _without_cache_control(message: RichMessage) -> RichMessage: ) +RETRY_BACKOFF_SECONDS = 2.0 + + +def retried( + call: Callable[[], Result[SessionMessagesResponse]], + attempts: int, + backoff_seconds: float = RETRY_BACKOFF_SECONDS, + sleep: Callable[[float], None] = time.sleep, +) -> Result[SessionMessagesResponse]: + result = call() + if isinstance(result, Success) or attempts <= 1: + return result + sleep(backoff_seconds) + return retried(call, attempts - 1, backoff_seconds, sleep) + + def _metric( result: Result[SessionMessagesResponse], turn_index: int, latency_seconds: float ) -> TurnMetric: @@ -144,6 +160,7 @@ def _drive_turns( history: tuple[RichMessage, ...], turn_index: int, remaining_turns: int, + attempts_per_turn: int, ) -> tuple[TurnMetric, ...]: if remaining_turns == 0: return () @@ -156,15 +173,18 @@ def _drive_turns( ], ) started = time.monotonic() - result = transport.post( - "/v1/messages", - headers=transport.bearer(key), - json=SessionMessagesRequest( - model=model, - system=[system_block], - messages=[*history, user_turn], + result = retried( + lambda: transport.post( + "/v1/messages", + headers=transport.bearer(key), + json=SessionMessagesRequest( + model=model, + system=[system_block], + messages=[*history, user_turn], + ), + response_type=SessionMessagesResponse, ), - response_type=SessionMessagesResponse, + attempts_per_turn, ) turn = _metric(result, turn_index, time.monotonic() - started) if not isinstance(result, Success): @@ -188,12 +208,13 @@ def _drive_turns( ), turn_index + 1, remaining_turns - 1, + attempts_per_turn, ), ) def run_session( - transport: Transport, key: str, model: str, turns: int + transport: Transport, key: str, model: str, turns: int, attempts_per_turn: int ) -> tuple[TurnMetric, ...]: marker = unique_marker() return _drive_turns( @@ -205,15 +226,23 @@ def run_session( (), 1, turns, + attempts_per_turn, ) def run_concurrent_sessions( - transport: Transport, key: str, model: str, sessions: int, turns_per_session: int + transport: Transport, + key: str, + model: str, + sessions: int, + turns_per_session: int, + attempts_per_turn: int, ) -> tuple[TurnMetric, ...]: with ThreadPoolExecutor(max_workers=sessions) as pool: futures = [ - pool.submit(run_session, transport, key, model, turns_per_session) + pool.submit( + run_session, transport, key, model, turns_per_session, attempts_per_turn + ) for _ in range(sessions) ] return tuple(turn for future in futures for turn in future.result()) diff --git a/tests/e2e/load/test_session_anomaly.py b/tests/e2e/load/test_session_anomaly.py index 9acedbac424..80f8aff3ba4 100644 --- a/tests/e2e/load/test_session_anomaly.py +++ b/tests/e2e/load/test_session_anomaly.py @@ -4,7 +4,14 @@ from itertools import count, repeat import pytest -from session_anomaly import TurnMetric, settled_spend, summarize +from e2e_http import NetworkError, Success +from session_anomaly import ( + SessionMessagesResponse, + TurnMetric, + retried, + settled_spend, + summarize, +) def _ok_turn(turn_index: int) -> TurnMetric: @@ -53,6 +60,42 @@ class TestSummarizePlannedTurns: assert report.error_ratio == 0.0 +class TestRetried: + def test_transient_failures_then_success_returns_the_success(self) -> None: + outcome = Success(data=SessionMessagesResponse()) + calls = iter( + (NetworkError(message="overloaded"), NetworkError(message="overloaded"), outcome) + ) + + result = retried(lambda: next(calls), attempts=3, sleep=lambda _: None) + + assert result is outcome + + def test_exhausted_attempts_return_the_last_failure(self) -> None: + last_attempt = NetworkError(message="still overloaded") + never_reached = NetworkError(message="a fourth attempt would break the budget") + calls = iter( + (NetworkError(message="overloaded"), last_attempt, never_reached) + ) + + result = retried(lambda: next(calls), attempts=2, sleep=lambda _: None) + + assert result is last_attempt + assert next(calls) is never_reached + + def test_first_try_success_never_sleeps(self) -> None: + def sleep_means_retry(_: float) -> None: + raise AssertionError("slept after a successful attempt") + + result = retried( + lambda: Success(data=SessionMessagesResponse()), + attempts=3, + sleep=sleep_means_retry, + ) + + assert isinstance(result, Success) + + class TestSettledSpend: def test_partial_total_between_batch_flushes_is_not_accepted_as_final(self) -> None: reads = iter((0.1, 0.1, 0.1, 0.35, 0.35, 0.35, 0.35, 0.35)) diff --git a/tests/e2e/load/test_weekly_session_anomaly_e2e.py b/tests/e2e/load/test_weekly_session_anomaly_e2e.py index de7ca33ba7c..d4ef883702e 100644 --- a/tests/e2e/load/test_weekly_session_anomaly_e2e.py +++ b/tests/e2e/load/test_weekly_session_anomaly_e2e.py @@ -11,6 +11,7 @@ from e2e_config import ( ANOMALY_MIN_WARM_CACHE_READ_SHARE, ANOMALY_SESSIONS, ANOMALY_SPEND_SETTLE_SECONDS, + ANOMALY_TURN_ATTEMPTS, ANOMALY_TURNS_PER_SESSION, unique_marker, ) @@ -77,6 +78,7 @@ class TestWeeklySessionAnomaly: model_name, ANOMALY_SESSIONS, ANOMALY_TURNS_PER_SESSION, + ANOMALY_TURN_ATTEMPTS, ) report = summarize(turns, ANOMALY_SESSIONS * ANOMALY_TURNS_PER_SESSION) failures = tuple(turn.failure for turn in turns if turn.failure)