From 975e79dcefb599f7efd5cc6f1d83d77eff9f5959 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 19 Aug 2026 14:55:53 -0700 Subject: [PATCH] fix(e2e): make concurrent replay consumption race-free Greptile flagged that lazy per-slug pool initialization could double-build under concurrent replay calls, splitting consumption across a discarded pool. Pools are now built once at ReplaySource construction and per-key consumption is a single atomic deque pop, with a barrier-synchronized regression test that fails 10/10 under the lazy-init mutant --- tests/e2e/fixture_transport.py | 28 +++++++++++-------- tests/e2e/test_fixture_transport.py | 43 +++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 11 deletions(-) diff --git a/tests/e2e/fixture_transport.py b/tests/e2e/fixture_transport.py index d49ac63787a..ce4eec701ca 100644 --- a/tests/e2e/fixture_transport.py +++ b/tests/e2e/fixture_transport.py @@ -391,18 +391,23 @@ def _miss_message(test_key: str, slug: str, canonical: CanonicalRequest, bundle: @dataclass(slots=True) class ReplaySource: """One shared pool per test over a loaded bundle, so every client built in - the session consumes the same recorded interactions. Calls match by - canonical content key: order-independent across distinct keys (concurrent - tests interleave calls nondeterministically), FIFO within one key (a poll - loop replays its recorded responses in recorded order).""" + the session consumes the same recorded interactions. Every pool is built + once at construction and per-key consumption is a single atomic deque pop, + so concurrent replay calls never race. Calls match by canonical content + key: order-independent across distinct keys (concurrent tests interleave + calls nondeterministically), FIFO within one key (a poll loop replays its + recorded responses in recorded order).""" bundle: LoadedBundle - _pools: dict[str, dict[str, deque[Interaction]]] = field(default_factory=dict) + _pools: dict[str, dict[str, deque[Interaction]]] = field(init=False) + + def __post_init__(self) -> None: + self._pools = { + slug: _build_pool(recorded) for slug, recorded in self.bundle.interactions.items() + } def _pool(self, slug: str) -> dict[str, deque[Interaction]]: - if slug not in self._pools: - self._pools[slug] = _build_pool(self.bundle.interactions.get(slug, ())) - return self._pools[slug] + return self._pools.get(slug, {}) def next_interaction(self, request: RecordedRequest) -> Interaction: test_key: Final = current_test_key() @@ -412,12 +417,13 @@ class ReplaySource: queue: Final = pool.get(canonical.key) if queue is None: raise ReplayMiss(_miss_message(test_key, slug, canonical, self.bundle)) - if not queue: + try: + return queue.popleft() + except IndexError: raise ReplayMiss( f"replay exhausted for {test_key}: every recorded interaction for key " f"{canonical.key} is already consumed; re-record with E2E_FIXTURE_MODE=record" - ) - return queue.popleft() + ) from None def leftover_error(self, test_key: str) -> str | None: """Non-None when the test consumed fewer interactions than were recorded, diff --git a/tests/e2e/test_fixture_transport.py b/tests/e2e/test_fixture_transport.py index 814155f64d1..e61088d841c 100644 --- a/tests/e2e/test_fixture_transport.py +++ b/tests/e2e/test_fixture_transport.py @@ -14,6 +14,9 @@ pinned here too, including the stale message that names the bundle's age. from __future__ import annotations import hashlib +import sys +import threading +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path @@ -458,6 +461,46 @@ class TestReplayTransport: assert first == Success(status_code=200, data=Payload(value="first")) assert second == Success(status_code=200, data=Payload(value="second")) + def test_concurrent_replays_of_one_key_serve_each_recording_exactly_once(self, tmp_path: Path) -> None: + """A burst of parallel identical calls consumes one shared pool: no + response duplicated, none forgotten, nothing left over at teardown. + The tiny switch interval forces thread preemption inside pool setup + and consumption, so a non-atomic pool build or pop fails this test.""" + root = tmp_path / "bundle" + recorder = make_recorder(root) + for ordinal in range(32): + recorder.record( + test_key=current_test_key(), + request=recorded_request( + "get", "/v1/models", headers=AuthHeaders(authorization="Bearer sk-x"), params=Query(q="all") + ), + response=RecordedResult(kind="success", status_code=200, data={"value": f"v{ordinal:02d}"}), + ) + source = replay_source(root) + replay: Transport = ReplayTransport(source=source, master_key="sk-1234") + barrier = threading.Barrier(8) + + def consume_one() -> str: + result = replay.get( + "/v1/models", headers=replay.master, params=Query(q="all"), response_type=Payload + ) + assert isinstance(result, Success) + return result.data.value + + def consume(_: int) -> tuple[str, ...]: + barrier.wait() + return tuple(consume_one() for _call in range(4)) + + previous_interval = sys.getswitchinterval() + sys.setswitchinterval(1e-6) + try: + with ThreadPoolExecutor(max_workers=8) as executor: + served = sorted(value for values in executor.map(consume, range(8)) for value in values) + finally: + sys.setswitchinterval(previous_interval) + assert served == [f"v{ordinal:02d}" for ordinal in range(32)] + assert source.leftover_error(current_test_key()) is None + class TestRecordedKeySets: def test_two_separate_recordings_of_one_flow_produce_identical_key_sets(