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,

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."""

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