fix(proxy): restore broad is_database_connection_error; add is_database_transport_error for reconnect

Any PrismaError should be treated as a DB connection error for the
allow_requests_on_db_unavailable feature and 503 responses. The narrow
keyword-based check is now in is_database_transport_error, which is
what the reconnect logic in auth_checks.py should use.

Fixes test_delete_access_group_503_on_db_connection_error and
test_handle_authentication_error_db_unavailable failures caused by
PR #21706 narrowing is_database_connection_error.
This commit is contained in:
Ishaan Jaffer 2026-02-21 11:55:34 -08:00
parent d95c3e9cd4
commit 6f71cd46f6
3 changed files with 26 additions and 6 deletions

View file

@ -2000,7 +2000,7 @@ async def _fetch_key_object_from_db_with_reconnect(
proxy_logging_obj=proxy_logging_obj,
)
except Exception as e:
if PrismaDBExceptionHandler.is_database_connection_error(e):
if PrismaDBExceptionHandler.is_database_transport_error(e):
did_reconnect = False
if hasattr(prisma_client, "attempt_db_reconnect"):
auth_reconnect_timeout = getattr(

View file

@ -32,7 +32,28 @@ class PrismaDBExceptionHandler:
@staticmethod
def is_database_connection_error(e: Exception) -> bool:
"""
Returns True if the exception is from a database outage / connection error
Returns True if the exception is from a database outage / connection error.
Any PrismaError qualifies — the DB failed to serve the request.
Used by allow_requests_on_db_unavailable logic and endpoint 503 responses.
"""
import prisma
if isinstance(e, DB_CONNECTION_ERROR_TYPES):
return True
if isinstance(e, prisma.errors.PrismaError):
return True
if isinstance(e, ProxyException) and e.type == ProxyErrorTypes.no_db_connection:
return True
return False
@staticmethod
def is_database_transport_error(e: Exception) -> bool:
"""
Returns True only for transport/connectivity failures where a reconnect
attempt makes sense (e.g. DB is unreachable, connection dropped).
Use this for reconnect logic — data-layer errors like UniqueViolationError
mean the DB IS reachable, so reconnecting would be pointless.
"""
import prisma
@ -44,8 +65,6 @@ class PrismaDBExceptionHandler:
return True
if isinstance(e, prisma.errors.PrismaError):
error_message = str(e).lower()
# Treat generic PrismaError as connection error only when its text
# clearly indicates transport/connectivity failure.
connection_keywords = (
"can't reach database server",
"cannot reach database server",

View file

@ -72,8 +72,9 @@ def test_is_database_connection_error_prisma_connection_errors(prisma_error):
),
],
)
def test_is_database_connection_error_non_connection_prisma_errors(prisma_error):
assert PrismaDBExceptionHandler.is_database_connection_error(prisma_error) == False
def test_is_database_transport_error_non_connection_prisma_errors(prisma_error):
"""Data-layer errors should not trigger reconnect — DB is reachable when these occur."""
assert PrismaDBExceptionHandler.is_database_transport_error(prisma_error) == False
def test_is_database_connection_generic_errors():