livekit_api: Fix LiveKit token revocation timestamps (#60157)

LiveKit Cloud rejects room-join tokens as revoked when their `nbf`
predates a participant revocation. Zed generated those LiveKit JWTs with
`nbf: 0`, so a fresh participant token minted after stale connection
cleanup could still appear older than the cleanup and leave a user
joined at the collab layer without audio or screen sharing. This sets
`nbf` to the issuance time for room-join tokens while leaving admin/API
tokens unchanged, and adds regression coverage at the token, mock
LiveKit, and channel rejoin layers.

Closes FR-83

Release Notes:

- Fixed calls getting stuck without audio or screen sharing after
restarting Zed and rejoining a channel.

Author: rtfeldman

crates/collab/tests/integration/channel_tests.rs

@@ -589,6 +589,95 @@ async fn test_channel_room(
     );
 }
 
+#[gpui::test]
+async fn test_rejoining_channel_after_stale_connection_cleanup_connects_livekit(
+    executor: BackgroundExecutor,
+    cx_a: &mut TestAppContext,
+    cx_a2: &mut TestAppContext,
+    cx_b: &mut TestAppContext,
+) {
+    let mut server = TestServer::start(executor.clone()).await;
+    let client_a = server.create_client(cx_a, "user_a").await;
+    let client_b = server.create_client(cx_b, "user_b").await;
+
+    let channel_id = server
+        .make_channel("zed", None, (&client_a, cx_a), &mut [(&client_b, cx_b)])
+        .await;
+
+    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();
+
+    let active_call_b = cx_b.read(ActiveCall::global);
+    active_call_b
+        .update(cx_b, |active_call, cx| {
+            active_call.join_channel(channel_id, cx)
+        })
+        .await
+        .unwrap();
+
+    executor.run_until_parked();
+
+    let old_room_a =
+        cx_a.read(|cx| active_call_a.read_with(cx, |call, _| call.room().unwrap().clone()));
+    cx_a.read(|cx| old_room_a.read_with(cx, |room, cx| assert!(room.is_connected(cx))));
+
+    server.disconnect_client(client_a.peer_id().unwrap());
+    executor.run_until_parked();
+    server.advance_livekit_timestamp();
+
+    let client_a2 = server.create_client(cx_a2, "user_a").await;
+    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();
+
+    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))));
+    assert_eq!(
+        room_participants(&room_a2, cx_a2),
+        RoomParticipants {
+            remote: vec!["user_b".to_string()],
+            pending: vec![]
+        }
+    );
+
+    let room_b =
+        cx_b.read(|cx| active_call_b.read_with(cx, |call, _| call.room().unwrap().clone()));
+    cx_b.read(|cx| room_b.read_with(cx, |room, cx| assert!(room.is_connected(cx))));
+    assert_eq!(
+        room_participants(&room_b, cx_b),
+        RoomParticipants {
+            remote: vec!["user_a".to_string()],
+            pending: vec![]
+        }
+    );
+
+    cx_a2.read(|cx| {
+        client_a2.channel_store().read_with(cx, |channels, _| {
+            let mut participant_ids = channels
+                .channel_participants(channel_id)
+                .iter()
+                .map(|participant| participant.legacy_id)
+                .collect::<Vec<_>>();
+            participant_ids.sort_unstable();
+            let mut expected_ids = vec![client_a2.user_id().unwrap(), client_b.user_id().unwrap()];
+            expected_ids.sort_unstable();
+            assert_eq!(participant_ids, expected_ids);
+        })
+    });
+}
+
 #[gpui::test]
 async fn test_channel_jumping(executor: BackgroundExecutor, cx_a: &mut TestAppContext) {
     let mut server = TestServer::start(executor.clone()).await;

crates/collab/tests/integration/test_server.rs

@@ -48,14 +48,15 @@ use std::{
 use util::path;
 use workspace::{MultiWorkspace, Workspace, WorkspaceStore};
 
-use livekit_client::test::TestServer as LivekitTestServer;
+use livekit_client::test::{ManualUnixTimestampSource, TestServer as LivekitTestServer};
 
 use crate::db_tests::TestDb;
 
 pub struct TestServer {
     pub app_state: Arc<AppState>,
     pub test_livekit_server: Arc<LivekitTestServer>,
     pub test_db: TestDb,
+    livekit_timestamp_source: Arc<ManualUnixTimestampSource>,
     server: Arc<Server>,
     next_github_user_id: i32,
     connection_killers: Arc<Mutex<HashMap<PeerId, Arc<AtomicBool>>>>,
@@ -96,11 +97,13 @@ impl TestServer {
             TestDb::sqlite(deterministic.clone())
         };
         let livekit_server_id = NEXT_LIVEKIT_SERVER_ID.fetch_add(1, SeqCst);
-        let livekit_server = LivekitTestServer::create(
+        let livekit_timestamp_source = Arc::new(ManualUnixTimestampSource::new(1_234_567));
+        let livekit_server = LivekitTestServer::create_with_timestamp_source(
             format!("http://livekit.{}.test", livekit_server_id),
             format!("devkey-{}", livekit_server_id),
             format!("secret-{}", livekit_server_id),
             deterministic.clone(),
+            livekit_timestamp_source.clone(),
         )
         .unwrap();
         let executor = Executor::Deterministic(deterministic.clone());
@@ -121,10 +124,15 @@ impl TestServer {
             forbid_connections: Default::default(),
             next_github_user_id: 0,
             test_db,
+            livekit_timestamp_source,
             test_livekit_server: livekit_server,
         }
     }
 
+    pub fn advance_livekit_timestamp(&self) {
+        self.livekit_timestamp_source.advance();
+    }
+
     pub async fn start2(
         cx_a: &mut TestAppContext,
         cx_b: &mut TestAppContext,

crates/livekit_api/Cargo.toml

@@ -13,6 +13,9 @@ workspace = true
 path = "src/livekit_api.rs"
 doctest = false
 
+[features]
+test-support = []
+
 [dependencies]
 anyhow.workspace = true
 async-trait.workspace = true

crates/livekit_api/src/livekit_api.rs

@@ -31,10 +31,25 @@ pub struct LiveKitClient {
     url: Arc<str>,
     key: Arc<str>,
     secret: Arc<str>,
+    timestamp_source: Arc<dyn token::UnixTimestampSource>,
 }
 
 impl LiveKitClient {
-    pub fn new(mut url: String, key: String, secret: String) -> Self {
+    pub fn new(url: String, key: String, secret: String) -> Self {
+        Self::new_with_timestamp_source(
+            url,
+            key,
+            secret,
+            Arc::new(token::SystemUnixTimestampSource),
+        )
+    }
+
+    pub(crate) fn new_with_timestamp_source(
+        mut url: String,
+        key: String,
+        secret: String,
+        timestamp_source: Arc<dyn token::UnixTimestampSource>,
+    ) -> Self {
         if url.ends_with('/') {
             url.pop();
         }
@@ -47,6 +62,7 @@ impl LiveKitClient {
             url: url.into(),
             key: key.into(),
             secret: secret.into(),
+            timestamp_source,
         }
     }
 
@@ -61,7 +77,13 @@ impl LiveKitClient {
         Res: Default + Message,
     {
         let client = self.http.clone();
-        let token = token::create(&self.key, &self.secret, None, grant);
+        let token = token::create_with_timestamp_source(
+            &self.key,
+            &self.secret,
+            None,
+            grant,
+            self.timestamp_source.as_ref(),
+        );
         let url = format!("{}/{}", self.url, path);
         log::info!("Request {}: {:?}", url, body);
         async move {
@@ -163,20 +185,22 @@ impl Client for LiveKitClient {
     }
 
     fn room_token(&self, room: &str, identity: &str) -> Result<String> {
-        token::create(
+        token::create_with_timestamp_source(
             &self.key,
             &self.secret,
             Some(identity),
             token::VideoGrant::to_join(room),
+            self.timestamp_source.as_ref(),
         )
     }
 
     fn guest_token(&self, room: &str, identity: &str) -> Result<String> {
-        token::create(
+        token::create_with_timestamp_source(
             &self.key,
             &self.secret,
             Some(identity),
             token::VideoGrant::for_guest(room),
+            self.timestamp_source.as_ref(),
         )
     }
 }