fix(dates): resume cursor so unconvertible rows can't stall the #118 drain

Review follow-up: rows that exist but can't be faithfully resolved spent
the cap and were re-collected first every tick. Carry a per-library
cursor across ticks (wrapping at end of table), share the cap parser
between the two date drains, and name the conversion tuple.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CLAUDE.md

@@ -446,9 +446,11 @@ partial index `idx_image_exif_utc_backfill`) by re-running
 batch) and writing **only** the new-contract triple — the legacy pair
 stays frozen. Legacy `manual` rows convert via
 `camera_tz::manual_date_resolution` instead of being re-resolved from
-the file. Missing-file rows are paged past without spending the cap;
+the file. Missing-file rows are paged past without spending the cap, and
-EXIF/exiftool-legacy rows wait when exiftool is absent (they'd otherwise
+a per-library resume cursor (carried across ticks by the watcher, wraps
-fall to fs_time). When a converted row's display-zone day differs from
+at end of table) keeps unconvertible rows from stalling the rows behind
+them; EXIF/exiftool-legacy rows wait when exiftool is absent (they'd
+otherwise fall to fs_time). When a converted row's display-zone day differs from
 the day its legacy value rendered on, the `day`-span `precomputed_reels`
 rows covering either month-day (this library + `all`) are deleted and
 unlinked; the nightly pre-gen regenerates. Daily conversation summaries

src/backfill.rs

@@ -124,6 +124,16 @@ pub fn backfill_unhashed_backlog(
     backfilled
 }
 
+/// `DATE_BACKFILL_MAX_PER_TICK` (default 500): the per-tick cap shared by
+/// both date drains.
+fn date_backfill_cap() -> i64 {
+    dotenv::var("DATE_BACKFILL_MAX_PER_TICK")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .filter(|n: &i64| *n > 0)
+        .unwrap_or(500)
+}
+
 /// Drain image_exif rows whose `date_taken` was never resolved or was
 /// resolved by the weakest fallback (`fs_time`). Runs the canonical-date
 /// waterfall — exiftool batch (one subprocess for the whole tick's
