From 212ab630b81957610b9708af7fb85b827538979e Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Mon, 21 Sep 2026 15:44:26 -0700 Subject: [PATCH 1/3] fix(logging_worker): make flush() survive an event loop change flush() awaited join() on whatever queue the worker held, even one bound to an event loop that has since closed. Its unfinished counter is never decremented on the new loop, so the first flush() after a loop change hung until pytest-timeout killed it and every later one raised "is bound to a different event loop" from the queue's Event. The CircleCI unit job has been red on every branch since the first tests that flush without enqueueing landed, and an SDK script that flushes from a second asyncio.run() hangs the same way. flush() now goes through start() first, which carries the tasks stranded on the previous loop onto the current one and guarantees a worker there to drain them, the same loop-change handling every other entry point already had. --- litellm/litellm_core_utils/logging_worker.py | 6 ++++ .../litellm_core_utils/test_logging_worker.py | 34 +++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/litellm/litellm_core_utils/logging_worker.py b/litellm/litellm_core_utils/logging_worker.py index 5ccc5632646..cb3d8bf4fe5 100644 --- a/litellm/litellm_core_utils/logging_worker.py +++ b/litellm/litellm_core_utils/logging_worker.py @@ -484,9 +484,15 @@ class LoggingWorker: so it correctly handles items that have been dequeued but whose callback hasn't finished yet — ``queue.empty()`` would return True in that window and cause us to skip the wait. + + ``start()`` runs first so that, after an event loop change, the tasks + still on the previous loop's queue move onto this loop and a worker + here drains them; joining the old queue directly would wait on a + counter nothing on this loop ever decrements. """ if self._queue is None: return + self.start() await self._queue.join() async def clear_queue(self): diff --git a/tests/test_litellm/litellm_core_utils/test_logging_worker.py b/tests/test_litellm/litellm_core_utils/test_logging_worker.py index 1553e788472..891bd0686b2 100644 --- a/tests/test_litellm/litellm_core_utils/test_logging_worker.py +++ b/tests/test_litellm/litellm_core_utils/test_logging_worker.py @@ -180,6 +180,40 @@ class TestLoggingWorker: assert sorted(fired) == ["first", "second"] + @pytest.mark.parametrize("stranded", ["still_queued", "dequeued_never_started"]) + def test_flush_on_new_loop_drains_tasks_stranded_on_previous_loop(self, stranded): + """ + Regression: ``flush()`` from a new event loop used to ``join()`` the queue bound to the + previous loop, whose unfinished counter nothing on the new loop ever decrements. The first + such flush hung until pytest-timeout killed it and every later one raised + ``RuntimeError: ... is bound to a different event loop`` from the queue's Event. + """ + worker = LoggingWorker(timeout=1.0, max_queue_size=10) + fired = [] + + async def marker(): + fired.append(True) + + async def enqueue_on_first_loop(): + if stranded == "still_queued": + worker._ensure_queue() + worker.enqueue(marker()) + return + worker.ensure_initialized_and_enqueue(marker()) + + asyncio.run(enqueue_on_first_loop()) + assert worker._queue is not None + expected_shape = (1, 0) if stranded == "still_queued" else (0, 1) + assert (worker._queue.qsize(), len(worker._unstarted_dequeued_tasks())) == expected_shape + assert fired == [], "precondition: the callback never ran before the first loop closed" + + async def flush_on_second_loop(): + await asyncio.wait_for(worker.flush(), timeout=5) + + asyncio.run(flush_on_second_loop()) + + assert fired == [True] + def test_flush_on_exit_swallows_cancellation_and_drains_remaining(self): """A callback raising CancelledError must not abort the atexit flush of later events.""" worker = LoggingWorker(timeout=1.0, max_queue_size=10) From e86ba8bbebfad37f558b91bb67e5fe55a71493f2 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Mon, 21 Sep 2026 15:56:09 -0700 Subject: [PATCH 2/3] test(logging_worker): cover a same-loop flush and a repeated flush after a loop change --- litellm/litellm_core_utils/logging_worker.py | 6 ++--- .../litellm_core_utils/test_logging_worker.py | 23 +++++++++++++++++-- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/litellm/litellm_core_utils/logging_worker.py b/litellm/litellm_core_utils/logging_worker.py index cb3d8bf4fe5..2f8e7bdccea 100644 --- a/litellm/litellm_core_utils/logging_worker.py +++ b/litellm/litellm_core_utils/logging_worker.py @@ -485,10 +485,8 @@ class LoggingWorker: callback hasn't finished yet — ``queue.empty()`` would return True in that window and cause us to skip the wait. - ``start()`` runs first so that, after an event loop change, the tasks - still on the previous loop's queue move onto this loop and a worker - here drains them; joining the old queue directly would wait on a - counter nothing on this loop ever decrements. + ``start()`` runs first so a queue left behind by a previous event loop + is carried onto this one and drained here instead of joined forever. """ if self._queue is None: return diff --git a/tests/test_litellm/litellm_core_utils/test_logging_worker.py b/tests/test_litellm/litellm_core_utils/test_logging_worker.py index 891bd0686b2..336067976fa 100644 --- a/tests/test_litellm/litellm_core_utils/test_logging_worker.py +++ b/tests/test_litellm/litellm_core_utils/test_logging_worker.py @@ -207,10 +207,29 @@ class TestLoggingWorker: assert (worker._queue.qsize(), len(worker._unstarted_dequeued_tasks())) == expected_shape assert fired == [], "precondition: the callback never ran before the first loop closed" - async def flush_on_second_loop(): + async def flush_twice_on_second_loop(): + await asyncio.wait_for(worker.flush(), timeout=5) await asyncio.wait_for(worker.flush(), timeout=5) - asyncio.run(flush_on_second_loop()) + asyncio.run(flush_twice_on_second_loop()) + + assert fired == [True] + + def test_flush_starts_a_worker_when_the_queue_has_none(self): + """``flush()`` must drain a queue that exists on the current loop without a running worker.""" + worker = LoggingWorker(timeout=1.0, max_queue_size=10) + fired = [] + + async def marker(): + fired.append(True) + + async def enqueue_then_flush(): + worker._ensure_queue() + worker.enqueue(marker()) + assert worker._worker_task is None, "precondition: nothing is draining the queue yet" + await asyncio.wait_for(worker.flush(), timeout=3) + + asyncio.run(enqueue_then_flush()) assert fired == [True] From 9388602f467b4c1d3ce2f137191c4a8638a8a5d1 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Mon, 21 Sep 2026 16:13:09 -0700 Subject: [PATCH 3/3] test(logging_worker): track callback runs with AsyncMock instead of a mutated list --- .../litellm_core_utils/test_logging_worker.py | 22 +++++++------------ 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/tests/test_litellm/litellm_core_utils/test_logging_worker.py b/tests/test_litellm/litellm_core_utils/test_logging_worker.py index 336067976fa..2bb93a58531 100644 --- a/tests/test_litellm/litellm_core_utils/test_logging_worker.py +++ b/tests/test_litellm/litellm_core_utils/test_logging_worker.py @@ -189,23 +189,20 @@ class TestLoggingWorker: ``RuntimeError: ... is bound to a different event loop`` from the queue's Event. """ worker = LoggingWorker(timeout=1.0, max_queue_size=10) - fired = [] - - async def marker(): - fired.append(True) + callback = AsyncMock() async def enqueue_on_first_loop(): if stranded == "still_queued": worker._ensure_queue() - worker.enqueue(marker()) + worker.enqueue(callback()) return - worker.ensure_initialized_and_enqueue(marker()) + worker.ensure_initialized_and_enqueue(callback()) asyncio.run(enqueue_on_first_loop()) assert worker._queue is not None expected_shape = (1, 0) if stranded == "still_queued" else (0, 1) assert (worker._queue.qsize(), len(worker._unstarted_dequeued_tasks())) == expected_shape - assert fired == [], "precondition: the callback never ran before the first loop closed" + assert callback.await_count == 0, "precondition: the callback never ran before the first loop closed" async def flush_twice_on_second_loop(): await asyncio.wait_for(worker.flush(), timeout=5) @@ -213,25 +210,22 @@ class TestLoggingWorker: asyncio.run(flush_twice_on_second_loop()) - assert fired == [True] + assert callback.await_count == 1 def test_flush_starts_a_worker_when_the_queue_has_none(self): """``flush()`` must drain a queue that exists on the current loop without a running worker.""" worker = LoggingWorker(timeout=1.0, max_queue_size=10) - fired = [] - - async def marker(): - fired.append(True) + callback = AsyncMock() async def enqueue_then_flush(): worker._ensure_queue() - worker.enqueue(marker()) + worker.enqueue(callback()) assert worker._worker_task is None, "precondition: nothing is draining the queue yet" await asyncio.wait_for(worker.flush(), timeout=3) asyncio.run(enqueue_then_flush()) - assert fired == [True] + assert callback.await_count == 1 def test_flush_on_exit_swallows_cancellation_and_drains_remaining(self): """A callback raising CancelledError must not abort the atexit flush of later events."""