fix(logging_worker): cancel orphaned tasks on event loop change

litellm/litellm_core_utils/logging_worker.py

@@ -73,9 +73,8 @@ class LoggingWorker:
             )
             # Cancel orphaned tasks bound to the old (likely closed) event
             # loop and suppress "Task was destroyed but it is pending!"
-            # warnings.  We cannot await these tasks because their event loop
-            # is no longer running, so we mark them to skip the __del__
-            # warning instead.
+            # warnings. We cannot await these tasks because their event loop
+            # is no longer running.
             self._discard_orphaned_tasks()
             # Clear old state - these are bound to the old loop
             self._queue = None
@@ -418,25 +417,25 @@ class LoggingWorker:
                 break
 
     def _discard_orphaned_tasks(self) -> None:
-        """Cancel orphaned tasks and suppress their destroy warnings.
+        """Cancel orphaned tasks and suppress destroy warnings.
 
-        When the event loop changes or the process is exiting, pending tasks
-        bound to the old (likely closed) loop cannot be properly awaited.
-        We attempt to cancel them and, regardless of whether cancel()
-        succeeds (it may raise ``RuntimeError`` if the loop is already
-        closed), suppress the ``"Task was destroyed but it is pending!"``
-        warning by setting ``_log_destroy_pending = False``.
+        When the event loop changes or the process exits, pending tasks bound to
+        the old loop cannot be awaited. We still cancel them best-effort and
+        disable asyncio's pending-task destroy warning.
         """
         all_tasks = list(self._running_tasks)
         if self._worker_task is not None:
             all_tasks.append(self._worker_task)
-        for t in all_tasks:
-            if not t.done():
-                try:
-                    t.cancel()
-                except RuntimeError:
-                    pass  # Event loop is already closed
-                t._log_destroy_pending = False
+
+        for task in all_tasks:
+            if task.done():
+                continue
+            try:
+                task.cancel()
+            except RuntimeError:
+                pass  # Event loop is already closed.
+            task._log_destroy_pending = False
+
         self._worker_task = None
         self._running_tasks.clear()
 
@@ -495,9 +494,8 @@ class LoggingWorker:
         Note: All logging in this method is wrapped to handle cases where
         logging handlers are closed during shutdown.
         """
-        # Cancel the old worker task — its event loop is already closed.
-        # Suppress "Task was destroyed but it is pending!" warnings since
-        # the closed loop cannot process cancellation.
+        # The original worker loop is bound to the old event loop, which is
+        # often already closed by the time atexit runs.
         self._discard_orphaned_tasks()
 
         if self._queue is None:

tests/test_litellm/litellm_core_utils/test_logging_worker.py

@@ -363,61 +363,36 @@ class TestLoggingWorker:
 
     @pytest.mark.asyncio
     async def test_event_loop_change_cancels_orphaned_tasks(self):
-        """Test that switching event loops cancels old tasks and suppresses warnings.
-
-        When the event loop changes (e.g. between asyncio.run() calls or in
-        test suites), _ensure_queue() must cancel the old worker task and set
-        _log_destroy_pending = False so that garbage-collecting the orphaned
-        task does not emit "Task was destroyed but it is pending!" warnings.
-        """
+        """Restarting on a new loop should cancel and discard the old worker."""
         worker = LoggingWorker(timeout=1.0, max_queue_size=10)
-
-        # Start the worker — creates _worker_task on the current loop
         worker.start()
         await asyncio.sleep(0.05)
 
         old_task = worker._worker_task
-        assert old_task is not None and not old_task.done()
+        assert old_task is not None
+        assert old_task.done() is False
 
-        # Simulate an event loop change: bind worker to a different loop
-        # so that the next _ensure_queue() call detects the mismatch.
         worker._bound_loop = asyncio.new_event_loop()
-
-        # Re-start triggers _ensure_queue() which should cancel the old task
         worker.start()
         await asyncio.sleep(0.05)
 
-        # The old task should have been cancelled and marked to suppress
-        # the "Task was destroyed but it is pending!" warning.
-        assert old_task.cancelled() or old_task.done()
+        assert old_task.done() or old_task.cancelled()
         assert old_task._log_destroy_pending is False
-
-        # The worker should be running on the current loop now
         assert worker._worker_task is not None
         assert worker._worker_task is not old_task
 
         await worker.stop()
 
     def test_flush_on_exit_cancels_worker_task(self):
-        """Test that _flush_on_exit cancels the worker task to avoid warnings.
-
-        When the atexit handler fires, the original event loop is closed.
-        _flush_on_exit must cancel the orphaned worker task and suppress
-        its destroy warning before creating a new loop to drain the queue.
-        """
+        """_flush_on_exit should discard the old worker bound to a closed loop."""
         loop = asyncio.new_event_loop()
         worker = LoggingWorker(timeout=1.0, max_queue_size=10)
 
-        # Start the worker on a loop, then close it (simulating process exit)
         loop.run_until_complete(self._start_worker(worker))
         old_task = worker._worker_task
         assert old_task is not None
 
-        # Close the loop — this is what happens before atexit fires
         loop.close()
-
-        # Now _flush_on_exit should clean up the orphaned task without
-        # raising RuntimeError from the closed event loop.
         worker._flush_on_exit()
 
         assert old_task._log_destroy_pending is False