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.
This commit is contained in:
2026-09-19 05:47:16 +02:00
parent 582f20fab6
commit 1fa6a0bea8
3 changed files with 63 additions and 4 deletions
+27 -4
View File
@@ -11,7 +11,7 @@ from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta, timezone 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 sqlalchemy.orm import Session
from app import logs, progress, quota, sysconfig from app import logs, progress, quota, sysconfig
@@ -32,7 +32,7 @@ from app.sync.videos import (
run_shorts_classification, run_shorts_classification,
stub_from_playlist_item, 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 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 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 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 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 `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") hours = sysconfig.get_int(db, "rss_fallback_hours")
if hours <= 0 or not channel_ids: if hours <= 0 or not channel_ids:
return {"read": 0, "new": 0, "failed": 0} 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) data = yt.get_playlist_items(channel.uploads_playlist_id, None)
# silent-ok: counted, and the one summary line below names the count — the client # silent-ok: counted, and the one summary line below names the count — the client
# already logged the call's own failure (`youtube.api.error`). # already logged the call's own failure (`youtube.api.error`).
except YouTubeError: except YouTubeError as exc:
failed += 1 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 continue
rows = [ rows = [
s 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"), extra=logs.event("sync.rss.fallback_failed"),
) )
failed += 1 failed += 1
_stamp_fallback_attempt(db, channel.id)
continue continue
read += 1 read += 1
log.info( 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} 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: def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
user = get_service_user(db) user = get_service_user(db)
if user is None: if user is None:
+8
View File
@@ -93,6 +93,14 @@ def _upstream_failure(
# The `mine` reads' answer for a Google account that has no YouTube channel. # The `mine` reads' answer for a Google account that has no YouTube channel.
_NO_CHANNEL = frozenset({"channelNotFound"}) _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]: def _chunks(items: list, size: int) -> Iterator[list]:
for i in range(0, len(items), size): for i in range(0, len(items), size):
+28
View File
@@ -8,12 +8,16 @@ import pytest
from app import sysconfig from app import sysconfig
from app.models import Channel, Video from app.models import Channel, Video
from app.sync import runner from app.sync import runner
from app.youtube.client import YouTubeError
NOW = datetime.now(timezone.utc) NOW = datetime.now(timezone.utc)
_PUB = datetime(2020, 1, 1, tzinfo=timezone.utc) _PUB = datetime(2020, 1, 1, tzinfo=timezone.utc)
class _FakeYt: class _FakeYt:
# playlist id -> the reason its read fails with (the tests put entries here)
failing: dict[str, str] = {}
def __init__(self, calls): def __init__(self, calls):
self.calls = calls self.calls = calls
@@ -25,6 +29,8 @@ class _FakeYt:
def get_playlist_items(self, playlist_id, page_token=None): def get_playlist_items(self, playlist_id, page_token=None):
self.calls.append(playlist_id) self.calls.append(playlist_id)
if playlist_id in self.failing:
raise YouTubeError("refused", reason=self.failing[playlist_id])
cid = "UC" + playlist_id[2:] cid = "UC" + playlist_id[2:]
return { return {
"items": [ "items": [
@@ -42,6 +48,7 @@ def calls(monkeypatch):
monkeypatch.setattr(runner, "get_service_user", lambda db: object()) monkeypatch.setattr(runner, "get_service_user", lambda db: object())
monkeypatch.setattr(runner, "YouTubeClient", lambda db, user: _FakeYt(made)) monkeypatch.setattr(runner, "YouTubeClient", lambda db, user: _FakeYt(made))
monkeypatch.setattr(runner.quota, "backfill_reserve_reached", lambda: False) monkeypatch.setattr(runner.quota, "backfill_reserve_reached", lambda: False)
monkeypatch.setattr(_FakeYt, "failing", {})
return made 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 runner.run_feed_fallback(db, ["UCdark"])["read"] == 0
assert calls == [] 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"]