diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 979676d..167a1e0 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -13,6 +13,7 @@ from __future__ import annotations import asyncio import logging +import time from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -39,7 +40,7 @@ from .gallery_dl import ( walk_completed, ) from .importer import Importer -from .ingest_core import DEFAULT_REVISIT_DAYS +from .ingest_core import DEFAULT_REVISIT_DAYS, PHASE3_DEADLINE_SECONDS from .platforms import auth_type_for from .scheduler_service import set_platform_cooldown @@ -82,6 +83,7 @@ class DownloadService: async def download_source(self, source_id: int) -> int: """Returns DownloadEvent.id. Idempotent: in-flight events are returned as-is.""" + started = time.monotonic() setup = await self._phase1_setup(source_id) if setup["status"] in ("skipped", "in_flight"): return setup["event_id"] @@ -174,6 +176,7 @@ class DownloadService: return await self._phase3_persist( ctx["event_id"], ctx, dl_result, resolved_campaign_id, + started=started, ) async def _run_download( @@ -268,6 +271,8 @@ class DownloadService: ctx: dict, dl_result, resolved_campaign_id: str | None, + *, + started: float | None = None, ) -> int: # Cache Patreon campaign id on the source. if resolved_campaign_id: @@ -303,7 +308,23 @@ class DownloadService: loop = asyncio.get_running_loop() - for path_str in dl_result.written_paths or []: + # Phase 3 shares the task's soft limit with the walk, and an import's + # cost is only estimated there. Native platforms stop starting imports + # at the deadline: what is left stays on disk with no record, which the + # next chunk's walk imports. gallery-dl platforms have no such pickup, + # so they import everything. + deadline = ( + started + PHASE3_DEADLINE_SECONDS + if started is not None and uses_native_ingester(ctx["platform"]) + else None + ) + written_paths = list(dl_result.written_paths or []) + deferred_from: int | None = None + + for idx, path_str in enumerate(written_paths): + if deadline is not None and time.monotonic() >= deadline: + deferred_from = idx + break if path_str in (dl_result.quarantined_paths or []): continue path = Path(path_str) @@ -427,8 +448,16 @@ class DownloadService: # has been through the importer. Had the run died before here they stay # unmarked, and the next walk imports them from disk (ingest_core). mark_seen = getattr(dl_result, "mark_seen_after_import", None) + deferred = written_paths[deferred_from:] if deferred_from is not None else [] if mark_seen is not None: - await loop.run_in_executor(None, mark_seen) + if deferred: + await loop.run_in_executor( + None, lambda: mark_seen(skip_paths=frozenset(deferred)), + ) + else: + await loop.run_in_executor(None, mark_seen) + if deferred: + self._defer_unimported(dl_result, deferred_from, len(deferred)) # #830 recapture: backfill source_filehash on EXISTING on-disk images so # their post-body inline remaps to the local copy. A @@ -530,6 +559,33 @@ class DownloadService: await self.async_session.commit() return event_id + @staticmethod + def _defer_unimported(dl_result, first: int, count: int) -> None: + """Turn this chunk into a chunk boundary at the first file phase 3 left. + + The next chunk resumes on the page that file came from, so its walk + meets every left file again, on disk with no record, and imports it. A + walk that reached its end has not finished while files wait, so a clean + result becomes PARTIAL; a real failure keeps its own error. + """ + cursors = getattr(dl_result, "written_cursors", None) or [] + if first < len(cursors): + dl_result.cursor = cursors[first] + dl_result.phase3_imported_any = first > 0 + msg = ( + f"Imports stopped at the chunk deadline: {count} file(s) left " + "for the next chunk" + ) + dl_result.stdout = f"{dl_result.stdout}\n{msg}" if dl_result.stdout else msg + log.info("%s", msg) + if dl_result.error_type in (None, ErrorType.PARTIAL, ErrorType.TIMEOUT) or ( + is_informational(dl_result.error_type) + ): + dl_result.success = False + dl_result.return_code = -1 + dl_result.error_type = ErrorType.PARTIAL + dl_result.error_message = msg + async def _apply_backfill_lifecycle(self, ctx: dict, dl_result) -> None: """Backfill state machine (plan #693, building on the cursor of #689). @@ -597,9 +653,12 @@ class DownloadService: new_cursor = ( dl_result.cursor if uses_native_ingester(ctx["platform"]) else None ) + # Imports phase 3 finished before its deadline are progress even when + # the cursor went back to the page holding the files it left. advanced = bool( (new_cursor and new_cursor != old_cursor) or dl_result.files_downloaded > 0 + or getattr(dl_result, "phase3_imported_any", False) ) if advanced: if new_cursor: diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py index 4ee68ae..98ce034 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -169,6 +169,10 @@ class DownloadResult: # backfilled (inline-image localization) WITHOUT re-download or unlink. Empty # on the gallery-dl path and outside recapture. relink_source_paths: list[tuple[str, str, str]] = field(default_factory=list) + # Native ingester only: one entry per `written_paths` entry, the backfill + # cursor of the page that file came from. Phase 3 resumes the next chunk + # there when it leaves files unimported at its deadline. Empty on gallery-dl. + written_cursors: list[str | None] = field(default_factory=list) stdout: str = "" stderr: str = "" return_code: int = 0 @@ -194,8 +198,10 @@ class DownloadResult: # but not yet imported is invisible to every later walk, so a run killed in # between orphaned it for good (TamadaHeijun's 12PCG post lost 7 of 13 # images to a stranded run, 2026-09-24). Unmarked, the next walk finds the - # file on disk with no ImageRecord and imports it. None on gallery-dl. - mark_seen_after_import: Callable[[], None] | None = None + # file on disk with no ImageRecord and imports it. Takes `skip_paths`: the + # written paths phase 3 did not import, which stay unmarked for the same + # reason. None on gallery-dl. + mark_seen_after_import: Callable[..., None] | None = None def extract_errors_warnings(stderr: str) -> str: diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 2c8a859..2ee9154 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -114,13 +114,29 @@ _CANARY_MIN_SAMPLE = 30 # on-disk files, and died at the soft limit. # # So the walk also stops when its elapsed time PLUS phase 3's estimated cost -# would pass CHUNK_TOTAL_SECONDS. Costs measured on the live instance: 431 -# imports took 976s (~2.3s each: hash, pHash, sidecar, provenance); a relink is -# a sha256 over NFS, 0.15s for the 8.8 MB average file, plus a lookup. -# test_download_source_task pins CHUNK_TOTAL_SECONDS under the soft limit. +# would pass CHUNK_TOTAL_SECONDS. An import (hash, pHash, sidecar, provenance, +# thumbnail enqueue) runs 2-5s depending on the files; the estimate takes the +# slow end. A relink is a sha256 over NFS plus a lookup. +# +# The estimate is a plan, not a guard. PHASE3_DEADLINE_SECONDS is the guard: +# phase 3 starts no import past it (download_service), and leaves the rest on +# disk, unmarked, for the next chunk. It sits under the soft limit by enough to +# finish the import in hand, upsert post records and finalize the event. +# test_download_source_task pins both under the soft limit. CHUNK_TOTAL_SECONDS = 1200.0 -PHASE3_IMPORT_SECONDS = 2.5 +PHASE3_IMPORT_SECONDS = 5.0 PHASE3_RELINK_SECONDS = 0.25 +PHASE3_DEADLINE_SECONDS = 1200.0 + + +def _unskipped( + fetched: list[tuple[str, str]], + written_keys: dict[str, tuple[str, str]], + skip_paths, +) -> list[tuple[str, str]]: + """`fetched` without the ledger keys of files phase 3 left unimported.""" + skip = {written_keys[p] for p in skip_paths if p in written_keys} + return [kv for kv in fetched if kv not in skip] if skip else list(fetched) def _parse_published(raw: object) -> datetime | None: @@ -289,6 +305,12 @@ class Ingester: # Media handed to phase 3 for import. Marked seen by phase 3 once the # import has run (`mark_seen_after_import`), not here — see there. fetched: list[tuple[str, str]] = [] + # Per entry of `written`: the cursor of the page it came from, and its + # ledger key. Phase 3 can stop importing at its deadline; the files it + # leaves must stay unmarked, and the next chunk has to resume on the + # page holding the first of them, or no walk sees them again. + written_cursors: list[str | None] = [] + written_keys: dict[str, tuple[str, str]] = {} # Post-record keys written this walk. Marked with the media, after # phase 3 has upserted the records — marking them at write time left a # walk killed before phase 3 with posts the ledger calls recorded that @@ -347,8 +369,10 @@ class Ingester: written_paths=written, post_record_paths=list(post_records), relink_source_paths=list(relink), - mark_seen_after_import=lambda: self._mark_seen( - source_id, fetched + recorded, + written_cursors=list(written_cursors), + mark_seen_after_import=lambda skip_paths=(): self._mark_seen( + source_id, + _unskipped(fetched, written_keys, skip_paths) + recorded, ), stdout="\n".join(log_lines), stderr="", @@ -577,6 +601,8 @@ class Ingester: downloaded += 1 if outcome.path is not None: written.append(str(outcome.path)) + written_cursors.append(emitted_cursor) + written_keys[str(outcome.path)] = (key, media_item.post_id) fetched.append((key, media_item.post_id)) to_clear.append(key) consecutive_seen = 0 @@ -590,6 +616,8 @@ class Ingester: # attach_in_place because no record owns this path — # the unlink below is about a file that IS the record. written.append(str(outcome.path)) + written_cursors.append(emitted_cursor) + written_keys[str(outcome.path)] = (key, media_item.post_id) fetched.append((key, media_item.post_id)) to_clear.append(key) skipped_count += 1 diff --git a/tests/test_download_service.py b/tests/test_download_service.py index a347646..b8f1dc2 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -570,6 +570,64 @@ async def test_backfill_chunk_progress_advances_cursor( assert remaining == 4 +@pytest.mark.asyncio +async def test_phase_3_leaves_imports_past_its_deadline_for_the_next_chunk( + db, db_sync, tmp_path, seed_artist_and_source, monkeypatch, +): + """A walk that reached its end handed phase 3 more imports than fit before + the deadline. Phase 3 imports what fits, leaves the rest on disk and + unmarked, keeps the backfill running, and resumes it on the page holding + the first file it left.""" + import backend.app.services.download_service as ds + + _artist, source = seed_artist_and_source + source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:PAGE1"} + source.backfill_runs_remaining = 5 + await db.commit() + + images_root = tmp_path / "images" + f1 = images_root / "alice" / "patreon" / "post" / "a.jpg" + f2 = images_root / "alice" / "patreon" / "post" / "b.jpg" + _make_jpg(f1, split="h") + _make_jpg(f2, split="v") + result = _make_fake_dl_result( + success=True, written_paths=[str(f1), str(f2)], cursor="03:PAGE3", + ) + result.written_cursors = ["03:PAGE1", "03:PAGE2"] + left_unmarked = [] + result.mark_seen_after_import = lambda skip_paths=(): left_unmarked.append( + set(skip_paths) + ) + + # The clock is past the deadline once the first import is done. + past = ds.PHASE3_DEADLINE_SECONDS + 1 + ticks = iter([0.0, 0.0]) + monkeypatch.setattr(ds, "time", SimpleNamespace(monotonic=lambda: next(ticks, past))) + + svc, _ = _backfill_svc(db, db_sync, tmp_path, result) + event_id = await svc.download_source(source.id) + + ev = (await db.execute( + select(DownloadEvent).where(DownloadEvent.id == event_id) + )).scalar_one() + assert ev.status == "ok" + assert ev.files_count == 1 + assert "1 file(s) left for the next chunk" in ev.metadata_["stdout"] + assert left_unmarked == [{str(f2)}] + assert f2.exists() + assert db_sync.execute( + select(func.count(ImageRecord.id)).where(ImageRecord.path == str(f2)) + ).scalar_one() == 0 + + remaining, co = (await db.execute( + select(Source.backfill_runs_remaining, Source.config_overrides) + .where(Source.id == source.id) + )).one() + assert co.get("_backfill_state") == "running" + assert co.get("_backfill_cursor") == "03:PAGE2" + assert remaining == 4 + + @pytest.mark.asyncio async def test_backfill_state_running_selects_backfill_mode_and_resumes( db, db_sync, tmp_path, seed_artist_and_source, diff --git a/tests/test_download_source_task.py b/tests/test_download_source_task.py index 1f50dd0..6f31167 100644 --- a/tests/test_download_source_task.py +++ b/tests/test_download_source_task.py @@ -55,6 +55,15 @@ def test_the_native_chunk_leaves_room_to_tear_down_before_the_soft_limit(): assert CHUNK_TOTAL_SECONDS <= DOWNLOAD_SOFT_TIME_LIMIT - 120 +def test_phase_3_stops_importing_with_room_to_finalize(): + """Past its deadline phase 3 starts no import; what remains after it (the + import in hand, post records, the event) has to fit before the soft limit.""" + from backend.app.services.ingest_core import PHASE3_DEADLINE_SECONDS + from backend.app.tasks.download import DOWNLOAD_SOFT_TIME_LIMIT + + assert PHASE3_DEADLINE_SECONDS <= DOWNLOAD_SOFT_TIME_LIMIT - 120 + + def test_decorated_limits_match_module_constants(): """The @celery.task decorator must use the audited constants, not drifted literals.""" diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 7ff79c7..c857069 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -984,6 +984,31 @@ async def test_recapture_imports_a_seen_file_nothing_imported( assert downloader.download_calls == 0 +@pytest.mark.asyncio +async def test_files_phase_3_leaves_unimported_stay_unmarked( + source_id, sync_engine, tmp_path, +): + """Phase 3 can stop importing at its deadline. Each written file names the + page it came from, so the next chunk can resume there; the files left keep + no ledger mark, so that walk meets them on disk and imports them.""" + m1, m2 = _media("p1", 1), _media("p2", 1) + pages = [("CUR1", [("p1", [m1])]), ("CUR2", [("p2", [m2])])] + ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path)) + result = ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="backfill") + assert result.written_cursors == ["CUR1", "CUR2"] + + result.mark_seen_after_import(skip_paths={str(tmp_path / "p2_1.jpg")}) + assert _count_ledger(sync_engine, source_id) == 3 # m1 and both post records + + ing2 = _ingester(sync_engine, tmp_path, _FakeClient(pages), + _FakeDownloader(tmp_path, on_disk={_ledger_key(m1), _ledger_key(m2)})) + again = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="backfill") + assert again.written_paths == [str(tmp_path / "p2_1.jpg")] + assert "on disk but never imported: p2_1.jpg" in again.stdout + + # --- dead-letter ledger (plan #705 #7) ------------------------------------