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

@@ -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 <id>.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}