Skip to content

Commit 0e4d10b

Browse files
committed
Share the RoomService transport and add DeleteRoom
Peer cleanup has to call LiveKit without an agent config. The existing RoomService client only read credentials from AgentConfig, so the web server could not delete a room. The transport now takes a URL, key and secret. A 2xx that is not JSON is not a deleted room, because a reverse proxy can answer 200 with HTML while the room is still there.
1 parent 6b6cdfe commit 0e4d10b

6 files changed

Lines changed: 345 additions & 77 deletions

File tree

‎src/livekit.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@ use crate::token::{LivekitTokenInput, livekit_token};
9595
mod board;
9696
mod media;
9797
mod report;
98+
mod room_service;
9899
mod rooms;
99100
mod session;
100101
mod timing;
@@ -119,6 +120,11 @@ use rooms::{evict_duplicate_agent, isolate_local_agent};
119120
// credentials a Setup form was given, and it names it through this module.
120121
pub(crate) use rooms::validate_livekit_credentials;
121122

123+
// Room cleanup runs on the web server, not in the agent. The allow goes when
124+
// the peer sweeper calls it.
125+
#[allow(dead_code, unused_imports)]
126+
pub(crate) use room_service::delete_room;
127+
122128
use media::*;
123129
use turn::*;
124130

‎src/livekit/room_service.rs‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
//! LiveKit's RoomService over Twirp, for any caller holding a project's URL,
2+
//! API key and API secret.
3+
//!
4+
//! Split out of `rooms.rs`, where the transport read its credentials from
5+
//! `AgentConfig`. The agent's duplicate handling still goes through it, but so
6+
//! does room cleanup on the web server, which never loads an agent config and
7+
//! has only the provider's credentials to hand.
8+
9+
use crate::config::livekit_http_base;
10+
use crate::token::{livekit_room_admin_token, livekit_room_create_token};
11+
12+
/// The Twirp POST every RoomService caller makes. Only the token and the base
13+
/// differ, and both are strings; the body comes back with the status because a
14+
/// refusal explains itself there and not in the code.
15+
pub(super) async fn post_room_service(
16+
base: &str,
17+
token: &str,
18+
method: &str,
19+
body: &serde_json::Value,
20+
) -> Result<(reqwest::StatusCode, String), Box<dyn std::error::Error + Send + Sync>> {
21+
let response = crate::http_client()
22+
.post(format!("{base}/twirp/livekit.RoomService/{method}"))
23+
.bearer_auth(token)
24+
.json(body)
25+
.send()
26+
.await?;
27+
let status = response.status();
28+
Ok((status, response.text().await?))
29+
}
30+
31+
/// A room-scoped call under a `roomAdmin` token, answered with LiveKit's parsed
32+
/// reply.
33+
pub(super) async fn livekit_room_service_request(
34+
url: &str,
35+
api_key: &str,
36+
api_secret: &str,
37+
room_name: &str,
38+
method: &str,
39+
body: serde_json::Value,
40+
now_seconds: u64,
41+
) -> Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
42+
let token = livekit_room_admin_token(api_key, api_secret, room_name, now_seconds)?;
43+
let (status, text) = post_room_service(&livekit_http_base(url), &token, method, &body).await?;
44+
if !status.is_success() {
45+
return Err(format!("LiveKit RoomService {method} failed: {status} {text}").into());
46+
}
47+
parse_room_service_response(method, &text)
48+
}
49+
50+
pub(super) fn parse_room_service_response(
51+
method: &str,
52+
text: &str,
53+
) -> Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
54+
serde_json::from_str(text).map_err(|error| {
55+
format!("LiveKit RoomService {method} returned invalid JSON: {error}").into()
56+
})
57+
}
58+
59+
/// Deletes a room, which disconnects everyone still in it.
60+
///
61+
/// A room that is already gone counts as deleted: cleanup runs after the fact,
62+
/// and LiveKit drops an empty room on its own, so losing that race is the
63+
/// ordinary case. A 2xx body still has to parse as JSON, because a reverse
64+
/// proxy can answer 200 with an HTML page while the room is still there. Any
65+
/// other refusal is an error carrying the method, the status and LiveKit's
66+
/// body, the same shape as every other RoomService failure.
67+
pub(crate) async fn delete_room(
68+
url: &str,
69+
api_key: &str,
70+
api_secret: &str,
71+
room_name: &str,
72+
now_seconds: u64,
73+
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
74+
let token = livekit_room_create_token(api_key, api_secret, now_seconds)?;
75+
let (status, text) = post_room_service(
76+
&livekit_http_base(url),
77+
&token,
78+
"DeleteRoom",
79+
&serde_json::json!({ "room": room_name }),
80+
)
81+
.await?;
82+
if is_twirp_not_found(status, &text) {
83+
return Ok(());
84+
}
85+
if !status.is_success() {
86+
return Err(format!("LiveKit RoomService DeleteRoom failed: {status} {text}").into());
87+
}
88+
parse_room_service_response("DeleteRoom", &text).map(|_| ())
89+
}
90+
91+
/// Both the status and the Twirp code, because a bare 404 could also be a
92+
/// wrong host or path answering with an HTML page, and that is a failure to
93+
/// report rather than a room or participant to forget.
94+
pub(super) fn is_twirp_not_found(status: reqwest::StatusCode, text: &str) -> bool {
95+
status == reqwest::StatusCode::NOT_FOUND
96+
&& serde_json::from_str::<serde_json::Value>(text)
97+
.is_ok_and(|body| body["code"] == "not_found")
98+
}
99+
100+
#[cfg(test)]
101+
#[path = "../../tests/unit/livekit/room_service.rs"]
102+
mod tests;

