Fix OOM issues across integrations

.semgrep/rules/python/unbounded-memory.yml

@@ -1,15 +1,14 @@
 # Unbounded memory growth – data structures without a clear max limit
-# Can lead to OOM under load or with unbounded input.
+# Can lead to OOM under load.
 
 rules:
-  # asyncio.Queue() with no maxsize is unbounded – bad pattern for integrations (log queues, etc.)
   - id: unbounded-asyncio-queue
     message: asyncio.Queue() with no maxsize can grow unbounded. Use asyncio.Queue(maxsize=N) for integrations (e.g. log queues).
-    severity: WARNING
+    severity: ERROR
     languages: [python]
     pattern-either:
       - pattern: asyncio.Queue()
       - pattern: asyncio.Queue(maxsize=0)
     metadata:
       category: correctness
-      cwe: "CWE-400: Uncontrolled Resource Consumption"
+      cwe: "CWE-400: Uncontrolled Resource Consumption"

cookbook/nova_sonic_realtime.py

@@ -16,10 +16,14 @@ Usage:
 import asyncio
 import base64
 import json
+import os
 import pyaudio
 import websockets
 from typing import Optional
 
+# Bounded queue size for audio chunks (configurable via env to avoid unbounded memory)
+AUDIO_QUEUE_MAXSIZE = int(os.getenv("LITELLM_ASYNCIO_QUEUE_MAXSIZE", 10_000))
+
 # Audio configuration (matching Nova Sonic requirements)
 INPUT_SAMPLE_RATE = 16000  # Nova Sonic expects 16kHz input
 OUTPUT_SAMPLE_RATE = 24000  # Nova Sonic outputs 24kHz
@@ -40,7 +44,7 @@ class RealtimeClient:
         self.api_key = api_key
         self.ws: Optional[websockets.WebSocketClientProtocol] = None
         self.is_active = False
-        self.audio_queue = asyncio.Queue()
+        self.audio_queue = asyncio.Queue(maxsize=AUDIO_QUEUE_MAXSIZE)
         self.pyaudio = pyaudio.PyAudio()
         self.input_stream = None
         self.output_stream = None

docs/my-website/docs/proxy/config_settings.md

@@ -784,6 +784,7 @@ router_settings:
 | LITELLM_USER_AGENT | Custom user agent string for LiteLLM API requests. Used for partner telemetry attribution
 | LITELLM_PRINT_STANDARD_LOGGING_PAYLOAD | If true, prints the standard logging payload to the console - useful for debugging
 | LITELM_ENVIRONMENT | Environment for LiteLLM Instance. This is currently only logged to DeepEval to determine the environment for DeepEval integration.
+| LITELLM_ASYNCIO_QUEUE_MAXSIZE | Maximum size for asyncio queues (e.g. log queues, spend update queues, and cookbook examples such as realtime audio in `nova_sonic_realtime.py`). Bounds in-memory growth to prevent OOM. Default is 1000.
 | LOGFIRE_TOKEN | Token for Logfire logging service
 | LOGFIRE_BASE_URL | Base URL for Logfire logging service (useful for self hosted deployments)
 | LOGGING_WORKER_CONCURRENCY | Maximum number of concurrent coroutine slots for the logging worker on the asyncio event loop. Default is 100. Setting too high will flood the event loop with logging tasks which will lower the overall latency of the requests.

litellm/constants.py

@@ -213,6 +213,10 @@ REDIS_DAILY_AGENT_SPEND_UPDATE_BUFFER_KEY = "litellm_daily_agent_spend_update_bu
 REDIS_DAILY_TAG_SPEND_UPDATE_BUFFER_KEY = "litellm_daily_tag_spend_update_buffer"
 MAX_REDIS_BUFFER_DEQUEUE_COUNT = int(os.getenv("MAX_REDIS_BUFFER_DEQUEUE_COUNT", 100))
 MAX_SIZE_IN_MEMORY_QUEUE = int(os.getenv("MAX_SIZE_IN_MEMORY_QUEUE", 2000))
+# Bounds asyncio.Queue() instances (log queues, spend update queues, etc.) to prevent unbounded memory growth
+LITELLM_ASYNCIO_QUEUE_MAXSIZE = int(
+    os.getenv("LITELLM_ASYNCIO_QUEUE_MAXSIZE", 1000)
+)
 MAX_IN_MEMORY_QUEUE_FLUSH_COUNT = int(
     os.getenv("MAX_IN_MEMORY_QUEUE_FLUSH_COUNT", 1000)
 )

litellm/integrations/gcs_bucket/gcs_bucket.py

@@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
 from urllib.parse import quote
 
 from litellm._logging import verbose_logger
+from litellm.constants import LITELLM_ASYNCIO_QUEUE_MAXSIZE
 from litellm.integrations.additional_logging_utils import AdditionalLoggingUtils
 from litellm.integrations.gcs_bucket.gcs_bucket_base import GCSBucketBase
 from litellm.proxy._types import CommonProxyErrors
