fix(api): a waiting reorder deals from the fresh order; the commit check sees staged writes

Round 2 of the deep review, on round 1's own fixes:
- `ORDER BY position … FOR UPDATE` sorts before it locks, so the reorder that waited got the
  fresh rows in its STALE order and dealt the pins it left out (a tab with Plex off) back to their
  old slots. It locks in id order — one global order, so two reorders cannot deadlock — and sorts
  the fresh rows itself. The race test now sends that partial list and fails unless tab B really
  blocked.
- The request check saw only writes the handler flushed: a `db.delete` left staged, or a flush
  followed by a released savepoint, passed. Staged changes count at the end of the request, and
  only the ROOT transaction's end clears it (a savepoint test drives the dependency directly).
- The `ok` test exempted every call as a serializer, hiding four domain REPORTS (YouTube sync,
  revert, push; Plex sync) without `ok`. They are spread into `{"ok": True, …}`; the exemption is a
  named list of real serializers.

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 = (
+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())
         )
-    return list(db.execute(stmt.with_for_update() if lock else stmt).scalars().all())
+        .scalars()
+        .all()
+    )
 
 
 def next_position(db: Session, user: User) -> int:

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)

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:

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):
+        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

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]

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

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)}