From 697a86d31c88befe05bcc9846c09d496eb7cbd43 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 09:56:26 -0400 Subject: [PATCH] fix(ingester): close #5 within-chunk live posts + #8 video transient retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/services/download_service.py | 16 ++++--- backend/app/services/ingest_core.py | 43 +++++++++++++++++- backend/app/services/patreon_downloader.py | 52 +++++++++++++++++----- tests/test_download_service.py | 13 +++--- tests/test_patreon_downloader.py | 51 +++++++++++++++++++++ tests/test_patreon_ingester.py | 44 ++++++++++++++++++ 6 files changed, 194 insertions(+), 25 deletions(-) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 47cd4c0..390d854 100644 --- a/backend/app/services/download_service.py +++ b/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 diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 099f634..0e6fb5d 100644 --- a/backend/app/services/ingest_core.py +++ b/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. diff --git a/backend/app/services/patreon_downloader.py b/backend/app/services/patreon_downloader.py index 7125a7e..ad001c8 100644 --- a/backend/app/services/patreon_downloader.py +++ b/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: diff --git a/tests/test_download_service.py b/tests/test_download_service.py index 253c29e..d08eb12 100644 --- a/tests/test_download_service.py +++ b/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 diff --git a/tests/test_patreon_downloader.py b/tests/test_patreon_downloader.py index 1d42285..e606489 100644 --- a/tests/test_patreon_downloader.py +++ b/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): diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 32db179..3ca19a5 100644 --- a/tests/test_patreon_ingester.py +++ b/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."""