collab: Fix unable to rejoin shared project after leaving a call (#50630)

When a downstream project was disconnected from the host (e.g. the guest
left the call), `disconnected_from_host_internal` did not clear
`client_subscriptions`. These subscriptions hold entries in the
`Client`'s entity subscription map, so a subsequent
`join_remote_project` with the same project ID would fail with "already
subscribed to entity".

The fix adds `self.client_subscriptions.clear()` to
`disconnected_from_host_internal`, matching what `unshare_internal`
already does for the host side.

Before you mark this PR as ready for review, make sure that you have:
- [x] Added a solid test coverage and/or screenshots from doing manual
testing
- [x] Done a self-review taking into account security and performance
aspects

Release Notes:

- collab: Fix unable to rejoin project bug ("already subscribed to
entity")

crates/collab/tests/integration/git_tests.rs

@@ -9,7 +9,6 @@ use serde_json::json;
 use util::{path, rel_path::rel_path};
 use workspace::{MultiWorkspace, Workspace};
 
-//
 use crate::TestServer;
 
 #[gpui::test]

crates/collab/tests/integration/integration_tests.rs

@@ -7205,3 +7205,89 @@ async fn test_remote_git_branches(
 
     assert_eq!(host_branch.name(), "totally-new-branch");
 }
+
+#[gpui::test]
+async fn test_guest_can_rejoin_shared_project_after_leaving_call(
+    executor: BackgroundExecutor,
+    cx_a: &mut TestAppContext,
+    cx_b: &mut TestAppContext,
+    cx_c: &mut TestAppContext,
+) {
+    let mut server = TestServer::start(executor.clone()).await;
+    let client_a = server.create_client(cx_a, "user_a").await;
+    let client_b = server.create_client(cx_b, "user_b").await;
+    let client_c = server.create_client(cx_c, "user_c").await;
+
+    server
+        .create_room(&mut [(&client_a, cx_a), (&client_b, cx_b), (&client_c, cx_c)])
+        .await;
+
+    client_a
+        .fs()
+        .insert_tree(
+            path!("/project"),
+            json!({
+                "file.txt": "hello\n",
+            }),
+        )
+        .await;
+
+    let (project_a, _worktree_id) = client_a.build_local_project(path!("/project"), cx_a).await;
+    let active_call_a = cx_a.read(ActiveCall::global);
+    let project_id = active_call_a
+        .update(cx_a, |call, cx| call.share_project(project_a.clone(), cx))
+        .await
+        .unwrap();
+
+    let _project_b = client_b.join_remote_project(project_id, cx_b).await;
+    executor.run_until_parked();
+
+    // third client joins call to prevent room from being torn down
+    let _project_c = client_c.join_remote_project(project_id, cx_c).await;
+    executor.run_until_parked();
+
+    let active_call_b = cx_b.read(ActiveCall::global);
+    active_call_b
+        .update(cx_b, |call, cx| call.hang_up(cx))
+        .await
+        .unwrap();
+    executor.run_until_parked();
+
+    let user_id_b = client_b.current_user_id(cx_b).to_proto();
+    let active_call_a = cx_a.read(ActiveCall::global);
+    active_call_a
+        .update(cx_a, |call, cx| call.invite(user_id_b, None, cx))
+        .await
+        .unwrap();
+    executor.run_until_parked();
+    let active_call_b = cx_b.read(ActiveCall::global);
+    active_call_b
+        .update(cx_b, |call, cx| call.accept_incoming(cx))
+        .await
+        .unwrap();
+    executor.run_until_parked();
+
+    let _project_b2 = client_b.join_remote_project(project_id, cx_b).await;
+    executor.run_until_parked();
+
+    project_a.read_with(cx_a, |project, _| {
+        let guest_count = project
+            .collaborators()
+            .values()
+            .filter(|c| !c.is_host)
+            .count();
+
+        assert_eq!(
+            guest_count, 2,
+            "host should have exactly one guest collaborator after rejoin"
+        );
+    });
+
+    _project_b.read_with(cx_b, |project, _| {
+        assert_eq!(
+            project.client_subscriptions().len(),
+            0,
+            "We should clear all host subscriptions after leaving the project"
+        );
+    })
+}

crates/project/src/project.rs

@@ -1942,6 +1942,11 @@ impl Project {
         }
     }
 
+    #[cfg(feature = "test-support")]
+    pub fn client_subscriptions(&self) -> &Vec<client::Subscription> {
+        &self.client_subscriptions
+    }
+
     #[cfg(feature = "test-support")]
     pub async fn example(
         root_paths: impl IntoIterator<Item = &Path>,
@@ -2741,6 +2746,7 @@ impl Project {
         } = &mut self.client_state
         {
             *sharing_has_stopped = true;
+            self.client_subscriptions.clear();
             self.collaborators.clear();
             self.worktree_store.update(cx, |store, cx| {
                 store.disconnected_from_host(cx);