feat(download): cursor-paged Patreon backfill for large catalogs
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=<resume> (PLATFORM_DEFAULTS leave log-only True for a fresh run); parse_last_cursor() pulls the last emitted 'Cursor: <token>' 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) <noreply@anthropic.com>
This commit is contained in:
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user