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.

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,7 +48,7 @@ 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;
 
@@ -56,6 +56,7 @@ 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(),
         )
     }
 }

crates/livekit_api/src/token.rs

@@ -1,14 +1,25 @@
-use anyhow::Result;
+use anyhow::{Context as _, Result};
 use jsonwebtoken::{DecodingKey, EncodingKey, Header, Validation};
 use serde::{Deserialize, Serialize};
 use std::{
     borrow::Cow,
-    ops::Add,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
 const DEFAULT_TTL: Duration = Duration::from_secs(6 * 60 * 60); // 6 hours
 
+pub trait UnixTimestampSource: Send + Sync {
+    fn unix_timestamp(&self) -> Result<u64>;
+}
+
+pub struct SystemUnixTimestampSource;
+
+impl UnixTimestampSource for SystemUnixTimestampSource {
+    fn unix_timestamp(&self) -> Result<u64> {
+        Ok(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs())
+    }
+}
+
 #[derive(Default, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ClaimGrants<'a> {
@@ -73,22 +84,55 @@ pub fn create(
     identity: Option<&str>,
     video_grant: VideoGrant,
 ) -> Result<String> {
-    if video_grant.room_join.is_some() && identity.is_none() {
+    create_with_timestamp_source(
+        api_key,
+        secret_key,
+        identity,
+        video_grant,
+        &SystemUnixTimestampSource,
+    )
+}
+
+pub fn create_with_timestamp_source(
+    api_key: &str,
+    secret_key: &str,
+    identity: Option<&str>,
+    video_grant: VideoGrant,
+    timestamp_source: &dyn UnixTimestampSource,
+) -> Result<String> {
+    let issued_at = timestamp_source.unix_timestamp()?;
+    create_with_issued_at(api_key, secret_key, identity, video_grant, issued_at)
+}
+
+fn create_with_issued_at(
+    api_key: &str,
+    secret_key: &str,
+    identity: Option<&str>,
+    video_grant: VideoGrant,
+    issued_at: u64,
+) -> Result<String> {
+    let room_join = video_grant.room_join.unwrap_or(false);
+    if room_join && identity.is_none() {
         anyhow::bail!("identity is required for room_join grant, but it is none");
     }
 
-    let now = SystemTime::now();
+    let expires_at = issued_at
+        .checked_add(DEFAULT_TTL.as_secs())
+        .context("token expiration overflow")?;
+    let not_before = if room_join {
+        // LiveKit Cloud applies participant revocations by comparing the
+        // revocation timestamp to room-join token `nbf`.
+        issued_at
+    } else {
+        0
+    };
 
     let claims = ClaimGrants {
         iss: Cow::Borrowed(api_key),
         sub: identity.map(Cow::Borrowed),
-        iat: now.duration_since(UNIX_EPOCH).unwrap().as_secs(),
-        exp: now
-            .add(DEFAULT_TTL)
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_secs(),
-        nbf: 0,
+        iat: issued_at,
+        exp: expires_at,
+        nbf: not_before,
         jwtid: identity.map(Cow::Borrowed),
         video: video_grant,
     };
@@ -108,3 +152,127 @@ pub fn validate<'a>(token: &'a str, secret_key: &str) -> Result<ClaimGrants<'a>>
 
     Ok(token.claims)
 }
+
+#[cfg(any(test, feature = "test-support"))]
+pub fn validate_with_timestamp_source<'a>(
+    token: &'a str,
+    secret_key: &str,
+    timestamp_source: &dyn UnixTimestampSource,
+) -> Result<ClaimGrants<'a>> {
+    let mut validation = Validation::default();
+    validation.validate_exp = false;
+    validation.validate_nbf = false;
+    let token: jsonwebtoken::TokenData<ClaimGrants<'_>> = jsonwebtoken::decode(
+        token,
+        &DecodingKey::from_secret(secret_key.as_ref()),
+        &validation,
+    )?;
+    let claims = token.claims;
+    let timestamp = timestamp_source.unix_timestamp()?;
+
+    anyhow::ensure!(claims.nbf <= timestamp, "token is not yet valid");
+    anyhow::ensure!(claims.exp > timestamp, "token has expired");
+
+    Ok(claims)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{Client as _, LiveKitClient};
+    use std::sync::Arc;
+
+    const ISSUED_AT: u64 = 1_234_567;
+
+    struct FixedUnixTimestampSource(u64);
+
+    impl UnixTimestampSource for FixedUnixTimestampSource {
+        fn unix_timestamp(&self) -> Result<u64> {
+            Ok(self.0)
+        }
+    }
+
+    #[test]
+    fn token_not_before_matches_issue_time() -> Result<()> {
+        let token = create_with_timestamp_source(
+            "api-key",
+            "secret-key",
+            Some("participant"),
+            VideoGrant::to_join("room"),
+            &FixedUnixTimestampSource(ISSUED_AT),
+        )?;
+
+        assert_claims_timestamp(&token, ISSUED_AT, ISSUED_AT)?;
+        Ok(())
+    }
+
+    #[test]
+    fn room_token_not_before_matches_issue_time() -> Result<()> {
+        let client = LiveKitClient::new_with_timestamp_source(
+            "http://livekit.test".into(),
+            "api-key".into(),
+            "secret-key".into(),
+            Arc::new(FixedUnixTimestampSource(ISSUED_AT)),
+        );
+        let token = client.room_token("room", "participant")?;
+
+        let claims = assert_claims_timestamp(&token, ISSUED_AT, ISSUED_AT)?;
+        assert_eq!(claims.video.room_join, Some(true));
+        assert_eq!(claims.video.can_publish, Some(true));
+        assert_eq!(claims.video.can_subscribe, Some(true));
+
+        Ok(())
+    }
+
+    #[test]
+    fn guest_token_not_before_matches_issue_time() -> Result<()> {
+        let client = LiveKitClient::new_with_timestamp_source(
+            "http://livekit.test".into(),
+            "api-key".into(),
+            "secret-key".into(),
+            Arc::new(FixedUnixTimestampSource(ISSUED_AT)),
+        );
+        let token = client.guest_token("room", "participant")?;
+
+        let claims = assert_claims_timestamp(&token, ISSUED_AT, ISSUED_AT)?;
+        assert_eq!(claims.video.room_join, Some(true));
+        assert_eq!(claims.video.can_publish, Some(false));
+        assert_eq!(claims.video.can_subscribe, Some(true));
+
+        Ok(())
+    }
+
+    #[test]
+    fn admin_token_not_before_remains_unset() -> Result<()> {
+        let token = create_with_timestamp_source(
+            "api-key",
+            "secret-key",
+            None,
+            VideoGrant::to_admin("room"),
+            &FixedUnixTimestampSource(ISSUED_AT),
+        )?;
+
+        let claims = assert_claims_timestamp(&token, ISSUED_AT, 0)?;
+        assert_eq!(claims.video.room_admin, Some(true));
+
+        Ok(())
+    }
+
+    fn assert_claims_timestamp(
+        token: &str,
+        issued_at: u64,
+        expected_not_before: u64,
+    ) -> Result<ClaimGrants<'_>> {
+        let claims = validate_with_timestamp_source(
+            token,
+            "secret-key",
+            &FixedUnixTimestampSource(issued_at),
+        )?;
+
+        assert_eq!(claims.iat, issued_at);
+        assert_eq!(claims.nbf, expected_not_before);
+        assert_eq!(claims.exp, issued_at + DEFAULT_TTL.as_secs());
+
+        Ok(claims)
+    }
+}

