From 09a98f55052f03d78aeea4a6fe6e3916078b8220 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Fri, 7 Aug 2026 19:36:30 -0700 Subject: [PATCH] test(e2e): settle control-plane writes across every replica, not just one The suite already waits for a new model or agent to become servable before handing it back, but that wait returns on the first successful read. Every request opens a fresh connection (e2e_http calls requests.* with no Session), so a load-balanced Service routes each one independently: one successful read proves one replica converged, and the caller's next request re-rolls and can land on a replica that has not reloaded yet. At replicaCount: 2 this surfaced as 30 failures on a SHA that is green at 1 replica -- 400 "Invalid model name passed", 404 "Guardrail not found", "no healthy deployments for this model", and a /model/info listing that contained one of two models created moments apart. Add PROPAGATION_TIMEOUT (default 15s, override E2E_PROPAGATION_TIMEOUT) and settle_propagation(), sized off the proxy's proxy_config_reload_interval_seconds (30s by default, 7s on the e2e stack) plus margin, and settle after every control-plane create whose object the suite then uses: - ProxyClient.create_model and A2AClient.register_agent, after their existing polls -- the poll still fails loudly if the object never appears at all - GuardrailsClient.register, which had no barrier; create_content_filter_guardrail and create_bedrock_guardrail now route through it instead of POSTing directly - the guardrail creates in mcp_client and logging_client - the vertex passthrough model, whose body cannot go through create_model Left alone: the /model/new calls that assert a 403 or read back a status code, since they never use the model. --- tests/e2e/a2a/a2a_client.py | 7 +- tests/e2e/e2e_config.py | 26 +++++++ tests/e2e/guardrails/guardrails_client.py | 76 ++++++++----------- .../test_vertex_passthrough_e2e.py | 14 +++- tests/e2e/logging/logging_client.py | 3 +- tests/e2e/mcp/mcp_client.py | 5 +- tests/e2e/proxy_client.py | 22 ++++-- 7 files changed, 98 insertions(+), 55 deletions(-) diff --git a/tests/e2e/a2a/a2a_client.py b/tests/e2e/a2a/a2a_client.py index e83897025a3..605dd8fb7e5 100644 --- a/tests/e2e/a2a/a2a_client.py +++ b/tests/e2e/a2a/a2a_client.py @@ -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: diff --git a/tests/e2e/e2e_config.py b/tests/e2e/e2e_config.py index 34770bc596b..277478eebaf 100644 --- a/tests/e2e/e2e_config.py +++ b/tests/e2e/e2e_config.py @@ -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) diff --git a/tests/e2e/guardrails/guardrails_client.py b/tests/e2e/guardrails/guardrails_client.py index 93861d19922..85964529ada 100644 --- a/tests/e2e/guardrails/guardrails_client.py +++ b/tests/e2e/guardrails/guardrails_client.py @@ -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( diff --git a/tests/e2e/llm_translation/test_vertex_passthrough_e2e.py b/tests/e2e/llm_translation/test_vertex_passthrough_e2e.py index b6d90b7f6a2..5e9c9f614e5 100644 --- a/tests/e2e/llm_translation/test_vertex_passthrough_e2e.py +++ b/tests/e2e/llm_translation/test_vertex_passthrough_e2e.py @@ -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: diff --git a/tests/e2e/logging/logging_client.py b/tests/e2e/logging/logging_client.py index 7053bd1dfd2..d76f7b356b2 100644 --- a/tests/e2e/logging/logging_client.py +++ b/tests/e2e/logging/logging_client.py @@ -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: diff --git a/tests/e2e/mcp/mcp_client.py b/tests/e2e/mcp/mcp_client.py index 33ec557c339..73453478e5a 100644 --- a/tests/e2e/mcp/mcp_client.py +++ b/tests/e2e/mcp/mcp_client.py @@ -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( diff --git a/tests/e2e/proxy_client.py b/tests/e2e/proxy_client.py index 6c6b948e29c..2627bdb8038 100644 --- a/tests/e2e/proxy_client.py +++ b/tests/e2e/proxy_client.py @@ -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: