From 4ac1cc8c966cf614d4c61159d8152c55936522bb Mon Sep 17 00:00:00 2001 From: mubashir1osmani Date: Tue, 30 Jun 2026 12:54:47 -0700 Subject: [PATCH] fix(logging): route realtime success logging through the bounded worker (#31733) RealTimeStreaming.log_messages dispatched the success handler with a bare asyncio.create_task, bypassing GLOBAL_LOGGING_WORKER (which gives a per-coroutine timeout and a concurrency cap). On a long-lived realtime websocket a slow logging callback left one suspended task per logged turn, each pinning that turn's assembled response, accumulating without bound (~12-15k in-flight under load in a repro) until OOM. Route realtime success logging through the bounded worker so in-flight logging is capped and a hung callback is cancelled at the worker timeout. The chat and responses streaming success-logging paths are intentionally left unchanged: their success callbacks must complete within the call's event-loop run (the non-streaming path pairs the worker with a synchronous callback; the streaming path has no such companion), so deferring them through the worker would drop logs for one-shot SDK calls and breaks test_async_custom_handler_stream. Bounding those paths needs a load-shedding approach and is left to a follow-up. (cherry picked from commit d4c33b2b5922cdc780c7dac31c73a2a21f342540) --- .../litellm_core_utils/realtime_streaming.py | 7 ++++-- .../test_realtime_streaming.py | 23 +++++++++++++++++++ 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 772f058d9bbd..9ce71009f33a 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -5,6 +5,7 @@ import litellm from litellm._logging import verbose_logger +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig from litellm.types.llms.openai import ( OpenAIRealtimeEvents, @@ -268,8 +269,10 @@ async def log_messages(self): self.tool_calls ) ## ASYNC LOGGING - # Create an event loop for the new thread - asyncio.create_task(self.logging_obj.async_success_handler(self.messages)) + # Route through the bounded logging worker (per-coroutine timeout + + # concurrency cap) instead of a bare create_task, so a slow callback + # can't leave suspended tasks pinning each call's response in memory. + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(self.logging_obj.async_success_handler(self.messages)) ## SYNC LOGGING executor.submit(self.logging_obj.success_handler(self.messages)) diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 3424bfd801c8..688a75e6248b 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -2110,3 +2110,26 @@ async def test_deferred_setup_caps_non_audio_buffered_bytes(monkeypatch): assert ( streaming._pending_messages_byte_total <= RealTimeStreaming._MAX_BUFFERED_BYTES ) + + +@pytest.mark.asyncio +async def test_log_messages_routes_async_logging_through_bounded_worker(): + """Realtime success logging must go through GLOBAL_LOGGING_WORKER (bounded + queue + per-coroutine timeout), not a bare asyncio.create_task. A bare task + has no timeout/concurrency cap, so when a logging callback is slow every + realtime turn leaves a suspended task pinning its response in memory -> an + unbounded leak. Regression for that fix.""" + logging_obj = MagicMock() + streaming = RealTimeStreaming(MagicMock(), MagicMock(), logging_obj) + streaming.messages = [{"type": "session.created"}] + + with ( + patch("litellm.litellm_core_utils.realtime_streaming.GLOBAL_LOGGING_WORKER") as mock_worker, + patch("litellm.litellm_core_utils.realtime_streaming.asyncio.create_task") as mock_create_task, + patch("litellm.litellm_core_utils.realtime_streaming.executor.submit"), + ): + await streaming.log_messages() + + mock_worker.ensure_initialized_and_enqueue.assert_called_once() + # the bare create_task path must no longer be used for success logging + mock_create_task.assert_not_called()