diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 8bb1e3d..6030cad 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -106,6 +106,22 @@ _LIVE_PROGRESS_INTERVAL = 5.0 # recapture (the operator's schema-test flow) reaches the sample. _CANARY_MIN_SAMPLE = 30 +# The walk's time budget covers only the walk, but phase 3 runs in the SAME +# Celery task, under the same soft limit (tasks/download.py: 1350s). A walk that +# finds a lot of work for phase 3 must stop early and leave it to the next chunk, +# or phase 3 is killed mid-import: TamadaHeijun's recapture, 2026-09-24, walked +# for ~2 min and then spent 20 min importing 431 orphans and relinking ~3000 +# 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. +CHUNK_TOTAL_SECONDS = 1200.0 +PHASE3_IMPORT_SECONDS = 2.5 +PHASE3_RELINK_SECONDS = 0.25 + def _parse_published(raw: object) -> datetime | None: """An ISO-8601 post date from either native client, as aware UTC. @@ -393,7 +409,17 @@ class Ingester: # Time-box check at the post boundary (coarse, like a gallery-dl # chunk). Backfill/recovery resume from emitted_cursor next chunk. - if time.monotonic() - start >= time_budget_seconds: + # The second half is phase 3's share of the task — see + # CHUNK_TOTAL_SECONDS. A mid-page stop resumes the same page. + elapsed = time.monotonic() - start + phase3 = ( + len(written) * PHASE3_IMPORT_SECONDS + + len(relink) * PHASE3_RELINK_SECONDS + ) + if ( + elapsed >= time_budget_seconds + or elapsed + phase3 >= CHUNK_TOTAL_SECONDS + ): budget_hit = True break @@ -711,7 +737,13 @@ class Ingester: # next chunk resumes from the emitted cursor. No progress → TIMEOUT, # which feeds download_service's backfill stall-guard. rc<0 mirrors # subprocess TimeoutExpired so completion detection stays false. - made_progress = downloaded > 0 or emitted_cursor != resume_cursor + # Work handed to phase 3 is progress too: a recapture chunk that + # stopped for its imports downloaded nothing, and may not have left + # its first page. + made_progress = ( + downloaded > 0 or bool(written) or bool(relink) + or emitted_cursor != resume_cursor + ) if made_progress: return _result( success=False, return_code=-1, diff --git a/tests/test_download_source_task.py b/tests/test_download_source_task.py index 22d59dd..1f50dd0 100644 --- a/tests/test_download_source_task.py +++ b/tests/test_download_source_task.py @@ -46,6 +46,15 @@ def test_timeout_ladder_keeps_subprocess_budgets_under_soft_limit(): assert DOWNLOAD_SOFT_TIME_LIMIT < DOWNLOAD_HARD_TIME_LIMIT +def test_the_native_chunk_leaves_room_to_tear_down_before_the_soft_limit(): + """The native walk sizes itself against CHUNK_TOTAL_SECONDS, walk plus + phase 3; that total has to leave the task time to finalize its event.""" + from backend.app.services.ingest_core import CHUNK_TOTAL_SECONDS + from backend.app.tasks.download import DOWNLOAD_SOFT_TIME_LIMIT + + assert CHUNK_TOTAL_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 47cf97f..1e344d7 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -846,6 +846,30 @@ async def test_recapture_does_not_refetch_seen_media_missing_from_disk( assert result.relink_source_paths == [] +@pytest.mark.asyncio +async def test_a_chunk_stops_walking_when_phase_3_would_overrun_the_task( + source_id, sync_engine, tmp_path, monkeypatch, +): + """Phase 3 shares the task's soft limit with the walk. TamadaHeijun's + recapture walked for two minutes and then died importing what it found. + With room for about one import, the walk takes two posts and stops.""" + import backend.app.services.ingest_core as core + monkeypatch.setattr(core, "CHUNK_TOTAL_SECONDS", core.PHASE3_IMPORT_SECONDS + 0.5) + + pages = [("CUR1", [(f"p{i}", [_media(f"p{i}", 1)]) for i in range(1, 4)])] + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), downloader) + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="backfill", time_budget_seconds=600.0, + ) + + assert result.files_downloaded == 2 + assert downloader.download_calls == 2 # p3 left for the next chunk + assert result.error_type == ErrorType.PARTIAL # a chunk boundary, not a failure + assert result.cursor == "CUR1" # resumes the page it was cut on + + # --- a run that dies between download and import --------------------------- # TamadaHeijun's 【12PCG】 post, 2026-09-24: 13 files on disk, 5 in the library. # A run wrote 01–08, marked them seen, and was killed before phase 3 imported