perf: turn pool_pre_ping off by default and rely on postgres_retry

app/settings.py

@@ -20,7 +20,7 @@ class Settings(pydantic_settings.BaseSettings):
     db_dsn: str = "postgresql+asyncpg://postgres:password@db/postgres"
     db_pool_size: int = 5
     db_max_overflow: int = 0
-    db_pool_pre_ping: bool = True
+    db_pool_pre_ping: bool = False
 
     app_host: str = "0.0.0.0"  # noqa: S104
     app_port: int = 8000

app/use_cases/create_chat.py

@@ -15,6 +15,14 @@
 _DIRECT_CHAT_MEMBER_COUNT: typing.Final = 2
 
 
+def _direct_key(member_ids: set[int]) -> str:
+    if len(member_ids) != _DIRECT_CHAT_MEMBER_COUNT:
+        msg = "A direct chat must have exactly two distinct members"
+        raise ValidationError(msg)
+    low, high = sorted(member_ids)
+    return tables.build_direct_key(low, high)
+
+
 @dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
 class CreateChatUseCase:
     transaction: Transaction
@@ -24,22 +32,16 @@ class CreateChatUseCase:
     @postgres_retry
     async def __call__(self, *, actor: Actor, data: CreateChatRequest) -> tuple[tables.ChatsTable, bool]:
         member_ids: typing.Final = {actor.id, *data.member_ids}
-        direct_key: str | None = None
-
-        if data.chat_type is tables.ChatType.DIRECT:
-            if len(member_ids) != _DIRECT_CHAT_MEMBER_COUNT:
-                msg = "A direct chat must have exactly two distinct members"
-                raise ValidationError(msg)
-            low, high = sorted(member_ids)
-            direct_key = tables.build_direct_key(low, high)
-
-            # Outside the block: __aexit__ rolls back an uncommitted transaction and detaches this.
-            existing = await self.chats_repository.fetch_direct_by_key(direct_key)
-            if existing is not None:
-                return existing, False
+        direct_key: typing.Final = _direct_key(member_ids) if data.chat_type is tables.ChatType.DIRECT else None
 
         chat: tables.ChatsTable | None = None
         async with self.transaction:
