Keep a run marker under the detector that found the faces

`record_updates` — the write behind the quality, eye and crop passes —
re-marked the image as indexed under the pipeline the pass ran as, and
left the faces it had updated under the id of the detector that found
them. On a desktop set to Thorough that put `scrfd_10g+w600k_mbf` over
faces spelled `w600k_mbf`; on the tablet, `scrfd_10g_i8+w600k_mbf` over
faces it had adopted from the desktop's thorough pass.

Every reader takes the marker and the faces to agree. `marker_under`
reads the marker as the detector having examined the image, so the
upgrade repair never revisits it. The shard store keys each face by
its pipeline id, so `export_to_shards` selects an image's faces by the
marker's id, finds none, and sends an entry that says the thorough
detector looked and found nothing — over photographs with named faces
on them. The desktop's shard index holds 54 such entries beside real
faces; the tablet's eye pass over the faces it had adopted made 430
more, and both devices have exchanged them. `held_model` takes the
newest entry for an image, which is the empty one. Nothing has been
lost yet only because the two spellings of the thorough detector rank
equal and neither side adopts the other's; a third device, or either
one after a reinstall, would adopt "nothing here" for 484 images. And
the desktop's eye pass is 4,739 images from doing the same to every
face from before V14 — which are the ones that only exist on the
desktop, and would then never reach anywhere.

The marker now takes the id the faces carry; the pass's own id is used
only when it dropped the last of them and there is no detector left to
name. A stale marker under another spelling of the same embedder is
removed in the same transaction, so one embedder has one marker.

core/dr-catalog/src/faces.rs

@@ -556,6 +556,19 @@ impl FaceUpdate {
 /// bookkeeping: `face_shard::export_to_shards` re-exports an image whose
 /// marker is newer than the store's copy, which is how what was written
 /// here reaches the other devices.
+///
+/// It is re-written under the pipeline id the **faces carry**, not the one
+/// this pass ran as. `model_id` names the pass only through its embedder;
+/// the detector half of a marker is a statement about who drew the boxes,
+/// and this pass drew none. Every reader takes the two to agree: the export
+/// selects an image's faces by the marker's id, `marker_under` takes a
+/// marker as proof the detector has been over the image, and the shard
+/// store keys each face by it. When the marker was written as
+/// `scrfd_10g+w600k_mbf` over faces still spelled `w600k_mbf`, the export
+/// found no faces under it and sent the other devices an entry saying the
+/// thorough detector had looked and found nothing — over photographs with
+/// named faces on them. With no faces left, the pass's own id is the only
+/// one there is, and the marker says so.
 pub fn record_updates(
     conn: &Connection,
     image_id: ImageId,
@@ -602,13 +615,25 @@ pub fn record_updates(
     for f in dropped {
         tx.execute("DELETE FROM faces WHERE id = ?1", [f.0 as i64])?;
     }
-    let remaining: i64 = tx.query_row(
+    let (remaining, found_by): (i64, Option<String>) = tx.query_row(
         &format!(
-            "SELECT COUNT(*) FROM faces WHERE image_id = ?1 AND {} = ?2",
+            "SELECT COUNT(*), MIN(model_id) FROM faces WHERE image_id = ?1 AND {} = ?2",
             embedder_sql("model_id")
         ),
         rusqlite::params![image_id.0 as i64, embedder_of(model_id)],
-        |r| r.get(0),
+        |r| Ok((r.get(0)?, r.get(1)?)),
+    )?;
+    let marker = found_by.as_deref().unwrap_or(model_id);
+    // One marker per embedder: a stale one under another spelling would
+    // keep saying that detector had been here, which is the claim the
+    // faces' own id is now making in its place.
+    tx.execute(
+        &format!(
+            "DELETE FROM face_index
+              WHERE image_id = ?1 AND model_id != ?2 AND {} = ?3",
+            embedder_sql("model_id")
+        ),
+        rusqlite::params![image_id.0 as i64, marker, embedder_of(model_id)],
     )?;
     tx.execute(
         "INSERT INTO face_index (image_id, model_id, indexed_at, faces_found, source_edge)
@@ -619,7 +644,7 @@ pub fn record_updates(
              source_edge = excluded.source_edge",
         rusqlite::params![
             image_id.0 as i64,
-            model_id,
+            marker,
             now_secs(),
             remaining,
             source_edge as i64,
@@ -1557,7 +1582,7 @@ fn iou(a: (f32, f32, f32, f32), b: (f32, f32, f32, f32)) -> f32 {
     }
 }
 
-fn now_secs() -> i64 {
+pub(crate) fn now_secs() -> i64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .map(|d| d.as_secs() as i64)
@@ -1890,6 +1915,86 @@ mod tests {
         assert!(at >= marked_at, "the marker was not refreshed");
     }
 
+    /// The marker a per-face pass leaves names the detector that drew the
+    /// boxes, whatever pipeline the pass itself ran as. A marker under the
+    /// pass's id over faces spelled another way is one the export finds no
+    /// faces under — and it sent every other device "nothing here".
+    #[test]
+    fn an_update_keeps_the_marker_under_the_detector_that_found_the_faces() {
+        let c = db();
+        let img = image(&c, 1);
+        let ids = record_detections(
+            &c,
+            img,
+            "w600k_mbf",
+            1024,
+            &[DetectedFace {
+                quality: None,
+                ..face(1)
+            }],
+        )
+        .unwrap();
+        // The state V14 leaves: the faces, and no marker at all.
+        c.execute("DELETE FROM face_index", []).unwrap();
+
+        record_updates(
+            &c,
+            img,
+            "scrfd_10g+w600k_mbf",
+            6000,
+            &[FaceUpdate {
+                embedding: Some((vec![9; 1024], 21.5)),
+                ..FaceUpdate::for_face(ids[0])
+            }],
+            &[],
+        )
+        .unwrap();
+
+        let markers: Vec<(String, i64)> = c
+            .prepare("SELECT model_id, faces_found FROM face_index")
+            .unwrap()
+            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
+            .unwrap()
+            .map(Result::unwrap)
+            .collect();
+        assert_eq!(markers, vec![("w600k_mbf".to_string(), 1)]);
+
+        // A marker already there under the pass's own id is replaced, not
+        // kept beside the right one.
+        c.execute(
+            "INSERT INTO face_index(image_id, model_id, indexed_at, faces_found, source_edge)
+             VALUES (1, 'scrfd_10g+w600k_mbf', 0, 0, 6000)",
+            [],
+        )
+        .unwrap();
+        record_updates(
+            &c,
+            img,
+            "scrfd_10g+w600k_mbf",
+            6000,
+            &[FaceUpdate {
+                crop: Some(vec![1, 2, 3]),
+                ..FaceUpdate::for_face(ids[0])
+            }],
+            &[],
+        )
+        .unwrap();
+        let n: i64 = c
+            .query_row("SELECT COUNT(*) FROM face_index", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(n, 1, "a second marker survived");
+
+        // With every face dropped there is no detector left to name, and
+        // the pass's own id records that it looked.
+        record_updates(&c, img, "scrfd_10g+w600k_mbf", 6000, &[], &ids).unwrap();
+        let marker: (String, i64) = c
+            .query_row("SELECT model_id, faces_found FROM face_index", [], |r| {
+                Ok((r.get(0)?, r.get(1)?))
+            })
+            .unwrap();
+        assert_eq!(marker, ("scrfd_10g+w600k_mbf".to_string(), 0));
+    }
+
     /// Re-detection is coalesced per image, so it must replace rather than
     /// append — otherwise every re-index doubles the library's face count.
     #[test]