fix(patreon): enforce the backfill time-box mid-post (stop overrunning to the soft limit)

A backfill chunk's time-box (BACKFILL_CHUNK_SECONDS=600) was only checked
between POSTS, but download_post downloads ALL of one post's media
synchronously — so a single media-heavy post could run the chunk far past 600s,
all the way to the Celery soft time limit (1350s), where it was killed and
finalized as error (Pocketacer, event #41330: ran the full 22.5 min).

download_post now polls a should_stop() deadline BEFORE each media item and the
engine passes `now - start >= time_budget_seconds`, so a heavy post stops at the
budget and the remaining media (never marked seen) re-fetch next chunk. Bounds
chunk overrun to one media download instead of one whole post.

Also genericized the soft-limit salvage message — it claimed the "gallery-dl
subprocess" failed, which is wrong for a native Patreon walk; it now describes
the time-budget overrun + per-page checkpoint resume in platform-neutral terms.

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

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]] = []

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)

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 {}),

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))

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: