fix(proxy): spend reads wait for the pool, keep stricter statement_timeout, and classify cancels honestly

query_raw_within_statement_timeout now passes Prisma max_wait equal to the URL's pool_timeout, only lowers statement_timeout when the session's is not already stricter, and reports the limit the read actually ran under. is_statement_timeout_error no longer treats every SQLSTATE 57014 as a statement timeout, so pg_cancel_backend raises as the generic error it is. Prisma P2028 and the in-transaction P2024 message classify as database capacity errors.
This commit is contained in:
devin-ai-integration[bot] 2026-11-04 15:50:52 +00:00
parent 03c26c6e07
commit f0d33aae24
12 changed files with 229 additions and 36 deletions

View file

@ -44,12 +44,13 @@ import sys
import tempfile
import urllib.parse
from collections.abc import Callable, Mapping, Sequence
from datetime import timedelta
from functools import partial
from pathlib import Path
from types import MappingProxyType
from typing import Annotated, Final, Protocol, TypeAlias, cast
from pydantic import AliasChoices, BeforeValidator, Field
from pydantic import AliasChoices, BeforeValidator, Field, TypeAdapter, ValidationError
from pydantic_settings import BaseSettings, SettingsConfigDict
from litellm.proxy.db.pgbouncer import database_url_is_pooled
@ -300,6 +301,23 @@ def connection_params_from_url(url: str) -> Mapping[str, str | int | float]:
)
PRISMA_DEFAULT_POOL_TIMEOUT: Final = timedelta(seconds=10)
_POOL_TIMEOUT_SECONDS: Final = TypeAdapter(float)
def pool_timeout_from_url(url: str) -> timedelta:
"""How long Prisma waits for a pooled connection on ``url``: its ``pool_timeout`` param, else Prisma's default.
``pool_timeout=0`` tells Prisma to wait forever, which a transaction's ``max_wait`` cannot express, so it falls
back to the default too.
"""
try:
seconds: Final = _POOL_TIMEOUT_SECONDS.validate_python(connection_params_from_url(url).get("pool_timeout"))
except ValidationError:
return PRISMA_DEFAULT_POOL_TIMEOUT
return timedelta(seconds=seconds) if seconds > 0 else PRISMA_DEFAULT_POOL_TIMEOUT
# A re-minted token URL replaces a URL to the same database, so unlike the
# reader allowlist it may also carry ``options``: that is where the server-side
# timeouts (statement, lock, idle-in-transaction) live, and a refresh that

View file

