From fb4b1e728a18a56f8ec93d72ade3c237c6bbddb6 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Thu, 3 Sep 2026 12:16:08 -0700 Subject: [PATCH 1/3] test(team-race): wait on pg_locks instead of a fixed sleep The three race tests claimed to pin the interleaving deterministically, but lock_acquired.set() ran as the first statement of the task coroutine, before the awaited endpoint call, so it only signalled that the task had started. The real synchronisation was `await asyncio.sleep(0.2)` followed by `assert not task.done()`, which is a timing assumption on a box running four xdist workers against one Postgres. Take the blocking connection's advisory lock key straight out of pg_locks, then poll from the unblocked watcher connection until a non-granted lock on that same key appears. That is the condition the sleep was standing in for, and it holds however slow the machine is. If the endpoint returns without ever queueing, the helper now fails with the endpoint's own exception chained on instead of a bare assert. The two tests reported failing in CI used sleep(0.2); the third used sleep(0.3) and was not reported, which is consistent with the margin being the cause. --- .../test_team_delete_member_add_race.py | 74 +++++++++++++------ 1 file changed, 53 insertions(+), 21 deletions(-) diff --git a/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py b/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py index 7577570be48..9af742b0dfe 100644 --- a/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py +++ b/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py @@ -9,14 +9,16 @@ be forced by a sequential script: it needs one request to be genuinely mid-fligh other commits. A mocked prisma cannot arbitrate that either, since the property under test is whether Postgres's own advisory lock actually serializes the two requests. -These tests pin the interleaving the same way test_access_group_team_sync.py does: a second -real connection holds the team's advisory lock in its own transaction, so the function under -test is provably blocked on it rather than hoping a sleep lands in the right gap. +These tests pin the interleaving without a timing assumption: a second real connection holds +the team's advisory lock in its own transaction, and the test then waits for Postgres itself +to report the endpoint queued behind that exact lock. A sleep can only guess whether the +endpoint has reached the lock yet; pg_locks answers it. """ import asyncio import json import os +import time import uuid from contextlib import asynccontextmanager from datetime import timedelta @@ -39,6 +41,48 @@ _DELETE_SEEDED = 'DELETE FROM "LiteLLM_TeamMembership" WHERE team_id = $1' _DELETE_USER = 'DELETE FROM "LiteLLM_UserTable" WHERE user_id = $1' _DELETE_TEAM = 'DELETE FROM "LiteLLM_TeamTable" WHERE team_id = $1' _LOCK_SQL = "SELECT pg_advisory_xact_lock(hashtext($1)) IS NULL AS locked" +_HELD_LOCK_KEY_SQL = ( + "SELECT classid::bigint AS classid, objid::bigint AS objid FROM pg_locks " + "WHERE locktype = 'advisory' AND granted AND pid = pg_backend_pid()" +) +_LOCK_WAITER_SQL = ( + "SELECT count(*)::int AS waiters FROM pg_locks " + "WHERE locktype = 'advisory' AND NOT granted " + "AND classid::bigint = $1 AND objid::bigint = $2" +) +_LOCK_WAIT_TIMEOUT_SECONDS = 20.0 +_LOCK_POLL_SECONDS = 0.01 + + +async def _hold_team_lock(held, team_id: str) -> tuple[int, int]: + """Take the team's advisory lock and return its pg_locks key. + + Reading the key back off our own backend avoids re-deriving hashtext()'s signed + 32-bit split here, and pins the watcher to this lock rather than to any advisory + lock another xdist worker happens to hold on the same database.""" + await held.query_raw(_LOCK_SQL, team_id) + rows = await held.query_raw(_HELD_LOCK_KEY_SQL) + assert len(rows) == 1, f"expected exactly one advisory lock on the blocking connection, got {rows}" + return rows[0]["classid"], rows[0]["objid"] + + +async def _await_lock_contention(watcher, lock_key: tuple[int, int], task, what: str) -> None: + """Block until Postgres reports `task` queued behind the held lock. + + This is the assertion that the endpoint serializes on the team's advisory lock, and it + is what a fixed sleep was standing in for: the endpoint is only provably waiting once a + non-granted advisory lock on the same key exists. `watcher` must be a connection that is + not itself blocked, so it can observe the queue.""" + classid, objid = lock_key + deadline = time.monotonic() + _LOCK_WAIT_TIMEOUT_SECONDS + while time.monotonic() < deadline: + if task.done(): + raise AssertionError(f"{what} returned without waiting on the team's advisory lock") from task.exception() + rows = await watcher.query_raw(_LOCK_WAITER_SQL, classid, objid) + if rows[0]["waiters"]: + return + await asyncio.sleep(_LOCK_POLL_SECONDS) + raise AssertionError(f"{what} never queued on the team's advisory lock within {_LOCK_WAIT_TIMEOUT_SECONDS}s") def _race_ids() -> tuple[str, str]: @@ -110,10 +154,8 @@ async def test_member_add_blocked_by_delete_writes_no_dangling_reference(): blocker = Prisma() await blocker.connect() - lock_acquired = asyncio.Event() async def add_member(): - lock_acquired.set() await _add_team_members_to_team( data=TeamMemberAddRequest( team_id=team_id, @@ -128,11 +170,9 @@ async def test_member_add_blocked_by_delete_writes_no_dangling_reference(): try: async with blocker.tx(timeout=timedelta(seconds=30)) as held: - await held.query_raw(_LOCK_SQL, team_id) + lock_key = await _hold_team_lock(held, team_id) task = asyncio.create_task(add_member()) - await lock_acquired.wait() - await asyncio.sleep(0.2) - assert not task.done(), "member_add did not wait on the team's advisory lock" + await _await_lock_contention(db, lock_key, task, "member_add") # the delete wins the race: strip the team row while the lock is held await held.execute_raw(_DELETE_TEAM, team_id) @@ -185,10 +225,8 @@ async def test_member_delete_blocked_by_member_add_removes_from_the_fresh_roster blocker = Prisma() await blocker.connect() - lock_acquired = asyncio.Event() async def run_delete(): - lock_acquired.set() return await team_member_delete( data=TeamMemberDeleteRequest(team_id=team_id, user_id=user_id), user_api_key_dict=_admin_auth(), @@ -196,11 +234,9 @@ async def test_member_delete_blocked_by_member_add_removes_from_the_fresh_roster try: async with blocker.tx(timeout=timedelta(seconds=30)) as held: - await held.query_raw(_LOCK_SQL, team_id) + lock_key = await _hold_team_lock(held, team_id) task = asyncio.create_task(run_delete()) - await lock_acquired.wait() - await asyncio.sleep(0.2) - assert not task.done(), "member_delete did not wait on the team's advisory lock" + await _await_lock_contention(db, lock_key, task, "member_delete") # member_add wins the race: it adds `other_user` while holding the lock await held.litellm_teamtable.update( @@ -265,10 +301,8 @@ async def test_delete_blocked_by_member_add_sweeps_the_fresh_reference(): blocker = Prisma() await blocker.connect() - lock_acquired = asyncio.Event() async def run_delete(): - lock_acquired.set() return await delete_team( data=DeleteTeamRequest(team_ids=[team_id]), http_request=MagicMock(), @@ -278,11 +312,9 @@ async def test_delete_blocked_by_member_add_sweeps_the_fresh_reference(): try: async with blocker.tx(timeout=timedelta(seconds=30)) as held: - await held.query_raw(_LOCK_SQL, team_id) + lock_key = await _hold_team_lock(held, team_id) task = asyncio.create_task(run_delete()) - await lock_acquired.wait() - await asyncio.sleep(0.3) - assert not task.done(), "delete_team did not wait on the team's advisory lock" + await _await_lock_contention(db, lock_key, task, "delete_team") # member_add wins the race: write the reference while holding the lock await held.litellm_usertable.upsert( From 8ac704a855cc2a27887138a5a2dd121f4a81b1ae Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Thu, 3 Sep 2026 12:16:16 -0700 Subject: [PATCH 2/3] test(responses): cancel the streaming response while it is still in flight test_cancel_streaming_response drained the whole stream and only then called cancel, by which point the response had finished and the cancel was expected to fail. The test passed only because the surrounding except matched the literal string "Cannot cancel a completed response", which is upstream OpenAI's wording, not ours: it appears nowhere in this repo. Any change to that text, a different status code, or the background job still running flipped the result. The test also never exercised cancellation, and its only positive assertion was hasattr(cancel_response, "id"). Break out of the stream at the first chunk carrying a response id and cancel there, with a prompt long enough that the response cannot have completed in the meantime. Both proxy cancel paths, the polling handler and the provider passthrough, settle on status "cancelled", so assert that and the returned id rather than a provider error string. --- .../test_e2e_openai_responses_api.py | 54 +++++++++---------- 1 file changed, 24 insertions(+), 30 deletions(-) diff --git a/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py b/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py index abae26e02cd..b24ac0bdb96 100644 --- a/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py +++ b/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py @@ -154,42 +154,36 @@ def test_cancel_response(): def test_cancel_streaming_response(): - try: - client = get_test_client() - from litellm.types.llms.openai import ResponsesAPIResponse + """Cancel a background streaming response while it is still generating. - stream = client.responses.create( - model="gpt-5.5", - input="just respond with the word 'ping'", - stream=True, - background=True, - ) + The prompt is deliberately long-running and the stream is abandoned at the first + chunk carrying a response id, so the response is provably still in flight when the + cancel lands. Draining the stream first would finish the response, making the cancel + fail and leaving nothing but the provider's error wording to assert on. + """ + client = get_test_client() - collected_chunks = [] + with client.responses.create( + model="gpt-5.5", + input="write a 2000 word essay on the history of the printing press", + stream=True, + background=True, + ) as stream: + chunk_count = 0 response_id = None for chunk in stream: - print("stream chunk=", chunk) - collected_chunks.append(chunk) - # Extract response ID from the first chunk that has it - if ( - response_id is None - and hasattr(chunk, "response") - and hasattr(chunk.response, "id") - ): - response_id = chunk.response.id + chunk_count += 1 + response_id = getattr(getattr(chunk, "response", None), "id", None) + if response_id is not None: + break - assert len(collected_chunks) > 0 + assert chunk_count > 0, "stream produced no chunks" + assert response_id is not None, "no streamed chunk carried a response id to cancel" - # cancel the response if we got a response ID - if response_id: - cancel_response = client.responses.cancel(response_id) - print("CANCEL streaming response=", cancel_response) - assert hasattr(cancel_response, "id") - except Exception as e: - if "Cannot cancel a completed response" in str(e): - pass - else: - raise e + cancel_response = client.responses.cancel(response_id) + print("CANCEL streaming response=", cancel_response) + assert cancel_response.id == response_id + assert cancel_response.status == "cancelled" def test_cancel_invalid_response_id(): From 52e24aebbabdd4d889dda96f3ebeab0e3bd8c3b1 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Thu, 3 Sep 2026 13:49:43 -0700 Subject: [PATCH 3/3] refactor(tests): assign the streamed id and lock poll once instead of rebinding The cancel test accumulated chunk_count and reassigned response_id on every iteration, and the lock watcher rebound its query result on every poll. Both are the mutable-local pattern the repo avoids. The stream now drains through a generator that stops at the first chunk carrying a response id, so the caller binds streamed_ids once and reads the id off the tail. Empty stream, no-id stream and first-chunk-id all behave exactly as the loop did. The watcher inlines its poll result. --- .../test_e2e_openai_responses_api.py | 20 +++++++++++-------- .../test_team_delete_member_add_race.py | 3 +-- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py b/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py index b24ac0bdb96..755f17c394e 100644 --- a/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py +++ b/tests/openai_endpoints_tests/test_e2e_openai_responses_api.py @@ -153,6 +153,15 @@ def test_cancel_response(): raise e +def _response_ids_until_first(stream): + """Yield each streamed chunk's response id, stopping at the first chunk that carries one.""" + for chunk in stream: + response_id = getattr(getattr(chunk, "response", None), "id", None) + yield response_id + if response_id is not None: + return + + def test_cancel_streaming_response(): """Cancel a background streaming response while it is still generating. @@ -169,15 +178,10 @@ def test_cancel_streaming_response(): stream=True, background=True, ) as stream: - chunk_count = 0 - response_id = None - for chunk in stream: - chunk_count += 1 - response_id = getattr(getattr(chunk, "response", None), "id", None) - if response_id is not None: - break + streamed_ids = tuple(_response_ids_until_first(stream)) - assert chunk_count > 0, "stream produced no chunks" + assert streamed_ids, "stream produced no chunks" + response_id = streamed_ids[-1] assert response_id is not None, "no streamed chunk carried a response id to cancel" cancel_response = client.responses.cancel(response_id) diff --git a/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py b/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py index 9af742b0dfe..d3ffcf2445e 100644 --- a/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py +++ b/tests/proxy_admin_ui_tests/test_team_delete_member_add_race.py @@ -78,8 +78,7 @@ async def _await_lock_contention(watcher, lock_key: tuple[int, int], task, what: while time.monotonic() < deadline: if task.done(): raise AssertionError(f"{what} returned without waiting on the team's advisory lock") from task.exception() - rows = await watcher.query_raw(_LOCK_WAITER_SQL, classid, objid) - if rows[0]["waiters"]: + if (await watcher.query_raw(_LOCK_WAITER_SQL, classid, objid))[0]["waiters"]: return await asyncio.sleep(_LOCK_POLL_SECONDS) raise AssertionError(f"{what} never queued on the team's advisory lock within {_LOCK_WAIT_TIMEOUT_SECONDS}s")