Skip to content

Commit 1cc521f

Browse files
authored
fix: let PUT cards swap fronts and lock cards in id order (#101)
Make card_deck_id_front_uc DEFERRABLE INITIALLY IMMEDIATE so uniqueness is checked at the end of the statement and one PUT can swap two cards' fronts. Join the UPDATE to a locked CTE that selects the rows ORDER BY id FOR UPDATE, so concurrent PUTs on overlapping cards lock them in the same order instead of deadlocking.
1 parent 8017334 commit 1cc521f

8 files changed

Lines changed: 143 additions & 23 deletions

File tree

‎AGENTS.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
4747
- `dynamic_engine` — `Scope.REQUEST`; `choose_sa_engine` returns the read engine for `GET`/`HEAD` requests, the primary otherwise (including when there is no request). GET handlers cannot write, and a read right after a write may see replica lag.
4848
- `session` — `Scope.REQUEST`, bound to `dynamic_engine`, finalized by `close_session`.
4949
- `transaction` — `Scope.REQUEST`, a `db_retry.Transaction` over `session`. Leaving its block rolls back anything uncommitted and closes the session.
50-
- `autocommit_transaction` — the same `Transaction`, opened at `AUTOCOMMIT` (`autocommit_isolation_level`). `CreateDeckUseCase` and `UpdateDeckUseCase` use it because each sends exactly one statement, which Postgres runs atomically on its own. The asyncpg adapter then sends no BEGIN or COMMIT, so those writes cost one round trip instead of three. A use case with two or more statements, or one that relies on rolling back after a check, as `UpdateCardsUseCase` does, keeps `transaction`. The `db_session` fixture overrides `autocommit_isolation_level` to `None`, because switching the shared test connection to `AUTOCOMMIT` would roll back its outer transaction. `tests/test_autocommit_writes.py` runs both use cases against a real engine instead.
50+
- `autocommit_transaction` — the same `Transaction`, opened at `AUTOCOMMIT` (`autocommit_isolation_level`). `CreateDeckUseCase` and `UpdateDeckUseCase` use it because each sends exactly one statement, which Postgres runs atomically on its own. The asyncpg adapter then sends no BEGIN or COMMIT, so those writes cost one round trip instead of three. A use case with two or more statements, or one that relies on rolling back after a check, as `UpdateCardsUseCase` does, keeps `transaction`. The `db_session` fixture overrides `autocommit_isolation_level` to `None`, because switching the shared test connection to `AUTOCOMMIT` would roll back its outer transaction. `tests/test_autocommit_writes.py` runs both use cases against a real engine instead, through the `committing_engine` fixture. `tests/test_card_locking.py` uses the same fixture, because a lock wait needs separate connections; both delete the rows they commit.
5151

5252
`Repositories` (`Scope.REQUEST`): `decks_repository` and `cards_repository`, on `Database.session` with `auto_commit=False`. The use case commits through `transaction.commit()`.
5353

@@ -89,7 +89,7 @@ class ListDecksUseCase:
8989
### Database layer
9090

9191
- `app/models.py` — `BigIntAuditBase` from `advanced_alchemy` (auto `id`, `created_at`, `updated_at`). The module aliases `orm_registry.metadata` onto `orm.DeclarativeBase.metadata` so Alembic autogenerate sees both. New models go here.
92-
- `app/repositories.py` — Subclass `SQLAlchemyAsyncRepositoryService[Model]` with a nested `BaseRepository(SQLAlchemyAsyncRepository[Model])`. Use cases call the service methods (`get_many`, `get_one`, `create_many`) for reads and plain inserts. Writes that advanced-alchemy would spread over several round trips are one statement on the session instead: its `update` reads the row twice and refreshes it after the flush, and `upsert_many` selects the rows before and after writing them. `update_deck` sends `UPDATE ... RETURNING` and `get_or_create` sends `INSERT ... ON CONFLICT ... RETURNING`. `update_cards` sends one `UPDATE cards ... FROM unnest(:ids, :fronts, :backs, :hints) AS updates(...) WHERE cards.id = updates.id AND cards.deck_id = :deck_id RETURNING *`. It binds one array per column, so the SQL text stays the same for any number of cards: SQLAlchemy caches it and asyncpg prepares it once per connection, which a multi-row `VALUES` would not allow. It raises `NotFoundError` (404) if fewer rows come back than ids were sent, which rolls the whole request back. PUT never inserts. An `INSERT ... ON CONFLICT (id)` upsert would move a card in from another deck and would insert client-chosen ids without advancing `cards_id_seq`, so a later POST would collide on the primary key. `schemas.CardCreates` and `schemas.CardUpdates` cap a request at `MAX_CARDS_PER_REQUEST` cards, and `schemas.CardUpdates` rejects duplicate ids, both with a 422 before any query runs. Ids in paths, `after` and card bodies are bounded to Postgres `bigint` (`schemas.MAX_ID`), so an oversized id is a 422 and never reaches asyncpg. A POST to a deck that does not exist fails on the foreign key, and `ForeignKeyError` maps to 404. A statement that can hit a unique constraint runs inside `wrap_sqlalchemy_exception` so that it still raises `DuplicateKeyError`.
92+
- `app/repositories.py` — Subclass `SQLAlchemyAsyncRepositoryService[Model]` with a nested `BaseRepository(SQLAlchemyAsyncRepository[Model])`. Use cases call the service methods (`get_many`, `get_one`, `create_many`) for reads and plain inserts. Writes that advanced-alchemy would spread over several round trips are one statement on the session instead: its `update` reads the row twice and refreshes it after the flush, and `upsert_many` selects the rows before and after writing them. `update_deck` sends `UPDATE ... RETURNING` and `get_or_create` sends `INSERT ... ON CONFLICT ... RETURNING`. `update_cards` sends one `WITH locked AS (SELECT id FROM cards WHERE id = ANY(:ids) AND deck_id = :deck_id ORDER BY id FOR UPDATE) UPDATE cards ... FROM unnest(:ids, :fronts, :backs, :hints) AS updates(...), locked WHERE cards.id = updates.id AND cards.id = locked.id AND cards.deck_id = :deck_id RETURNING *`. It binds one array per column, so the SQL text stays the same for any number of cards: SQLAlchemy caches it and asyncpg prepares it once per connection, which a multi-row `VALUES` would not allow. Every updated row must match a `locked` row, and the CTE locks its rows in id order (lazily, one before each update), so concurrent PUTs on overlapping cards always lock them in the same order and cannot deadlock. Without it the lock order follows the plan, which can be the order the client sent; `tests/test_card_locking.py` forces that interleaving. `card_deck_id_front_uc` is `DEFERRABLE INITIALLY IMMEDIATE`, so uniqueness is checked at the end of the statement and one PUT can swap two cards' fronts. It raises `NotFoundError` (404) if fewer rows come back than ids were sent, which rolls the whole request back. PUT never inserts. An `INSERT ... ON CONFLICT (id)` upsert would move a card in from another deck and would insert client-chosen ids without advancing `cards_id_seq`, so a later POST would collide on the primary key. `schemas.CardCreates` and `schemas.CardUpdates` cap a request at `MAX_CARDS_PER_REQUEST` cards, and `schemas.CardUpdates` rejects duplicate ids, both with a 422 before any query runs. Ids in paths, `after` and card bodies are bounded to Postgres `bigint` (`schemas.MAX_ID`), so an oversized id is a 422 and never reaches asyncpg. A POST to a deck that does not exist fails on the foreign key, and `ForeignKeyError` maps to 404. A statement that can hit a unique constraint runs inside `wrap_sqlalchemy_exception` so that it still raises `DuplicateKeyError`.
9393
- `app/resources/db.py` — engine factories live here and are DI creators: modern-di resolves their parameter annotations at runtime, so `pyproject.toml` exempts `app/resources/` from ruff's `TC` rules. `create_session` passes `join_transaction_mode="create_savepoint"`. This is inert in production (the session binds to an engine) but enables the test rollback pattern below: when a test binds the session to a connection already in a transaction, the session owns its own savepoint so the outer transaction survives commits — do not "fix" it.
9494
- `migrations/env.py` swaps the asyncpg driver for the sync `postgresql` driver and uses `app.models.METADATA` as `target_metadata`.
9595

‎app/models.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ class Deck(BigIntAuditBase):
2323
class Card(BigIntAuditBase):
2424
__tablename__ = "cards"
2525
__table_args__ = (
26-
sa.UniqueConstraint("deck_id", "front", name="card_deck_id_front_uc"),
26+
sa.UniqueConstraint("deck_id", "front", name="card_deck_id_front_uc", deferrable=True, initially="IMMEDIATE"),
2727
sa.Index("ix_cards_deck_id_id", "deck_id", "id"),
2828
)
2929

‎app/repositories.py‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -75,9 +75,17 @@ async def add_cards(self, deck_id: int, cards: list[schemas.CardCreate]) -> Sequ
7575
async def update_cards(self, deck_id: int, cards: list[schemas.CardUpdate]) -> Sequence[models.Card]:
7676
if not cards:
7777
return []
78+
ids: typing.Final = sa.bindparam("ids", [card.id for card in cards], type_=postgresql.ARRAY(sa.BigInteger))
79+
locked: typing.Final = (
80+
sa.select(models.Card.id)
81+
.where(models.Card.id == sa.any_(ids), models.Card.deck_id == deck_id)
82+
.order_by(models.Card.id)
83+
.with_for_update()
84+
.cte("locked")
85+
)
7886
updates: typing.Final = (
7987
sa.func.unnest(
80-
sa.bindparam("ids", [card.id for card in cards], type_=postgresql.ARRAY(sa.BigInteger)),
88+
ids,
8189
sa.bindparam("fronts", [card.front for card in cards], type_=postgresql.ARRAY(sa.String)),
8290
sa.bindparam("backs", [card.back for card in cards], type_=postgresql.ARRAY(sa.String)),
8391
sa.bindparam("hints", [card.hint for card in cards], type_=postgresql.ARRAY(sa.String)),
@@ -92,7 +100,7 @@ async def update_cards(self, deck_id: int, cards: list[schemas.CardUpdate]) -> S
92100
)
93101
statement: typing.Final = (
94102
sa.update(models.Card)
95-
.where(models.Card.id == updates.c.id, models.Card.deck_id == deck_id)
103+
.where(models.Card.id == updates.c.id, models.Card.id == locked.c.id, models.Card.deck_id == deck_id)
96104
.values(front=updates.c.front, back=updates.c.back, hint=updates.c.hint)
97105
.returning(models.Card)
98106
)
Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
"""deferrable card front unique.
2+
3+
Revision ID: 456d08f790fb
4+
Revises: 4004ffe4656f
5+
Create Date: 2026-10-17 15:00:00.000000
6+
7+
"""
8+
9+
from alembic import op
10+
11+
12+
# revision identifiers, used by Alembic.
13+
revision = "456d08f790fb"
14+
down_revision = "4004ffe4656f"
15+
branch_labels = None
16+
depends_on = None
17+
18+
19+
def upgrade() -> None:
20+
op.drop_constraint("card_deck_id_front_uc", "cards", type_="unique")
21+
op.create_unique_constraint(
22+
"card_deck_id_front_uc", "cards", ["deck_id", "front"], deferrable=True, initially="IMMEDIATE"
23+
)
24+
25+
26+
def downgrade() -> None:
27+
op.drop_constraint("card_deck_id_front_uc", "cards", type_="unique")
28+
op.create_unique_constraint("card_deck_id_front_uc", "cards", ["deck_id", "front"])

‎tests/conftest.py‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
import fastapi
1717
import modern_di
1818
from sqlalchemy.engine import Connection
19-
from sqlalchemy.ext.asyncio import AsyncSession
19+
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
2020

2121

2222
SAVEPOINT_PREFIXES: typing.Final = ("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO SAVEPOINT")
@@ -92,3 +92,14 @@ def record(*args: object) -> None:
9292
yield recorded
9393
finally:
9494
event.remove(sync_connection, "before_cursor_execute", record)
95+
96+
97+
@pytest.fixture
98+
async def committing_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
99+
engine = create_primary_sa_engine()
100+
di_container.override(ioc.Database.dynamic_engine, engine)
101+
try:
102+
yield engine
103+
finally:
104+
di_container.reset_override(ioc.Database.dynamic_engine)
105+
await engine.dispose()

‎tests/test_autocommit_writes.py‎

Lines changed: 5 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -9,24 +9,12 @@
99
from sqlalchemy import event
1010

1111
from app import ioc, models, schemas
12-
from app.resources.db import create_primary_sa_engine
1312

1413

1514
if typing.TYPE_CHECKING:
1615
from sqlalchemy.ext.asyncio import AsyncEngine
1716

1817

19-
@pytest.fixture
20-
async def primary_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
21-
engine = create_primary_sa_engine()
22-
di_container.override(ioc.Database.dynamic_engine, engine)
23-
try:
24-
yield engine
25-
finally:
26-
di_container.reset_override(ioc.Database.dynamic_engine)
27-
await engine.dispose()
28-
29-
3018
@contextlib.contextmanager
3119
def record_in_transaction(engine: AsyncEngine) -> typing.Iterator[list[bool]]:
3220
in_transaction: typing.Final[list[bool]] = []
@@ -44,25 +32,25 @@ def record(connection: sa.Connection, *_: object) -> None:
4432

4533

4634
async def test_create_deck_sends_no_begin_or_commit(
47-
di_container: modern_di.Container, primary_engine: AsyncEngine
35+
di_container: modern_di.Container, committing_engine: AsyncEngine
4836
) -> None:
4937
key: typing.Final = uuid.uuid4()
5038
try:
51-
with record_in_transaction(primary_engine) as in_transaction:
39+
with record_in_transaction(committing_engine) as in_transaction:
5240
async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
5341
use_case = container.resolve_provider(ioc.UseCases.create_deck_use_case)
5442
await use_case(data=schemas.DeckCreate(name="autocommit", idempotency_key=key))
5543
finally:
56-
async with primary_engine.begin() as connection:
44+
async with committing_engine.begin() as connection:
5745
await connection.execute(sa.delete(models.Deck).where(models.Deck.idempotency_key == key))
5846

5947
assert in_transaction == [False]
6048

6149

6250
async def test_update_deck_sends_no_begin_or_commit(
63-
di_container: modern_di.Container, primary_engine: AsyncEngine
51+
di_container: modern_di.Container, committing_engine: AsyncEngine
6452
) -> None:
65-
with record_in_transaction(primary_engine) as in_transaction:
53+
with record_in_transaction(committing_engine) as in_transaction:
6654
async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
6755
use_case = container.resolve_provider(ioc.UseCases.update_deck_use_case)
6856
with pytest.raises(NotFoundError):

‎tests/test_card_locking.py‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
import asyncio
2+
import typing
3+
import uuid
4+
5+
import sqlalchemy as sa
6+
from fastapi import status
7+
8+
from app import models
9+
10+
11+
if typing.TYPE_CHECKING:
12+
from httpx import AsyncClient, Response
13+
from sqlalchemy.ext.asyncio import AsyncEngine
14+
15+
16+
async def wait_for_lock_waiters(engine: AsyncEngine, count: int, request: asyncio.Task[typing.Any]) -> None:
17+
statement: typing.Final = sa.text(
18+
"SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() AND wait_event_type = 'Lock'"
19+
)
20+
async with asyncio.timeout(5):
21+
while not request.done():
22+
async with engine.connect() as connection:
23+
if await connection.scalar(statement) >= count:
24+
return
25+
await asyncio.sleep(0.01)
26+
27+
28+
async def test_concurrent_card_updates_in_opposite_orders_do_not_deadlock(
29+
client: AsyncClient, committing_engine: AsyncEngine
30+
) -> None:
31+
async with committing_engine.begin() as connection:
32+
deck_id = await connection.scalar(
33+
sa.insert(models.Deck).values(name="locking", idempotency_key=uuid.uuid4()).returning(models.Deck.id)
34+
)
35+
card1, card2, card3 = (
36+
await connection.scalars(
37+
sa.insert(models.Card)
38+
.values([{"deck_id": deck_id, "front": f"card{i}"} for i in range(1, 4)])
39+
.returning(models.Card.id)
40+
)
41+
).all()
42+
url: typing.Final = f"/api/decks/{deck_id}/cards/"
43+
requests: typing.Final[list[asyncio.Task[Response]]] = []
44+
try:
45+
async with committing_engine.connect() as blocker:
46+
await blocker.execute(sa.text("SELECT 1 FROM cards WHERE id = :id FOR UPDATE"), {"id": card3})
47+
requests.append(
48+
asyncio.create_task(
49+
client.put(
50+
url,
51+
json=[
52+
{"id": card1, "front": "a1"},
53+
{"id": card3, "front": "a3"},
54+
{"id": card2, "front": "a2"},
55+
],
56+
)
57+
)
58+
)
59+
await wait_for_lock_waiters(committing_engine, 1, requests[0])
60+
requests.append(
61+
asyncio.create_task(client.put(url, json=[{"id": card2, "front": "b2"}, {"id": card1, "front": "b1"}]))
62+
)
63+
await wait_for_lock_waiters(committing_engine, 2, requests[1])
64+
responses = await asyncio.gather(*requests)
65+
finally:
66+
for request in requests:
67+
request.cancel()
68+
await asyncio.gather(*requests, return_exceptions=True)
69+
async with committing_engine.begin() as connection:
70+
await connection.execute(sa.delete(models.Card).where(models.Card.deck_id == deck_id))
71+
await connection.execute(sa.delete(models.Deck).where(models.Deck.id == deck_id))
72+
73+
assert [response.status_code for response in responses] == [status.HTTP_200_OK, status.HTTP_200_OK]

‎tests/test_cards.py‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -321,3 +321,15 @@ async def test_card_endpoints_reject_ids_out_of_bigint_range(
321321
) -> None:
322322
response = await client.request(method, path, json=body, params=params)
323323
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
324+
325+
326+
async def test_update_cards_swaps_fronts(client: AsyncClient) -> None:
327+
deck = await factories.DeckModelFactory.create_async()
328+
card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)
329+
330+
response = await client.put(
331+
f"/api/decks/{deck.id}/cards/",
332+
json=[{"id": card1.id, "front": card2.front}, {"id": card2.id, "front": card1.front}],
333+
)
334+
assert response.status_code == status.HTTP_200_OK
335+
assert [x["front"] for x in response.json()["items"]] == [card2.front, card1.front]

0 commit comments

Comments
 (0)