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>
This commit is contained in:
1 parent
1fad08dea3
commit
b7b59c1ba2
3 files changed
+153
-43
No files matched your search
@@ -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;
|
||||
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 file. Missing-file rows are paged past without spending the cap, and
|
||||
a per-library resume cursor (carried across ticks by the watcher, wraps
|
||||
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
|
||||
|
||||
+144
-40
@@ -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")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.filter(|n: &i64| *n > 0)
|
||||
.unwrap_or(500);
|
||||
let cap = date_backfill_cap();
|
||||
|
||||
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")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.filter(|n: &i64| *n > 0)
|
||||
.unwrap_or(500);
|
||||
backfill_utc_dates_capped(context, library, exif_dao, reel_dao, display_tz, cap)
|
||||
let cap = date_backfill_cap();
|
||||
backfill_utc_dates_capped(
|
||||
context, library, exif_dao, reel_dao, display_tz, cursor, 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
|
||||
// can't convert this tick are paged past, not counted against the
|
||||
// cap, so a pile of missing files at low ids can't stall the drain.
|
||||
// Page by id from the resume cursor until `cap` convertible rows are
|
||||
// collected. Rows that can't convert this tick are paged past, not
|
||||
// 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<(UtcBackfillCandidate, i64, Option<i32>, &'static str)> =
|
||||
Vec::with_capacity(manual.len() + filed.len());
|
||||
let mut conversions: Vec<Conversion> = 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)) => {
|
||||
conversions.push((row, utc, Some(offset), camera_tz::TzSource::Manual.as_str()))
|
||||
}
|
||||
Some((utc, offset)) => conversions.push(Conversion {
|
||||
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)) => {
|
||||
conversions.push((row, utc, rd.camera_offset_minutes, tz.as_str()))
|
||||
}
|
||||
(Some(utc), Some(tz)) => conversions.push(Conversion {
|
||||
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 {
|
||||
match dao.backfill_date_utc(context, row.id, *utc, *offset, tz_source) {
|
||||
for Conversion {
|
||||
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
|
||||
/// has stored seconds rather than minutes; an out-of-range offset falls
|
||||
/// back to the display zone.
|
||||
/// on, in the zone offset it was generated with. The pre-gen fallback
|
||||
/// `reels::fixed_tz_offset` (no prefs row, no env override) stores
|
||||
/// `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);
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user