Merge pull request 'Patreon: enforce the backfill time-box mid-post (stop soft-limit overruns)' (#78) from dev into main
Build images / sign-extension (push) Successful in 3s
CI / lint (push) Successful in 3s
Build images / build-ml (push) Successful in 8s
Build images / build-web (push) Successful in 8s
CI / frontend-build (push) Successful in 19s
CI / backend-lint-and-test (push) Successful in 27s
CI / integration (push) Successful in 3m3s
Build images / sign-extension (push) Successful in 3s
CI / lint (push) Successful in 3s
Build images / build-ml (push) Successful in 8s
Build images / build-web (push) Successful in 8s
CI / frontend-build (push) Successful in 19s
CI / backend-lint-and-test (push) Successful in 27s
CI / integration (push) Successful in 3m3s
This commit was merged in pull request #78.
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,
|
||||
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]] = []
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 {}),
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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:
|
||||
@@ -346,11 +349,14 @@ async def test_tick_does_not_checkpoint_cursor(source_id, sync_engine, tmp_path,
|
||||
async def test_backfill_budget_cut_returns_partial_with_progress(
|
||||
source_id, sync_engine, tmp_path, monkeypatch,
|
||||
):
|
||||
# Deterministic clock: start=0, post1 check=10 (ok), post2 check=200 (>budget).
|
||||
# run() lives in ingest_core now, so patch the clock there (the same global
|
||||
# time module object, but we reference it through the module that uses it).
|
||||
# Deterministic clock: start=0; post1 gate=10 (ok); post1's per-media
|
||||
# should_stop=20 (still ok, so post1's item downloads); post1 live-progress
|
||||
# read=200; post2 gate=250 (>budget → cut). The per-media should_stop check
|
||||
# (added 2026-06-07 to bound media-dense posts) reads the clock once more per
|
||||
# post, so the sequence carries the extra tick. run() lives in ingest_core
|
||||
# now, so patch the clock there.
|
||||
import backend.app.services.ingest_core as core
|
||||
ticks = iter([0.0, 10.0, 200.0, 250.0])
|
||||
ticks = iter([0.0, 10.0, 20.0, 200.0, 250.0])
|
||||
last = [0.0]
|
||||
|
||||
def fake_monotonic():
|
||||
|
||||
Reference in New Issue
Block a user