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>
This commit is contained in:
@@ -9,9 +9,11 @@ construction by a thin adapter (e.g. `PatreonIngester`):
|
|||||||
|
|
||||||
- `client` — `.iter_posts(feed_id, cursor)` yielding `(post, included,
|
- `client` — `.iter_posts(feed_id, cursor)` yielding `(post, included,
|
||||||
page_cursor)` + `.extract_media(post, included) -> [media]`.
|
page_cursor)` + `.extract_media(post, included) -> [media]`.
|
||||||
- `downloader`— `.download_post(post, media, artist_slug, is_seen) ->
|
- `downloader`— `.download_post(post, media, artist_slug, is_seen,
|
||||||
[MediaOutcome]` (status in downloaded/skipped_seen/skipped_disk
|
should_stop) -> [MediaOutcome]` (status in downloaded/
|
||||||
/quarantined/error; `.path`/`.error`/`.post_id`).
|
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
|
- ledger — `seen_model` + `failed_model` SQLAlchemy models (+ their
|
||||||
on-conflict UNIQUE constraint names) and a `ledger_key(media)`.
|
on-conflict UNIQUE constraint names) and a `ledger_key(media)`.
|
||||||
- failure map — the adapter overrides `_failure_result` (platform exception
|
- failure map — the adapter overrides `_failure_result` (platform exception
|
||||||
@@ -237,8 +239,12 @@ class Ingester:
|
|||||||
def _is_skip(m, _skip=skip) -> bool:
|
def _is_skip(m, _skip=skip) -> bool:
|
||||||
return ledger_key(m) in _skip
|
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(
|
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]] = []
|
to_mark: list[tuple[str, str]] = []
|
||||||
|
|||||||
@@ -190,6 +190,7 @@ class PatreonDownloader:
|
|||||||
artist_slug: str,
|
artist_slug: str,
|
||||||
*,
|
*,
|
||||||
is_seen: Callable[[object], bool] = lambda m: False,
|
is_seen: Callable[[object], bool] = lambda m: False,
|
||||||
|
should_stop: Callable[[], bool] = lambda: False,
|
||||||
) -> list[MediaOutcome]:
|
) -> list[MediaOutcome]:
|
||||||
"""Download every media item of one post; return per-item outcomes.
|
"""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,
|
yt-dlp for video), writes the sidecar for each freshly-downloaded item,
|
||||||
validates, and returns outcomes. Resilient: one media's failure yields
|
validates, and returns outcomes. Resilient: one media's failure yields
|
||||||
an "error" outcome for that item; the rest proceed.
|
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)
|
post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post)
|
||||||
outcomes: list[MediaOutcome] = []
|
outcomes: list[MediaOutcome] = []
|
||||||
|
|
||||||
for i, media in enumerate(media_items, start=1):
|
for i, media in enumerate(media_items, start=1):
|
||||||
|
if should_stop():
|
||||||
|
break
|
||||||
try:
|
try:
|
||||||
outcomes.append(
|
outcomes.append(
|
||||||
self._download_one(post, media, post_dir, artist_slug, i, is_seen)
|
self._download_one(post, media, post_dir, artist_slug, i, is_seen)
|
||||||
|
|||||||
@@ -94,10 +94,11 @@ def _finalize_soft_limited(session: SyncSession, source_id: int) -> None:
|
|||||||
ev.finished_at = now
|
ev.finished_at = now
|
||||||
ev.error = (
|
ev.error = (
|
||||||
f"killed by Celery soft time limit ({DOWNLOAD_SOFT_TIME_LIMIT}s) "
|
f"killed by Celery soft time limit ({DOWNLOAD_SOFT_TIME_LIMIT}s) "
|
||||||
"before the gallery-dl subprocess returned — the run exceeded its "
|
"before the download finished — the run exceeded its time budget. "
|
||||||
"budget and its stdout/stderr were lost with the worker thread. "
|
"Progress is checkpointed per page, so the next tick resumes near "
|
||||||
"If this recurs, the source is too large for one run; the backfill "
|
"the cut; the backfill budget was decremented so it walks less. If "
|
||||||
"budget was decremented so the next tick walks less."
|
"this recurs, a single post is heavy enough to overrun the chunk "
|
||||||
|
"time-box on its own."
|
||||||
)
|
)
|
||||||
ev.metadata_ = {
|
ev.metadata_ = {
|
||||||
**(ev.metadata_ or {}),
|
**(ev.metadata_ or {}),
|
||||||
|
|||||||
@@ -397,6 +397,25 @@ def test_skips_do_not_pace(tmp_path, monkeypatch):
|
|||||||
assert slept == []
|
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):
|
def test_media_429_retried_then_succeeds(tmp_path, monkeypatch):
|
||||||
slept: list[float] = []
|
slept: list[float] = []
|
||||||
monkeypatch.setattr(pd_mod.time, "sleep", lambda s: slept.append(s))
|
monkeypatch.setattr(pd_mod.time, "sleep", lambda s: slept.append(s))
|
||||||
|
|||||||
@@ -81,9 +81,12 @@ class _FakeDownloader:
|
|||||||
self.error = set(error or ())
|
self.error = set(error or ())
|
||||||
self.download_calls = 0
|
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 = []
|
outcomes = []
|
||||||
for m in media_items:
|
for m in media_items:
|
||||||
|
if should_stop():
|
||||||
|
break
|
||||||
if is_seen(m):
|
if is_seen(m):
|
||||||
outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None))
|
outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None))
|
||||||
elif _ledger_key(m) in self.on_disk:
|
elif _ledger_key(m) in self.on_disk:
|
||||||
|
|||||||
Reference in New Issue
Block a user