audio: Fix phantom presence in channels

crates/call/src/call_impl/room.rs

@@ -93,6 +93,7 @@ pub struct Room {
     room_update_completed_rx: watch::Receiver<Option<()>>,
     pending_room_update: Option<Task<()>>,
     maintain_connection: Option<Task<Option<()>>>,
+    livekit_connection_task: Option<Task<()>>,
     created: Instant,
 }
 
@@ -123,7 +124,7 @@ impl Room {
         user_store: Entity<UserStore>,
         cx: &mut Context<Self>,
     ) -> Self {
-        spawn_room_connection(livekit_connection_info, cx);
+        let livekit_connection_task = spawn_room_connection(livekit_connection_info, cx);
 
         let maintain_connection = cx.spawn({
             let client = client.clone();
@@ -164,6 +165,7 @@ impl Room {
             user_store,
             follows_by_leader_id_project_id: Default::default(),
             maintain_connection: Some(maintain_connection),
+            livekit_connection_task,
             room_update_completed_tx,
             room_update_completed_rx,
             created: cx.background_executor().now(),
@@ -360,6 +362,7 @@ impl Room {
         self.diagnostics.take();
         self.pending_room_update.take();
         self.maintain_connection.take();
+        self.livekit_connection_task.take();
     }
 
     fn emit_video_track_unsubscribed_events(&self, cx: &mut Context<Self>) {
@@ -400,7 +403,10 @@ impl Room {
 
                             let Some(this) = this.upgrade() else { break };
                             let task = this.update(cx, |this, cx| this.rejoin(cx));
-                            if task.await.log_err().is_some() {
+                            if let Some(live_kit_connection_info) = task.await.log_err() {
+                                this.update(cx, |this, cx| {
+                                    this.ensure_livekit_connection(live_kit_connection_info, cx);
+                                });
                                 return true;
                             } else {
                                 remaining_attempts -= 1;
@@ -447,7 +453,10 @@ impl Room {
         anyhow::bail!("can't reconnect to room: client failed to re-establish connection");
     }
 
-    fn rejoin(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
+    fn rejoin(
+        &mut self,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<Option<proto::LiveKitConnectionInfo>>> {
         let mut projects = HashMap::default();
         let mut reshared_projects = Vec::new();
         let mut rejoined_projects = Vec::new();
@@ -507,7 +516,8 @@ impl Room {
         cx.spawn(async move |this, cx| {
             let response = response.await?;
             let message_id = response.message_id;
-            let response = response.payload;
+            let mut response = response.payload;
+            let live_kit_connection_info = response.live_kit_connection_info.take();
             let room_proto = response.room.context("invalid room")?;
             this.update(cx, |this, cx| {
                 this.status = RoomStatus::Online;
@@ -530,10 +540,22 @@ impl Room {
                 }
 
                 anyhow::Ok(())
-            })?
+            })??;
+            Ok(live_kit_connection_info)
         })
     }
 
+    fn ensure_livekit_connection(
+        &mut self,
+        connection_info: Option<proto::LiveKitConnectionInfo>,
+        cx: &mut Context<Self>,
+    ) {
+        if connection_info.is_none() || self.is_connected(cx) {
+            return;
+        }
+        self.livekit_connection_task = spawn_room_connection(connection_info, cx);
+    }
+
     pub fn id(&self) -> u64 {
         self.id
     }
@@ -1754,60 +1776,93 @@ impl Room {
 fn spawn_room_connection(
     livekit_connection_info: Option<proto::LiveKitConnectionInfo>,
     cx: &mut Context<Room>,
-) {
-    if let Some(connection_info) = livekit_connection_info {
-        cx.spawn(async move |this, cx| {
-            let (room, mut events) =
-                livekit::Room::connect(connection_info.server_url, connection_info.token, cx)
-                    .await?;
+) -> Option<Task<()>> {
+    let mut connection_info = livekit_connection_info?;
+    Some(cx.spawn(async move |this, cx| {
+        let mut backoff = Duration::from_secs(1);
+        loop {
+            match livekit::Room::connect(
+                connection_info.server_url.clone(),
+                connection_info.token.clone(),
+                cx,
+            )
+            .await
+            {
+                Ok((room, mut events)) => {
+                    let weak_room = this.clone();
+                    let Ok(share_microphone) = this.update(cx, |this, cx| {
+                        let _handle_updates = cx.spawn(async move |this, cx| {
+                            while let Some(event) = events.next().await {
+                                if this
+                                    .update(cx, |this, cx| {
+                                        this.livekit_room_updated(event, cx).warn_on_err();
+                                    })
+                                    .is_err()
+                                {
+                                    break;
+                                }
+                            }
+                        });
 
-            let weak_room = this.clone();
-            this.update(cx, |this, cx| {
-                let _handle_updates = cx.spawn(async move |this, cx| {
-                    while let Some(event) = events.next().await {
-                        if this
-                            .update(cx, |this, cx| {
-                                this.livekit_room_updated(event, cx).warn_on_err();
-                            })
-                            .is_err()
-                        {
-                            break;
+                        let muted_by_user = Room::mute_on_join(cx);
+                        this.live_kit = Some(LiveKitRoom {
+                            room: Rc::new(room),
+                            screen_track: LocalTrack::None,
+                            microphone_track: LocalTrack::None,
+                            input_lag_us: None,
+                            next_publish_id: 0,
+                            muted_by_user,
+                            deafened: false,
+                            speaking: false,
+                            _handle_updates,
+                        });
+                        this.diagnostics = Some(cx.new(|cx| CallDiagnostics::new(weak_room, cx)));
+                        cx.notify();
+
+                        // Always open the microphone track on join, even when
+                        // `muted_by_user` is set. Note that the microphone will still
+                        // be muted, as it is still gated in `share_microphone` by
+                        // `muted_by_user`. For users that have `mute_on_join` enabled,
+                        // this moves the Bluetooth profile switch (A2DP -> HFP) (which
+                        // can cause 1-2 seconds of audio silence on some Bluetooth
+                        // headphones) from first unmute to channel join, where
+                        // instability is expected.
+                        if this.can_use_microphone() {
+                            this.share_microphone(cx)
+                        } else {
+                            Task::ready(Ok(()))
                         }
-                    }
-                });
+                    }) else {
+                        return;
+                    };
+                    share_microphone.await.log_err();
+                    return;
+                }
+                Err(error) => {
+                    log::error!("failed to connect to LiveKit room: {error:#}");
+                }
+            }
 
-                let muted_by_user = Room::mute_on_join(cx);
-                this.live_kit = Some(LiveKitRoom {
-                    room: Rc::new(room),
-                    screen_track: LocalTrack::None,
-                    microphone_track: LocalTrack::None,
-                    input_lag_us: None,
-                    next_publish_id: 0,
-                    muted_by_user,
-                    deafened: false,
-                    speaking: false,
-                    _handle_updates,
-                });
-                this.diagnostics = Some(cx.new(|cx| CallDiagnostics::new(weak_room, cx)));
-
-                // Always open the microphone track on join, even when
-                // `muted_by_user` is set. Note that the microphone will still
-                // be muted, as it is still gated in `share_microphone` by
-                // `muted_by_user`. For users that have `mute_on_join` enabled,
-                // this moves the Bluetooth profile switch (A2DP -> HFP) (which
-                // can cause 1-2 seconds of audio silence on some Bluetooth
-                // headphones) from first unmute to channel join, where
-                // instability is expected.
-                if this.can_use_microphone() {
-                    this.share_microphone(cx)
-                } else {
-                    Task::ready(Ok(()))
+            cx.background_executor().timer(backoff).await;
+            backoff = (backoff * 2).min(Duration::from_secs(30));
+
+            // The token we were given may no longer be valid: LiveKit revokes
+            // it when this user's stale connection is cleaned up around the
+            // time the token is issued (e.g. when rejoining a channel right
+            // after a crash). Rejoin the room to obtain fresh connection info
+            // before trying again.
+            let Ok(rejoin) = this.update(cx, |this, cx| this.rejoin(cx)) else {
+                return;
+            };
+            match rejoin.await {
+                Ok(Some(new_connection_info)) => connection_info = new_connection_info,
+                Ok(None) => {}
+                Err(error) => {
+                    log::error!("failed to refresh LiveKit connection info: {error:#}");
                 }
-            })?
-            .await
-        })
-        .detach_and_log_err(cx);
-    }
+            }
+        }
+    }))
 }
 
 struct LiveKitRoom {

crates/collab/src/db.rs

@@ -487,6 +487,7 @@ pub struct RejoinedRoom {
     pub rejoined_projects: Vec<RejoinedProject>,
     pub reshared_projects: Vec<ResharedProject>,
     pub channel: Option<channel::Model>,
+    pub role: ChannelRole,
 }
 
 pub struct ResharedProject {

crates/collab/src/db/queries/rooms.rs

@@ -499,6 +499,16 @@ impl Database {
                 return Err(anyhow!("room does not exist or was already joined"))?;
             }
 
+            let participant = room_participant::Entity::find()
+                .filter(
+                    Condition::all()
+                        .add(room_participant::Column::RoomId.eq(room_id))
+                        .add(room_participant::Column::UserId.eq(user_id)),
+                )
+                .one(&*tx)
+                .await?
+                .context("participant not found")?;
+
             let mut reshared_projects = Vec::new();
             for reshared_project in &rejoin_room.reshared_projects {
                 let project_id = ProjectId::from_proto(reshared_project.project_id);
@@ -591,6 +601,7 @@ impl Database {
                 channel,
                 rejoined_projects,
                 reshared_projects,
+                role: participant.role.unwrap_or(ChannelRole::Member),
             })
         })
         .await

crates/collab/src/rpc.rs

@@ -1493,6 +1493,44 @@ async fn rejoin_room(
             .rejoin_room(request, session.user_id(), session.connection_id)
             .await?;
 
+        // Include fresh LiveKit connection info so that clients whose LiveKit
+        // connection failed (e.g. because their token was revoked by a stale
+        // connection cleanup) can re-establish it.
+        let live_kit_connection_info =
+            session
+                .app_state
+                .livekit_client
+                .as_ref()
+                .and_then(|live_kit| {
+                    let (can_publish, token) = if rejoined_room.role == ChannelRole::Guest {
+                        (
+                            false,
+                            live_kit
+                                .guest_token(
+                                    &rejoined_room.room.livekit_room,
+                                    &session.user_id().to_string(),
+                                )
+                                .trace_err()?,
+                        )
+                    } else {
+                        (
+                            true,
+                            live_kit
+                                .room_token(
+                                    &rejoined_room.room.livekit_room,
+                                    &session.user_id().to_string(),
+                                )
+                                .trace_err()?,
+                        )
+                    };
+
+                    Some(LiveKitConnectionInfo {
+                        server_url: live_kit.url().into(),
+                        token,
+                        can_publish,
+                    })
+                });
+
         response.send(proto::RejoinRoomResponse {
             room: Some(rejoined_room.room.clone()),
             reshared_projects: rejoined_room
@@ -1512,6 +1550,7 @@ async fn rejoin_room(
                 .iter()
                 .map(|rejoined_project| rejoined_project.to_proto())
                 .collect(),
+            live_kit_connection_info,
         })?;
         room_updated(&rejoined_room.room, &session.peer);
 

crates/collab/tests/integration/channel_tests.rs

@@ -366,6 +366,72 @@ async fn test_joining_channel_ancestor_member(
     );
 }
 
+#[gpui::test]
+async fn test_channel_call_recovers_from_livekit_connect_failure(
+    executor: BackgroundExecutor,
+    cx_a: &mut TestAppContext,
+) {
+    let mut server = TestServer::start(executor.clone()).await;
+    let client_a = server.create_client(cx_a, "user_a").await;
+    client_a.initialize_channel_store(cx_a);
+
+    let channel_id = server
+        .make_channel("zed", None, (&client_a, cx_a), &mut [])
+        .await;
+
+    // Simulate LiveKit rejecting the token issued for this join. In
+    // production this happens when the collab server's stale connection
+    // cleanup (`leave_room_for_session`) calls `remove_participant` for this
+    // user's identity just before issuing a new token for the same identity
+    // and room (e.g. rejoining a channel right after an abrupt disconnect or
+    // restart): LiveKit Cloud revokes the freshly issued token and the
+    // client's initial connection fails with "401 invalid token: revoked".
+    let identity = client_a.user_id().unwrap().to_string();
+    server
+        .test_livekit_server
+        .set_token_revoked(&identity, true);
+
+    let active_call_a = cx_a.read(ActiveCall::global);
+    active_call_a
+        .update(cx_a, |active_call, cx| {
+            active_call.join_channel(channel_id, cx)
+        })
+        .await
+        .unwrap();
+    executor.run_until_parked();
+
+    // The collab server considers user A a participant in the call...
+    let room_a =
+        cx_a.read(|cx| active_call_a.read_with(cx, |call, _| call.room().unwrap().clone()));
+    cx_a.read(|cx| {
+        client_a.channel_store().read_with(cx, |channels, _| {
+            assert_participants_eq(
+                channels.channel_participants(channel_id),
+                &[client_a.user_id().unwrap()],
+            );
+        })
+    });
+    // ...but the LiveKit connection failed, so they have no audio.
+    cx_a.read(|cx| room_a.read_with(cx, |room, cx| assert!(!room.is_connected(cx))));
+
+    // Once LiveKit accepts the user's tokens again, the client should
+    // re-establish its LiveKit connection instead of silently staying in the
+    // call without audio.
+    server
+        .test_livekit_server
+        .set_token_revoked(&identity, false);
+    executor.advance_clock(RECONNECT_TIMEOUT);
+    executor.run_until_parked();
+    cx_a.read(|cx| {
+        room_a.read_with(cx, |room, cx| {
+            assert!(
+                room.is_connected(cx),
+                "client should have re-established its LiveKit connection"
+            )
+        })
+    });
+}
+
 #[gpui::test]
 async fn test_channel_room(
     executor: BackgroundExecutor,