From 130a9d31d7ba1c605e0bb03d143f9dc91e582e93 Mon Sep 17 00:00:00 2001 From: Piotr Osiewicz <24362066+osiewicz@users.noreply.github.com> Date: Fri, 12 Jun 2026 14:54:29 +0200 Subject: [PATCH] audio: Fix phantom presence in channels (#59195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit # Objective You know that feeling? You join the channel, crack a banger joke to break the ice - yet, nobody laughs. Worse yet, you're swarmed with Slack messages like "Piotr, we can't hear you". After a while you realize that maybe it's for the best as the joke was not as good as you've initially thought, but still: it's possible to get into a state where other participants see you as muted. You don't hear anybody. Bummer. That's what happened when we were hoping to pair with @dinocosta today. I could not hear him. However, he had some logs to share which indicated that while collab thought he's in the room with me, live-kit was rejecting his token as it was revoked. What clanker helped us figure out is that there's a race-y code path where if you disconnect from a given room abruptly, collab can put you in a limbo.
Clanker's idea of what's going on

Found the whole story. Here's what those logs mean and where the bug surface is. What's happening The error at room.rs:1809 is the detach_and_log_err on the task spawned by spawn_room_connection (crates/call/src/call_impl/room.rs:1754-1810). So this is the initial livekit::Room::connect call failing — the LiveKit SDK internally retried 3 times (that's the retrying... (1/3) lines), got 401 invalid token: revoked every time, gave up, and the error propagated out of the ? at line 1762 and was logged-and-dropped. The "revoked" comes from the collab server: leave_room_for_session (crates/collab/src/rpc.rs:4091-4095) calls live_kit.remove_participant(livekit_room, session.user_id().to_string()), and LiveKit Cloud treats a removed participant's token as revoked. Crucially, the LiveKit identity is just the user id, not the connection id, while the DB-side leave_room(connection_id) is per-connection. leave_room_for_session fires from three places: 1. Explicit leave (rpc.rs:1666) 2. connection_lost, after RECONNECT_TIMEOUT expires (rpc.rs:1342) 3. Stale-connection cleanup when re-joining a channel (rpc.rs:3401) So there's a race: an old connection's delayed cleanup (path 2) can fire after the same user has rejoined the room on a new connection and been issued a fresh token — and remove_participant(room, user_id) kicks/revokes the user's current LiveKit participant, because the identity is shared across connections. Quick restart of Zed, two instances, or a reconnect that straddles the server's RECONNECT_TIMEOUT all set this up. Client-side consequences (this connects to your first question) When that initial connect fails: - this.live_kit stays None and this.diagnostics is never created — lines 1780-1791 only run on success. So in this failure mode CallDiagnostics doesn't exist at all; the "stats never overridden" path I described earlier isn't even reached. - There is no retry and no token refresh. The failure is detach_and_log_err'd and that's the end of it. The user stays in the room at the collab level (status is still Online, they appear in the channel), but with no audio and no way to recover short of leaving and rejoining. - Even a collab-level reconnect doesn't help: RejoinRoomResponse (crates/proto/proto/call.proto:58) carries no live_kit_connection_info, and rejoin() never calls spawn_room_connection — the LiveKit connection is established exactly once, in Room::new, with the token from the original join response. So the user-visible symptom is: you're "in" the call, everyone sees you, but you can't hear or be heard, and there's no error surfaced in the UI — just this log line. If you want to fix it, the two angles are server-side (make leave_room_for_session skip the LiveKit removal when the user still has a live connection in that room, or scope LiveKit identities per-connection) and client-side (treat a failed spawn_room_connection as fatal-but-recoverable: surface it, and have rejoin request fresh connection info instead of silently carrying on). Happy to dig into either if you want.

So tl;dr: our token was swapped out underneath us. One way to "fix" it was to just go into another channel, but that was a bummer. We have a good and reliable repro for it though: `kill -9 $ZED_PID` followed by an attempt to rejoin the same channel within 30s would consistently put us in that state. Put another way: you panic and thus do not send a clean "leave room" message to LK. if you rejoin the channel within 30s (after restarting), LK will invalidate the very token you're attempting to use (as it cleans up old tokens). It may also happen without crashing, but that's the most reliable way to repro the issue. Long story short your LK token gets tainted and you can't share the audio anyhow. ## Solution The fix is both client and server-side. On client's side, we now retry the reconnection with a back-off and we grab a fresh token off of reconnect attempts. On server's side, we share the current token on room reconnect attempts. We've also tweaked the call diagnostics to not use `unwrap_or_default` so much: it was hard for us to tell that we're in a totally bogus state and we've only reached that conclusion based on source code analysis. We now show some generic "ok you're in a borked state pls report a bug" message instead. ## Testing We added tests that present the flacky scenario. We've also tried to setup an infra to test the new behaviour, but sadly, local LK instance seems to revoke tokens more leniently than the prod.. OTOH, we are quite confident that once new collab is deployed, we'll be able to mince the new tokens and life is going to be good. Even when running against the current prod instance we could see that our code changed the behaviour for the better, as now we'll actually try to use a new token if one is provided by collab. For that to happen we need to redeploy though. ## Self-Review Checklist: - [x] I've reviewed my own diff for quality, security, and reliability - [x] Unsafe blocks (if any) have justifying comments - [x] The content adheres to Zed's UI standards ([UX/UI](https://github.com/zed-industries/zed/blob/main/CONTRIBUTING.md#uiux-checklist) and [icon](https://github.com/zed-industries/zed/blob/main/crates/icons/README.md) guidelines) - [x] Tests cover the new/changed behavior - [x] Performance impact has been considered and is acceptable Release Notes: - Fixed a race condition that caused collab users to not receive/send any audio to their peers. --------- Co-authored-by: dino Co-authored-by: Dino Co-authored-by: zed-zippy[bot] <234243425+zed-zippy[bot]@users.noreply.github.com> --- crates/call/src/call_impl/room.rs | 171 ++++++++++++------ crates/collab/src/db.rs | 1 + crates/collab/src/db/queries/rooms.rs | 11 ++ crates/collab/src/rpc.rs | 39 ++++ .../collab/tests/integration/channel_tests.rs | 66 +++++++ crates/collab_ui/src/call_stats_modal.rs | 24 ++- crates/livekit_client/src/test.rs | 21 +++ crates/proto/proto/call.proto | 1 + 8 files changed, 270 insertions(+), 64 deletions(-) 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 {