fix(ingester): close #5 within-chunk live posts + #8 video transient retry

Review of the #1–#9 ingester roadmap found two real-but-small gaps; this closes
both.

#5 (live posts progress) shipped at per-chunk granularity — _apply_backfill_
lifecycle accumulated DownloadResult.posts_processed AFTER each chunk, so the
badge didn't move during a chunk (up to ~14.5 min) and over-counted the
re-walked resume page. The plan called for within-chunk live updates. Move
ownership of _backfill_posts into the ingester: ingest_core writes a monotonic
absolute (posts_base + net-new) via _checkpoint_posts at each page boundary and
once at the end, EXCLUDING the resumed page so it no longer inflates across
chunks. download_service seeds posts_base from prior chunks and stops touching
the key (the lifecycle now carries the ingester's committed value forward).

#8 (per-media transient/permanent retry) covered only the plain-GET path
(_fetch_to_file); the Mux/video path returned None on any yt-dlp failure with no
retry. Give _run_ytdlp the same split: TimeoutExpired/OSError are transient
(back off + retry up to _MAX_MEDIA_RETRIES), a non-zero exit (CalledProcessError)
is permanent (yt-dlp already did its own network retries) → fail fast to the
per-item/dead-letter path.

Tests: live-posts absolute + resume-page exclusion + tick-doesn't-persist
(test_patreon_ingester); lifecycle-leaves-posts-to-ingester rewrite
(test_download_service); video transient-retry + permanent-fail-fast
(test_patreon_downloader).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/services/download_service.py

@@ -238,6 +238,9 @@ class DownloadService:
                 mode=mode,
                 resume_cursor=source_config.resume_cursor,
                 time_budget_seconds=source_config.timeout,
+                # plan #704 #5: seed the live posts badge from prior chunks so the
+                # ingester writes a monotonic absolute mid-walk (it owns the key).
+                posts_base=int(overrides.get("_backfill_posts", 0)),
             ),
         )
         return dl_result, resolved_campaign_id
@@ -515,13 +518,12 @@ class DownloadService:
         new_overrides = dict(src.config_overrides or {})
         chunks = int(new_overrides.get("_backfill_chunks", 0)) + 1
         new_overrides["_backfill_chunks"] = chunks
