From 008dd0e7f4d1f57b60fbc73020f633b6ca2d4a31 Mon Sep 17 00:00:00 2001 From: Cameron Cordes Date: Fri, 30 Oct 2026 19:20:44 -0400 Subject: [PATCH] Page the version name sweep with a resume cursor (#194) Co-Authored-By: Claude Opus 5.5 --- src/database/version_dao.rs | 33 ++++++++--- src/versions/integration_tests.rs | 40 ++++++++++++++ src/versions/mod.rs | 91 +++++++++++++++++++++++++------ 3 files changed, 139 insertions(+), 25 deletions(-) diff --git a/src/database/version_dao.rs b/src/database/version_dao.rs index 355312a..6a4ba8f 100644 --- a/src/database/version_dao.rs +++ b/src/database/version_dao.rs @@ -230,12 +230,15 @@ pub trait VersionDao: Sync + Send { /// Rows in a library whose file name follows the version convention /// (`.[.].`), have a hash and aren't linked yet. - /// Returns `(rel_path, content_hash)`, oldest first. + /// Returns `(id, rel_path, content_hash)` ordered by row id, starting + /// after `after_id` (the caller's resume cursor, so rows that never + /// resolve can't starve the ones behind them). fn unlinked_name_candidates( &mut self, library_id: i32, + after_id: i32, limit: i64, - ) -> Result, DbError>; + ) -> Result, DbError>; /// Copy the original's date, GPS and camera fields onto every version /// row whose copy is stale. Returns the number of rows updated. @@ -723,11 +726,14 @@ impl VersionDao for SqliteVersionDao { fn unlinked_name_candidates( &mut self, library_id: i32, + after_id: i32, limit: i64, - ) -> Result, DbError> { + ) -> Result, DbError> { trace_db_call("query", "unlinked_name_candidates", |_span| { #[derive(QueryableByName)] struct Row { + #[diesel(sql_type = Integer)] + id: i32, #[diesel(sql_type = Text)] rel_path: String, #[diesel(sql_type = Text)] @@ -739,8 +745,8 @@ impl VersionDao for SqliteVersionDao { .collect::>() .join(" OR "); let sql = format!( - "SELECT rel_path, content_hash FROM image_exif \ - WHERE library_id = ? AND content_hash IS NOT NULL AND ({kinds}) \ + "SELECT id, rel_path, content_hash FROM image_exif \ + WHERE library_id = ? AND id > ? AND content_hash IS NOT NULL AND ({kinds}) \ AND NOT EXISTS (SELECT 1 FROM photo_versions pv \ WHERE pv.version_hash = image_exif.content_hash) \ ORDER BY id LIMIT ?" @@ -748,11 +754,12 @@ impl VersionDao for SqliteVersionDao { let mut conn = self.connection.lock().expect("Unable to get VersionDao"); sql_query(sql) .bind::(library_id) + .bind::(after_id) .bind::(limit) .load::(&mut *conn) .map(|rows| { rows.into_iter() - .map(|r| (r.rel_path, r.content_hash)) + .map(|r| (r.id, r.rel_path, r.content_hash)) .collect() }) .map_err(|e| anyhow::anyhow!("Query error: {}", e)) @@ -1089,8 +1096,18 @@ mod tests { insert_exif(&c, 2, "a/y.jpg.restored.png", "h4"); dao.link_version(&link("h2", "ox", "a/x.jpg.edit.2.png")) .unwrap_or(LinkOutcome::Linked); - let got = dao.unlinked_name_candidates(1, 100).unwrap(); - assert_eq!(got, vec![("a/x.jpg.restored.png".to_string(), "h1".into())]); + let got = dao.unlinked_name_candidates(1, 0, 100).unwrap(); + assert_eq!(got.len(), 1); + assert_eq!( + (got[0].1.as_str(), got[0].2.as_str()), + ("a/x.jpg.restored.png", "h1") + ); + // The cursor skips rows at or before it. + assert!( + dao.unlinked_name_candidates(1, got[0].0, 100) + .unwrap() + .is_empty() + ); } #[test] diff --git a/src/versions/integration_tests.rs b/src/versions/integration_tests.rs index 0ebae04..b3f27e6 100644 --- a/src/versions/integration_tests.rs +++ b/src/versions/integration_tests.rs @@ -253,6 +253,46 @@ async fn name_link_pairs_raw_and_jpeg_with_the_right_original() { ); } +#[actix_rt::test] +async fn unresolvable_name_candidates_do_not_starve_later_ones() { + let fx = fx(); + // Many matching rows with no original anywhere, then one linkable pair. + for i in 0..120 { + fx.put( + &format!("album_a/aiedits/orphan{i}.jpg.edit.png"), + format!("orphan {i}").as_bytes(), + ); + } + let original = fx.put("album_b/real.jpg", b"real original"); + let version = fx.put("album_b/aiedits/real.jpg.restored.png", b"real version"); + + let exif = fx.exif_arc(); + let mut cursor = 0; + let mut total = 0; + // One 50-row page per call: the old "ORDER BY id LIMIT n" would return + // the same 50 orphans forever. + for _ in 0..5 { + total += sweep_name_links_from( + &fx.library, + &exif, + &fx.state.version_dao, + &mut cursor, + 50, + 1, + ); + } + assert_eq!(total, 1); + let link = fx + .state + .version_dao + .lock() + .unwrap() + .get_link(&version) + .unwrap() + .unwrap(); + assert_eq!(link.original_hash, original); +} + #[actix_rt::test] async fn name_link_waits_for_the_original_to_be_hashed() { let fx = fx(); diff --git a/src/versions/mod.rs b/src/versions/mod.rs index c46359a..4b7a048 100644 --- a/src/versions/mod.rs +++ b/src/versions/mod.rs @@ -266,26 +266,83 @@ pub fn sweep_name_links( exif_dao: &Arc>>, version_dao: &Arc>>, ) -> usize { - let candidates = match version_dao + let mut cursor = name_sweep_cursors() .lock() - .expect("Unable to lock VersionDao") - .unlinked_name_candidates(library.id, NAME_SWEEP_LIMIT) - { - Ok(c) => c, - Err(e) => { - warn!( - "versions: name sweep failed for '{}': {:?}", - library.name, e - ); - return 0; - } - }; + .expect("name sweep cursors") + .get(&library.id) + .copied() + .unwrap_or(0); + let linked = sweep_name_links_from( + library, + exif_dao, + version_dao, + &mut cursor, + NAME_SWEEP_LIMIT, + NAME_SWEEP_PAGES, + ); + name_sweep_cursors() + .lock() + .expect("name sweep cursors") + .insert(library.id, cursor); + linked +} + +/// Pages examined per sweep call. +const NAME_SWEEP_PAGES: usize = 4; + +/// Per-library resume point (last candidate row id visited) for the name +/// sweep. Candidates that never resolve (no original in the library) stay +/// matching forever; paging past them keeps them from starving the rows +/// behind them. In memory only: a restart just rescans from the start. +fn name_sweep_cursors() -> &'static Mutex> { + static CURSORS: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + CURSORS.get_or_init(Default::default) +} + +/// One sweep: up to `max_pages` pages of `page` candidates from `cursor`, +/// advancing it and wrapping to 0 past the last candidate (then continuing +/// from the start, so a call never ends having examined nothing). +fn sweep_name_links_from( + library: &Library, + exif_dao: &Arc>>, + version_dao: &Arc>>, + cursor: &mut i32, + page: i64, + max_pages: usize, +) -> usize { let mut linked = 0; - for (rel, hash) in candidates { - if let Some(link) = name_link(library, &rel, &hash, exif_dao) - && record_link(version_dao, &link) + let mut wrapped = *cursor == 0; + for _ in 0..max_pages { + let candidates = match version_dao + .lock() + .expect("Unable to lock VersionDao") + .unlinked_name_candidates(library.id, *cursor, page) { - linked += 1; + Ok(c) => c, + Err(e) => { + warn!( + "versions: name sweep failed for '{}': {:?}", + library.name, e + ); + return linked; + } + }; + let short = (candidates.len() as i64) < page; + for (id, rel, hash) in candidates { + *cursor = id; + if let Some(link) = name_link(library, &rel, &hash, exif_dao) + && record_link(version_dao, &link) + { + linked += 1; + } + } + if short { + *cursor = 0; + if wrapped { + break; + } + wrapped = true; } } linked