Indexing: Add scan bytes counter + heal-contract fix

- Add `ScanProgress.bytes_scanned` (resolved post-dedup physical bytes), incremented once per stored entry after the hardlink-dedup match so it tracks the stored physical-size sums apples-to-apples. `snapshot()` now returns a `ScanProgressSnapshot` struct.
- Add `ScanSummary.total_physical_bytes` at both construction sites, and persist it via `UpdateMeta` in the completion handler.
- Fix the broken "interrupted scan → fresh rescan" contract: new `WriteMessage::DeleteMeta` + `IndexStore::delete_meta` clear the previous `scan_completed_at` at scan start (before `TruncateData`), and all completion-handler meta writes are now gated behind `!summary.was_cancelled`. Calibration keys stay intact so they keep describing the last completed scan; the reconcile/live transition is unchanged. `DeleteMeta` doesn't bump the writer generation.
- Add `ScanCalibration` + the connection-taking `IndexStore::read_scan_calibration` helper (reads `total_entries`/`total_physical_bytes`/`scan_duration_ms` as `Option`s).
- Promote `get_space_info_for_path` to `pub(crate)` for reuse.
- Tests: bytes-counter parity with hardlinks, summary == final counter, writer `UpdateMeta`/`DeleteMeta` round-trips, calibration read helper (seeded + missing keys).

Author: vdavid

apps/desktop/src-tauri/src/file_system/volume/backends/local_posix.rs

@@ -796,7 +796,7 @@ impl VolumeReadStream for LocalPosixReadStream {
 /// On macOS, uses `NSURLVolumeAvailableCapacityForImportantUsageKey` which includes purgeable
 /// space (APFS snapshots, iCloud caches), matching what Finder reports. Falls back to `statvfs`
 /// if the NSURL query fails. On Linux, uses `statvfs` directly (no purgeable space concept).
-fn get_space_info_for_path(path: &Path) -> Result<SpaceInfo, VolumeError> {
+pub(crate) fn get_space_info_for_path(path: &Path) -> Result<SpaceInfo, VolumeError> {
     // On macOS, prefer the NSURL API that accounts for purgeable space.
     #[cfg(target_os = "macos")]
     {

apps/desktop/src-tauri/src/indexing/manager.rs

@@ -303,7 +303,22 @@ impl IndexManager {
             return Err("Scan already running".to_string());
         }
 
-        // Step 0: Truncate entries + dir_stats so the scan inserts into an empty DB.
+        // Step 0a: Clear the previous scan's completion marker BEFORE truncating.
+        // Without this, a rescan killed mid-way (power loss, `kill -9`) leaves the
+        // PREVIOUS scan's `scan_completed_at` in meta on top of a truncated/partial
+        // `entries` table, so the next startup takes the journal-replay path over a
+        // gutted index instead of the `IncompletePreviousScan` fresh rescan. The
+        // calibration keys (`total_entries`, `total_physical_bytes`, `scan_duration_ms`)
+        // are intentionally left intact so they keep describing the last COMPLETED
+        // scan throughout this one. The same flush below covers both sends.
+        if let Err(e) = self
+            .writer
+            .send(WriteMessage::DeleteMeta("scan_completed_at".to_string()))
+        {
+            log::warn!("Failed to send DeleteMeta(scan_completed_at): {e}");
+        }
+
+        // Step 0b: Truncate entries + dir_stats so the scan inserts into an empty DB.
         // Without this, INSERT OR REPLACE on a populated table with the `platform_case`
         // collation is ~12x slower (30 min vs 2.5 min), and old rows with stale IDs
         // accumulate as orphaned subtrees, bloating the DB 3-4x per scan cycle.
@@ -382,13 +397,13 @@ impl IndexManager {
                 if scan_done_progress.load(Ordering::Relaxed) {
                     break;
                 }
-                let (entries, dirs) = progress.snapshot();
+                let snap = progress.snapshot();
                 let _ = app_progress.emit(
                     "index-scan-progress",
                     IndexScanProgressEvent {
                         volume_id: volume_id_progress.clone(),
-                        entries_scanned: entries,
-                        dirs_found: dirs,
+                        entries_scanned: snap.entries_scanned,
+                        dirs_found: snap.dirs_found,
                     },
                 );
 
@@ -550,26 +565,40 @@ impl IndexManager {
                     // can fail (e.g. "database is locked") and cause an early return.
                     // Without this, scan_completed_at is never persisted and the next
                     // startup triggers a full rescan of the entire volume.
-                    let now = std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .map(|d| d.as_secs().to_string())
-                        .unwrap_or_default();
-                    let _ = writer.send(WriteMessage::UpdateMeta {
-                        key: "scan_completed_at".to_string(),
-                        value: now,
-                    });
-                    let _ = writer.send(WriteMessage::UpdateMeta {
-                        key: "scan_duration_ms".to_string(),
-                        value: summary.duration_ms.to_string(),
-                    });
-                    let _ = writer.send(WriteMessage::UpdateMeta {
-                        key: "total_entries".to_string(),
-                        value: summary.total_entries.to_string(),
-                    });
-                    let _ = writer.send(WriteMessage::UpdateMeta {
-                        key: "volume_path".to_string(),
-                        value: "/".to_string(),
-                    });
+                    //
+                    // Gate ALL meta writes behind `!was_cancelled`: a user-stopped scan
+                    // holds only partial totals, and writing `scan_completed_at` for it
+                    // would mark a partial index as complete — the next startup would skip
+                    // the `IncompletePreviousScan` fresh rescan. With the clear-at-start
+                    // above, a cancelled scan leaves NO completion marker, so it heals on
+                    // restart. The reconcile/live transition below is intentionally NOT
+                    // gated; only the meta writes are.
+                    if !summary.was_cancelled {
+                        let now = std::time::SystemTime::now()
+                            .duration_since(std::time::UNIX_EPOCH)
+                            .map(|d| d.as_secs().to_string())
+                            .unwrap_or_default();
+                        let _ = writer.send(WriteMessage::UpdateMeta {
+                            key: "scan_completed_at".to_string(),
+                            value: now,
+                        });
+                        let _ = writer.send(WriteMessage::UpdateMeta {
+                            key: "scan_duration_ms".to_string(),
+                            value: summary.duration_ms.to_string(),
+                        });
+                        let _ = writer.send(WriteMessage::UpdateMeta {
+                            key: "total_entries".to_string(),
+                            value: summary.total_entries.to_string(),
+                        });
+                        let _ = writer.send(WriteMessage::UpdateMeta {
+                            key: "total_physical_bytes".to_string(),
+                            value: summary.total_physical_bytes.to_string(),
+                        });
+                        let _ = writer.send(WriteMessage::UpdateMeta {
+                            key: "volume_path".to_string(),
+                            value: "/".to_string(),
+                        });
+                    }
 
                     // Open a read connection for path resolution during replay
                     let replay_conn = match IndexStore::open_read_connection(&writer.db_path()) {
@@ -682,11 +711,9 @@ impl IndexManager {
 
         let db_file_size = self.store.db_file_size().ok();
 
-        let (entries_scanned, dirs_found) = self
-            .scan_handle
-            .as_ref()
-            .map(|h| h.progress.snapshot())
-            .unwrap_or((0, 0));
+        let snap = self.scan_handle.as_ref().map(|h| h.progress.snapshot());
+        let entries_scanned = snap.map(|s| s.entries_scanned).unwrap_or(0);
+        let dirs_found = snap.map(|s| s.dirs_found).unwrap_or(0);
 
         Ok(IndexStatusResponse {
             initialized: true,

apps/desktop/src-tauri/src/indexing/scanner.rs

@@ -104,22 +104,37 @@ impl Default for ScanConfig {
 pub struct ScanProgress {
     pub entries_scanned: Arc<AtomicU64>,
     pub dirs_found: Arc<AtomicU64>,
+    /// Resolved post-dedup physical bytes seen so far. Each entry contributes its
+    /// `physical_size.unwrap_or(0)` after hardlink dedup, so the live numerator
+    /// follows the exact same rules as the stored physical-size sums (directories,
+    /// symlinks, and second+ hardlinks contribute 0).
+    pub bytes_scanned: Arc<AtomicU64>,
+}
+
+/// A point-in-time read of an active scan's progress counters.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ScanProgressSnapshot {
+    pub entries_scanned: u64,
+    pub dirs_found: u64,
+    pub bytes_scanned: u64,
 }
 
 impl ScanProgress {
     fn new() -> Self {
         Self {
             entries_scanned: Arc::new(AtomicU64::new(0)),
             dirs_found: Arc::new(AtomicU64::new(0)),
+            bytes_scanned: Arc::new(AtomicU64::new(0)),
         }
     }
 
     /// Read current progress snapshot.
-    pub fn snapshot(&self) -> (u64, u64) {
-        (
-            self.entries_scanned.load(Ordering::Relaxed),
-            self.dirs_found.load(Ordering::Relaxed),
-        )
+    pub fn snapshot(&self) -> ScanProgressSnapshot {
+        ScanProgressSnapshot {
+            entries_scanned: self.entries_scanned.load(Ordering::Relaxed),
+            dirs_found: self.dirs_found.load(Ordering::Relaxed),
+            bytes_scanned: self.bytes_scanned.load(Ordering::Relaxed),
+        }
     }
 }
 
@@ -141,6 +156,9 @@ impl ScanHandle {
 pub struct ScanSummary {
     pub total_entries: u64,
     pub total_dirs: u64,
+    /// Resolved post-dedup physical bytes the scan summed (the final value of the
+    /// `bytes_scanned` counter). Apples-to-apples with the stored physical-size sums.
+    pub total_physical_bytes: u64,
     pub duration_ms: u64,
     pub was_cancelled: bool,
 }
@@ -267,6 +285,7 @@ fn run_scan(
     let mut batch: Vec<EntryRow> = Vec::with_capacity(batch_size);
     let mut total_entries: u64 = 0;
     let mut total_dirs: u64 = 0;
+    let mut total_physical_bytes: u64 = 0;
     // Tracks inodes with nlink > 1 so each hardlinked file's size is counted only once.
     // Files with nlink == 1 (the vast majority) skip the set entirely.
     let mut seen_inodes: HashSet<u64> = HashSet::new();
@@ -305,6 +324,7 @@ fn run_scan(
             return Ok(ScanSummary {
                 total_entries,
                 total_dirs,
+                total_physical_bytes,
                 duration_ms: start.elapsed().as_millis() as u64,
                 was_cancelled: true,
             });
@@ -417,6 +437,14 @@ fn run_scan(
         total_entries += 1;
         progress.entries_scanned.fetch_add(1, Ordering::Relaxed);
 
+        // Accumulate the resolved post-dedup physical bytes once per stored entry.
+        // Placed here (after the hardlink-dedup match, alongside the entry counters)
+        // so it follows the exact dedup rules of the stored sums: directories,
+        // symlinks, and second+ hardlinks resolved to `None` contribute 0.
+        let entry_physical = physical_size.unwrap_or(0);
+        total_physical_bytes += entry_physical;
+        progress.bytes_scanned.fetch_add(entry_physical, Ordering::Relaxed);
+
         batch.push(scanned);
         if batch.len() >= batch_size {
             flush_batch(&mut batch, writer)?;
@@ -435,6 +463,7 @@ fn run_scan(
     Ok(ScanSummary {
         total_entries,
         total_dirs,
+        total_physical_bytes,
         duration_ms: start.elapsed().as_millis() as u64,
         was_cancelled: false,
     })
@@ -741,9 +770,10 @@ mod tests {
         assert!(summary.duration_ms < 10_000, "scan should complete quickly");
 
         // Verify progress matches summary
-        let (entries, dirs) = handle.progress.snapshot();
-        assert_eq!(entries, summary.total_entries);
-        assert_eq!(dirs, summary.total_dirs);
+        let snap = handle.progress.snapshot();
+        assert_eq!(snap.entries_scanned, summary.total_entries);
+        assert_eq!(snap.dirs_found, summary.total_dirs);
+        assert_eq!(snap.bytes_scanned, summary.total_physical_bytes);
 
         // Wait for writer to process all messages + aggregation
         writer.flush_blocking().unwrap();
@@ -1101,4 +1131,95 @@ mod tests {
         // Should resolve to the actual entry ID, NOT ROOT_ID
         assert_eq!(ctx.lookup_parent(subtree_root), Some(noname_id));
     }
+
+    /// Sum every stored row's `physical_size` (NULLs count as 0), matching how the
+    /// aggregator treats per-entry physical bytes.
+    fn sum_stored_physical_bytes(db_path: &Path) -> u64 {
+        let conn = IndexStore::open_read_connection(db_path).unwrap();
+        conn.query_row("SELECT COALESCE(SUM(physical_size), 0) FROM entries", [], |row| {
+            row.get::<_, i64>(0)
+        })
+        .unwrap() as u64
+    }
+
+    /// Build a tree with BOTH plain single-link files AND a hardlink pair. The
+    /// single-link files are what catch a "bytes increment placed inside the
+    /// dedup arm" bug: that arm fires only for `nlink > 1`, so single-link files
+    /// would contribute nothing and near-zero the counter.
+    #[cfg(unix)]
+    fn create_tree_with_hardlinks(dir: &Path) {
+        // Plain single-link files (the majority).
+        fs::write(dir.join("plain1.bin"), vec![0u8; 4096]).unwrap();
+        fs::write(dir.join("plain2.bin"), vec![0u8; 12288]).unwrap();
+        let sub = dir.join("sub");
+        fs::create_dir_all(&sub).unwrap();
+        fs::write(sub.join("plain3.bin"), vec![0u8; 8192]).unwrap();
+
+        // A hardlink pair: two directory entries, one inode. Only the first link's
+        // size should be counted; the second resolves to None.
+        let target = dir.join("linked.bin");
+        fs::write(&target, vec![0u8; 16384]).unwrap();
+        fs::hard_link(&target, dir.join("linked-alias.bin")).unwrap();
+    }
+
+    #[test]
+    #[cfg(unix)]
+    fn bytes_scanned_matches_stored_physical_sum_with_hardlinks() {
+        let scan_root = scan_test_tempdir();
+        create_tree_with_hardlinks(scan_root.path());
+
+        let (writer, db_path, _db_dir) = setup_writer();
+        let config = ScanConfig {
+            root: scan_root.path().to_path_buf(),
+            batch_size: 100,
+            num_threads: 1,
+        };
+
+        let (handle, join_handle) = scan_volume(config, &writer).unwrap();
+        let summary = join_handle.join().expect("scan thread panicked").unwrap();
+        assert!(!summary.was_cancelled);
+
+        writer.flush_blocking().unwrap();
+        writer.shutdown();
+
+        let counter_total = handle.progress.snapshot().bytes_scanned;
+        let stored_total = sum_stored_physical_bytes(&db_path);
+
+        // The live counter follows the exact post-dedup rules of the stored rows.
+        assert_eq!(
+            counter_total, stored_total,
+            "bytes_scanned counter must equal the sum of stored physical sizes"
+        );
+        // Sanity: the plain single-link files alone exceed any single hardlink, so a
+        // counter that only ran inside the dedup arm would fall well below this.
+        assert!(
+            counter_total >= 4096 + 12288 + 8192,
+            "counter must include the single-link files, not just the hardlink"
+        );
+    }
+
+    #[test]
+    #[cfg(unix)]
+    fn scan_summary_total_physical_bytes_equals_final_counter() {
+        let scan_root = scan_test_tempdir();
+        create_tree_with_hardlinks(scan_root.path());
+
+        let (writer, _db_path, _db_dir) = setup_writer();
+        let config = ScanConfig {
+            root: scan_root.path().to_path_buf(),
+            batch_size: 100,
+            num_threads: 1,
+        };
+
+        let (handle, join_handle) = scan_volume(config, &writer).unwrap();
+        let summary = join_handle.join().expect("scan thread panicked").unwrap();
+        writer.shutdown();
+
+        assert_eq!(
+            summary.total_physical_bytes,
+            handle.progress.snapshot().bytes_scanned,
+            "summary.total_physical_bytes must equal the final counter value"
+        );
+        assert!(summary.total_physical_bytes > 0, "scan should sum some physical bytes");
+    }
 }