feat(patreon): dead-letter ledger for permanently-failing media — #705 step 2 (#7)

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) <noreply@anthropic.com>

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

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

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()
+    )

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,

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