From bd8fe59b49e162dc20d603901f3a857583233cb9 Mon Sep 17 00:00:00 2001 From: mischa Date: Sat, 27 Jun 2026 18:49:24 +0200 Subject: [PATCH] fix(ingest): don't advance incremental cursor past a failed/sampled asset A later successful asset could push last_ingest_at past an earlier asset whose thumbnail download failed, permanently excluding it from later incremental runs (only --full recovered it). Now the cursor never advances to/past the earliest failed asset, never below the prior cursor, and not at all on a --subset run. Regression tests added. (review finding #9) --- apps/trip-cluster/app/ingest.py | 23 ++++++++++++---- apps/trip-cluster/tests/test_ingest.py | 36 ++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 5 deletions(-) diff --git a/apps/trip-cluster/app/ingest.py b/apps/trip-cluster/app/ingest.py index 87e2ec6..c451b04 100644 --- a/apps/trip-cluster/app/ingest.py +++ b/apps/trip-cluster/app/ingest.py @@ -31,8 +31,9 @@ def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None, assets = assets[:subset] processed_marked = 0 - max_updated = updated_after or "" tag_counts: dict = {} + successful_updated: list = [] # updated_at of assets ingested this run + failed_updated: list = [] # updated_at of assets we could NOT ingest for a in assets: # Revision (2): sanitize the Immich-provided id before using it as a @@ -47,6 +48,7 @@ def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None, data = client.download_thumbnail(a["id"]) except Exception: log.exception("Thumbnail download failed for asset %s", a["id"]) + failed_updated.append(a["updated_at"]) continue # Write to a temp path and os.replace into place so a failure # never leaves a 0-byte .jpg behind; clean up the temp file @@ -62,6 +64,7 @@ def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None, os.unlink(tmp_path) except OSError: pass + failed_updated.append(a["updated_at"]) continue store.upsert_asset(Asset( @@ -77,13 +80,23 @@ def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None, store.mark_processed(a["id"]) processed_marked += 1 - if a["updated_at"] and a["updated_at"] > max_updated: - max_updated = a["updated_at"] + successful_updated.append(a["updated_at"]) for name, count in tag_counts.items(): store.upsert_tag(name, count=count) - if max_updated: - store.set_meta("last_ingest_at", max_updated) + # Advance the incremental cursor, but never past an asset we failed to ingest + # (a later success must not strand an earlier failure on the next run), never + # below the prior cursor, and not at all on a --subset sampling run. + if subset is None: + new_watermark = updated_after or "" + floor = min((u for u in failed_updated if u), default=None) + for u in successful_updated: + if not u or (floor is not None and u >= floor): + continue + if u > new_watermark: + new_watermark = u + if new_watermark: + store.set_meta("last_ingest_at", new_watermark) return {"fetched": len(assets), "processed_marked": processed_marked} diff --git a/apps/trip-cluster/tests/test_ingest.py b/apps/trip-cluster/tests/test_ingest.py index 25896b8..c0a7384 100644 --- a/apps/trip-cluster/tests/test_ingest.py +++ b/apps/trip-cluster/tests/test_ingest.py @@ -102,3 +102,39 @@ def test_ingest_resilient_to_thumb_failure_and_zero_byte(tmp_path): # The 0-byte file for "b" was replaced with real bytes. assert os.path.getsize(os.path.join(thumbs, "b.jpg")) > 0 store.close() + + +def test_ingest_watermark_not_advanced_past_failed_asset(tmp_path): + # A later successful asset must NOT advance last_ingest_at past an earlier + # asset whose thumbnail failed, or that asset is stranded on the next + # incremental run (review finding #9). + store = Store(str(tmp_path / "t.db")).connect() + thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True) + + class FailOld(FakeImmich): + def download_thumbnail(self, asset_id): + if asset_id == "old": + raise RuntimeError("boom") + return self._thumb + + client = FailOld([_asset("old", "2008-06-01", updated="2026-01-01T00:00:00Z"), + _asset("new", "2024-03-01", updated="2026-03-01T00:00:00Z")]) + run_ingest(client, store, thumbs) + assert store.get_asset("old") is None and store.get_asset("new") is not None + wm = store.get_meta("last_ingest_at") + # cursor must stay below the failed asset (here: not advanced at all) + assert wm is None or wm < "2026-01-01T00:00:00Z" + assert wm != "2026-03-01T00:00:00Z" + store.close() + + +def test_ingest_subset_run_does_not_advance_watermark(tmp_path): + # --subset is a sampling run; it must not move the incremental cursor or it + # would strand the un-fetched remainder. + store = Store(str(tmp_path / "t.db")).connect() + thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True) + client = FakeImmich([_asset("a", "2019-06-01", updated="2026-02-01T00:00:00Z"), + _asset("b", "2019-06-02", updated="2026-02-02T00:00:00Z")]) + run_ingest(client, store, thumbs, subset=1) + assert store.get_meta("last_ingest_at") is None + store.close()