@ -30,11 +30,12 @@ _CONNECTION_CAPACITY_PHRASES: Final = (
"too many connections for role",
"too many connections for database",
"remaining connection slots are reserved",
"timed out fetching a new connection from the connection pool",
"unable to start a transaction in the given time",
)
_PRISMA_POOL_TIMEOUT_CODE: Final = "P2024"
_LOCK_TIMEOUT_SQLSTATE: Final = "55P03"
_LOCK_TIMEOUT_PHRASE: Final = "canceling statement due to lock timeout"
_STATEMENT_TIMEOUT_SQLSTATE: Final = "57014"
_STATEMENT_TIMEOUT_PHRASE: Final = "canceling statement due to statement timeout"
@ -274,10 +275,11 @@ class PrismaDBExceptionHandler:
def is_database_capacity_error(e: Exception) -> bool:
"""True iff Postgres refused the pool a new connection (SQLSTATE 53300:
``too many clients already``, a reserved slot, or a per-role limit) or
prisma's own pool timed out handing one over (P2024). The server is up
but full, so the failure is neither a transport error (which would tear
the engine down and open yet more connections against it) nor a
rejection of the rows being written."""
prisma's own pool timed out handing one over (P2024 for a query, P2028
``Unable to start a transaction in the given time`` for a transaction).
The server is up but full, so the failure is neither a transport error
(which would tear the engine down and open yet more connections against
it) nor a rejection of the rows being written."""
import prisma
if not isinstance(e, _exception_types(prisma.errors.PrismaError)):
@ -306,13 +308,14 @@ class PrismaDBExceptionHandler:
@staticmethod
def is_statement_timeout_error(e: Exception) -> bool:
"""True iff Postgres cancelled a statement for exceeding ``statement_timeout`` (SQLSTATE 57014)."""
"""True iff Postgres cancelled a statement for exceeding ``statement_timeout``.
Only the message decides: its SQLSTATE 57014 is the same one
``pg_cancel_backend`` and a client-side cancel raise."""
import prisma
if not isinstance(e, _exception_types(prisma.errors.PrismaError)):
return False
if PrismaDBExceptionHandler.postgres_sqlstate(e) == _STATEMENT_TIMEOUT_SQLSTATE:
return True
return _STATEMENT_TIMEOUT_PHRASE in str(e).lower()
@staticmethod

View file

@ -20,7 +20,11 @@ from pydantic import TypeAdapter
from litellm._logging import verbose_proxy_logger
from litellm.proxy.db.db_span import db_span
from litellm.proxy.db.db_url_settings import add_missing_query_params, token_refresh_params_from_url
from litellm.proxy.db.db_url_settings import (
add_missing_query_params,
pool_timeout_from_url,
token_refresh_params_from_url,
)
from litellm.proxy.db.log_db_metrics import db_io_claimed, record_db_io
from litellm.proxy.db.prisma_query_span import parse_prisma_query
from litellm.proxy.db.token_auth import (
@ -904,6 +908,11 @@ class PrismaWrapper:
async def disconnect(self, timeout: float | timedelta | None = None) -> None: ...
@property
def pool_timeout(self) -> timedelta:
"""How long this wrapper's engine waits for a pooled connection, read from its URL's ``pool_timeout``."""
return pool_timeout_from_url(os.getenv(self._db_url_env_var, ""))
def __getattr__(self, name: str):
"""
Proxy attribute access to the underlying Prisma client.

View file

