diff --git a/litellm/constants.py b/litellm/constants.py index dbc79b69a67..3819cc3b2dc 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -1344,6 +1344,11 @@ SPEND_LOG_RUN_LOOPS = int(os.getenv("SPEND_LOG_RUN_LOOPS", 500)) SPEND_LOG_CLEANUP_BATCH_SIZE = int(os.getenv("SPEND_LOG_CLEANUP_BATCH_SIZE", 1000)) SPEND_LOG_QUEUE_SIZE_THRESHOLD = int(os.getenv("SPEND_LOG_QUEUE_SIZE_THRESHOLD", 100)) SPEND_LOG_QUEUE_POLL_INTERVAL = float(os.getenv("SPEND_LOG_QUEUE_POLL_INTERVAL", 2.0)) +# Maximum number of spend log entries to hold in-memory before dropping oldest. +# Prevents unbounded memory growth when DB writes are slow or failing. +MAX_SPEND_LOG_TRANSACTIONS_QUEUE_SIZE = int( + os.getenv("MAX_SPEND_LOG_TRANSACTIONS_QUEUE_SIZE", 10000) +) DEFAULT_CRON_JOB_LOCK_TTL_SECONDS = int( os.getenv("DEFAULT_CRON_JOB_LOCK_TTL_SECONDS", 60) ) # 1 minute diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py index a305d5be1e6..e1b3ab5ac6b 100644 --- a/litellm/proxy/db/db_spend_update_writer.py +++ b/litellm/proxy/db/db_spend_update_writer.py @@ -720,16 +720,29 @@ class DBSpendUpdateWriter: prisma_client: Optional[PrismaClient] = None, spend_logs_url: Optional[str] = os.getenv("SPEND_LOGS_URL"), ) -> Optional[PrismaClient]: + from litellm.constants import MAX_SPEND_LOG_TRANSACTIONS_QUEUE_SIZE + verbose_proxy_logger.debug( "Writing spend log to db - request_id: {}, spend: {}".format( payload.get("request_id"), payload.get("spend") ) ) - if prisma_client is not None and spend_logs_url is not None: - async with prisma_client._spend_log_transactions_lock: - prisma_client.spend_log_transactions.append(payload) - elif prisma_client is not None: + if prisma_client is not None: async with prisma_client._spend_log_transactions_lock: + # Prevent unbounded queue growth when DB writes are slow/failing + if ( + len(prisma_client.spend_log_transactions) + >= MAX_SPEND_LOG_TRANSACTIONS_QUEUE_SIZE + ): + drop_count = MAX_SPEND_LOG_TRANSACTIONS_QUEUE_SIZE // 10 + prisma_client.spend_log_transactions = ( + prisma_client.spend_log_transactions[drop_count:] + ) + verbose_proxy_logger.warning( + "Spend log queue exceeded %d entries, dropped oldest %d to prevent memory leak", + MAX_SPEND_LOG_TRANSACTIONS_QUEUE_SIZE, + drop_count, + ) prisma_client.spend_log_transactions.append(payload) else: verbose_proxy_logger.debug(