WIP

crates/collab/migrations/20230103200902_replace_is_completed_with_completed_scan_id.sql

@@ -0,0 +1,3 @@
+ALTER TABLE worktrees
+    DROP COLUMN is_complete,
+    ADD COLUMN completed_scan_id INT8;

crates/collab/src/db.rs

@@ -1443,7 +1443,7 @@ impl Database {
                         removed_entries: Default::default(),
                         diagnostic_summaries: Default::default(),
                         scan_id: db_worktree.scan_id as u64,
-                        is_complete: db_worktree.is_complete,
+                        completed_scan_id: db_worktree.completed_scan_id as u64,
                     };
 
                     let rejoined_worktree = rejoined_project
@@ -1997,7 +1997,7 @@ impl Database {
                         root_name: ActiveValue::set(worktree.root_name.clone()),
                         visible: ActiveValue::set(worktree.visible),
                         scan_id: ActiveValue::set(0),
-                        is_complete: ActiveValue::set(false),
+                        completed_scan_id: ActiveValue::set(0),
                     }
                 }))
                 .exec(&*tx)
@@ -2091,7 +2091,7 @@ impl Database {
                 root_name: ActiveValue::set(worktree.root_name.clone()),
                 visible: ActiveValue::set(worktree.visible),
                 scan_id: ActiveValue::set(0),
-                is_complete: ActiveValue::set(false),
+                completed_scan_id: ActiveValue::set(0),
             }))
             .on_conflict(
                 OnConflict::columns([worktree::Column::ProjectId, worktree::Column::Id])
@@ -2141,7 +2141,11 @@ impl Database {
                 project_id: ActiveValue::set(project_id),
                 root_name: ActiveValue::set(update.root_name.clone()),
                 scan_id: ActiveValue::set(update.scan_id as i64),
-                is_complete: ActiveValue::set(update.is_last_update),
+                completed_scan_id: if update.is_last_update {
+                    ActiveValue::set(update.scan_id as i64)
+                } else {
+                    ActiveValue::default()
+                },
                 abs_path: ActiveValue::set(update.abs_path.clone()),
                 ..Default::default()
             })
@@ -2381,7 +2385,7 @@ impl Database {
                             entries: Default::default(),
                             diagnostic_summaries: Default::default(),
                             scan_id: db_worktree.scan_id as u64,
-                            is_complete: db_worktree.is_complete,
+                            completed_scan_id: db_worktree.completed_scan_id as u64,
                         },
                     )
                 })
@@ -3039,7 +3043,7 @@ pub struct RejoinedWorktree {
     pub removed_entries: Vec<u64>,
     pub diagnostic_summaries: Vec<proto::DiagnosticSummary>,
     pub scan_id: u64,
-    pub is_complete: bool,
+    pub completed_scan_id: u64,
 }
 
 pub struct LeftRoom {
@@ -3093,7 +3097,7 @@ pub struct Worktree {
     pub entries: Vec<proto::Entry>,
     pub diagnostic_summaries: Vec<proto::DiagnosticSummary>,
     pub scan_id: u64,
-    pub is_complete: bool,
+    pub completed_scan_id: u64,
 }
 
 #[cfg(test)]

crates/collab/src/db/worktree.rs

@@ -11,8 +11,10 @@ pub struct Model {
     pub abs_path: String,
     pub root_name: String,
     pub visible: bool,
+    /// The last scan for which we've observed entries. It may be in progress.
     pub scan_id: i64,
-    pub is_complete: bool,
+    /// The last scan that fully completed.
+    pub completed_scan_id: i64,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

crates/collab/src/rpc.rs

@@ -1048,7 +1048,7 @@ async fn rejoin_room(
                     updated_entries: worktree.updated_entries,
                     removed_entries: worktree.removed_entries,
                     scan_id: worktree.scan_id,
-                    is_last_update: worktree.is_complete,
+                    is_last_update: worktree.completed_scan_id == worktree.scan_id,
                 };
                 for update in proto::split_worktree_update(message, MAX_CHUNK_SIZE) {
                     session.peer.send(session.connection_id, update.clone())?;
@@ -1361,7 +1361,7 @@ async fn join_project(
             updated_entries: worktree.entries,
             removed_entries: Default::default(),
             scan_id: worktree.scan_id,
-            is_last_update: worktree.is_complete,
+            is_last_update: worktree.scan_id == worktree.completed_scan_id,
         };
         for update in proto::split_worktree_update(message, MAX_CHUNK_SIZE) {
             session.peer.send(session.connection_id, update.clone())?;

crates/collab/src/tests/randomized_integration_tests.rs

@@ -17,7 +17,7 @@ use project::{search::SearchQuery, Project};
 use rand::prelude::*;
 use std::{env, path::PathBuf, sync::Arc};
 
-#[gpui::test(iterations = 1, seed = 4742)]
+#[gpui::test(iterations = 100)]
 async fn test_random_collaboration(
     cx: &mut TestAppContext,
     deterministic: Arc<Deterministic>,

crates/project/src/worktree.rs

@@ -94,7 +94,7 @@ pub struct Snapshot {
     entries_by_path: SumTree<Entry>,
     entries_by_id: SumTree<PathEntry>,
     scan_id: usize,
-    is_complete: bool,
+    completed_scan_id: usize,
 }
 
 #[derive(Clone)]
@@ -230,7 +230,7 @@ impl Worktree {
             entries_by_path: Default::default(),
             entries_by_id: Default::default(),
             scan_id: 0,
-            is_complete: false,
+            completed_scan_id: 0,
         };
 
         let (updates_tx, mut updates_rx) = mpsc::unbounded();
@@ -423,8 +423,8 @@ impl LocalWorktree {
                     root_char_bag,
                     entries_by_path: Default::default(),
                     entries_by_id: Default::default(),
-                    scan_id: 0,
+                    scan_id: 1,
-                    is_complete: true,
+                    completed_scan_id: 0,
                 },
             };
             if let Some(metadata) = metadata {
@@ -1002,7 +1002,7 @@ impl LocalWorktree {
                             entries_by_path: Default::default(),
                             entries_by_id: Default::default(),
                             scan_id: 0,
-                            is_complete: true,
+                            completed_scan_id: 0,
                         },
                     };
                     while let Some(snapshot) = snapshots_rx.recv().await {
@@ -1091,7 +1091,7 @@ impl RemoteWorktree {
     }
 
     fn observed_snapshot(&self, scan_id: usize) -> bool {
-        self.scan_id > scan_id || (self.scan_id == scan_id && self.is_complete)
+        self.completed_scan_id >= scan_id
     }
 
     fn wait_for_snapshot(&mut self, scan_id: usize) -> impl Future<Output = Result<()>> {
@@ -1254,7 +1254,9 @@ impl Snapshot {
         self.entries_by_path.edit(entries_by_path_edits, &());
         self.entries_by_id.edit(entries_by_id_edits, &());
         self.scan_id = update.scan_id as usize;
-        self.is_complete = update.is_last_update;
+        if update.is_last_update {
+            self.completed_scan_id = update.scan_id as usize;
+        }
 
         Ok(())
     }
@@ -1466,7 +1468,7 @@ impl LocalSnapshot {
             updated_entries,
             removed_entries,
             scan_id: self.scan_id as u64,
-            is_last_update: true,
+            is_last_update: self.completed_scan_id == self.scan_id,
         }
     }
 
@@ -3437,7 +3439,7 @@ mod tests {
                 root_name: Default::default(),
                 root_char_bag: Default::default(),
                 scan_id: 0,
-                is_complete: true,
+                completed_scan_id: 0,
             },
         };
         initial_snapshot.insert_entry(