Skip to content

Commit bc5db96

Browse files
authored
fix: let PUT cards swap fronts and lock cards in id order (#78)
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 46c00ea commit bc5db96

8 files changed

Lines changed: 143 additions & 22 deletions

File tree

‎AGENTS.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ Python is 3.14, dependencies managed by `uv`. The API is exposed on `:8000`.
3030
- `dynamic_engine` — request-scoped; `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
3131
- `session` — request-scoped, bound to `dynamic_engine`, finalizer closes the session
3232
- `transaction` — request-scoped `db_retry.Transaction` over `session`. Leaving its block rolls back anything uncommitted and closes the session
33-
- `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.
33+
- `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.
3434

3535
`Repositories` group (request-scoped): `decks_repository` and `cards_repository`, on `Database.session` with `auto_commit=False`. The use case commits through `transaction.commit()`.
3636

@@ -61,7 +61,7 @@ class ListDecksUseCase:
6161
- `CreateCardsUseCase` relies on the `(deck_id, front)` unique constraint. A replay is rejected as a duplicate-key error rather than creating the cards twice.
6262
- Updates (`UpdateDeckUseCase`, `UpdateCardsUseCase`) write the same values again, so a replay is harmless.
6363

64-
**Persistence**: Models inherit `advanced_alchemy.base.BigIntAuditBase` (gives `id: BigInt`, `created_at`, `updated_at`). Metadata is shared with `orm.DeclarativeBase.metadata` in `app/models.py` so Alembic autogen sees everything. Repositories are `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. The card handlers cap a request at `schemas.MAX_CARDS_PER_REQUEST` cards, and `update_cards` rejects duplicate ids, both with a 400 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 400 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`. `create_session` in `app/resources/db.py` builds a plain `AsyncSession` with `join_transaction_mode="create_savepoint"`. This is inert in production (the session binds to an engine) but enables the per-test rollback fixture: 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.
64+
**Persistence**: Models inherit `advanced_alchemy.base.BigIntAuditBase` (gives `id: BigInt`, `created_at`, `updated_at`). Metadata is shared with `orm.DeclarativeBase.metadata` in `app/models.py` so Alembic autogen sees everything. Repositories are `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. The card handlers cap a request at `schemas.MAX_CARDS_PER_REQUEST` cards, and `update_cards` rejects duplicate ids, both with a 400 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 400 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`. `create_session` in `app/resources/db.py` builds a plain `AsyncSession` with `join_transaction_mode="create_savepoint"`. This is inert in production (the session binds to an engine) but enables the per-test rollback fixture: 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.
6565

6666
**Test isolation** (`tests/conftest.py`): `db_session` fixture opens a connection, starts a transaction, then **overrides** `Database.dynamic_engine` in the DI container to return that connection. All requests in the test reuse this connection. Each session built against it uses `join_transaction_mode="create_savepoint"`, so `transaction.commit()` releases the session's own savepoint while the outer transaction is rolled back in teardown. DB state is clean between tests with no truncation needed. `app` and `client` fixtures build the real app and run it through `httpx.ASGITransport` + `asgi_lifespan.LifespanManager`. Polyfactory `SQLAlchemyFactory` is wired up via `set_async_session_in_base_sqlalchemy_factory`.
6767

‎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 & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import litestar
1818
import modern_di
1919
from sqlalchemy.engine import Connection
20+
from sqlalchemy.ext.asyncio import AsyncEngine
2021

2122

2223
SAVEPOINT_PREFIXES: typing.Final = ("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO SAVEPOINT")
@@ -92,3 +93,14 @@ def record(*args: object) -> None:
9293
yield recorded
9394
finally:
9495
event.remove(sync_connection, "before_cursor_execute", record)
96+
97+
98+
@pytest.fixture
99+
async def committing_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
100+
engine = create_primary_sa_engine()
101+
di_container.override(ioc.Database.dynamic_engine, engine)
102+
try:
103+
yield engine
104+
finally:
105+
di_container.reset_override(ioc.Database.dynamic_engine)
106+
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 litestar import status_codes as 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_codes.HTTP_400_BAD_REQUEST
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_codes.HTTP_200_OK
335+
assert [x["front"] for x in response.json()["items"]] == [card2.front, card1.front]

0 commit comments

Comments
 (0)