From 37ba0e45199b9cac22968992c67f39355e58a508 Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Thu, 15 Oct 2026 03:28:43 +0900 Subject: [PATCH 1/2] refactor(web): route stream key resolution to one loop --- src/servers/web.rs | 40 ++++++++++++++++++++-------------------- src/sfu/room.rs | 20 +++++++++++++------- src/shards/types.rs | 14 +++++++------- src/types.rs | 6 +++--- tests/control_api.rs | 31 +++++++++++++++++++++++++++++++ 5 files changed, 74 insertions(+), 37 deletions(-) diff --git a/src/servers/web.rs b/src/servers/web.rs index fea3e93..d234ee5 100644 --- a/src/servers/web.rs +++ b/src/servers/web.rs @@ -211,26 +211,26 @@ impl FromRequestParts for AuthStreamKey { let hashed_stream_key = hash_string(bearer.token()); - let (tx, mut rx) = tokio::sync::mpsc::channel::>(state.shards.shards.len()); + let (tx, rx) = oneshot::channel::>(); - for sfu_tx in state.shards.senders() { + if let Some(sfu_tx) = state.shards.sender_by_key(bearer.token()) { let msg = SfuMessage::ResolveStreamKey { hashed_stream_key, - reply: tx.clone(), + reply: tx, }; send_to_sfu(sfu_tx, msg).map_err(IntoResponse::into_response)?; + } else { + return Err(unauthorized()); } - drop(tx); - - while let Some(reply) = rx.recv().await { - if let Some(room_id) = reply { - return Ok(AuthStreamKey(room_id)); - } - } - - Err(unauthorized()) + rx.await + .map_err(|e| { + error!("send to sfu loop failed: {}", e); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + })? + .map(AuthStreamKey) + .ok_or_else(unauthorized) } } @@ -245,7 +245,7 @@ async fn create_room(State(state): State) -> Result>(); let room_id = RoomId(room_id); - let Some(sfu_tx) = state.shards.sender_for(&room_id) else { + let Some(sfu_tx) = state.shards.sender_by_key(&room_id.0) else { return Err(StatusCode::NOT_FOUND); }; @@ -323,7 +323,7 @@ async fn update_room( let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); - let Some(sfu_tx) = state.shards.sender_for(&room_id) else { + let Some(sfu_tx) = state.shards.sender_by_key(&room_id.0) else { return Err(StatusCode::NOT_FOUND); }; @@ -350,7 +350,7 @@ async fn delete_room( let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); - let Some(sfu_tx) = state.shards.sender_for(&room_id) else { + let Some(sfu_tx) = state.shards.sender_by_key(&room_id.0) else { return Err(StatusCode::NOT_FOUND); }; @@ -378,7 +378,7 @@ async fn reissue_stream_key( let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); - let Some(sfu_tx) = state.shards.sender_for(&room_id) else { + let Some(sfu_tx) = state.shards.sender_by_key(&room_id.0) else { return Err(StatusCode::NOT_FOUND); }; @@ -423,7 +423,7 @@ async fn sdp_offer( let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); - let Some(sfu_tx) = state.shards.sender_for(&room_id) else { + let Some(sfu_tx) = state.shards.sender_by_key(&room_id.0) else { return Err(StatusCode::NOT_FOUND); }; @@ -461,7 +461,7 @@ async fn whip_sdp_offer( let location = format!("/whip/sessions/{}", session_id); let (tx, rx) = oneshot::channel::>(); - let Some(sfu_tx) = state.shards.sender_for(&room_id) else { + let Some(sfu_tx) = state.shards.sender_by_key(&room_id.0) else { return Err(StatusCode::INTERNAL_SERVER_ERROR); }; @@ -491,7 +491,7 @@ async fn whip_terminate_session( ) -> Result { let (tx, rx) = oneshot::channel::>(); - let Some(sfu_tx) = state.shards.sender_for(&room_id) else { + let Some(sfu_tx) = state.shards.sender_by_key(&room_id.0) else { return Err(StatusCode::INTERNAL_SERVER_ERROR); }; diff --git a/src/sfu/room.rs b/src/sfu/room.rs index 6d8d53e..9b8d9c3 100644 --- a/src/sfu/room.rs +++ b/src/sfu/room.rs @@ -13,7 +13,7 @@ use tokio::sync::oneshot; use tracing::{error, info, warn}; use uuid::Uuid; -use crate::{metrics, types::Viewer}; +use crate::{metrics, shards::types::extract_shard_id, types::Viewer}; use crate::{ shards::types::ShardRouteMsg, types::{ @@ -61,7 +61,11 @@ impl Rooms { } pub fn create(&mut self, room_id: RoomId, reply: oneshot::Sender<(RoomId, StreamKey)>) { - let stream_key = StreamKey::new(); + let Some(shard_id) = extract_shard_id(&room_id.0) else { + return; + }; + + let stream_key = StreamKey::new(shard_id); let hashed_stream_key = stream_key.hashed(); self.insert( @@ -116,13 +120,13 @@ impl Rooms { pub fn resolve_stream_key( &self, hashed_stream_key: [u8; 32], - reply: tokio::sync::mpsc::Sender>, + reply: oneshot::Sender>, ) { let room_id = self.iter().find_map(|(room_id, room)| { bool::from(hashed_stream_key.ct_eq(&room.hashed_stream_key)).then_some(room_id.clone()) }); - reply.try_send(room_id).ok(); + reply.send(room_id).ok(); } pub fn terminate_session( @@ -148,8 +152,10 @@ impl Rooms { room_id: RoomId, reply: oneshot::Sender>, ) { - if let Some(room) = self.get_mut(&room_id) { - let stream_key = StreamKey::new(); + if let Some(room) = self.get_mut(&room_id) + && let Some(shard_id) = extract_shard_id(&room_id.0) + { + let stream_key = StreamKey::new(shard_id); let hashed_stream_key = stream_key.hashed(); room.hashed_stream_key = hashed_stream_key; @@ -355,7 +361,7 @@ mod tests { } fn resolve(rooms: &Rooms, hashed_stream_key: [u8; 32]) -> Option { - let (tx, mut rx) = tokio::sync::mpsc::channel(1); + let (tx, rx) = oneshot::channel(); rooms.resolve_stream_key(hashed_stream_key, tx); rx.blocking_recv().unwrap() } diff --git a/src/shards/types.rs b/src/shards/types.rs index 3abc201..47eb597 100644 --- a/src/shards/types.rs +++ b/src/shards/types.rs @@ -8,11 +8,7 @@ use std::{ use tracing::error; -use crate::{ - config::tuning, - sfu, - types::{RoomId, SfuMessage}, -}; +use crate::{config::tuning, sfu, types::SfuMessage}; pub struct Shard { pub sender: SyncSender, @@ -34,8 +30,8 @@ impl Shards { self.next_shard.fetch_add(1, Ordering::Relaxed) % self.shards.len() } - pub fn sender_for(&self, room_id: &RoomId) -> Option<&SyncSender> { - let shard_id = room_id.0.split("_").nth(1)?.parse::().ok(); + pub fn sender_by_key(&self, key: &str) -> Option<&SyncSender> { + let shard_id = extract_shard_id(key); if let Some(shard_id) = shard_id { self.shards.get(shard_id).map(|s| &s.sender) @@ -76,3 +72,7 @@ impl Shards { shards } } + +pub fn extract_shard_id(key: &str) -> Option { + key.split("_").nth(1)?.parse::().ok() +} diff --git a/src/types.rs b/src/types.rs index 31e7e48..141c457 100644 --- a/src/types.rs +++ b/src/types.rs @@ -237,7 +237,7 @@ pub enum SfuMessage { }, ResolveStreamKey { hashed_stream_key: [u8; 32], - reply: tokio::sync::mpsc::Sender>, + reply: oneshot::Sender>, }, TerminateSession { room_id: RoomId, @@ -354,8 +354,8 @@ impl RoomId { } impl StreamKey { - pub fn new() -> StreamKey { - StreamKey(format!("SK_{}", random_string(STREAM_KEY_LEN))) + pub fn new(shard_id: usize) -> StreamKey { + StreamKey(format!("SK_{}_{}", shard_id, random_string(STREAM_KEY_LEN))) } pub fn hashed(&self) -> [u8; 32] { diff --git a/tests/control_api.rs b/tests/control_api.rs index 0dda49b..49b2c89 100644 --- a/tests/control_api.rs +++ b/tests/control_api.rs @@ -59,6 +59,37 @@ async fn publishing_requires_a_stream_key() { assert_eq!(wrong_key.status(), StatusCode::UNAUTHORIZED, "wrong key"); } +#[tokio::test] +async fn a_wrong_stream_key_is_refused_whatever_loop_it_names() { + let server = Server::default(); + // A key whose loop is never asked must still get an answer, so a hang + // fails the test instead of stalling it. + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .expect("client"); + + let keys = [ + ( + format!("SK_{}_wrong", common::shards()), + "a loop that does not exist", + ), + ("SK_0_wrong".to_string(), "a loop without the key"), + ]; + + for (key, case) in keys { + let response = client + .post(server.sdp_url("/whip")) + .bearer_auth(&key) + .header("Content-Type", "application/sdp") + .body("v=0\r\n") + .send() + .await + .expect("POST /whip"); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "{case}"); + } +} + #[tokio::test] async fn publishing_rejects_a_body_that_is_not_sdp() { let server = Server::default(); From 49e7d0a3b3034ec0a6d4f0fe500d078b1a69eb8b Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Thu, 15 Oct 2026 03:28:43 +0900 Subject: [PATCH 2/2] docs(adr): record that a stream key carries its loop --- docs/architecture/overview.md | 2 +- docs/decisions/0012-reply-over-oneshot.md | 2 +- .../0013-stream-key-carries-its-loop.md | 34 +++++++++++++++++++ 3 files changed, 36 insertions(+), 2 deletions(-) create mode 100644 docs/decisions/0013-stream-key-carries-its-loop.md diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 1ba7ff3..af864d5 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -41,7 +41,7 @@ Media is forwarded per room: a streamer's RTP fans out to every viewer in the sa ## Placement and routing -A new room goes to the next loop in turn. Its id carries the loop number (`RM_2_...`), so every later command reaches the right loop without a table. Resolving a stream key has no room id to go on, so it asks every loop. +A new room goes to the next loop in turn. Its id carries the loop number (`RM_2_...`), so every later command reaches the right loop without a table. A stream key carries the same number (`SK_2_...`), so resolving one asks that loop alone. A datagram carries only its source address, so routing takes two maps: diff --git a/docs/decisions/0012-reply-over-oneshot.md b/docs/decisions/0012-reply-over-oneshot.md index 6993984..6989dd0 100644 --- a/docs/decisions/0012-reply-over-oneshot.md +++ b/docs/decisions/0012-reply-over-oneshot.md @@ -2,7 +2,7 @@ ## Status -Accepted (2026-10-08). Supersedes [ADR 0007](0007-defer-sync-mpsc-recv-blocking.md). +Stream key resolution superseded by [ADR 0013](0013-stream-key-carries-its-loop.md) (2026-10-15). Accepted (2026-10-08). Supersedes [ADR 0007](0007-defer-sync-mpsc-recv-blocking.md). ## Context diff --git a/docs/decisions/0013-stream-key-carries-its-loop.md b/docs/decisions/0013-stream-key-carries-its-loop.md new file mode 100644 index 0000000..0aeda9c --- /dev/null +++ b/docs/decisions/0013-stream-key-carries-its-loop.md @@ -0,0 +1,34 @@ +# 0013. A stream key carries its room's loop + +## Status + +Accepted (2026-10-15). Supersedes the stream key resolution in [ADR 0012](0012-reply-over-oneshot.md). + +## Context + +Since rooms spread across media loops, a room id carries its loop number, so every command about a room goes to one loop. A stream key carried nothing, so resolving one asked every loop. + +It was the only HTTP request that fanned out. Its reply needed an `mpsc` where every other reply uses a `oneshot`, it answered `503` if any loop was full, and one wrong key cost a message on every loop. + +[ADR 0010](0010-stream-key-authentication.md) calls a key "a random string with no structure". Its alternatives show what that ruled out: claims and a signature that every integrator would have to handle. + +## Decision + +- A stream key is `SK__`. The loop is its room's, read from the room id when the key is issued or reissued. +- Resolution goes to that loop alone, with a `oneshot` reply. +- A key without a loop number, or with one past the last loop, gets `401` without a loop being asked. +- The random part, and the hash taken over the whole key, are unchanged. The number routes and grants nothing: a client can write any number, and a wrong one reaches a loop that does not hold the key. + +## Alternatives considered + +- **Keep asking every loop.** One request routed unlike every other, and a wrong key costs every loop a message. +- **The room id inside the key, as a selector.** The loop would find the room directly instead of comparing hashes across its rooms, but a loop holds few rooms, and the key would grow by a whole room id. +- **An index from key hash to loop in the web layer.** Shared state that create, reissue and delete would have to keep in step with the loops. + +## Consequences + +**Gain**: every HTTP request reaches one loop and waits on a `oneshot`; a wrong key costs at most one message; resolution needs no `503` rule of its own. + +**Lose**: a key is tied to where its room lives, as a room id already is. If the loop count changes across a restart once rooms persist, both need the same remapping, and a key is the harder one to replace, since it sits in an encoder's settings. + +**Trigger to revisit**: rooms persisting across restarts, or rooms moving between loops.