perf: send a message with one INSERT ... ON CONFLICT DO UPDATE

AGENTS.md

@@ -135,10 +135,12 @@ back. `app`/`client` fixtures build the real app and run it through
 container_fixture="request_container")` (`tests/use_cases/conftest.py`)
 exposes every repository/use case provider as a same-named pytest fixture —
 the template predates this and hand-assembles dependencies instead. The
-race-simulation pattern used to test the concurrent-retry paths without a second
+race-simulation pattern used to test the concurrent-retry path without a second
 real connection is the `_Racing*Repository` classes in
-`tests/use_cases/test_create_chat.py` and `tests/use_cases/test_create_message.py`;
-the invariant each one pins is in the docstring on the test that uses it.
+`tests/use_cases/test_create_chat.py`. `tests/use_cases/test_create_message.py`
+uses a second real connection instead: it holds an uncommitted insert and polls
+`pg_stat_activity` until the use case waits on its lock. The invariant each one
+pins is in the docstring on the test that uses it.
 
 **Migrations**: `migrations/env.py` reads the shared `METADATA` and rewrites
 the DSN driver from `postgresql+asyncpg` → `postgresql` (Alembic uses sync

app/repositories/messages_repository.py

@@ -6,6 +6,7 @@
 from advanced_alchemy.filters import LimitOffset
 from advanced_alchemy.repository import SQLAlchemyAsyncRepository
 from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService
+from sqlalchemy.dialects import postgresql
 
 from app.database import tables
 
@@ -16,8 +17,22 @@ class BaseRepository(SQLAlchemyAsyncRepository[tables.MessagesTable]):
 
     repository_type = BaseRepository
 
-    async def fetch_by_idempotency_key(self, chat_id: int, idempotency_key: uuid.UUID) -> tables.MessagesTable | None:
-        return await self.get_one_or_none(chat_id=chat_id, idempotency_key=idempotency_key)
+    async def get_or_create(
+        self, *, chat_id: int, user_id: int, idempotency_key: uuid.UUID, text: str
+    ) -> tuple[tables.MessagesTable, bool]:
+        statement = postgresql.insert(tables.MessagesTable).values(
+            chat_id=chat_id, user_id=user_id, idempotency_key=idempotency_key, text=text
+        )
+        statement = statement.on_conflict_do_update(
+            constraint="uk_messages_chat_id_idempotency_key",
+            set_={"idempotency_key": statement.excluded.idempotency_key},
+        )
+        # xmax is 0 only on a row version this statement inserted.
+        created: typing.Final = sa.literal_column("xmax = 0", sa.Boolean)
+        message, was_created = (
+            await self.repository.session.execute(statement.returning(tables.MessagesTable, created))
+        ).one()
+        return message, was_created
 
     async def fetch_latest_active(self, chat_id: int) -> tables.MessagesTable | None:
         """Return the newest non-deleted message in a chat, or None if none remains.

app/use_cases/create_message.py

@@ -1,7 +1,5 @@
 import dataclasses
-import typing
 
-from advanced_alchemy.exceptions import DuplicateKeyError
 from db_retry import Transaction, postgres_retry
 
 from app.actor import Actor
@@ -24,44 +22,19 @@ class CreateMessageUseCase:
     async def __call__(
         self, *, actor: Actor, chat_id: int, data: SendMessageRequest
     ) -> tuple[tables.MessagesTable, bool]:
-        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
+            message, created = await self.messages_repository.get_or_create(
+                chat_id=chat_id, user_id=actor.id, idempotency_key=data.idempotency_key, text=data.text
             )
-            if existing is not None:
-                await self.transaction.commit()
-                return existing, False
-
-            try:
-                message = await self.messages_repository.create(
-                    tables.MessagesTable(
-                        chat_id=chat_id,
-                        user_id=actor.id,
-                        idempotency_key=data.idempotency_key,
-                        text=data.text,
-                    )
-                )
-            except DuplicateKeyError:
-                # Re-read outside the block: __aexit__ rolls back and detaches on an open transaction.
-                await self.transaction.rollback()
-            else:
+            if created:
                 await self.chats_repository.update(
                     tables.ChatsTable(id=chat_id, last_message_id=message.id),
                     item_id=chat_id,
                     attribute_names=["last_message_id"],
                 )
-                await self.transaction.commit()
-
-        if message is None:
-            duplicate = await self.messages_repository.fetch_by_idempotency_key(chat_id, data.idempotency_key)
-            if duplicate is None:
-                msg = "Message send raced but the resulting row could not be found"
-                raise RuntimeError(msg)
-            return duplicate, False
-
-        return message, True
+            await self.transaction.commit()
+        return message, created

docs/adr/0005-upsert-via-duplicate-key-recovery.md

@@ -1,7 +1,7 @@
 # Upsert by recovering from DuplicateKeyError
 
-`CreateChatUseCase` and `CreateMessageUseCase` both read first to see whether
-the row already exists and then insert, but the read is only an optimisation:
+`CreateChatUseCase` reads first to see whether the direct chat already exists
+and then inserts, but the read is only an optimisation:
 the correctness guarantee is the `except DuplicateKeyError` branch, which rolls
 back and re-reads the row the winner committed. The pre-check alone does not
 make the insert safe. At READ COMMITTED two concurrent "open a DM with Bob"
@@ -10,7 +10,8 @@ requests both miss the read, both insert, and the loser violates
 have received the existing chat. `@postgres_retry` does not rescue it either,
 because `db-retry` retries serialization and connection failures, not integrity
 violations, and `SELECT ... FOR UPDATE` has nothing to take, since the race is
-between two inserts of a row that does not yet exist. Both recovery paths are
-exercised by repository subclasses (`_RacingChatsRepository`,
-`_RacingMessagesRepository`) whose `create` raises and whose lookup misses once,
-simulating the database condition at a seam the tests already own.
+between two inserts of a row that does not yet exist. The recovery path is
+exercised by a repository subclass, `_RacingChatsRepository`, whose `create`
+raises and whose lookup misses once, simulating the database condition at a seam
+the tests already own. Message sends used this pattern too until
+[0012](0012-message-send-is-one-upsert.md) replaced it with a single statement.

docs/adr/0006-idempotency-scoped-per-chat.md

@@ -1,15 +1,14 @@
 # Idempotency is scoped per chat
 
 `messages` carries `UniqueConstraint("chat_id", "idempotency_key")`, and the
-pre-check lookup, the constraint and the `DuplicateKeyError` recovery re-read
-all filter on that same pair. Idempotency is a property of an operation, and the
+send's `ON CONFLICT` names that constraint. Idempotency is a property of an operation, and the
 operation is "send this message to this chat", so two chats are two operations
 and a key reused across them is not a retry of anything. A global constraint on
 `idempotency_key` shipped first and made a reachable state look unreachable:
 cross-chat key reuse under concurrency raised `DuplicateKeyError` from the
 insert, the chat-scoped re-read missed, and control reached a guard whose only
 justification was "the unique constraint guarantees a match here". Worse, a
 client reusing a key across chats received the other chat's message while its
-intended message was never written, a silent wrong-row return. The three
-surfaces must agree, and aligning the constraint with the lookup was both the
-cheaper direction and the one that matches the domain.
+intended message was never written, a silent wrong-row return. The surfaces
+had to agree, and aligning the constraint with the lookup was both the cheaper
+direction and the one that matches the domain.

docs/adr/0009-coverage-exclusions-structural-only.md

@@ -7,8 +7,8 @@ explain itself, plus `exclude_also` for `if typing.TYPE_CHECKING:` blocks. No
 `# pragma: no cover` appears in `app/`, `tests/` or `migrations/`. "Awkward to
 reach" is not a warrant: where a branch looked untestable the answer was a
 repository subclass that raises the condition the database would raise, which is