crates/livekit_client/Cargo.toml

@@ -17,7 +17,7 @@ doctest = false
 name = "test_app"
 
 [features]
-test-support = ["collections/test-support", "gpui/test-support"]
+test-support = ["collections/test-support", "gpui/test-support", "livekit_api/test-support"]
 
 [dependencies]
 anyhow.workspace = true
@@ -65,6 +65,7 @@ objc.workspace = true
 collections = { workspace = true, features = ["test-support"] }
 gpui = { workspace = true, features = ["test-support"] }
 gpui_platform.workspace = true
+livekit_api = { workspace = true, features = ["test-support"] }
 simplelog.workspace = true
 
 [build-dependencies]

crates/livekit_client/src/test.rs

@@ -57,6 +57,25 @@ pub struct TestServer {
     pub secret_key: String,
     rooms: Mutex<HashMap<String, TestServerRoom>>,
     executor: BackgroundExecutor,
+    timestamp_source: Arc<dyn token::UnixTimestampSource>,
+}
+
+pub struct ManualUnixTimestampSource(AtomicU64);
+
+impl ManualUnixTimestampSource {
+    pub fn new(timestamp: u64) -> Self {
+        Self(AtomicU64::new(timestamp))
+    }
+
+    pub fn advance(&self) {
+        self.0.fetch_add(1, SeqCst);
+    }
+}
+
+impl token::UnixTimestampSource for ManualUnixTimestampSource {
+    fn unix_timestamp(&self) -> Result<u64> {
+        Ok(self.0.load(SeqCst))
+    }
 }
 
 impl TestServer {
@@ -65,6 +84,22 @@ impl TestServer {
         api_key: String,
         secret_key: String,
         executor: BackgroundExecutor,
+    ) -> Result<Arc<TestServer>> {
+        Self::create_with_timestamp_source(
+            url,
+            api_key,
+            secret_key,
+            executor,
+            Arc::new(token::SystemUnixTimestampSource),
+        )
+    }
+
+    pub fn create_with_timestamp_source(
+        url: String,
+        api_key: String,
+        secret_key: String,
+        executor: BackgroundExecutor,
+        timestamp_source: Arc<dyn token::UnixTimestampSource>,
     ) -> Result<Arc<TestServer>> {
         let mut servers = SERVERS.lock();
         if let BTreeEntry::Vacant(e) = servers.entry(url.clone()) {
@@ -74,6 +109,7 @@ impl TestServer {
                 secret_key,
                 rooms: Default::default(),
                 executor,
+                timestamp_source,
             });
             e.insert(server.clone());
             Ok(server)
@@ -104,6 +140,20 @@ impl TestServer {
         }
     }
 
+    #[cfg(any(test, feature = "test-support"))]
+    fn validate_token<'a>(&self, token: &'a str) -> Result<token::ClaimGrants<'a>> {
+        token::validate_with_timestamp_source(
+            token,
+            &self.secret_key,
+            self.timestamp_source.as_ref(),
+        )
+    }
+
+    #[cfg(not(any(test, feature = "test-support")))]
+    fn validate_token<'a>(&self, token: &'a str) -> Result<token::ClaimGrants<'a>> {
+        token::validate(token, &self.secret_key)
+    }
+
     pub async fn create_room(&self, room: String) -> Result<()> {
         self.simulate_random_delay().await;
 
@@ -129,11 +179,19 @@ impl TestServer {
     async fn join_room(&self, token: String, client_room: Room) -> Result<ParticipantIdentity> {
         self.simulate_random_delay().await;
 
-        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();
+        let claims = self.validate_token(&token)?;
+        let identity = ParticipantIdentity(
+            claims
+                .sub
+                .context("missing participant identity")?
+                .to_string(),
+        );
+        let room_name = claims.video.room.context("missing room name")?.to_string();
         let mut server_rooms = self.rooms.lock();
-        let room = (*server_rooms).entry(room_name.to_string()).or_default();
+        let room = (*server_rooms).entry(room_name.clone()).or_default();
+        if let Some(revoked_before) = room.token_revocations.get(&identity) {
+            anyhow::ensure!(claims.nbf >= *revoked_before, "invalid token: revoked");
+        }
 
         if let Entry::Vacant(e) = room.client_rooms.entry(identity.clone()) {
             for server_track in &room.video_tracks {
@@ -192,7 +250,7 @@ impl TestServer {
     async fn leave_room(&self, token: String) -> Result<()> {
         self.simulate_random_delay().await;
 
-        let claims = livekit_api::token::validate(&token, &self.secret_key)?;
+        let claims = self.validate_token(&token)?;
         let identity = ParticipantIdentity(claims.sub.unwrap().to_string());
         let room_name = claims.video.room.unwrap();
         let mut server_rooms = self.rooms.lock();
@@ -209,7 +267,7 @@ impl TestServer {
         &self,
         token: String,
     ) -> Result<HashMap<ParticipantIdentity, RemoteParticipant>> {
-        let claims = livekit_api::token::validate(&token, &self.secret_key)?;
+        let claims = self.validate_token(&token)?;
         let local_identity = ParticipantIdentity(claims.sub.unwrap().to_string());
         let room_name = claims.video.room.unwrap().to_string();
 
@@ -244,14 +302,25 @@ impl TestServer {
         identity: ParticipantIdentity,
     ) -> Result<()> {
         self.simulate_random_delay().await;
+        let revoked_before = self.timestamp_source.unix_timestamp()?;
 
         let mut server_rooms = self.rooms.lock();
         let room = server_rooms
             .get_mut(&room_name)
             .with_context(|| format!("room {room_name} does not exist"))?;
-        room.client_rooms
+        let removed_room = room
+            .client_rooms
             .remove(&identity)
             .with_context(|| format!("participant {identity:?} did not join room {room_name:?}"))?;
+        room.token_revocations.insert(identity, revoked_before);
+        let mut removed_room = removed_room.0.lock();
+        removed_room.connection_state = ConnectionState::Disconnected;
+        removed_room
+            .updates_tx
+            .blocking_send(RoomEvent::Disconnected {
+                reason: "PARTICIPANT_REMOVED",
+            })
+            .ok();
         Ok(())
     }
 
@@ -262,13 +331,18 @@ impl TestServer {
         permission: proto::ParticipantPermission,
     ) -> Result<()> {
         self.simulate_random_delay().await;
+        let revoked_before = self.timestamp_source.unix_timestamp()?;
 
         let mut server_rooms = self.rooms.lock();
         let room = server_rooms
             .get_mut(&room_name)
             .with_context(|| format!("room {room_name} does not exist"))?;
+        let identity = ParticipantIdentity(identity);
         room.participant_permissions
-            .insert(ParticipantIdentity(identity), permission);
+            .insert(identity.clone(), permission);
+        // Permission changes in LiveKit Cloud invalidate existing participant
+        // tokens, so the mock needs to reject tokens minted before the update.
+        room.token_revocations.insert(identity, revoked_before);
         Ok(())
     }
 
@@ -298,7 +372,7 @@ impl TestServer {
     ) -> Result<TrackSid> {
         self.simulate_random_delay().await;
 
-        let claims = livekit_api::token::validate(&token, &self.secret_key)?;
+        let claims = self.validate_token(&token)?;
         let identity = ParticipantIdentity(claims.sub.unwrap().to_string());
         let room_name = claims.video.room.unwrap();
 
@@ -362,7 +436,7 @@ impl TestServer {
     ) -> Result<TrackSid> {
         self.simulate_random_delay().await;
 
-        let claims = livekit_api::token::validate(&token, &self.secret_key)?;
+        let claims = self.validate_token(&token)?;
         let identity = ParticipantIdentity(claims.sub.unwrap().to_string());
         let room_name = claims.video.room.unwrap();
 
@@ -421,7 +495,7 @@ impl TestServer {
     }
 
     pub(crate) async fn unpublish_track(&self, token: String, track_sid: &TrackSid) -> Result<()> {
-        let claims = livekit_api::token::validate(&token, &self.secret_key)?;
+        let claims = self.validate_token(&token)?;
         let identity = ParticipantIdentity(claims.sub.unwrap().to_string());
         let room_name = claims.video.room.unwrap();
 
@@ -503,7 +577,7 @@ impl TestServer {
         track_sid: &TrackSid,
         muted: bool,
     ) -> Result<()> {
-        let claims = livekit_api::token::validate(token, &self.secret_key)?;
+        let claims = self.validate_token(token)?;
         let room_name = claims.video.room.unwrap();
         let identity = ParticipantIdentity(claims.sub.unwrap().to_string());
         let mut server_rooms = self.rooms.lock();
@@ -557,7 +631,7 @@ impl TestServer {
     }
 
     pub(crate) fn is_track_muted(&self, token: &str, track_sid: &TrackSid) -> Option<bool> {
-        let claims = livekit_api::token::validate(token, &self.secret_key).ok()?;
+        let claims = self.validate_token(token).ok()?;
         let room_name = claims.video.room.unwrap();
 
         let mut server_rooms = self.rooms.lock();
@@ -572,7 +646,7 @@ impl TestServer {
     }
 
     pub(crate) fn video_tracks(&self, token: String) -> Result<Vec<RemoteVideoTrack>> {
-        let claims = livekit_api::token::validate(&token, &self.secret_key)?;
+        let claims = self.validate_token(&token)?;
         let room_name = claims.video.room.unwrap();
         let identity = ParticipantIdentity(claims.sub.unwrap().to_string());
 
@@ -595,7 +669,7 @@ impl TestServer {
     }
 
     pub(crate) fn audio_tracks(&self, token: String) -> Result<Vec<RemoteAudioTrack>> {
-        let claims = livekit_api::token::validate(&token, &self.secret_key)?;
+        let claims = self.validate_token(&token)?;
         let room_name = claims.video.room.unwrap();
         let identity = ParticipantIdentity(claims.sub.unwrap().to_string());
 
@@ -629,6 +703,7 @@ struct TestServerRoom {
     video_tracks: Vec<Arc<TestServerVideoTrack>>,
     audio_tracks: Vec<Arc<TestServerAudioTrack>>,
     participant_permissions: HashMap<ParticipantIdentity, proto::ParticipantPermission>,
+    token_revocations: HashMap<ParticipantIdentity, u64>,
 }
 
 #[derive(Debug)]
@@ -690,21 +765,23 @@ impl livekit_api::Client for TestApiClient {
 
     fn room_token(&self, room: &str, identity: &str) -> Result<String> {
         let server = TestServer::get(&self.url)?;
-        token::create(
+        token::create_with_timestamp_source(
             &server.api_key,
             &server.secret_key,
             Some(identity),
             token::VideoGrant::to_join(room),
+            server.timestamp_source.as_ref(),
         )
     }
 
     fn guest_token(&self, room: &str, identity: &str) -> Result<String> {
         let server = TestServer::get(&self.url)?;
-        token::create(
+        token::create_with_timestamp_source(
             &server.api_key,
             &server.secret_key,
             Some(identity),
             token::VideoGrant::for_guest(room),
+            server.timestamp_source.as_ref(),
         )
     }
 }
@@ -853,3 +930,168 @@ impl WeakRoom {
         self.0.upgrade().map(Room)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use gpui::TestAppContext;
+    use livekit_api::Client as _;
+    use std::{ops::Deref, sync::atomic::AtomicUsize};
+
+    struct TestServerGuard {
+        server: Arc<TestServer>,
+        timestamp_source: Arc<ManualUnixTimestampSource>,
+    }
+
+    impl TestServerGuard {
+        fn advance_timestamp(&self) {
+            self.timestamp_source.advance();
+        }
+    }
+
+    impl Deref for TestServerGuard {
+        type Target = TestServer;
+
+        fn deref(&self) -> &Self::Target {
+            self.server.as_ref()
+        }
+    }
+
+    impl Drop for TestServerGuard {
+        fn drop(&mut self) {
+            self.server.teardown().ok();
+        }
+    }
+
+    fn create_test_server(name: &str, executor: BackgroundExecutor) -> TestServerGuard {
+        static NEXT_SERVER_ID: AtomicUsize = AtomicUsize::new(0);
+        let server_id = NEXT_SERVER_ID.fetch_add(1, SeqCst);
+        let timestamp_source = Arc::new(ManualUnixTimestampSource::new(1_234_567));
+        let server = TestServer::create_with_timestamp_source(
+            format!("http://livekit-{name}-{server_id}.test"),
+            format!("api-key-{server_id}"),
+            format!("secret-key-{server_id}"),
+            executor,
+            timestamp_source.clone(),
+        )
+        .expect("create LiveKit test server");
+        TestServerGuard {
+            server,
+            timestamp_source,
+        }
+    }
+
+    async fn assert_token_was_revoked(server: &TestServer, token: String, cx: &mut TestAppContext) {
+        match Room::connect(server.url.clone(), token, &mut cx.to_async()).await {
+            Ok(_) => panic!("revoked token unexpectedly connected"),
+            Err(error) => {
+                let error = format!("{error:#}");
+                assert!(
+                    error.contains("invalid token: revoked"),
+                    "expected revoked token error, got {error}"
+                );
+            }
+        }
+    }
+
+    #[gpui::test]
+    async fn token_created_after_participant_removal_can_join(
+        executor: BackgroundExecutor,
+        cx: &mut TestAppContext,
+    ) {
+        let server = create_test_server("room-token", executor);
+        server
+            .create_room("room".into())
+            .await
+            .expect("create LiveKit test room");
+        let api_client = server.create_api_client();
+
+        let initial_token = api_client
+            .room_token("room", "participant")
+            .expect("create initial room token");
+        let (initial_room, _) = Room::connect(
+            server.url.clone(),
+            initial_token.clone(),
+            &mut cx.to_async(),
+        )
+        .await
+        .expect("connect with initial room token");
+
+        server.advance_timestamp();
+        api_client
+            .remove_participant("room".into(), "participant".into())
+            .await
+            .expect("remove participant");
+
+        assert_eq!(
+            initial_room.connection_state(),
+            ConnectionState::Disconnected
+        );
+        assert_token_was_revoked(&server, initial_token, cx).await;
+
+        let fresh_token = api_client
+            .room_token("room", "participant")
+            .expect("create fresh room token");
+        let (fresh_room, _) = Room::connect(server.url.clone(), fresh_token, &mut cx.to_async())
+            .await
+            .expect("connect with fresh room token");
+
+        assert_eq!(fresh_room.connection_state(), ConnectionState::Connected);
+    }
+
+    #[gpui::test]
+    async fn guest_token_created_after_permission_update_can_join(
+        executor: BackgroundExecutor,
+        cx: &mut TestAppContext,
+    ) {
+        let server = create_test_server("guest-token", executor);
+        server
+            .create_room("room".into())
+            .await
+            .expect("create LiveKit test room");
+        let api_client = server.create_api_client();
+
+        let initial_token = api_client
+            .guest_token("room", "participant")
+            .expect("create initial guest token");
+        let (initial_room, _) = Room::connect(
+            server.url.clone(),
+            initial_token.clone(),
+            &mut cx.to_async(),
+        )
+        .await
+        .expect("connect with initial guest token");
+
+        server.advance_timestamp();
+        api_client
+            .update_participant(
+                "room".into(),
+                "participant".into(),
+                proto::ParticipantPermission {
+                    can_subscribe: true,
+                    can_publish: true,
+                    can_publish_data: true,
+                    hidden: false,
+                    recorder: false,
+                },
+            )
+            .await
+            .expect("update participant permissions");
+        assert_token_was_revoked(&server, initial_token, cx).await;
+
+        server.disconnect_client("participant".into()).await;
+        assert_eq!(
+            initial_room.connection_state(),
+            ConnectionState::Disconnected
+        );
+
+        let fresh_token = api_client
+            .guest_token("room", "participant")
+            .expect("create fresh guest token");
+        let (fresh_room, _) = Room::connect(server.url.clone(), fresh_token, &mut cx.to_async())
+            .await
+            .expect("connect with fresh guest token");
+
+        assert_eq!(fresh_room.connection_state(), ConnectionState::Connected);
+    }
+}