test(e2e): run the memory cell alone on the shared stack (#42518)

* test(e2e): run the memory cell alone on the shared stack

* test(e2e): hold the stack lock for every collected test, marker or not

* test(e2e): prove the stack lock's reader sharing and writer preference across processes

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

tests/e2e/AGENTS.md

@@ -97,7 +97,7 @@ Each suite provides its own `client` fixture (see `llm_translation/passthrough_c
 
 Request and response bodies are typed pydantic models in `models.py`; only the fields a test reads are modelled, and nothing passes raw dicts. Outcomes come back as a `Result[R]` tagged union (`Success`, `NetworkError`, `UnauthorizedError`, `RateLimitedError`, `ValidationError`, `UnknownApiError`). Handle them with `match`, or call `unwrap(...)` when a non-success should fail the test. The harness hard-fails and never skips: a test marked `e2e` fails when no proxy answers its liveness probe, and once a request reaches the proxy any wrong behavior is likewise a hard failure, so a missing proxy turns the run red instead of being mistaken for a pass
 
-Mark live tests with `@pytest.mark.e2e` (on the class or the module). Use `scoped_key` for a fresh all-models key that auto-deletes, `resources` when you need to create and tear down more than a key, and `unique_marker()` from `e2e_config` to keep prompts, tags, and customer ids from colliding across concurrent runs and the shared response cache
+Mark live tests with `@pytest.mark.e2e` (on the class or the module). Add `@pytest.mark.quiet_stack` to a test that measures the proxy itself (RSS, latency): the shared stack lock in `stack_lock.py` then runs it while no other test on the host is hitting the stack, marked or not, so the reading depends only on the test's own traffic. Use `scoped_key` for a fresh all-models key that auto-deletes, `resources` when you need to create and tear down more than a key, and `unique_marker()` from `e2e_config` to keep prompts, tags, and customer ids from colliding across concurrent runs and the shared response cache
 
 ## Record and replay fixtures
 

tests/e2e/conftest.py

@@ -51,6 +51,7 @@ from models import TeamNewBody, UserNewBody, UserNewResponse
 from provider_cache_routing import LIVE_PROVIDER_REQUIRED
 from provider_edge import replay_leftover_error
 from proxy_client import ProxyClient, build_proxy_client
+from stack_lock import stack_lock
 
 _E2E_TEST_RAN = pytest.StashKey[bool]()
 _CALL_PASSED = pytest.StashKey[bool]()
@@ -148,6 +149,11 @@ def pytest_configure(config: pytest.Config) -> None:
         "redis_chaos: load test that pauses the proxy's Redis outright mid-run; needs a proxy booted from "
         "gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set",
     )
+    config.addinivalue_line(
+        "markers",
+        "quiet_stack: measures the proxy itself, so it runs while no other test on this host is hitting the stack; "
+        "every other test waits for it to finish",
+    )
     config.addinivalue_line(
         "markers",
         "mcp_oauth_live: real Linear OAuth consent via a captured browser session; deselected unless "
@@ -172,9 +178,7 @@ def pytest_sessionstart(session: pytest.Session) -> None:
     """Abort before collection when E2E_FIXTURE_MODE can never work: an unknown
     mode value, or replay against a missing, unreadable, or stale bundle (the
     stale message names the bundle's age). Live and record modes pass through."""
-    reason = fixture_mode_collection_error(
-        FIXTURE_MODE_RAW, FIXTURE_DIR, now=datetime.now(timezone.utc)
-    )
+    reason = fixture_mode_collection_error(FIXTURE_MODE_RAW, FIXTURE_DIR, now=datetime.now(timezone.utc))
     if reason is not None:
         raise pytest.UsageError(reason)
 
@@ -267,6 +271,12 @@ def _proxy_fail_reason() -> str | None:
     return None
 
 
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_protocol(item: pytest.Item, nextitem: pytest.Item | None) -> Generator[None, object, object]:
+    with stack_lock(exclusive=item.get_closest_marker("quiet_stack") is not None):
+        return (yield)
+
+
 @pytest.hookimpl(tryfirst=True)
 def pytest_runtest_setup(item: pytest.Item) -> None:
     """Hard-fail `e2e`-marked tests unless a proxy answers its liveness probe.
@@ -326,9 +336,7 @@ def pytest_runtest_teardown(item: pytest.Item) -> Generator[None, None, None]:
     LIVE_PROVIDER_REQUIRED.set(False)
     if not item.stash.get(_CALL_PASSED, False):
         return result
-    reason = replay_leftover_error(
-        mode_raw=FIXTURE_MODE_RAW, bundle_dir=FIXTURE_DIR, test_key=item.nodeid
-    )
+    reason = replay_leftover_error(mode_raw=FIXTURE_MODE_RAW, bundle_dir=FIXTURE_DIR, test_key=item.nodeid)
     if reason is not None:
         pytest.fail(reason)
     return result

tests/e2e/pytest.ini

@@ -12,6 +12,7 @@ markers =
     prompt_caching_stack: needs a proxy running with router_settings.optional_pre_call_checks including prompt_caching; deselected unless E2E_PROMPT_CACHING_STACK is set
     cli_determinism: drives the real claude CLI for several seconds; deselected unless E2E_CLI_DETERMINISM is set
     redis_chaos: load test that pauses the proxy's Redis outright mid-run; needs a proxy booted from gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set
+    quiet_stack: measures the proxy itself, so it runs while no other test on this host is hitting the stack; every other test waits for it to finish
     mcp_oauth_live: real Linear OAuth consent via a captured browser session; deselected unless E2E_MCP_OAUTH_LIVE is set
     provider_edge_host: routes provider traffic through the pytest host's edge in every fixture mode, so the gateway must reach the pytest host; deselected unless E2E_PROVIDER_EDGE_HOST_REACHABLE is set
     otel_v2: needs a proxy running with LITELLM_OTEL_V2=true; deselected unless E2E_OTEL_V2 is set

tests/e2e/router/test_reliability_memory_e2e.py

@@ -86,7 +86,7 @@ from models import ChatMessage, RouterSettingsOverride, SpendLogRow
 from proxy_client import ProxyClient
 from reliability_support import chat_override, create_never_benched_refusing_deployment
 
-pytestmark = pytest.mark.e2e
+pytestmark = [pytest.mark.e2e, pytest.mark.quiet_stack]
 
 DEPLOYMENTS_PER_GROUP: Final = 2
 RSS_SAMPLE_CAP: Final = 4 * MEMORY_RSS_SETTLE_SAMPLES

tests/e2e/stack_lock.py

@@ -0,0 +1,45 @@
+"""Cross-process reader/writer lock over the proxy stack every xdist worker shares.
+Every collected test holds it shared, marker or not, since the Claude Code cells and
+other unmarked suites drive the same stack; a `quiet_stack` test holds it exclusive,
+and the `gate` file makes a waiting exclusive holder win over readers that arrive
+after it."""
+
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import tempfile
+from collections.abc import Generator
+from contextlib import ExitStack, contextmanager
+from pathlib import Path
+from typing import Final
+
+from e2e_config import PROXY_BASE_URL
+
+STACK_DIGEST: Final = hashlib.sha256(PROXY_BASE_URL.encode()).hexdigest()[:12]
+LOCK_DIR: Final = Path(tempfile.gettempdir()) / f"litellm-e2e-stack-{STACK_DIGEST}"
+GATE_FILE: Final = LOCK_DIR / "gate"
+STACK_FILE: Final = LOCK_DIR / "stack"
+
+
+@contextmanager
+def _flock(path: Path, operation: int) -> Generator[None]:
+    with path.open("a") as handle:
+        fcntl.flock(handle, operation)
+        try:
+            yield
+        finally:
+            fcntl.flock(handle, fcntl.LOCK_UN)
+
+
+@contextmanager
+def stack_lock(exclusive: bool) -> Generator[None]:
+    LOCK_DIR.mkdir(parents=True, exist_ok=True)
+    if exclusive:
+        with _flock(GATE_FILE, fcntl.LOCK_EX), _flock(STACK_FILE, fcntl.LOCK_EX):
+            yield
+        return
+    with ExitStack() as held:
+        with _flock(GATE_FILE, fcntl.LOCK_SH):
+            held.enter_context(_flock(STACK_FILE, fcntl.LOCK_SH))
+        yield

tests/e2e/test_stack_lock.py

@@ -0,0 +1,117 @@
+"""Cross-process behavior of the stack lock: readers share it, an exclusive holder waits for
+every reader and keeps them out, and a reader arriving behind a waiting exclusive holder
+queues behind it instead of starving it."""
+
+from __future__ import annotations
+
+import fcntl
+import os
+import subprocess
+import sys
+import time
+from contextlib import ExitStack
+from pathlib import Path
+from typing import Final
+
+import pytest
+
+from stack_lock import STACK_DIGEST
+
+HARNESS_DIR: Final = Path(__file__).resolve().parent
+DEADLINE_SECONDS: Final = 30.0
+SETTLE_SECONDS: Final = 0.5
+HOLDER_SCRIPT: Final = """
+import sys, time
+from pathlib import Path
+from stack_lock import stack_lock
+name, mode, release_path, log_path = sys.argv[1:]
+
+
+def record(event):
+    with Path(log_path).open("a") as log:
+        log.write(f"{name} {event}\\n")
+
+
+record("waiting")
+with stack_lock(exclusive=mode == "exclusive"):
+    record("enter")
+    while not Path(release_path).exists():
+        time.sleep(0.02)
+    record("exit")
+"""
+
+
+def _events(log_path: Path) -> tuple[str, ...]:
+    return tuple(log_path.read_text().splitlines()) if log_path.exists() else ()
+
+
+def _wait_for_event(log_path: Path, event: str) -> None:
+    deadline: Final = time.monotonic() + DEADLINE_SECONDS
+    while event not in _events(log_path):
+        if time.monotonic() > deadline:
+            pytest.fail(f"{event!r} never appeared; events so far: {_events(log_path)}")
+        time.sleep(0.02)
+
+
+def _wait_until_gate_is_held_exclusively(gate_path: Path) -> None:
+    deadline: Final = time.monotonic() + DEADLINE_SECONDS
+    with gate_path.open("a") as handle:
+        while True:
+            try:
+                fcntl.flock(handle, fcntl.LOCK_SH | fcntl.LOCK_NB)
+            except BlockingIOError:
+                return
+            fcntl.flock(handle, fcntl.LOCK_UN)
+            if time.monotonic() > deadline:
+                pytest.fail("no exclusive holder ever took the gate")
+            time.sleep(0.02)
+
+
+def _start_holder(held: ExitStack, tmp_path: Path, name: str, mode: str) -> subprocess.Popen[bytes]:
+    holder: Final = held.enter_context(
+        subprocess.Popen(
+            (
+                sys.executable,
+                "-P",
+                "-c",
+                HOLDER_SCRIPT,
+                name,
+                mode,
+                str(tmp_path / f"release-{name}"),
+                str(tmp_path / "events"),
+            ),
+            cwd=HARNESS_DIR,
+            env={**os.environ, "TMPDIR": str(tmp_path), "PYTHONPATH": str(HARNESS_DIR)},
+        )
+    )
+    held.callback(holder.kill)
+    return holder
+
+
+def test_readers_share_exclusive_waits_and_a_waiting_exclusive_beats_later_readers(tmp_path: Path) -> None:
+    lock_dir: Final = tmp_path / f"litellm-e2e-stack-{STACK_DIGEST}"
+    lock_dir.mkdir()
+    log_path: Final = tmp_path / "events"
+    with ExitStack() as held:
+        first_reader: Final = _start_holder(held, tmp_path, "A", "shared")
+        _wait_for_event(log_path, "A enter")
+        second_reader: Final = _start_holder(held, tmp_path, "R", "shared")
+        _wait_for_event(log_path, "R enter")
+        (tmp_path / "release-R").touch()
+        _wait_for_event(log_path, "R exit")
+        writer: Final = _start_holder(held, tmp_path, "W", "exclusive")
+        _wait_until_gate_is_held_exclusively(lock_dir / "gate")
+        late_reader: Final = _start_holder(held, tmp_path, "B", "shared")
+        _wait_for_event(log_path, "B waiting")
+        time.sleep(SETTLE_SECONDS)
+        (tmp_path / "release-A").touch()
+        _wait_for_event(log_path, "W enter")
+        (tmp_path / "release-W").touch()
+        _wait_for_event(log_path, "B enter")
+        (tmp_path / "release-B").touch()
+        for holder in (first_reader, second_reader, writer, late_reader):
+            assert holder.wait(timeout=DEADLINE_SECONDS) == 0
+    events: Final = _events(log_path)
+    assert events.index("R enter") < events.index("A exit")
+    assert events.index("W enter") > events.index("A exit")
+    assert events.index("B enter") > events.index("W exit")