fix(playlists): every sync path locks the list before its items, and holds it to the write

The mirror refresh re-ingested inside its row lock, and that ingest's commit let go of the lock
before the replace: a remove made then was undone and the list read as synced. Push, re-pull and a
delete touched items before the list row, a deadlock beside the playlist-first route edits. A push
now re-plans from the items under the lock and derives dirty after its YouTube writes.
lock_playlists moves to the sync package. Tests pin both ids of a move's lock and the claim's
own connection.

backend/tests/test_playlist_move.py

@@ -285,7 +285,7 @@ def test_no_add_can_land_in_the_target_while_a_move_holds_it(db, make_user):
     _seed_videos(db, ["v1"])
     src = _playlist(db, user, "Source", ["v1"])
     dst = _playlist(db, user, "Target", [])
-    plroutes._lock_playlists(db, src.id, dst.id)  # what the move holds until its commit
+    plroutes.lock_playlists(db, src.id, dst.id)  # what the move holds until its commit
     with SessionLocal() as other:
         other.execute(text("SET LOCAL lock_timeout = '300ms'"))
         other.add(PlaylistItem(playlist_id=dst.id, video_id="v1", position=0))

backend/tests/test_playlist_position_lock.py

@@ -183,8 +183,10 @@ PLEX_ITEMS = PlexPlaylistItem.__table__.name
             f"/api/playlists/{pid}/items/v1/move", json={"target_playlist_id": other}
         ),
         lambda c, pid, other: c.put(f"/api/playlists/{pid}/order", json={"video_ids": ["v1"]}),
+        # the ORM cascade deletes the items before the list row
+        lambda c, pid, other: c.delete(f"/api/playlists/{pid}"),
     ],
-    ids=["add-present", "add-at", "remove", "move", "reorder"],
+    ids=["add-present", "add-at", "remove", "move", "reorder", "delete"],
 )
 def test_a_youtube_item_write_locks_the_playlist_before_any_item(
     db, make_user, as_user, statements, call
@@ -208,7 +210,11 @@ def test_a_move_locks_both_lists_in_one_ordered_statement(db, make_user, as_user
     )
     assert r.status_code == 200, r.text
     lock = _locked_before_items(statements, YT, YT_ITEMS)
-    assert lock and " IN (" in lock and "ORDER BY" in lock, lock
+    assert lock and "ORDER BY" in lock, lock
+    # BOTH lists in that one statement: one bound id per list in its IN (…) (a single-id lock
+    # renders the same IN and ORDER BY)
+    listed = re.search(r" IN \((.*?)\) ORDER BY", lock)
+    assert listed and len(re.findall(r"%\(\w+\)s", listed.group(1))) == 2, lock
 
 
 def test_watch_later_add_and_remove_lock_it_before_any_item(db, make_user, as_user, statements):

backend/tests/test_playlist_sync_claim.py

@@ -92,6 +92,18 @@ def test_a_claim_is_released_when_its_block_raises(db, make_user):
     assert held == 0
 
 
+def test_a_claim_draws_no_connection_from_the_app_pool(db, make_user):
+    """A claim is held for a whole push beside the request's pooled session; drawn from the pool
+    too, a burst of pushes ran it dry and stalled every other request for pool_timeout."""
+    from app.db import engine
+
+    pl = _linked(db, make_user())
+    before = engine.pool.checkedout()
+    with plsync.sync_claim(db, pl.id) as free:
+        assert free
+        assert engine.pool.checkedout() == before
+
+
 def test_the_scheduled_refresh_skips_a_mirror_being_synced(db, make_user, monkeypatch):
     user = make_user()
     busy = _linked(db, user, source="youtube")
@@ -100,7 +112,7 @@ def test_the_scheduled_refresh_skips_a_mirror_being_synced(db, make_user, monkey
     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)
+        plsync, "_replace_items_from_youtube", lambda d, p, ids: replaced.append(p.id)
     )
 
     with plsync.sync_claim(db, busy.id):
@@ -134,7 +146,7 @@ def test_a_bulk_pull_runs_once_per_user_at_a_time(db, as_user, make_user, monkey
     monkeypatch.setattr(plsync, "has_read_scope", lambda u: True)
     monkeypatch.setattr(plroutes, "has_read_scope", lambda u: True)
     monkeypatch.setattr(plsync, "YouTubeClient", FakeYT)
-    monkeypatch.setattr(plsync, "_replace_items_from_youtube", lambda d, yt, p, ids: None)
+    monkeypatch.setattr(plsync, "_replace_items_from_youtube", lambda d, p, ids: None)
     c = as_user(user)
     with plsync.user_sync_claim(db, user.id):
         assert plsync.sync_user_playlists(db, user) == {"synced": 0, "reason": "busy"}
@@ -184,7 +196,7 @@ def test_a_mirror_edited_during_the_pull_is_not_replaced(db, make_user, monkeypa
     monkeypatch.setattr(plsync, "has_read_scope", lambda u: True)
     monkeypatch.setattr(plsync, "YouTubeClient", EditedMeanwhile)
     monkeypatch.setattr(
-        plsync, "_replace_items_from_youtube", lambda d, yt, p, ids: replaced.append(p.id)
+        plsync, "_replace_items_from_youtube", lambda d, p, ids: replaced.append(p.id)
     )
 
     assert plsync.sync_user_playlists(db, user) == {"synced": 0}

backend/tests/test_playlist_sync_lock_order.py

@@ -0,0 +1,242 @@
+"""E28 review round 3 — the YouTube sync paths take the playlist row BEFORE they touch an item, and
+hold it to the commit that writes them.
+
+Round 2 made every route edit lock the playlist first. The sync paths were left item-first (a push's
+and a re-pull's item replace, a delete's cascade), so a push beside a remove could deadlock (a 500),
+and the scheduled mirror refresh re-ingested inside its lock — its commit let go of the lock before
+the replace, so a remove made then was undone and the list read as synced. A push also replaced the
+items from a snapshot read before its YouTube calls, and marked the list clean over an edit made
+during its YouTube writes. The "other tab" here is a second session that commits while the push is
+talking to YouTube — when the push holds no lock, as a real one does not.
+"""
+
+import re
+from datetime import datetime, timezone
+
+import pytest
+from sqlalchemy import event
+
+from app.db import SessionLocal
+from app.models import Channel, Playlist, PlaylistItem, Video
+from app.sync import playlists as pl_mod
+
+ITEMS = PlaylistItem.__table__.name
+LISTS = Playlist.__table__.name
+
+
+@pytest.fixture
+def trace(db):
+    """Every statement and every COMMIT on the test engine, in order."""
+    seen: list[str] = []
+
+    def _stmt(conn, cursor, statement, *_):
+        seen.append(" ".join(statement.split()))
+
+    def _commit(conn):
+        seen.append("COMMIT")
+
+    engine = db.get_bind()
+    event.listen(engine, "before_cursor_execute", _stmt)
+    event.listen(engine, "commit", _commit)
+    yield seen
+    event.remove(engine, "before_cursor_execute", _stmt)
+    event.remove(engine, "commit", _commit)
+
+
+def _held_to_the_item_write(seen: list[str], verb: str) -> bool:
+    """The first `<verb> … playlist_items` comes after a `FOR UPDATE` of playlists, with no COMMIT
+    in between (a commit ends the lock)."""
+    locked = False
+    for st in seen:
+        if "FOR UPDATE" in st and re.search(rf"FROM {LISTS}\b", st):
+            locked = True
+        elif st == "COMMIT":
+            locked = False
+        elif st.startswith(verb) and re.search(rf"\b{ITEMS}\b", st):
+            return locked
+    return False
+
+
+class FakeYT:
+    """YouTube for the sync paths: one playlist, its items, and ids the Data API will not return."""
+
+    def __init__(self, items, *, unavailable=(), title="Mix", on_get=None, on_add=None):
+        self.items = [(f"it{i}", v) for i, v in enumerate(items)]
+        self.unavailable = set(unavailable)
+        self.title = title
+        self.get_calls = 0
+        self._on_get = on_get
+        self._on_add = on_add
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        return False
+
+    def iter_my_playlists(self):
+        return [{"id": "PLyt", "title": self.title}]
+
+    def iter_my_playlist_video_ids(self, _pid):
+        return [v for _, v in self.items]
+
+    def iter_playlist_items_with_ids(self, _pid):
+        return [{"item_id": i, "video_id": v} for i, v in self.items]
+
+    def get_playlist_snippet(self, _pid):
+        return {"title": self.title}
+
+    def get_videos(self, ids):
+        self.get_calls += 1
+        if self._on_get:
+            self._on_get()
+        return [
+            {"id": v, "snippet": {"channelId": "UCtest", "publishedAt": "2020-01-01T00:00:00Z"}}
+            for v in ids
+            if v not in self.unavailable
+        ]
+
+    def create_playlist(self, title, *_, **__):
+        return "PLyt"
+
+    def update_playlist_title(self, _pid, title):
+        self.title = title
+
+    def add_playlist_item(self, _pid, vid, position=None):
+        if self._on_add:
+            self._on_add()
+        iid = f"it{len(self.items) + 100}"
+        self.items.append((iid, vid))
+        return iid
+
+    def delete_playlist_item(self, item_id):
+        self.items = [r for r in self.items if r[0] != item_id]
+
+    def move_playlist_item(self, item_id, _pid, _vid, position):
+        row = next(r for r in self.items if r[0] == item_id)
+        self.items.remove(row)
+        self.items.insert(position, row)
+
+
+@pytest.fixture
+def fake_yt(monkeypatch):
+    def install(fake):
+        monkeypatch.setattr(pl_mod, "YouTubeClient", lambda *_: fake)
+        return fake
+
+    return install
+
+
+def _seed(db, ids):
+    db.add(Channel(id="UCtest", title="Test"))
+    now = datetime.now(timezone.utc)
+    for v in ids:
+        db.add(Video(id=v, channel_id="UCtest", published_at=now))
+    db.commit()
+
+
+def _playlist(db, user, local, *, source="local", baseline=None):
+    pl = Playlist(
+        user_id=user.id,
+        name="Mix",
+        source=source,
+        yt_playlist_id="PLyt",
+        synced_video_ids=None if baseline is None else pl_mod._dump_ids(baseline),
+    )
+    db.add(pl)
+    db.flush()
+    for pos, v in enumerate(local):
+        db.add(PlaylistItem(playlist_id=pl.id, video_id=v, position=pos))
+    db.commit()
+    return pl
+
+
+def _other_tab(pl_id, *, add=None, remove=None):
+    """A second session's edit, committed at once — what another tab's request does."""
+    with SessionLocal() as s:
+        if add:
+            pos = len(s.query(PlaylistItem).filter_by(playlist_id=pl_id).all())
+            s.add(PlaylistItem(playlist_id=pl_id, video_id=add, position=pos))
+        if remove:
+            s.query(PlaylistItem).filter_by(playlist_id=pl_id, video_id=remove).delete()
+        s.commit()
+
+
+def test_the_mirror_refresh_holds_its_lock_through_a_list_with_a_private_video(
+    db, make_user, fake_yt, trace
+):
+    user = make_user()
+    _seed(db, ["a"])
+    pl = _playlist(db, user, ["a"], source="youtube")
+    fake = fake_yt(FakeYT(["a", "ghost"], unavailable={"ghost"}))
+    trace.clear()
+    pl_mod.pull_user_playlists(db, user)
+    assert _held_to_the_item_write(trace, "DELETE"), trace
+    # the private video is asked for once, not again under the lock
+    assert fake.get_calls == 1
+    assert pl_mod._local_video_ids(db, pl) == ["a"]
+
+
+def test_the_mirror_drop_locks_the_list_before_its_items(db, make_user, fake_yt, trace):
+    user = make_user()
+    _seed(db, ["a"])
+    pl = _playlist(db, user, ["a"], source="youtube")
+    pl.yt_playlist_id = "PLgone"
+    db.commit()
+    fake_yt(FakeYT([]))
+    trace.clear()
+    pl_mod.pull_user_playlists(db, user)
+    assert _held_to_the_item_write(trace, "DELETE"), trace
+
+
+def test_a_repull_ingests_before_its_lock_and_holds_it_to_the_replace(
+    db, make_user, fake_yt, trace
+):
+    user = make_user()
+    _seed(db, ["a"])
+    pl = _playlist(db, user, ["a"])
+    fake = fake_yt(FakeYT(["b", "a", "ghost"], unavailable={"ghost"}, title="Renamed"))
+    trace.clear()
+    pl_mod.repull_playlist(db, user, pl)
+    assert _held_to_the_item_write(trace, "DELETE"), trace
+    assert fake.get_calls == 1
+    db.refresh(pl)
+    assert (pl.name, pl.dirty) == ("Renamed", False)
+    assert pl_mod._local_video_ids(db, pl) == ["b", "a"]
+
+
+def test_a_push_locks_the_list_before_it_replaces_the_items(db, make_user, fake_yt, trace):
+    user = make_user()
+    _seed(db, ["a", "b"])
+    pl = _playlist(db, user, ["a", "b"], baseline=["a"])
+    fake_yt(FakeYT(["a", "y"]))
+    trace.clear()
+    pl_mod.merge_playlist(db, user, pl)
+    assert _held_to_the_item_write(trace, "DELETE"), trace
+
+
+def test_an_add_made_while_a_push_reads_youtube_survives_it(db, make_user, fake_yt):
+    user = make_user()
+    _seed(db, ["a", "x"])
+    pl = _playlist(db, user, ["a"], baseline=["a"])
+    # 'y' is new on YouTube, so the push ingests it — and the other tab adds 'x' meanwhile.
+    fake = fake_yt(FakeYT(["a", "y"], on_get=lambda: _other_tab(pl.id, add="x")))
+    pl_mod.merge_playlist(db, user, pl)
+    local = pl_mod._local_video_ids(db, pl)
+    assert set(local) == {"a", "x", "y"}
+    assert "x" in [v for _, v in fake.items]  # and it reached YouTube as a local add
+
+
+@pytest.mark.parametrize("linked", [True, False], ids=["linked", "first-push"])
+def test_an_edit_made_during_a_push_s_youtube_writes_stays_dirty(db, make_user, fake_yt, linked):
+    user = make_user()
+    _seed(db, ["a", "b"])
+    pl = _playlist(db, user, ["a", "b"], baseline=["a"] if linked else None)
+    if not linked:
+        pl.yt_playlist_id = None
+        db.commit()
+    fake_yt(FakeYT(["a"] if linked else [], on_add=lambda: _other_tab(pl.id, remove="a")))
+    pl_mod.merge_playlist(db, user, pl)
+    db.refresh(pl)
+    assert pl_mod._local_video_ids(db, pl) == ["b"]
+    assert pl.dirty is True  # the removal has not reached YouTube; the list must say so