Drive index: keep min_subtree_epoch honest under live mutation

The live-watch and verifier paths (where the local index spends ~all its life) now maintain coverage correctly: `propagate_delta_by_id` carries `min_subtree_epoch` through a pure size/count delta unchanged (so a file write into a complete dir no longer flips it to a lower bound), and a new `propagate_min_subtree_epoch` re-min walk fires only on tree-shape changes — new dir (ancestors drop to 0), delete/delete-subtree (a gone incomplete child can raise coverage), and cross-parent move (both chains). `reconcile_subtree` (the real live fill path) now marks every dir it lists and re-mins up; the verifier lifts ancestor coverage after a subtree scan. Regression pinned: a live file-add keeps an exact dir exact.

Author: vdavid

apps/desktop/src-tauri/src/indexing/DETAILS.md

@@ -403,6 +403,16 @@ pub(super) fn reconcile_subtree(
     let mut removed: u64 = 0;
     let mut updated: u64 = 0;
 
+    // The epoch every dir we successfully list this pass is stamped with. A
+    // reconcile *stamps* with the current epoch; it never bumps it. Read once.
+    let epoch = IndexStore::read_current_epoch(conn).unwrap_or(1);
+    // Every dir whose direct contents we successfully list (incl. empty), so we
+    // can `MarkDirsListed` them after the walk and lift ancestor coverage.
+    // Without this, a reconcile-discovered subtree stays `listed_epoch = 0`
+    // forever and drags every ancestor to incomplete — the exact local-live-path
+    // regression this milestone guards against.
+    let mut listed_dir_ids: Vec<i64> = Vec::new();
+
     let root_str = firmlinks::normalize_path(&root.to_string_lossy());
     let root_id = match store::resolve_path(conn, &root_str) {
         Ok(Some(id)) => id,
@@ -495,6 +505,9 @@ pub(super) fn reconcile_subtree(
             Some(c) => c,
             None => continue,
         };
+        // We successfully listed this dir's direct contents (an empty listing
+        // still counts). Stamp it at the current epoch after the walk.
+        listed_dir_ids.push(dir_id);
 
         let db_children =
             IndexStore::list_children_on(dir_id, conn).map_err(|e| format!("list_children_on({dir_id}): {e}"))?;
@@ -600,6 +613,27 @@ pub(super) fn reconcile_subtree(
         }
     }
 
+    // Stamp every dir we listed at the current epoch, then lift ancestor
+    // coverage. The walk collected ids shallow→deep (BFS), so recompute
+    // deepest-first: a parent's `min_subtree_epoch` reads its children's stored
+    // values, which must already reflect this pass. `propagate_min_subtree_epoch`
+    // short-circuits once a value stabilizes, so the repeated up-walks are cheap.
+    if !listed_dir_ids.is_empty() {
+        // Chunk under SQLite's bound-parameter ceiling (+1 for the epoch param).
+        const MARK_CHUNK: usize = 900;
+        for chunk in listed_dir_ids.chunks(MARK_CHUNK) {
+            if let Err(e) = writer.send(WriteMessage::MarkDirsListed {
+                ids: chunk.to_vec(),
+                epoch,
+            }) {
+                log::warn!("reconcile_subtree: failed to send MarkDirsListed: {e}");
+            }
+        }
+        for dir_id in listed_dir_ids.iter().rev() {
+            let _ = writer.send(WriteMessage::PropagateMinSubtreeEpoch(*dir_id));
+        }
+    }
+
     Ok(ReconcileSummary {
         added,
         removed,
@@ -1873,6 +1907,69 @@ mod tests {
         assert_eq!(children.len(), 2, "both child files should be indexed");
     }
 
+    /// A reconcile-discovered subtree must be stamped `listed_epoch = current`
+    /// for every dir it lists (including empty ones), and ancestor coverage must
+    /// lift. Without the mark, the subtree stays `listed_epoch = 0` forever and
+    /// drags ancestors to incomplete — the exact local-live-path regression M1
+    /// guards against.
+    #[test]
+    fn reconcile_subtree_marks_listed_dirs_at_current_epoch() {
+        let (writer, dir, conn) = setup_test_writer();
+        let db_path = dir.path().join("test-reconciler.db");
+
+        // Stamp the volume's current epoch at 5.
+        {
+            let wconn = IndexStore::open_write_connection(&db_path).unwrap();
+            IndexStore::update_meta(&wconn, "current_epoch", "5").unwrap();
+        }
+
+        // On disk: a new tree with a child dir (non-empty) and an empty dir.
+        let test_dir = non_excluded_tempdir();
+        let new_dir = test_dir.path().join("tree");
+        std::fs::create_dir(&new_dir).unwrap();
+        std::fs::create_dir(new_dir.join("sub")).unwrap();
+        std::fs::write(new_dir.join("sub").join("f.txt"), "x").unwrap();
+        std::fs::create_dir(new_dir.join("empty")).unwrap();
+
+        // Only the parent of `tree` is in the DB (mimics must_scan_sub_dirs).
+        ensure_path_in_db(&db_path, &test_dir.path().to_string_lossy(), &writer);
+
+        let cancelled = AtomicBool::new(false);
+        reconcile_subtree(&new_dir, &conn, &writer, &cancelled).unwrap();
+        writer.flush_blocking().unwrap();
+        writer.shutdown();
+
+        let store = IndexStore::open(&db_path).unwrap();
+        let rconn = store.read_conn();
+        let resolve = |p: &Path| {
+            store::resolve_path(rconn, &p.to_string_lossy())
+                .unwrap()
+                .unwrap_or_else(|| panic!("{} should be in DB", p.display()))
+        };
+        let tree_id = resolve(&new_dir);
+        let sub_id = resolve(&new_dir.join("sub"));
+        let empty_id = resolve(&new_dir.join("empty"));
+
+        // Every listed dir (including the empty one) is stamped at epoch 5.
+        for (label, id) in [("tree", tree_id), ("sub", sub_id), ("empty", empty_id)] {
+            assert_eq!(
+                IndexStore::get_listed_epoch_by_id(rconn, id).unwrap(),
+                Some(5),
+                "{label} must be listed at the current epoch"
+            );
+        }
+
+        // Coverage lifted: the whole reconciled subtree is complete at epoch 5.
+        assert_eq!(
+            IndexStore::get_dir_stats_by_id(rconn, tree_id)
+                .unwrap()
+                .unwrap()
+                .min_subtree_epoch,
+            5,
+            "tree's min_subtree_epoch lifts to 5 (fully listed)"
+        );
+    }
+
     /// Bug 2: after a MustScanSubDirs rescan completes, pending queued rescans
     /// should be started automatically. Previously the spawned task set
     /// rescan_active=false but never called start_next_rescan, so queued paths

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

@@ -1134,13 +1134,54 @@ impl IndexStore {
     }
 
     /// Read an entry's `listed_epoch` by id. `None` if the entry doesn't exist.
-    #[cfg(test)]
     pub fn get_listed_epoch_by_id(conn: &Connection, id: i64) -> Result<Option<u64>, IndexStoreError> {
         let mut stmt = conn.prepare_cached("SELECT listed_epoch FROM entries WHERE id = ?1")?;
         let result = stmt.query_row(params![id], |row| row.get::<_, u64>(0)).optional()?;
         Ok(result)
     }
 
+    /// Recompute a directory's `min_subtree_epoch` from its own `listed_epoch`
+    /// and every child directory's stored `min_subtree_epoch`, taking the
+    /// 0-absorbing `min` (any `0` ⇒ `0`).
+    ///
+    /// This is the per-dir step of `delta::propagate_min_subtree_epoch`, kept in
+    /// the store as a single SQL pass so the walk does one round trip per
+    /// ancestor instead of N+1. A dir with no `entries` row is treated as
+    /// unlisted (`0`). The 0-absorbing semantics differ from a plain `MIN()`:
+    /// SQL `MIN` over `{1, 0}` is `0` (correct here by luck), but `MIN` over an
+    /// empty child set is `NULL`, and `MIN` ignoring NULLs could mask a missing
+    /// `dir_stats` row — so the CTE coalesces a missing child stat to `0`.
+    pub fn recompute_min_subtree_epoch(conn: &Connection, dir_id: i64) -> Result<u64, IndexStoreError> {
+        // Own listed_epoch (0 if the row is gone).
+        let own: u64 = conn
+            .prepare_cached("SELECT listed_epoch FROM entries WHERE id = ?1")?
+            .query_row(params![dir_id], |row| row.get::<_, u64>(0))
+            .optional()?
+            .unwrap_or(0);
+        if own == 0 {
+            return Ok(0);
+        }
+
+        // 0-absorbing min over child dirs' min_subtree_epoch. A child dir with no
+        // dir_stats row counts as 0 (unknown coverage), so LEFT JOIN + COALESCE.
+        // If ANY child is 0, MIN is 0. No child dirs ⇒ MIN over empty ⇒ NULL ⇒
+        // keep `own` (a listed-but-childless dir is fully covered at its epoch).
+        let child_min: Option<u64> = conn
+            .prepare_cached(
+                "SELECT MIN(COALESCE(ds.min_subtree_epoch, 0))
+                 FROM entries c
+                 LEFT JOIN dir_stats ds ON ds.entry_id = c.id
+                 WHERE c.parent_id = ?1 AND c.is_directory = 1",
+            )?
+            .query_row(params![dir_id], |row| row.get::<_, Option<u64>>(0))?;
+
+        Ok(match child_min {
+            None => own,             // no child dirs: covered at own epoch
+            Some(0) => 0,            // some child subtree is unlisted
+            Some(cm) => own.min(cm), // weakest link across self + children
+        })
+    }
+
     /// Get all directory paths from the entries table.
     ///
     /// Reconstructs full paths from the integer-keyed tree for backward compat.
@@ -1484,6 +1525,72 @@ mod tests {
         assert_eq!(IndexStore::read_current_epoch(&conn).unwrap(), 2);
     }
 
+    /// `recompute_min_subtree_epoch`: the 0-absorbing min over the dir's own
+    /// `listed_epoch` and every child dir's stored `min_subtree_epoch`.
+    #[test]
+    fn recompute_min_subtree_epoch_cases() {
+        let (store, _dir) = open_temp_store();
+        let conn = IndexStore::open_write_connection(store.db_path()).unwrap();
+
+        // An unlisted dir (listed_epoch = 0) is always 0, regardless of children.
+        let unlisted = insert_entry(&conn, ROOT_ID, "unlisted", true, None);
+        assert_eq!(IndexStore::recompute_min_subtree_epoch(&conn, unlisted).unwrap(), 0);
+
+        // A listed dir with NO child dirs is covered at its own epoch.
+        let leaf = insert_entry(&conn, ROOT_ID, "leaf", true, None);
+        IndexStore::mark_dirs_listed(&conn, &[leaf], 5).unwrap();
+        assert_eq!(
+            IndexStore::recompute_min_subtree_epoch(&conn, leaf).unwrap(),
+            5,
+            "listed-childless ⇒ own epoch"
+        );
+
+        // A listed parent with one complete child (epoch 4) and one incomplete
+        // child (epoch 0) ⇒ 0 (the 0 absorbs).
+        let parent = insert_entry(&conn, ROOT_ID, "parent", true, None);
+        IndexStore::mark_dirs_listed(&conn, &[parent], 9).unwrap();
+        let complete = insert_entry(&conn, parent, "complete", true, None);
+        let incomplete = insert_entry(&conn, parent, "incomplete", true, None);
+        IndexStore::upsert_dir_stats_by_id(
+            &conn,
+            &[
+                DirStatsById {
+                    entry_id: complete,
+                    min_subtree_epoch: 4,
+                    ..Default::default()
+                },
+                DirStatsById {
+                    entry_id: incomplete,
+                    min_subtree_epoch: 0,
+                    ..Default::default()
+                },
+            ],
+        )
+        .unwrap();
+        assert_eq!(
+            IndexStore::recompute_min_subtree_epoch(&conn, parent).unwrap(),
+            0,
+            "an incomplete child absorbs to 0"
+        );
+
+        // With both children complete (4 and 6), the parent is the weakest link
+        // across self (9) and children ⇒ 4.
+        IndexStore::upsert_dir_stats_by_id(
+            &conn,
+            &[DirStatsById {
+                entry_id: incomplete,
+                min_subtree_epoch: 6,
+                ..Default::default()
+            }],
+        )
+        .unwrap();
+        assert_eq!(
+            IndexStore::recompute_min_subtree_epoch(&conn, parent).unwrap(),
+            4,
+            "weakest link = min(own=9, 4, 6) = 4"
+        );
+    }
+
     /// A schema-version mismatch drops + rebuilds; the rebuilt DB still has the
     /// new v13 columns (a write/read round-trip through them succeeds).
     #[test]

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

@@ -436,6 +436,12 @@ async fn verify_and_correct(dir_path: &str, writer: &IndexWriter) -> Vec<String>
                 file_count_delta: stats.recursive_file_count as i32,
                 dir_count_delta: stats.recursive_dir_count as i32,
             });
+            // `scan_subtree` stamped the new dir's `listed_epoch` and its
+            // `ComputeSubtreeAggregates` set its `min_subtree_epoch`, but the
+            // `UpsertEntryV2` that created it earlier dropped every ancestor's
+            // coverage to 0. Recompute up from the parent so a now-fully-listed
+            // subtree lifts ancestor coverage back to exact.
+            let _ = writer.send(WriteMessage::PropagateMinSubtreeEpoch(*p_id));
         }
     }