diff --git a/crates/call/src/call_impl/room.rs b/crates/call/src/call_impl/room.rs index e72f375847..108c5a4630 100644 --- a/crates/call/src/call_impl/room.rs +++ b/crates/call/src/call_impl/room.rs @@ -93,7 +93,6 @@ pub struct Room { room_update_completed_rx: watch::Receiver>, pending_room_update: Option>, maintain_connection: Option>>, - livekit_connection_task: Option>, created: Instant, } @@ -124,7 +123,7 @@ impl Room { user_store: Entity, cx: &mut Context, ) -> Self { - let livekit_connection_task = spawn_room_connection(livekit_connection_info, cx); + spawn_room_connection(livekit_connection_info, cx); let maintain_connection = cx.spawn({ let client = client.clone(); @@ -165,7 +164,6 @@ 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(), @@ -362,7 +360,6 @@ 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) { @@ -403,10 +400,7 @@ impl Room { let Some(this) = this.upgrade() else { break }; let task = this.update(cx, |this, cx| this.rejoin(cx)); - 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); - }); + if task.await.log_err().is_some() { return true; } else { remaining_attempts -= 1; @@ -453,10 +447,7 @@ 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(); @@ -516,8 +507,7 @@ impl Room { cx.spawn(async move |this, cx| { let response = response.await?; let message_id = response.message_id; - let mut response = response.payload; - let live_kit_connection_info = response.live_kit_connection_info.take(); + let response = response.payload; let room_proto = response.room.context("invalid room")?; this.update(cx, |this, cx| { this.status = RoomStatus::Online; @@ -540,22 +530,10 @@ 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 } @@ -1773,100 +1751,63 @@ impl Room { } } -/// The number of times we attempt to establish the LiveKit connection before -/// giving up. Retries cover transient failures (e.g. a network blip); a -/// genuinely bad token won't be fixed by retrying, so the bound keeps us from -/// hammering LiveKit indefinitely. -const LIVEKIT_CONNECT_ATTEMPTS: usize = 5; - fn spawn_room_connection( livekit_connection_info: Option, cx: &mut Context, -) -> Option> { - let connection_info = livekit_connection_info?; - Some(cx.spawn(async move |this, cx| { - // Retry the LiveKit connection with backoff, but deliberately do NOT - // ask collab for a fresh token here: a refreshed token is only needed - // when the collab connection itself is re-established, which is handled - // separately in `maintain_connection` via `ensure_livekit_connection`. - // Coupling this loop to `rejoin` previously turned an isolated LiveKit - // failure into a channel-wide `RejoinRoom` storm. - let mut backoff = Duration::from_secs(1); - for attempt in 1..=LIVEKIT_CONNECT_ATTEMPTS { - 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; - } - } - }); +) { + 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 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(())) + 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; } - }) else { - return; - }; - share_microphone.await.log_err(); - return; - } - Err(error) => { - log::error!( - "failed to connect to LiveKit room (attempt {attempt}/{LIVEKIT_CONNECT_ATTEMPTS}): {error:#}" - ); - } - } + } + }); - if attempt < LIVEKIT_CONNECT_ATTEMPTS { - cx.background_executor().timer(backoff).await; - backoff = (backoff * 2).min(Duration::from_secs(10)); - } - } + 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))); - log::error!( - "giving up on LiveKit room connection after {LIVEKIT_CONNECT_ATTEMPTS} attempts" - ); - })) + // 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(())) + } + })? + .await + }) + .detach_and_log_err(cx); + } } struct LiveKitRoom { diff --git a/crates/collab/src/db.rs b/crates/collab/src/db.rs index 8a9beac933..60e49fc39a 100644 --- a/crates/collab/src/db.rs +++ b/crates/collab/src/db.rs @@ -487,7 +487,6 @@ 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 8661523a14..a04cd53410 100644 --- a/crates/collab/src/db/queries/rooms.rs +++ b/crates/collab/src/db/queries/rooms.rs @@ -499,16 +499,6 @@ 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); @@ -601,7 +591,6 @@ 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 337e363db6..a591e1a57a 100644 --- a/crates/collab/src/rpc.rs +++ b/crates/collab/src/rpc.rs @@ -1340,7 +1340,7 @@ async fn connection_lost( _ = executor.sleep(RECONNECT_TIMEOUT).fuse() => { log::info!("connection lost, removing all resources for user:{}, connection:{:?}", session.user_id(), session.connection_id); - leave_room_for_session(&session, session.connection_id, None).await.trace_err(); + leave_room_for_session(&session, session.connection_id).await.trace_err(); leave_channel_buffers_for_session(&session) .await .trace_err(); @@ -1494,44 +1494,6 @@ 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 @@ -1551,7 +1513,6 @@ async fn rejoin_room( .iter() .map(|rejoined_project| rejoined_project.to_proto()) .collect(), - live_kit_connection_info, })?; room_updated(&rejoined_room.room, &session.peer); @@ -1703,7 +1664,7 @@ async fn leave_room( response: Response, session: MessageContext, ) -> Result<()> { - leave_room_for_session(&session, session.connection_id, None).await?; + leave_room_for_session(&session, session.connection_id).await?; response.send(proto::Ack {})?; Ok(()) } @@ -3438,7 +3399,7 @@ async fn join_channel_internal( "cleaning up stale connection", ); drop(db); - leave_room_for_session(&session, connection, Some(channel_id)).await?; + leave_room_for_session(&session, connection).await?; db = session.db().await; } @@ -4068,11 +4029,7 @@ async fn update_user_contacts(user_id: UserId, session: &Session) -> Result<()> Ok(()) } -async fn leave_room_for_session( - session: &Session, - connection_id: ConnectionId, - rejoining_channel_id: Option, -) -> Result<()> { +async fn leave_room_for_session(session: &Session, connection_id: ConnectionId) -> Result<()> { let mut contacts_to_update = HashSet::default(); let room_id; @@ -4081,7 +4038,6 @@ async fn leave_room_for_session( let delete_livekit_room; let room; let channel; - let left_channel_id; if let Some(mut left_room) = session.db().await.leave_room(connection_id).await? { contacts_to_update.insert(session.user_id()); @@ -4096,23 +4052,12 @@ async fn leave_room_for_session( delete_livekit_room = left_room.deleted; room = mem::take(&mut left_room.room); channel = mem::take(&mut left_room.channel); - left_channel_id = channel.as_ref().map(|channel| channel.id); room_updated(&room, &session.peer); } else { return Ok(()); } - // When this cleanup is part of rejoining the same channel, the user is - // immediately re-entering this LiveKit room with the same identity, which - // LiveKit handles by evicting the prior session. Calling - // `remove_participant` here is therefore redundant, and worse: on LiveKit - // Cloud it revokes the identity's tokens (including the freshly issued one - // for the rejoin), leaving the user connected to collab but unable to join - // audio. Skip the removal in that case. - let skip_livekit_removal = - rejoining_channel_id.is_some() && left_channel_id == rejoining_channel_id; - if let Some(channel) = channel { channel_updated( &channel, @@ -4145,12 +4090,10 @@ async fn leave_room_for_session( } if let Some(live_kit) = session.app_state.livekit_client.as_ref() { - if !skip_livekit_removal { - live_kit - .remove_participant(livekit_room.clone(), session.user_id().to_string()) - .await - .trace_err(); - } + live_kit + .remove_participant(livekit_room.clone(), session.user_id().to_string()) + .await + .trace_err(); if delete_livekit_room { live_kit.delete_room(livekit_room).await.trace_err(); diff --git a/crates/collab/tests/integration/channel_tests.rs b/crates/collab/tests/integration/channel_tests.rs index ab4754082e..329478819e 100644 --- a/crates/collab/tests/integration/channel_tests.rs +++ b/crates/collab/tests/integration/channel_tests.rs @@ -366,76 +366,6 @@ async fn test_joining_channel_ancestor_member( ); } -#[gpui::test] -async fn test_rejoining_channel_does_not_remove_livekit_participant( - executor: BackgroundExecutor, - cx_a1: &mut TestAppContext, - cx_a2: &mut TestAppContext, -) { - let mut server = TestServer::start(executor.clone()).await; - // Two connections for the same user. The second one joining the same - // channel reproduces the production scenario where the server runs - // stale-connection cleanup for the first connection (as it does after an - // abrupt disconnect/crash followed by a rejoin within RECONNECT_TIMEOUT). - // Mirror LiveKit Cloud: removing a participant revokes their tokens. This - // is what turns the redundant `remove_participant` during stale cleanup - // into a user-visible failure. - server - .test_livekit_server - .set_revoke_tokens_on_removal(true); - - let client_a1 = server.create_client(cx_a1, "user_a").await; - let client_a2 = server.create_client(cx_a2, "user_a").await; - client_a1.initialize_channel_store(cx_a1); - - let channel_id = server - .make_channel("zed", None, (&client_a1, cx_a1), &mut []) - .await; - - let active_call_a1 = cx_a1.read(ActiveCall::global); - active_call_a1 - .update(cx_a1, |active_call, cx| { - active_call.join_channel(channel_id, cx) - }) - .await - .unwrap(); - executor.run_until_parked(); - - // The second connection joins the same channel, triggering stale-connection - // cleanup of the first connection on the server. - let active_call_a2 = cx_a2.read(ActiveCall::global); - active_call_a2 - .update(cx_a2, |active_call, cx| { - active_call.join_channel(channel_id, cx) - }) - .await - .unwrap(); - executor.run_until_parked(); - - // The user-visible symptom: with the bug, stale cleanup revokes the - // rejoining user's token and they end up in the call with no audio. - let room_a2 = - cx_a2.read(|cx| active_call_a2.read_with(cx, |call, _| call.room().unwrap().clone())); - cx_a2.read(|cx| { - room_a2.read_with(cx, |room, cx| { - assert!( - room.is_connected(cx), - "rejoining the same channel should not break audio" - ) - }) - }); - - // The mechanism: that cleanup must NOT have removed the user from the - // LiveKit room, since they are immediately rejoining it. - let identity = client_a2.user_id().unwrap().to_string(); - assert!( - !server - .test_livekit_server - .participant_was_removed(&identity), - "rejoining the same channel should not remove the LiveKit participant" - ); -} - #[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 1481fc1602..cfbdf82c3b 100644 --- a/crates/collab_ui/src/call_stats_modal.rs +++ b/crates/collab_ui/src/call_stats_modal.rs @@ -144,24 +144,12 @@ 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 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 stats = room + .and_then(|room| { + let diagnostics = room.read(cx).diagnostics()?; + Some(diagnostics.read(cx).stats().clone()) + }) + .unwrap_or_default(); 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 b07087bbc2..955f92dc19 100644 --- a/crates/livekit_client/src/test.rs +++ b/crates/livekit_client/src/test.rs @@ -56,14 +56,6 @@ pub struct TestServer { pub api_key: String, pub secret_key: String, rooms: Mutex>, - revoked_identities: Mutex>, - /// Log of `(room, identity)` pairs passed to `remove_participant`, for - /// tests that need to assert whether a participant was forcibly removed. - removed_participants: Mutex>, - /// When set, `remove_participant` revokes the identity's tokens, mirroring - /// LiveKit Cloud (whose docs state removal "immediately revokes their - /// access token"). Off by default so unrelated tests are unaffected. - revoke_tokens_on_removal: AtomicBool, executor: BackgroundExecutor, } @@ -81,9 +73,6 @@ impl TestServer { api_key, secret_key, rooms: Default::default(), - revoked_identities: Default::default(), - removed_participants: Default::default(), - revoke_tokens_on_removal: AtomicBool::new(false), executor, }); e.insert(server.clone()); @@ -115,20 +104,6 @@ impl TestServer { } } - /// Makes `remove_participant` revoke the removed identity's tokens, as - /// LiveKit Cloud does. Used to reproduce the revoked-token failure. - pub fn set_revoke_tokens_on_removal(&self, revoke: bool) { - self.revoke_tokens_on_removal.store(revoke, SeqCst); - } - - /// Whether `remove_participant` was ever called for the given identity. - pub fn participant_was_removed(&self, identity: &str) -> bool { - self.removed_participants - .lock() - .iter() - .any(|(_, removed_identity)| removed_identity == identity) - } - pub async fn create_room(&self, room: String) -> Result<()> { self.simulate_random_delay().await; @@ -157,13 +132,6 @@ 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(); @@ -277,14 +245,6 @@ impl TestServer { ) -> Result<()> { self.simulate_random_delay().await; - self.removed_participants - .lock() - .push((room_name.clone(), identity.0.clone())); - - if self.revoke_tokens_on_removal.load(SeqCst) { - self.revoked_identities.lock().insert(identity.0.clone()); - } - let mut server_rooms = self.rooms.lock(); let room = server_rooms .get_mut(&room_name) diff --git a/crates/proto/proto/call.proto b/crates/proto/proto/call.proto index 23a2f0b7e6..aa79014959 100644 --- a/crates/proto/proto/call.proto +++ b/crates/proto/proto/call.proto @@ -59,7 +59,11 @@ message RejoinRoomResponse { Room room = 1; repeated ResharedProject reshared_projects = 2; repeated RejoinedProject rejoined_projects = 3; - optional LiveKitConnectionInfo live_kit_connection_info = 4; + // Field 4 (live_kit_connection_info) was used by the reverted channel-rejoin + // LiveKit token-refresh fix (#59388/#59195). Reserved so the number is not + // reused with an incompatible type if/when that fix is reattempted. + reserved 4; + reserved "live_kit_connection_info"; } message ResharedProject {