fix(playlists): one YouTube sync of a list at a time, across tabs and the scheduler

The client's in-flight flag covers the menus of one page only. A push reads YouTube, then writes
both sides for seconds and commits step by step, so a row lock ends at its first commit: two tabs,
or a push and the scheduled mirror refresh, could interleave into a doubled YouTube item or a
baseline recorded over the other's writes. `sync_claim` takes a session-level advisory lock on a
connection of its own (it survives those commits; a crashed worker's lock dies with its
connection, so no claim column is left stuck). Push and revert answer 409 `sync_busy`; the
scheduled refresh skips a list being synced and picks it up next run.

backend/app/routes/playlists.py

@@ -19,6 +19,7 @@ from app.sync.playlists import (
     plan_merge,
     recompute_dirty,
     repull_playlist,
+    sync_claim,
     sync_user_playlists,
 )
 from app.ratelimit import RateLimiter
@@ -347,13 +348,16 @@ def revert_youtube(
         raise _refused(400, "not_on_youtube")
     if not has_read_scope(user):
         raise _refused(403, "youtube_read_scope")
-    _check_pull_cooldown(user)
-    try:
-        with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL):
-            return {"ok": True, **repull_playlist(db, user, pl)}
-    except YouTubeError as exc:
-        db.rollback()
-        raise youtube_refusal(exc)
+    with sync_claim(db, pl.id) as free:
+        if not free:
+            raise _refused(409, "sync_busy")
+        _check_pull_cooldown(user)
+        try:
+            with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL):
+                return {"ok": True, **repull_playlist(db, user, pl)}
+        except YouTubeError as exc:
+            db.rollback()
+            raise youtube_refusal(exc)
 
 
 def _pushable(db: Session, user: User, playlist_id: int) -> Playlist:
@@ -395,25 +399,30 @@ def push(
     neither side loses an item). Refuses if the estimated quota exceeds what's left today, so a
     big sync can't strand half-applied. Write scope required (require_write_scope)."""
     pl = _pushable(db, user, playlist_id)
-    try:
-        with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PUSH):
-            # Read the live playlist ONCE (linked playlists only) and share it with both the
-            # plan and the merge, halving read-quota and closing the second read→write TOCTOU
-            # window. A fresh export (no yt_playlist_id) reads nothing.
-            live = None
-            if pl.yt_playlist_id:
-                with YouTubeClient(db, user) as yt:
-                    live = (
-                        list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id)),
-                        yt.get_playlist_snippet(pl.yt_playlist_id),
-                    )
-            plan = plan_merge(db, user, pl, live=live)
-            if plan["units_estimate"] > quota.remaining_today(db):
-                raise _refused(429, "quota_exceeded")
-            return {"ok": True, **merge_playlist(db, user, pl, live=live)}
-    except YouTubeError as exc:
-        db.rollback()
-        raise youtube_refusal(exc)
+    # One sync of a list at a time, across tabs, devices and the scheduler (`sync_claim`): the
+    # client's own in-flight flag only covers the menus of one page.
+    with sync_claim(db, pl.id) as free:
+        if not free:
+            raise _refused(409, "sync_busy")
+        try:
+            with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PUSH):
+                # Read the live playlist ONCE (linked playlists only) and share it with both the
+                # plan and the merge, halving read-quota and closing the second read→write TOCTOU
+                # window. A fresh export (no yt_playlist_id) reads nothing.
+                live = None
+                if pl.yt_playlist_id:
+                    with YouTubeClient(db, user) as yt:
+                        live = (
+                            list(yt.iter_playlist_items_with_ids(pl.yt_playlist_id)),
+                            yt.get_playlist_snippet(pl.yt_playlist_id),
+                        )
+                plan = plan_merge(db, user, pl, live=live)
+                if plan["units_estimate"] > quota.remaining_today(db):
+                    raise _refused(429, "quota_exceeded")
+                return {"ok": True, **merge_playlist(db, user, pl, live=live)}
+        except YouTubeError as exc:
+            db.rollback()
+            raise youtube_refusal(exc)
 
 
 @router.get("/{playlist_id}")

backend/app/sync/playlists.py

@@ -8,12 +8,13 @@ ingested (with a stub channel) so the mirror is faithful."""
 
 import hashlib
 import json
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import TypeVar
 
-from sqlalchemy import delete, select
+from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
 from app import logs, progress, quota
@@ -27,6 +28,37 @@ log = logs.get_logger(__name__)
 # Cost of a single YouTube write op (playlists/playlistItems insert/update/delete).
 WRITE_UNIT_COST = 50
 
+# The advisory-lock namespace of `sync_claim` ("PL"), so its per-playlist key cannot meet another
+# feature's lock on the same number (Postgres keeps the two-int keys apart from the one-bigint ones).
+_SYNC_LOCK_NS = 0x504C
+
+
+@contextmanager
+def sync_claim(db: Session, playlist_id: int) -> Iterator[bool]:
+    """Claim one playlist's YouTube sync (push, re-pull, the scheduled mirror refresh) for the block;
+    yields whether it was free. A push reads YouTube, then writes both sides for seconds and commits
+    step by step, so a row lock ends at its first commit and two tabs (or a push and the scheduler)
+    interleave into a doubled YouTube item or a baseline recorded over the other's writes. A
+    session-level advisory lock on a connection of its own survives those commits, and a crashed
+    worker's lock dies with its connection — no claim column that a crash leaves stuck."""
+    conn = db.get_bind().engine.connect()
+    held = False
+    try:
+        held = bool(conn.scalar(select(func.pg_try_advisory_lock(_SYNC_LOCK_NS, playlist_id))))
+        conn.commit()
+        yield held
+    finally:
+        try:
+            if held:
+                conn.scalar(select(func.pg_advisory_unlock(_SYNC_LOCK_NS, playlist_id)))
+                conn.commit()
+        except Exception:
+            # A lock whose unlock failed must not ride back into the pool on this connection.
+            conn.invalidate()
+            raise
+        finally:
+            conn.close()
+
 
 def fingerprint(name: str, ids: list[str]) -> str:
     """A stable checksum of a playlist's name + ordered video ids. Stored as the synced
@@ -245,12 +277,16 @@ def sync_user_playlists(db: Session, user: User) -> dict:
                 )
                 db.add(pl)
                 db.flush()
-            else:
+            with sync_claim(db, pl.id) as free:
+                if not free:
+                    # The user is pushing or re-pulling this very list: it writes the same rows.
+                    # The next scheduled run refreshes it.
+                    continue
                 pl.name = p.get("title") or pl.name
-            # Mirror is authoritative: replace the items to match YouTube's order.
-            ids = list(yt.iter_my_playlist_video_ids(ytid))
-            _replace_items_from_youtube(db, yt, pl, ids)
-            db.commit()
+                # Mirror is authoritative: replace the items to match YouTube's order.
+                ids = list(yt.iter_my_playlist_video_ids(ytid))
+                _replace_items_from_youtube(db, yt, pl, ids)
+                db.commit()
             synced += 1
     return {"synced": synced}
 

backend/tests/test_playlist_sync_claim.py

@@ -0,0 +1,111 @@
+"""E28 review C1 — one YouTube sync of a playlist at a time, whoever starts it.
+
+A push reads YouTube and then writes both sides for seconds, committing step by step, so the
+client's in-flight flag (one page's menus) and a row lock (gone at the first commit) both leave two
+tabs, or a push and the scheduled mirror refresh, free to interleave. The claim is an advisory lock
+on a connection of its own; the test holds it from outside, the way a second request would.
+"""
+
+from sqlalchemy import text
+
+from app.auth import require_write_scope
+from app.models import Playlist
+from app.routes import playlists as plroutes
+from app.sync import playlists as plsync
+
+
+def _linked(db, user, *, source="local"):
+    pl = Playlist(user_id=user.id, name="L", source=source, yt_playlist_id="PL123")
+    db.add(pl)
+    db.commit()
+    return pl
+
+
+class FakeYT:
+    """YouTube as far as these routes read it: two playlists, none with items."""
+
+    def __init__(self, *_):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        return False
+
+    def iter_my_playlists(self):
+        return [{"id": "PL123", "title": "Renamed"}, {"id": "PL999", "title": "New"}]
+
+    def iter_my_playlist_video_ids(self, _ytid):
+        return []
+
+    def iter_playlist_items_with_ids(self, _ytid):
+        return []
+
+    def get_playlist_snippet(self, _ytid):
+        return {}
+
+
+def test_push_and_revert_refuse_while_the_list_is_claimed(
+    db, as_user, make_user, override_dep, monkeypatch
+):
+    user = make_user()
+    pl = _linked(db, user)
+    override_dep(require_write_scope, lambda: user)
+    monkeypatch.setattr(plroutes, "has_read_scope", lambda u: True)
+    ran: list[str] = []
+    monkeypatch.setattr(plroutes, "YouTubeClient", FakeYT)
+    monkeypatch.setattr(plroutes, "plan_merge", lambda d, u, p, live: {"units_estimate": 0})
+    monkeypatch.setattr(plroutes, "merge_playlist", lambda d, u, p, live: ran.append("push") or {})
+    monkeypatch.setattr(plroutes, "repull_playlist", lambda d, u, p: ran.append("revert") or {})
+    c = as_user(user)
+
+    with plsync.sync_claim(db, pl.id) as free:
+        assert free
+        for path in ("push", "revert-youtube"):
+            r = c.post(f"/api/playlists/{pl.id}/{path}")
+            assert r.status_code == 409
+            assert r.json()["detail"] == {"code": "playlist", "reason": "sync_busy"}
+    assert ran == []
+
+    # Released on the way out: the same list syncs again, and a claim is per list.
+    assert c.post(f"/api/playlists/{pl.id}/push").status_code == 200
+    assert c.post(f"/api/playlists/{pl.id}/revert-youtube").status_code == 200
+    assert ran == ["push", "revert"]
+
+
+def test_a_claim_is_released_when_its_block_raises(db, make_user):
+    pl = _linked(db, make_user())
+    try:
+        with plsync.sync_claim(db, pl.id) as free:
+            assert free
+            raise RuntimeError("boom")
+    except RuntimeError:
+        pass
+    # Asked of pg_locks, not by claiming again: a lock leaked on a pooled connection would be granted
+    # again to that same connection (advisory locks are re-entrant per session).
+    held = db.scalar(
+        text("SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND objid = :id"),
+        {"id": pl.id},
+    )
+    assert held == 0
+
+
+def test_the_scheduled_refresh_skips_a_mirror_being_synced(db, make_user, monkeypatch):
+    user = make_user()
+    busy = _linked(db, user, source="youtube")
+    replaced: list[int] = []
+
+    monkeypatch.setattr(plsync, "has_read_scope", lambda u: True)
+    monkeypatch.setattr(plsync, "YouTubeClient", FakeYT)
+    monkeypatch.setattr(
+        plsync, "_replace_items_from_youtube", lambda d, yt, p, ids: replaced.append(p.id)
+    )
+
+    with plsync.sync_claim(db, busy.id):
+        out = plsync.sync_user_playlists(db, user)
+
+    assert out == {"synced": 1}
+    assert busy.id not in replaced and len(replaced) == 1
+    db.refresh(busy)
+    assert busy.name == "L"