scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

fix(e2e): wait one default DB reload interval of continuous listing

Browse source 
create_model returned after the first /v1/models hit that listed the model,
so chat could still land on a cold gateway worker (numWorkers>1 / peer pod)
and 400 Invalid model name. Require continuous listing for the product
default add_deployment interval (30s) after first sight so every worker has
synced from the DB; first listing still bounded at 40s
This commit is contained in:
mubashir1osmani 2026-07-28 17:09:20 -07:00
parent c082a0e648
commit 7d1ee2ff86
2 changed files with 148 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
tests/e2e/proxy_client.py
View file

@ -75,14 +75,17 @@ 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.
# After /model/new, the control-plane writer reloads itself immediately, but every
# other gateway worker (and peer pod) only picks the model up on its add_deployment
# job. That job runs every proxy_config_reload_interval_seconds (product default 30).
# A single /v1/models hit can land on a hot worker while the next /chat hits a cold
# one ("Invalid model name"). Wait for first listing within MODEL_SERVABLE_TIMEOUT,
# then require continuous listing for MODEL_SERVABLE_DB_SYNC_SECONDS (the default
# reload interval) so every worker has had a chance to sync from the DB.
MODEL_SERVABLE_TIMEOUT = 40.0
MODEL_SERVABLE_DB_SYNC_SECONDS = 30.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.
# Cap each /v1/models poll so one slow request cannot outlast the remaining budget.
MODEL_SERVABLE_REQUEST_TIMEOUT = 5.0
@ -112,29 +115,55 @@ def await_servable(
timeout: float,
interval: float,
request_timeout: float,
db_sync_seconds: float,
now: Callable[[], float],
sleep: Callable[[float], None],
) -> ServableOutcome:
"""Poll `list_models` until the data plane lists `model_name` or `timeout` elapses.
"""Poll until `model_name` is listed long enough for every worker to DB-sync.
`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
First listing must happen within `timeout`. After that, the model must stay
listed continuously for `db_sync_seconds` (any miss resets the continuous
window). `db_sync_seconds=0` returns on the first listing. Each poll's request
timeout is clamped to the remaining budget. Clock and sleep are injected."""
started = now()
first_seen_at: float | None = None
last_result: Result[ModelsListResponse] | None = None
while True:
remaining = deadline - now()
t = now()
if first_seen_at is None:
deadline = started + timeout
else:
deadline = first_seen_at + db_sync_seconds
remaining = deadline - t
if remaining <= 0 and last_result is not None:
if first_seen_at is not None and db_sync_seconds <= 0:
return Servable()
if first_seen_at is not None and t - first_seen_at >= db_sync_seconds:
return Servable()
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(
listed = isinstance(last_result, Success) and any(
entry.id == model_name for entry in last_result.data.data
):
)
t = now()
if not listed:
first_seen_at = None
elif first_seen_at is None:
first_seen_at = t
if db_sync_seconds <= 0:
return Servable()
elif t - first_seen_at >= db_sync_seconds:
return Servable()
if now() + interval >= deadline:
return NotServable(last_result=last_result)
if first_seen_at is None:
if now() + interval >= started + timeout:
return NotServable(last_result=last_result)
elif now() + interval >= first_seen_at + db_sync_seconds:
# Final stretch: sleep only the remainder of the continuous window.
remainder = first_seen_at + db_sync_seconds - now()
if remainder > 0:
sleep(remainder)
continue
sleep(interval)
@ -142,6 +171,7 @@ def servable_timeout_message(
*,
model_name: str,
timeout: float,
db_sync_seconds: float,
last_result: Result[ModelsListResponse] | None,
) -> str:
last_error = (
@ -151,7 +181,8 @@ def servable_timeout_message(
)
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"plane within {timeout}s of first listing (plus {db_sync_seconds}s continuous "
f"DB sync) after /model/new (control/data-plane propagation or "
f"STORE_MODEL_IN_DB reload issue){last_error}"
)
@ -162,6 +193,7 @@ class ProxyClient:
poll_timeout: float = 120.0
poll_interval: float = 5.0
model_servable_timeout: float = MODEL_SERVABLE_TIMEOUT
model_servable_db_sync_seconds: float = MODEL_SERVABLE_DB_SYNC_SECONDS
model_servable_interval: float = MODEL_SERVABLE_INTERVAL
model_servable_request_timeout: float = MODEL_SERVABLE_REQUEST_TIMEOUT
@ -252,10 +284,10 @@ class ProxyClient:
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.
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)."""
First listing must arrive within `model_servable_timeout` (not the longer
spend `poll_timeout`). The model must then stay listed for
`model_servable_db_sync_seconds` (product default DB reload interval) so every
gateway worker has run add_deployment before callers use the model."""
model_id = unwrap(
self.transport.post(
"/model/new",
@ -272,9 +304,10 @@ class ProxyClient:
return model_id
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 model_servable_timeout (a real propagation/config problem,
surfaced here instead of as a downstream "Invalid model name passed")."""
"""Block until the data plane lists `model_name` long enough for DB sync.
Fails if first listing misses model_servable_timeout, or if continuous listing
for model_servable_db_sync_seconds never holds (multi-worker / peer reload)."""
outcome = await_servable(
lambda poll_timeout: self.transport.get(
"/v1/models",
@ -287,6 +320,7 @@ class ProxyClient:
timeout=self.model_servable_timeout,
interval=self.model_servable_interval,
request_timeout=self.model_servable_request_timeout,
db_sync_seconds=self.model_servable_db_sync_seconds,
now=time.monotonic,
sleep=time.sleep,
)
@ -298,6 +332,7 @@ class ProxyClient:
servable_timeout_message(
model_name=model_name,
timeout=self.model_servable_timeout,
db_sync_seconds=self.model_servable_db_sync_seconds,
last_result=last_result,
)
)

125
tests/e2e/test_proxy_client_model_servable.py
View file

@ -1,8 +1,9 @@
"""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.
create_model must wait for the product default DB reload interval of continuous
listing so multi-worker gateways finish add_deployment before callers use the
model. Clock and sleep are injected so these assert the deadline arithmetic
without wall-clock waits.
"""
from __future__ import annotations
@ -12,6 +13,7 @@ from dataclasses import dataclass, field
from e2e_http import NetworkError, Result, Success
from models import ModelListEntry, ModelsListResponse
from proxy_client import (
MODEL_SERVABLE_DB_SYNC_SECONDS,
MODEL_SERVABLE_REQUEST_TIMEOUT,
MODEL_SERVABLE_TIMEOUT,
NotServable,
@ -30,8 +32,6 @@ def _listing(*model_names: str) -> Result[ModelsListResponse]:
@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
@ -45,8 +45,6 @@ class FakeClock:
@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
@ -58,7 +56,7 @@ class FakeModelList:
return response
def test_returns_servable_on_first_listing_without_sleeping() -> None:
def test_returns_on_first_listing_when_db_sync_is_zero() -> None:
clock = FakeClock()
list_models = FakeModelList(responses=(_listing("my-model"),))
@ -68,17 +66,64 @@ def test_returns_servable_on_first_listing_without_sleeping() -> None:
timeout=40.0,
interval=2.0,
request_timeout=5.0,
db_sync_seconds=0.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:
def test_requires_continuous_listing_for_default_db_sync_interval() -> 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,
db_sync_seconds=MODEL_SERVABLE_DB_SYNC_SECONDS,
now=clock.now,
sleep=clock.sleep,
)
assert outcome == Servable()
assert clock.seconds >= MODEL_SERVABLE_DB_SYNC_SECONDS
assert list_models.calls >= 2
def test_resets_db_sync_window_when_a_poll_misses() -> None:
clock = FakeClock()
list_models = FakeModelList(
responses=(
_listing("my-model"),
_listing("my-model"),
_listing("other"),
_listing("my-model"),
)
)
outcome = await_servable(
list_models,
model_name="my-model",
timeout=40.0,
interval=2.0,
request_timeout=5.0,
db_sync_seconds=6.0,
now=clock.now,
sleep=clock.sleep,
)
assert outcome == Servable()
assert list_models.calls >= 4
assert clock.seconds >= 6.0
def test_polls_until_the_model_first_appears() -> None:
clock = FakeClock()
list_models = FakeModelList(responses=(_listing("other"), _listing("other"), _listing("other", "my-model")))
@ -88,6 +133,7 @@ def test_polls_until_the_model_appears() -> None:
timeout=40.0,
interval=2.0,
request_timeout=5.0,
db_sync_seconds=0.0,
now=clock.now,
sleep=clock.sleep,
)
@ -97,7 +143,7 @@ def test_polls_until_the_model_appears() -> None:
assert clock.seconds == 4.0
def test_gives_up_at_the_deadline_rather_than_polling_forever() -> None:
def test_gives_up_if_first_listing_never_arrives() -> None:
clock = FakeClock()
list_models = FakeModelList(responses=(_listing("other"),))
@ -107,6 +153,7 @@ def test_gives_up_at_the_deadline_rather_than_polling_forever() -> None:
timeout=10.0,
interval=2.0,
request_timeout=5.0,
db_sync_seconds=30.0,
now=clock.now,
sleep=clock.sleep,
)
@ -114,32 +161,9 @@ def test_gives_up_at_the_deadline_rather_than_polling_forever() -> None:
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] = []
@ -154,6 +178,7 @@ def test_clamps_request_timeout_to_remaining_deadline() -> None:
timeout=10.0,
interval=2.0,
request_timeout=5.0,
db_sync_seconds=30.0,
now=clock.now,
sleep=clock.sleep,
)
@ -161,10 +186,27 @@ def test_clamps_request_timeout_to_remaining_deadline() -> None:
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_does_not_wait_past_first_listing_budget_when_missing() -> 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,
db_sync_seconds=MODEL_SERVABLE_DB_SYNC_SECONDS,
now=clock.now,
sleep=clock.sleep,
)
assert isinstance(outcome, NotServable)
assert clock.seconds <= MODEL_SERVABLE_TIMEOUT
def test_reports_a_failed_read_distinctly_from_a_missing_model() -> None:
clock = FakeClock()
unreachable: Result[ModelsListResponse] = NetworkError(message="connection refused")
@ -175,16 +217,25 @@ def test_reports_a_failed_read_distinctly_from_a_missing_model() -> None:
timeout=1.0,
interval=0.5,
request_timeout=5.0,
db_sync_seconds=0.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)
message = servable_timeout_message(
model_name="my-model",
timeout=1.0,
db_sync_seconds=0.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
model_name="my-model",
timeout=1.0,
db_sync_seconds=30.0,
last_result=listed_without_model,
)
assert "did not succeed" not in propagation_message