-how both `DuplicateKeyError` recovery paths and both defensive `is None` guards
-became executed code. Pragmas on unreachable-in-production guards were argued
+how the `DuplicateKeyError` recovery path in `CreateChatUseCase` and its
+defensive `is None` guard became executed code. Pragmas on unreachable-in-production guards were argued
 for twice and refused, because an excluded branch is one nobody notices when it
 stops being unreachable, and the coverage number then asserts something untrue
 about what the tests exercise; tests written only to move the number were

docs/adr/0012-message-send-is-one-upsert.md

@@ -0,0 +1,19 @@
+# A message send is one INSERT ... ON CONFLICT DO UPDATE
+
+`MessagesRepository.get_or_create` writes a send as one `INSERT ... ON CONFLICT
+ON CONSTRAINT uk_messages_chat_id_idempotency_key DO UPDATE SET idempotency_key
+= EXCLUDED.idempotency_key RETURNING ..., xmax = 0`. The no-op update makes a
+replay return the existing row from the same statement, and a request racing an
+uncommitted insert of the same key waits on that row's lock, then gets the
+winner's row. `xmax = 0` is true only for a row version the statement inserted,
+so the use case can answer 201 or 200 and skip the `last_message_id` update on a
+replay. The read-then-insert with `DuplicateKeyError` recovery from
+[0005](0005-upsert-via-duplicate-key-recovery.md) cost two round trips per new
+message and re-read the winner outside the retried transaction. `ON CONFLICT DO
+NOTHING` returns no row to the loser, and advanced-alchemy's `upsert` reads
+first. The cost: every replay writes a row version and consumes a sequence
+value, so a retried send leaves a gap in message ids that
+[0001](0001-sequence-ids-not-snowflakes.md)'s client gap detection answers with
+one empty catch-up fetch. Postgres does not document `xmax` in `RETURNING`; the
+replay and concurrent tests in `tests/use_cases/test_create_message.py` would
+catch a change.