-        # plan #704 (#5): accumulate posts-processed across chunks so the badge
-        # shows live walk progress, not just the chunk count. Cleared on
-        # (re)start. (Slight over-count: each chunk re-walks the resumed page —
-        # fine for a progress indicator.)
-        new_overrides["_backfill_posts"] = int(
-            new_overrides.get("_backfill_posts", 0)
-        ) + int(getattr(dl_result, "posts_processed", 0) or 0)
+        # plan #704 (#5): _backfill_posts (the live progress badge) is OWNED by the
+        # ingester now — it writes a monotonic absolute mid-walk at each page
+        # boundary (ingest_core._checkpoint_posts), so the badge climbs DURING a
+        # chunk instead of jumping once per chunk here, and the re-walked resume
+        # page is no longer double-counted. new_overrides (read fresh above)
+        # carries the ingester's committed value forward untouched.
 
         completed = (
             dl_result.success

backend/app/services/ingest_core.py

@@ -91,6 +91,7 @@ class Ingester:
         resume_cursor: str | None = None,
         time_budget_seconds: float = 870.0,
         seen_threshold: int = _TICK_SEEN_THRESHOLD,
+        posts_base: int = 0,
     ) -> DownloadResult:
         """Walk + download for one source, returning a gallery-dl-shaped result.
 
@@ -117,6 +118,11 @@ class Ingester:
         dead_lettered = 0
         skipped_count = 0
         posts_processed = 0
+        # Net-new posts THIS chunk for the live progress badge (plan #704 #5);
+        # excludes the re-walked resume page so _backfill_posts stays a monotonic
+        # absolute across chunks instead of an inflating sum. posts_processed
+        # stays the gross per-chunk count used for the run summary.
+        chunk_new_posts = 0
         consecutive_seen = 0
         emitted_cursor: str | None = None
         reached_bottom = False
@@ -171,9 +177,12 @@ class Ingester:
                     # plan #705 #6: persist the cursor at each page boundary so a
                     # worker SIGKILL mid-chunk resumes near the crash, not the
                     # chunk start. (phase 3 still writes the final cursor — same
-                    # value; this is the crash-safety net.)
+                    # value; this is the crash-safety net.) plan #704 #5: persist
+                    # the live posts count alongside it so the badge climbs DURING
+                    # the chunk, not only when it ends.
                     if checkpoint:
                         self._checkpoint_cursor(source_id, emitted_cursor)
+                        self._checkpoint_posts(source_id, posts_base + chunk_new_posts)
 
                 # Time-box check at the post boundary (coarse, like a gallery-dl
                 # chunk). Backfill/recovery resume from emitted_cursor next chunk.
@@ -182,6 +191,12 @@ class Ingester:
                     break
 
                 posts_processed += 1
+                # The resume page (its cursor == resume_cursor) was already
+                # counted by the chunk that checkpointed it — don't re-count it
+                # into the persisted badge (plan #704 #5). First chunk has
+                # resume_cursor None, so everything counts.
+                if not (resume_cursor and page_cursor == resume_cursor):
+                    chunk_new_posts += 1
                 media = self.client.extract_media(post, included)
                 if not media:
                     continue
@@ -270,6 +285,11 @@ class Ingester:
             # maps it to a typed error.
             return self._failure_result(exc, _result)
 
+        # Final authoritative posts count for the badge — captures the last page
+        # after the last boundary write and the time-box break (plan #704 #5).
+        if checkpoint:
+            self._checkpoint_posts(source_id, posts_base + chunk_new_posts)
+
         if errors:
             log_lines.append(f"{errors} media item(s) failed")
         if quarantined:
@@ -371,6 +391,27 @@ class Ingester:
             )
             session.commit()
 
+    def _checkpoint_posts(self, source_id: int, posts: int) -> None:
+        """Persist the live backfill posts-processed count mid-walk (plan #704 #5).
+
+        Same atomic single-key jsonb_set dance as _checkpoint_cursor, on the
+        `_backfill_posts` key (cast to a JSON number) — so the progress badge
+        climbs DURING a chunk without clobbering operator config or the cursor.
+        The ingester OWNS this key now; download_service no longer accumulates it
+        post-chunk (which lagged a whole chunk and over-counted the resume page).
+        """
+        with self.session_factory() as session:
+            session.execute(
+                text(
+                    "UPDATE source SET config_overrides = jsonb_set("
+                    "  coalesce(config_overrides::jsonb, '{}'::jsonb),"
+                    "  '{_backfill_posts}', to_jsonb(cast(:posts AS int))"
+                    ")::json WHERE id = :sid"
+                ),
+                {"posts": posts, "sid": source_id},
+            )
+            session.commit()
+
     def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None:
         """Idempotent upsert of (filehash, post_id) seen-ledger rows for a page.
 

backend/app/services/patreon_downloader.py

@@ -353,6 +353,14 @@ class PatreonDownloader:
         The output template uses `dest` without its extension; yt-dlp appends
         the chosen container extension. We pass Referer/Origin (Mux JWT policy)
         and the cookies file.
+
+        Mirrors the GET path's transient/permanent split (plan #705 #8): a hung
+        fetch (TimeoutExpired) or a spawn failure (OSError) is TRANSIENT — back
+        off and retry. A non-zero yt-dlp exit (CalledProcessError) is treated as
+        PERMANENT for this pass — yt-dlp already does its OWN internal network
+        retries, so a non-zero exit is effectively a real failure (private/gone/
+        geo-blocked), like a 4xx on the GET path: fail fast to the per-item error
+        → dead-letter path.
         """
         dest = Path(dest)
         out_template = str(dest.with_suffix("")) + ".%(ext)s"
@@ -362,17 +370,39 @@ class PatreonDownloader:
         if self.cookies_path and os.path.isfile(self.cookies_path):
             cmd += ["--cookies", self.cookies_path]
         cmd.append(url)
-        try:
-            subprocess.run(
-                cmd,
-                check=True,
-                capture_output=True,
-                text=True,
-                timeout=_TIMEOUT_SECONDS,
-            )
-        except (OSError, subprocess.SubprocessError) as exc:
-            log.warning("yt-dlp failed for %s: %s", url, exc)
-            return None
+        attempt = 0
+        while True:
+            try:
+                subprocess.run(
+                    cmd,
+                    check=True,
+                    capture_output=True,
+                    text=True,
+                    timeout=_TIMEOUT_SECONDS,
+                )
+                break
+            except subprocess.CalledProcessError as exc:
+                # Permanent for this pass — fail fast (no retry).
+                log.warning(
+                    "yt-dlp failed (exit %s) for %s: %s",
+                    exc.returncode, url, (exc.stderr or "").strip() or exc,
+                )
+                return None
+            except (OSError, subprocess.TimeoutExpired) as exc:
+                # Transient — back off and retry, like a transport blip on a GET.
+                if attempt >= _MAX_MEDIA_RETRIES:
+                    log.warning(
+                        "yt-dlp transient failure exhausted for %s: %s", url, exc
+                    )
+                    return None
+                attempt += 1
+                delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS)
+                log.warning(
+                    "yt-dlp transient failure (%s) — backing off %.1fs "
+                    "(retry %d/%d): %s",
+                    url, delay, attempt, _MAX_MEDIA_RETRIES, exc,
+                )
+                time.sleep(delay)
         return self._existing_video_output(dest)
 
     def _existing_video_output(self, dest: Path) -> Path | None:

tests/test_download_service.py

@@ -649,17 +649,18 @@ async def test_backfill_stuck_guard_stalls_after_two_no_advance(
 
 
 @pytest.mark.asyncio
-async def test_backfill_accumulates_posts_processed(
+async def test_backfill_lifecycle_leaves_posts_to_ingester(
     db, db_sync, tmp_path, seed_artist_and_source,
 ):
-    """plan #704 (#5): the lifecycle accumulates DownloadResult.posts_processed
-    into _backfill_posts across chunks for the live progress badge."""
+    """plan #704 (#5), revised: _backfill_posts is OWNED by the ingester now (it
+    writes a monotonic absolute live mid-walk), so the post-chunk lifecycle must
+    NOT touch it — it carries the ingester's committed value forward unchanged."""
     from unittest.mock import MagicMock
 
     from backend.app.services.download_service import DownloadService
 
     _artist, source = seed_artist_and_source