@ -5315,6 +5315,7 @@ _PRISMA_DEFAULT_TX_TIMEOUT: Final = timedelta(seconds=5)
_PRISMA_DEFAULT_TX_MAX_WAIT: Final = timedelta(seconds=2)
_SPEND_LOG_READ_STATEMENT_TIMEOUT: Final = timedelta(seconds=SPEND_LOG_READ_STATEMENT_TIMEOUT_SECONDS)
_STATEMENT_TIMEOUT_TRANSACTION_GRACE: Final = timedelta(seconds=5)
_STATEMENT_TIMEOUT_MS: Final = TypeAdapter(int)
class _ReadTransactionClient(Protocol):
@ -5322,6 +5323,27 @@ class _ReadTransactionClient(Protocol):
def read_db(self) -> PrismaWrapper: ...
class _RawReadTransaction(Protocol):
async def query_raw(self, query: str, *args: object) -> Sequence[Mapping[str, object]]: ...
async def execute_raw(self, query: str, *args: object) -> int: ...
async def _lower_statement_timeout(transaction: _RawReadTransaction, timeout_ms: int) -> int:
"""``SET LOCAL statement_timeout`` to ``timeout_ms`` unless the session already runs under a stricter one.
Returns the limit the read runs under, in milliseconds. 0 means the session has none.
"""
rows: Final = await transaction.query_raw(
"SELECT setting::int AS ms FROM pg_settings WHERE name = 'statement_timeout'"
)
current_ms: Final = _STATEMENT_TIMEOUT_MS.validate_python(next(iter(rows), {}).get("ms", 0))
if 0 < current_ms <= timeout_ms:
return current_ms
await transaction.execute_raw(f"SET LOCAL statement_timeout = {timeout_ms}")
return timeout_ms
async def query_raw_within_statement_timeout(
prisma_client: _ReadTransactionClient,
sql: str,
@ -5333,19 +5355,24 @@ async def query_raw_within_statement_timeout(
Prisma's HTTP client gives up on a statement after 30s anyway, so without
this the Admin UI's error is the same but Postgres keeps running the scan.
A cancelled read surfaces as a 503 asking for a narrower date range.
A cancelled read surfaces as a 503 asking for a narrower date range. The
transaction waits for a pooled connection as long as a plain read would,
and never loosens a ``statement_timeout`` the role or database already set.
"""
try:
async with prisma_client.read_db.tx(timeout=timeout + _STATEMENT_TIMEOUT_TRANSACTION_GRACE) as transaction:
await transaction.execute_raw(f"SET LOCAL statement_timeout = {int(timeout / timedelta(milliseconds=1))}")
read_db: Final = prisma_client.read_db
async with read_db.tx(
timeout=timeout + _STATEMENT_TIMEOUT_TRANSACTION_GRACE, max_wait=read_db.pool_timeout
) as transaction:
effective_ms: Final = await _lower_statement_timeout(transaction, int(timeout / timedelta(milliseconds=1)))
try:
return await transaction.query_raw(sql, *params)
except Exception as e:
if not PrismaDBExceptionHandler.is_statement_timeout_error(e):
raise
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail={"error": f"{what} took longer than {int(timeout.total_seconds())}s. Pick a shorter date range"},
) from e
except Exception as e:
if not PrismaDBExceptionHandler.is_statement_timeout_error(e):
raise
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail={"error": f"{what} took longer than {effective_ms / 1000:g}s. Pick a shorter date range"},
) from e
async def _lookup_deprecated_key(

View file

@ -77,7 +77,7 @@ def build_prisma(query_raw: AsyncMock) -> MagicMock:
def _spend_log_reads(prisma: MagicMock) -> AsyncMock:
"""The raw reads the cache dashboard ran under the statement timeout."""
return prisma.read_db.tx.return_value.query_raw
return prisma.read_db.tx.return_value.reads
def dispatching_query_raw() -> AsyncMock:

View file

@ -2,6 +2,7 @@
marking the DB I/O witness the way ``_TrackedPrismaEngine`` does, so the producer under test
emits its service event."""
from datetime import timedelta
from typing import Final, TypeVar
from unittest.mock import AsyncMock, MagicMock
@ -21,11 +22,20 @@ def engine_call(return_value: _T | None = None) -> AsyncMock:
class RecordingReadTransaction:
"""The transaction ``query_raw_within_statement_timeout`` opens on ``read_db``, keeping its ``SET LOCAL`` statements."""
"""The transaction ``query_raw_within_statement_timeout`` opens on ``read_db``, keeping its ``SET LOCAL`` statements.
def __init__(self, query_raw: AsyncMock) -> None:
The helper's ``pg_settings`` probe sees ``statement_timeout_ms`` as the session's current ``statement_timeout``
(0 means unlimited), every other raw read is answered by ``reads``."""
def __init__(self, reads: AsyncMock, statement_timeout_ms: int = 0) -> None:
self.statements: tuple[str, ...] = ()
self.query_raw: AsyncMock = query_raw
self.reads: AsyncMock = reads
self.statement_timeout_ms: Final = statement_timeout_ms
async def query_raw(self, sql: str, *args: object) -> object:
if "FROM pg_settings" in sql:
return [{"ms": self.statement_timeout_ms}]
return await self.reads(sql, *args)
async def execute_raw(self, sql: str) -> int:
self.statements = (*self.statements, sql)
@ -38,9 +48,11 @@ class RecordingReadTransaction:
pass
def statement_timeout_reads(prisma: MagicMock, query_raw: AsyncMock) -> RecordingReadTransaction:
def statement_timeout_reads(
prisma: MagicMock, query_raw: AsyncMock, statement_timeout_ms: int = 0
) -> RecordingReadTransaction:
"""Make ``prisma.read_db.tx()`` hand out one recording transaction whose raw reads answer with ``query_raw``."""
transaction: Final = RecordingReadTransaction(query_raw)
transaction: Final = RecordingReadTransaction(query_raw, statement_timeout_ms)
prisma.read_db.tx = MagicMock(return_value=transaction)
return transaction
@ -71,6 +83,8 @@ class _DelegatingReadTransaction:
return 0
async def query_raw(self, sql: str, *args: object) -> object:
if "FROM pg_settings" in sql:
return [{"ms": 0}]
return await getattr(self.db, "query_raw")(sql, *args)
async def __aenter__(self) -> "_DelegatingReadTransaction":
@ -83,6 +97,8 @@ class _DelegatingReadTransaction:
class ReadTargetOf:
"""What a prisma fake returns from ``read_db``: ``tx()`` opens a transaction over the fake's ``db``."""
pool_timeout: Final = timedelta(seconds=10)
def __init__(self, db: object) -> None:
self.db: Final = db

View file

@ -33,7 +33,9 @@ from pydantic import ValidationError
from litellm.proxy.db.db_url_settings import (
PG_SSL_REQUEST,
PRISMA_DEFAULT_POOL_TIMEOUT,
DatabaseURLSettings,
pool_timeout_from_url,
token_refresh_params_from_url,
translate_libpq_ssl_params,
unsupported_db_scheme,
@ -1314,3 +1316,18 @@ def test_connection_budget_message_survives_an_unparseable_limit() -> None:
message: Final = postgres_connection_budget_message(writer_limit="ten", reader_limit=None, num_workers="2")
assert "writer='ten'" in message
@pytest.mark.parametrize(
("url", "expected"),
[
("postgresql://u:p@db:5432/litellm?connection_limit=10&pool_timeout=60", datetime.timedelta(seconds=60)),
("postgresql://u:p@db:5432/litellm?pool_timeout=2.5", datetime.timedelta(seconds=2.5)),
("postgresql://u:p@db:5432/litellm?connection_limit=10", PRISMA_DEFAULT_POOL_TIMEOUT),
("postgresql://u:p@db:5432/litellm?pool_timeout=0", PRISMA_DEFAULT_POOL_TIMEOUT),
("postgresql://u:p@db:5432/litellm?pool_timeout=soon", PRISMA_DEFAULT_POOL_TIMEOUT),
],
ids=["set", "fractional", "absent", "zero_waits_forever", "garbage"],
)
def test_pool_timeout_from_url(url: str, expected: datetime.timedelta) -> None:
assert pool_timeout_from_url(url) == expected

View file

@ -842,11 +842,27 @@ def _pool_timeout_error() -> DataError:
)
def _transaction_start_timeout_error() -> EngineRequestError:
"""The shape prisma raises when ``tx()`` could not get a pooled connection within ``max_wait`` (P2028)."""
return EngineRequestError(
MagicMock(status=504),
'{"is_panic":false,"message":"Transaction API error: Unable to start a transaction in the given time.",'
'"error_code":"P2028"}',
)
@pytest.mark.parametrize(
"error",
[
_capacity_error("sorry, too many clients already"),
_pool_timeout_error(),
_transaction_start_timeout_error(),
EngineRequestError(
MagicMock(status=500),
'{"is_panic":false,"message":"Timed out fetching a new connection from the connection pool. More info: '
'http://pris.ly/d/connection-pool (Current connection pool timeout: 1, connection limit: 10)",'
'"error_code":"P2024"}',
),
EngineRequestError(
MagicMock(status=500),
'{"is_panic":false,"message":"Error in connector: Error querying the database: FATAL: sorry, too many clients already","backtrace":null}',
@ -861,7 +877,14 @@ def _pool_timeout_error() -> DataError:
}
),
],
ids=["53300_connector_dataerror", "P2024_pool_timeout", "engine_500", "raw_query_53300"],
ids=[
"53300_connector_dataerror",
"P2024_pool_timeout",
"P2028_transaction_start",
"P2024_inside_transaction",
"engine_500",
"raw_query_53300",
],
)
def test_is_database_capacity_error_recognises_postgres_and_pool_exhaustion(error: Exception) -> None:
"""SQLSTATE 53300 and prisma P2024 mean the statement was never sent because
@ -958,11 +981,21 @@ def test_is_lock_timeout_error_excludes_other_failures(error: Exception) -> None
[
(_lock_timeout_error(message="canceling statement due to statement timeout", code="57014"), True),
(_lock_timeout_error(message="canceling statement due to statement timeout", code=None), True),
(_lock_timeout_error(message="canceling statement due to user request", code="57014"), False),
(_lock_timeout_error(), False),
(_capacity_error("sorry, too many clients already"), False),
(_transaction_start_timeout_error(), False),
(RuntimeError("canceling statement due to statement timeout"), False),
],
ids=["meta_57014", "message_only", "lock_timeout_55P03", "capacity_53300", "not_prisma"],
ids=[
"meta_57014",
"message_only",
"pg_cancel_backend_57014",
"lock_timeout_55P03",
"capacity_53300",
"P2028_transaction_start",
"not_prisma",
],
)
def test_is_statement_timeout_error(error: Exception, expected: bool) -> None:
assert PrismaDBExceptionHandler.is_statement_timeout_error(error) is expected

View file

@ -602,3 +602,16 @@ def test_migration_dry_run_summary_needs_database_url(unset_database_url):
from litellm.proxy.db.prisma_client import PrismaManager
assert PrismaManager.migration_dry_run_summary(use_migrate=True) == (False, "DATABASE_URL is not set.")
def test_prisma_wrapper_pool_timeout_comes_from_its_own_database_url(monkeypatch: pytest.MonkeyPatch) -> None:
from datetime import timedelta
monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@primary:5432/litellm?pool_timeout=60")
monkeypatch.setenv("DATABASE_URL_READ_REPLICA", "postgresql://u:p@replica:5432/litellm?pool_timeout=7")
wrapper: Final = PrismaWrapper(
original_prisma=MagicMock(), iam_token_db_auth=False, db_url_env_var="DATABASE_URL_READ_REPLICA"
)
assert wrapper.pool_timeout == timedelta(seconds=7)

View file

@ -17307,7 +17307,7 @@ async def test_get_team_spend_by_user_admin_groups_spend_logs_by_team_and_user(m
assert any("statement_timeout = 30000" in statement for statement in transaction.statements)
mock_db_client.db.tx.assert_not_called()
sql, *params = transaction.query_raw.call_args.args
sql, *params = transaction.reads.call_args.args
assert params == ["2026-09-01", "2026-09-04", "team-alpha", "team-beta"]
assert 'FROM "LiteLLM_SpendLogs" sl' in sql
assert 'sl."startTime" >= $1::timestamp' in sql
@ -17343,7 +17343,7 @@ async def test_get_team_spend_by_user_team_admin_sees_every_member(mock_db_clien
user_api_key_dict=caller, team_ids="team-alpha", start_date="2026-09-01", end_date="2026-09-04"
)
sql, *params = transaction.query_raw.call_args.args
sql, *params = transaction.reads.call_args.args
assert params == ["2026-09-01", "2026-09-04", "team-alpha"]
assert 'sl."user" = $' not in sql
@ -17367,7 +17367,7 @@ async def test_get_team_spend_by_user_plain_member_only_sees_own_row(mock_db_cli
user_api_key_dict=caller, team_ids="team-alpha", start_date="2026-09-01", end_date="2026-09-04"
)
sql, *params = transaction.query_raw.call_args.args
sql, *params = transaction.reads.call_args.args
assert params == ["2026-09-01", "2026-09-04", "team-alpha", "bob"]
assert "sl.team_id IN ($3)" in sql
assert 'AND sl."user" = $4' in sql
@ -17390,7 +17390,7 @@ async def test_get_team_spend_by_user_member_of_other_team_gets_404(mock_db_clie
)
assert exc_info.value.status_code == 404
transaction.query_raw.assert_not_called()
transaction.reads.assert_not_called()
@pytest.mark.asyncio
@ -17418,7 +17418,7 @@ async def test_get_team_spend_by_user_rejects_bad_input(mock_db_client, team_ids
assert exc_info.value.status_code == 400
assert expected_error in str(exc_info.value.detail)
transaction.query_raw.assert_not_called()
transaction.reads.assert_not_called()
@pytest.mark.asyncio

View file

@ -205,7 +205,7 @@ async def test_spend_logs_ui_wraps_params_in_at_time_zone_utc(monkeypatch):
def _spend_log_reads(mock_prisma: MagicMock) -> AsyncMock:
"""The raw reads ``/spend/logs/ui`` ran under the statement timeout."""
return mock_prisma.read_db.tx.return_value.query_raw
return mock_prisma.read_db.tx.return_value.reads
def _make_ui_spend_logs_mock(count_total, page_rows):

View file

@ -3734,10 +3734,67 @@ async def test_query_raw_within_statement_timeout_has_postgres_cancel_the_read_a
assert rows == [{"spend": 1.5}]
assert transaction.statements == ("SET LOCAL statement_timeout = 30000",)
transaction.query_raw.assert_awaited_once_with('SELECT spend FROM "LiteLLM_SpendLogs" WHERE "user" = $1', "u1")
transaction.reads.assert_awaited_once_with('SELECT spend FROM "LiteLLM_SpendLogs" WHERE "user" = $1', "u1")
assert prisma.read_db.tx.call_args.kwargs["timeout"] > timedelta(seconds=30), "Prisma must outlive the Postgres cancel"
@pytest.mark.asyncio
async def test_query_raw_within_statement_timeout_waits_for_a_pooled_connection_as_long_as_a_plain_read() -> None:
"""A transaction start must not give up on the pool before a plain ``query_raw`` would, or a busy pool turns
25 concurrent spend reads into Prisma P2028 failures that the plain path served fine."""
from datetime import timedelta
from litellm.proxy.utils import query_raw_within_statement_timeout
from tests.unit.proxy.db.fake_prisma_engine import statement_timeout_reads
prisma: Final = MagicMock()
prisma.read_db.pool_timeout = timedelta(seconds=60)
statement_timeout_reads(prisma, AsyncMock(return_value=[]))
await query_raw_within_statement_timeout(prisma, "SELECT 1")
assert prisma.read_db.tx.call_args.kwargs["max_wait"] == timedelta(seconds=60)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("session_statement_timeout_ms", "expected_statements"),
[
(0, ("SET LOCAL statement_timeout = 30000",)),
(60_000, ("SET LOCAL statement_timeout = 30000",)),
(30_000, ()),
(500, ()),
],
ids=["unlimited", "looser", "equal", "stricter"],
)
async def test_query_raw_within_statement_timeout_only_lowers_the_session_statement_timeout(
session_statement_timeout_ms: int, expected_statements: tuple[str, ...]
) -> None:
from litellm.proxy.utils import query_raw_within_statement_timeout
from tests.unit.proxy.db.fake_prisma_engine import statement_timeout_reads
prisma: Final = MagicMock()
transaction: Final = statement_timeout_reads(prisma, AsyncMock(return_value=[]), session_statement_timeout_ms)
await query_raw_within_statement_timeout(prisma, "SELECT 1")
assert transaction.statements == expected_statements
@pytest.mark.asyncio
async def test_query_raw_within_statement_timeout_reports_the_stricter_limit_the_read_ran_under() -> None:
from litellm.proxy.utils import query_raw_within_statement_timeout
from tests.unit.proxy.db.fake_prisma_engine import statement_timeout_error, statement_timeout_reads
prisma: Final = MagicMock()
statement_timeout_reads(prisma, AsyncMock(side_effect=statement_timeout_error()), statement_timeout_ms=500)
with pytest.raises(HTTPException) as exc_info:
await query_raw_within_statement_timeout(prisma, "SELECT 1", what="Spend per user")
assert exc_info.value.detail == {"error": "Spend per user took longer than 0.5s. Pick a shorter date range"}
@pytest.mark.asyncio
async def test_query_raw_within_statement_timeout_reports_a_cancelled_read_as_503() -> None:
from litellm.proxy.utils import query_raw_within_statement_timeout