Stamp the backfill on the newest rows, not only their ids

The backfill stamp read max(id) of images and versions and max(rowid)
of keywords. None of those tables is AUTOINCREMENT, so SQLite hands a
freed newest id out again: empty the trash of the newest photograph and
scan a new one, or let a local folder's walk delete a renamed file's row
and insert the new name in the same pass, and the new image takes the
old id. max(id) does not move, nor does count(*), and when a newer
version elsewhere keeps max(versions.id) still too, the stamp matched
and the open skipped the backfill.

That row is exactly one that needs it. Neither scan path creates the
default version: scan::persist and walk insert the image and leave the
version, the RAW/JPEG pairing and the keyword terms to the next open.
Skipped, the image went without them until the app restarted, so a
rating or a pulled sidecar judgement had no version to land on and a
JPEG beside its RAW showed twice.

The stamp now carries the newest row's content: the newest image's id,
path, added time and whether it has a version; the newest version's id
and image; the newest assignment's rowid, version and word. Whether the
newest image has a version is the part that cannot be fooled - after a
backfill every image has one, and a row that has just taken a freed id
has none - so the two stamps differ even when the same file comes back
at the same id in the same second. Still one statement: three reverse
rowid scans that stop at the first row, and one probe of versions_image.
An open that skips still costs ~1 ms on the reference catalog copy.

This closes the hole in the stamp itself rather than by a forget() at
each delete site, so a delete path added later, or one in another
process, cannot reopen it. Two tests delete the newest image and insert
another at the freed id on a separate connection, with a newer version
elsewhere holding max(versions.id); both fail against the old stamp.

core/dr-catalog/src/backfilled.rs

@@ -21,10 +21,34 @@
 //! may carry a minted uuid; a keyword assignment merged from a remote may name
 //! a word with no term. So the question "is any work owed?" is answered by
 //! whether those tables have gained rows since the last backfill, and that is
-//! three `max(rowid)` lookups — each the last page of a b-tree — rather than a
+//! a read of each table's last row — the last page of its b-tree — rather than
-//! scan.
+//! a scan.
 //!
-//! The [`Stamp`] is those maxima, the schema version, and the file's identity.
+//! # Why the last row, and not only its id
+//!
+//! None of these tables is `AUTOINCREMENT`, so SQLite hands out the largest
+//! rowid plus one, and an id freed by deleting the newest row is handed out
+//! again. That is an ordinary sequence, not a contrived one: emptying the
+//! trash of the newest photograph and then scanning a new one, or a local
+//! folder's walk removing a renamed file's row and inserting the new name in
+//! the same pass. `max(id)` does not move, and neither does `count(*)`. And
+//! the row that took the id is exactly one that needs the backfill, because
+//! neither scan creates default versions — `persist` and the walk insert the
+//! image and leave the version, the pairing and the keyword terms to the next
+//! open. Skipped, it would go without them until the app restarted: a rating
+//! or a keyword with nowhere to land, a JPEG beside its RAW shown twice.
+//!
+//! So the stamp carries the last row's content as well as its id: the newest
+//! image's path, when it was added, and **whether it has a version**; the
+//! newest version's image; the newest assignment's word and version. Whether
+//! the newest image has a version is the part that cannot be fooled: once the
+//! backfill has run, every image has one, and a row that has just taken a
+//! freed id has none, so the two stamps differ whatever the path and the time
+//! say. The others make the newest version or assignment a different row
+//! whenever a different one took its id; one that is the same content at the
+//! same id is the same row as far as the backfill is concerned.
+//!
+//! The [`Stamp`] is those, the schema version, and the file's identity.
 //! An open whose stamp matches the one recorded at the last backfill of the
 //! same path skips it; anything else runs it. That covers the cases that must
 //! run it:
@@ -83,9 +107,12 @@ pub(crate) struct Stamp {
     /// catalog. `None` where the platform has no such thing.
     file: Option<(u64, u64)>,
     user_version: i64,
-    max_image: i64,
+    /// The newest image: id, path, when added, and whether it has a version.
-    max_version: i64,
+    last_image: Option<String>,
-    max_keyword: i64,
+    /// The newest version: id and the image it belongs to.
+    last_version: Option<String>,
+    /// The newest keyword assignment: rowid, version and word.
+    last_keyword: Option<String>,
 }
 
 /// The stamp recorded at the last backfill, per catalog file.
