diff --git a/tests/e2e/proxy_client.py b/tests/e2e/proxy_client.py index 6c6b948e29c..87693175e62 100644 --- a/tests/e2e/proxy_client.py +++ b/tests/e2e/proxy_client.py @@ -75,12 +75,95 @@ from transport import HttpTransport, SplitTransport, Transport RowsPredicate = Callable[[list[SpendLogRow]], bool] +# After /model/new, poll the data plane until the model is listed (or fail). +# Shorter than poll_timeout (spend/log read-backs ~120s); longer than a single +# request. 40s is the harness middle ground: happy path returns on the first +# poll, a stuck reload fails in under a minute instead of two. +MODEL_SERVABLE_TIMEOUT = 40.0 +MODEL_SERVABLE_INTERVAL = 2.0 +# Cap each /v1/models poll so one slow request cannot outlast the budget. +# Clamped further to remaining deadline inside await_servable. +MODEL_SERVABLE_REQUEST_TIMEOUT = 5.0 + + +@dataclass(frozen=True, slots=True) +class Servable: + """The data plane listed the model within the deadline.""" + + +@dataclass(frozen=True, slots=True) +class NotServable: + """The deadline passed without the data plane listing the model. + + `last_result` is the final /v1/models read, so the caller can tell "the proxy + answered but omitted the model" (propagation) from "the read itself failed" + (network/auth) when reporting.""" + + last_result: Result[ModelsListResponse] | None + + +ServableOutcome = Servable | NotServable + + +def await_servable( + list_models: Callable[[float], Result[ModelsListResponse]], + *, + model_name: str, + timeout: float, + interval: float, + request_timeout: float, + now: Callable[[], float], + sleep: Callable[[float], None], +) -> ServableOutcome: + """Poll `list_models` until the data plane lists `model_name` or `timeout` elapses. + + `list_models` receives the per-poll request timeout, clamped to the remaining + deadline so a slow final poll cannot overrun the overall budget. Clock and sleep + are injected so this is exercised without wall-clock waits. Always polls at least + once when the loop starts with a positive budget.""" + deadline = now() + timeout + last_result: Result[ModelsListResponse] | None = None + while True: + remaining = deadline - now() + if remaining <= 0 and last_result is not None: + return NotServable(last_result=last_result) + poll_timeout = min(request_timeout, remaining) if remaining > 0 else request_timeout + last_result = list_models(poll_timeout) + if isinstance(last_result, Success) and any( + entry.id == model_name for entry in last_result.data.data + ): + return Servable() + if now() + interval >= deadline: + return NotServable(last_result=last_result) + sleep(interval) + + +def servable_timeout_message( + *, + model_name: str, + timeout: float, + last_result: Result[ModelsListResponse] | None, +) -> str: + last_error = ( + f"; last /v1/models poll did not succeed: {last_result}" + if last_result is not None and not isinstance(last_result, Success) + else "" + ) + return ( + f"model {model_name!r} was created but never became servable on the data " + f"plane within {timeout}s of /model/new (control/data-plane propagation or " + f"STORE_MODEL_IN_DB reload issue){last_error}" + ) + @dataclass(frozen=True, slots=True) class ProxyClient: transport: Transport poll_timeout: float = 120.0 poll_interval: float = 5.0 + model_servable_timeout: float = MODEL_SERVABLE_TIMEOUT + model_servable_interval: float = MODEL_SERVABLE_INTERVAL + model_servable_request_timeout: float = MODEL_SERVABLE_REQUEST_TIMEOUT # ---- keys / customers (satisfies lifecycle.ResourceClient) ---------- @@ -167,7 +250,12 @@ class ProxyClient: this returns can race the reload and 400 with "Invalid model name passed". We therefore poll the data-plane /v1/models until the model appears before handing back, so callers can invoke it immediately. In the monolithic case - it is already present on the first poll, so this adds one request.""" + it is already present on the first poll, so this adds one request. + + The wait is bounded by `model_servable_timeout` rather than the much longer + `poll_timeout` used for batched read-backs, so a stuck reload fails in under + a minute instead of two. Happy path still returns as soon as /v1/models lists + the model (usually the first poll).""" model_id = unwrap( self.transport.post( "/model/new", @@ -185,32 +273,34 @@ class ProxyClient: def _await_model_servable(self, model_name: str) -> None: """Block until the data plane lists `model_name`, or fail loudly if it does - not within poll_timeout (a real propagation/config problem, surfaced here - instead of as a downstream "Invalid model name passed").""" - deadline = time.monotonic() + self.poll_timeout - last_result: Result[ModelsListResponse] | None = None - while time.monotonic() < deadline: - last_result = self.transport.get( + not within model_servable_timeout (a real propagation/config problem, + surfaced here instead of as a downstream "Invalid model name passed").""" + outcome = await_servable( + lambda poll_timeout: self.transport.get( "/v1/models", headers=self.transport.master, params=NoBody(), response_type=ModelsListResponse, - ) - if isinstance(last_result, Success) and any( - entry.id == model_name for entry in last_result.data.data - ): + timeout=poll_timeout, + ), + model_name=model_name, + timeout=self.model_servable_timeout, + interval=self.model_servable_interval, + request_timeout=self.model_servable_request_timeout, + now=time.monotonic, + sleep=time.sleep, + ) + match outcome: + case Servable(): return - time.sleep(self.poll_interval) - last_error = ( - f"; last /v1/models poll did not succeed: {last_result}" - if last_result is not None and not isinstance(last_result, Success) - else "" - ) - raise AssertionError( - f"model {model_name!r} was created but never became servable on the data " - f"plane within {self.poll_timeout}s of /model/new (control/data-plane " - f"propagation or STORE_MODEL_IN_DB reload issue){last_error}" - ) + case NotServable(last_result=last_result): + raise AssertionError( + servable_timeout_message( + model_name=model_name, + timeout=self.model_servable_timeout, + last_result=last_result, + ) + ) def update_model(self, model_id: str, litellm_params: LiteLLMParamsBody) -> None: """Merge `litellm_params` over the deployment `model_id`'s stored params via diff --git a/tests/e2e/test_proxy_client_model_servable.py b/tests/e2e/test_proxy_client_model_servable.py new file mode 100644 index 00000000000..0cc63f882e2 --- /dev/null +++ b/tests/e2e/test_proxy_client_model_servable.py @@ -0,0 +1,190 @@ +"""Harness coverage for the bounded wait after /model/new (no live proxy). + +Model propagation is polled to a deadline so a stuck control/data-plane reload fails +fast instead of stalling every test that creates a model. The clock and sleep are +injected, so these assert the deadline arithmetic without waiting. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from e2e_http import NetworkError, Result, Success +from models import ModelListEntry, ModelsListResponse +from proxy_client import ( + MODEL_SERVABLE_REQUEST_TIMEOUT, + MODEL_SERVABLE_TIMEOUT, + NotServable, + Servable, + await_servable, + servable_timeout_message, +) + + +def _listing(*model_names: str) -> Result[ModelsListResponse]: + return Success( + status_code=200, + data=ModelsListResponse(data=tuple(ModelListEntry(id=name) for name in model_names)), + ) + + +@dataclass(slots=True) +class FakeClock: + """A clock that only advances when the code under test sleeps or a slow poll runs.""" + + seconds: float = 0.0 + slept: list[float] = field(default_factory=list) # mutable-ok: records calls for assertions + + def now(self) -> float: + return self.seconds + + def sleep(self, duration: float) -> None: + self.slept.append(duration) + self.seconds += duration + + +@dataclass(slots=True) +class FakeModelList: + """Returns each queued /v1/models read in turn, repeating the last forever.""" + + responses: tuple[Result[ModelsListResponse], ...] + calls: int = 0 + timeouts: list[float] = field(default_factory=list) # mutable-ok: records call timeouts + + def __call__(self, request_timeout: float) -> Result[ModelsListResponse]: + self.timeouts.append(request_timeout) + response = self.responses[min(self.calls, len(self.responses) - 1)] + self.calls += 1 + return response + + +def test_returns_servable_on_first_listing_without_sleeping() -> None: + clock = FakeClock() + list_models = FakeModelList(responses=(_listing("my-model"),)) + + outcome = await_servable( + list_models, + model_name="my-model", + timeout=40.0, + interval=2.0, + request_timeout=5.0, + now=clock.now, + sleep=clock.sleep, + ) + + assert outcome == Servable() + assert list_models.calls == 1 + assert list_models.timeouts == [5.0] + assert clock.slept == [] + + +def test_polls_until_the_model_appears() -> None: + clock = FakeClock() + list_models = FakeModelList(responses=(_listing("other"), _listing("other"), _listing("other", "my-model"))) + + outcome = await_servable( + list_models, + model_name="my-model", + timeout=40.0, + interval=2.0, + request_timeout=5.0, + now=clock.now, + sleep=clock.sleep, + ) + + assert outcome == Servable() + assert list_models.calls == 3 + assert clock.seconds == 4.0 + + +def test_gives_up_at_the_deadline_rather_than_polling_forever() -> None: + clock = FakeClock() + list_models = FakeModelList(responses=(_listing("other"),)) + + outcome = await_servable( + list_models, + model_name="my-model", + timeout=10.0, + interval=2.0, + request_timeout=5.0, + now=clock.now, + sleep=clock.sleep, + ) + + assert isinstance(outcome, NotServable) + assert clock.seconds == 8.0 + assert list_models.calls == 5 + assert list_models.timeouts == [5.0, 5.0, 5.0, 4.0, 2.0] + + +def test_does_not_wait_past_the_overall_budget() -> None: + clock = FakeClock() + + outcome = await_servable( + FakeModelList(responses=(_listing("other"),)), + model_name="my-model", + timeout=MODEL_SERVABLE_TIMEOUT, + interval=2.0, + request_timeout=MODEL_SERVABLE_REQUEST_TIMEOUT, + now=clock.now, + sleep=clock.sleep, + ) + + assert isinstance(outcome, NotServable) + assert clock.seconds <= MODEL_SERVABLE_TIMEOUT + + +def test_clamps_request_timeout_to_remaining_deadline() -> None: + """A slow final poll must not receive the full request cap when less budget remains. + + Without the clamp, remaining=3 and cap=5 lets the transport block for 5s and the + overall wait overruns model_servable_timeout by up to ~cap seconds. + """ + clock = FakeClock() + timeouts: list[float] = [] + + def list_models(request_timeout: float) -> Result[ModelsListResponse]: + timeouts.append(request_timeout) + clock.seconds += request_timeout + return _listing("other") + + outcome = await_servable( + list_models, + model_name="my-model", + timeout=10.0, + interval=2.0, + request_timeout=5.0, + now=clock.now, + sleep=clock.sleep, + ) + + assert isinstance(outcome, NotServable) + assert timeouts[0] == 5.0 + assert any(timeout < 5.0 for timeout in timeouts) + assert timeouts[-1] == 3.0 + assert clock.seconds <= 10.0 + + +def test_reports_a_failed_read_distinctly_from_a_missing_model() -> None: + clock = FakeClock() + unreachable: Result[ModelsListResponse] = NetworkError(message="connection refused") + + outcome = await_servable( + FakeModelList(responses=(unreachable,)), + model_name="my-model", + timeout=1.0, + interval=0.5, + request_timeout=5.0, + now=clock.now, + sleep=clock.sleep, + ) + + assert outcome == NotServable(last_result=unreachable) + message = servable_timeout_message(model_name="my-model", timeout=1.0, last_result=unreachable) + assert "connection refused" in message + + listed_without_model = _listing("other") + propagation_message = servable_timeout_message( + model_name="my-model", timeout=1.0, last_result=listed_without_model + ) + assert "did not succeed" not in propagation_message diff --git a/tests/e2e/transport.py b/tests/e2e/transport.py index a6adf83ed1f..27b11befc8e 100644 --- a/tests/e2e/transport.py +++ b/tests/e2e/transport.py @@ -58,6 +58,7 @@ class Transport(Protocol): headers: BaseModel, params: BaseModel, response_type: type[R], + timeout: float | None = None, ) -> Result[R]: ... def delete[R: BaseModel]( @@ -136,13 +137,16 @@ class HttpTransport: headers: BaseModel, params: BaseModel, response_type: type[R], + timeout: float | None = None, ) -> Result[R]: + """`timeout` overrides the transport-wide request_timeout for this call, for + pollers whose own deadline is shorter than it.""" return e2e_http.get( self._url(path), headers=headers, params=params, response_type=response_type, - timeout=self.request_timeout, + timeout=self.request_timeout if timeout is None else timeout, ) def delete[R: BaseModel]( @@ -336,9 +340,14 @@ class SplitTransport: headers: BaseModel, params: BaseModel, response_type: type[R], + timeout: float | None = None, ) -> Result[R]: return self._route(path).get( - path, headers=headers, params=params, response_type=response_type + path, + headers=headers, + params=params, + response_type=response_type, + timeout=timeout, ) def delete[R: BaseModel](