Merge: repair shards that predate the crop and index-time columns

The schema batch is all CREATE IF NOT EXISTS, so two columns added to it never
reached any shard already on disk — and every export into one failed on "no
such column", quietly, logged at warn while the sync reported success. That,
rather than anything in the export logic, is why this library's shard stayed at
1,807 faces while the catalog reached 15,194.

Shards are now upgraded when opened, and a peer's read-only shard is read as it
stands.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

core/dr-catalog/src/face_shard.rs

@@ -311,6 +311,7 @@ impl FaceShardStore {
         }
         let conn = Connection::open(&path)?;
         conn.execute_batch(SHARD_SCHEMA)?;
+        upgrade_shard(&conn)?;
         Ok(conn)
     }
 
@@ -380,11 +381,12 @@ impl FaceShardStore {
             if self.contains(file_id as u64, &model_id) {
                 continue;
             }
-            let mut fq = src.prepare(
-                "SELECT file_id, model_id, x, y, w, h, landmarks, confidence,
-                        embedding, crop_px, crop
-                   FROM faces WHERE file_id = ?1 AND model_id = ?2",
-            )?;
+            let mut fq = src.prepare(&format!(
+                "SELECT f.file_id, f.model_id, f.x, f.y, f.w, f.h, f.landmarks,
+                        f.confidence, f.embedding, f.crop_px, {}
+                   FROM faces f WHERE f.file_id = ?1 AND f.model_id = ?2",
+                crop_column(&src)
+            ))?;
             let faces: Vec<SharedFace> = fq
                 .query_map(rusqlite::params![file_id, &model_id], read_shared_face)?
                 .collect::<Result<_, _>>()?;
@@ -436,6 +438,56 @@ impl FaceShardStore {
     }
 }
 
+/// Add the columns a shard written by an older build is missing.
+///
+/// [`SHARD_SCHEMA`] is entirely `CREATE ... IF NOT EXISTS`, which does exactly
+/// nothing to a table that already exists. So every column added to it *after*
+/// a shard was first written stays invisible to that shard — and the `INSERT`
+/// naming the column then fails with "no such column", which takes the whole
+/// export down with it.
+///
+/// That is not hypothetical, and it is the reason this function exists rather
+/// than the schema being left to speak for itself. `crop` and `indexed_at` were
+/// both added to the batch above, and on any library already holding a shard —
+/// which is every library that had ever synced a face — face export stopped
+/// working entirely and silently, logged as a warning while the catalog went on
+/// filling up with faces no other device would ever see.
+///
+/// Both columns are additive and nullable, so catching up is one `ALTER` each.
+/// There is deliberately no version counter: "does this column exist" is the
+/// question actually being asked, and asking it directly cannot fall out of
+/// step with reality the way a counter can.
+fn upgrade_shard(conn: &Connection) -> Result<(), CatalogError> {
+    for (table, column, decl) in [
+        ("faces", "crop", "BLOB"),
+        ("indexed", "indexed_at", "INTEGER"),
+    ] {
+        if !has_column(conn, table, column)? {
+            conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {decl}"))?;
+        }
+    }
+    Ok(())
+}
+
+/// Whether a table in this connection's own database holds a column.
+fn has_column(conn: &Connection, table: &str, column: &str) -> Result<bool, CatalogError> {
+    let mut stmt = conn.prepare("SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2")?;
+    Ok(stmt.exists(rusqlite::params![table, column])?)
+}
+
+/// `f.crop`, or a `NULL` standing in for it.
+///
+/// A shard downloaded from a peer is opened **read-only** and cannot be
+/// upgraded, so one written before crops existed has to be read as it is rather
+/// than repaired. Selecting a literal keeps the column count the same, which is
+/// what lets [`read_shared_face`] stay a single function.
+fn crop_column(conn: &Connection) -> &'static str {
+    match has_column(conn, "faces", "crop") {
+        Ok(true) => "f.crop",
+        _ => "NULL",
+    }
+}
+
 /// Copy this device's indexed faces into the shard store, ready to upload.
 ///
 /// Only what the store does not already hold, so a call after a partial sweep
