peter
b59fd38a9c
The log now has a stored, queryable copy for the diagnostics console: every WARNING and above plus the request line of every request, kept for log_retention_days (default 7, no "forever") and pruned hourly by the new log_gc job. Rows are built on the logging thread from the same redacted fields as the stdout line, queued without blocking and written in batches by one background thread on its own connection. A full queue or a failed write drops and reports instead of retrying. Flood control keeps 30 of any one message per minute plus a count of the rest; anonymous request lines are capped at 60 a minute per client, and the client address is never stored. Deleting an account deletes its stored events. Measured on prod first: a YouTube RSS outage wrote 11k warnings in four hours, one per channel per poll. The poll now writes one warning with per-reason counts and a few examples, and a failure on our side gets one traceback per poll instead of one per channel.
275 lines
11 KiB
Python
275 lines
11 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_)
|
|
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}
|
|
|
|
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}
|
|
|
|
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}, "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}
|
|
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} # 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_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}
|
|
|
|
|
|
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)
|
|
|
|
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"
|