From 7a872a3619e1fdd6535063266d933b2bcff3f9ee Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 00:04:24 -0400 Subject: [PATCH] =?UTF-8?q?feat(patreon):=20dead-letter=20ledger=20for=20p?= =?UTF-8?q?ermanently-failing=20media=20=E2=80=94=20#705=20step=202=20(#7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A media that fails every walk (404'd CDN, deleted post, geo-blocked Mux, persistently-corrupt bytes) used to re-error forever and re-burn chunks. New `patreon_failed_media` table (alembic 0038, chains 0037) records per-media attempts; once attempts reach DEAD_LETTER_THRESHOLD (3) the ingester skips it on routine tick/backfill walks (tier-1.5, folded into the seen/skip predicate). Recovery BYPASSES it (the operator's "try everything again" re-attempts dead media). A clean download clears the row (recovered); errors/quarantines upsert-increment it. Surfaced as run_stats.dead_lettered_count. - New PatreonFailedMedia model + migration; ingester _dead_keys / _record_failures (on_conflict increment) / _clear_failures. - skip = seen | dead (empty in recovery); failures recorded post-fetch on short sessions (same pattern as the seen-ledger). Tests: a media erroring 3× is dead-lettered + skipped (no download attempt); recovery re-attempts a dead media and clears it on success; a clean download clears a sub-threshold failure. Co-Authored-By: Claude Opus 4.8 (1M context) --- alembic/versions/0038_patreon_failed_media.py | 58 +++++++++ backend/app/models/__init__.py | 2 + backend/app/models/patreon_failed_media.py | 45 +++++++ backend/app/services/patreon_ingester.py | 110 ++++++++++++++++-- tests/test_patreon_ingester.py | 106 ++++++++++++++++- 5 files changed, 311 insertions(+), 10 deletions(-) create mode 100644 alembic/versions/0038_patreon_failed_media.py create mode 100644 backend/app/models/patreon_failed_media.py diff --git a/alembic/versions/0038_patreon_failed_media.py b/alembic/versions/0038_patreon_failed_media.py new file mode 100644 index 0000000..e907ae1 --- /dev/null +++ b/alembic/versions/0038_patreon_failed_media.py @@ -0,0 +1,58 @@ +"""patreon_failed_media: per-source dead-letter ledger for failing Patreon media + +Revision ID: 0038 +Revises: 0037 +Create Date: 2026-06-06 + +Plan #705 (#7). Media that keeps failing to download/validate (404'd CDN, +deleted post, geo-blocked Mux, persistently-corrupt bytes) gets recorded here +with an attempt counter; once it crosses the dead-letter threshold the ingester +skips it on routine walks (recovery still re-attempts). A clean download clears +the row. UNIQUE (source_id, filehash) is the upsert key (same media key the +seen-ledger uses). +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0038" +down_revision: Union[str, None] = "0037" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "patreon_failed_media", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column( + "source_id", + sa.Integer, + sa.ForeignKey("source.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + sa.Column("filehash", sa.String(128), nullable=False), + sa.Column("attempts", sa.Integer, nullable=False, server_default="1"), + sa.Column("last_error", sa.Text, nullable=True), + sa.Column( + "first_failed_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column( + "last_failed_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.UniqueConstraint( + "source_id", "filehash", name="uq_patreon_failed_media_source_id" + ), + ) + + +def downgrade() -> None: + op.drop_table("patreon_failed_media") diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 1445788..b8854b1 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -14,6 +14,7 @@ from .import_settings import ImportSettings from .import_task import ImportTask from .library_audit_run import LibraryAuditRun from .ml_settings import MLSettings +from .patreon_failed_media import PatreonFailedMedia from .patreon_seen_media import PatreonSeenMedia from .post import Post from .post_attachment import PostAttachment @@ -34,6 +35,7 @@ __all__ = [ "BackupRun", "Source", "Credential", + "PatreonFailedMedia", "PatreonSeenMedia", "Post", "PostAttachment", diff --git a/backend/app/models/patreon_failed_media.py b/backend/app/models/patreon_failed_media.py new file mode 100644 index 0000000..79976ef --- /dev/null +++ b/backend/app/models/patreon_failed_media.py @@ -0,0 +1,45 @@ +"""PatreonFailedMedia — per-source dead-letter ledger of Patreon media that +keeps failing to download/validate. + +Plan #705 (#7). A media that fails every walk (404'd CDN URL, deleted post, +geo-blocked Mux stream, persistently-corrupt bytes) would otherwise re-error +forever and re-burn backfill chunks. After ``attempts`` reaches the dead-letter +threshold the ingester skips it on routine tick/backfill walks (recovery still +re-attempts it — the operator's "try everything again"). A later clean download +clears the row (the media recovered). + +`filehash` is the same per-media key the seen-ledger uses (32-hex CDN MD5, or a +``video:`` / ``post:filename`` synthesized key) — hence String(128). UNIQUE +(source_id, filehash) is the upsert key. +""" + +from datetime import datetime + +from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.types import DateTime + +from .base import Base + + +class PatreonFailedMedia(Base): + __tablename__ = "patreon_failed_media" + __table_args__ = ( + UniqueConstraint( + "source_id", "filehash", name="uq_patreon_failed_media_source_id" + ), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + source_id: Mapped[int] = mapped_column( + ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True + ) + filehash: Mapped[str] = mapped_column(String(128), nullable=False) + attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + first_failed_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + last_failed_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/backend/app/services/patreon_ingester.py b/backend/app/services/patreon_ingester.py index 5207ff6..206b555 100644 --- a/backend/app/services/patreon_ingester.py +++ b/backend/app/services/patreon_ingester.py @@ -44,10 +44,10 @@ import time from collections.abc import Callable from pathlib import Path -from sqlalchemy import select, text +from sqlalchemy import delete, func, select, text from sqlalchemy.dialects.postgresql import insert as pg_insert -from ..models import PatreonSeenMedia +from ..models import PatreonFailedMedia, PatreonSeenMedia from .gallery_dl import DownloadResult, ErrorType from .patreon_client import ( MediaItem, @@ -72,6 +72,14 @@ _TICK_SEEN_THRESHOLD = 20 # synthesized key so a pathologically long file_name can't overflow the column. _LEDGER_KEY_MAX = 128 +# plan #705 #7: after this many failed download/validate attempts a media is +# "dead-lettered" and skipped on routine tick/backfill walks (recovery still +# re-attempts it). Stops a permanently-broken media (404'd CDN, deleted post) +# re-erroring forever and re-burning chunks. +DEAD_LETTER_THRESHOLD = 3 +# last_error is Text but bound it so a giant traceback doesn't bloat the row. +_ERROR_MAX = 1000 + def _ledger_key(media: MediaItem) -> str: """Stable per-media identity for the cross-run seen-ledger. @@ -163,6 +171,7 @@ class PatreonIngester: downloaded = 0 errors = 0 quarantined = 0 + dead_lettered = 0 skipped_count = 0 posts_processed = 0 consecutive_seen = 0 @@ -203,6 +212,7 @@ class PatreonIngester: "warning_count": 0, "tier_gated_count": 0, "quarantined_count": quarantined, + "dead_lettered_count": dead_lettered, }, ) @@ -235,27 +245,37 @@ class PatreonIngester: continue keys = [_ledger_key(m) for m in media] + # Recovery bypasses BOTH the seen-ledger AND the dead-letter + # ledger (the operator's "try everything again"); routine walks + # skip seen + dead media (tier-1 + tier-1.5, plan #705 #7). + dead = set() if bypass_seen else self._dead_keys(source_id, keys) seen = ( set() if bypass_seen else self._seen_keys(source_id, keys) ) + skip = seen | dead - def _is_seen(m: MediaItem, _seen=seen) -> bool: - return _ledger_key(m) in _seen + def _is_skip(m: MediaItem, _skip=skip) -> bool: + return _ledger_key(m) in _skip outcomes = self.downloader.download_post( - post, media, artist_slug, is_seen=_is_seen + post, media, artist_slug, is_seen=_is_skip ) to_mark: list[tuple[str, str]] = [] + to_clear: list[str] = [] # recovered → drop any dead-letter row + to_fail: list[tuple[str, str, str]] = [] # (key, post_id, error) for media_item, outcome in zip(media, outcomes, strict=False): key = _ledger_key(media_item) + if key in dead: + dead_lettered += 1 # skipped because previously dead if outcome.status == "downloaded": downloaded += 1 if outcome.path is not None: written.append(str(outcome.path)) to_mark.append((key, media_item.post_id)) + to_clear.append(key) consecutive_seen = 0 elif outcome.status == "skipped_disk": # Already on disk (a prior run). Reconcile the ledger so a @@ -263,6 +283,7 @@ class PatreonIngester: # do NOT re-feed it to phase 3 — attach_in_place would see # the duplicate sha256 and unlink the on-disk copy. to_mark.append((key, media_item.post_id)) + to_clear.append(key) skipped_count += 1 consecutive_seen += 1 elif outcome.status == "skipped_seen": @@ -272,22 +293,31 @@ class PatreonIngester: # New content that failed validation (corrupt) — counted # distinctly so the run surfaces a real quarantined total. # Not marked seen (a later walk may re-fetch a fixed file); - # it IS new content, so it breaks the run-of-seen. + # it IS new content, so it breaks the run-of-seen. Counts + # toward the dead-letter ledger (plan #705 #7). quarantined += 1 if outcome.path is not None: quarantined_paths.append(str(outcome.path)) + to_fail.append((key, media_item.post_id, outcome.error or "quarantined")) consecutive_seen = 0 elif outcome.status == "error": errors += 1 + to_fail.append((key, media_item.post_id, outcome.error or "error")) # An error neither advances nor resets the run-of-seen. if mode == "tick" and consecutive_seen >= seen_threshold: early_out = True break - # Mark seen AFTER the network fetch, on its own short session. + # Persist ledger changes AFTER the network fetch, on short + # sessions: mark downloaded/on-disk seen, clear any dead-letter + # for recovered media, and record failures (plan #705 #7). if to_mark: self._mark_seen(source_id, to_mark) + if to_clear: + self._clear_failures(source_id, to_clear) + if to_fail: + self._record_failures(source_id, to_fail) if early_out: break @@ -302,10 +332,13 @@ class PatreonIngester: log_lines.append(f"{errors} media item(s) failed") if quarantined: log_lines.append(f"{quarantined} media item(s) quarantined (invalid)") + if dead_lettered: + log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)") log_lines.append( f"Patreon ingest ({mode}): {downloaded} downloaded, " f"{skipped_count} skipped, {quarantined} quarantined, " - f"{errors} error(s), {posts_processed} post(s)" + f"{dead_lettered} dead-lettered, {errors} error(s), " + f"{posts_processed} post(s)" + (", reached end" if reached_bottom else "") + (", time-boxed" if budget_hit else "") ) @@ -453,6 +486,67 @@ class PatreonIngester: session.execute(stmt) session.commit() + # -- dead-letter ledger (plan #705 #7) --------------------------------- + + def _dead_keys(self, source_id: int, keys: list[str]) -> set[str]: + """Which of `keys` have failed >= DEAD_LETTER_THRESHOLD times (dead). + One short SELECT; recovery never calls this (it re-attempts dead media).""" + if not keys: + return set() + with self.session_factory() as session: + rows = session.execute( + select(PatreonFailedMedia.filehash).where( + PatreonFailedMedia.source_id == source_id, + PatreonFailedMedia.filehash.in_(keys), + PatreonFailedMedia.attempts >= DEAD_LETTER_THRESHOLD, + ) + ).scalars().all() + return set(rows) + + def _record_failures( + self, source_id: int, items: list[tuple[str, str, str]] + ) -> None: + """Upsert-increment the dead-letter ledger for failed media. On conflict + bump `attempts` and refresh last_error/last_failed_at (UNIQUE backs the + upsert — no check-then-insert, [[scalar_one_or_none-duplicates]]). De-dup + the batch locally (one row per key, last error wins).""" + by_key: dict[str, str] = {} + for key, _post_id, err in items: + by_key[key] = (err or "")[:_ERROR_MAX] + if not by_key: + return + values = [ + {"source_id": source_id, "filehash": k, "attempts": 1, "last_error": e} + for k, e in by_key.items() + ] + with self.session_factory() as session: + stmt = pg_insert(PatreonFailedMedia).values(values) + stmt = stmt.on_conflict_do_update( + constraint="uq_patreon_failed_media_source_id", + set_={ + "attempts": PatreonFailedMedia.attempts + 1, + "last_error": stmt.excluded.last_error, + "last_failed_at": func.now(), + }, + ) + session.execute(stmt) + session.commit() + + def _clear_failures(self, source_id: int, keys: list[str]) -> None: + """Drop dead-letter rows for media that just downloaded cleanly — they + recovered. A no-op DELETE for keys that were never failing.""" + unique = list(dict.fromkeys(keys)) + if not unique: + return + with self.session_factory() as session: + session.execute( + delete(PatreonFailedMedia).where( + PatreonFailedMedia.source_id == source_id, + PatreonFailedMedia.filehash.in_(unique), + ) + ) + session.commit() + async def verify_patreon_credential( url: str, diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index e5a0015..0b3df18 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -11,7 +11,12 @@ import pytest from sqlalchemy import func, select from sqlalchemy.orm import sessionmaker -from backend.app.models import Artist, PatreonSeenMedia, Source +from backend.app.models import ( + Artist, + PatreonFailedMedia, + PatreonSeenMedia, + Source, +) from backend.app.services.gallery_dl import ErrorType from backend.app.services.patreon_client import ( MediaItem, @@ -66,10 +71,11 @@ class _FakeDownloader: `on_disk` report skipped_disk (tier-2); media in `quarantine` report quarantined; everything else downloads.""" - def __init__(self, tmp_path, on_disk=None, quarantine=None): + def __init__(self, tmp_path, on_disk=None, quarantine=None, error=None): self.tmp_path = tmp_path self.on_disk = set(on_disk or ()) self.quarantine = set(quarantine or ()) + self.error = set(error or ()) self.download_calls = 0 def download_post(self, post, media_items, artist_slug, *, is_seen): @@ -80,6 +86,9 @@ class _FakeDownloader: elif _ledger_key(m) in self.on_disk: p = self.tmp_path / f"{m.post_id}_{m.filename}" outcomes.append(MediaOutcome(media=m, status="skipped_disk", path=p, error=None)) + elif _ledger_key(m) in self.error: + self.download_calls += 1 + outcomes.append(MediaOutcome(media=m, status="error", path=None, error="boom")) elif _ledger_key(m) in self.quarantine: self.download_calls += 1 q = self.tmp_path / "_quarantine" / f"{m.post_id}_{m.filename}" @@ -370,6 +379,99 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path) assert _count_ledger(sync_engine, source_id) == 1 +# --- dead-letter ledger (plan #705 #7) ------------------------------------ + + +def _count_failed(sync_engine, source_id): + factory = sessionmaker(sync_engine, expire_on_commit=False) + with factory() as s: + return s.execute( + select(func.count(PatreonFailedMedia.id)).where( + PatreonFailedMedia.source_id == source_id + ) + ).scalar_one() + + +def _seed_failed(sync_engine, source_id, key, attempts): + factory = sessionmaker(sync_engine, expire_on_commit=False) + with factory() as s: + s.add(PatreonFailedMedia(source_id=source_id, filehash=key, attempts=attempts)) + s.commit() + + +@pytest.mark.asyncio +async def test_repeated_failures_dead_letter_then_skip(source_id, sync_engine, tmp_path): + """A media that errors DEAD_LETTER_THRESHOLD times is recorded; the next walk + skips it (no download attempt) and counts it dead-lettered.""" + from backend.app.services.patreon_ingester import DEAD_LETTER_THRESHOLD + + m1 = _media("p1", 1) + + def _run(): + ing = _ingester( + sync_engine, tmp_path, + _FakeClient([(None, [("p1", [m1])])]), + _FakeDownloader(tmp_path, error={_ledger_key(m1)}), + ) + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", + ) + return result, ing.downloader + + # Fail it up to the threshold — each run attempts the download and errors. + for _ in range(DEAD_LETTER_THRESHOLD): + result, dl = _run() + assert dl.download_calls == 1 + assert _count_failed(sync_engine, source_id) == 1 + + # Now attempts == threshold → dead: the next walk skips it without trying. + result, dl = _run() + assert dl.download_calls == 0 + assert result.run_stats["dead_lettered_count"] == 1 + + +@pytest.mark.asyncio +async def test_recovery_reattempts_dead_lettered_and_clears_on_success( + source_id, sync_engine, tmp_path, +): + """Recovery bypasses the dead-letter ledger (re-attempts dead media); a clean + download clears the row (it recovered).""" + from backend.app.services.patreon_ingester import DEAD_LETTER_THRESHOLD + + m1 = _media("p1", 1) + _seed_failed(sync_engine, source_id, _ledger_key(m1), DEAD_LETTER_THRESHOLD) + + downloader = _FakeDownloader(tmp_path) # succeeds + ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader) + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="recovery", + ) + assert downloader.download_calls == 1 # re-attempted despite being dead + assert result.files_downloaded == 1 + assert _count_failed(sync_engine, source_id) == 0 # cleared on success + + +@pytest.mark.asyncio +async def test_clean_download_clears_below_threshold_failure( + source_id, sync_engine, tmp_path, +): + """A media with a sub-threshold failure that now downloads cleanly (tick) has + its dead-letter row cleared.""" + m1 = _media("p1", 1) + _seed_failed(sync_engine, source_id, _ledger_key(m1), 1) # not yet dead + + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader) + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", + ) + assert result.files_downloaded == 1 + assert _count_failed(sync_engine, source_id) == 0 + + # --- ledger + drift -------------------------------------------------------