@@ -101,23 +128,29 @@ fn key(path: &Path) -> PathBuf {
 
 /// Read the stamp of the catalog behind `conn`, which was opened from `path`.
 ///
-/// One statement of three `max()`s over rowids — each answered from the last
+/// One statement: the last row of each of three tables, each found by
-/// page of its table — plus a `stat` of the file.
+/// descending its rowid b-tree to the last page, plus one probe of
+/// `versions_image` for the newest image — and a `stat` of the file.
 pub(crate) fn stamp(conn: &Connection, path: &Path) -> Result<Stamp, CatalogError> {
-    let (user_version, max_image, max_version, max_keyword) = conn.query_row(
+    let (user_version, last_image, last_version, last_keyword) = conn.query_row(
         "SELECT (SELECT user_version FROM pragma_user_version),
-                coalesce((SELECT max(id) FROM images), 0),
+                (SELECT printf('%d|%d|%d|%s', i.id, i.added_at,
-                coalesce((SELECT max(id) FROM versions), 0),
+                               EXISTS (SELECT 1 FROM versions v WHERE v.image_id = i.id),
-                coalesce((SELECT max(rowid) FROM keywords), 0)",
+                               i.source_ref)
+                   FROM images i ORDER BY i.id DESC LIMIT 1),
+                (SELECT printf('%d|%d', id, image_id)
+                   FROM versions ORDER BY id DESC LIMIT 1),
+                (SELECT printf('%d|%d|%s', rowid, version_id, keyword)
+                   FROM keywords ORDER BY rowid DESC LIMIT 1)",
         [],
         |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
     )?;
     Ok(Stamp {
         file: file_identity(path),
         user_version,
-        max_image,
+        last_image,
-        max_version,
+        last_version,
-        max_keyword,
+        last_keyword,
     })
 }
 
@@ -246,6 +279,95 @@ mod tests {
         assert!(uuid(&path, 2).is_some(), "the new image got its version");
     }
 
+    /// The id of a deleted newest row is handed out again, so `max(id)` is
+    /// the same before and after — the sequence emptying the trash and then
+    /// scanning makes. The image that took the id still needs its version.
+    ///
+    /// A virtual copy on the older image holds the newest version id, so
+    /// the deletion does not move `max(versions.id)` either: nothing the old
+    /// stamp read changes, which is the case that went unrepaired.
+    #[test]
+    fn an_image_that_reuses_a_deleted_id_is_backfilled_on_the_next_open() {
+        let path = catalog("reused");
+        rusqlite::Connection::open(&path)
+            .unwrap()
+            .execute(
+                "INSERT INTO images(id, root_id, source_ref, added_at)
+                 VALUES (2, 1, 'Photos/b.CR3', 0)",
+                [],
+            )
+            .unwrap();
+        assert!(uuid(&path, 2).is_some());
+        rusqlite::Connection::open(&path)
+            .unwrap()
+            .execute(
+                "INSERT INTO versions(image_id, uuid, name, is_default)
+                 VALUES (1, 'copy', 'Crop', 0)",
+                [],
+            )
+            .unwrap();
+        // Settle: one open backfills after the new version, one more runs
+        // the redundant pass and records the stamp that stands.
+        assert!(uuid(&path, 2).is_some());
+        assert!(uuid(&path, 2).is_some());
+
+        let c = rusqlite::Connection::open(&path).unwrap();
+        let before: (i64, i64) = c
+            .query_row(
+                "SELECT (SELECT max(id) FROM images), (SELECT max(id) FROM versions)",
+                [],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .unwrap();
+        c.execute("DELETE FROM images WHERE id = 2", []).unwrap();
+        c.execute(
+            "INSERT INTO images(root_id, source_ref, added_at)
+             VALUES (1, 'Photos/c.CR3', 0)",
+            [],
+        )
+        .unwrap();
+        let after: (i64, i64) = c
+            .query_row(
+                "SELECT (SELECT max(id) FROM images), (SELECT max(id) FROM versions)",
+                [],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .unwrap();
+        assert_eq!(before, after, "SQLite handed the freed id out again");
+        drop(c);
+        assert!(uuid(&path, 2).is_some(), "the new image got its version");
+    }
+
+    /// The same, with the same file coming back at the same id in the same
+    /// second: path and time match, and only the missing version tells.
+    #[test]
+    fn the_same_file_back_at_the_same_id_is_backfilled_on_the_next_open() {
+        let path = catalog("returned");
+        let c = rusqlite::Connection::open(&path).unwrap();
+        // As above: a newer version on another image keeps the deletion
+        // from moving `max(versions.id)`.
+        c.execute_batch(
+            "INSERT INTO images(id, root_id, source_ref, added_at)
+                  VALUES (0, 1, 'Photos/0.CR3', 0);
+             INSERT INTO versions(image_id, uuid, name, is_default)
+                  VALUES (0, 'copy', 'Crop', 0);",
+        )
+        .unwrap();
+        drop(c);
+        assert!(uuid(&path, 1).is_some());
+        assert!(uuid(&path, 1).is_some());
+        let c = rusqlite::Connection::open(&path).unwrap();
+        c.execute_batch(
+            "DELETE FROM images WHERE id = 1;
+             INSERT INTO images(id, root_id, source_ref, added_at)
+                  VALUES (1, 1, 'Photos/a.CR3', 0);
+             INSERT INTO remote(image_id, file_id) VALUES (1, 77);",
+        )
+        .unwrap();
+        drop(c);
+        assert_eq!(uuid(&path, 1), Some(derived_version_uuid(77)));
+    }
+
     #[test]
     fn the_first_open_after_a_migration_backfills() {
         let path = catalog("migrated");