diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 210a801..7de059c 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -9,9 +9,11 @@ construction by a thin adapter (e.g. `PatreonIngester`): - `client` — `.iter_posts(feed_id, cursor)` yielding `(post, included, page_cursor)` + `.extract_media(post, included) -> [media]`. - - `downloader`— `.download_post(post, media, artist_slug, is_seen) -> - [MediaOutcome]` (status in downloaded/skipped_seen/skipped_disk - /quarantined/error; `.path`/`.error`/`.post_id`). + - `downloader`— `.download_post(post, media, artist_slug, is_seen, + should_stop) -> [MediaOutcome]` (status in downloaded/ + skipped_seen/skipped_disk/quarantined/error; + `.path`/`.error`/`.post_id`). `should_stop()` is polled + between media so the time-box is honoured mid-post. - ledger — `seen_model` + `failed_model` SQLAlchemy models (+ their on-conflict UNIQUE constraint names) and a `ledger_key(media)`. - failure map — the adapter overrides `_failure_result` (platform exception @@ -237,8 +239,12 @@ class Ingester: def _is_skip(m, _skip=skip) -> bool: return ledger_key(m) in _skip + # Honour the time-box DURING a media-dense post too, not only at + # the per-post boundary below — else one heavy post can blow the + # chunk budget out to the Celery soft limit (Pocketacer, 2026-06-07). outcomes = self.downloader.download_post( - post, media, artist_slug, is_seen=_is_skip + post, media, artist_slug, is_seen=_is_skip, + should_stop=lambda: time.monotonic() - start >= time_budget_seconds, ) to_mark: list[tuple[str, str]] = [] diff --git a/backend/app/services/patreon_downloader.py b/backend/app/services/patreon_downloader.py index b5a9dea..7abc2df 100644 --- a/backend/app/services/patreon_downloader.py +++ b/backend/app/services/patreon_downloader.py @@ -190,6 +190,7 @@ class PatreonDownloader: artist_slug: str, *, is_seen: Callable[[object], bool] = lambda m: False, + should_stop: Callable[[], bool] = lambda: False, ) -> list[MediaOutcome]: """Download every media item of one post; return per-item outcomes. @@ -198,11 +199,19 @@ class PatreonDownloader: yt-dlp for video), writes the sidecar for each freshly-downloaded item, validates, and returns outcomes. Resilient: one media's failure yields an "error" outcome for that item; the rest proceed. + + `should_stop()` is polled BEFORE each media item: a media-dense post can + otherwise run a backfill chunk far past its time-box (the engine only + re-checks the budget between posts), so we honour the deadline mid-post + and return the items done so far — the rest re-fetch next chunk (they + were never marked seen). Bounds chunk overrun to one media download. """ post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post) outcomes: list[MediaOutcome] = [] for i, media in enumerate(media_items, start=1): + if should_stop(): + break try: outcomes.append( self._download_one(post, media, post_dir, artist_slug, i, is_seen) diff --git a/backend/app/tasks/download.py b/backend/app/tasks/download.py index 17d04e3..49af07e 100644 --- a/backend/app/tasks/download.py +++ b/backend/app/tasks/download.py @@ -94,10 +94,11 @@ def _finalize_soft_limited(session: SyncSession, source_id: int) -> None: ev.finished_at = now ev.error = ( f"killed by Celery soft time limit ({DOWNLOAD_SOFT_TIME_LIMIT}s) " - "before the gallery-dl subprocess returned — the run exceeded its " - "budget and its stdout/stderr were lost with the worker thread. " - "If this recurs, the source is too large for one run; the backfill " - "budget was decremented so the next tick walks less." + "before the download finished — the run exceeded its time budget. " + "Progress is checkpointed per page, so the next tick resumes near " + "the cut; the backfill budget was decremented so it walks less. If " + "this recurs, a single post is heavy enough to overrun the chunk " + "time-box on its own." ) ev.metadata_ = { **(ev.metadata_ or {}), diff --git a/tests/test_patreon_downloader.py b/tests/test_patreon_downloader.py index 0522807..4b19873 100644 --- a/tests/test_patreon_downloader.py +++ b/tests/test_patreon_downloader.py @@ -397,6 +397,25 @@ def test_skips_do_not_pace(tmp_path, monkeypatch): assert slept == [] +def test_download_post_honors_should_stop_mid_post(tmp_path): + """The time-box is polled BEFORE each media item, so a heavy post can't run + the whole chunk past its budget (Pocketacer soft-limit, 2026-06-07).""" + dl = PatreonDownloader( + images_root=tmp_path, cookies_path=None, validate=False, + session=_FakeSession(), + ) + items = [_img("a.png"), _img("b.png"), _img("c.png")] + polls = {"n": 0} + + def _stop(): + polls["n"] += 1 + return polls["n"] > 1 # allow the first item, stop before the second + + outcomes = dl.download_post(_post(), items, "artist-x", should_stop=_stop) + assert len(outcomes) == 1 # only the first item ran; the rest re-fetch later + assert outcomes[0].status == "downloaded" + + def test_media_429_retried_then_succeeds(tmp_path, monkeypatch): slept: list[float] = [] monkeypatch.setattr(pd_mod.time, "sleep", lambda s: slept.append(s)) diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 2e32075..bfe20a0 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -81,9 +81,12 @@ class _FakeDownloader: self.error = set(error or ()) self.download_calls = 0 - def download_post(self, post, media_items, artist_slug, *, is_seen): + def download_post(self, post, media_items, artist_slug, *, is_seen, + should_stop=lambda: False): outcomes = [] for m in media_items: + if should_stop(): + break if is_seen(m): outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None)) elif _ledger_key(m) in self.on_disk: