Take the server's shards and dates when the scan completes, not after the sweep

The derived sync fired only after the metadata sweep, so a fresh device
re-derived every thumbnail it scrolled past, re-detected faces and re-read
every header for hours before adopting the shards and snapshot that held
all of it. It now fires as soon as the scan completes — the first moment
the rows the merges key on exist — and the sweep starts behind it. In
steady state that pass is one listing.

The catalog merge gains a fourth half: capture metadata (captured_at,
offset, camera, lens, ISO) for images still at metadata_state < 2, matched
by oc:fileid from a remote row at 2. A date is a fact about the file's
bytes, not local state, and the snapshot already carried it. The sweep's
per-chunk query then finds nothing left, and the timeline is whole on a
fresh device without a header fetch.

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

core/dr-catalog/src/merge.rs

@@ -119,6 +119,8 @@ pub struct MergeReport {
     pub keywords_fused: usize,
     /// Keyword assignments taken from the remote.
     pub keywords_assigned: usize,
+    /// Images whose capture metadata was taken from the remote.
+    pub metadata_adopted: usize,
 }
 
 impl MergeReport {
@@ -133,6 +135,7 @@ impl MergeReport {
             || self.keywords_deleted > 0
             || self.keywords_fused > 0
             || self.keywords_assigned > 0
+            || self.metadata_adopted > 0
     }
 
     /// Whether the local catalog holds anything the remote did not, and so
@@ -193,10 +196,67 @@ pub fn merge_all(conn: &Connection) -> Result<MergeReport, CatalogError> {
     merge_collections_within(&tx, &mut report)?;
     merge_keywords_within(&tx, &mut report)?;
     merge_people_within(&tx, &mut report)?;
+    merge_metadata_within(&tx, &mut report)?;
     tx.commit()?;
     Ok(report)
 }
 
+/// Adopt capture metadata from an attached catalog, on its own.
+pub fn merge_metadata(conn: &Connection) -> Result<MergeReport, CatalogError> {
+    let tx = conn.unchecked_transaction()?;
+    let mut report = MergeReport::default();
+    merge_metadata_within(&tx, &mut report)?;
+    tx.commit()?;
+    Ok(report)
+}
+
+/// Capture metadata a peer's sweep already read, for images this device has
+/// not dated yet.
+///
+/// The `images` table is local state and the merge leaves it alone — except
+/// for these columns, which are not: a capture time, an offset, a camera, a
+/// lens and an ISO are facts about the file's bytes, identical on every
+/// device, and read by fetching a header per image across the whole library
+/// (`dr_ui::library::spawn_sweep`). A fresh device inherits its peers'
+/// thumbnails and faces from the shards and then spent hours re-reading
+/// every header for the timeline; the snapshot it had just merged held
+/// every one of those dates.
+///
+/// Matched by `oc:fileid`, as collection membership is. Only rows still at
+/// `metadata_state < 2` take anything, and only from a remote row at 2: a
+/// date this device read for itself is never overwritten, and a peer that
+/// has not read one has nothing to give. The sweep's own query
+/// (`metadata_state < 2`) then finds nothing left to do for them.
+const METADATA_BY_FILE_ID: &str = "
+    UPDATE main.images
+       SET captured_at     = r.captured_at,
+           captured_offset = coalesce(main.images.captured_offset, r.captured_offset),
+           camera          = coalesce(main.images.camera, r.camera),
+           lens            = coalesce(main.images.lens, r.lens),
+           iso             = coalesce(main.images.iso, r.iso),
+           metadata_state  = 2
+      FROM (SELECT lr.image_id, ri.captured_at, ri.captured_offset,
+                   ri.camera, ri.lens, ri.iso
+              FROM remote_cat.images ri
+              JOIN remote_cat.remote rr ON rr.image_id = ri.id
+              JOIN main.remote       lr ON lr.file_id = rr.file_id
+             WHERE ri.metadata_state >= 2 AND ri.captured_at IS NOT NULL) AS r
+     WHERE main.images.id = r.image_id
+       AND main.images.metadata_state < 2";
+
+fn merge_metadata_within(tx: &Connection, report: &mut MergeReport) -> Result<(), CatalogError> {
+    // A snapshot from before these columns, or from a library with no server
+    // behind it, has nothing to join on.
+    if !remote_has(tx, "remote")?
+        || !remote_has_column(tx, "images", "metadata_state")?
+        || !remote_has_column(tx, "images", "captured_offset")?
+    {
+        return Ok(());
+    }
+    report.metadata_adopted = tx.execute(METADATA_BY_FILE_ID, [])?;
+    Ok(())
+}
+
 /// Merge people and identity judgements from an attached catalog.
 ///
 /// The people half of [`merge_all`], on its own, for the same reason the other
@@ -1168,6 +1228,57 @@ mod tests {
 
     // ---- integration over two real catalogs ------------------------------
 
+    /// A fresh device takes the capture dates a peer's sweep read, matched by
+    /// `oc:fileid`, and never overwrites a date it read for itself.
+    #[test]
+    fn capture_metadata_arrives_for_undated_images_only() {
+        let c = two_catalogs();
+        // Three photographs on both devices: 1 undated here and dated there;
+        // 2 dated on both, differently; 3 undated on both.
+        for id in 1..=3 {
+            add_image_without_hash(&c, "main", id);
+            add_image_without_hash(&c, "remote_cat", id + 10);
+            add_remote_id(&c, "main", id, 100 + id);
+            add_remote_id(&c, "remote_cat", id + 10, 100 + id);
+        }
+        c.execute(
+            "UPDATE remote_cat.images
+                SET captured_at = 1000, captured_offset = 60, camera = 'X', metadata_state = 2
+              WHERE id = 11",
+            [],
+        )
+        .unwrap();
+        c.execute(
+            "UPDATE remote_cat.images SET captured_at = 2000, metadata_state = 2 WHERE id = 12",
+            [],
+        )
+        .unwrap();
+        c.execute(
+            "UPDATE main.images SET captured_at = 2222, metadata_state = 2 WHERE id = 2",
+            [],
+        )
+        .unwrap();
+
+        let report = merge_metadata(&c).unwrap();
+        assert_eq!(report.metadata_adopted, 1);
+
+        let row = |id: i64| -> (Option<i64>, Option<i64>, Option<String>, i64) {
+            c.query_row(
+                "SELECT captured_at, captured_offset, camera, metadata_state
+                   FROM main.images WHERE id = ?1",
+                [id],
+                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
+            )
+            .unwrap()
+        };
+        assert_eq!(row(1), (Some(1000), Some(60), Some("X".into()), 2));
+        assert_eq!(row(2), (Some(2222), None, None, 2));
+        assert_eq!(row(3), (None, None, None, 0));
+
+        // Idempotent: a second pass finds nothing left to take.
+        assert_eq!(merge_metadata(&c).unwrap().metadata_adopted, 0);
+    }
+
     fn two_catalogs() -> Connection {
         attached_remote(schema::for_attached("remote_cat"))
     }