fix(sync): an empty answer no longer empties a mirror; maintenance checks each video once a run

The catalog prune and the subscription import have long refused to reconcile against an empty
listing that contradicts what we hold (should_prune: a 200 with no items is a blip far more often
than a user who removed everything). Two mirrors did not: the YouTube playlist mirrors (dropped
every mirror) and the Plex collections (dropped every collection with its user-set `editable`
flag, and every member's keys — the guard landed in plex/sync.py with the previous commit). Both
now hold the same rule; test_empty_listing_guards covers each way (seen red).

sync/maintenance (19% covered) gets a lifecycle test: recover, grace, delete + one notification
per user, live stream kept, abandoned premiere. It found the rolling revalidation re-fetching the
same videos until `checked` reached the batch (15000 by default) whenever fewer videos were
available — quota spent on answers it already had; a run now skips what it has checked.

backend/app/sync/maintenance.py

@@ -22,7 +22,7 @@ backfill reserve as the other jobs so interactive syncs never starve.
 
 from datetime import timedelta
 
-from sqlalchemy import select
+from sqlalchemy import or_, select
 from sqlalchemy.orm import Session
 
 from app import logs, progress, quota, sysconfig
@@ -190,6 +190,10 @@ def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict:
             db.execute(
                 select(Video)
                 .where(Video.unavailable_since.is_(None))
+                # Not one this run already checked: with fewer available videos than the batch, the
+                # loop used to wrap around and re-fetch the same ones until `checked` reached the
+                # batch, spending quota on answers it already had (E32 S2).
+                .where(or_(Video.last_checked_at.is_(None), Video.last_checked_at < now))
                 # NULLS FIRST = never-checked first, then least-recently-checked; equivalent to the old
                 # `(last_checked_at IS NULL) DESC, last_checked_at ASC` but in the exact shape of the
                 # partial ix_videos_recheck_order (migration 0069, PERF-18) so it rides that index.

backend/app/sync/playlists.py

@@ -22,6 +22,7 @@ from app import logs, progress, quota
 from app.auth import has_read_scope
 from app.models import Channel, Playlist, PlaylistItem, User, Video
 from app.sync.runner import token_users
+from app.sync.subscriptions import should_prune
 from app.sync.videos import apply_video_details, parse_dt
 from app.youtube.client import YouTubeClient, YouTubeError
 
@@ -342,7 +343,13 @@ def pull_user_playlists(db: Session, user: User) -> dict:
             .all()
         )
         # 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]
+        # Never on an empty answer that contradicts the mirrors we hold: the same 200-with-no-items
+        # blip `should_prune` guards the subscriptions against would drop every mirror (E32 S2).
+        gone = (
+            [pl for ytid, pl in existing.items() if ytid not in yt_ids]
+            if should_prune(len(yt_ids), len(existing))
+            else []
+        )
         if gone:
             lock_playlists(db, *(pl.id for pl in gone))
         for pl in gone:

backend/tests/test_empty_listing_guards.py

@@ -0,0 +1,111 @@
+"""An empty answer is not an emptied library (E32 S2).
+
+Every mirror here reconciles: what the remote no longer lists is removed locally. The catalog
+(`_prune_stale`) and the subscriptions (`should_prune`) had long refused to do that on an empty
+listing that contradicts what we hold — a 200 with no items is a blip far more often than a user
+who removed everything. Two mirrors did not: the Plex collections (a row carries the user-set
+`editable` flag, and every member its keys) and the YouTube playlist mirrors. These hold both.
+"""
+
+from sqlalchemy import select
+
+from app.models import Playlist, PlexCollection, PlexItem, PlexLibrary
+from app.plex import sync as plex_sync
+from app.sync import playlists as plsync
+
+
+class _Plex:
+    def __init__(self, collections: list[dict], children: dict[str, list[dict]]):
+        self._collections = collections
+        self._children = children
+
+    def collections(self, section_key):
+        return self._collections
+
+    def collection_children(self, rk):
+        return self._children.get(rk, [])
+
+
+def _library_with_a_collection(db):
+    lib = PlexLibrary(plex_key="4", title="Movies", kind="movie")
+    db.add(lib)
+    db.flush()
+    db.add(PlexCollection(rating_key="c1", library_id=lib.id, title="Bond", editable=True))
+    movie = PlexItem(
+        rating_key="m1", library_id=lib.id, kind="movie", title="Dr. No", collection_keys=["c1"]
+    )
+    db.add(movie)
+    db.commit()
+    return lib, movie
+
+
+def test_an_empty_collection_listing_keeps_the_collections_and_their_members(db):
+    lib, movie = _library_with_a_collection(db)
+    stats = {"collections": 0}
+    plex_sync._sync_collections(db, _Plex([], {}), lib, stats)
+    db.commit()
+    db.expire_all()
+    col = db.execute(select(PlexCollection).where(PlexCollection.rating_key == "c1")).scalar_one()
+    assert col.editable is True
+    assert db.get(PlexItem, movie.id).collection_keys == ["c1"]
+
+
+def test_a_listing_with_something_in_it_still_prunes_what_it_no_longer_has(db):
+    lib, movie = _library_with_a_collection(db)
+    stats = {"collections": 0}
+    other = {"ratingKey": "c2", "title": "Alien"}
+    plex_sync._sync_collections(db, _Plex([other], {"c2": []}), lib, stats)
+    db.commit()
+    db.expire_all()
+    keys = set(db.execute(select(PlexCollection.rating_key)).scalars())
+    assert keys == {"c2"}
+    assert db.get(PlexItem, movie.id).collection_keys is None
+
+
+class _YT:
+    def __init__(self, playlists: list[dict]):
+        self._playlists = playlists
+
+    def __call__(self, *_):
+        return self
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        return False
+
+    def iter_my_playlists(self):
+        return self._playlists
+
+    def iter_playlist_items_with_ids(self, _ytid):
+        return []
+
+    def iter_my_playlist_video_ids(self, _ytid):
+        return []
+
+
+def test_an_empty_playlist_answer_keeps_the_youtube_mirrors(db, make_user, monkeypatch):
+    user = make_user()
+    db.add(Playlist(user_id=user.id, name="Mixes", source="youtube", yt_playlist_id="PL1"))
+    db.commit()
+    monkeypatch.setattr(plsync, "YouTubeClient", _YT([]))
+    plsync.pull_user_playlists(db, user)
+    db.expire_all()
+    names = db.execute(select(Playlist.name).where(Playlist.user_id == user.id)).scalars().all()
+    assert "Mixes" in names
+
+
+def test_a_playlist_answer_with_something_in_it_still_drops_a_vanished_mirror(
+    db, make_user, monkeypatch
+):
+    user = make_user()
+    db.add(Playlist(user_id=user.id, name="Mixes", source="youtube", yt_playlist_id="PL1"))
+    db.commit()
+    monkeypatch.setattr(plsync, "YouTubeClient", _YT([{"id": "PL2", "title": "New"}]))
+    plsync.pull_user_playlists(db, user)
+    db.expire_all()
+    ytids = set(
+        db.execute(select(Playlist.yt_playlist_id).where(Playlist.source == "youtube")).scalars()
+    )
+    assert "PL1" not in ytids

backend/tests/test_maintenance_lifecycle.py

@@ -0,0 +1,147 @@
+"""sync/maintenance — the job that HARD-DELETES videos (E32 S2: 19% covered).
+
+Its lifecycle: a video missing from videos.list is flagged and hidden; after the grace period a
+still-missing one is deleted (cascading to every playlist that held it) and each affected user
+gets one notification; a video that answers again is unhidden. A live broadcast is never a delete
+trigger, and an upcoming premiere that never went live is. YouTube is a fake that answers
+videos.list from a dict.
+"""
+
+from datetime import timedelta
+
+import pytest
+from sqlalchemy import select
+
+from app.models import Channel, Notification, Playlist, PlaylistItem, Video
+from app.sync import maintenance
+from app.utils import now_utc
+
+
+class _YT:
+    def __init__(self, items: dict[str, dict]):
+        self.items = items
+        self.asked: list[list[str]] = []
+
+    def get_videos(self, ids):
+        self.asked.append(list(ids))
+        return [self.items[i] for i in ids if i in self.items]
+
+
+def _item(vid: str, **snippet) -> dict:
+    return {"id": vid, "snippet": {"title": vid, **snippet}}
+
+
+@pytest.fixture
+def channel(db):
+    ch = Channel(id="UC_maint", title="Maint")
+    db.add(ch)
+    db.flush()
+    return ch
+
+
+def test_a_flagged_video_that_answers_again_is_unhidden(db, channel):
+    v = Video(
+        id="v1", channel_id=channel.id, unavailable_since=now_utc(), unavailable_reason="removed"
+    )
+    db.add(v)
+    db.commit()
+    out = maintenance._recheck_flagged(db, _YT({"v1": _item("v1")}))
+    db.refresh(v)
+    assert v.unavailable_since is None and v.unavailable_reason is None
+    assert out == {"checked": 1, "recovered": 1, "deleted": 0, "notified": 0}
+
+
+def test_a_video_still_missing_inside_its_grace_period_is_kept_hidden(db, channel):
+    v = Video(
+        id="v1",
+        channel_id=channel.id,
+        unavailable_since=now_utc() - timedelta(days=maintenance.DEFAULT_GRACE_DAYS - 1),
+        unavailable_reason="removed",
+    )
+    db.add(v)
+    db.commit()
+    out = maintenance._recheck_flagged(db, _YT({}))
+    assert db.get(Video, "v1") is not None
+    assert out["deleted"] == 0
+
+
+def test_past_the_grace_period_it_is_deleted_and_each_user_who_kept_it_told_once(
+    db, channel, make_user
+):
+    a, b = make_user(), make_user()
+    gone_since = now_utc() - timedelta(days=maintenance.DEFAULT_GRACE_DAYS + 1)
+    for vid in ("v1", "v2"):
+        db.add(
+            Video(
+                id=vid,
+                channel_id=channel.id,
+                title=f"T {vid}",
+                unavailable_since=gone_since,
+                unavailable_reason="removed",
+            )
+        )
+    pa = Playlist(user_id=a.id, name="A")
+    pb = Playlist(user_id=b.id, name="B")
+    db.add_all([pa, pb])
+    db.flush()
+    db.add_all(
+        [
+            PlaylistItem(playlist_id=pa.id, video_id="v1", position=0),
+            PlaylistItem(playlist_id=pa.id, video_id="v2", position=1),
+            PlaylistItem(playlist_id=pb.id, video_id="v2", position=0),
+        ]
+    )
+    db.commit()
+    out = maintenance._recheck_flagged(db, _YT({}))
+    db.expire_all()
+    assert db.get(Video, "v1") is None and db.get(Video, "v2") is None
+    assert db.execute(select(PlaylistItem)).scalars().all() == []  # the cascade
+    notes = {
+        n.user_id: n.data["count"]
+        for n in db.execute(select(Notification).where(Notification.type == "maintenance"))
+        .scalars()
+        .all()
+    }
+    assert notes == {a.id: 2, b.id: 1}  # one per user, not one per video
+    assert out == {"checked": 2, "recovered": 0, "deleted": 2, "notified": 2}
+
+
+def test_rolling_revalidation_flags_the_missing_and_keeps_a_live_stream(db, channel, monkeypatch):
+    monkeypatch.setattr(maintenance.sysconfig, "get_int", lambda db, key: 10)
+    monkeypatch.setattr(maintenance.quota, "backfill_reserve_reached", lambda: False)
+    db.add_all(
+        [
+            Video(id="ok", channel_id=channel.id),
+            Video(id="gone", channel_id=channel.id),
+            Video(id="live", channel_id=channel.id),
+        ]
+    )
+    db.commit()
+    yt = _YT({"ok": _item("ok"), "live": _item("live", liveBroadcastContent="live")})
+    out = maintenance._revalidate_rolling(db, yt)
+    db.expire_all()
+    assert db.get(Video, "gone").unavailable_reason == "removed"
+    assert db.get(Video, "ok").unavailable_since is None
+    assert db.get(Video, "live").unavailable_since is None
+    assert out == {"checked": 3, "flagged": 1}
+    assert all(db.get(Video, v).last_checked_at is not None for v in ("ok", "gone", "live"))
+
+
+def test_an_upcoming_premiere_counts_as_abandoned_only_when_it_never_went_live():
+    past = (now_utc() - timedelta(days=3)).isoformat()
+    soon = (now_utc() - timedelta(days=1)).isoformat()
+
+    def item(**live):
+        return {"snippet": {"liveBroadcastContent": "upcoming"}, "liveStreamingDetails": live}
+
+    assert maintenance._is_abandoned_upcoming(item(scheduledStartTime=past)) is True
+    assert maintenance._is_abandoned_upcoming(item(scheduledStartTime=soon)) is False
+    assert (
+        maintenance._is_abandoned_upcoming(item(scheduledStartTime=past, actualStartTime=past))
+        is False
+    )
+    assert (
+        maintenance._is_abandoned_upcoming({"snippet": {"liveBroadcastContent": "none"}}) is False
+    )
+    # …and an abandoned premiere has no grace period: the confirming sweep deletes it.
+    assert maintenance._grace_days("abandoned") == 0