fix(sync): R5 S2 — sync guards

- never prune subscriptions on an empty fetch: one anomalous empty-but-200
  response wiped every subscription (and its deep_requested flag) in one run
- push_playlist: stop after 5 consecutive write failures — a quota-dead or
  scope-revoked account otherwise burns 50 units per remaining item
- RSS: raise RssError instead of returning [] on a transport/HTTP failure, so a
  failed poll no longer stamps last_rss_at and a 404 feed stops looking healthy
- worker: exit when the schema never appears instead of "continuing" into a
  crash one call later

backend/app/sync/playlists.py

@@ -263,6 +263,40 @@ def plan_push(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
     }
 
 
+# After this many consecutive write failures we stop calling YouTube for the rest of the push.
+# The failure modes that matter here are all account-wide, not per-video: quota exhausted, the
+# write scope revoked, the token dead. Every remaining write then fails identically — but each
+# one still SPENDS 50 units, so a long playlist burns through the whole (already dead) budget,
+# and tomorrow's too, to produce a failure list we knew after the fifth item.
+_MAX_CONSECUTIVE_FAILURES = 5
+
+
+class _Breaker:
+    """Consecutive-failure circuit breaker for a single push run. Any success re-arms it."""
+
+    def __init__(self, limit: int = _MAX_CONSECUTIVE_FAILURES) -> None:
+        self.limit = limit
+        self.streak = 0
+        self.tripped = False
+
+    def open(self) -> bool:
+        """True while writes may still be attempted."""
+        return not self.tripped
+
+    def hit(self, ok: bool) -> None:
+        if ok:
+            self.streak = 0
+            return
+        self.streak += 1
+        if self.streak >= self.limit and not self.tripped:
+            self.tripped = True
+            log.warning(
+                "playlist push: %d consecutive YouTube write failures — stopping this run "
+                "(quota, scope or token); the remaining items are reported as failures",
+                self.limit,
+            )
+
+
 def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
     """Push a local playlist to YouTube: create it if needed, then reconcile membership
     and order so YouTube matches local (local wins). Marks the playlist clean on success.
@@ -271,6 +305,7 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
     desired = _local_video_ids(db, pl)
     inserted = deleted = reordered = created = 0
     failures: list[str] = []
+    breaker = _Breaker()
     with YouTubeClient(db, user) as yt:
         if not pl.yt_playlist_id:
             pl.yt_playlist_id = yt.create_playlist(pl.name)
@@ -278,11 +313,16 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
             created = 1
             # Brand-new playlist: just append every video in order.
             for vid in desired:
+                if not breaker.open():
+                    failures.append(vid)
+                    continue
                 try:
                     yt.add_playlist_item(pl.yt_playlist_id, vid)
                     inserted += 1
+                    breaker.hit(True)
                 except YouTubeError:
                     failures.append(vid)
+                    breaker.hit(False)
             if not failures:
                 pl.synced_fingerprint = fingerprint(pl.name, desired)
                 pl.dirty = False
@@ -293,6 +333,7 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
                 "deleted": deleted,
                 "reordered": reordered,
                 "failures": failures,
+                "aborted": breaker.tripped,
                 "yt_playlist_id": pl.yt_playlist_id,
             }
 
@@ -313,20 +354,30 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
         # Delete extras (local is authoritative).
         for it in current:
             if it["video_id"] not in desired_set:
+                if not breaker.open():
+                    failures.append(it["video_id"])
+                    continue
                 try:
                     yt.delete_playlist_item(it["item_id"])
                     deleted += 1
+                    breaker.hit(True)
                 except YouTubeError:
                     failures.append(it["video_id"])
+                    breaker.hit(False)
         # Insert missing (append; the reorder pass below fixes positions).
         cur_set = set(cur_vids)
         for vid in desired:
             if vid not in cur_set:
+                if not breaker.open():
+                    failures.append(vid)
+                    continue
                 try:
                     item_id_by_vid[vid] = yt.add_playlist_item(pl.yt_playlist_id, vid)
                     inserted += 1
+                    breaker.hit(True)
                 except YouTubeError:
                     failures.append(vid)
+                    breaker.hit(False)
         # Reorder to match `desired` via insertion-sort moves over a local model.
         model = [v for v in cur_vids if v in desired_set] + [
             v for v in desired if v not in cur_set and v in item_id_by_vid
@@ -336,11 +387,16 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
                 continue  # failed insert; skip
             if i < len(model) and model[i] == want:
                 continue
+            if not breaker.open():
+                failures.append(want)
+                continue
             try:
                 yt.move_playlist_item(item_id_by_vid[want], pl.yt_playlist_id, want, i)
                 reordered += 1
+                breaker.hit(True)
             except YouTubeError:
                 failures.append(want)
+                breaker.hit(False)
                 continue
             j = model.index(want)
             model.pop(j)
@@ -355,6 +411,7 @@ def push_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
         "deleted": deleted,
         "reordered": reordered,
         "failures": failures,
+        "aborted": breaker.tripped,
         "yt_playlist_id": pl.yt_playlist_id,
     }
 

backend/app/sync/runner.py

@@ -25,7 +25,7 @@ from app.sync.videos import (
     run_shorts_classification,
 )
 from app.youtube.client import YouTubeClient
-from app.youtube.rss import fetch_channel_feed
+from app.youtube.rss import RssError, fetch_channel_feed
 
 # RSS feeds are free and network-bound, so fetch many at once; DB writes stay single-threaded.
 RSS_POLL_WORKERS = 16
@@ -54,6 +54,7 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
     total = len(channels)
     new = 0
     done = 0
+    failed = 0
     # Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as
     # each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a
     # plain str) to the workers; the ORM instance is touched only here on the main thread.
@@ -65,9 +66,20 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
             channel = futures[fut]
             try:
                 entries = fut.result()
+            except RssError as exc:
+                # A failed poll is NOT an empty feed: skip the apply so `last_rss_at` stays where
+                # it was. Stamping it here made a permanently-broken feed look freshly polled.
+                log.warning("RSS poll skipped for channel %s: %s", channel.id, exc)
+                failed += 1
+                done += 1
+                progress.report(done, total, "rss_poll")
+                continue
             except Exception:
                 log.exception("RSS fetch failed for channel %s", channel.id)
-                entries = []
+                failed += 1
+                done += 1
+                progress.report(done, total, "rss_poll")
+                continue
             try:
                 new += apply_rss_feed(db, channel, entries)
             except Exception:
@@ -75,6 +87,8 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
                 log.exception("RSS apply failed for channel %s", channel.id)
             done += 1
             progress.report(done, total, "rss_poll")
+    if failed:
+        log.warning("RSS poll: %d/%d channel feeds could not be read", failed, total)
     return new
 
 

backend/app/sync/subscriptions.py

@@ -108,17 +108,29 @@ def import_subscriptions(db: Session, user: User) -> dict:
             sub_row.yt_subscription_id = sub.get("yt_subscription_id")
         db.flush()
 
-        # Prune subscriptions removed on YouTube.
+        # Prune subscriptions removed on YouTube — but NEVER on an empty fetch. The API can answer
+        # 200 with no items (a transient blip, a scope/token edge), and pruning against that wipes
+        # every subscription the user has, taking each row's `deep_requested` flag with it, in one
+        # run. Someone who genuinely unsubscribed from everything simply keeps their rows until a
+        # sync that returns something.
         removed = 0
         existing = (
             db.execute(select(Subscription).where(Subscription.user_id == user.id))
             .scalars()
             .all()
         )
-        for sub_row in existing:
+        prune_skipped = bool(existing) and not fetched
-            if sub_row.channel_id not in fetched:
+        if prune_skipped:
-                db.delete(sub_row)
+            log.warning(
-                removed += 1
+                "Subscription import (user %s): YouTube returned 0 subscriptions while %d exist "
+                "locally — skipping the prune (treating it as a transient empty response)",
+                user.id, len(existing),
+            )
+        else:
+            for sub_row in existing:
+                if sub_row.channel_id not in fetched:
+                    db.delete(sub_row)
+                    removed += 1
         db.commit()
 
         # Fetch details for channels that don't have them yet.
@@ -144,6 +156,7 @@ def import_subscriptions(db: Session, user: User) -> dict:
         "channels_new": new_channels,
         "channels_detailed": detailed,
         "removed_stale": removed,
+        "prune_skipped": prune_skipped,
     }
     log.info("Subscription import (user %s): %s", user.id, result)
     return result

backend/app/worker.py

@@ -93,24 +93,33 @@ def _parse_upload_date(raw) -> date | None:
 
 # --- claim + recovery -----------------------------------------------------------------------
 
-def _wait_for_schema(timeout: float = 300.0) -> None:
+def _wait_for_schema(timeout: float = 300.0) -> bool:
-    """Block until the download tables exist before doing any DB work.
+    """Block until the download tables exist before doing any DB work. Returns False if they never
+    appeared (or we were asked to stop while waiting).
 
     The API applies migrations on its OWN startup, and the worker is a separate container that may
     boot first (they start in parallel). Without this, the worker would hit a missing `download_jobs`
-    table and crash-loop until the API's migrations land. Poll patiently instead."""
+    table and crash-loop until the API's migrations land. Poll patiently instead — and if the schema
+    is still absent when the timeout runs out, say so and let the caller exit, because "continuing"
+    only moved the same crash one call further down (into `_recover_orphans`), where it read as an
+    unrelated failure."""
     deadline = time.monotonic() + timeout
     while not _stop.is_set():
         try:
             with SessionLocal() as db:
                 db.execute(text("SELECT 1 FROM download_jobs LIMIT 1"))
-            return
+            return True
         except Exception as exc:  # noqa: BLE001 — table not migrated yet / DB not up yet
             if time.monotonic() > deadline:
-                log.warning("download schema still absent after %ss; continuing: %s", timeout, str(exc)[:120])
+                log.error(
-                return
+                    "download schema still absent after %ss — the API's migrations have not landed; "
+                    "exiting so the container restarts: %s",
+                    timeout, str(exc)[:120],
+                )
+                return False
             log.info("waiting for the database schema (API migrations)…")
             _stop.wait(3.0)
+    return False
 
 
 def _recover_orphans() -> None:
@@ -719,8 +728,9 @@ def main() -> None:
         log.info("WORKER_ENABLED is off; worker exiting (nothing to do).")
         return
 
-    _wait_for_schema()  # the API may still be applying migrations when we boot
+    # The API may still be applying migrations when we boot. If they never arrive, exit instead of
-    if _stop.is_set():
+    # pressing on into a guaranteed crash — the container's restart policy retries the whole wait.
+    if not _wait_for_schema():
         return
     _recover_orphans()
     n = max(1, settings.download_worker_concurrency)

backend/app/youtube/rss.py

@@ -8,14 +8,24 @@ import httpx
 RSS_URL = "https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
 
 
+class RssError(Exception):
+    """The feed could not be read (transport failure or a non-200 answer).
+
+    Deliberately NOT the same as an empty feed: a failed poll used to return `[]`, which the
+    caller stamped as a successful `last_rss_at` — so a channel whose feed 404s permanently
+    (deleted, renamed id) looked like it was being polled fine forever, and nothing surfaced it."""
+
+
 def fetch_channel_feed(channel_id: str) -> list[dict]:
+    """The channel's recent uploads as stubs. `[]` means the feed really is empty; a failure
+    raises RssError so the caller can leave the channel un-stamped and log it."""
     url = RSS_URL.format(channel_id=channel_id)
     try:
         resp = httpx.get(url, timeout=20.0, headers={"User-Agent": "Siftlode/1.0"})
-    except httpx.HTTPError:
+    except httpx.HTTPError as exc:
-        return []
+        raise RssError(f"RSS fetch failed for {channel_id}: {exc}") from exc
     if resp.status_code != 200:
-        return []
+        raise RssError(f"RSS fetch for {channel_id} returned HTTP {resp.status_code}")
 
     parsed = feedparser.parse(resp.content)
     out: list[dict] = []

backend/tests/test_sync_guards.py

@@ -0,0 +1,28 @@
+"""Guards that stop one bad YouTube answer from cascading: the playlist-push circuit breaker."""
+from app.sync.playlists import _Breaker
+
+
+class TestBreaker:
+    def test_open_until_the_limit_is_reached(self):
+        b = _Breaker(limit=3)
+        for _ in range(2):
+            b.hit(False)
+            assert b.open()
+        b.hit(False)
+        assert not b.open() and b.tripped
+
+    def test_a_success_re_arms_the_streak(self):
+        b = _Breaker(limit=3)
+        b.hit(False)
+        b.hit(False)
+        b.hit(True)  # one write went through — this isn't an account-wide failure
+        b.hit(False)
+        b.hit(False)
+        assert b.open()
+
+    def test_stays_tripped_once_tripped(self):
+        b = _Breaker(limit=2)
+        b.hit(False)
+        b.hit(False)
+        b.hit(True)
+        assert not b.open()