Merge branch 'main' into window_context_2

crates/collab/Cargo.toml

@@ -3,7 +3,7 @@ authors = ["Nathan Sobo <nathan@zed.dev>"]
 default-run = "collab"
 edition = "2021"
 name = "collab"
-version = "0.8.2"
+version = "0.8.3"
 publish = false
 
 [[bin]]

crates/collab/src/db.rs

@@ -175,25 +175,39 @@ impl Database {
                 .map(|participant| participant.user_id)
                 .collect::<Vec<_>>();
 
-            // Delete participants who failed to reconnect.
+            // Delete participants who failed to reconnect and cancel their calls.
+            let mut canceled_calls_to_user_ids = Vec::new();
             room_participant::Entity::delete_many()
                 .filter(stale_participant_filter)
                 .exec(&*tx)
                 .await?;
+            let called_participants = room_participant::Entity::find()
+                .filter(
+                    Condition::all()
+                        .add(
+                            room_participant::Column::CallingUserId
+                                .is_in(stale_participant_user_ids.iter().copied()),
+                        )
+                        .add(room_participant::Column::AnsweringConnectionId.is_null()),
+                )
+                .all(&*tx)
+                .await?;
+            room_participant::Entity::delete_many()
+                .filter(
+                    room_participant::Column::Id
+                        .is_in(called_participants.iter().map(|participant| participant.id)),
+                )
+                .exec(&*tx)
+                .await?;
+            canceled_calls_to_user_ids.extend(
+                called_participants
+                    .into_iter()
+                    .map(|participant| participant.user_id),
+            );
 
             let room = self.get_room(room_id, &tx).await?;
-            let mut canceled_calls_to_user_ids = Vec::new();
-            // Delete the room if it becomes empty and cancel pending calls.
+            // Delete the room if it becomes empty.
             if room.participants.is_empty() {
-                canceled_calls_to_user_ids.extend(
-                    room.pending_participants
-                        .iter()
-                        .map(|pending_participant| UserId::from_proto(pending_participant.user_id)),
-                );
-                room_participant::Entity::delete_many()
-                    .filter(room_participant::Column::RoomId.eq(room_id))
-                    .exec(&*tx)
-                    .await?;
                 project::Entity::delete_many()
                     .filter(project::Column::RoomId.eq(room_id))
                     .exec(&*tx)

crates/collab/src/rpc.rs

@@ -228,7 +228,7 @@ impl Server {
             .add_message_handler(update_buffer_file)
             .add_message_handler(buffer_reloaded)
             .add_message_handler(buffer_saved)
-            .add_request_handler(save_buffer)
+            .add_request_handler(forward_project_request::<proto::SaveBuffer>)
             .add_request_handler(get_users)
             .add_request_handler(fuzzy_search_users)
             .add_request_handler(request_contact)
@@ -1591,51 +1591,6 @@ where
     Ok(())
 }
 
-async fn save_buffer(
-    request: proto::SaveBuffer,
-    response: Response<proto::SaveBuffer>,
-    session: Session,
-) -> Result<()> {
-    let project_id = ProjectId::from_proto(request.project_id);
-    let host_connection_id = {
-        let collaborators = session
-            .db()
-            .await
-            .project_collaborators(project_id, session.connection_id)
-            .await?;
-        collaborators
-            .iter()
-            .find(|collaborator| collaborator.is_host)
-            .ok_or_else(|| anyhow!("host not found"))?
-            .connection_id
-    };
-    let response_payload = session
-        .peer
-        .forward_request(session.connection_id, host_connection_id, request.clone())
-        .await?;
-
-    let mut collaborators = session
-        .db()
-        .await
-        .project_collaborators(project_id, session.connection_id)
-        .await?;
-    collaborators.retain(|collaborator| collaborator.connection_id != session.connection_id);
-    let project_connection_ids = collaborators
-        .iter()
-        .map(|collaborator| collaborator.connection_id);
-    broadcast(
-        Some(host_connection_id),
-        project_connection_ids,
-        |conn_id| {
-            session
-                .peer
-                .forward_send(host_connection_id, conn_id, response_payload.clone())
-        },
-    );
-    response.send(response_payload)?;
-    Ok(())
-}
-
 async fn create_buffer_for_peer(
     request: proto::CreateBufferForPeer,
     session: Session,
@@ -1655,23 +1610,42 @@ async fn update_buffer(
 ) -> Result<()> {
     session.executor.record_backtrace();
     let project_id = ProjectId::from_proto(request.project_id);
-    let project_connection_ids = session
-        .db()
-        .await
-        .project_connection_ids(project_id, session.connection_id)
-        .await?;
+    let mut guest_connection_ids;
+    let mut host_connection_id = None;
+    {
+        let collaborators = session
+            .db()
+            .await
+            .project_collaborators(project_id, session.connection_id)
+            .await?;
+        guest_connection_ids = Vec::with_capacity(collaborators.len() - 1);
+        for collaborator in collaborators.iter() {
+            if collaborator.is_host {
+                host_connection_id = Some(collaborator.connection_id);
+            } else {
+                guest_connection_ids.push(collaborator.connection_id);
+            }
+        }
+    }
+    let host_connection_id = host_connection_id.ok_or_else(|| anyhow!("host not found"))?;
 
     session.executor.record_backtrace();
-
     broadcast(
         Some(session.connection_id),
-        project_connection_ids.iter().copied(),
+        guest_connection_ids,
         |connection_id| {
             session
                 .peer
                 .forward_send(session.connection_id, connection_id, request.clone())
         },
     );
+    if host_connection_id != session.connection_id {
+        session
+            .peer
+            .forward_request(session.connection_id, host_connection_id, request.clone())
+            .await?;
+    }
+
     response.send(proto::Ack {})?;
     Ok(())
 }

crates/collab/src/tests.rs

@@ -18,9 +18,10 @@ use parking_lot::Mutex;
 use project::{Project, WorktreeId};
 use settings::Settings;
 use std::{
+    cell::{Ref, RefCell, RefMut},
     env,
-    ops::Deref,
-    path::{Path, PathBuf},
+    ops::{Deref, DerefMut},
+    path::Path,
     sync::{
         atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
         Arc,
@@ -209,13 +210,10 @@ impl TestServer {
         let client = TestClient {
             client,
             username: name.to_string(),
-            local_projects: Default::default(),
-            remote_projects: Default::default(),
-            next_root_dir_id: 0,
+            state: Default::default(),
             user_store,
             fs,
             language_registry: Arc::new(LanguageRegistry::test()),
-            buffers: Default::default(),
         };
         client.wait_for_current_user(cx).await;
         client
@@ -314,12 +312,16 @@ impl Drop for TestServer {
 struct TestClient {
     client: Arc<Client>,
     username: String,
-    local_projects: Vec<ModelHandle<Project>>,
-    remote_projects: Vec<ModelHandle<Project>>,
-    next_root_dir_id: usize,
+    state: RefCell<TestClientState>,
     pub user_store: ModelHandle<UserStore>,
     language_registry: Arc<LanguageRegistry>,
     fs: Arc<FakeFs>,
+}
+
+#[derive(Default)]
+struct TestClientState {
+    local_projects: Vec<ModelHandle<Project>>,
+    remote_projects: Vec<ModelHandle<Project>>,
     buffers: HashMap<ModelHandle<Project>, HashSet<ModelHandle<language::Buffer>>>,
 }
 
@@ -358,6 +360,38 @@ impl TestClient {
             .await;
     }
 
+    fn local_projects<'a>(&'a self) -> impl Deref<Target = Vec<ModelHandle<Project>>> + 'a {
+        Ref::map(self.state.borrow(), |state| &state.local_projects)
+    }
+
+    fn remote_projects<'a>(&'a self) -> impl Deref<Target = Vec<ModelHandle<Project>>> + 'a {
+        Ref::map(self.state.borrow(), |state| &state.remote_projects)
+    }
+
+    fn local_projects_mut<'a>(&'a self) -> impl DerefMut<Target = Vec<ModelHandle<Project>>> + 'a {
+        RefMut::map(self.state.borrow_mut(), |state| &mut state.local_projects)
+    }
+
+    fn remote_projects_mut<'a>(&'a self) -> impl DerefMut<Target = Vec<ModelHandle<Project>>> + 'a {
+        RefMut::map(self.state.borrow_mut(), |state| &mut state.remote_projects)
+    }
+
+    fn buffers_for_project<'a>(
+        &'a self,
+        project: &ModelHandle<Project>,
+    ) -> impl DerefMut<Target = HashSet<ModelHandle<language::Buffer>>> + 'a {
+        RefMut::map(self.state.borrow_mut(), |state| {
+            state.buffers.entry(project.clone()).or_default()
+        })
+    }
+
+    fn buffers<'a>(
+        &'a self,
+    ) -> impl DerefMut<Target = HashMap<ModelHandle<Project>, HashSet<ModelHandle<language::Buffer>>>> + 'a
+    {
+        RefMut::map(self.state.borrow_mut(), |state| &mut state.buffers)
+    }
+
     fn summarize_contacts(&self, cx: &TestAppContext) -> ContactsSummary {
         self.user_store.read_with(cx, |store, _| ContactsSummary {
             current: store
@@ -431,15 +465,6 @@ impl TestClient {
         let (_, root_view) = cx.add_window(|_| EmptyView);
         cx.add_view(&root_view, |cx| Workspace::test_new(project.clone(), cx))
     }
-
-    fn create_new_root_dir(&mut self) -> PathBuf {
-        format!(
-            "/{}-root-{}",
-            self.username,
-            util::post_inc(&mut self.next_root_dir_id)
-        )
-        .into()
-    }
 }
 
 impl Drop for TestClient {

crates/collab/src/tests/integration_tests.rs

@@ -1629,9 +1629,7 @@ async fn test_project_reconnect(
         })
         .await
         .unwrap();
-    worktree_a2
-        .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
-        .await;
+    deterministic.run_until_parked();
     let worktree2_id = worktree_a2.read_with(cx_a, |tree, _| {
         assert!(tree.as_local().unwrap().is_shared());
         tree.id()
@@ -1692,11 +1690,9 @@ async fn test_project_reconnect(
         .unwrap();
 
     // While client A is disconnected, add and remove worktrees from client A's project.
-    project_a1
-        .update(cx_a, |project, cx| {
-            project.remove_worktree(worktree2_id, cx)
-        })
-        .await;
+    project_a1.update(cx_a, |project, cx| {
+        project.remove_worktree(worktree2_id, cx)
+    });
     let (worktree_a3, _) = project_a1
         .update(cx_a, |p, cx| {
             p.find_or_create_local_worktree("/root-1/dir3", true, cx)
@@ -1820,18 +1816,14 @@ async fn test_project_reconnect(
         })
         .await
         .unwrap();
-    worktree_a4
-        .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
-        .await;
+    deterministic.run_until_parked();
     let worktree4_id = worktree_a4.read_with(cx_a, |tree, _| {
         assert!(tree.as_local().unwrap().is_shared());
         tree.id()
     });
-    project_a1
-        .update(cx_a, |project, cx| {
-            project.remove_worktree(worktree3_id, cx)
-        })
-        .await;
+    project_a1.update(cx_a, |project, cx| {
+        project.remove_worktree(worktree3_id, cx)
+    });
     deterministic.run_until_parked();
 
     // While client B is disconnected, mutate a buffer on both the host and the guest.