-    source.config_overrides = {"_backfill_state": "running", "_backfill_posts": 5}
+    source.config_overrides = {"_backfill_state": "running", "_backfill_posts": 12}
     source.backfill_runs_remaining = 5
     await db.commit()
 
@@ -669,7 +670,7 @@ async def test_backfill_accumulates_posts_processed(
     )
     ctx = {
         "source_id": source.id, "platform": "patreon",
-        "config_overrides": {"_backfill_state": "running", "_backfill_posts": 5},
+        "config_overrides": {"_backfill_state": "running", "_backfill_posts": 12},
         "backfill_runs_remaining": 5,
     }
     await svc._apply_backfill_lifecycle(
@@ -679,7 +680,7 @@ async def test_backfill_accumulates_posts_processed(
     co = (await db.execute(
         select(Source.config_overrides).where(Source.id == source.id)
     )).scalar_one()
-    assert co.get("_backfill_posts") == 15  # 5 (prior) + 10 (this chunk)
+    assert co.get("_backfill_posts") == 12  # untouched — the ingester owns it
 
 
 @pytest.mark.asyncio

tests/test_patreon_downloader.py

@@ -216,6 +216,57 @@ def test_video_routes_to_ytdlp(tmp_path, monkeypatch):
     assert find_sidecar(outcomes[0].path) is not None
 
 
+def _video_item():
+    return MediaItem(
+        url="https://stream.mux.com/abc123.m3u8", filename="clip.mp4",
+        kind="postfile", filehash=None, post_id="1001",
+    )
+
+
+def test_video_transient_failure_retried_then_succeeds(tmp_path, monkeypatch):
+    """plan #705 #8: a TRANSIENT yt-dlp failure (TimeoutExpired) is retried within
+    the pass; a later success yields a downloaded outcome. (The GET path already
+    retried transients; the video path didn't until #8 closed the gap.)"""
+    calls = {"n": 0}
+
+    def fake_run(cmd, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise pd_mod.subprocess.TimeoutExpired(cmd, kwargs.get("timeout"))
+        out = Path(cmd[cmd.index("-o") + 1].replace(".%(ext)s", ".mp4"))
+        out.parent.mkdir(parents=True, exist_ok=True)
+        out.write_bytes(b"video-bytes")
+        return pd_mod.subprocess.CompletedProcess(cmd, 0, "", "")
+
+    monkeypatch.setattr(pd_mod.subprocess, "run", fake_run)
+    monkeypatch.setattr(pd_mod.time, "sleep", lambda *_a, **_k: None)
+
+    dl = _downloader(tmp_path, session=_FakeSession())
+    outcomes = dl.download_post(_post(), [_video_item()], "artist-x")
+    assert calls["n"] == 2  # retried once after the transient blip
+    assert outcomes[0].status == "downloaded"
+    assert outcomes[0].path.read_bytes() == b"video-bytes"
+
+
+def test_video_permanent_failure_fails_fast(tmp_path, monkeypatch):
+    """plan #705 #8: a non-zero yt-dlp exit (CalledProcessError) is PERMANENT —
+    yt-dlp already did its own network retries — so we don't re-spawn; the item
+    errors straight to the per-item/dead-letter path."""
+    calls = {"n": 0}
+
+    def fake_run(cmd, **kwargs):
+        calls["n"] += 1
+        raise pd_mod.subprocess.CalledProcessError(1, cmd, stderr="ERROR: gone")
+
+    monkeypatch.setattr(pd_mod.subprocess, "run", fake_run)
+    monkeypatch.setattr(pd_mod.time, "sleep", lambda *_a, **_k: None)
+
+    dl = _downloader(tmp_path, session=_FakeSession())
+    outcomes = dl.download_post(_post(), [_video_item()], "artist-x")
+    assert calls["n"] == 1  # no retry on a permanent failure
+    assert outcomes[0].status == "error"
+
+
 def test_one_failure_isolated(tmp_path):
     class _BoomSession(_FakeSession):
         def get(self, url, stream=False, timeout=None):

tests/test_patreon_ingester.py

@@ -277,6 +277,50 @@ async def test_backfill_checkpoints_cursor_mid_walk(source_id, sync_engine, tmp_
     assert co.get("_backfill_cursor") == "CUR3"
 
 
+@pytest.mark.asyncio
+async def test_backfill_persists_live_posts_count(source_id, sync_engine, tmp_path, db):
+    """plan #704 #5: the ingester writes a monotonic _backfill_posts absolute
+    DURING the walk (seeded from prior chunks via posts_base), EXCLUDING the
+    re-walked resume page so the count doesn't inflate across chunks."""
+    # Resume from CUR2 (a prior chunk left off here): its page is re-walked and
+    # must NOT be re-counted. posts_base=4 stands in for the prior chunks.
+    pages = [
+        ("CUR2", [("p1", [_media("p1", 1)])]),   # the resumed page — not counted
+        ("CUR3", [("p2", [_media("p2", 1)])]),   # net-new
+        ("CUR4", [("p3", [_media("p3", 1)])]),   # net-new
+    ]
+    ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path))
+    ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="backfill",
+        resume_cursor="CUR2", posts_base=4,
+    )
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == source_id)
+    )).scalar_one()
+    # 4 prior + 2 net-new (p2, p3); p1 on the resumed CUR2 page is excluded.
+    assert co.get("_backfill_posts") == 6
+
+
+@pytest.mark.asyncio
+async def test_tick_does_not_persist_posts(source_id, sync_engine, tmp_path, db):
+    """A tick has no resumable backfill state, so it must not write _backfill_posts
+    (the badge is a backfill/recovery concept)."""
+    ing = _ingester(
+        sync_engine, tmp_path,
+        _FakeClient([("CUR2", [("p1", [_media("p1", 1)])])]),
+        _FakeDownloader(tmp_path),
+    )
+    ing.run(
+        source_id=source_id, campaign_id="c1", artist_slug="ingest",
+        url="https://patreon.com/ingest", mode="tick",
+    )
+    co = (await db.execute(
+        select(Source.config_overrides).where(Source.id == source_id)
+    )).scalar_one()
+    assert "_backfill_posts" not in (co or {})
+
+
 @pytest.mark.asyncio
 async def test_tick_does_not_checkpoint_cursor(source_id, sync_engine, tmp_path, db):
     """A tick has no resumable backfill state, so it must not write a cursor."""