diff --git a/core/dr-catalog/src/face_shard.rs b/core/dr-catalog/src/face_shard.rs index c6aedbd..1354895 100644 --- a/core/dr-catalog/src/face_shard.rs +++ b/core/dr-catalog/src/face_shard.rs @@ -474,14 +474,26 @@ impl FaceShardStore { rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX, )?; - let mut q = - src.prepare("SELECT file_id, model_id, faces_found, source_edge FROM indexed")?; - let images: Vec<(i64, String, i64, i64)> = q - .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))? + // The peer's marker travels with the image: it is what lets + // `import_from_shards` record the adoption under the time the peer + // indexed it, and so what keeps `export_to_shards` from reading the + // adoption as a re-index and sending the peer's faces back out under + // this device's name. A shard from before the column has none. + let mut q = src.prepare(&format!( + "SELECT file_id, model_id, faces_found, source_edge, {} FROM indexed", + match has_column(&src, "indexed", "indexed_at") { + Ok(true) => "indexed_at", + _ => "NULL", + } + ))?; + let images: Vec<(i64, String, i64, i64, Option)> = q + .query_map([], |r| { + Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)) + })? .collect::>()?; let mut adopted = 0; - for (file_id, model_id, _found, edge) in images { + for (file_id, model_id, _found, edge, indexed_at) in images { if self.contains(file_id as u64, &model_id) { continue; } @@ -501,12 +513,22 @@ impl FaceShardStore { let faces: Vec = fq .query_map(rusqlite::params![file_id, &model_id], read_shared_face)? .collect::>()?; - self.put_image(file_id as u64, &model_id, edge as u32, &faces)?; + self.put_image_at(file_id as u64, &model_id, edge as u32, &faces, indexed_at)?; adopted += 1; } Ok(adopted) } + /// Record when the catalog indexed a held image, for an entry that + /// arrived without a marker — a peer's shard from before the column. + pub fn set_indexed_at(&self, file_id: u64, model_id: &str, at: i64) -> Result<(), CatalogError> { + self.index.execute( + "UPDATE entries SET indexed_at = ?3 WHERE file_id = ?1 AND model_id = ?2", + rusqlite::params![file_id as i64, model_id, at], + )?; + Ok(()) + } + /// Read back everything held for one image. pub fn get_image( &self, @@ -865,6 +887,31 @@ pub fn import_from_shards( edge, &local, )?; + // The peer's marker, not this moment. `record_detections` stamps the + // run as now, and `export_to_shards` reads a marker newer than the + // shard's as a re-index — so every adopted image went straight back + // out as this device's own work: 14,100 adopted, 15,457 "newly + // indexed" on the next pass, and twenty-two shards of a peer's faces + // uploaded again under a second name. Where the peer's shard carried + // no marker, the store takes the catalog's, so the two agree either + // way and the export sees nothing to send. + match store.indexed_at(file_id as u64, &held) { + Some(theirs) => { + conn.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( + "SELECT indexed_at FROM face_index WHERE image_id = ?1 AND model_id = ?2", + rusqlite::params![image_id, held], + |r| r.get(0), + )?; + store.set_indexed_at(file_id as u64, &held, ours)?; + } + } adopted += 1; } Ok(adopted) @@ -1322,6 +1369,11 @@ mod catalog_round_trip { assert!((got[0].landmarks[2].0 - 0.15).abs() < 1e-5); let emb = faces::embeddings(&b, "w600k_mbf").unwrap(); assert!(emb.iter().any(|e| e.embedding[0] == 1)); + + // And what B adopted is not B's work: its next export sends nothing. + // Adopting used to stamp the run as now, so every adopted image went + // back out under B's name as a re-index. + assert_eq!(export_to_shards(&b, &mut store_b, "w600k_mbf").unwrap(), 0); } /// The desktop switched to a stronger detector part-way through the