diff --git a/core/dr-catalog/src/face_shard.rs b/core/dr-catalog/src/face_shard.rs index ca20cdb..3bd52df 100644 --- a/core/dr-catalog/src/face_shard.rs +++ b/core/dr-catalog/src/face_shard.rs @@ -178,6 +178,67 @@ impl FaceShardStore { .flatten() } + /// The other pipelines this file is held under that share `model_id`'s + /// embedder — the generations a put of `model_id` may supersede. + fn siblings(&self, file_id: u64, model_id: &str) -> Vec { + let mut stmt = match self.index.prepare(&format!( + "SELECT model_id FROM entries + WHERE file_id = ?1 AND model_id != ?2 AND {} = ?3", + crate::faces::embedder_sql("model_id") + )) { + Ok(s) => s, + Err(_) => return Vec::new(), + }; + stmt.query_map( + rusqlite::params![ + file_id as i64, + model_id, + crate::faces::embedder_of(model_id) + ], + |r| r.get::<_, String>(0), + ) + .map(|rows| rows.filter_map(|r| r.ok()).collect()) + .unwrap_or_default() + } + + /// Whether a pass this file is already held under outranks `model_id`, + /// so a put of `model_id` would add a generation nobody would adopt. + pub fn outranked(&self, file_id: u64, model_id: &str) -> bool { + use dr_types::FaceDetector; + let Some(incoming) = FaceDetector::for_model_id(model_id) else { + return false; + }; + self.siblings(file_id, model_id) + .iter() + .filter_map(|m| FaceDetector::for_model_id(m)) + .any(|held| held.outranks(incoming)) + } + + /// Forget the index entries for generations of this file that `model_id` + /// outranks. The bytes stay where they are — a sealed shard is + /// immutable — but the store stops offering them, and a later export or + /// merge writes nothing for them again. + fn supersede(&self, file_id: u64, model_id: &str) -> Result<(), CatalogError> { + use dr_types::FaceDetector; + let Some(incoming) = FaceDetector::for_model_id(model_id) else { + return Ok(()); + }; + for held in self.siblings(file_id, model_id) { + let weaker = FaceDetector::for_model_id(&held).is_some_and(|h| incoming.outranks(h)); + if weaker { + self.index.execute( + "DELETE FROM entries WHERE file_id = ?1 AND model_id = ?2", + rusqlite::params![file_id as i64, held], + )?; + self.index.execute( + "DELETE FROM faces_meta WHERE file_id = ?1 AND model_id = ?2", + rusqlite::params![file_id as i64, held], + )?; + } + } + Ok(()) + } + pub fn contains(&self, file_id: u64, model_id: &str) -> bool { self.index .query_row( @@ -233,6 +294,15 @@ impl FaceShardStore { faces: &[SharedFace], indexed_at: Option, ) -> Result { + // One generation per image per embedder. A store carried every pass + // — 24,123 entries for 19,089 images on the reference library, a + // third of its 293 MB — and only the strongest was ever adopted. + // A weaker pass arriving after a stronger one is not written; a + // stronger one arriving retires the weaker from the index. + if self.outranked(file_id, model_id) { + return Ok(0); + } + self.supersede(file_id, model_id)?; let incoming = faces .iter() .map(|f| BYTES_PER_FACE + if f.crop.is_empty() { 0 } else { BYTES_PER_CROP }) @@ -494,7 +564,8 @@ impl FaceShardStore { let mut adopted = 0; for (file_id, model_id, _found, edge, indexed_at) in images { - if self.contains(file_id as u64, &model_id) { + if self.contains(file_id as u64, &model_id) || self.outranked(file_id as u64, &model_id) + { continue; } let mut fq = src.prepare(&format!( @@ -825,8 +896,21 @@ pub fn import_from_shards( })? .collect::>()?; + /// Images per write transaction. Large enough that fourteen thousand + /// adoptions are a hundred and forty commits rather than fourteen + /// thousand; small enough that a read on the UI thread, queued behind + /// the lock, waits a fraction of a second and not the whole import. + const CHUNK: usize = 100; + let mut adopted = 0; + let mut tx = conn.unchecked_transaction()?; + let mut in_chunk = 0; for (file_id, image_id, local) in candidates { + if in_chunk == CHUNK { + tx.commit()?; + tx = conn.unchecked_transaction()?; + in_chunk = 0; + } let Some(held) = store.held_model(file_id as u64, model_id) else { continue; }; @@ -885,8 +969,8 @@ pub fn import_from_shards( }) .collect(); - crate::faces::record_detections( - conn, + crate::faces::record_detections_within( + &tx, dr_types::ImageId(image_id as u64), &held, edge, @@ -902,14 +986,14 @@ pub fn import_from_shards( // way and the export sees nothing to send. match store.indexed_at(file_id as u64, &held) { Some(theirs) => { - conn.execute( + tx.execute( "UPDATE face_index SET indexed_at = ?3 WHERE image_id = ?1 AND model_id = ?2", rusqlite::params![image_id, held, theirs], )?; } None => { - let ours: i64 = conn.query_row( + let ours: i64 = tx.query_row( "SELECT indexed_at FROM face_index WHERE image_id = ?1 AND model_id = ?2", rusqlite::params![image_id, held], |r| r.get(0), @@ -918,7 +1002,9 @@ pub fn import_from_shards( } } adopted += 1; + in_chunk += 1; } + tx.commit()?; Ok(adopted) } @@ -1154,6 +1240,32 @@ mod tests { assert!(!s.contains(1, "lvface")); } + /// One generation per image per embedder: a stronger detector's pass + /// retires a weaker one from the index, and a weaker pass arriving after + /// a stronger is not written at all. + #[test] + fn a_stronger_pass_retires_a_weaker_one_and_a_weaker_is_not_added() { + let dir = tempdir(); + let mut s = FaceShardStore::open(&dir).unwrap(); + s.put_image(1, "w600k_mbf", 1024, &[face(1, 1)]).unwrap(); + s.put_image(1, "scrfd_10g+w600k_mbf", 1024, &[face(1, 2)]) + .unwrap(); + assert!(s.contains(1, "scrfd_10g+w600k_mbf")); + assert!(!s.contains(1, "w600k_mbf"), "the fast pass was not retired"); + assert_eq!(s.len(), 1, "faces_meta still counts the retired pass"); + + s.put_image(1, "scrfd_2.5g+w600k_mbf", 1024, &[face(1, 3)]) + .unwrap(); + assert!( + !s.contains(1, "scrfd_2.5g+w600k_mbf"), + "a weaker pass was added" + ); + assert_eq!( + s.held_model(1, "w600k_mbf").as_deref(), + Some("scrfd_10g+w600k_mbf") + ); + } + #[test] fn re_storing_an_image_replaces_rather_than_doubling_it() { let dir = tempdir(); diff --git a/core/dr-catalog/src/faces.rs b/core/dr-catalog/src/faces.rs index cb2af93..0c094ab 100644 --- a/core/dr-catalog/src/faces.rs +++ b/core/dr-catalog/src/faces.rs @@ -279,7 +279,25 @@ pub fn record_detections( faces: &[DetectedFace], ) -> Result, CatalogError> { let tx = conn.unchecked_transaction()?; + let ids = record_detections_within(&tx, image_id, model_id, source_edge, faces)?; + tx.commit()?; + Ok(ids) +} +/// [`record_detections`] inside a transaction the caller owns. +/// +/// For a caller recording many images at once — the shard import adopts +/// fourteen thousand in one pass — where a commit per image is fourteen +/// thousand fsyncs and fourteen thousand turns at the write lock that every +/// read on the UI thread queues behind. `unchecked_transaction` cannot nest, +/// so the batching has to be offered here rather than wrapped from above. +pub fn record_detections_within( + tx: &Connection, + image_id: ImageId, + model_id: &str, + source_edge: u32, + faces: &[DetectedFace], +) -> Result, CatalogError> { // Everything the old faces knew, so it can be carried across the // replacement. Read only when there is something to carry it onto: a // pass that found nothing has nothing to match, and decoding a vector @@ -287,7 +305,7 @@ pub fn record_detections( let prior = if faces.is_empty() { Vec::new() } else { - read_priors(&tx, image_id)? + read_priors(tx, image_id)? }; tx.execute("DELETE FROM faces WHERE image_id = ?1", [image_id.0 as i64])?; @@ -389,7 +407,6 @@ pub fn record_detections( ], )?; - tx.commit()?; Ok(ids) }