diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index a4e0e741..e67a0160 100644 --- a/backend/app/routes/playlists.py +++ b/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}") diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py index 2a889234..db3417fa 100644 --- a/backend/app/sync/playlists.py +++ b/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} diff --git a/backend/tests/test_playlist_sync_claim.py b/backend/tests/test_playlist_sync_claim.py new file mode 100644 index 00000000..c0bc451b --- /dev/null +++ b/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" diff --git a/frontend/src/i18n/locales/en/playlists.json b/frontend/src/i18n/locales/en/playlists.json index 8c5205a3..201a0218 100644 --- a/frontend/src/i18n/locales/en/playlists.json +++ b/frontend/src/i18n/locales/en/playlists.json @@ -86,7 +86,8 @@ "watch_later_not_pushable": "The Watch later list can't be sent to YouTube", "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" + "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" }, "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 0f01ee8b..9ae6c844 100644 --- a/frontend/src/i18n/locales/hu/playlists.json +++ b/frontend/src/i18n/locales/hu/playlists.json @@ -86,7 +86,8 @@ "watch_later_not_pushable": "A „Megnézem később” listát nem lehet a YouTube-ra küldeni", "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" + "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" }, "moveErrors": { "not_in_playlist": "Ez a videó már nincs ebben a listában — időközben máshol megváltozott",