fix(sync): read a dark RSS feed through the API instead

YouTube's RSS goes dark per channel for hours (prod 2026-09-18/19: 393 of
396 feeds answered 404 for four hours, sticky per channel, so a retry does
not help), and nothing else read a followed channel's new uploads. A
channel whose feed has not answered for `rss_fallback_hours` (admin knob,
default 2, 0 = off) now gets one page of its uploads playlist through the
Data API (1 unit), at most once per window, yielding to the backfill
reserve. Spend is attributed to its own quota action.

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 == []

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"]
 
 

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: