litellm/tests/proxy_unit_tests/test_update_spend.py
Yassin Kortam 40b8300ac2
fix(spend): bound each spend-log write statement by row count as well as bytes (#37758)
The Prisma query engine is a separate process whose resident memory grows with
what it is asked to hold and glibc never returns it, so a pod's memory floor
ratchets up to its worst statement and stays there for the life of the worker.
#34956 bounded a spend-log flush by payload bytes, which caps that floor when
prompts are stored and does nothing when they are not: rows carrying only
attribution metadata run about 1.2 KB, so a 1000-row statement is roughly
1.2 MB, the 2 MB byte budget never binds, and every statement stays at 1000
rows forever.

The engine charges per row as well as per byte. Measured on a container running
the same engine build (5.4.2) against real Postgres, with rows shaped like a
store_prompts_in_spend_logs=false deployment, writing the same 200,000 rows:

  rows/statement   engine RSS still resident after the flush
  1000             179 MB
  500               91 MB
  250               41 MB
  100               19 MB

None of those statements came near the byte budget, so the whole difference is
row count. The floor is a plateau rather than a leak: 1,000,000 rows written at
1000 per statement settles around 229 MB and stops climbing.

Adds SPEND_LOG_WRITE_BATCH_MAX_ROWS, default 100, applied alongside the
existing byte budget so whichever binds first splits the statement. Both are
needed, since bytes are what track a prompt-carrying row and rows are what
track the engine's per-row bookkeeping.

One consequence worth naming: a flush now issues more statements, and a
statement that fails under a poison flood costs one insert before any
isolation runs, so the irreducible floor rises by the statement count. The
isolation budget still caps the amplification on top of that, and the tests
assert the bound derived from the configured row cap rather than a constant.
2026-08-21 09:49:51 -07:00

322 lines
10 KiB
Python

import asyncio
import os
import sys
from unittest.mock import Mock
from litellm.proxy.utils import _get_redoc_url, _get_docs_url
import pytest
from fastapi import Request
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import litellm
from unittest.mock import MagicMock, patch, AsyncMock
import httpx
import math
from litellm.constants import SPEND_LOG_WRITE_BATCH_MAX_ROWS
from litellm.proxy.utils import update_spend
# The flush chunks the queue by BATCH_SIZE and then splits each chunk by the row
# budget, so statement counts below are derived from both rather than hardcoded.
_OUTER_BATCH_SIZE = 1000
def _statements_for(rows: int) -> int:
full, remainder = divmod(rows, _OUTER_BATCH_SIZE)
chunks = [_OUTER_BATCH_SIZE] * full + ([remainder] if remainder else [])
return sum(math.ceil(chunk / SPEND_LOG_WRITE_BATCH_MAX_ROWS) for chunk in chunks)
class MockPrismaClient:
def __init__(self):
# Create AsyncMock for db operations
self.db = AsyncMock()
self.db.litellm_spendlogs = AsyncMock()
self.db.litellm_spendlogs.create_many = AsyncMock()
# Initialize transaction lists
self.spend_log_transactions = []
self.daily_user_spend_transactions = {}
self.tool_usage_transactions = []
self.autorouter_turn_transactions = []
# Add locks for the transaction queues (matches real PrismaClient)
import asyncio
self._spend_log_transactions_lock = asyncio.Lock()
self._tool_usage_transactions_lock = asyncio.Lock()
self._autorouter_turn_transactions_lock = asyncio.Lock()
def jsonify_object(self, obj):
return obj
def add_spend_log_transaction_to_daily_user_transaction(self, payload):
# Mock implementation
pass
def create_mock_proxy_logging():
print("creating mock proxy logging")
proxy_logging_obj = MagicMock()
proxy_logging_obj.failure_handler = AsyncMock()
proxy_logging_obj.db_spend_update_writer = AsyncMock()
proxy_logging_obj.db_spend_update_writer.db_update_spend_transaction_handler = (
AsyncMock()
)
print("returning proxy logging obj")
return proxy_logging_obj
@pytest.mark.asyncio
@pytest.mark.parametrize(
"error_type",
[
httpx.ConnectError("Failed to connect"),
httpx.ReadError("Failed to read response"),
httpx.ReadTimeout("Request timed out"),
],
)
async def test_update_spend_logs_connection_errors(error_type):
"""Test retry mechanism for different connection error types"""
# Setup
prisma_client = MockPrismaClient()
proxy_logging_obj = create_mock_proxy_logging()
# Create AsyncMock for db_spend_update_writer
proxy_logging_obj.db_spend_update_writer = AsyncMock()
proxy_logging_obj.db_spend_update_writer.db_update_spend_transaction_handler = (
AsyncMock()
)
# Add test spend logs
prisma_client.spend_log_transactions = [
{"id": "1", "spend": 10},
{"id": "2", "spend": 20},
]
# Mock the database to fail with connection error twice then succeed
create_many_mock = AsyncMock()
create_many_mock.side_effect = [
error_type, # First attempt fails
error_type, # Second attempt fails
error_type, # Third attempt fails
None, # Fourth attempt succeeds
]
prisma_client.db.litellm_spendlogs.create_many = create_many_mock
# Execute
await update_spend(prisma_client, None, proxy_logging_obj)
# Verify
assert create_many_mock.call_count == 4 # Should have tried 3 times
assert (
len(prisma_client.spend_log_transactions) == 0
) # Should have cleared after success
@pytest.mark.asyncio
@pytest.mark.parametrize(
"error_type",
[
httpx.ConnectError("Failed to connect"),
httpx.ReadError("Failed to read response"),
httpx.ReadTimeout("Request timed out"),
],
)
async def test_update_spend_logs_max_retries_exceeded(error_type):
"""Test that each connection error type properly fails after max retries"""
# Setup
prisma_client = MockPrismaClient()
proxy_logging_obj = create_mock_proxy_logging()
# Add test spend logs
prisma_client.spend_log_transactions = [
{"id": "1", "spend": 10},
{"id": "2", "spend": 20},
]
# Mock the database to always fail
create_many_mock = AsyncMock(side_effect=error_type)
prisma_client.db.litellm_spendlogs.create_many = create_many_mock
# Execute and verify it raises after max retries
with pytest.raises(type(error_type)) as exc_info:
await update_spend(prisma_client, None, proxy_logging_obj)
# Verify error message matches
assert str(exc_info.value) == str(error_type)
# Verify retry attempts (initial try + 4 retries)
assert create_many_mock.call_count == 4
await asyncio.sleep(2)
# Verify failure handler was called
assert proxy_logging_obj.failure_handler.call_count == 1
@pytest.mark.asyncio
async def test_update_spend_logs_non_connection_error():
"""Test handling of non-connection related errors"""
# Setup
prisma_client = MockPrismaClient()
proxy_logging_obj = create_mock_proxy_logging()
# Add test spend logs
prisma_client.spend_log_transactions = [
{"id": "1", "spend": 10},
{"id": "2", "spend": 20},
]
# Mock a different type of error (not connection-related)
unexpected_error = ValueError("Unexpected database error")
create_many_mock = AsyncMock(side_effect=unexpected_error)
prisma_client.db.litellm_spendlogs.create_many = create_many_mock
# Execute and verify it raises immediately without retrying
with pytest.raises(ValueError, match='Unexpected database error') as exc_info:
await update_spend(prisma_client, None, proxy_logging_obj)
# Verify error message
assert str(exc_info.value) == "Unexpected database error"
# Verify only tried once (no retries for non-connection errors)
assert create_many_mock.call_count == 1
# Verify failure handler was called
assert proxy_logging_obj.failure_handler.called
@pytest.mark.asyncio
async def test_update_spend_logs_exponential_backoff():
"""Test that exponential backoff is working correctly"""
# Setup
prisma_client = MockPrismaClient()
proxy_logging_obj = create_mock_proxy_logging()
# Add test spend logs
prisma_client.spend_log_transactions = [{"id": "1", "spend": 10}]
# Track sleep times
sleep_times = []
# Mock asyncio.sleep to track delay times
async def mock_sleep(seconds):
sleep_times.append(seconds)
# Mock the database to fail with connection errors
create_many_mock = AsyncMock(
side_effect=[
httpx.ConnectError("Failed to connect"), # First attempt
httpx.ConnectError("Failed to connect"), # Second attempt
None, # Third attempt succeeds
]
)
prisma_client.db.litellm_spendlogs.create_many = create_many_mock
# Apply mocks
with patch("asyncio.sleep", mock_sleep):
await update_spend(prisma_client, None, proxy_logging_obj)
# Verify exponential backoff
assert len(sleep_times) == 2 # Should have slept twice
assert (
sleep_times[0] >= 1 and sleep_times[0] <= 2
) # First retry after 2^0~2^1 seconds
assert (
sleep_times[1] >= 2 and sleep_times[1] <= 4
) # Second retry after 2^1~2^2 seconds
@pytest.mark.asyncio
async def test_update_spend_logs_multiple_batches_success():
"""
Test successful processing of multiple batches of spend logs
Code sets batch size to 1000. This test creates 1500 logs, so it should make 2 batches.
"""
# Setup
prisma_client = MockPrismaClient()
proxy_logging_obj = create_mock_proxy_logging()
# Create 1500 test spend logs (1.5x BATCH_SIZE)
prisma_client.spend_log_transactions = [
{"id": str(i), "spend": 10} for i in range(1500)
]
create_many_mock = AsyncMock(return_value=None)
prisma_client.db.litellm_spendlogs.create_many = create_many_mock
# Execute
await update_spend(prisma_client, None, proxy_logging_obj)
# Verify
assert create_many_mock.call_count == _statements_for(1500)
# No statement may exceed the row budget, which is what bounds the query
# engine's resident memory.
batches = [call[1]["data"] for call in create_many_mock.call_args_list]
assert all(len(batch) <= SPEND_LOG_WRITE_BATCH_MAX_ROWS for batch in batches)
# Every row is written exactly once and in order, whatever the split.
written_ids = [item["id"] for batch in batches for item in batch]
assert written_ids == [str(i) for i in range(1500)]
# Verify all logs were processed
assert len(prisma_client.spend_log_transactions) == 0
@pytest.mark.asyncio
async def test_update_spend_logs_multiple_batches_with_failure():
"""
Test processing of multiple batches where one batch fails.
Creates 4000 logs (4 batches) with one batch failing but eventually succeeding after retry.
"""
# Setup
prisma_client = MockPrismaClient()
proxy_logging_obj = create_mock_proxy_logging()
# Create 4000 test spend logs (4x BATCH_SIZE)
prisma_client.spend_log_transactions = [
{"id": str(i), "spend": 10} for i in range(4000)
]
# Mock to fail on second batch first attempt, then succeed
call_count = 0
async def create_many_side_effect(**kwargs):
nonlocal call_count
call_count += 1
# Fail on the second batch's first attempt
if call_count == 2:
raise httpx.ConnectError("Failed to connect")
return None
create_many_mock = AsyncMock(side_effect=create_many_side_effect)
prisma_client.db.litellm_spendlogs.create_many = create_many_mock
# Execute
await update_spend(prisma_client, None, proxy_logging_obj)
# The first attempt aborts on its second statement, then the whole flush
# replays, so the total is those two calls plus one complete pass.
assert create_many_mock.call_count == 2 + _statements_for(4000)
# Verify all batches were processed
all_processed_logs = []
for call in create_many_mock.call_args_list:
all_processed_logs.extend(call[1]["data"])
# Verify all IDs were processed
processed_ids = {item["id"] for item in all_processed_logs}
# these should have ids 0-3999
print("all processed ids", sorted(processed_ids, key=int))
expected_ids = {str(i) for i in range(4000)}
assert processed_ids == expected_ids
# Verify all logs were cleared from transactions
assert len(prisma_client.spend_log_transactions) == 0