@@ -1072,4 +1124,197 @@ mod catalog_round_trip {
         let mut store = FaceShardStore::open(&tempdir("localonly")).unwrap();
         assert_eq!(export_to_shards(&a, &mut store, "w600k_mbf").unwrap(), 1);
     }
+
+    // ── shards written by an older build ──────────────────────────────────
+
+    /// The exact `indexed`/`faces` shape shipped before `crop` and
+    /// `indexed_at` existed. Written out in full rather than derived from
+    /// `SHARD_SCHEMA`, because the whole point is that it is *not* the current
+    /// schema and must not track it.
+    const SHARD_SCHEMA_BEFORE_CROPS: &str = r#"
+    CREATE TABLE IF NOT EXISTS faces (
+        file_id    INTEGER NOT NULL,
+        model_id   TEXT    NOT NULL,
+        x REAL NOT NULL, y REAL NOT NULL, w REAL NOT NULL, h REAL NOT NULL,
+        landmarks  BLOB NOT NULL,
+        confidence REAL NOT NULL,
+        embedding  BLOB NOT NULL,
+        crop_px    REAL NOT NULL
+    );
+    CREATE INDEX IF NOT EXISTS faces_file ON faces(file_id, model_id);
+    CREATE TABLE IF NOT EXISTS indexed (
+        file_id     INTEGER NOT NULL,
+        model_id    TEXT    NOT NULL,
+        faces_found INTEGER NOT NULL,
+        source_edge INTEGER NOT NULL,
+        PRIMARY KEY(file_id, model_id)
+    );
+    "#;
+
+    /// Stand up a store whose shard 0 is in the pre-crop shape, the way every
+    /// library that had ever synced a face actually was.
+    /// A face, in the shape this module's other tests use.
+    fn old_face(file_id: u64, seed: u8) -> SharedFace {
+        SharedFace {
+            file_id,
+            model_id: "w600k_mbf".into(),
+            x: 0.1,
+            y: 0.2,
+            w: 0.15,
+            h: 0.2,
+            landmarks: vec![seed; 40],
+            confidence: 0.87,
+            embedding: vec![seed; 1024],
+            crop_px: 180.0,
+            crop: vec![seed; 64],
+        }
+    }
+
+    fn store_with_an_old_shard() -> (PathBuf, FaceShardStore) {
+        let dir = tempdir("old-shard");
+        let store = FaceShardStore::open(&dir).unwrap();
+        let path = store.shard_path(0);
+        {
+            let c = Connection::open(&path).unwrap();
+            c.execute_batch(SHARD_SCHEMA_BEFORE_CROPS).unwrap();
+            c.execute(
+                "INSERT INTO faces
+                     (file_id, model_id, x, y, w, h, landmarks, confidence,
+                      embedding, crop_px)
+                 VALUES (1, 'w600k_mbf', 0.1, 0.2, 0.15, 0.2, X'00', 0.87, X'00', 180.0)",
+                [],
+            )
+            .unwrap();
+            c.execute(
+                "INSERT INTO indexed (file_id, model_id, faces_found, source_edge)
+                 VALUES (1, 'w600k_mbf', 1, 1024)",
+                [],
+            )
+            .unwrap();
+        }
+        // The index has to believe shard 0 is the open one, or the store would
+        // start a fresh shard and never touch the old file — and it has to
+        // know the image is in there, which is what `contains` reads.
+        store
+            .index
+            .execute(
+                "INSERT INTO shards (id, bytes, sealed) VALUES (0, 100, 0)",
+                [],
+            )
+            .unwrap();
+        store
+            .index
+            .execute(
+                "INSERT INTO entries (file_id, model_id, shard, bytes)
+                 VALUES (1, 'w600k_mbf', 0, 100)",
+                [],
+            )
+            .unwrap();
+        store
+            .index
+            .execute(
+                "INSERT INTO faces_meta (file_id, model_id) VALUES (1, 'w600k_mbf')",
+                [],
+            )
+            .unwrap();
+        (dir, store)
+    }
+
+    /// The regression this exists for: `SHARD_SCHEMA` is all
+    /// `CREATE ... IF NOT EXISTS`, so adding `crop` to it left every existing
+    /// shard without the column — and the next export died on "no such column"
+    /// rather than writing anything. Face sync stopped, quietly, for every
+    /// library that already had a shard.
+    #[test]
+    fn a_shard_from_before_crops_can_still_be_written_to() {
+        let (dir, mut store) = store_with_an_old_shard();
+
+        store
+            .put_image(2, "w600k_mbf", 1024, &[old_face(2, 9)])
+            .expect("writing into a pre-crop shard failed");
+
+        let (back, edge) = store.get_image(2, "w600k_mbf").unwrap().unwrap();
+        assert_eq!(edge, 1024);
+        assert_eq!(back.len(), 1);
+        assert_eq!(back[0].crop, vec![9; 64]);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    /// And the rows already in it survive the upgrade rather than being
+    /// rewritten or lost.
+    #[test]
+    fn upgrading_a_shard_keeps_what_it_already_held() {
+        let (dir, mut store) = store_with_an_old_shard();
+        store
+            .put_image(2, "w600k_mbf", 1024, &[old_face(2, 9)])
+            .unwrap();
+
+        let c = Connection::open(store.shard_path(0)).unwrap();
+        let faces: i64 = c
+            .query_row("SELECT COUNT(*) FROM faces", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(faces, 2, "the pre-existing face was lost");
+        // The old row has no crop, which is exactly right — nothing invented
+        // one for it.
+        let without: i64 = c
+            .query_row("SELECT COUNT(*) FROM faces WHERE crop IS NULL", [], |r| {
+                r.get(0)
+            })
+            .unwrap();
+        assert_eq!(without, 1);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    /// An image already in an old shard has no `indexed_at`, so the export
+    /// cannot vouch for it and must send it again. That is what carries a
+    /// re-indexed library across after an upgrade.
+    #[test]
+    fn an_old_shards_images_have_no_index_time_so_they_re_export() {
+        let (dir, store) = store_with_an_old_shard();
+        assert!(store.contains(1, "w600k_mbf"));
+        assert_eq!(
+            store.indexed_at(1, "w600k_mbf"),
+            None,
+            "a pre-upgrade image claimed an index time it cannot have"
+        );
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    /// A peer's shard is opened read-only and cannot be repaired, so an older
+    /// one has to be readable as it stands.
+    #[test]
+    fn a_peers_shard_from_before_crops_can_still_be_adopted() {
+        let dir = tempdir("old-shard");
+        let old = dir.join("peer.sqlite");
+        {
+            let c = Connection::open(&old).unwrap();
+            c.execute_batch(SHARD_SCHEMA_BEFORE_CROPS).unwrap();
+            c.execute(
+                "INSERT INTO faces
+                     (file_id, model_id, x, y, w, h, landmarks, confidence,
+                      embedding, crop_px)
+                 VALUES (77, 'w600k_mbf', 0.1, 0.2, 0.15, 0.2, X'00', 0.87, X'00', 180.0)",
+                [],
+            )
+            .unwrap();
+            c.execute(
+                "INSERT INTO indexed (file_id, model_id, faces_found, source_edge)
+                 VALUES (77, 'w600k_mbf', 1, 1024)",
+                [],
+            )
+            .unwrap();
+        }
+
+        let mine = dir.join("mine");
+        let mut store = FaceShardStore::open(&mine).unwrap();
+        let adopted = store
+            .merge_shard(&old)
+            .expect("adopting an old shard failed");
+        assert_eq!(adopted, 1);
+
+        let (faces, _) = store.get_image(77, "w600k_mbf").unwrap().unwrap();
+        assert_eq!(faces.len(), 1);
+        assert!(faces[0].crop.is_empty(), "a crop was invented from nowhere");
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }