diff --git a/crates/call/src/call_impl/room.rs b/crates/call/src/call_impl/room.rs index 108c5a4630..fd5c7ac78a 100644 --- a/crates/call/src/call_impl/room.rs +++ b/crates/call/src/call_impl/room.rs @@ -93,6 +93,7 @@ pub struct Room { room_update_completed_rx: watch::Receiver>, pending_room_update: Option>, maintain_connection: Option>>, + livekit_connection_task: Option>, created: Instant, } @@ -123,7 +124,7 @@ impl Room { user_store: Entity, cx: &mut Context, ) -> 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) { @@ -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) -> Task> { + fn rejoin( + &mut self, + cx: &mut Context, + ) -> Task>> { 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, + cx: &mut Context, + ) { + 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, cx: &mut Context, -) { - 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?; - - 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))); - - // 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(())) - } - })? +) -> Option> { + 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 - }) - .detach_and_log_err(cx); - } + { + 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 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:#}"); + } + } + + 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:#}"); + } + } + } + })) } struct LiveKitRoom { diff --git a/crates/collab/src/db.rs b/crates/collab/src/db.rs index 60e49fc39a..8a9beac933 100644 --- a/crates/collab/src/db.rs +++ b/crates/collab/src/db.rs @@ -487,6 +487,7 @@ pub struct RejoinedRoom { pub rejoined_projects: Vec, pub reshared_projects: Vec, pub channel: Option, + pub role: ChannelRole, } pub struct ResharedProject { diff --git a/crates/collab/src/db/queries/rooms.rs b/crates/collab/src/db/queries/rooms.rs index a04cd53410..8661523a14 100644 --- a/crates/collab/src/db/queries/rooms.rs +++ b/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 diff --git a/crates/collab/src/rpc.rs b/crates/collab/src/rpc.rs index ef46033d12..e89257b70d 100644 --- a/crates/collab/src/rpc.rs +++ b/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); diff --git a/crates/collab/tests/integration/channel_tests.rs b/crates/collab/tests/integration/channel_tests.rs index 329478819e..eac9f57952 100644 --- a/crates/collab/tests/integration/channel_tests.rs +++ b/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, diff --git a/crates/collab_ui/src/call_stats_modal.rs b/crates/collab_ui/src/call_stats_modal.rs index cfbdf82c3b..1481fc1602 100644 --- a/crates/collab_ui/src/call_stats_modal.rs +++ b/crates/collab_ui/src/call_stats_modal.rs @@ -144,12 +144,24 @@ impl Render for CallStatsModal { fn render(&mut self, _window: &mut Window, cx: &mut Context) -> impl IntoElement { let room = active_room(cx); let is_connected = room.is_some(); - let stats = room - .and_then(|room| { - let diagnostics = room.read(cx).diagnostics()?; - Some(diagnostics.read(cx).stats().clone()) - }) - .unwrap_or_default(); + let Some(stats) = room.and_then(|room| { + let diagnostics = room.read(cx).diagnostics()?; + Some(diagnostics.read(cx).stats().clone()) + }) else { + return v_flex() + .key_context("CallStatsModal") + .on_action(cx.listener(Self::dismiss)) + .track_focus(&self.focus_handle) + .elevation_3(cx) + .w(rems(24.)) + .p_4() + .gap_3() + .child( + Label::new("Unable to fetch call statistics") + .size(LabelSize::Large) + .color(Color::Error), + ); + }; let (quality_text, quality_color) = quality_label(stats.connection_quality); diff --git a/crates/livekit_client/src/test.rs b/crates/livekit_client/src/test.rs index 955f92dc19..eff26af425 100644 --- a/crates/livekit_client/src/test.rs +++ b/crates/livekit_client/src/test.rs @@ -56,6 +56,7 @@ pub struct TestServer { pub api_key: String, pub secret_key: String, rooms: Mutex>, + revoked_identities: Mutex>, executor: BackgroundExecutor, } @@ -73,6 +74,7 @@ impl TestServer { api_key, secret_key, rooms: Default::default(), + revoked_identities: Default::default(), executor, }); e.insert(server.clone()); @@ -104,6 +106,18 @@ impl TestServer { } } + /// Simulates LiveKit Cloud revoking an identity's tokens, as happens when + /// `remove_participant` is called for that identity (e.g. by the collab + /// server's stale connection cleanup) around the time a token is issued. + pub fn set_token_revoked(&self, identity: &str, revoked: bool) { + let mut revoked_identities = self.revoked_identities.lock(); + if revoked { + revoked_identities.insert(identity.to_string()); + } else { + revoked_identities.remove(identity); + } + } + pub async fn create_room(&self, room: String) -> Result<()> { self.simulate_random_delay().await; @@ -132,6 +146,13 @@ impl TestServer { let claims = livekit_api::token::validate(&token, &self.secret_key)?; let identity = ParticipantIdentity(claims.sub.unwrap().to_string()); let room_name = claims.video.room.unwrap(); + + if self.revoked_identities.lock().contains(&identity.0) { + anyhow::bail!( + "signal failure: client error: 401 Unauthorized - invalid token: revoked" + ); + } + let mut server_rooms = self.rooms.lock(); let room = (*server_rooms).entry(room_name.to_string()).or_default(); diff --git a/crates/proto/proto/call.proto b/crates/proto/proto/call.proto index 71351fb74c..23a2f0b7e6 100644 --- a/crates/proto/proto/call.proto +++ b/crates/proto/proto/call.proto @@ -59,6 +59,7 @@ message RejoinRoomResponse { Room room = 1; repeated ResharedProject reshared_projects = 2; repeated RejoinedProject rejoined_projects = 3; + optional LiveKitConnectionInfo live_kit_connection_info = 4; } message ResharedProject {