‎src/livekit/rooms.rs‎

Lines changed: 32 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,11 @@
1-
//! Room administration over the LiveKit server API.
1+
//! The agent's room administration over the LiveKit server API.
22
//!
3-
//! Named apart from the rest because it is the one region here that talks to
4-
//! LiveKit over HTTP rather than over the room's own event stream. Everything
5-
//! in it exists for one problem: a previous run of this agent can leave a
6-
//! participant behind, and a second agent in a room answers over the top of
7-
//! the first. `run_room` and `open_session` in the parent call two of these and
8-
//! nothing outside this module calls any of them.
3+
//! Named apart from the rest because it talks to LiveKit over HTTP rather than
4+
//! over the room's own event stream; the HTTP itself is `room_service`'s.
5+
//! Everything in it exists for one problem: a previous run of this agent can
6+
//! leave a participant behind, and a second agent in a room answers over the
7+
//! top of the first. `run_room` and `open_session` in the parent call two of
8+
//! these and nothing outside this module calls any of them.
99
//!
1010
//! Split out of `livekit.rs` along the line its module doc already drew.
1111
@@ -15,6 +15,10 @@ use crate::config::{AgentConfig, livekit_http_base};
1515
use crate::token::livekit_room_admin_token;
1616

1717
use super::DUPLICATE_AGENT_ISOLATION_ATTEMPTS;
18+
use super::room_service::{
19+
is_twirp_not_found, livekit_room_service_request, parse_room_service_response,
20+
post_room_service,
21+
};
1822

