From b7ce1d467c1e5c002b85317a8d2b47a777a93cb3 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sun, 27 Sep 2026 19:42:53 +0200 Subject: [PATCH] fix(playlists): item writes lock the list first; one bulk pull per user; the claim on its own connection - Every playlist item write (add, remove, move, reorder, rename; Plex remove and remove-bulk) row-locks the playlist before it touches an item, a move both lists in id order: remove beside reorder was a deadlock, and two adds of one video a unique-key 500. - The scheduled mirror refresh re-reads dirty under that lock, so an edit made during its YouTube calls is not replaced by YouTube's order. - A user's bulk pull holds a per-user advisory claim: two at once created duplicate mirrors. The manual sync answers 409 sync_all_busy instead of '0 synced', and like the re-pull spends its cooldown only on a pull that runs. - Deleted mirrors are committed at once, not only when a later list commits. - The claims open an unpooled connection of their own instead of a third pooled one per push. --- backend/app/db.py | 4 +- backend/app/routes/playlists.py | 71 +++++++---- backend/app/routes/plex/playlists.py | 6 +- backend/app/sync/playlists.py | 89 +++++++++++--- backend/tests/test_abuse_limits.py | 2 +- backend/tests/test_playlist_move.py | 35 +++--- backend/tests/test_playlist_position_lock.py | 120 +++++++++++++++++++ backend/tests/test_playlist_sync_claim.py | 82 +++++++++++++ backend/tests/test_youtube_refusals.py | 2 +- frontend/src/i18n/locales/en/playlists.json | 3 +- frontend/src/i18n/locales/hu/playlists.json | 3 +- 11 files changed, 346 insertions(+), 71 deletions(-) diff --git a/backend/app/db.py b/backend/app/db.py index aba4339d..e12291b7 100644 --- a/backend/app/db.py +++ b/backend/app/db.py @@ -12,7 +12,9 @@ class Base(DeclarativeBase): # Pool sized above the sync-endpoint threadpool (~40) so a burst of short DB sessions can't starve on # checkout. The image proxies deliberately DON'T hold a connection during their slow external fetch -# (see routes/plex/), so this headroom is only for the many brief sessions, not long ones. +# (see routes/plex/), so this headroom is only for the many brief sessions, not long ones. The one +# long hold — a playlist sync's advisory claim, for a whole push — opens its own unpooled connection +# (sync/playlists.py `_claim_engine`) instead of drawing a third one from here. engine = create_engine( settings.database_url, pool_pre_ping=True, diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index e67a0160..62a6f49a 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -6,7 +6,6 @@ out (`/{id}/push-plan`, `/{id}/revert-youtube`), behind the optional write scope from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy import func, select, and_ -from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, aliased from app import quota @@ -18,9 +17,10 @@ from app.sync.playlists import ( merge_playlist, plan_merge, recompute_dirty, + pull_user_playlists, repull_playlist, sync_claim, - sync_user_playlists, + user_sync_claim, ) from app.ratelimit import RateLimiter from app.routes._common import get_or_404, owned_or_404, youtube_refusal @@ -79,11 +79,25 @@ 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.""" - db.execute(select(Playlist.id).where(Playlist.id == playlist_id).with_for_update()) + _lock_playlists(db, playlist_id) return ( db.scalar( select(func.coalesce(func.max(PlaylistItem.position), -1)).where( @@ -96,7 +110,10 @@ def _next_item_position(db: Session, playlist_id: int) -> int: def _add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool: """Append a video to a playlist if not already present (idempotent). Returns whether it - was added. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later).""" + 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) exists = db.scalar( select(PlaylistItem).where( PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id @@ -117,6 +134,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) item = db.scalar( select(PlaylistItem).where( PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id @@ -325,13 +343,18 @@ def sync_youtube( Also runs automatically on the scheduler; this is the manual 'sync now'.""" 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, **sync_user_playlists(db, user)} - except YouTubeError as exc: - db.rollback() - raise youtube_refusal(exc) + with user_sync_claim(db, user.id) as free: + if not free: + # Another run (the scheduler, another tab) is pulling right now: "0 synced" would lie. + raise _refused(409, "sync_all_busy") + # After the claim, like the re-pull's: a refused attempt must not spend one of the pulls. + _check_pull_cooldown(user) + try: + with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL): + return {"ok": True, **pull_user_playlists(db, user)} + except YouTubeError as exc: + db.rollback() + raise youtube_refusal(exc) @router.post("/{playlist_id}/revert-youtube") @@ -351,6 +374,8 @@ def revert_youtube( with sync_claim(db, pl.id) as free: if not free: raise _refused(409, "sync_busy") + # The cooldown is spent only by a pull that runs: checked after the claim, so clicking + # Revert while a push is in flight does not use up the three pulls per ten minutes. _check_pull_cooldown(user) try: with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL): @@ -464,7 +489,8 @@ def rename_playlist( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") + # Locked: the new name moves `dirty`, which the scheduled mirror refresh reads under this lock. + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist", lock=True) if "name" in body.model_fields_set: name = (body.name or "").strip() if not name: @@ -598,6 +624,8 @@ def move_item( raise HTTPException( 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) item = db.scalar( select(PlaylistItem).where( PlaylistItem.playlist_id == src.id, PlaylistItem.video_id == video_id @@ -617,20 +645,11 @@ def move_item( if already is None: # Re-parent the existing row rather than delete+insert: `added_at` ("when I saved this") # survives the move, and there is no window where the video belongs to neither list. The - # check above is not a lock: a second tab's add or the scheduler's YouTube merge can insert - # the same (playlist, video) in between, and uq_playlist_video then refuses the re-parent. - # That is the already-present outcome arriving late, not a failure — answer it the same way - # instead of a raw 500. - try: - with db.begin_nested(): - item.playlist_id = dst.id - item.position = _next_item_position(db, dst.id) - db.flush() - dst_changed = True - # silent-ok: the conflict IS the answer — the target already holds the video, which is the - # idempotent outcome; the savepoint put the row back in the source list, dropped below. - except IntegrityError: - pass + # check above holds: both lists are locked, and an insert into the target (another tab's + # add, a YouTube merge) waits for the lock — its foreign key needs the target row. + item.playlist_id = dst.id + item.position = _next_item_position(db, dst.id) + dst_changed = True if not dst_changed: db.delete(item) # The source always changed; the target only when the row landed in it. `_touch` on a target diff --git a/backend/app/routes/plex/playlists.py b/backend/app/routes/plex/playlists.py index 0f618cf9..0e742652 100644 --- a/backend/app/routes/plex/playlists.py +++ b/backend/app/routes/plex/playlists.py @@ -252,7 +252,9 @@ def playlist_remove_item( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") + # Locked like the adds and the reorder: the list row first, then its items — a remove that + # locked an item row before a reorder locked the list deadlocked against it. + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist", lock=True) it = by_rating_key(db, PlexItem, item_rating_key) if it is not None: db.execute( @@ -399,7 +401,7 @@ def playlist_remove_bulk( ) -> dict: """Remove many items at once (a whole season/show, or a multi-select). Returns how many were removed + the new total.""" - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist", lock=True) ids = [it.id for it in _items_by_rks(db, body.item_rating_keys)] removed = 0 if ids: diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py index db3417fa..f9e2e5c2 100644 --- a/backend/app/sync/playlists.py +++ b/backend/app/sync/playlists.py @@ -9,13 +9,14 @@ ingested (with a stub channel) so the mirror is faithful.""" import hashlib import json from collections.abc import Callable, Iterator -from contextlib import contextmanager +from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass from datetime import datetime, timezone from typing import TypeVar -from sqlalchemy import delete, func, select +from sqlalchemy import Engine, create_engine, delete, func, select from sqlalchemy.orm import Session +from sqlalchemy.pool import NullPool from app import logs, progress, quota from app.auth import has_read_scope @@ -28,38 +29,55 @@ 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 +# The advisory-lock namespaces ("PL" per playlist, "PU" per user), so a 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 +_USER_SYNC_LOCK_NS = 0x5055 + +# The claims' own connections, outside the app pool: a claim is held for a whole push (seconds of +# YouTube calls) beside the request's pooled session and quota's, so a burst of pushes drawing a +# third pooled connection each ran the pool dry and stalled every other request for pool_timeout. +# NullPool opens one per claim and closes it after, so a lock can never ride back into a pool either. +_claim_engines: dict[str, Engine] = {} + + +def _claim_engine(db: Session) -> Engine: + url = db.get_bind().engine.url + key = url.render_as_string(hide_password=False) + eng = _claim_engines.get(key) + if eng is None: + eng = _claim_engines[key] = create_engine(url, poolclass=NullPool, hide_parameters=True) + return eng @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() +def _advisory_claim(db: Session, ns: int, key: int) -> Iterator[bool]: + """Try a session-level advisory lock for the block on a connection of its own; yields whether it + was free. It survives the block's commits, and a crashed worker's lock dies with its connection + — no claim column that a crash leaves stuck.""" + conn = _claim_engine(db).connect() held = False try: - held = bool(conn.scalar(select(func.pg_try_advisory_lock(_SYNC_LOCK_NS, playlist_id)))) + held = bool(conn.scalar(select(func.pg_try_advisory_lock(ns, key)))) conn.commit() yield held finally: try: if held: - conn.scalar(select(func.pg_advisory_unlock(_SYNC_LOCK_NS, playlist_id))) + conn.scalar(select(func.pg_advisory_unlock(ns, key))) 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 sync_claim(db: Session, playlist_id: int) -> AbstractContextManager[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.""" + return _advisory_claim(db, _SYNC_LOCK_NS, playlist_id) + + 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`.""" @@ -228,6 +246,21 @@ def sync_user_playlists(db: Session, user: User) -> dict: the user's local playlists and the built-in Watch later untouched.""" if not has_read_scope(user): return {"synced": 0, "reason": "no read scope"} + with user_sync_claim(db, user.id) as free: + if not free: + return {"synced": 0, "reason": "busy"} + return pull_user_playlists(db, user) + + +def user_sync_claim(db: Session, user_id: int) -> AbstractContextManager[bool]: + """Claim a user's bulk mirror pull for the block; yields whether it was free. Two runs at once + (the scheduler beside a manual "sync now", two tabs) both saw a YouTube list unmirrored and + each created its mirror — a duplicate row.""" + return _advisory_claim(db, _USER_SYNC_LOCK_NS, user_id) + + +def pull_user_playlists(db: Session, user: User) -> dict: + """The mirror pull itself; the caller holds `user_sync_claim`.""" synced = 0 with YouTubeClient(db, user) as yt: yt_playlists = list(yt.iter_my_playlists()) @@ -255,7 +288,9 @@ def sync_user_playlists(db: Session, user: User) -> dict: for ytid, pl in existing.items(): if ytid not in yt_ids: db.delete(pl) - db.flush() + # Committed now: a run where no list below commits (all dirty, or all claimed) would + # otherwise drop the deletions with the session. + db.commit() for idx, p in enumerate(yt_playlists): ytid = p.get("id") @@ -282,9 +317,25 @@ def sync_user_playlists(db: Session, user: User) -> dict: # The user is pushing or re-pulling this very list: it writes the same rows. # The next scheduled run refreshes it. continue + ids = list(yt.iter_my_playlist_video_ids(ytid)) + # Ingest first — it commits — so the row lock below is held from the dirty check + # to the commit that replaces the items. + _ensure_videos(db, yt, ids) + # `dirty` read above is a snapshot from before the YouTube calls (sessions keep + # 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 + 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. - ids = list(yt.iter_my_playlist_video_ids(ytid)) _replace_items_from_youtube(db, yt, pl, ids) db.commit() synced += 1 diff --git a/backend/tests/test_abuse_limits.py b/backend/tests/test_abuse_limits.py index a6e9c80e..6ff1ca84 100644 --- a/backend/tests/test_abuse_limits.py +++ b/backend/tests/test_abuse_limits.py @@ -100,7 +100,7 @@ def test_the_single_playlist_repull_shares_the_pull_cooldown(db, as_user, make_u db.commit() monkeypatch.setattr(plroutes, "has_read_scope", lambda u: True) monkeypatch.setattr(plroutes, "repull_playlist", lambda d, u, p: {"ok": True}) - monkeypatch.setattr(plroutes, "sync_user_playlists", lambda d, u: {"ok": True}) + monkeypatch.setattr(plroutes, "pull_user_playlists", lambda d, u: {"ok": True}) c = as_user(user) codes = [c.post(f"/api/playlists/{pl.id}/revert-youtube").status_code for _ in range(4)] diff --git a/backend/tests/test_playlist_move.py b/backend/tests/test_playlist_move.py index 3f9a5fc2..30988bfc 100644 --- a/backend/tests/test_playlist_move.py +++ b/backend/tests/test_playlist_move.py @@ -13,6 +13,9 @@ Two properties are load-bearing and neither is visible from the happy path alone from datetime import datetime, timedelta, timezone +import pytest +from sqlalchemy.exc import OperationalError + from app.models import Channel, Playlist, PlaylistItem, Video @@ -269,12 +272,12 @@ def test_a_target_that_already_had_it_is_not_recently_changed(db, make_user, as_ assert db.get(Playlist, dst.id).updated_at == old -def test_a_concurrent_add_to_the_target_lands_as_the_already_present_move( - db, make_user, as_user, monkeypatch -): - """The already-in-target check is a read, not a lock: another tab's add (or the scheduler's - YouTube merge) can commit the same video to the target in between, and the re-parent then - hits uq_playlist_video. That is the idempotent outcome arriving late — not a raw 500.""" +def test_no_add_can_land_in_the_target_while_a_move_holds_it(db, make_user): + """The already-in-target check is a read; it holds because the move locks both lists first and + an insert into the target — another tab's add, a YouTube merge — needs the target row for its + foreign key, so it waits for the move's commit instead of hitting uq_playlist_video under it.""" + from sqlalchemy import text + from app.db import SessionLocal from app.routes import playlists as plroutes @@ -282,16 +285,10 @@ def test_a_concurrent_add_to_the_target_lands_as_the_already_present_move( _seed_videos(db, ["v1"]) src = _playlist(db, user, "Source", ["v1"]) dst = _playlist(db, user, "Target", []) - real = plroutes._next_item_position - - def add_first(session, playlist_id): - with SessionLocal() as other: # the other tab, committing between check and re-parent - other.add(PlaylistItem(playlist_id=dst.id, video_id="v1", position=0)) - other.commit() - return real(session, playlist_id) - - monkeypatch.setattr(plroutes, "_next_item_position", add_first) - assert _move(as_user(user), src.id, "v1", dst.id).status_code == 200 - db.expire_all() - assert _ids(db, src.id) == [] - assert _ids(db, dst.id) == ["v1"] + 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)) + with pytest.raises(OperationalError, match="lock timeout"): + other.flush() + db.rollback() diff --git a/backend/tests/test_playlist_position_lock.py b/backend/tests/test_playlist_position_lock.py index 8a2e8007..18a943fb 100644 --- a/backend/tests/test_playlist_position_lock.py +++ b/backend/tests/test_playlist_position_lock.py @@ -6,6 +6,11 @@ on one position, and a reorder racing an add collides with it — the list's ord rows is then whatever the planner returns. A true two-connection race is not reproducible on demand, so the test holds the mechanism: the parent row is taken `FOR UPDATE` by each such route. A Plex reorder that omits a row (added in another tab meanwhile) must not collide with it either. + +Round 2: the lock also comes FIRST — before the request reads or writes a single item row. A remove +that deleted an item and then touched the playlist beside a reorder that went the other way was a +deadlock (a 500 for one of them), and an add that checked "already there?" before the lock let two +adds of one video both insert (a 500 on the unique key). """ import re @@ -41,6 +46,32 @@ def locks(db): event.remove(engine, "before_cursor_execute", _record) +@pytest.fixture +def statements(db): + """Every statement a request ran, in order.""" + seen: list[str] = [] + + def _record(conn, cursor, statement, *_): + seen.append(" ".join(statement.split())) + + engine = db.get_bind() + event.listen(engine, "before_cursor_execute", _record) + yield seen + event.remove(engine, "before_cursor_execute", _record) + + +def _locked_before_items(seen: list[str], parent: str, items: str) -> str | None: + """The first statement touching `items` must come after a `FOR UPDATE` of `parent`; returns + that lock statement (None when it came too late or never).""" + lock = None + for st in seen: + if lock is None and "FOR UPDATE" in st and re.search(rf"FROM {parent}\b", st): + lock = st + if re.search(rf"\b{items}\b", st): + return lock + return lock + + def _yt(db, user): db.add(Channel(id="UCtest", title="Test")) now = datetime.now(timezone.utc) @@ -134,3 +165,92 @@ def test_a_plex_reorder_that_omits_a_row_does_not_collide_with_it(db, make_user, by_id = {it.id: it.rating_key for it in db.query(PlexItem).all()} got = sorted((r.position, by_id[r.item_id]) for r in rows) assert got == [(0, "c"), (1, "b"), (2, "a")] + + +YT_ITEMS = PlaylistItem.__table__.name +PLEX_ITEMS = PlexPlaylistItem.__table__.name + + +@pytest.mark.parametrize( + "call", + [ + lambda c, pid, other: c.post(f"/api/playlists/{pid}/items", json={"video_id": "v1"}), + lambda c, pid, other: c.post( + f"/api/playlists/{pid}/items", json={"video_id": "v2", "position": 0} + ), + lambda c, pid, other: c.delete(f"/api/playlists/{pid}/items/v1"), + lambda c, pid, other: c.post( + 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"]}), + ], + ids=["add-present", "add-at", "remove", "move", "reorder"], +) +def test_a_youtube_item_write_locks_the_playlist_before_any_item( + db, make_user, as_user, statements, call +): + user = make_user() + pid, other = _yt(db, user) + c = as_user(user) + statements.clear() + r = call(c, pid, other) + assert r.status_code == 200, r.text + assert _locked_before_items(statements, YT, YT_ITEMS), statements + + +def test_a_move_locks_both_lists_in_one_ordered_statement(db, make_user, as_user, statements): + """Two moves A→B and B→A each locking "its" list first would lock crosswise.""" + user = make_user() + pid, other = _yt(db, user) + statements.clear() + r = as_user(user).post( + f"/api/playlists/{pid}/items/v1/move", json={"target_playlist_id": other} + ) + 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 + + +def test_watch_later_add_and_remove_lock_it_before_any_item(db, make_user, as_user, statements): + user = make_user() + _yt(db, user) + c = as_user(user) + assert c.post("/api/playlists/watch-later", json={"video_id": "v2"}).status_code == 200 + statements.clear() + assert c.post("/api/playlists/watch-later", json={"video_id": "v2"}).status_code == 200 + assert _locked_before_items(statements, YT, YT_ITEMS), statements + statements.clear() + assert c.delete("/api/playlists/watch-later/v2").status_code == 200 + assert _locked_before_items(statements, YT, YT_ITEMS), statements + + +def test_a_rename_locks_the_playlist(db, make_user, as_user, locks): + """A rename moves `dirty`, which the scheduled mirror refresh re-reads under this lock.""" + user = make_user() + pid, _ = _yt(db, user) + r = as_user(user).patch(f"/api/playlists/{pid}", json={"name": "New"}) + assert r.status_code == 200, r.text + assert YT in locks + + +@pytest.mark.parametrize( + "call", + [ + lambda c, pid: c.delete(f"/api/plex/playlists/{pid}/items/a"), + lambda c, pid: c.post( + f"/api/plex/playlists/{pid}/items/remove-bulk", json={"item_rating_keys": ["a", "b"]} + ), + lambda c, pid: c.post(f"/api/plex/playlists/{pid}/items", json={"item_rating_key": "a"}), + ], + ids=["remove", "remove-bulk", "add-present"], +) +def test_a_plex_item_write_locks_the_playlist_before_any_item( + db, make_user, as_user, statements, call +): + user = make_user() + pid = _plex(db, user) + c = as_user(user) + statements.clear() + r = call(c, pid) + assert r.status_code == 200, r.text + assert _locked_before_items(statements, PLEX, PLEX_ITEMS), statements diff --git a/backend/tests/test_playlist_sync_claim.py b/backend/tests/test_playlist_sync_claim.py index c0bc451b..c35231d7 100644 --- a/backend/tests/test_playlist_sync_claim.py +++ b/backend/tests/test_playlist_sync_claim.py @@ -9,6 +9,7 @@ on a connection of its own; the test holds it from outside, the way a second req from sqlalchemy import text from app.auth import require_write_scope +from app.db import SessionLocal from app.models import Playlist from app.routes import playlists as plroutes from app.sync import playlists as plsync @@ -109,3 +110,84 @@ def test_the_scheduled_refresh_skips_a_mirror_being_synced(db, make_user, monkey assert busy.id not in replaced and len(replaced) == 1 db.refresh(busy) assert busy.name == "L" + + +def test_refused_reverts_do_not_spend_the_pull_cooldown( + db, as_user, make_user, override_dep, monkeypatch +): + """Round 2: the cooldown (three pulls per ten minutes) is checked after the claim, so clicking + Revert while a push runs is refused as busy without using up a pull.""" + user = make_user() + pl = _linked(db, user) + monkeypatch.setattr(plroutes, "has_read_scope", lambda u: True) + monkeypatch.setattr(plroutes, "repull_playlist", lambda d, u, p: {}) + c = as_user(user) + with plsync.sync_claim(db, pl.id): + for _ in range(4): + assert c.post(f"/api/playlists/{pl.id}/revert-youtube").status_code == 409 + assert c.post(f"/api/playlists/{pl.id}/revert-youtube").status_code == 200 + + +def test_a_bulk_pull_runs_once_per_user_at_a_time(db, as_user, make_user, monkeypatch): + """Two bulk pulls at once both saw a YouTube list unmirrored and each created its mirror.""" + user = make_user() + 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) + c = as_user(user) + with plsync.user_sync_claim(db, user.id): + assert plsync.sync_user_playlists(db, user) == {"synced": 0, "reason": "busy"} + for _ in range(4): + r = c.post("/api/playlists/sync-youtube") + assert r.status_code == 409 + assert r.json()["detail"] == {"code": "playlist", "reason": "sync_all_busy"} + assert c.post("/api/playlists/sync-youtube").json() == {"ok": True, "synced": 2} + + +def test_a_gone_mirror_is_deleted_even_when_no_list_refreshes(db, make_user, monkeypatch): + """Every remaining mirror has unpushed edits, so nothing below the deletion commits.""" + user = make_user() + gone = Playlist(user_id=user.id, name="Gone", source="youtube", yt_playlist_id="PLgone") + kept = Playlist(user_id=user.id, name="K", source="youtube", yt_playlist_id="PL123", dirty=True) + other = Playlist( + user_id=user.id, name="O", source="youtube", yt_playlist_id="PL999", dirty=True + ) + db.add_all([gone, kept, other]) + db.commit() + gone_id = gone.id + monkeypatch.setattr(plsync, "has_read_scope", lambda u: True) + monkeypatch.setattr(plsync, "YouTubeClient", FakeYT) + + assert plsync.sync_user_playlists(db, user) == {"synced": 0} + db.rollback() # what the request's session close does to an uncommitted deletion + assert db.get(Playlist, gone_id) is None + + +def test_a_mirror_edited_during_the_pull_is_not_replaced(db, make_user, monkeypatch): + """The loop's `dirty` is a snapshot from before the YouTube calls; a remove committed by another + request meanwhile must win over YouTube's order.""" + user = make_user() + pl = _linked(db, user, source="youtube") + replaced: list[int] = [] + + class EditedMeanwhile(FakeYT): + def iter_my_playlists(self): + return [{"id": "PL123", "title": "Renamed"}] + + def iter_my_playlist_video_ids(self, _ytid): + with SessionLocal() as other: + other.get(Playlist, pl.id).dirty = True + other.commit() + return [] + + 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) + ) + + assert plsync.sync_user_playlists(db, user) == {"synced": 0} + assert replaced == [] + db.expire_all() + assert db.get(Playlist, pl.id).name == "L" diff --git a/backend/tests/test_youtube_refusals.py b/backend/tests/test_youtube_refusals.py index a87cfb78..b4043757 100644 --- a/backend/tests/test_youtube_refusals.py +++ b/backend/tests/test_youtube_refusals.py @@ -37,7 +37,7 @@ def test_a_failure_is_sorted_by_what_the_user_can_do(exc, reason): ("route", "target"), [ ("/api/sync/subscriptions", "app.routes.sync.import_subscriptions"), - ("/api/playlists/sync-youtube", "app.routes.playlists.sync_user_playlists"), + ("/api/playlists/sync-youtube", "app.routes.playlists.pull_user_playlists"), ], ) def test_a_manual_sync_youtube_refuses_is_a_422_not_a_500( diff --git a/frontend/src/i18n/locales/en/playlists.json b/frontend/src/i18n/locales/en/playlists.json index 201a0218..e6adfa49 100644 --- a/frontend/src/i18n/locales/en/playlists.json +++ b/frontend/src/i18n/locales/en/playlists.json @@ -87,7 +87,8 @@ "watch_later_undeletable": "The Watch later list can't be deleted", "youtube_editing_off": "Enable YouTube editing in Settings first", "quota_exceeded": "Not enough YouTube quota left today for this sync", - "sync_busy": "This playlist is already syncing with YouTube — try again when it finishes" + "sync_busy": "This playlist is already syncing with YouTube — try again when it finishes", + "sync_all_busy": "Your playlists are already syncing with YouTube — try again when it finishes" }, "moveErrors": { "not_in_playlist": "That video isn't in this playlist any more — it changed somewhere else", diff --git a/frontend/src/i18n/locales/hu/playlists.json b/frontend/src/i18n/locales/hu/playlists.json index 9ae6c844..4d3771aa 100644 --- a/frontend/src/i18n/locales/hu/playlists.json +++ b/frontend/src/i18n/locales/hu/playlists.json @@ -87,7 +87,8 @@ "watch_later_undeletable": "A „Megnézem később” lista nem törölhető", "youtube_editing_off": "Előbb engedélyezd a YouTube-szerkesztést a Beállításokban", "quota_exceeded": "Mára nem maradt elég YouTube-keret ehhez a szinkronhoz", - "sync_busy": "Ez a lista épp szinkronizál a YouTube-bal — próbáld újra, ha végzett" + "sync_busy": "Ez a lista épp szinkronizál a YouTube-bal — próbáld újra, ha végzett", + "sync_all_busy": "A listáid épp szinkronizálnak a YouTube-bal — próbáld újra, ha végzett" }, "moveErrors": { "not_in_playlist": "Ez a videó már nincs ebben a listában — időközben máshol megváltozott",