+            if direct_key is not None:
+                existing = await self.chats_repository.fetch_direct_by_key(direct_key)
+                if existing is not None:
+                    await self.transaction.commit()
+                    return existing, False
+
             try:
                 chat = await self.chats_repository.create(
                     tables.ChatsTable(

app/use_cases/create_message.py

@@ -24,16 +24,19 @@ class CreateMessageUseCase:
     async def __call__(
         self, *, actor: Actor, chat_id: int, data: SendMessageRequest
     ) -> tuple[tables.MessagesTable, bool]:
-        if not await self.chat_members_repository.is_member(chat_id, actor.id):
-            msg = "Not a member of this chat"
-            raise PermissionDeniedError(msg)
-
-        existing: typing.Final = await self.messages_repository.fetch_by_idempotency_key(chat_id, data.idempotency_key)
-        if existing is not None:
-            return existing, False
-
         message: tables.MessagesTable | None = None
         async with self.transaction:
+            if not await self.chat_members_repository.is_member(chat_id, actor.id):
+                msg = "Not a member of this chat"
+                raise PermissionDeniedError(msg)
+
+            existing: typing.Final = await self.messages_repository.fetch_by_idempotency_key(
+                chat_id, data.idempotency_key
+            )
+            if existing is not None:
+                await self.transaction.commit()
+                return existing, False
+
             try:
                 message = await self.messages_repository.create(
                     tables.MessagesTable(

app/use_cases/mark_read.py

@@ -19,17 +19,19 @@ class MarkReadUseCase:
 
     @postgres_retry
     async def __call__(self, *, actor: Actor, chat_id: int, data: MarkReadRequest) -> tables.ChatMembersTable:
-        member: typing.Final = await self.chat_members_repository.fetch_member(chat_id, actor.id)
-        if member is None:
-            msg = "Not a member of this chat"
-            raise PermissionDeniedError(msg)
+        async with self.transaction:
+            member: typing.Final = await self.chat_members_repository.fetch_member(chat_id, actor.id)
+            if member is None:
+                msg = "Not a member of this chat"
+                raise PermissionDeniedError(msg)
 
-        target_message = await self.messages_repository.get_one_or_none(id=data.last_read_message_id, chat_id=chat_id)
-        if target_message is None:
-            msg = "last_read_message_id does not name a message in this chat"
-            raise ValidationError(msg)
+            target_message = await self.messages_repository.get_one_or_none(
+                id=data.last_read_message_id, chat_id=chat_id
+            )
+            if target_message is None:
+                msg = "last_read_message_id does not name a message in this chat"
+                raise ValidationError(msg)
 
-        async with self.transaction:
             updated = await self.chat_members_repository.mark_read(member.id, data.last_read_message_id)
             await self.transaction.commit()
             # Safe inside the block: commit() ended the transaction, so __aexit__ only closes.

pyproject.toml

@@ -14,7 +14,7 @@ dependencies = [
     "pydantic-settings",
     "granian[uvloop]",
     "argon2-cffi",
-    "db-retry>=0.7.1",
+    "db-retry>=0.8.0",
     # database
     "alembic",
     "alembic-postgresql-enum",

tests/test_settings.py

@@ -15,6 +15,10 @@ def test_sync_db_dsn_parsed_names_the_sync_driver() -> None:
     assert settings.sync_db_dsn_parsed.database == "dbname"
 
 
+def test_pool_pre_ping_is_off_by_default() -> None:
+    assert Settings().db_pool_pre_ping is False
+
+
 def test_api_bootstrapper_config_carries_service_identity() -> None:
     settings = Settings(service_name="svc", service_version="9.9.9")
     config = settings.api_bootstrapper_config

tests/use_cases/test_connection_loss.py

@@ -1,16 +1,20 @@
 import asyncio
 import typing
+import uuid
 
 import asyncpg
 import modern_di
 import pytest
 import sqlalchemy as sa
-from sqlalchemy.ext.asyncio import AsyncSession
+from advanced_alchemy.exceptions import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession
 
 from app import ioc
 from app.actor import Actor
+from app.database import tables
 from app.database.resources import create_database_engine
 from app.exceptions import PermissionDeniedError
+from app.schemas.api import CreateChatRequest, MarkReadRequest, SendMessageRequest
 
 
 UNKNOWN_ACTOR: typing.Final = Actor(id=-1)
@@ -28,19 +32,18 @@ async def request_container(di_container: modern_di.Container) -> typing.AsyncIt
         await engine.dispose()
 
 
-@pytest.fixture
-async def lost_connection(request_container: modern_di.Container) -> None:
-    """Leave the request session holding a pooled connection whose backend was terminated."""
-    session: typing.Final = request_container.resolve_provider(ioc.Database.database_session)
-    assert isinstance(session, AsyncSession)
-    connection: typing.Final = await session.connection()
+async def _watch_backend(connection: AsyncConnection) -> tuple[int, asyncio.Event]:
     pid: typing.Final = await connection.scalar(sa.text("SELECT pg_backend_pid()"))
     driver_connection: typing.Final = (await connection.get_raw_connection()).driver_connection
     assert driver_connection is not None
     closed: typing.Final = asyncio.Event()
     driver_connection.add_termination_listener(lambda _: closed.set())
+    return pid, closed
+
+
+async def _terminate_backend(engine: AsyncEngine, pid: int, closed: asyncio.Event) -> None:
     killer: typing.Final = await asyncpg.connect(
-        connection.engine.url.set(drivername="postgresql").render_as_string(hide_password=False)
+        engine.url.set(drivername="postgresql").render_as_string(hide_password=False)
     )
     try:
         await killer.execute("SELECT pg_terminate_backend($1)", pid)
@@ -49,8 +52,26 @@ async def lost_connection(request_container: modern_di.Container) -> None:
     await asyncio.wait_for(closed.wait(), timeout=5)
 
 
-@pytest.mark.usefixtures("lost_connection")
-@pytest.mark.parametrize(
+@pytest.fixture
+async def lost_connection(request_container: modern_di.Container) -> None:
+    """Leave the request session holding a pooled connection whose backend was terminated."""
+    session: typing.Final = request_container.resolve_provider(ioc.Database.database_session)
+    assert isinstance(session, AsyncSession)
+    connection: typing.Final = await session.connection()
+    await _terminate_backend(connection.engine, *await _watch_backend(connection))
+
+
+@pytest.fixture
+async def lost_pooled_connection(request_container: modern_di.Container) -> None:
+    """Leave the engine's pool holding one idle connection whose backend was terminated."""
+    engine: typing.Final = request_container.resolve_provider(ioc.Database.database_engine)
+    assert isinstance(engine, AsyncEngine)
+    async with engine.connect() as connection:
+        watched: typing.Final = await _watch_backend(connection)
+    await _terminate_backend(engine, *watched)
+
+
+READ_USE_CASES: typing.Final = pytest.mark.parametrize(
     ("provider", "kwargs", "expected"),
     [
         (ioc.UseCases.authenticate_user_use_case, {"username": "nobody", "password": "hunter2hunter2"}, None),
@@ -59,6 +80,10 @@ async def lost_connection(request_container: modern_di.Container) -> None:
         (ioc.UseCases.fetch_user_use_case, {"actor": UNKNOWN_ACTOR}, None),
     ],
 )
+
+
+@pytest.mark.usefixtures("lost_connection")
+@READ_USE_CASES
 async def test_read_use_case_recovers_from_a_lost_connection(
     request_container: modern_di.Container,
     provider: modern_di.providers.Factory[typing.Any],
@@ -69,11 +94,60 @@ async def test_read_use_case_recovers_from_a_lost_connection(
     assert await use_case(**kwargs) == expected
 
 
+@pytest.mark.usefixtures("lost_pooled_connection")
+@READ_USE_CASES
+async def test_read_use_case_recovers_from_a_connection_lost_while_pooled(
+    request_container: modern_di.Container,
+    provider: modern_di.providers.Factory[typing.Any],
+    kwargs: dict[str, typing.Any],
+    expected: object,
+) -> None:
+    use_case: typing.Final = request_container.resolve_provider(provider)
+    assert await use_case(**kwargs) == expected
+
+
+MEMBERSHIP_GATED_USE_CASES: typing.Final = pytest.mark.parametrize(
+    ("provider", "kwargs"),
+    [
+        (ioc.UseCases.fetch_chat_use_case, {}),
+        (ioc.UseCases.fetch_messages_use_case, {}),
+        (
+            ioc.UseCases.create_message_use_case,
+            {"data": SendMessageRequest(idempotency_key=uuid.uuid4(), text="hello")},
+        ),
+        (ioc.UseCases.mark_read_use_case, {"data": MarkReadRequest(last_read_message_id=1)}),
+    ],
+)
+
+
 @pytest.mark.usefixtures("lost_connection")
-@pytest.mark.parametrize("provider", [ioc.UseCases.fetch_chat_use_case, ioc.UseCases.fetch_messages_use_case])
+@MEMBERSHIP_GATED_USE_CASES
 async def test_membership_gated_use_case_recovers_from_a_lost_connection(
-    request_container: modern_di.Container, provider: modern_di.providers.Factory[typing.Any]
+    request_container: modern_di.Container,
+    provider: modern_di.providers.Factory[typing.Any],
+    kwargs: dict[str, typing.Any],
 ) -> None:
     use_case: typing.Final = request_container.resolve_provider(provider)
     with pytest.raises(PermissionDeniedError):
-        await use_case(actor=UNKNOWN_ACTOR, chat_id=-1)
+        await use_case(actor=UNKNOWN_ACTOR, chat_id=-1, **kwargs)
+
+
+@pytest.mark.usefixtures("lost_pooled_connection")
+@MEMBERSHIP_GATED_USE_CASES
+async def test_membership_gated_use_case_recovers_from_a_connection_lost_while_pooled(
+    request_container: modern_di.Container,
+    provider: modern_di.providers.Factory[typing.Any],
+    kwargs: dict[str, typing.Any],
+) -> None:
+    use_case: typing.Final = request_container.resolve_provider(provider)
+    with pytest.raises(PermissionDeniedError):
+        await use_case(actor=UNKNOWN_ACTOR, chat_id=-1, **kwargs)
+
+
+@pytest.mark.usefixtures("lost_pooled_connection")
+async def test_direct_chat_lookup_recovers_from_a_connection_lost_while_pooled(
+    request_container: modern_di.Container,
+) -> None:
+    use_case: typing.Final = request_container.resolve_provider(ioc.UseCases.create_chat_use_case)
+    with pytest.raises(IntegrityError):
+        await use_case(actor=UNKNOWN_ACTOR, data=CreateChatRequest(chat_type=tables.ChatType.DIRECT, member_ids=[1]))