diff --git a/core/dr-catalog/src/face_shard.rs b/core/dr-catalog/src/face_shard.rs index d4a72ff..b127e79 100644 --- a/core/dr-catalog/src/face_shard.rs +++ b/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 = fq .query_map(rusqlite::params![file_id, &model_id], read_shared_face)? .collect::>()?; @@ -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 { + 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); + } }