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 35

Merge pull request #36247 from BerriAI/litellm_/e2e-propagation-barrier

Browse source 
test(e2e): settle control-plane writes across every replica, not just one
This commit is contained in:
yuneng-jiang 2026-08-07 19:51:02 -07:00 • committed by GitHub
commit 5d2b695f00
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 98 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
tests/e2e/a2a/a2a_client.py
View file

@ -17,6 +17,7 @@ from dataclasses import dataclass
from pydantic import BaseModel, ConfigDict, Field
from e2e_config import settle_propagation
from e2e_http import NoBody, Result, Success, get_external, is_ok
from proxy_client import ProxyClient
@ -298,7 +299,9 @@ class A2AClient:
the next DB reload. A card read or message/send issued the instant this
returns can therefore 404 on the agent it just created. Waiting here keeps
every caller from having to poll, the same way ProxyClient.create_model
waits for a new model to become servable.
waits for a new model to become servable -- including the settle that
covers the other replicas, since one successful card read only proves the
replica that answered it has the agent.
"""
result = self.proxy.transport.post(
"/v1/agents",
@ -307,7 +310,9 @@ class A2AClient:
response_type=AgentResponse,
)
if isinstance(result, Success):
written_at = time.monotonic()
self._await_agent_servable(result.data.agent_id)
settle_propagation(written_at)
return result
def _await_agent_servable(self, agent_id: str) -> None:

26
tests/e2e/e2e_config.py
View file

@ -7,6 +7,7 @@ environment so the same tests run against localhost or a deployed proxy.
from __future__ import annotations
import os
import time
import uuid
from pathlib import Path
@ -75,6 +76,18 @@ POLL_TIMEOUT = float(os.environ.get("E2E_POLL_TIMEOUT", "120"))
POLL_INTERVAL = float(os.environ.get("E2E_POLL_INTERVAL", "5"))
REQUEST_TIMEOUT = float(os.environ.get("E2E_REQUEST_TIMEOUT", "60"))
# How long a control-plane write (/model/new, /guardrails, /v1/agents) may take to
# reach EVERY replica. Distinct from POLL_TIMEOUT, which is sized for spend-row
# flush; this one is sized for the proxy's config reload
# (`proxy_config_reload_interval_seconds`, 30s by default and 7s on the e2e stack)
# plus margin.
#
# The barriers below wait this out instead of returning on first sight, because a
# single successful read only proves ONE replica converged: every request opens a
# fresh connection, so a load-balanced Service routes each one independently and
# the next call re-rolls. See ProxyClient._await_model_servable.
PROPAGATION_TIMEOUT = float(os.environ.get("E2E_PROPAGATION_TIMEOUT", "15"))
EXPECT_RUST = os.environ.get("E2E_EXPECT_RUST", "").strip().lower() in ("1", "true", "yes")
# Deliberately modest concurrency. The suite shares its proxy with every other
@ -137,3 +150,16 @@ def unique_marker() -> str:
"""A short unique token per call/run, so concurrent runs and the shared
response cache never collide on prompts, tags, or customer ids."""
return uuid.uuid4().hex[:12]
def settle_propagation(written_at: float) -> None:
"""Block until PROPAGATION_TIMEOUT has elapsed since `written_at`, a
`time.monotonic()` stamp taken the moment a control-plane write returned.
Callers that already polled for the object still need this: the poll proves one
replica has it, not all of them. Waiting out the config-reload budget is what
makes the object safe to use on whichever replica the next request lands on.
"""
remaining = PROPAGATION_TIMEOUT - (time.monotonic() - written_at)
if remaining > 0:
time.sleep(remaining)

76
tests/e2e/guardrails/guardrails_client.py
View file

@ -11,7 +11,7 @@ from typing import Literal
from pydantic import BaseModel
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT, unique_marker
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT, settle_propagation, unique_marker
from e2e_http import NoBody, Result, Success, unwrap
from lifecycle import ResourceManager
from models import (
@ -104,25 +104,14 @@ class GuardrailsClient:
proxy: ProxyClient
def create_content_filter_guardrail(self, name: str, blocked_keyword: str) -> str:
return unwrap(
self.proxy.transport.post(
"/guardrails",
headers=self.proxy.transport.master,
json=GuardrailCreateBody(
guardrail=GuardrailSpecBody(
guardrail_name=name,
litellm_params=ContentFilterParamsBody(
mode="pre_call",
default_on=True,
blocked_words=[
BlockedWordBody(keyword=blocked_keyword, action="BLOCK")
],
),
)
),
response_type=GuardrailCreateResponse,
)
).guardrail_id
return self.register(
name,
ContentFilterParamsBody(
mode="pre_call",
default_on=True,
blocked_words=[BlockedWordBody(keyword=blocked_keyword, action="BLOCK")],
),
)
def create_bedrock_guardrail(
self,
@ -141,24 +130,15 @@ class GuardrailsClient:
test takes out whatever else is running. Callers select the guardrail
per-request instead, which keeps the blast radius to the test that wants it.
"""
return unwrap(
self.proxy.transport.post(
"/guardrails",
headers=self.proxy.transport.master,
json=GuardrailCreateBody(
guardrail=GuardrailSpecBody(
guardrail_name=name,
litellm_params=BedrockGuardrailParamsBody(
mode="pre_call",
default_on=default_on,
guardrailIdentifier=identifier,
guardrailVersion=version,
),
)
),
response_type=GuardrailCreateResponse,
)
).guardrail_id
return self.register(
name,
BedrockGuardrailParamsBody(
mode="pre_call",
default_on=default_on,
guardrailIdentifier=identifier,
guardrailVersion=version,
),
)
def create_backend_model(self, resources: ResourceManager, prefix: str = "e2e-guard-backend") -> str:
"""Register a gemini chat deployment for a guardrail test to run against
@ -174,11 +154,19 @@ class GuardrailsClient:
return model_name
def register(self, name: str, params: GuardrailParamsBody) -> str:
"""Register any guardrail via POST /guardrails and return its id. New
built-ins register with default_on=False and are opted into per request
via the chat body's `guardrails` list, so one guardrail under test never
intercepts unrelated traffic on the shared proxy."""
return unwrap(
"""Register any guardrail via POST /guardrails and return its id, once every
replica can be expected to serve it. New built-ins register with
default_on=False and are opted into per request via the chat body's
`guardrails` list, so one guardrail under test never intercepts unrelated
traffic on the shared proxy.
/guardrails is a control-plane route and guardrails reach the data plane on
the config reload, so a request naming this guardrail the instant the POST
returns can 404 with "Guardrail not found" on a replica that has not
reloaded. There is no data-plane read that lists guardrails, so unlike
ProxyClient.create_model this settles on the propagation budget alone with
nothing to poll first."""
guardrail_id = unwrap(
self.proxy.transport.post(
"/guardrails",
headers=self.proxy.transport.master,
@ -188,6 +176,8 @@ class GuardrailsClient:
response_type=GuardrailCreateResponse,
)
).guardrail_id
settle_propagation(time.monotonic())
return guardrail_id
def delete_guardrail(self, guardrail_id: str) -> None:
_ = self.proxy.transport.delete(

14
tests/e2e/llm_translation/test_vertex_passthrough_e2e.py
View file

@ -23,11 +23,12 @@ model, spend > 0), correlated by the x-litellm-call-id header.
"""
import os
import time
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_config import settle_propagation, unique_marker
from e2e_http import NoBody, require_successful_call, unwrap
from lifecycle import ResourceManager
from models import SpendLogRow
@ -90,7 +91,14 @@ class _ModelDeleteBody(BaseModel):
def _add_vertex_passthrough_model(
client: PassthroughClient, model_name: str, project: str, credentials: str
) -> str:
return unwrap(
"""Register the passthrough deployment and settle before the caller uses it.
This body carries `use_in_pass_through` and a pinned `model_info.id`, so it
cannot go through ProxyClient.create_model -- but it needs that helper's
propagation settle just the same, or the passthrough call can land on a replica
that has not reloaded yet.
"""
model_id = unwrap(
client.proxy.transport.post(
"/model/new",
headers=client.proxy.transport.master,
@ -108,6 +116,8 @@ def _add_vertex_passthrough_model(
response_type=_ModelNewResponse,
)
).model_id
settle_propagation(time.monotonic())
return model_id
def _delete_model(client: PassthroughClient, model_id: str) -> None:

3
tests/e2e/logging/logging_client.py
View file

@ -23,7 +23,7 @@ from typing import Callable, Literal
import pytest
from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT, settle_propagation
from proxy_client import ProxyClient
from e2e_http import (
URL,
@ -409,6 +409,7 @@ class LoggingClient:
)
guardrail_id = response.guardrail_id
assert guardrail_id, f"create guardrail returned no id: {response!r}"
settle_propagation(time.monotonic())
return guardrail_id
def delete_guardrail(self, guardrail_id: str) -> None:

5
tests/e2e/mcp/mcp_client.py
View file

@ -18,6 +18,7 @@ from dataclasses import dataclass
from pydantic import BaseModel, ConfigDict, Field, RootModel
from e2e_config import settle_propagation
from e2e_http import Headers, NoBody, Result, Success, UnknownApiError, unwrap
from models import KeyGenerateBody, ObjectPermission
from proxy_client import ProxyClient
@ -330,7 +331,7 @@ class McpClient:
tool-call hook (pre_mcp_call) and blocks a single keyword. The keyword is
unique per test, so default_on only ever intercepts this test's own
banned tool call on the shared proxy."""
return unwrap(
guardrail_id = unwrap(
self.proxy.transport.post(
"/guardrails",
headers=self.proxy.transport.master,
@ -345,6 +346,8 @@ class McpClient:
response_type=GuardrailCreateResponse,
)
).guardrail_id
settle_propagation(time.monotonic())
return guardrail_id
def delete_guardrail(self, guardrail_id: str) -> None:
_ = self.proxy.transport.delete(

22
tests/e2e/proxy_client.py
View file

@ -70,6 +70,7 @@ from e2e_config import (
POLL_TIMEOUT,
PROXY_BASE_URL,
REQUEST_TIMEOUT,
settle_propagation,
)
from transport import HttpTransport, SplitTransport, Transport
@ -161,13 +162,18 @@ class ProxyClient:
"""Register a deployment under `model_name` and return its proxy-assigned
model_id, once the model is actually servable on the data plane.
/model/new is a control-plane route; in a split control/data-plane
deployment the gateway (data plane, which serves /chat, /ocr, ...) only
picks the new model up on its next DB reload, so a call issued the instant
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."""
/model/new is a control-plane route; the data plane (which serves /chat,
/ocr, ...) only picks the new model up on its next DB reload, so a call
issued the instant this returns can race the reload and 400 with "Invalid
model name passed". We poll the data-plane /v1/models until the model
appears, then settle for the remainder of the propagation budget.
Both steps are needed, and the second is the one that matters at >1 replica.
The poll proves *a* replica is serving the model; it cannot prove they all
are, because every request opens a fresh connection and a load-balanced
Service routes each one independently -- so the caller's next request
re-rolls and can land on a replica that has not reloaded yet. Waiting out
PROPAGATION_TIMEOUT is what makes the model safe to use anywhere."""
model_id = unwrap(
self.transport.post(
"/model/new",
@ -180,7 +186,9 @@ class ProxyClient:
response_type=ModelNewResponse,
)
).model_id
written_at = time.monotonic()
self._await_model_servable(model_name)
settle_propagation(written_at)
return model_id
def _await_model_servable(self, model_name: str) -> None: