diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index 62a6f49a..98e89d65 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -14,6 +14,7 @@ from app.db import get_db from app.feedquery import feed_columns, serialize from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState from app.sync.playlists import ( + lock_playlists, merge_playlist, plan_merge, recompute_dirty, @@ -79,25 +80,11 @@ def _next_playlist_position(db: Session, user_id: int) -> int: ) -def _lock_playlists(db: Session, *playlist_ids: int) -> None: - """Row-lock playlists before touching their items, several in id order. Every item write takes - the PLAYLIST first: a writer that locked an item row first and the playlist second (a remove - beside a reorder) deadlocks against one that went the other way, and the loser is a 500. The - rows are re-read, so `dirty` and the sync baseline are the ones a finished sync committed.""" - db.scalars( - select(Playlist) - .where(Playlist.id.in_(sorted(set(playlist_ids)))) - .order_by(Playlist.id) - .with_for_update() - .execution_options(populate_existing=True) - ).all() # consumed: rows fetched are what refreshes the loaded objects - - def _next_item_position(db: Session, playlist_id: int) -> int: """Append position for a new item at the end of a playlist. Locks the playlist row first: two adds (two tabs, the rail and the page) otherwise both read the same max and land on one position, and the list's order between them is then arbitrary.""" - _lock_playlists(db, playlist_id) + lock_playlists(db, playlist_id) return ( db.scalar( select(func.coalesce(func.max(PlaylistItem.position), -1)).where( @@ -113,7 +100,7 @@ def _add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> was added. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later). Locked BEFORE the presence check: two adds of one video both saw "absent" and the second one's insert hit uq_playlist_video as a 500.""" - _lock_playlists(db, pl.id) + lock_playlists(db, pl.id) exists = db.scalar( select(PlaylistItem).where( PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id @@ -134,7 +121,7 @@ def _add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> def _remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool: """Remove a video from a playlist if present (idempotent). Returns whether it was removed. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later).""" - _lock_playlists(db, pl.id) + lock_playlists(db, pl.id) item = db.scalar( select(PlaylistItem).where( PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id @@ -541,6 +528,8 @@ def delete_playlist( yt.delete_playlist(pl.yt_playlist_id) except YouTubeError as exc: raise youtube_refusal(exc) + # The playlist first, as every item write: the cascade deletes the items before the list row. + lock_playlists(db, pl.id) db.delete(pl) # items cascade db.commit() return {"deleted": playlist_id} @@ -625,7 +614,7 @@ def move_item( status_code=400, detail={"code": "playlist_move", "reason": "same_playlist"} ) # Both lists, in id order: a move A→B beside a move B→A would otherwise lock crosswise. - _lock_playlists(db, src.id, dst.id) + lock_playlists(db, src.id, dst.id) item = db.scalar( select(PlaylistItem).where( PlaylistItem.playlist_id == src.id, PlaylistItem.video_id == video_id diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py index f9e2e5c2..bde91b00 100644 --- a/backend/app/sync/playlists.py +++ b/backend/app/sync/playlists.py @@ -78,6 +78,23 @@ def sync_claim(db: Session, playlist_id: int) -> AbstractContextManager[bool]: return _advisory_claim(db, _SYNC_LOCK_NS, playlist_id) +def lock_playlists(db: Session, *playlist_ids: int) -> None: + """Row-lock playlists before touching their items, several in id order. Every item write takes + the PLAYLIST first — the route edits, a push, a re-pull, the scheduled mirror refresh and a + delete alike: a writer that locked an item row first and the playlist second (a push's item + replace beside a remove) deadlocks against one that went the other way, and the loser is a + 500. The rows are re-read, so `dirty`, the name and the sync baseline are the ones a finished + writer committed. Held to the caller's next commit — so no commit may come between this and + the write it guards.""" + db.scalars( + select(Playlist) + .where(Playlist.id.in_(sorted(set(playlist_ids)))) + .order_by(Playlist.id) + .with_for_update() + .execution_options(populate_existing=True) + ).all() # consumed: rows fetched are what refreshes the loaded objects + + def fingerprint(name: str, ids: list[str]) -> str: """A stable checksum of a playlist's name + ordered video ids. Stored as the synced baseline; the current fingerprint is compared to it to derive `dirty`.""" @@ -153,6 +170,18 @@ def _set_baseline(pl: Playlist, ids: list[str], unavailable: list[str] | None = pl.dirty = False +def _set_baseline_under_lock( + db: Session, pl: Playlist, ids: list[str], unavailable: list[str] | None = None +) -> None: + """A push's closing baseline: `ids` is what YouTube now holds. The push spent seconds on + YouTube writes since it replaced the local items, so an edit may have landed meanwhile — lock + the list and derive `dirty` from what is there, or the edit would read as synced and never + reach YouTube.""" + lock_playlists(db, pl.id) + _set_baseline(pl, ids, unavailable) + recompute_dirty(db, pl) + + def recompute_dirty(db: Session, pl: Playlist) -> None: """Set `dirty` by comparing the current state to the synced baseline. Unlinked lists (and Watch later) are never dirty; a missing baseline counts as dirty (unknown YT state).""" @@ -209,15 +238,13 @@ def _ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None db.commit() -def _replace_items_from_youtube( - db: Session, yt: YouTubeClient, pl: Playlist, ids: list[str] -) -> list[str]: +def _replace_items_from_youtube(db: Session, pl: Playlist, ids: list[str]) -> list[str]: """Mirror a YouTube playlist's ordered video ids into `pl`'s local items (authoritative - replace): ensure the videos exist in the catalog, keep only those present in their YT - order (de-duplicated), swap the items out, and reset the synced fingerprint + dirty flag. - `pl.name` must already be set to the desired name. The caller commits. Returns the final - ordered ids.""" - _ensure_videos(db, yt, ids) + replace): keep only those present in the catalog in their YT order (de-duplicated), swap the + items out, and reset the synced fingerprint + dirty flag. The caller has ingested `ids` + (`_ensure_videos`, which commits) and THEN taken `lock_playlists`, set `pl.name`, and commits + after: an ingest in here re-fetched every id YouTube will not return (a private or deleted + video) and its commit let go of the lock before the replace. Returns the final ordered ids.""" present = ( set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars().all()) if ids else set() ) @@ -284,10 +311,12 @@ def pull_user_playlists(db: Session, user: User) -> dict: .scalars() .all() ) - # Drop mirrors whose YouTube playlist no longer exists. - for ytid, pl in existing.items(): - if ytid not in yt_ids: - db.delete(pl) + # Drop mirrors whose YouTube playlist no longer exists — locked first, as every item write. + gone = [pl for ytid, pl in existing.items() if ytid not in yt_ids] + if gone: + lock_playlists(db, *(pl.id for pl in gone)) + for pl in gone: + db.delete(pl) # Committed now: a run where no list below commits (all dirty, or all claimed) would # otherwise drop the deletions with the session. db.commit() @@ -325,18 +354,13 @@ def pull_user_playlists(db: Session, user: User) -> dict: # loaded rows past a commit): a remove or reorder made meanwhile must not be # replaced by YouTube's order. Item writes lock the playlist row, so this waits for # one in flight and sees what it committed. - db.scalars( - select(Playlist) - .where(Playlist.id == pl.id) - .with_for_update() - .execution_options(populate_existing=True) - ).all() # consumed: rows fetched are what refreshes the loaded objects + lock_playlists(db, pl.id) if pl.dirty: db.rollback() continue pl.name = p.get("title") or pl.name # Mirror is authoritative: replace the items to match YouTube's order. - _replace_items_from_youtube(db, yt, pl, ids) + _replace_items_from_youtube(db, pl, ids) db.commit() synced += 1 return {"synced": synced} @@ -588,7 +612,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict: if not isinstance(added, _Failed): inserted += 1 if not failures: - _set_baseline(pl, local) + _set_baseline_under_lock(db, pl, local) db.commit() return { "created": created, @@ -629,6 +653,13 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict: # merged order authoritatively. Do this first so local reflects the merge even if a later # YouTube write trips the breaker. _ensure_videos(db, yt, mp.add_to_local) + # `local` was read before the YouTube calls above: an add or remove committed meanwhile + # (another tab) would be undone by the replace below. Lock the list — the ingest's commit is + # behind us — and plan again from what is there now. Nothing new needs ingesting: a video + # the re-read adds is already a local row. + lock_playlists(db, pl.id) + local = _local_video_ids(db, pl) + mp = merge_plan(parse_baseline(pl), local, yt_ids, effective_unavailable) # Only the ids that actually exist in the catalog can become local rows (a private/deleted # YouTube video won't ingest); keep the merged order, dropping any that didn't materialize. present = ( @@ -713,7 +744,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict: (effective_unavailable | {v for v in mp.add_to_local if v not in present}) & yt_id_set ) - _set_baseline(pl, final_local, unavailable) + _set_baseline_under_lock(db, pl, final_local, unavailable) db.commit() return { "created": created, @@ -739,9 +770,11 @@ def repull_playlist(db: Session, user: User, pl: Playlist) -> dict: snippet = yt.get_playlist_snippet(pl.yt_playlist_id) if snippet is None: raise YouTubeError("Playlist no longer exists on YouTube") - pl.name = snippet.get("title") or pl.name ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id)) - ordered = _replace_items_from_youtube(db, yt, pl, ids) + _ensure_videos(db, yt, ids) + lock_playlists(db, pl.id) + pl.name = snippet.get("title") or pl.name + ordered = _replace_items_from_youtube(db, pl, ids) db.commit() return {"items": len(ordered)} diff --git a/backend/tests/test_playlist_move.py b/backend/tests/test_playlist_move.py index 30988bfc..9d6470f7 100644 --- a/backend/tests/test_playlist_move.py +++ b/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)) diff --git a/backend/tests/test_playlist_position_lock.py b/backend/tests/test_playlist_position_lock.py index 18a943fb..74e69cab 100644 --- a/backend/tests/test_playlist_position_lock.py +++ b/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): diff --git a/backend/tests/test_playlist_sync_claim.py b/backend/tests/test_playlist_sync_claim.py index c35231d7..083e610f 100644 --- a/backend/tests/test_playlist_sync_claim.py +++ b/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} diff --git a/backend/tests/test_playlist_sync_lock_order.py b/backend/tests/test_playlist_sync_lock_order.py new file mode 100644 index 00000000..c3a141d4 --- /dev/null +++ b/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 ` … 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