peter
29094f5f51
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.
368 lines
15 KiB
Python
368 lines
15 KiB
Python
"""Guards that stop one bad YouTube answer from cascading: the subscription prune guard, the
|
|
playlist-push circuit breaker, and RSS telling a failure apart from an empty feed."""
|
|
|
|
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.sync import runner
|
|
from app.sync.playlists import FAILED, _Breaker
|
|
from app.sync.subscriptions import should_prune
|
|
from app.youtube import rss
|
|
from app.youtube.client import YouTubeError
|
|
|
|
|
|
class TestShouldPrune:
|
|
def test_a_normal_fetch_prunes(self):
|
|
assert should_prune(fetched_count=12, existing_count=15) is True
|
|
|
|
def test_an_empty_fetch_against_existing_rows_never_prunes(self):
|
|
# The data-loss case: one anomalous empty-but-200 response must not wipe 15 rows.
|
|
assert should_prune(fetched_count=0, existing_count=15) is False
|
|
|
|
def test_an_empty_fetch_with_nothing_local_is_fine(self):
|
|
# A brand-new account: nothing to lose, so the (no-op) prune may run.
|
|
assert should_prune(fetched_count=0, existing_count=0) is True
|
|
|
|
def test_a_single_remaining_subscription_still_prunes(self):
|
|
assert should_prune(fetched_count=1, existing_count=200) is True
|
|
|
|
|
|
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()
|
|
|
|
|
|
def _quota_dead():
|
|
"""A write op that fails the way a quota-exhausted / scope-revoked account fails."""
|
|
raise YouTubeError("quota exceeded")
|
|
|
|
|
|
class TestBreakerRun:
|
|
def test_returns_the_ops_result_and_leaves_failures_alone(self):
|
|
b, failed = _Breaker(), []
|
|
assert b.run(lambda: "item-1", lambda: failed.append("v1")) == "item-1"
|
|
assert failed == []
|
|
|
|
def test_a_youtube_error_records_the_failure_and_counts_towards_the_trip(self):
|
|
b, failed = _Breaker(limit=2), []
|
|
assert b.run(_quota_dead, lambda: failed.append("v1")) is FAILED
|
|
assert failed == ["v1"] and b.open()
|
|
b.run(_quota_dead, lambda: failed.append("v2"))
|
|
assert not b.open()
|
|
|
|
def test_a_tripped_breaker_never_calls_the_op(self):
|
|
b, failed, calls = _Breaker(limit=1), [], []
|
|
b.run(_quota_dead, lambda: failed.append("v1"))
|
|
assert b.run(lambda: calls.append("called"), lambda: failed.append("v2")) is FAILED
|
|
assert calls == [] # the whole point: no more quota spent
|
|
assert failed == ["v1", "v2"] # …but the item is still reported
|
|
|
|
def test_other_exceptions_are_not_swallowed(self):
|
|
# Only YouTubeError is a "write failed" signal; a bug must still surface.
|
|
def boom():
|
|
raise ValueError("bug")
|
|
|
|
with pytest.raises(ValueError):
|
|
_Breaker().run(boom, lambda: None)
|
|
|
|
|
|
class _StubDb:
|
|
"""Just enough Session for run_rss_poll when the caller supplies the channel list."""
|
|
|
|
def __init__(self):
|
|
self.rollbacks = 0
|
|
|
|
def rollback(self):
|
|
self.rollbacks += 1
|
|
|
|
|
|
class _StubChannel:
|
|
def __init__(self, cid):
|
|
self.id = cid
|
|
|
|
|
|
class TestRssPollResult:
|
|
"""`failed` must count every channel that did not get polled — feed-side AND apply-side.
|
|
Counting only the fetch failures would move the "a broken run looks like a quiet hour" blind
|
|
spot rather than remove it."""
|
|
|
|
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, "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.
|
|
def fetch(cid):
|
|
if cid == "UC_dead":
|
|
raise rss.RssError("404", "http_404")
|
|
return [{"id": "v"}]
|
|
|
|
out = self._run(
|
|
monkeypatch, fetch=fetch, apply_=lambda db, ch, e: 1, channels=("UC1", "UC_dead")
|
|
)
|
|
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.
|
|
def apply_(db, ch, entries):
|
|
if ch.id == "UC_bad":
|
|
raise RuntimeError("constraint")
|
|
return 1
|
|
|
|
out = self._run(
|
|
monkeypatch,
|
|
fetch=lambda cid: [{"id": "v"}],
|
|
apply_=apply_,
|
|
channels=("UC1", "UC_bad"),
|
|
)
|
|
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
|
|
):
|
|
# YouTube's RSS outage of 2026-09-16: ~900 of 979 feeds failed on every 20-minute poll and
|
|
# each wrote its own WARNING — 11k lines in four hours, which the log_events table would
|
|
# have stored too. One summary per poll, per-channel detail at DEBUG.
|
|
channels = [f"UC{i}" for i in range(40)]
|
|
|
|
def fetch(cid):
|
|
n = int(cid[2:])
|
|
if n < 30:
|
|
raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404", "http_404")
|
|
if n < 35:
|
|
raise rss.RssError(f"RSS fetch failed for {cid}: timed out", "transport")
|
|
return [{"id": "v"}]
|
|
|
|
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, "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"
|
|
assert warnings[0].attrs["reasons"] == {"http_404": 30, "transport": 5}
|
|
assert len(warnings[0].attrs["examples"]) == runner.RSS_FAILURE_EXAMPLES
|
|
assert sum(r.levelno == logging.DEBUG for r in caplog.records) == 35
|
|
|
|
def test_a_failure_on_our_side_gets_one_traceback_per_poll_not_one_per_channel(
|
|
self, monkeypatch, caplog
|
|
):
|
|
# A read-only database fails every channel's apply identically. The first gets a traceback
|
|
# at ERROR; repeating it for each of 979 channels would be the RSS flood all over again.
|
|
def apply_(db, ch, entries):
|
|
raise RuntimeError("read-only transaction")
|
|
|
|
caplog.set_level(logging.DEBUG, logger="siftlode.sync")
|
|
with pytest.raises(RuntimeError):
|
|
self._run(
|
|
monkeypatch,
|
|
fetch=lambda cid: [{"id": "v"}],
|
|
apply_=apply_,
|
|
channels=("A", "B", "C"),
|
|
)
|
|
errors = [r for r in caplog.records if r.levelno == logging.ERROR]
|
|
assert len(errors) == 1 and errors[0].exc_info is not None
|
|
assert (
|
|
sum(r.levelno == logging.DEBUG and r.exc_info is not None for r in caplog.records) == 2
|
|
)
|
|
|
|
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,
|
|
"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
|
|
# here recorded the scheduler run as "ok", with the outage only visible inside the summary
|
|
# text — a green pill for a run that did nothing at all.
|
|
def boom(cid):
|
|
raise rss.RssError("Connection refused", "transport")
|
|
|
|
with pytest.raises(RuntimeError, match="all 2 channels"):
|
|
self._run(monkeypatch, fetch=boom, apply_=lambda db, ch, e: 1)
|
|
|
|
def test_the_outage_message_carries_the_underlying_error(self, monkeypatch):
|
|
# "all N channels failed" is ALSO what one dead feed id looks like on a small instance.
|
|
# The card has to say which: `HTTP 404` = fix the subscription, `Connection refused` =
|
|
# the egress is down.
|
|
def dead(cid):
|
|
raise rss.RssError(f"RSS fetch for {cid} returned HTTP 404", "http_404")
|
|
|
|
with pytest.raises(RuntimeError, match="HTTP 404"):
|
|
self._run(monkeypatch, fetch=dead, apply_=lambda db, ch, e: 1, channels=("UC_dead",))
|
|
|
|
def test_a_read_only_database_is_a_total_outage_too(self, monkeypatch):
|
|
# Every feed reads fine, nothing can be stored: just as un-polled, just as red.
|
|
def boom(db, ch, entries):
|
|
raise RuntimeError("read-only transaction")
|
|
|
|
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, "fallback": 0}
|
|
|
|
|
|
class TestRssErrorVsEmpty:
|
|
"""A failed poll must RAISE (so the caller leaves `last_rss_at` alone); only a genuinely
|
|
empty feed may return []."""
|
|
|
|
def _patch(self, monkeypatch, **kwargs):
|
|
def fake_get(url, **_):
|
|
if "exc" in kwargs:
|
|
raise kwargs["exc"]
|
|
return httpx.Response(
|
|
kwargs["status"],
|
|
content=kwargs.get("body", b""),
|
|
request=httpx.Request("GET", url),
|
|
)
|
|
|
|
monkeypatch.setattr(rss.httpx, "get", fake_get)
|
|
monkeypatch.setattr(rss.time, "sleep", lambda s: None)
|
|
|
|
def _answers(self, monkeypatch, *answers):
|
|
"""The endpoint answers each try with the next of `answers` (a status or an exception);
|
|
returns the list of tries made and the waits between them."""
|
|
tries, waits = [], []
|
|
queue = list(answers)
|
|
|
|
def fake_get(url, **_):
|
|
answer = queue.pop(0)
|
|
tries.append(answer)
|
|
if isinstance(answer, Exception):
|
|
raise answer
|
|
body = self.FEED if answer == 200 else b""
|
|
return httpx.Response(answer, content=body, request=httpx.Request("GET", url))
|
|
|
|
monkeypatch.setattr(rss.httpx, "get", fake_get)
|
|
monkeypatch.setattr(rss.time, "sleep", waits.append)
|
|
return tries, waits
|
|
|
|
FEED = b"""<?xml version="1.0"?>
|
|
<feed xmlns="http://www.w3.org/2005/Atom" xmlns:yt="http://www.youtube.com/xml/schemas/2015">
|
|
<entry><yt:videoId>abc123</yt:videoId><title>Hello</title></entry>
|
|
</feed>"""
|
|
|
|
def test_a_feed_that_fails_and_then_answers_is_read(self, monkeypatch):
|
|
# The endpoint's measured behaviour: the same channel 404s, 500s, then answers.
|
|
tries, waits = self._answers(monkeypatch, 404, 500, 200)
|
|
assert [e["id"] for e in rss.fetch_channel_feed("UC_flaky")] == ["abc123"]
|
|
assert (len(tries), waits) == (3, list(rss.RETRY_DELAYS))
|
|
|
|
def test_a_dropped_connection_is_tried_again_too(self, monkeypatch):
|
|
tries, _ = self._answers(monkeypatch, httpx.ConnectError("reset"), 200)
|
|
assert rss.fetch_channel_feed("UC_blip")
|
|
assert len(tries) == 2
|
|
|
|
def test_a_timeout_is_not_tried_again(self, monkeypatch):
|
|
# A hung request already cost the full timeout; retrying it would triple a poll over a
|
|
# dead egress.
|
|
tries, waits = self._answers(monkeypatch, httpx.ReadTimeout("hung"), 200)
|
|
with pytest.raises(rss.RssError) as err:
|
|
rss.fetch_channel_feed("UC_hung")
|
|
assert (len(tries), waits, err.value.reason) == (1, [], "transport")
|
|
|
|
def test_a_feed_that_keeps_failing_raises_its_last_failure(self, monkeypatch):
|
|
tries, _ = self._answers(monkeypatch, 500, 404, 404)
|
|
with pytest.raises(rss.RssError) as err:
|
|
rss.fetch_channel_feed("UC_gone")
|
|
assert (len(tries), err.value.reason) == (3, "http_404")
|
|
|
|
def test_a_refusal_is_not_tried_again(self, monkeypatch):
|
|
tries, waits = self._answers(monkeypatch, 403)
|
|
with pytest.raises(rss.RssError) as err:
|
|
rss.fetch_channel_feed("UC_forbidden")
|
|
assert (len(tries), waits, err.value.reason) == (1, [], "http_403")
|
|
|
|
def test_non_200_raises(self, monkeypatch):
|
|
self._patch(monkeypatch, status=404)
|
|
with pytest.raises(rss.RssError):
|
|
rss.fetch_channel_feed("UC_dead")
|
|
|
|
def test_transport_error_raises(self, monkeypatch):
|
|
self._patch(monkeypatch, exc=httpx.ConnectError("no route"))
|
|
with pytest.raises(rss.RssError):
|
|
rss.fetch_channel_feed("UC_offline")
|
|
|
|
def test_an_empty_but_valid_feed_returns_empty(self, monkeypatch):
|
|
feed = b"""<?xml version="1.0"?><feed xmlns="http://www.w3.org/2005/Atom"></feed>"""
|
|
self._patch(monkeypatch, status=200, body=feed)
|
|
assert rss.fetch_channel_feed("UC_quiet") == []
|
|
|
|
def test_entries_are_parsed(self, monkeypatch):
|
|
feed = b"""<?xml version="1.0"?>
|
|
<feed xmlns="http://www.w3.org/2005/Atom" xmlns:yt="http://www.youtube.com/xml/schemas/2015">
|
|
<entry><yt:videoId>abc123</yt:videoId><title>Hello</title>
|
|
<published>2026-07-23T10:00:00+00:00</published></entry>
|
|
</feed>"""
|
|
self._patch(monkeypatch, status=200, body=feed)
|
|
out = rss.fetch_channel_feed("UC_live")
|
|
assert [e["id"] for e in out] == ["abc123"]
|
|
assert out[0]["channel_id"] == "UC_live"
|