Introduce completed_scan_id to worktree

We need to know the most recent scan id we have actually completed. This is to
handle the case where a guest disconnects when we're in the middle of streaming
worktree entries to them. When they reconnect, they need to report a scan_id
from before we started streaming the entries, because we have no record of when
the stream was interrupted.

Next failure:
SEED=5051 ITERATIONS=1 OPERATIONS=200 cargo test --release --package=collab random -- --nocapture

crates/call/src/room.rs

@@ -382,7 +382,7 @@ impl Room {
                                 let worktree = worktree.read(cx);
                                 proto::RejoinWorktree {
                                     id: worktree.id().to_proto(),
-                                    scan_id: worktree.scan_id() as u64,
+                                    scan_id: worktree.completed_scan_id() as u64,
                                 }
                             })
                             .collect(),

crates/collab/migrations.sqlite/20221109000000_test_schema.sql

@@ -57,7 +57,7 @@ CREATE TABLE "worktrees" (
     "abs_path" VARCHAR NOT NULL,
     "visible" BOOL NOT NULL,
     "scan_id" INTEGER NOT NULL,
-    "is_complete" BOOL NOT NULL,
+    "completed_scan_id" INTEGER NOT NULL,
     PRIMARY KEY(project_id, id)
 );
 CREATE INDEX "index_worktrees_on_project_id" ON "worktrees" ("project_id");

crates/project/src/worktree.rs

@@ -344,6 +344,13 @@ impl Worktree {
         }
     }
 
+    pub fn completed_scan_id(&self) -> usize {
+        match self {
+            Worktree::Local(worktree) => worktree.snapshot.completed_scan_id,
+            Worktree::Remote(worktree) => worktree.snapshot.completed_scan_id,
+        }
+    }
+
     pub fn is_visible(&self) -> bool {
         match self {
             Worktree::Local(worktree) => worktree.visible,
@@ -423,7 +430,7 @@ impl LocalWorktree {
                     root_char_bag,
                     entries_by_path: Default::default(),
                     entries_by_id: Default::default(),
-                    scan_id: 1,
+                    scan_id: 0,
                     completed_scan_id: 0,
                 },
             };
@@ -955,8 +962,9 @@ impl LocalWorktree {
                     if let Some(old_path) = old_path {
                         snapshot.remove_path(&old_path);
                     }
+                    snapshot.scan_started();
                     inserted_entry = snapshot.insert_entry(entry, fs.as_ref());
-                    snapshot.scan_id += 1;
+                    snapshot.scan_completed();
                 }
                 this.poll_snapshot(true, cx);
                 Ok(inserted_entry)
@@ -1345,6 +1353,14 @@ impl Snapshot {
         &self.root_name
     }
 
+    pub fn scan_started(&mut self) {
+        self.scan_id += 1;
+    }
+
+    pub fn scan_completed(&mut self) {
+        self.completed_scan_id = self.scan_id;
+    }
+
     pub fn scan_id(&self) -> usize {
         self.scan_id
     }
@@ -2250,7 +2266,8 @@ impl BackgroundScanner {
         let is_dir;
         let next_entry_id;
         {
-            let snapshot = self.snapshot.lock();
+            let mut snapshot = self.snapshot.lock();
+            snapshot.scan_started();
             root_char_bag = snapshot.root_char_bag;
             root_abs_path = snapshot.abs_path.clone();
             root_inode = snapshot.root_entry().map(|e| e.inode);
@@ -2316,6 +2333,8 @@ impl BackgroundScanner {
                     }
                 })
                 .await;
+
+            self.snapshot.lock().scan_completed();
         }
 
         Ok(())
@@ -2443,7 +2462,8 @@ impl BackgroundScanner {
         let root_abs_path;
         let next_entry_id;
         {
-            let snapshot = self.snapshot.lock();
+            let mut snapshot = self.snapshot.lock();
+            snapshot.scan_started();
             root_char_bag = snapshot.root_char_bag;
             root_abs_path = snapshot.abs_path.clone();
             next_entry_id = snapshot.next_entry_id.clone();
@@ -2468,7 +2488,6 @@ impl BackgroundScanner {
         let (scan_queue_tx, scan_queue_rx) = channel::unbounded();
         {
             let mut snapshot = self.snapshot.lock();
-            snapshot.scan_id += 1;
             for event in &events {
                 if let Ok(path) = event.path.strip_prefix(&root_canonical_path) {
                     snapshot.remove_path(path);
@@ -2555,6 +2574,7 @@ impl BackgroundScanner {
 
         self.update_ignore_statuses().await;
         self.update_git_repositories();
+        self.snapshot.lock().scan_completed();
         true
     }