diff --git a/backend/alembic/versions/0088_channels_feed_fallback_at.py b/backend/alembic/versions/0088_channels_feed_fallback_at.py new file mode 100644 index 00000000..8680606c --- /dev/null +++ b/backend/alembic/versions/0088_channels_feed_fallback_at.py @@ -0,0 +1,30 @@ +"""add channels.feed_fallback_at — when a channel's uploads were last read through the API because +its RSS feed would not answer + +Revision ID: 0088_channels_feed_fallback_at +Revises: 0087_users_diagnostics_level +Create Date: 2026-09-19 + +YouTube's RSS feeds go dark for hours at a time, per channel (2026-09-18/19: 393 of 396 answered 404 +for four hours; the same channel answers 404 to every retry). The RSS poll then reads the newest +uploads through the Data API instead, at a quota cost, and at most once per `rss_fallback_hours` per +channel: this column is that clock. NULL = never needed it. +""" + +import sqlalchemy as sa +from alembic import op + +revision: str = "0088_channels_feed_fallback_at" +down_revision: str | None = "0087_users_diagnostics_level" +branch_labels: str | None = None +depends_on: str | None = None + + +def upgrade() -> None: + op.add_column( + "channels", sa.Column("feed_fallback_at", sa.DateTime(timezone=True), nullable=True) + ) + + +def downgrade() -> None: + op.drop_column("channels", "feed_fallback_at") diff --git a/backend/app/config.py b/backend/app/config.py index 2b2ad824..2490b51a 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -119,6 +119,9 @@ class Settings(BaseSettings): # Recent-first backfill: how far back to fetch on the first pass per channel. backfill_recent_max_videos: int = 100 backfill_recent_max_days: int = 365 + # A channel whose RSS feed has not answered for this many hours gets its newest uploads read + # through the Data API instead (1 unit), at most once per this many hours. 0 = never. + rss_fallback_hours: int = 2 # Shorts are confirmed by probing youtube.com/shorts/. Only videos at or below # this duration (and not livestreams) are probed; longer videos are never Shorts. diff --git a/backend/app/models.py b/backend/app/models.py index 173d4b8a..b4273184 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -241,6 +241,9 @@ class Channel(Base, TimestampMixin): details_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) recent_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_rss_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + # When the newest uploads were last read through the Data API because the RSS feed would not + # answer (`sync/runner.py` `run_feed_fallback`): the once-per-`rss_fallback_hours` clock. + feed_fallback_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) backfill_cursor: Mapped[str | None] = mapped_column(String(128)) backfill_done: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # True when this channel only entered the catalog because a live YouTube search diff --git a/backend/app/quota.py b/backend/app/quota.py index e630a34c..c7d91b38 100644 --- a/backend/app/quota.py +++ b/backend/app/quota.py @@ -37,6 +37,7 @@ class QuotaAction: PLAYLISTS_DELETE = "playlists_delete" VIDEOS_BACKFILL_RECENT = "videos_backfill_recent" VIDEOS_BACKFILL_FULL = "videos_backfill_full" + VIDEOS_FEED_FALLBACK = "videos_feed_fallback" VIDEOS_ENRICH = "videos_enrich" VIDEOS_LOOKUP = "videos_lookup" VIDEOS_SEARCH = "videos_search" diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 7220620c..be7cef0b 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -9,8 +9,9 @@ import logging import threading from collections import Counter from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timedelta, timezone -from sqlalchemy import select +from sqlalchemy import func, or_, select from sqlalchemy.orm import Session from app import logs, progress, quota, sysconfig @@ -25,11 +26,13 @@ from app.sync.videos import ( backfill_channel_deep, backfill_channel_recent, enrich_pending, + insert_stubs, reconcile_full_history, refresh_live, run_shorts_classification, + stub_from_playlist_item, ) -from app.youtube.client import YouTubeClient +from app.youtube.client import YouTubeClient, YouTubeError 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. @@ -85,6 +88,9 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str # of 2026-09-16 into 11k warnings in four hours; each channel's own line is DEBUG now. reasons: Counter[str] = Counter() examples: list[str] = [] + # Channels whose FEED would not answer (not those whose rows failed to apply: the API would + # hand the same rows to the same broken write) — the feed fallback's candidates. + unread: list[str] = [] def fail(reason: str, detail: str, *, unexpected: str | None = None) -> None: """Count a channel that did not get polled. `unexpected` is the message for a failure that @@ -120,6 +126,7 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str except RssError as exc: log.debug("RSS poll skipped for channel %s: %s", channel.id, exc) fail(exc.reason, f"{channel.id}: {exc}") + unread.append(channel.id) # silent-ok: `fail(unexpected=…)` is the report — it logs at ERROR with the # traceback, once per reason per poll, which is the flood control this loop needs. except Exception as exc: # noqa: BLE001 — one channel must not stop the poll @@ -143,6 +150,7 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str # One place that advances the counter, whatever happened above. done += 1 progress.report(done, total, "rss_poll") + fallback = run_feed_fallback(db, unread) if failed: log.warning( "RSS poll: %d/%d channels could not be polled", @@ -164,8 +172,90 @@ def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> dict[str # subscriptions "all channels failed" is also what ONE dead feed id looks like, and a red # card the user can't act on is its own kind of dishonest. `channel: HTTP 404` says # "fix this subscription"; `Connection refused` says "the egress is down". - raise RuntimeError(f"RSS poll failed for all {total} channels — first: {first_error}") - return {"new": new, "failed": failed} + # The fallback's count rides along: a red card during an RSS outage should still say + # whether the uploads are arriving another way. + raise RuntimeError( + f"RSS poll failed for all {total} channels — first: {first_error}" + f" (read through the API instead: {fallback['read']})" + ) + return {"new": new + fallback["new"], "failed": failed, "fallback": fallback["read"]} + + +def run_feed_fallback(db: Session, channel_ids: list[str]) -> dict[str, int]: + """Read the newest uploads through the Data API for the channels among `channel_ids` whose feed + has not answered for `rss_fallback_hours` — and that were not read this way within as long. + + YouTube's RSS goes dark per channel for hours at a time (2026-09-18/19: 393 of 396 feeds answered + 404 for four hours, and a retry of the same channel gets the same 404), and nothing else reads a + followed channel's new uploads: recent backfill runs once per channel, ever. One page of 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 + `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.""" + hours = sysconfig.get_int(db, "rss_fallback_hours") + if hours <= 0 or not channel_ids: + return {"read": 0, "new": 0, "failed": 0} + now = datetime.now(timezone.utc) + last_read = func.greatest(Channel.last_rss_at, Channel.feed_fallback_at) + due = ( + db.execute( + select(Channel) + .where( + Channel.id.in_(channel_ids), + Channel.uploads_playlist_id.is_not(None), + or_(last_read.is_(None), last_read < now - timedelta(hours=hours)), + ) + .order_by(Channel.id) + ) + .scalars() + .all() + ) + user = get_service_user(db) if due else None + if user is None: + return {"read": 0, "new": 0, "failed": 0} + read = new = failed = 0 + with ( + quota.attribute(None, quota.QuotaAction.VIDEOS_FEED_FALLBACK), + YouTubeClient(db, user) as yt, + ): + for channel in due: + if quota.backfill_reserve_reached(): + break + try: + 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: + failed += 1 + continue + rows = [ + s + for s in (stub_from_playlist_item(it, channel.id) for it in data.get("items", [])) + if s is not None + ] + try: + new += insert_stubs(db, rows) + channel.feed_fallback_at = datetime.now(timezone.utc) + db.add(channel) + db.commit() + except Exception: # noqa: BLE001 — one channel must not stop the rest + db.rollback() + log.exception( + "Feed fallback could not store channel %s", + channel.id, + extra=logs.event("sync.rss.fallback_failed"), + ) + failed += 1 + continue + read += 1 + log.info( + "Feed fallback: read %d of %d channels through the API, %d new videos", + read, + len(due), + new, + extra=logs.event("sync.rss.fallback", read=read, due=len(due), new=new, failed=failed), + ) + return {"read": read, "new": new, "failed": failed} def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int: diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py index 8b2f49b3..41f1b7bd 100644 --- a/backend/app/sysconfig.py +++ b/backend/app/sysconfig.py @@ -91,6 +91,7 @@ SPECS: tuple[ConfigSpec, ...] = ( ConfigSpec( "backfill_recent_max_days", "int", "backfill", "backfill_recent_max_days", min=1, max=36_500 ), + ConfigSpec("rss_fallback_hours", "int", "backfill", "rss_fallback_hours", min=0, max=168), # --- Shorts probe --- ConfigSpec( "shorts_probe_max_seconds", "int", "shorts", "shorts_probe_max_seconds", min=1, max=3_600 diff --git a/backend/tests/test_feed_fallback.py b/backend/tests/test_feed_fallback.py new file mode 100644 index 00000000..2c5432dd --- /dev/null +++ b/backend/tests/test_feed_fallback.py @@ -0,0 +1,115 @@ +"""The RSS poll's API fallback (`sync/runner.run_feed_fallback`), DB-lane: which channels are read +through the Data API when their feed stays dark, and how often.""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from app import sysconfig +from app.models import Channel, Video +from app.sync import runner + +NOW = datetime.now(timezone.utc) +_PUB = datetime(2020, 1, 1, tzinfo=timezone.utc) + + +class _FakeYt: + def __init__(self, calls): + self.calls = calls + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def get_playlist_items(self, playlist_id, page_token=None): + self.calls.append(playlist_id) + cid = "UC" + playlist_id[2:] + return { + "items": [ + { + "contentDetails": {"videoId": f"{cid}-v1", "videoPublishedAt": NOW.isoformat()}, + "snippet": {"title": "new upload"}, + } + ] + } + + +@pytest.fixture +def calls(monkeypatch): + made: list[str] = [] + 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) + return made + + +def _channel(db, cid, *, rss=None, fallback=None, uploads=True): + ch = Channel( + id=cid, + title=cid, + published_at=_PUB, + uploads_playlist_id=("UU" + cid[2:]) if uploads else None, + last_rss_at=rss, + feed_fallback_at=fallback, + ) + db.add(ch) + db.commit() + return ch + + +def test_a_feed_dark_past_the_window_is_read_through_the_api(db, calls): + _channel(db, "UCdark", rss=NOW - timedelta(hours=3)) + + out = runner.run_feed_fallback(db, ["UCdark"]) + + assert out == {"read": 1, "new": 1, "failed": 0} + assert calls == ["UUdark"] + assert db.get(Video, "UCdark-v1") is not None + assert db.get(Channel, "UCdark").feed_fallback_at is not None + + +@pytest.mark.parametrize( + ("rss", "fallback"), + [ + (NOW - timedelta(minutes=30), None), # the feed answered recently: a blip, not an outage + (NOW - timedelta(hours=5), NOW - timedelta(minutes=30)), # already read this window + ], +) +def test_a_channel_read_within_the_window_is_left_alone(db, calls, rss, fallback): + _channel(db, "UCrecent", rss=rss, fallback=fallback) + + assert runner.run_feed_fallback(db, ["UCrecent"])["read"] == 0 + assert calls == [] + + +def test_the_window_is_the_admins_and_zero_turns_it_off(db, calls): + _channel(db, "UCdark", rss=NOW - timedelta(hours=3)) + + sysconfig.set_value(db, "rss_fallback_hours", 4) + db.commit() + assert runner.run_feed_fallback(db, ["UCdark"])["read"] == 0 + + sysconfig.set_value(db, "rss_fallback_hours", 0) + db.commit() + _channel(db, "UCnever") + assert runner.run_feed_fallback(db, ["UCdark", "UCnever"])["read"] == 0 + assert calls == [] + + +def test_only_the_named_channels_and_only_those_with_an_uploads_playlist(db, calls): + _channel(db, "UCdark") + _channel(db, "UCother") + _channel(db, "UCnolist", uploads=False) + + assert runner.run_feed_fallback(db, ["UCdark", "UCnolist"])["read"] == 1 + assert calls == ["UUdark"] + + +def test_it_yields_to_the_backfill_reserve(db, calls, monkeypatch): + _channel(db, "UCdark") + monkeypatch.setattr(runner.quota, "backfill_reserve_reached", lambda: True) + + assert runner.run_feed_fallback(db, ["UCdark"])["read"] == 0 + assert calls == [] diff --git a/backend/tests/test_followed_channels.py b/backend/tests/test_followed_channels.py index 798b972f..5b914695 100644 --- a/backend/tests/test_followed_channels.py +++ b/backend/tests/test_followed_channels.py @@ -27,7 +27,7 @@ def test_the_rss_poll_skips_channels_nobody_follows(db, make_user, monkeypatch): polled: list[str] = [] monkeypatch.setattr(runner, "fetch_channel_feed", lambda cid: polled.append(cid) or []) monkeypatch.setattr(runner, "apply_rss_feed", lambda db, ch, entries: 0) - assert runner.run_rss_poll(db) == {"new": 0, "failed": 0} + assert runner.run_rss_poll(db) == {"new": 0, "failed": 0, "fallback": 0} assert sorted(polled) == ["UC_explored", "UC_sub"] diff --git a/backend/tests/test_sync_guards.py b/backend/tests/test_sync_guards.py index b36ce352..a43dbafd 100644 --- a/backend/tests/test_sync_guards.py +++ b/backend/tests/test_sync_guards.py @@ -112,11 +112,15 @@ class TestRssPollResult: def _run(self, monkeypatch, *, fetch, apply_, channels=("UC1", "UC2")): monkeypatch.setattr(runner, "fetch_channel_feed", fetch) monkeypatch.setattr(runner, "apply_rss_feed", apply_) + # The API fallback has its own DB-lane tests (test_feed_fallback.py); here it reads nothing. + monkeypatch.setattr( + runner, "run_feed_fallback", lambda db, ids: {"read": 0, "new": 0, "failed": 0} + ) return runner.run_rss_poll(_StubDb(), [_StubChannel(c) for c in channels]) def test_a_healthy_run_reports_no_failures(self, monkeypatch): out = self._run(monkeypatch, fetch=lambda cid: [{"id": "v"}], apply_=lambda db, ch, e: 1) - assert out == {"new": 2, "failed": 0} + assert out == {"new": 2, "failed": 0, "fallback": 0} def test_unreadable_feeds_are_counted(self, monkeypatch): # One dead channel among healthy ones: counted, but the run itself is still a run. @@ -128,7 +132,7 @@ class TestRssPollResult: out = self._run( monkeypatch, fetch=fetch, apply_=lambda db, ch, e: 1, channels=("UC1", "UC_dead") ) - assert out == {"new": 1, "failed": 1} + assert out == {"new": 1, "failed": 1, "fallback": 0} def test_apply_failures_are_counted_too(self, monkeypatch): # A per-row write failure (a constraint clash on one channel) while the rest store fine. @@ -143,7 +147,9 @@ class TestRssPollResult: apply_=apply_, channels=("UC1", "UC_bad"), ) - assert out == {"new": 1, "failed": 1}, "an unstored channel must not read as ok" + assert out == {"new": 1, "failed": 1, "fallback": 0}, ( + "an unstored channel must not read as ok" + ) def test_many_failures_write_one_warning_with_counts_not_one_per_channel( self, monkeypatch, caplog @@ -163,7 +169,7 @@ class TestRssPollResult: caplog.set_level(logging.DEBUG, logger="siftlode.sync") out = self._run(monkeypatch, fetch=fetch, apply_=lambda db, ch, e: 1, channels=channels) - assert out == {"new": 5, "failed": 35} + assert out == {"new": 5, "failed": 35, "fallback": 0} warnings = [r for r in caplog.records if r.levelno >= logging.WARNING] assert len(warnings) == 1 assert warnings[0].event == "sync.rss.failed" @@ -195,7 +201,11 @@ class TestRssPollResult: def test_a_quiet_hour_is_distinguishable_from_an_outage(self, monkeypatch): quiet = self._run(monkeypatch, fetch=lambda cid: [], apply_=lambda db, ch, e: 0) - assert quiet == {"new": 0, "failed": 0} # same "new", different "failed" — that's the point + assert quiet == { + "new": 0, + "failed": 0, + "fallback": 0, + } # same "new", different "failed" — that's the point def test_a_total_outage_raises_instead_of_returning_ok(self, monkeypatch): # Every feed unreachable (the tinyproxy-lost-the-boot-race incident). Returning a result @@ -225,10 +235,38 @@ class TestRssPollResult: with pytest.raises(RuntimeError, match="read-only transaction"): self._run(monkeypatch, fetch=lambda cid: [{"id": "v"}], apply_=boom) + def test_only_the_unreadable_feeds_go_to_the_api_and_what_it_reads_counts(self, monkeypatch): + # A feed that answered 404 is the fallback's; a channel whose rows failed to STORE is not — + # the API would hand the same rows to the same broken write. + seen = [] + + def fetch(cid): + if cid == "UC_dark": + raise rss.RssError("404", "http_404") + return [{"id": "v"}] + + def apply_(db, ch, entries): + if ch.id == "UC_badrow": + raise RuntimeError("constraint") + return 1 + + def fallback(db, ids): + seen.extend(ids) + return {"read": 1, "new": 3, "failed": 0} + + monkeypatch.setattr(runner, "fetch_channel_feed", fetch) + monkeypatch.setattr(runner, "apply_rss_feed", apply_) + monkeypatch.setattr(runner, "run_feed_fallback", fallback) + out = runner.run_rss_poll( + _StubDb(), [_StubChannel(c) for c in ("UC_ok", "UC_dark", "UC_badrow")] + ) + assert seen == ["UC_dark"] + assert out == {"new": 4, "failed": 2, "fallback": 1} + def test_no_channels_at_all_is_not_an_outage(self, monkeypatch): # A fresh instance with nothing subscribed must not report an error every poll. out = self._run(monkeypatch, fetch=lambda cid: [], apply_=lambda db, ch, e: 0, channels=()) - assert out == {"new": 0, "failed": 0} + assert out == {"new": 0, "failed": 0, "fallback": 0} class TestRssErrorVsEmpty: diff --git a/frontend/src/i18n/locales/en/config.json b/frontend/src/i18n/locales/en/config.json index abf6d62c..586f0393 100644 --- a/frontend/src/i18n/locales/en/config.json +++ b/frontend/src/i18n/locales/en/config.json @@ -63,6 +63,10 @@ "label": "Recent backfill — max age (days)", "hint": "How far back the first pass reaches per channel." }, + "rss_fallback_hours": { + "label": "Feed fallback after (hours)", + "hint": "When a channel's RSS feed has not answered for this many hours, its newest uploads are read through the API instead (1 unit), at most once per this many hours. 0 turns it off." + }, "shorts_probe_max_seconds": { "label": "Shorts probe — max duration (s)", "hint": "Only videos at or below this length are probed as possible Shorts." diff --git a/frontend/src/i18n/locales/en/quotaActions.json b/frontend/src/i18n/locales/en/quotaActions.json index 2d8d98e0..7bec3cd4 100644 --- a/frontend/src/i18n/locales/en/quotaActions.json +++ b/frontend/src/i18n/locales/en/quotaActions.json @@ -6,6 +6,7 @@ "playlists_delete": "Delete playlist", "videos_backfill_recent": "Recent backfill", "videos_backfill_full": "Full-history backfill", + "videos_feed_fallback": "Feed fallback (RSS down)", "videos_enrich": "Video enrichment", "videos_lookup": "Video lookup", "videos_search": "Live search", diff --git a/frontend/src/i18n/locales/hu/config.json b/frontend/src/i18n/locales/hu/config.json index c32c69b4..320d9f3c 100644 --- a/frontend/src/i18n/locales/hu/config.json +++ b/frontend/src/i18n/locales/hu/config.json @@ -63,6 +63,10 @@ "label": "Friss letöltés — max kor (nap)", "hint": "Az első kör csatornánként meddig nyúl vissza." }, + "rss_fallback_hours": { + "label": "Feed-pótlás ennyi óra után", + "hint": "Ha egy csatorna RSS-feedje ennyi órája nem válaszol, a legújabb feltöltéseit az API-n keresztül olvassuk be (1 egység), csatornánként legfeljebb ennyi óránként egyszer. 0 = kikapcsolva." + }, "shorts_probe_max_seconds": { "label": "Shorts-vizsgálat — max hossz (mp)", "hint": "Csak az ennél nem hosszabb videókat vizsgáljuk lehetséges Shortsként." diff --git a/frontend/src/i18n/locales/hu/quotaActions.json b/frontend/src/i18n/locales/hu/quotaActions.json index bd688665..fb567bce 100644 --- a/frontend/src/i18n/locales/hu/quotaActions.json +++ b/frontend/src/i18n/locales/hu/quotaActions.json @@ -6,6 +6,7 @@ "playlists_delete": "Lejátszási lista törlése", "videos_backfill_recent": "Friss feltöltések betöltése", "videos_backfill_full": "Teljes előzmény betöltése", + "videos_feed_fallback": "Feed-pótlás (RSS kiesés)", "videos_enrich": "Videó-gazdagítás", "videos_lookup": "Videó-lekérdezés", "videos_search": "Élő keresés",