refactor: Route stream key resolution to one loop

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:
 

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
 

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_<loop>_<random>`. 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.

src/servers/web.rs

@@ -211,26 +211,26 @@ impl FromRequestParts<SdpState> for AuthStreamKey {
 
         let hashed_stream_key = hash_string(bearer.token());
 
-        let (tx, mut rx) = tokio::sync::mpsc::channel::<Option<RoomId>>(state.shards.shards.len());
+        let (tx, rx) = oneshot::channel::<Option<RoomId>>();
 
-        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<ApiState>) -> Result<impl IntoResponse,
 
     let room_id = RoomId::new(state.shards.next_shard());
 
-    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);
     };
 
@@ -293,7 +293,7 @@ async fn get_room(
     let (tx, rx) = oneshot::channel::<Option<(usize, RoomState)>>();
     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::<Option<()>>();
     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::<Option<()>>();
     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::<Option<StreamKey>>();
     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::<Option<StatusCode>>();
     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::<Option<StatusCode>>();
 
-    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<StatusCode, StatusCode> {
     let (tx, rx) = oneshot::channel::<Option<()>>();
 
-    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);
     };
 

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<Option<RoomId>>,
+        reply: oneshot::Sender<Option<RoomId>>,
     ) {
         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<Option<StreamKey>>,
     ) {
-        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<RoomId> {
-        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()
     }

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<SfuMessage>,
@@ -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<SfuMessage>> {
-        let shard_id = room_id.0.split("_").nth(1)?.parse::<usize>().ok();
+    pub fn sender_by_key(&self, key: &str) -> Option<&SyncSender<SfuMessage>> {
+        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<usize> {
+    key.split("_").nth(1)?.parse::<usize>().ok()
+}

src/types.rs

@@ -237,7 +237,7 @@ pub enum SfuMessage {
     },
     ResolveStreamKey {
         hashed_stream_key: [u8; 32],
-        reply: tokio::sync::mpsc::Sender<Option<RoomId>>,
+        reply: oneshot::Sender<Option<RoomId>>,
     },
     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] {

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();