diff --git a/backend/app/discover/pins.py b/backend/app/discover/pins.py index 78f50e01..318b68ec 100644 --- a/backend/app/discover/pins.py +++ b/backend/app/discover/pins.py @@ -86,8 +86,18 @@ def reorder(db: Session, user: User, ids: list[int]) -> None: ignored rather than failing a stale replay wholesale. Flushes; the route commits.""" # Locked: the rows it leaves alone are part of the answer too. Two overlapping reorders each # skipped the rows that already matched in their own snapshot and interleaved into two pins on - # one slot; the second now waits and deals from the first one's result. - current = list_pins(db, user, lock=True) + # one slot; the second now waits and deals from the first one's result. Locked in id order and + # sorted HERE: with `ORDER BY position … FOR UPDATE` Postgres sorts before it locks, so the one + # that waited got the fresh rows in its stale order and dealt the pins it left out back to their + # old slots — and one global lock order means two reorders cannot deadlock. + locked = db.execute( + select(ComfortPin) + .where(ComfortPin.user_id == user.id) + .order_by(ComfortPin.id) + .with_for_update() + .execution_options(populate_existing=True) + ).scalars() + current = sorted(locked, key=lambda p: (p.position, p.id)) owned = {p.id for p in current} listed: list[int] = [] for i in ids: @@ -103,13 +113,16 @@ def reorder(db: Session, user: User, ids: list[int]) -> None: db.flush() -def list_pins(db: Session, user: User, *, lock: bool = False) -> list[ComfortPin]: - stmt = ( - select(ComfortPin) - .where(ComfortPin.user_id == user.id) - .order_by(ComfortPin.position.asc(), ComfortPin.id.asc()) +def list_pins(db: Session, user: User) -> list[ComfortPin]: + return list( + db.execute( + select(ComfortPin) + .where(ComfortPin.user_id == user.id) + .order_by(ComfortPin.position.asc(), ComfortPin.id.asc()) + ) + .scalars() + .all() ) - return list(db.execute(stmt.with_for_update() if lock else stmt).scalars().all()) def next_position(db: Session, user: User) -> int: diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index 95a0b2b1..f4a09fb0 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -324,7 +324,7 @@ def sync_youtube( _check_pull_cooldown(user) try: with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL): - return sync_user_playlists(db, user) + return {"ok": True, **sync_user_playlists(db, user)} except YouTubeError as exc: db.rollback() raise youtube_refusal(exc) @@ -347,7 +347,7 @@ def revert_youtube( _check_pull_cooldown(user) try: with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL): - return repull_playlist(db, user, pl) + return {"ok": True, **repull_playlist(db, user, pl)} except YouTubeError as exc: db.rollback() raise youtube_refusal(exc) @@ -407,7 +407,7 @@ def push( plan = plan_merge(db, user, pl, live=live) if plan["units_estimate"] > quota.remaining_today(db): raise _refused(429, "quota_exceeded") - return merge_playlist(db, user, pl, live=live) + return {"ok": True, **merge_playlist(db, user, pl, live=live)} except YouTubeError as exc: db.rollback() raise youtube_refusal(exc) diff --git a/backend/app/routes/plex/admin.py b/backend/app/routes/plex/admin.py index f724bc7f..82642f9e 100644 --- a/backend/app/routes/plex/admin.py +++ b/backend/app/routes/plex/admin.py @@ -103,7 +103,7 @@ def _media_check(db: Session, plex: PlexClient, sections: list[dict]) -> dict: def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: """Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a while on a large library); a background/scheduled sync also runs periodically.""" - return plex_sync.sync(db) + return {"ok": True, **plex_sync.sync(db)} def _link_dict(link: PlexLink | None) -> dict: diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index d1cc5483..d6c37c9a 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -108,8 +108,9 @@ def _request_dep(db, unfinished: list[str]): session after the request, which rolls back anything flushed but not committed — yet here every request shares the test's session, so a follow-up GET or an assertion reads the flushed write and the suite stays green over a handler that forgot to commit (deep review, debt burn). So a - request that ends normally with a flushed, uncommitted write is recorded, and the test fails on - it at teardown. The test's own staged rows are flushed before the request starts — they are the + request that ends normally with a write it did not commit — flushed, or still only staged — is + recorded, and the test fails on it at teardown. Only the ROOT transaction's end clears it: a + savepoint released inside the handler commits nothing. The test's own staged rows are flushed before the request starts — they are the test's business, not the handler's. A request that raises is exempt: its writes are meant to roll back.""" @@ -120,15 +121,17 @@ def _request_dep(db, unfinished: list[str]): def flushed(session, ctx): state["dirty"] = True - def ended(session): - state["dirty"] = False + def ended(session, transaction): + if transaction.parent is None: + state["dirty"] = False - hooks = (("after_flush", flushed), ("after_commit", ended), ("after_rollback", ended)) + hooks = (("after_flush", flushed), ("after_transaction_end", ended)) for name, fn in hooks: event.listen(db, name, fn) try: yield db - if state["dirty"]: + staged = db.new or db.deleted or any(db.is_modified(o) for o in db.dirty) + if state["dirty"] or staged: unfinished.append(f"{conn.scope.get('method', 'WS')} {conn.url.path}") finally: for name, fn in hooks: @@ -146,12 +149,13 @@ def db(_schema): session = SessionLocal() unfinished: list[str] = [] session.info["request"] = _request_dep(session, unfinished) + session.info["unfinished"] = unfinished # the requests recorded so far (the check's own test) try: yield session finally: session.rollback() session.close() - assert unfinished == [], "a handler flushed a write and never committed it" + assert unfinished == [], "a handler left a write uncommitted" @pytest.fixture diff --git a/backend/tests/test_discover.py b/backend/tests/test_discover.py index e1fc5872..1a6894a0 100644 --- a/backend/tests/test_discover.py +++ b/backend/tests/test_discover.py @@ -1074,8 +1074,10 @@ def test_a_hero_from_a_show_pin_names_the_pin_it_came_from(db, as_user): def test_two_overlapping_reorders_never_share_a_slot(db, as_user): """`reorder` writes only the rows that change. Unlocked, two tabs reordering from the same snapshot each skipped the rows that already matched in THEIR snapshot and interleaved row by - row: p1=1, p2=2, p3=1 — two pins on one slot (deep review, debt burn). Locked, the second waits - for the first and deals from its result.""" + row — two pins on one slot (deep review, debt burn). Locked, the second waits for the first and + deals from its result: here a partial list (a tab with Plex off, one pin hidden), whose left-out + pin must stay where the FIRST reorder put it — a lock taken in position order handed the waiter + fresh rows in its stale order, and it dealt that pin back to its old slot (round 2).""" import threading from app.db import SessionLocal @@ -1088,12 +1090,12 @@ def test_two_overlapping_reorders_never_share_a_slot(db, as_user): uid = user.id with SessionLocal() as tab_a, SessionLocal() as tab_b, SessionLocal() as probe: - pins.reorder(tab_a, tab_a.get(User, uid), [p2, p1, p3]) # written, not yet committed + pins.reorder(tab_a, tab_a.get(User, uid), [p2, p3, p1]) # written, not yet committed errors: list[BaseException] = [] def second_tab(): try: - pins.reorder(tab_b, tab_b.get(User, uid), [p1, p3, p2]) + pins.reorder(tab_b, tab_b.get(User, uid), [p3, p2]) tab_b.commit() except BaseException as exc: # noqa: BLE001 — re-raised in the test thread below errors.append(exc) @@ -1101,12 +1103,14 @@ def test_two_overlapping_reorders_never_share_a_slot(db, as_user): worker = threading.Thread(target=second_tab) worker.start() # Commit A only once B is blocked on a row lock, so B's read is the stale snapshot. - for _ in range(100): + waiting = 0 + for _ in range(200): waiting = probe.scalar(text("SELECT count(*) FROM pg_locks WHERE NOT granted")) probe.rollback() if waiting: break worker.join(0.05) + assert waiting, "tab B never blocked — the race was not exercised" tab_a.commit() worker.join(10) assert not worker.is_alive() and errors == [] @@ -1114,4 +1118,4 @@ def test_two_overlapping_reorders_never_share_a_slot(db, as_user): db.expire_all() positions = db.scalars(select(ComfortPin.position).where(ComfortPin.user_id == uid)).all() assert sorted(positions) == [0, 1, 2] - assert [p["pin_id"] for p in c.get("/api/discover/pins").json()["items"]] == [p1, p3, p2] + assert [p["pin_id"] for p in c.get("/api/discover/pins").json()["items"]] == [p3, p2, p1] diff --git a/backend/tests/test_override_dep_convention.py b/backend/tests/test_override_dep_convention.py index 222009a7..4f032823 100644 --- a/backend/tests/test_override_dep_convention.py +++ b/backend/tests/test_override_dep_convention.py @@ -35,3 +35,26 @@ def test_a_request_gets_the_session_that_checks_its_commit(): if "override_dep(get_db," in line and 'override_dep(get_db, db.info["request"])' not in line ] assert offenders == [], 'override get_db with db.info["request"]: ' + ", ".join(offenders) + + +def test_the_commit_check_counts_only_the_root_transaction(db): + """A savepoint released inside a handler commits nothing to the database; ending it must not + clear the check (round 2). Driven directly, as a request would drive the dependency.""" + import pytest + from starlette.requests import HTTPConnection + + from app.models import User + + conn = HTTPConnection( + {"type": "http", "method": "POST", "path": "/probe", "headers": [], "query_string": b""} + ) + request = db.info["request"](conn) + session = next(request) + session.add(User(email="savepoint-probe@example.com")) + session.flush() + with session.begin_nested(): + pass + with pytest.raises(StopIteration): + next(request) + assert db.info["unfinished"] == ["POST /probe"] + db.info["unfinished"].clear() # recorded as it should be — not this test's failure diff --git a/backend/tests/test_route_conventions.py b/backend/tests/test_route_conventions.py index 42aa81a9..a7ccefcd 100644 --- a/backend/tests/test_route_conventions.py +++ b/backend/tests/test_route_conventions.py @@ -38,6 +38,30 @@ def _modules(): FUNCTIONS = (ast.FunctionDef, ast.AsyncFunctionDef) +# The calls a mutation may answer with in place of an acknowledgement: each describes the resource +# it wrote (or is the logout's cookie-clearing response). A domain function's REPORT dict is not one +# of them — `{"synced": n}` is an acknowledgement, and gets spread into `{"ok": True, **report}`. +SERIALIZERS = { + "JSONResponse", + "_collection_card", + "_playlist_card", + "_serialize", + "_serialize_demo", + "_serialize_job", + "_serialize_msg", + "_serialize_profile", + "_serialize_user", + "_summary", + "admin_get_quota", + "describe", + "owner_view", +} + + +def _called(node: ast.Call) -> str: + f = node.func + return f.id if isinstance(f, ast.Name) else getattr(f, "attr", "") + def _is_route(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: return any( @@ -55,16 +79,19 @@ def _is_mutation_route(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: def test_every_mutation_answers_ok(): """`{"ok": true, …}` (or `{"deleted": id}`) for every mutation, sync or async — one field a - client can read the same way everywhere. A serializer's return (a call) is exempt: it describes - a resource, not an acknowledgement. A dict built elsewhere and returned by name is not: it hid - two acknowledgements without `ok` from the literal-only scan, so it is spread into a literal.""" + client can read the same way everywhere. A serializer's return (SERIALIZERS) is exempt: it + describes a resource, not an acknowledgement. A dict built elsewhere — returned by name, or + straight from a domain call — is not: both hid acknowledgements without `ok` from a + literal-only scan, so such a dict is spread into a literal.""" missing = [] for name, tree in _modules(): for fn in ast.walk(tree): if not (isinstance(fn, FUNCTIONS) and _is_mutation_route(fn)): continue for node in ast.walk(fn): - if not isinstance(node, ast.Return) or isinstance(node.value, ast.Call): + if not isinstance(node, ast.Return): + continue + if isinstance(node.value, ast.Call) and _called(node.value) in SERIALIZERS: continue keys = ( {k.value for k in node.value.keys if isinstance(k, ast.Constant)}