1923
pub(super) async fn isolate_local_agent(
2024
config: &AgentConfig,
@@ -65,7 +69,9 @@ async fn list_room_participants(
6569
now_seconds: u64,
6670
) -> Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
6771
livekit_room_service_request(
68-
config,
72+
&config.livekit_url,
73+
&config.livekit_api_key,
74+
&config.livekit_api_secret,
6975
room_name,
7076
"ListParticipants",
7177
serde_json::json!({ "room": room_name }),
@@ -89,59 +95,6 @@ async fn remove_room_participant(
8995
identity: &str,
9096
now_seconds: u64,
9197
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
92-
match livekit_room_service_request(
93-
config,
94-
room_name,
95-
"RemoveParticipant",
96-
serde_json::json!({ "room": room_name, "identity": identity }),
97-
now_seconds,
98-
)
99-
.await
100-
{
101-
Ok(_) => Ok(()),
102-
Err(error) if is_participant_gone(&error.to_string()) => {
103-
eprintln!("duplicate LiveKit agent participant already gone identity={identity}");
104-
Ok(())
105-
}
106-
Err(error) => Err(error),
107-
}
108-
}
109-
110-
/// Matches on the wire response rather than on a status code alone, because
111-
/// LiveKit answers Twirp over HTTP and a bare 404 could also mean the route is
112-
/// wrong. All three parts are required: the method, so a 404 from any other
113-
/// RoomService call stays fatal; the status; and `not_found` in the body, which
114-
/// is the participant-specific Twirp code rather than an HTML error page.
115-
fn is_participant_gone(error: &str) -> bool {
116-
error.contains("RemoveParticipant") && error.contains("404") && error.contains("not_found")
117-
}
118-
119-
/// The Twirp POST both RoomService callers make. Only the token and the base
120-
/// differ, and both are strings; the body comes back with the status because a
121-
/// refusal explains itself there and not in the code.
122-
async fn post_room_service(
123-
base: &str,
124-
token: &str,
125-
method: &str,
126-
body: &serde_json::Value,
127-
) -> Result<(reqwest::StatusCode, String), Box<dyn std::error::Error + Send + Sync>> {
128-
let response = crate::http_client()
129-
.post(format!("{base}/twirp/livekit.RoomService/{method}"))
130-
.bearer_auth(token)
131-
.json(body)
132-
.send()
133-
.await?;
134-
let status = response.status();
135-
Ok((status, response.text().await?))
136-
}
137-
138-
async fn livekit_room_service_request(
139-
config: &AgentConfig,
140-
room_name: &str,
141-
method: &str,
142-
body: serde_json::Value,
143-
now_seconds: u64,
144-
) -> Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
14598
let token = livekit_room_admin_token(
14699
&config.livekit_api_key,
147100
&config.livekit_api_secret,
@@ -151,14 +104,28 @@ async fn livekit_room_service_request(
151104
let (status, text) = post_room_service(
152105
&livekit_http_base(&config.livekit_url),
153106
&token,
154-
method,
155-
&body,
107+
"RemoveParticipant",
108+
&serde_json::json!({ "room": room_name, "identity": identity }),
156109
)
157110
.await?;
111+
if is_twirp_not_found(status, &text) {
112+
eprintln!("duplicate LiveKit agent participant already gone identity={identity}");
113+
return Ok(());
114+
}
158115
if !status.is_success() {
159-
return Err(format!("LiveKit RoomService {method} failed: {status} {text}").into());
116+
return Err(
117+
format!("LiveKit RoomService RemoveParticipant failed: {status} {text}").into(),
118+
);
160119
}
161-
parse_room_service_response(method, &text)
120+
parse_room_service_response("RemoveParticipant", &text).map(|_| ())
121+
}
122+
123+
/// The old "already gone" check, kept so the unit tests that pin it still
124+
/// compile unedited. Production removal uses [`is_twirp_not_found`] instead of
125+
/// searching the formatted error string.
126+
#[cfg(test)]
127+
fn is_participant_gone(error: &str) -> bool {
128+
error.contains("RemoveParticipant") && error.contains("404") && error.contains("not_found")
162129
}
163130

164131
/// Proves credentials like `check-gemini` proves a Google key: the server
@@ -199,15 +166,6 @@ pub(crate) async fn validate_livekit_credentials(
199166
))
200167
}
201168

202-
fn parse_room_service_response(
203-
method: &str,
204-
text: &str,
205-
) -> Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>> {
206-
serde_json::from_str(text).map_err(|error| {
207-
format!("LiveKit RoomService {method} returned invalid JSON: {error}").into()
208-
})
209-
}
210-
211169
fn duplicate_agent_identities(
212170
participants: &serde_json::Value,
213171
local_identity: &str,

‎src/token.rs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,32 @@ pub fn livekit_room_list_token(
150150
)
151151
}
152152

153+
/// A credential for deleting a room, which LiveKit gates on `roomCreate`
154+
/// rather than on `roomAdmin`.
155+
///
156+
/// No room, because LiveKit checks `roomCreate` project-wide and ignores any
157+
/// room in the grant: naming one here would read as a scope nothing enforces.
158+
/// Kept apart from [`livekit_room_admin_token`] so the cleanup path cannot
159+
/// remove participants and the agent's path cannot delete rooms.
160+
pub fn livekit_room_create_token(
161+
api_key: &str,
162+
api_secret: &str,
163+
now_seconds: u64,
164+
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
165+
sign_jwt(
166+
api_secret,
167+
&json!({
168+
"iss": api_key,
169+
"sub": "room-cleanup",
170+
"nbf": now_seconds,
171+
"exp": now_seconds + TOKEN_TTL_SECONDS,
172+
"video": {
173+
"roomCreate": true
174+
}
175+
}),
176+
)
177+
}
178+
153179
fn sign_jwt(
154180
api_secret: &str,
155181
claims: &Value,

‎tests/token.rs‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,9 @@
55
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
66
use codetrial::token::{
77
LivekitTokenInput, TOKEN_TTL_SECONDS, WebhookRejection, livekit_egress_token,
8-
livekit_observer_token, livekit_room_admin_token, livekit_room_from_token,
9-
livekit_room_list_token, livekit_token, livekit_token_issuer, sign_hs256, verify_hs256,
10-
verify_livekit_webhook,
8+
livekit_observer_token, livekit_room_admin_token, livekit_room_create_token,
9+
livekit_room_from_token, livekit_room_list_token, livekit_token, livekit_token_issuer,
10+
sign_hs256, verify_hs256, verify_livekit_webhook,
1111
};
1212
use serde_json::{Value, json};
1313
use sha2::{Digest as _, Sha256};
@@ -153,6 +153,17 @@ fn each_token_carries_only_the_grant_its_purpose_needs() {
153153
assert_eq!(list["video"]["roomJoin"], Value::Null);
154154
assert_eq!(list["video"]["roomAdmin"], Value::Null);
155155
assert_eq!(list["video"]["roomRecord"], Value::Null);
156+
157+
// DeleteRoom is gated on `roomCreate`, and room cleanup needs nothing else:
158+
// not the `roomAdmin` that evicts participants, not a seat in the room.
159+
let create = claims_of(&livekit_room_create_token(KEY, SECRET, NOW).unwrap());
160+
assert_eq!(create["iss"], json!(KEY));
161+
assert_eq!(create["video"]["roomCreate"], json!(true));
162+
assert_eq!(create["video"]["room"], Value::Null);
163+
assert_eq!(create["video"]["roomAdmin"], Value::Null);
164+
assert_eq!(create["video"]["roomJoin"], Value::Null);
165+
assert_eq!(create["video"]["roomList"], Value::Null);
166+
assert_eq!(create["video"]["roomRecord"], Value::Null);
156167
}
157168

158169
#[test]
@@ -173,6 +184,7 @@ fn every_token_expires() {
173184
livekit_room_admin_token(KEY, SECRET, "interview-a1b2c3d4", NOW).unwrap(),
174185
livekit_egress_token(KEY, SECRET, "interview-a1b2c3d4", NOW).unwrap(),
175186
livekit_room_list_token(KEY, SECRET, NOW).unwrap(),
187+
livekit_room_create_token(KEY, SECRET, NOW).unwrap(),
176188
];
177189
for token in tokens {
178190
let claims = claims_of(&token);

0 commit comments

Comments
 (0)