From da2091a875ed22ea5ed908157922fd473922d559 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 24 Sep 2026 14:13:12 -0400 Subject: [PATCH] fix: a download is marked seen only after it is imported A run killed between download and import left its files on disk and in the seen-ledger but never in the library, and every later walk trusted the ledger. TamadaHeijun's 12PCG post lost 7 of 13 images this way (stranded run 90402), which read as the duplicates filter. The ingester now hands phase 3 a mark_seen_after_import hook, called after the import loop. A file on disk with no ImageRecord at its path is fed to import, not reconciled into the ledger. Recapture reaches files already orphaned, because it looks past the ledger to the disk. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR --- backend/app/services/download_service.py | 7 ++ backend/app/services/gallery_dl.py | 8 +++ backend/app/services/ingest_core.py | 44 +++++++++++- .../subscriptions/SourceActions.vue | 5 +- tests/test_download_service.py | 13 +++- tests/test_patreon_ingester.py | 70 +++++++++++++++++++ 6 files changed, 141 insertions(+), 6 deletions(-) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 825bb82..979676d 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -423,6 +423,13 @@ class DownloadService: await loop.run_in_executor(None, _upsert) + # Only now is it safe to call this walk's media seen: every file above + # 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) + if mark_seen is not None: + await loop.run_in_executor(None, mark_seen) + # #830 recapture: backfill source_filehash on EXISTING on-disk images so # their post-body inline remaps to the local copy. A # SEPARATE non-deleting channel (NOT the import list — that would unlink diff --git a/backend/app/services/gallery_dl.py b/backend/app/services/gallery_dl.py index b07fb1b..90933bf 100644 --- a/backend/app/services/gallery_dl.py +++ b/backend/app/services/gallery_dl.py @@ -17,6 +17,7 @@ import subprocess import sys import tempfile import time +from collections.abc import Callable from dataclasses import dataclass, field from datetime import UTC, datetime from enum import StrEnum @@ -230,6 +231,13 @@ class DownloadResult: # the platform cooldown matches the hint instead of a flat default. None when # unknown (no header, or not a rate-limit failure). retry_after_seconds: float | None = None + # Native ingester only: marks this walk's fetched media seen in its ledger. + # Phase 3 calls it AFTER the import loop, never before — a file marked seen + # 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 def extract_errors_warnings(stderr: str) -> str: diff --git a/backend/app/services/ingest_core.py b/backend/app/services/ingest_core.py index 8105baf..8bb1e3d 100644 --- a/backend/app/services/ingest_core.py +++ b/backend/app/services/ingest_core.py @@ -36,6 +36,7 @@ from datetime import UTC, datetime, timedelta from sqlalchemy import delete, func, select, text from sqlalchemy.dialects.postgresql import insert as pg_insert +from ..models import ImageRecord from .gallery_dl import ( DownloadResult, ErrorType, @@ -261,6 +262,9 @@ class Ingester: # source_filehash and (b) link the on-disk image to its Post (#1288) — # WITHOUT re-downloading or unlinking the file. Empty outside recapture. relink: list[tuple[str, str, str]] = [] + # 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]] = [] downloaded = 0 errors = 0 quarantined = 0 @@ -313,6 +317,7 @@ 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), stdout="\n".join(log_lines), stderr="", return_code=return_code, @@ -512,6 +517,13 @@ class Ingester: recapture=recapture, ) + # An on-disk file is only "done" if something imported it. One + # with no ImageRecord at its path was written by a run that died + # before phase 3 — it goes to import, not to the ledger. + imported_paths = self._recorded_paths([ + str(o.path) for o in outcomes + if o.status == "skipped_disk" and o.path is not None + ]) to_mark: list[tuple[str, str]] = [] to_clear: list[str] = [] # recovered → drop any dead-letter row to_fail: list[tuple[str, str, str]] = [] # (key, post_id, error) @@ -523,11 +535,29 @@ class Ingester: downloaded += 1 if outcome.path is not None: written.append(str(outcome.path)) - to_mark.append((key, media_item.post_id)) + fetched.append((key, media_item.post_id)) to_clear.append(key) consecutive_seen = 0 + elif ( + outcome.status == "skipped_disk" + and outcome.path is not None + and str(outcome.path) not in imported_paths + ): + # On disk, never imported: a prior run wrote it and died + # before phase 3. Import it now. Safe to feed to + # 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)) + fetched.append((key, media_item.post_id)) + to_clear.append(key) + skipped_count += 1 + consecutive_seen += 1 + log_lines.append( + f" post {media_item.post_id} — on disk but never " + f"imported: {outcome.path.name}" + ) elif outcome.status == "skipped_disk": - # Already on disk (a prior run). Reconcile the ledger so a + # Already on disk and imported. Reconcile the ledger so a # later tick skips it at tier-1 without a disk stat, but # do NOT re-feed it to phase 3 — attach_in_place would see # the duplicate sha256 and unlink the on-disk copy. @@ -887,6 +917,16 @@ class Ingester: ) session.commit() + def _recorded_paths(self, paths: list[str]) -> set[str]: + """Which of `paths` an ImageRecord already points at.""" + if not paths: + return set() + with self.session_factory() as session: + rows = session.execute( + select(ImageRecord.path).where(ImageRecord.path.in_(paths)) + ).scalars().all() + return set(rows) + def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None: """Idempotent upsert of (filehash, post_id) seen-ledger rows for a page. diff --git a/frontend/src/components/subscriptions/SourceActions.vue b/frontend/src/components/subscriptions/SourceActions.vue index e026aed..34e484f 100644 --- a/frontend/src/components/subscriptions/SourceActions.vue +++ b/frontend/src/components/subscriptions/SourceActions.vue @@ -45,8 +45,9 @@ > Recapture post text & links - Re-grab every post's body + external links and localize inline images - already on disk — without re-downloading media + Re-grab every post's body + external links, localize inline images + already on disk, and import any downloaded file that never reached the + library — without re-downloading media diff --git a/tests/test_download_service.py b/tests/test_download_service.py index 7533dcb..a347646 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -9,9 +9,9 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest -from sqlalchemy import select +from sqlalchemy import func, select -from backend.app.models import Artist, DownloadEvent, ImportSettings, Source +from backend.app.models import Artist, DownloadEvent, ImageRecord, ImportSettings, Source from backend.app.services.credential_crypto import CredentialCrypto from backend.app.services.credential_service import CredentialService from backend.app.services.thumbnailer import Thumbnailer @@ -152,6 +152,14 @@ async def test_download_source_attaches_written_files( files_downloaded=2, stdout=f"{f1}\n{f2}\n", ) + # The ledger is marked only once the files are in: a run killed before + # import must leave them unmarked so the next walk imports them. + imported_when_marked = [] + result.mark_seen_after_import = lambda: imported_when_marked.append( + db_sync.execute( + select(func.count(ImageRecord.id)).where(ImageRecord.path.in_([str(f1), str(f2)])) + ).scalar_one() + ) fake_gdl = _fake_gdl_with_result(result) sync_settings = db_sync.execute( @@ -182,6 +190,7 @@ async def test_download_source_attaches_written_files( assert ev.files_count == 2 assert ev.metadata_["import_summary"]["attached"] == 2 assert ev.metadata_["run_stats"]["downloaded_count"] == 2 + assert imported_when_marked == [2] @pytest.mark.asyncio diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py index 774722b..7f56a96 100644 --- a/tests/test_patreon_ingester.py +++ b/tests/test_patreon_ingester.py @@ -7,6 +7,7 @@ real CDN. The ledger is real (a sync sessionmaker bound to the test engine), so the tier-1 skip and the idempotent mark-seen run against actual rows. """ +import hashlib from datetime import UTC, datetime, timedelta import pytest @@ -15,6 +16,7 @@ from sqlalchemy.orm import sessionmaker from backend.app.models import ( Artist, + ImageRecord, PatreonFailedMedia, PatreonSeenMedia, Source, @@ -237,6 +239,9 @@ async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_ # plan #704: structured run_stats carry the real counts. assert result.run_stats["downloaded_count"] == 2 assert result.posts_processed == 1 + # The media wait for phase 3 to import them; only the post key is in yet. + assert _count_ledger(sync_engine, source_id) == 1 + result.mark_seen_after_import() # 2 media keys + 1 synthetic post key (body/links recaptured per post). assert _count_ledger(sync_engine, source_id) == 3 # The post body + links are captured for media posts too (rides the walk). @@ -263,6 +268,7 @@ async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_ assert result.run_stats["quarantined_count"] == 1 assert result.run_stats["downloaded_count"] == 1 assert len(result.written_paths) == 1 # quarantined NOT written + result.mark_seen_after_import() # Quarantined media is NOT marked seen (a fixed file may be re-fetched); # m1 + the synthetic post key (body/links captured per post) = 2. assert _count_ledger(sync_engine, source_id) == 2 @@ -573,6 +579,7 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path) m1 = _media("p1", 1) client = _FakeClient([(None, [("p1", [m1])])]) # File still on disk (a kept image) → tier-2 spares it even under recovery. + _seed_record(sync_engine, tmp_path / "p1_1.jpg") downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)}) ing = _ingester(sync_engine, tmp_path, client, downloader) @@ -582,6 +589,7 @@ async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path) ) assert result.files_downloaded == 0 assert downloader.download_calls == 0 + assert result.written_paths == [] # Disk-skip reconciles the media key + the synthetic post key (recovery # recaptures the body/links per post) = 2. assert _count_ledger(sync_engine, source_id) == 2 @@ -609,6 +617,15 @@ async def test_backfill_recaptures_body_for_already_downloaded_post( assert downloader.post_records == 1 +def _seed_record(sync_engine, path): + """An ImageRecord at `path` — the file was imported, not just downloaded.""" + factory = sessionmaker(sync_engine, expire_on_commit=False) + with factory() as s: + s.add(ImageRecord(path=str(path), sha256=hashlib.sha256(str(path).encode()).hexdigest(), + size_bytes=1, mime="image/jpeg", origin="downloaded")) + s.commit() + + def _seed_seen(sync_engine, source_id, key, post_id=None): factory = sessionmaker(sync_engine, expire_on_commit=False) with factory() as s: @@ -629,6 +646,7 @@ async def test_backfill_skips_already_captured_post_but_recapture_forces_it( # Pre-seed BOTH the media key and the synthetic post key as already seen. _seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1") _seed_seen(sync_engine, source_id, "post:p1", post_id="p1") + _seed_record(sync_engine, tmp_path / "p1_1.jpg") # 1) Plain backfill: post key is seen → gate skips body recapture. client = _FakeClient([(None, [("p1", [m1])])]) @@ -696,6 +714,7 @@ async def test_gated_post_skipped_entirely_no_media_no_record( assert downloader.download_calls == 1 assert len(result.post_record_paths) == 1 # only the open post assert downloader.post_records == 1 + result.mark_seen_after_import() # The gated post left NO trace in the seen-ledger (no media key, no post key): # only the open post's media key + synthetic post key are recorded. assert _count_ledger(sync_engine, source_id) == 2 @@ -827,6 +846,57 @@ async def test_recapture_does_not_refetch_seen_media_missing_from_disk( assert result.relink_source_paths == [] +# --- 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 +# them; every later walk trusted the ledger and never looked again. + + +@pytest.mark.asyncio +async def test_a_run_that_dies_before_import_leaves_its_media_unmarked( + source_id, sync_engine, tmp_path, +): + m1 = _media("p1", 1) + ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), + _FakeDownloader(tmp_path)) + ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick") + # Phase 3 never ran, so `mark_seen_after_import` never did: only the post key. + assert _count_ledger(sync_engine, source_id) == 1 + + # The next walk finds the file on disk with no record, and imports it. + ing2 = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), + _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})) + result = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick") + assert result.written_paths == [str(tmp_path / "p1_1.jpg")] + assert result.files_downloaded == 0 # not fetched again + assert "on disk but never imported: p1_1.jpg" in result.stdout + result.mark_seen_after_import() + assert _count_ledger(sync_engine, source_id) == 2 + + +@pytest.mark.asyncio +async def test_recapture_imports_a_seen_file_nothing_imported( + source_id, sync_engine, tmp_path, +): + """The repair for files already orphaned: they are in the ledger, so only + a walk that looks past it — recapture — reaches them.""" + m1, m2 = _media("p1", 1), _media("p1", 2) + for m in (m1, m2): + _seed_seen(sync_engine, source_id, _ledger_key(m), post_id="p1") + _seed_record(sync_engine, tmp_path / "p1_2.jpg") # m2 made it in; m1 did not + downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1), _ledger_key(m2)}) + ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1, m2])])]), + downloader) + result = ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="recapture") + + assert result.written_paths == [str(tmp_path / "p1_1.jpg")] + assert [r[0] for r in result.relink_source_paths] == [str(tmp_path / "p1_2.jpg")] + assert downloader.download_calls == 0 + + # --- dead-letter ledger (plan #705 #7) ------------------------------------