yenru0/zed
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

collab: Revert livekit changes (#59733)

Browse Source 
- **Revert "livekit: Preserve tokens on channel rejoin (#59388)"**
- **Revert "call: Log LiveKit connection info refresh outcomes in retry
loop (#59205)"**
- **Revert "audio: Fix phantom presence in channels (#59195)"**
- **proto: Reserve RejoinRoomResponse field 4 after revert**

We've observed a spike of phantom collaborator issues after this fixes.
This sucks and I'm quite unhappy about it.

# Objective

- Describe the objective or issue this PR addresses.
- If you're fixing a specific issue, use "Fixes #X" for each issue as
[described in the GitHub
docs](https://docs.github.com/en/issues/tracking-your-work-with-issues/using-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword).

## Solution

- Describe the solution used to achieve the objective above.

## Testing

- Did you test these changes? If so, how?
- Are there any parts that need more testing?
- How can other people (reviewers) test your changes? Is there anything
specific they need to know?
- If relevant, what platforms did you test these changes on, and are
there any important ones you can't test?

## Self-Review Checklist:

- [ ] I've reviewed my own diff for quality, security, and reliability
- [ ] Unsafe blocks (if any) have justifying comments
- [ ] 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)
- [ ] Tests cover the new/changed behavior
- [ ] Performance impact has been considered and is acceptable

## Showcase

> This section is optional. If this PR does not include a visual change
or does not add a new user-facing feature, you can delete this section.

- Help others understand the result of this PR by showcasing your
awesome work!
- If this PR includes a visual change, consider adding a screenshot,
GIF, or video
- A before/after comparison is very useful for changes to existing
features!

While a showcase should aim to be brief and digestible, you can use a
toggleable section to save space on longer showcases:

<details>
  <summary>Click to view showcase</summary>

My super cool demos here

</details>

---

Release Notes:

- N/A

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Piotr OsiewiczandClaude Opus 4.8 authored and GitHub committed 2026-06-23 09:13:28 +00:00
1 parent 9c02d74cac
commit f743291f1c
8 files changed
+74 -320
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/call/src/call_impl/room.rs
+55 -114
View File
@@ -93,7 +93,6 @@ 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,
}
@@ -124,7 +123,7 @@ impl Room {
user_store: Entity<UserStore>,
cx: &mut Context<Self>,
) -> 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<Self>) {
@@ -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<Self>,
) -> Task<Result<Option<proto::LiveKitConnectionInfo>>> {
fn rejoin(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
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<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
}
@@ -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<proto::LiveKitConnectionInfo>,
cx: &mut Context<Room>,
) -> Option<Task<()>> {
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 {
crates/collab/src/db.rs
-1
View File
@@ -487,7 +487,6 @@ 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
-11
View File
@@ -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
crates/collab/src/rpc.rs
+8 -65
View File
@@ -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<proto::LeaveRoom>,
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<ChannelId>,
) -> 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();
crates/collab/tests/integration/channel_tests.rs
-70
View File
@@ -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,
crates/collab_ui/src/call_stats_modal.rs
+6 -18
View File
@@ -144,24 +144,12 @@ impl Render for CallStatsModal {
fn render(&mut self, _window: &mut Window, cx: &mut Context<Self>) -> 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);
crates/livekit_client/src/test.rs
-40
View File
@@ -56,14 +56,6 @@ pub struct TestServer {
pub api_key: String,
pub secret_key: String,
rooms: Mutex<HashMap<String, TestServerRoom>>,
revoked_identities: Mutex<HashSet<String>>,
/// Log of `(room, identity)` pairs passed to `remove_participant`, for
/// tests that need to assert whether a participant was forcibly removed.
removed_participants: Mutex<Vec<(String, String)>>,
/// 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)
crates/proto/proto/call.proto
+5 -1
View File
@@ -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 {