From 1fa6a0bea86a461243e2dd564f7c434055e43003 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Sat, 19 Sep 2026 05:47:16 +0200 Subject: [PATCH] fix(sync): a failed feed-fallback read waits its window out, and spent quota ends the round A channel whose uploads playlist also failed was never stamped, so a terminated channel was charged again on every 20-minute poll instead of once per rss_fallback_hours. An upstream quotaExceeded now stops the round instead of charging every remaining channel the same refusal. --- backend/app/sync/runner.py | 31 +++++++++++++++++++++++++---- backend/app/youtube/client.py | 8 ++++++++ backend/tests/test_feed_fallback.py | 28 ++++++++++++++++++++++++++ 3 files changed, 63 insertions(+), 4 deletions(-) diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 695977cc..ca971db4 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -11,7 +11,7 @@ from collections import Counter from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta, timezone -from sqlalchemy import func, or_, select +from sqlalchemy import func, or_, select, update from sqlalchemy.orm import Session from app import logs, progress, quota, sysconfig @@ -32,7 +32,7 @@ from app.sync.videos import ( run_shorts_classification, stub_from_playlist_item, ) -from app.youtube.client import YouTubeClient, YouTubeError +from app.youtube.client import YouTubeClient, YouTubeError, is_quota_refusal 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. @@ -197,7 +197,9 @@ def run_feed_fallback(db: Session, channel_ids: list[str]) -> dict[str, int]: uploads playlist is 1 unit, so a whole day of total outage costs 12 units per channel at the default 2 hours; it yields to the backfill reserve like every other background read. The clock is `last_rss_at` or `feed_fallback_at`, whichever is later, so a channel is read at most once per - window whether its feed came back in between or not.""" + window whether its feed came back in between or not. A FAILED read stamps the clock too: a + terminated channel's uploads playlist answers 404 for good, and without the stamp it would be + charged again on every 20-minute poll instead of once per window.""" hours = sysconfig.get_int(db, "rss_fallback_hours") if hours <= 0 or not channel_ids: return {"read": 0, "new": 0, "failed": 0} @@ -231,8 +233,12 @@ def run_feed_fallback(db: Session, channel_ids: list[str]) -> dict[str, int]: data = yt.get_playlist_items(channel.uploads_playlist_id, None) # silent-ok: counted, and the one summary line below names the count — the client # already logged the call's own failure (`youtube.api.error`). - except YouTubeError: + except YouTubeError as exc: failed += 1 + _stamp_fallback_attempt(db, channel.id) + # Out of quota upstream: every further channel would be charged the same refusal. + if is_quota_refusal(exc): + break continue rows = [ s @@ -252,6 +258,7 @@ def run_feed_fallback(db: Session, channel_ids: list[str]) -> dict[str, int]: extra=logs.event("sync.rss.fallback_failed"), ) failed += 1 + _stamp_fallback_attempt(db, channel.id) continue read += 1 log.info( @@ -264,6 +271,22 @@ def run_feed_fallback(db: Session, channel_ids: list[str]) -> dict[str, int]: return {"read": read, "new": new, "failed": failed} +def _stamp_fallback_attempt(db: Session, channel_id: str) -> None: + """Start a channel's fallback window after a read that did not land, so it waits the window out + like a read that did. Its own commit: the failure already rolled the channel's work back.""" + try: + db.execute( + update(Channel) + .where(Channel.id == channel_id) + .values(feed_fallback_at=datetime.now(timezone.utc)) + ) + db.commit() + # silent-ok: the failure that brought us here was already logged and counted; an unstamped + # channel is only read again at the next poll. + except Exception: # noqa: BLE001 + db.rollback() + + def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int: user = get_service_user(db) if user is None: diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py index 2e827c16..42187ca2 100644 --- a/backend/app/youtube/client.py +++ b/backend/app/youtube/client.py @@ -93,6 +93,14 @@ def _upstream_failure( # The `mine` reads' answer for a Google account that has no YouTube channel. _NO_CHANNEL = frozenset({"channelNotFound"}) +# The reasons that mean the project's quota is spent: every further call today gets the same answer. +_QUOTA_REASONS = frozenset({"quotaExceeded", "dailyLimitExceeded"}) + + +def is_quota_refusal(exc: YouTubeError) -> bool: + """Whether the API refused for want of quota. `reason` can list several codes, comma-joined.""" + return any(r in _QUOTA_REASONS for r in exc.reason.split(",")) + def _chunks(items: list, size: int) -> Iterator[list]: for i in range(0, len(items), size): diff --git a/backend/tests/test_feed_fallback.py b/backend/tests/test_feed_fallback.py index 2c5432dd..7dd16b8c 100644 --- a/backend/tests/test_feed_fallback.py +++ b/backend/tests/test_feed_fallback.py @@ -8,12 +8,16 @@ import pytest from app import sysconfig from app.models import Channel, Video from app.sync import runner +from app.youtube.client import YouTubeError NOW = datetime.now(timezone.utc) _PUB = datetime(2020, 1, 1, tzinfo=timezone.utc) class _FakeYt: + # playlist id -> the reason its read fails with (the tests put entries here) + failing: dict[str, str] = {} + def __init__(self, calls): self.calls = calls @@ -25,6 +29,8 @@ class _FakeYt: def get_playlist_items(self, playlist_id, page_token=None): self.calls.append(playlist_id) + if playlist_id in self.failing: + raise YouTubeError("refused", reason=self.failing[playlist_id]) cid = "UC" + playlist_id[2:] return { "items": [ @@ -42,6 +48,7 @@ def calls(monkeypatch): monkeypatch.setattr(runner, "get_service_user", lambda db: object()) monkeypatch.setattr(runner, "YouTubeClient", lambda db, user: _FakeYt(made)) monkeypatch.setattr(runner.quota, "backfill_reserve_reached", lambda: False) + monkeypatch.setattr(_FakeYt, "failing", {}) return made @@ -113,3 +120,24 @@ def test_it_yields_to_the_backfill_reserve(db, calls, monkeypatch): assert runner.run_feed_fallback(db, ["UCdark"])["read"] == 0 assert calls == [] + + +def test_a_read_that_fails_waits_the_window_out_too(db, calls): + # A terminated channel: its feed 404s and so does its uploads playlist, for good. Without a + # stamp it was charged on every 20-minute poll instead of once per window. + _channel(db, "UCgone", rss=NOW - timedelta(hours=3)) + _FakeYt.failing["UUgone"] = "playlistNotFound" + + assert runner.run_feed_fallback(db, ["UCgone"]) == {"read": 0, "new": 0, "failed": 1} + assert db.get(Channel, "UCgone").feed_fallback_at is not None + assert runner.run_feed_fallback(db, ["UCgone"])["failed"] == 0 + assert calls == ["UUgone"] + + +def test_out_of_quota_upstream_stops_the_round(db, calls): + _channel(db, "UCa") + _channel(db, "UCb") + _FakeYt.failing["UUa"] = "quotaExceeded" + + assert runner.run_feed_fallback(db, ["UCa", "UCb"])["failed"] == 1 + assert calls == ["UUa"]