Page the version name sweep with a resume cursor (#194)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
9191257fbb
commit
008dd0e7f4
3 files changed
+139
-25
No files matched your search
@@ -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();
|
||||
|
||||
+74
-17
@@ -266,26 +266,83 @@ pub fn sweep_name_links(
|
||||
exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
|
||||
version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
|
||||
) -> 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<std::collections::HashMap<i32, i32>> {
|
||||
static CURSORS: std::sync::OnceLock<Mutex<std::collections::HashMap<i32, i32>>> =
|
||||
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<Mutex<Box<dyn ExifDao>>>,
|
||||
version_dao: &Arc<Mutex<Box<dyn VersionDao>>>,
|
||||
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
|
||||
|
||||
Reference in new issue
Block a user