From 593f65c9ccdc4b53b6585198e7f366c9de8b0393 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 4 Jun 2026 23:44:31 -0400 Subject: [PATCH] feat(download): cursor-paged Patreon backfill for large catalogs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Large Patreon creators (Anduo: weekly 50-120-image Reports back months = thousands of files) couldn't backfill: each run re-walked newest→oldest from the top, and gallery-dl's polite ~0.75s/request HEAD walk alone exceeded the 1170s subprocess budget, so the run died during enumeration with 0 files written and NO forward progress — re-stranding every time (event #40411). Checkpoint gallery-dl's pagination cursor so each backfill window advances the frontier: - gallery_dl.py: SourceConfig.resume_cursor; _build_config_for_source sets extractor.patreon.cursor= (PLATFORM_DEFAULTS leave log-only True for a fresh run); parse_last_cursor() pulls the last emitted 'Cursor: ' from stdout+stderr — survives a timed-out run since the TimeoutExpired path returns partial output. - download_service.py: phase2 stays in BACKFILL mode while a cursor is pending (even after the run budget drains) and threads resume_cursor; _apply_backfill_lifecycle() checkpoints the advancing cursor each non-completing run, completes on a clean rc=0 finish (walk reached bottom), and a stuck-guard clears the cursor after 2 non-advancing runs so a wedged walk can't re-strand forever. patreon-only (sole platform with a resumable cursor); other platforms keep the simple counter semantics. Cursor state lives in config_overrides JSON (patreon_campaign_id precedent) — no migration. Time-budget ladder (1170/1350/1500) unchanged. Plan #689. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/download_service.py | 106 +++++++++++---- backend/app/services/gallery_dl.py | 35 +++++ tests/test_download_service.py | 160 +++++++++++++++++++++-- tests/test_gallery_dl_service.py | 36 +++++ 4 files changed, 297 insertions(+), 40 deletions(-) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index f965a56..a293745 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -32,6 +32,7 @@ from .gallery_dl import ( ErrorType, GalleryDLService, SourceConfig, + parse_last_cursor, ) from .importer import Importer from .patreon_resolver import resolve_campaign_id @@ -112,10 +113,20 @@ class DownloadService: # post history (skip: True + 1170s); when 0, exit gallery-dl after # 20 contiguous archived items (skip: "exit:20" + the default # 870s). Operator sets backfill via POST /api/sources/{id}/backfill. + overrides = ctx["config_overrides"] or {} backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0 - if backfill_remaining > 0: + # Cursor-paged backfill (plan #689): a pending checkpoint keeps the + # deep-walk in backfill mode across ticks until it reaches the + # bottom, even after the operator's run budget hits 0 — so one large + # creator (Anduo) finishes without re-triggering. The phase-3 + # stuck-guard bounds it. patreon-only: it's the sole platform with a + # resumable gallery-dl cursor. + pending_cursor = overrides.get("_backfill_cursor") + if backfill_remaining > 0 or pending_cursor: skip_value: bool | str = BACKFILL_SKIP_VALUE source_config.timeout = BACKFILL_TIMEOUT_SECONDS + if ctx["platform"] == "patreon" and pending_cursor: + source_config.resume_cursor = pending_cursor else: skip_value = TICK_SKIP_VALUE @@ -395,36 +406,75 @@ class DownloadService: source_id=ctx["source_id"], status=status, error_message=ev.error, error_type=dl_result.error_type.value if dl_result.error_type else None, ) - # Plan #544: backfill lifecycle — auto-complete when a clean - # backfill run drained the queue (gallery-dl exited 0 + zero files - # downloaded means there was nothing to fetch); otherwise decrement - # the counter. Next tick falls back to tick mode once it hits 0. - # - # Audit 2026-06-02 gating: VALIDATION_FAILED also exits the - # subprocess with return_code=0 and files_downloaded=0 (every - # file was quarantined), which used to match the auto-complete - # predicate exactly — zeroing the operator's armed budget on - # the FIRST quarantine run instead of decrementing. Require - # dl_result.success + no error_type so only genuinely-empty - # successful runs drain the counter. - backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0 - if backfill_remaining > 0: - src = (await self.async_session.execute( - select(Source).where(Source.id == ctx["source_id"]) - )).scalar_one() - queue_drained = ( - dl_result.success - and dl_result.error_type is None - and dl_result.return_code == 0 - and dl_result.files_downloaded == 0 - ) - if queue_drained: - src.backfill_runs_remaining = 0 - else: - src.backfill_runs_remaining = max(0, backfill_remaining - 1) + await self._apply_backfill_lifecycle(ctx, dl_result) await self.async_session.commit() return event_id + async def _apply_backfill_lifecycle(self, ctx: dict, dl_result) -> None: + """Cursor-paged backfill state machine (plan #689). + + A backfill is "in progress" while backfill_runs_remaining > 0 OR a + checkpoint cursor is stored. Completion is a CLEAN rc=0 finish — + gallery-dl with skip:True exits 0 only after exhausting the + newest→oldest walk; a run cut short by the subprocess budget returns + success=False / return_code=-1 (the TimeoutExpired path), so rc=0 is + an unambiguous "reached the bottom" signal regardless of how many + files this run fetched. (This replaces plan #544's files==0 predicate, + whose audit-2026-06-02 VALIDATION_FAILED guard still holds: those exit + non-zero or carry an error_type, so they are not "completed".) + + For patreon, each non-completing run checkpoints gallery-dl's + last-emitted pagination cursor — even a timed-out run, since its + partial stdout/stderr still carries it — so the next run RESUMES + instead of re-walking from the top. A stuck-guard clears the cursor + after two consecutive runs that fail to advance it, so a wedged walk + can't re-strand forever (cf. download soft-limit ladder). + """ + overrides = ctx["config_overrides"] or {} + old_cursor = overrides.get("_backfill_cursor") + backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0 + if backfill_remaining <= 0 and not old_cursor: + return # tick mode — counter untouched + + src = (await self.async_session.execute( + select(Source).where(Source.id == ctx["source_id"]) + )).scalar_one() + new_overrides = dict(src.config_overrides or {}) + + completed = ( + dl_result.success + and dl_result.error_type is None + and dl_result.return_code == 0 + ) + if completed: + new_overrides.pop("_backfill_cursor", None) + new_overrides.pop("_backfill_cursor_stalls", None) + src.config_overrides = new_overrides + src.backfill_runs_remaining = 0 + return + + # Non-completing run. Patreon checkpoints + resumes via cursor; + # other platforms have no resumable cursor, so just decrement. + if ctx["platform"] == "patreon": + new_cursor = parse_last_cursor(dl_result.stdout, dl_result.stderr) + if new_cursor and new_cursor != old_cursor: + new_overrides["_backfill_cursor"] = new_cursor + new_overrides.pop("_backfill_cursor_stalls", None) + else: + stalls = int(new_overrides.get("_backfill_cursor_stalls", 0)) + 1 + if stalls >= 2: + # Wedged: give up, drop back to tick mode. The surfaced + # error_type on the event tells the operator why. + new_overrides.pop("_backfill_cursor", None) + new_overrides.pop("_backfill_cursor_stalls", None) + src.config_overrides = new_overrides + src.backfill_runs_remaining = 0 + return + new_overrides["_backfill_cursor_stalls"] = stalls + src.config_overrides = new_overrides + + src.backfill_runs_remaining = max(0, backfill_remaining - 1) + async def _update_source_health( self, *, source_id: int, status: str, error_message: str | None, error_type: str | None = None, diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py index dac6a6c..42a4fdf 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -13,6 +13,7 @@ from __future__ import annotations import asyncio import json import logging +import re import subprocess import sys import tempfile @@ -98,6 +99,15 @@ class SourceConfig: Source.backfill_runs_remaining column at the download_service layer and is passed to _build_config_for_source as `skip_value`. Same for the per-run subprocess timeout. + + `resume_cursor` is RUNTIME state, not persisted operator config: the + cursor-paged backfill checkpoint (see parse_last_cursor + the + download_service backfill lifecycle). When set on a patreon backfill + run, gallery-dl resumes its newest→oldest walk from that pagination + cursor instead of restarting from the top. It is threaded by + download_service from config_overrides["_backfill_cursor"]; SourceConfig + deliberately does NOT read it in from_dict (config_overrides also holds + operator config, and resume_cursor must only apply in backfill mode). """ content_types: list[str] = field(default_factory=lambda: ["all"]) sleep: float | None = None @@ -106,6 +116,7 @@ class SourceConfig: filename_pattern: str | None = None save_metadata: bool = True timeout: int = _DEFAULT_GDL_TIMEOUT_SECONDS + resume_cursor: str | None = None @classmethod def from_dict(cls, data: dict) -> SourceConfig: @@ -154,6 +165,21 @@ def _summarize_validation_failures(failures: list[dict]) -> str: return f"{n} files quarantined ({top_count}× {top_reason}, mixed)" +# gallery-dl logs its pagination cursor (when extractor.*.cursor is truthy) +# as `Cursor: ` — once per fetched page, at debug verbosity. The LAST +# such line is the furthest-progressed page, i.e. the resume point. We scan +# both streams (the debug log lands on stderr, but be defensive) and take the +# final match; passing it back as extractor.patreon.cursor resumes the walk. +# Survives a timed-out run too: the TimeoutExpired path returns the partial +# stdout/stderr, so an interrupted backfill still yields its last cursor. +_CURSOR_RE = re.compile(r"Cursor:\s*(\S+)") + + +def parse_last_cursor(stdout: str, stderr: str) -> str | None: + matches = _CURSOR_RE.findall(f"{stdout or ''}\n{stderr or ''}") + return matches[-1] if matches else None + + class GalleryDLService: """Service for executing gallery-dl downloads.""" @@ -349,6 +375,15 @@ class GalleryDLService: ] else: platform_section["files"] = source_config.content_types + # Cursor-paged backfill: resume the newest→oldest walk from the + # checkpoint saved by the previous run instead of restarting from + # the top. Without this, a large catalog (Anduo's weekly + # image-dense Reports) never reaches its frontier inside the + # subprocess budget — the HEAD walk alone times out, 0 files + # written, no progress (event #40411). The PLATFORM_DEFAULTS leave + # `cursor: True` (log-only) for the first/fresh run. + if source_config.resume_cursor: + platform_section["cursor"] = source_config.resume_cursor elif platform == "hentaifoundry": if "pictures" in source_config.content_types or "all" in source_config.content_types: platform_section["include"] = "all" diff --git a/tests/test_download_service.py b/tests/test_download_service.py index 8d15e60..97eae7e 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -376,12 +376,13 @@ async def test_finalize_skipped_preserves_failures_clears_error(db): @pytest.mark.asyncio -async def test_backfill_decrements_after_run( +async def test_backfill_decrements_and_checkpoints_cursor( db, db_sync, tmp_path, seed_artist_and_source, ): - """When backfill_runs_remaining > 0 going in, a non-clean / non-empty - run decrements by 1 — operator gets N runs to complete the deep scan - before tick mode resumes.""" + """A backfill run that did NOT finish the walk (subprocess budget cut it + short → success=False / rc!=0) decrements the run budget AND checkpoints + gallery-dl's last-emitted cursor, so the next run resumes instead of + re-walking from the top (plan #689, Anduo #40411).""" from backend.app.services.download_service import DownloadService from backend.app.services.importer import Importer @@ -390,12 +391,9 @@ async def test_backfill_decrements_after_run( await db.commit() images_root = tmp_path / "images" - f1 = images_root / "alice" / "patreon" / "post" / "a.jpg" - _make_jpg(f1, split="h") - fake_gdl = _fake_gdl_with_result(_make_fake_dl_result( - success=True, written_paths=[str(f1)], files_downloaded=1, - stdout=f"{f1}\n", + success=False, written_paths=[], files_downloaded=0, + stderr="[patreon][debug] Cursor: 03:PAGE2:xyz\n", )) sync_settings = db_sync.execute( @@ -412,10 +410,148 @@ async def test_backfill_decrements_after_run( ) await svc.download_source(source.id) - remaining = (await db.execute( - select(Source.backfill_runs_remaining).where(Source.id == source.id) - )).scalar_one() + remaining, overrides = (await db.execute( + select(Source.backfill_runs_remaining, Source.config_overrides) + .where(Source.id == source.id) + )).one() assert remaining == 2 + assert overrides.get("_backfill_cursor") == "03:PAGE2:xyz" + + +@pytest.mark.asyncio +async def test_backfill_cursor_keeps_backfill_mode_after_budget_drained( + db, db_sync, tmp_path, seed_artist_and_source, +): + """A pending cursor with runs_remaining == 0 still runs in BACKFILL mode + (skip:True + the resume cursor threaded to gallery-dl), so a large walk + finishes across ticks without the operator re-triggering.""" + from backend.app.services.download_service import DownloadService + from backend.app.services.gallery_dl import BACKFILL_SKIP_VALUE + from backend.app.services.importer import Importer + + _artist, source = seed_artist_and_source + source.backfill_runs_remaining = 0 + source.config_overrides = {"_backfill_cursor": "03:RESUME:here"} + await db.commit() + + fake_gdl = _fake_gdl_with_result(_make_fake_dl_result( + success=False, written_paths=[], + stderr="[patreon][debug] Cursor: 03:RESUME2:next\n", + )) + sync_settings = db_sync.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + importer = Importer( + session=db_sync, images_root=tmp_path / "images", + import_root=tmp_path / "images", + thumbnailer=Thumbnailer(images_root=tmp_path / "images"), + settings=sync_settings, + ) + cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)) + svc = DownloadService( + async_session=db, sync_session=db_sync, + gdl=fake_gdl, importer=importer, cred_service=cred_service, + ) + await svc.download_source(source.id) + + kwargs = fake_gdl.download.call_args.kwargs + assert kwargs["skip_value"] is BACKFILL_SKIP_VALUE + assert kwargs["source_config"].resume_cursor == "03:RESUME:here" + + overrides = (await db.execute( + select(Source.config_overrides).where(Source.id == source.id) + )).scalar_one() + assert overrides.get("_backfill_cursor") == "03:RESUME2:next" + + +@pytest.mark.asyncio +async def test_backfill_completes_and_clears_cursor( + db, db_sync, tmp_path, seed_artist_and_source, +): + """A clean rc=0 finish means gallery-dl walked to the bottom — clear the + checkpoint and drop back to tick mode, regardless of files this run.""" + from backend.app.services.download_service import DownloadService + from backend.app.services.importer import Importer + + _artist, source = seed_artist_and_source + source.backfill_runs_remaining = 2 + source.config_overrides = {"_backfill_cursor": "03:NEAR:bottom"} + await db.commit() + + fake_gdl = _fake_gdl_with_result(_make_fake_dl_result( + success=True, written_paths=[], files_downloaded=0, + stderr="[patreon][debug] Cursor: 03:LAST:page\n", + )) + sync_settings = db_sync.execute( + select(ImportSettings).where(ImportSettings.id == 1) + ).scalar_one() + importer = Importer( + session=db_sync, images_root=tmp_path / "images", + import_root=tmp_path / "images", + thumbnailer=Thumbnailer(images_root=tmp_path / "images"), + settings=sync_settings, + ) + cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)) + svc = DownloadService( + async_session=db, sync_session=db_sync, + gdl=fake_gdl, importer=importer, cred_service=cred_service, + ) + await svc.download_source(source.id) + + remaining, overrides = (await db.execute( + select(Source.backfill_runs_remaining, Source.config_overrides) + .where(Source.id == source.id) + )).one() + assert remaining == 0 + assert "_backfill_cursor" not in (overrides or {}) + + +@pytest.mark.asyncio +async def test_backfill_cursor_stuck_guard_gives_up( + db, db_sync, tmp_path, seed_artist_and_source, +): + """If two consecutive non-completing runs fail to advance the cursor, + drop the checkpoint + zero the budget so a wedged walk can't re-strand + forever (cf. download soft-limit ladder).""" + from unittest.mock import MagicMock + + from backend.app.services.download_service import DownloadService + + _artist, source = seed_artist_and_source + source.backfill_runs_remaining = 0 + source.config_overrides = {"_backfill_cursor": "stuck"} + await db.commit() + + svc = DownloadService( + async_session=db, sync_session=db_sync, + gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(), + ) + ctx = { + "source_id": source.id, "platform": "patreon", + "config_overrides": {"_backfill_cursor": "stuck"}, + "backfill_runs_remaining": 0, + } + stuck = _make_fake_dl_result(success=False, stderr="Cursor: stuck\n") + + # Run 1: cursor unchanged → 1 stall, checkpoint retained. + await svc._apply_backfill_lifecycle(ctx, stuck) + await db.commit() + overrides = (await db.execute( + select(Source.config_overrides).where(Source.id == source.id) + )).scalar_one() + assert overrides.get("_backfill_cursor") == "stuck" + assert overrides.get("_backfill_cursor_stalls") == 1 + + # Run 2: still stuck → give up. + ctx["config_overrides"] = dict(overrides) + await svc._apply_backfill_lifecycle(ctx, stuck) + await db.commit() + remaining, overrides2 = (await db.execute( + select(Source.backfill_runs_remaining, Source.config_overrides) + .where(Source.id == source.id) + )).one() + assert remaining == 0 + assert "_backfill_cursor" not in (overrides2 or {}) @pytest.mark.asyncio diff --git a/tests/test_gallery_dl_service.py b/tests/test_gallery_dl_service.py index d3607a5..5eadee9 100644 --- a/tests/test_gallery_dl_service.py +++ b/tests/test_gallery_dl_service.py @@ -8,6 +8,7 @@ from backend.app.services.gallery_dl import ( ErrorType, GalleryDLService, SourceConfig, + parse_last_cursor, ) @@ -326,3 +327,38 @@ def test_default_config_forwards_patreon_referer_to_ytdl(gdl): headers = cfg["downloader"]["ytdl"]["raw-options"]["http_headers"] assert headers["Referer"] == "https://www.patreon.com/" assert headers["Origin"] == "https://www.patreon.com" + + +# --- Cursor-paged backfill (plan #689) ------------------------------------- + + +def test_parse_last_cursor_returns_last_match_across_streams(): + # gallery-dl logs `Cursor: ` per page (debug → stderr); the last + # is the furthest-progressed resume point. + stderr = ( + "[patreon][debug] Cursor: 03:AAAA:bbb\n" + "[urllib3] GET ...\n" + "[patreon][debug] Cursor: 03:CCCC:ddd\n" + ) + assert parse_last_cursor("", stderr) == "03:CCCC:ddd" + + +def test_parse_last_cursor_scans_stdout_too_and_none_when_absent(): + assert parse_last_cursor("Cursor: 99:ZZZ", "") == "99:ZZZ" + assert parse_last_cursor("no marker here", "still nothing") is None + assert parse_last_cursor("", "") is None + + +def test_build_config_patreon_resume_cursor_overrides_default(gdl): + # No resume cursor → PLATFORM_DEFAULTS leaves the log-only `cursor: True`. + fresh = gdl._build_config_for_source( + "patreon", SourceConfig(), "alice", skip_value=True, + ) + assert fresh["extractor"]["patreon"]["cursor"] is True + + # Resume cursor set → gallery-dl restarts the walk from that page. + resumed = gdl._build_config_for_source( + "patreon", SourceConfig(resume_cursor="03:CCCC:ddd"), "alice", + skip_value=True, + ) + assert resumed["extractor"]["patreon"]["cursor"] == "03:CCCC:ddd"