@@ -41,7 +42,9 @@ class GCSBucketLogger(GCSBucketBase, AdditionalLoggingUtils):
             batch_size=self.batch_size,
             flush_interval=self.flush_interval,
         )
-        self.log_queue: asyncio.Queue[GCSLogQueueItem] = asyncio.Queue()  # type: ignore[assignment]
+        self.log_queue: asyncio.Queue[GCSLogQueueItem] = asyncio.Queue(  # type: ignore[assignment]
+            maxsize=LITELLM_ASYNCIO_QUEUE_MAXSIZE
+        )
         asyncio.create_task(self.periodic_flush())
         AdditionalLoggingUtils.__init__(self)
 
@@ -69,6 +72,9 @@ class GCSBucketLogger(GCSBucketBase, AdditionalLoggingUtils):
             )
             if logging_payload is None:
                 raise ValueError("standard_logging_object not found in kwargs")
+            # When queue is at maxsize, flush immediately to make room (no blocking, no data dropped)
+            if self.log_queue.full():
+                await self.flush_queue()
             await self.log_queue.put(
                 GCSLogQueueItem(
                     payload=logging_payload, kwargs=kwargs, response_obj=response_obj
@@ -91,9 +97,9 @@ class GCSBucketLogger(GCSBucketBase, AdditionalLoggingUtils):
             )
             if logging_payload is None:
                 raise ValueError("standard_logging_object not found in kwargs")
-            # Add to logging queue - this will be flushed periodically
-            # Use asyncio.Queue.put() for thread-safe concurrent access
-            # If queue is full, this will block until space is available (backpressure)
+            # When queue is at maxsize, flush immediately to make room (no blocking, no data dropped)
+            if self.log_queue.full():
+                await self.flush_queue()
             await self.log_queue.put(
                 GCSLogQueueItem(
                     payload=logging_payload, kwargs=kwargs, response_obj=response_obj

litellm/proxy/db/db_transaction_queue/base_update_queue.py

@@ -10,14 +10,18 @@ from litellm._service_logger import ServiceLogging
 service_logger_obj = (
     ServiceLogging()
 )  # used for tracking metrics for In memory buffer, redis buffer, pod lock manager
-from litellm.constants import MAX_IN_MEMORY_QUEUE_FLUSH_COUNT, MAX_SIZE_IN_MEMORY_QUEUE
+from litellm.constants import (
+    LITELLM_ASYNCIO_QUEUE_MAXSIZE,
+    MAX_IN_MEMORY_QUEUE_FLUSH_COUNT,
+    MAX_SIZE_IN_MEMORY_QUEUE,
+)
 
 
 class BaseUpdateQueue:
     """Base class for in memory buffer for database transactions"""
 
     def __init__(self):
-        self.update_queue = asyncio.Queue()
+        self.update_queue = asyncio.Queue(maxsize=LITELLM_ASYNCIO_QUEUE_MAXSIZE)
         self.MAX_SIZE_IN_MEMORY_QUEUE = MAX_SIZE_IN_MEMORY_QUEUE
 
     async def add_update(self, update):

litellm/proxy/db/db_transaction_queue/daily_spend_update_queue.py

@@ -3,6 +3,7 @@ from copy import deepcopy
 from typing import Dict, List, Optional
 
 from litellm._logging import verbose_proxy_logger
+from litellm.constants import LITELLM_ASYNCIO_QUEUE_MAXSIZE
 from litellm.proxy._types import BaseDailySpendTransaction
 from litellm.proxy.db.db_transaction_queue.base_update_queue import (
     BaseUpdateQueue,
@@ -54,7 +55,7 @@ class DailySpendUpdateQueue(BaseUpdateQueue):
     def __init__(self):
         super().__init__()
         self.update_queue: asyncio.Queue[Dict[str, BaseDailySpendTransaction]] = (
-            asyncio.Queue()
+            asyncio.Queue(maxsize=LITELLM_ASYNCIO_QUEUE_MAXSIZE)
         )
 
     async def add_update(self, update: Dict[str, BaseDailySpendTransaction]):

litellm/proxy/db/db_transaction_queue/spend_update_queue.py

@@ -2,6 +2,7 @@ import asyncio
 from typing import Dict, List, Optional
 
 from litellm._logging import verbose_proxy_logger
+from litellm.constants import LITELLM_ASYNCIO_QUEUE_MAXSIZE
 from litellm.proxy._types import (
     DBSpendUpdateTransactions,
     Litellm_EntityType,
@@ -21,7 +22,9 @@ class SpendUpdateQueue(BaseUpdateQueue):
 
     def __init__(self):
         super().__init__()
-        self.update_queue: asyncio.Queue[SpendUpdateQueueItem] = asyncio.Queue()
+        self.update_queue: asyncio.Queue[SpendUpdateQueueItem] = asyncio.Queue(
+            maxsize=LITELLM_ASYNCIO_QUEUE_MAXSIZE
+        )
 
     async def flush_and_get_aggregated_db_spend_update_transactions(
         self,