@@ -142,11 +152,7 @@ pub fn backfill_missing_date_taken(
     library: &libraries::Library,
     exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
 ) -> usize {
-    let cap: i64 = dotenv::var("DATE_BACKFILL_MAX_PER_TICK")
+    let cap = date_backfill_cap();
-        .ok()
-        .and_then(|s| s.parse().ok())
-        .filter(|n: &i64| *n > 0)
-        .unwrap_or(500);
 
     let rows: Vec<(i32, String)> = {
         let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
@@ -241,6 +247,14 @@ lazy_static! {
 /// a full-table walk.
 const UTC_BACKFILL_SCAN_FACTOR: i64 = 20;
 
+/// One row's resolved new-contract triple, pending write.
+struct Conversion {
+    row: UtcBackfillCandidate,
+    date_utc: i64,
+    camera_offset_minutes: Option<i32>,
+    tz_source: camera_tz::TzSource,
+}
+
 /// What one tick of [`backfill_utc_dates`] did for one library.
 #[derive(Debug, Default, Clone, PartialEq, Eq)]
 pub struct UtcBackfillOutcome {
@@ -271,6 +285,14 @@ pub struct UtcBackfillOutcome {
 /// manual edit does today (`camera_tz::manual_date_resolution` in the
 /// display zone).
 ///
+/// `cursor` is the caller's per-library resume point (last row id
+/// visited), carried across ticks and wrapped to 0 at the end of the
+/// table. Rows that can't convert — missing files, no exiftool, no
+/// faithful resolution — are left behind the cursor instead of being
+/// re-collected first every tick, so they can never stall the rows after
+/// them (the "spun on the same lowest-id rows" failure of the legacy
+/// drain's first version).
+///
 /// Day-boundary invalidation: when a converted instant falls on a
 /// different display-zone calendar day than the legacy value did (as the
 /// legacy `/memories` SQL rendered it), the pre-generated `day` reels
@@ -283,13 +305,12 @@ pub fn backfill_utc_dates(
     exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
     reel_dao: &Arc<Mutex<Box<dyn PrecomputedReelDao>>>,
     display_tz: &str,
+    cursor: &mut i32,
 ) -> UtcBackfillOutcome {
-    let cap: i64 = dotenv::var("DATE_BACKFILL_MAX_PER_TICK")
+    let cap = date_backfill_cap();
-        .ok()
+    backfill_utc_dates_capped(
-        .and_then(|s| s.parse().ok())
+        context, library, exif_dao, reel_dao, display_tz, cursor, cap,
-        .filter(|n: &i64| *n > 0)
+    )
-        .unwrap_or(500);
-    backfill_utc_dates_capped(context, library, exif_dao, reel_dao, display_tz, cap)
 }
 
 fn backfill_utc_dates_capped(
@@ -298,6 +319,7 @@ fn backfill_utc_dates_capped(
     exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
     reel_dao: &Arc<Mutex<Box<dyn PrecomputedReelDao>>>,
     display_tz: &str,
+    cursor: &mut i32,
     cap: i64,
 ) -> UtcBackfillOutcome {
     let mut outcome = UtcBackfillOutcome::default();
@@ -319,22 +341,21 @@ fn backfill_utc_dates_capped(
     // them for a deploy that has it.
     let exiftool_ok = date_resolver::exiftool_available();
 
-    // Page by id until `cap` convertible rows are collected. Rows that
+    // Page by id from the resume cursor until `cap` convertible rows are
-    // can't convert this tick are paged past, not counted against the
+    // collected. Rows that can't convert this tick are paged past, not
-    // cap, so a pile of missing files at low ids can't stall the drain.
+    // counted against the cap.
     let mut manual: Vec<UtcBackfillCandidate> = Vec::new();
     let mut filed: Vec<(UtcBackfillCandidate, PathBuf)> = Vec::new();
-    let mut cursor = 0;
     let mut scanned = 0i64;
     'paging: loop {
         let page = {
             let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
-            dao.get_rows_needing_utc_backfill(context, library.id, cursor, cap)
+            dao.get_rows_needing_utc_backfill(context, library.id, *cursor, cap)
                 .unwrap_or_default()
         };
         let page_len = page.len() as i64;
         for row in page {
-            cursor = row.id;
+            *cursor = row.id;
             scanned += 1;
             match row.date_taken_source.as_deref() {
                 Some("manual") => manual.push(row),
@@ -353,20 +374,27 @@ fn backfill_utc_dates_capped(
                 break 'paging;
             }
         }
-        if page_len < cap || scanned >= cap * UTC_BACKFILL_SCAN_FACTOR {
+        if page_len < cap {
+            // End of the table: wrap so next tick revisits what was left
+            // behind (and anything that became convertible since).
+            *cursor = 0;
+            break;
+        }
+        if scanned >= cap * UTC_BACKFILL_SCAN_FACTOR {
             break;
         }
     }
     let more_remain = (manual.len() + filed.len()) as i64 >= cap;
 
-    // (candidate, date_utc, camera_offset_minutes, tz_source)
+    let mut conversions: Vec<Conversion> = Vec::with_capacity(manual.len() + filed.len());
-    let mut conversions: Vec<(UtcBackfillCandidate, i64, Option<i32>, &'static str)> =
-        Vec::with_capacity(manual.len() + filed.len());
     for row in manual {
         match camera_tz::manual_date_resolution(Some(&zone_name), row.date_taken) {
-            Some((utc, offset)) => {
+            Some((utc, offset)) => conversions.push(Conversion {
-                conversions.push((row, utc, Some(offset), camera_tz::TzSource::Manual.as_str()))
+                row,
-            }
+                date_utc: utc,
+                camera_offset_minutes: Some(offset),
+                tz_source: camera_tz::TzSource::Manual,
+            }),
             None => outcome.unresolved += 1,
         }
     }
@@ -383,9 +411,12 @@ fn backfill_utc_dates_capped(
             match resolved.get(&abs) {
                 Some(rd) if !(legacy_had_tag && rd.source == date_resolver::DateSource::FsTime) => {
                     match (rd.date_utc, rd.tz_source) {
-                        (Some(utc), Some(tz)) => {
+                        (Some(utc), Some(tz)) => conversions.push(Conversion {
-                            conversions.push((row, utc, rd.camera_offset_minutes, tz.as_str()))
+                            row,
-                        }
+                            date_utc: utc,
+                            camera_offset_minutes: rd.camera_offset_minutes,
+                            tz_source: tz,
+                        }),
                         _ => outcome.unresolved += 1,
                     }
                 }
@@ -406,8 +437,14 @@ fn backfill_utc_dates_capped(
     let mut affected_days: HashSet<(u32, u32)> = HashSet::new();
     {
         let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
-        for (row, utc, offset, tz_source) in &conversions {
+        for Conversion {
-            match dao.backfill_date_utc(context, row.id, *utc, *offset, tz_source) {
+            row,
+            date_utc: utc,
+            camera_offset_minutes: offset,
+            tz_source,
+        } in &conversions
+        {
+            match dao.backfill_date_utc(context, row.id, *utc, *offset, tz_source.as_str()) {
                 Ok(true) => {
                     outcome.converted += 1;
                     let before = display_day(zone, row.date_taken);
@@ -455,9 +492,10 @@ fn display_day(zone: chrono_tz::Tz, epoch: i64) -> Option<NaiveDate> {
 }
 
 /// The month-day an "on this day" reel covers: the date it was generated
-/// on, in the zone offset it was generated with. A pre-gen fallback path
+/// on, in the zone offset it was generated with. The pre-gen fallback
-/// has stored seconds rather than minutes; an out-of-range offset falls
+/// `reels::fixed_tz_offset` (no prefs row, no env override) stores
-/// back to the display zone.
+/// `local_minus_utc()` — seconds, not minutes — so an out-of-range offset
+/// falls back to the display zone.
 fn reel_covered_day(reel: &PrecomputedReel, zone: chrono_tz::Tz) -> Option<NaiveDate> {
     if reel.tz_offset_minutes.abs() <= 14 * 60 {
         DateTime::from_timestamp(reel.generated_at + reel.tz_offset_minutes as i64 * 60, 0)
@@ -1347,7 +1385,8 @@ mod tests {
         );
 
         assert_eq!(remaining(&exif_dao, 1), 5);
-        let out = backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, 500);
+        let out =
+            backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, &mut 0, 500);
         assert_eq!(out.converted, 5, "{out:?}");
         assert_eq!(out.unresolved, 0, "{out:?}");
 
@@ -1407,7 +1446,8 @@ mod tests {
                 .collect::<Vec<_>>()
         };
         let before = snapshot(&conn);
-        let again = backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, 500);
+        let again =
+            backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, &mut 0, 500);
         assert_eq!(again, UtcBackfillOutcome::default());
         assert_eq!(snapshot(&conn), before);
         let mut progress = HashMap::new();
@@ -1427,7 +1467,8 @@ mod tests {
         let undated = insert_legacy(&exif_dao, 1, "undated.jpg", None, None);
         let other = insert_legacy(&exif_dao, 2, "1500000000.jpg", Some(ts), Some("filename"));
 
-        let out = backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, 500);
+        let out =
+            backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, &mut 0, 500);
         assert_eq!(out.converted, 1);
         assert_eq!(row(&conn, converted).date_utc, Some(ts));
         assert_eq!(row(&conn, undated).date_utc, None);
@@ -1464,8 +1505,18 @@ mod tests {
 
         let mut progress = HashMap::new();
         let mut seen = Vec::new();
+        // Cursor carried across ticks, as the watcher does.
+        let mut cursor = 0;
         for _ in 0..4 {
-            let out = backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, 2);
+            let out = backfill_utc_dates_capped(
+                &ctx(),
+                &library,
+                &exif_dao,
+                &reel_dao,
+                NY,
+                &mut cursor,
+                2,
+            );
             refresh_utc_backfill_progress(&ctx(), &library, &exif_dao, &mut progress);
             seen.push((out.converted, progress[&1]));
         }
@@ -1485,6 +1536,55 @@ mod tests {
         assert_eq!(progress[&1], 0);
     }
 
+    #[test]
+    fn utc_backfill_unresolvable_rows_cannot_stall_the_rows_behind_them() {
+        let (tmp, library, conn, exif_dao, _face_dao) = setup();
+        let reel_dao = reel_dao_for(&conn);
+        // Present files whose legacy date came from EXIF but that carry no
+        // date anywhere now: the waterfall would only find fs_time, which
+        // the drain refuses to write over a tag date. More of them than
+        // the cap, all at the lowest ids.
+        for i in 0..4 {
+            let rel = format!("bad{i}.jpg");
+            write_image(tmp.path(), &rel, b"no-exif");
+            insert_legacy(&exif_dao, 1, &rel, Some(1_000_000_000), Some("exif"));
+        }
+        let mut good = Vec::new();
+        for i in 0..2 {
+            let ts = 1_500_000_000 + i;
+            let rel = format!("{ts}.jpg");
+            write_image(tmp.path(), &rel, b"z");
+            good.push(insert_legacy(
+                &exif_dao,
+                1,
+                &rel,
+                Some(ts),
+                Some("filename"),
+            ));
+        }
+
+        let mut cursor = 0;
+        let mut converted = 0;
+        for _ in 0..4 {
+            let out = backfill_utc_dates_capped(
+                &ctx(),
+                &library,
+                &exif_dao,
+                &reel_dao,
+                NY,
+                &mut cursor,
+                2,
+            );
+            converted += out.converted;
+        }
+        assert_eq!(converted, 2, "the rows behind the stuck ones still drain");
+        for id in good {
+            assert!(row(&conn, id).date_utc.is_some());
+        }
+        // The unresolvable rows stay pending (visible in progress), untouched.
+        assert_eq!(remaining(&exif_dao, 1), 4);
+    }
+
     #[test]
     fn utc_backfill_skips_unavailable_library_then_resumes() {
         let (tmp, library, conn, exif_dao, _face_dao) = setup();
@@ -1514,14 +1614,16 @@ mod tests {
         let health_map = libraries::new_health_map(std::slice::from_ref(&unmounted));
         assert!(!libraries::refresh_health(&health_map, &unmounted, true).is_online());
         // …and the drain itself converts nothing even if called.
-        let out = backfill_utc_dates_capped(&ctx(), &unmounted, &exif_dao, &reel_dao, NY, 500);
+        let out =
+            backfill_utc_dates_capped(&ctx(), &unmounted, &exif_dao, &reel_dao, NY, &mut 0, 500);
         assert_eq!(out, UtcBackfillOutcome::default());
         assert_eq!(row(&conn, filed).date_utc, None);
         assert_eq!(row(&conn, manual).date_utc, None);
 
         // Share returns → picked up next tick.
         assert!(libraries::refresh_health(&health_map, &library, true).is_online());
-        let out = backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, 500);
+        let out =
+            backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, &mut 0, 500);
         assert_eq!(out.converted, 2);
         assert_eq!(remaining(&exif_dao, 1), 0);
     }
@@ -1602,7 +1704,8 @@ mod tests {
         let (other_day, mp4_0301) = seed_reel(&reel_dao, reels_dir.path(), "day", "1", 3, 1);
         let (week, _) = seed_reel(&reel_dao, reels_dir.path(), "week", "1", 6, 12);
 
-        let out = backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, 500);
+        let out =
+            backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, &mut 0, 500);
         assert_eq!(out.converted, 2);
         assert_eq!(out.day_changed, 1, "{out:?}");
         assert_eq!(out.reels_invalidated, 2, "{out:?}");
@@ -1646,7 +1749,8 @@ mod tests {
         seed_reel(&reel_dao, reels_dir.path(), "day", "1", 6, 11);
         seed_reel(&reel_dao, reels_dir.path(), "day", "all", 6, 12);
 
-        let out = backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, 500);
+        let out =
+            backfill_utc_dates_capped(&ctx(), &library, &exif_dao, &reel_dao, NY, &mut 0, 500);
         assert_eq!(out.converted, 1);
         assert_eq!(out.day_changed, 0);
         assert_eq!(out.reels_invalidated, 0);

src/watcher.rs

@@ -349,6 +349,9 @@ pub fn watch_files(
         // Last-known per-library remaining count behind the
         // `imageserver_date_backfill_remaining` gauge.
         let mut utc_backfill_remaining: HashMap<i32, i64> = HashMap::new();
+        // Per-library resume cursor for the true-instant drain; in-memory,
+        // so a restart resumes from 0 (harmless — the drain is idempotent).
+        let mut utc_backfill_cursors: HashMap<i32, i32> = HashMap::new();
 
         let mut last_quick_scan = SystemTime::now();
         // Initialize to UNIX_EPOCH so the *first* tick is treated as a
@@ -530,6 +533,7 @@ pub fn watch_files(
                         &exif_dao,
                         &precomputed_reel_dao,
                         &display_tz,
+                        utc_backfill_cursors.entry(lib.id).or_insert(0),
                     );
                     backfill::refresh_utc_backfill_progress(
                         &context,