From 61f08f99c6842de670c9c35a98d278fadf13e316 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 16 Oct 2026 19:05:17 +0300 Subject: [PATCH 1/2] perf: turn pool_pre_ping off by default and rely on postgres_retry Requires db-retry 0.8.0, whose retry costs about 0.14 us per call, and adds a test that a connection terminated while idle in the pool is recovered by the retry. --- app/settings.py | 2 +- pyproject.toml | 2 +- tests/test_settings.py | 4 +++ tests/use_cases/test_connection_loss.py | 42 +++++++++++++++++++++++-- uv.lock | 18 +++-------- 5 files changed, 49 insertions(+), 19 deletions(-) diff --git a/app/settings.py b/app/settings.py index 50d31a9..66d8324 100644 --- a/app/settings.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 476fa14..e2a018c 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/test_settings.py b/tests/test_settings.py index f4b8669..8af4afa 100644 --- a/tests/test_settings.py +++ b/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 diff --git a/tests/use_cases/test_connection_loss.py b/tests/use_cases/test_connection_loss.py index ad8a19d..8668500 100644 --- a/tests/use_cases/test_connection_loss.py +++ b/tests/use_cases/test_connection_loss.py @@ -5,7 +5,7 @@ import modern_di import pytest import sqlalchemy as sa -from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession from app import ioc from app.actor import Actor @@ -49,8 +49,28 @@ 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_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: + 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()) + killer: typing.Final = await asyncpg.connect( + engine.url.set(drivername="postgresql").render_as_string(hide_password=False) + ) + try: + await killer.execute("SELECT pg_terminate_backend($1)", pid) + finally: + await killer.close() + await asyncio.wait_for(closed.wait(), timeout=5) + + +READ_USE_CASES: typing.Final = pytest.mark.parametrize( ("provider", "kwargs", "expected"), [ (ioc.UseCases.authenticate_user_use_case, {"username": "nobody", "password": "hunter2hunter2"}, None), @@ -59,6 +79,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,6 +93,18 @@ 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 + + @pytest.mark.usefixtures("lost_connection") @pytest.mark.parametrize("provider", [ioc.UseCases.fetch_chat_use_case, ioc.UseCases.fetch_messages_use_case]) async def test_membership_gated_use_case_recovers_from_a_lost_connection( diff --git a/uv.lock b/uv.lock index 491cd33..4ffc39d 100644 --- a/uv.lock +++ b/uv.lock @@ -299,7 +299,7 @@ requires-dist = [ { name = "alembic-postgresql-enum" }, { name = "argon2-cffi" }, { name = "asyncpg" }, - { name = "db-retry", specifier = ">=0.7.1" }, + { name = "db-retry", specifier = ">=0.8.0" }, { name = "granian", extras = ["uvloop"] }, { name = "lite-bootstrap", extras = ["litestar-all"] }, { name = "litestar", extras = ["jwt"] }, @@ -480,16 +480,15 @@ wheels = [ [[package]] name = "db-retry" -version = "0.7.1" +version = "0.8.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "asyncpg" }, { name = "sqlalchemy", extra = ["asyncio"] }, - { name = "tenacity" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/88/39/ab2e1d20e7f45749fcf5da1148aba7fb2961f2b6967d746fa5a5bec1f905/db_retry-0.7.1.tar.gz", hash = "sha256:516595b778d95927d02efc8021440d151b04484afcf973fcc96e74ef8f245a65", size = 8752, upload-time = "2026-10-16T15:17:53.629Z" } +sdist = { url = "https://files.pythonhosted.org/packages/cb/78/71a71ccabcd09175e4351c655b0d1439a6207f9401ce8df4189fa8bfcc84/db_retry-0.8.0.tar.gz", hash = "sha256:dccdc281422f8d01acc2b4f39e622d4a97e03b23adf3d3bf66b61c680c235b71", size = 8737, upload-time = "2026-10-16T15:55:58.283Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/8c/21/97f08a382f6f7684292630becef70e0657212bd7def1b0dc7efa422915ae/db_retry-0.7.1-py3-none-any.whl", hash = "sha256:a98d230c92f4de7b850ae79618698195ef491582d921399ce271ad6a6d33ece9", size = 9326, upload-time = "2026-10-16T15:17:52.641Z" }, + { url = "https://files.pythonhosted.org/packages/d0/1c/3becec5f059559f88c2b4e343673509bac363ebb255743d144da7af35f95/db_retry-0.8.0-py3-none-any.whl", hash = "sha256:edaf2aed767082eb84a6d22b8d79001d0f2c86d68853dda46a623c94fe819ad3", size = 9338, upload-time = "2026-10-16T15:55:57.339Z" }, ] [[package]] @@ -1669,15 +1668,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a9/18/489c97b834dfff9cf2fc2507cede4bcd4b11e67f84bc462acd1992496f86/structlog-26.1.0-py3-none-any.whl", hash = "sha256:e081a26d6c373e6d201eca24eede26d8ffab07f88f477822e679183428d3d91e", size = 73764, upload-time = "2026-06-06T07:33:38.046Z" }, ] -[[package]] -name = "tenacity" -version = "9.2.1" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/82/9e/497c1c8ebe5a5b5d1d4a7511aea22c0bb1a97e3170d98abdef0e1b34265a/tenacity-9.2.1.tar.gz", hash = "sha256:a606b5c808d0cded4a359d5b9932d867ff2a6a6b64d37350260fd01bbdf83839", size = 58261, upload-time = "2026-10-07T12:13:01.633Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/d6/26/1ff2b0721ac66a3ec5b1402b333110b352ab0a8724052ac279a7b82d40c4/tenacity-9.2.1-py3-none-any.whl", hash = "sha256:9e56f17539296baab7beabb08b92f6ee3d7be92d8be72d763360677c2ad6580e", size = 32310, upload-time = "2026-10-07T12:13:00.102Z" }, -] - [[package]] name = "ty" version = "0.0.89" From 689784c95c84c9cbf5da22d900d91ccdb1eb4aa1 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 16 Oct 2026 19:11:22 +0300 Subject: [PATCH 2/2] fix: run every query in create_chat, create_message and mark_read inside the Transaction A lost connection on a read before the block left the session in an invalid transaction that nothing rolled back, so the retry failed with PendingRollbackError. With pool_pre_ping off, an idle connection that died in the pool hits this on the first statement. --- app/use_cases/create_chat.py | 28 ++++---- app/use_cases/create_message.py | 19 +++--- app/use_cases/mark_read.py | 20 +++--- tests/use_cases/test_connection_loss.py | 86 ++++++++++++++++++------- 4 files changed, 99 insertions(+), 54 deletions(-) diff --git a/app/use_cases/create_chat.py b/app/use_cases/create_chat.py index 583f076..1348598 100644 --- a/app/use_cases/create_chat.py +++ b/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( diff --git a/app/use_cases/create_message.py b/app/use_cases/create_message.py index e58c489..babb16f 100644 --- a/app/use_cases/create_message.py +++ b/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( diff --git a/app/use_cases/mark_read.py b/app/use_cases/mark_read.py index 12bdb90..469593a 100644 --- a/app/use_cases/mark_read.py +++ b/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. diff --git a/tests/use_cases/test_connection_loss.py b/tests/use_cases/test_connection_loss.py index 8668500..91897d1 100644 --- a/tests/use_cases/test_connection_loss.py +++ b/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 AsyncEngine, 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,25 +52,23 @@ async def lost_connection(request_container: modern_di.Container) -> None: await asyncio.wait_for(closed.wait(), timeout=5) +@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: - 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()) - killer: typing.Final = await asyncpg.connect( - engine.url.set(drivername="postgresql").render_as_string(hide_password=False) - ) - try: - await killer.execute("SELECT pg_terminate_backend($1)", pid) - finally: - await killer.close() - await asyncio.wait_for(closed.wait(), timeout=5) + watched: typing.Final = await _watch_backend(connection) + await _terminate_backend(engine, *watched) READ_USE_CASES: typing.Final = pytest.mark.parametrize( @@ -105,11 +106,48 @@ async def test_read_use_case_recovers_from_a_connection_lost_while_pooled( 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]))