From 73c8588cde7511688e31f500e7311f7bd649b4dc Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Wed, 14 Oct 2026 18:06:46 +0900 Subject: [PATCH 1/6] refactor(web): await media loop replies instead of blocking a worker --- src/servers/web.rs | 58 +++++++++++++++---------------- src/sfu/room.rs | 87 ++++++++++++++++++++++++++++------------------ src/types.rs | 21 +++++------ 3 files changed, 94 insertions(+), 72 deletions(-) diff --git a/src/servers/web.rs b/src/servers/web.rs index 44b0d7e..8d2fa0a 100644 --- a/src/servers/web.rs +++ b/src/servers/web.rs @@ -1,9 +1,6 @@ use std::{ net::{Ipv4Addr, SocketAddr}, - sync::{ - Arc, - mpsc::{self, Receiver, SyncSender}, - }, + sync::{Arc, mpsc::SyncSender}, time::Instant, }; @@ -31,7 +28,7 @@ use str0m::{ change::{SdpAnswer, SdpOffer}, }; use subtle::ConstantTimeEq; -use tokio::net::TcpListener; +use tokio::{net::TcpListener, sync::oneshot}; use tower_http::cors::CorsLayer; use tracing::error; use uuid::Uuid; @@ -211,7 +208,7 @@ impl FromRequestParts for AuthStreamKey { let hashed_stream_key = hash_string(bearer.token()); - let (tx, rx) = mpsc::sync_channel::>(state.shards.shards.len()); + let (tx, mut rx) = tokio::sync::mpsc::channel::>(state.shards.shards.len()); for sfu_tx in state.shards.senders() { let msg = SfuMessage::ResolveStreamKey { @@ -226,10 +223,13 @@ impl FromRequestParts for AuthStreamKey { drop(tx); - rx.iter() - .find_map(|reply| reply) - .map(AuthStreamKey) - .ok_or_else(unauthorized) + while let Some(reply) = rx.recv().await { + if let Some(room_id) = reply { + return Ok(AuthStreamKey(room_id)); + } + } + + Err(unauthorized()) } } @@ -240,7 +240,7 @@ struct CreateRoomResponse { } async fn create_room(State(state): State) -> Result { - let (tx, rx) = mpsc::sync_channel::<(RoomId, StreamKey)>(1); + let (tx, rx) = oneshot::channel::<(RoomId, StreamKey)>(); let room_id = RoomId::new(state.shards.next_shard()); @@ -255,7 +255,7 @@ async fn create_room(State(state): State) -> Result, State(state): State, ) -> Result, StatusCode> { - let (tx, rx) = mpsc::sync_channel::>(1); + let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); let Some(sfu_tx) = state.shards.sender_for(&room_id) else { @@ -302,7 +302,7 @@ async fn get_room( sfu_tx.send(SfuMessage::GetRoom { room_id, reply: tx }).ok(); let (views, state) = rx - .recv() + .await .map_err(|e| { error!("send to sfu loop failed: {}", e); StatusCode::INTERNAL_SERVER_ERROR @@ -322,7 +322,7 @@ async fn update_room( Path(room_id): Path, Json(payload): Json, ) -> Result<(), StatusCode> { - let (tx, rx) = mpsc::sync_channel::>(1); + let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); let Some(sfu_tx) = state.shards.sender_for(&room_id) else { @@ -340,7 +340,7 @@ async fn update_room( StatusCode::INTERNAL_SERVER_ERROR })?; - rx.recv() + rx.await .map_err(|e| { error!("send to sfu loop failed: {}", e); StatusCode::INTERNAL_SERVER_ERROR @@ -352,7 +352,7 @@ async fn delete_room( Path(room_id): Path, State(state): State, ) -> Result<(), StatusCode> { - let (tx, rx) = mpsc::sync_channel::>(1); + let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); let Some(sfu_tx) = state.shards.sender_for(&room_id) else { @@ -366,7 +366,7 @@ async fn delete_room( StatusCode::INTERNAL_SERVER_ERROR })?; - rx.recv() + rx.await .map_err(|e| { error!("send to sfu loop failed: {}", e); StatusCode::INTERNAL_SERVER_ERROR @@ -383,7 +383,7 @@ async fn reissue_stream_key( State(state): State, Path(room_id): Path, ) -> Result { - let (tx, rx) = mpsc::sync_channel::>(1); + let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); let Some(sfu_tx) = state.shards.sender_for(&room_id) else { @@ -397,7 +397,7 @@ async fn reissue_stream_key( StatusCode::INTERNAL_SERVER_ERROR })?; - let stream_key = rx.recv().map_err(|e| { + let stream_key = rx.await.map_err(|e| { error!("send to sfu loop failed: {}", e); StatusCode::INTERNAL_SERVER_ERROR })?; @@ -431,7 +431,7 @@ async fn sdp_offer( let (rtc, answer) = viewer_rtc_from_offer(sdp, state.addr)?; let session_id = Uuid::new_v4(); - let (tx, rx) = mpsc::sync_channel::>(1); + let (tx, rx) = oneshot::channel::>(); let room_id = RoomId(room_id); let Some(sfu_tx) = state.shards.sender_for(&room_id) else { @@ -445,7 +445,7 @@ async fn sdp_offer( reply: tx, }; - register_client(msg, rx, sfu_tx.to_owned())?; + register_client(msg, rx, sfu_tx.to_owned()).await?; let value = serde_json::to_value(&answer).map_err(|e| { error!("json to value failed: {}", e); @@ -470,7 +470,7 @@ async fn whip_sdp_offer( let (rtc, answer) = streamer_rtc_from_offer(sdp, state.addr)?; let session_id = Uuid::new_v4(); let location = format!("/whip/sessions/{}", session_id); - let (tx, rx) = mpsc::sync_channel::>(1); + let (tx, rx) = oneshot::channel::>(); let Some(sfu_tx) = state.shards.sender_for(&room_id) else { return Err(StatusCode::INTERNAL_SERVER_ERROR); @@ -483,7 +483,7 @@ async fn whip_sdp_offer( reply: tx, }; - register_client(msg, rx, sfu_tx.to_owned())?; + register_client(msg, rx, sfu_tx.to_owned()).await?; Ok(( StatusCode::CREATED, @@ -500,7 +500,7 @@ async fn whip_terminate_session( AuthStreamKey(room_id): AuthStreamKey, State(state): State, ) -> Result { - let (tx, rx) = mpsc::sync_channel::>(1); + let (tx, rx) = oneshot::channel::>(); let Some(sfu_tx) = state.shards.sender_for(&room_id) else { return Err(StatusCode::INTERNAL_SERVER_ERROR); @@ -517,7 +517,7 @@ async fn whip_terminate_session( StatusCode::INTERNAL_SERVER_ERROR })?; - rx.recv() + rx.await .map_err(|e| { error!("send to sfu loop failed: {}", e); StatusCode::INTERNAL_SERVER_ERROR @@ -563,9 +563,9 @@ fn accept_offer( Ok((rtc, answer)) } -fn register_client( +async fn register_client( msg: SfuMessage, - rx: Receiver>, + rx: oneshot::Receiver>, sfu_tx: SyncSender, ) -> Result<(), StatusCode> { sfu_tx.send(msg).map_err(|e| { @@ -573,7 +573,7 @@ fn register_client( StatusCode::INTERNAL_SERVER_ERROR })?; - if let Some(status_code) = rx.recv().map_err(|e| { + if let Some(status_code) = rx.await.map_err(|e| { error!("send to sfu loop failed: {}", e); StatusCode::INTERNAL_SERVER_ERROR })? { diff --git a/src/sfu/room.rs b/src/sfu/room.rs index 10ba4f6..6d8d53e 100644 --- a/src/sfu/room.rs +++ b/src/sfu/room.rs @@ -2,13 +2,14 @@ use std::{ collections::HashMap, ops::{Deref, DerefMut}, rc::Rc, - sync::mpsc::{Sender, SyncSender}, + sync::mpsc::Sender, time::Instant, }; use axum::http::StatusCode; use str0m::Rtc; use subtle::ConstantTimeEq; +use tokio::sync::oneshot; use tracing::{error, info, warn}; use uuid::Uuid; @@ -34,7 +35,7 @@ impl Rooms { rtc: Rtc, room_id: RoomId, session_id: Uuid, - reply: SyncSender>, + reply: oneshot::Sender>, ) { if let Some(room) = self.get_mut(&room_id) { room.add_viewer(rtc, session_id, reply); @@ -49,7 +50,7 @@ impl Rooms { rtc: Rtc, room_id: RoomId, session_id: Uuid, - reply: SyncSender>, + reply: oneshot::Sender>, ) { if let Some(room) = self.get_mut(&room_id) { room.add_streamer(rtc, session_id, reply); @@ -59,7 +60,7 @@ impl Rooms { } } - pub fn create(&mut self, room_id: RoomId, reply: SyncSender<(RoomId, StreamKey)>) { + pub fn create(&mut self, room_id: RoomId, reply: oneshot::Sender<(RoomId, StreamKey)>) { let stream_key = StreamKey::new(); let hashed_stream_key = stream_key.hashed(); @@ -76,7 +77,7 @@ impl Rooms { reply.send((room_id, stream_key)).ok(); } - pub fn get_room(&self, room_id: RoomId, reply: SyncSender>) { + pub fn get_room(&self, room_id: RoomId, reply: oneshot::Sender>) { match self.get(&room_id) { Some(room) => reply.send(Some((room.view_count(), room.state))).ok(), None => reply.send(None).ok(), @@ -87,7 +88,7 @@ impl Rooms { &mut self, room_id: RoomId, state: RoomState, - reply: SyncSender>, + reply: oneshot::Sender>, ) { match self.rooms.get_mut(&room_id) { Some(room) => { @@ -102,7 +103,7 @@ impl Rooms { }; } - pub fn delete(&mut self, room_id: RoomId, reply: SyncSender>) { + pub fn delete(&mut self, room_id: RoomId, reply: oneshot::Sender>) { match self.rooms.remove(&room_id) { Some(_) => { self.socket_routes.clean_routes_by_room_id(&room_id); @@ -115,20 +116,20 @@ impl Rooms { pub fn resolve_stream_key( &self, hashed_stream_key: [u8; 32], - reply: SyncSender>, + reply: tokio::sync::mpsc::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.send(room_id).ok(); + reply.try_send(room_id).ok(); } pub fn terminate_session( &mut self, room_id: RoomId, session_id: Uuid, - reply: SyncSender>, + reply: oneshot::Sender>, ) { if let Some(room) = self.rooms.get_mut(&room_id) && let Some(streamer) = &room.streamer @@ -142,7 +143,11 @@ impl Rooms { }; } - pub fn reissue_stream_key(&mut self, room_id: RoomId, reply: SyncSender>) { + pub fn reissue_stream_key( + &mut self, + room_id: RoomId, + reply: oneshot::Sender>, + ) { if let Some(room) = self.get_mut(&room_id) { let stream_key = StreamKey::new(); let hashed_stream_key = stream_key.hashed(); @@ -232,7 +237,12 @@ impl Room { self.state = state; } - fn add_viewer(&mut self, rtc: Rtc, session_id: Uuid, reply: SyncSender>) { + fn add_viewer( + &mut self, + rtc: Rtc, + session_id: Uuid, + reply: oneshot::Sender>, + ) { let mut viewer = Viewer::new(rtc, session_id); match self.state { @@ -272,7 +282,12 @@ impl Room { } } - fn add_streamer(&mut self, rtc: Rtc, session_id: Uuid, reply: SyncSender>) { + fn add_streamer( + &mut self, + rtc: Rtc, + session_id: Uuid, + reply: oneshot::Sender>, + ) { if self.streamer.is_none() { self.state = RoomState::Preview; @@ -310,6 +325,7 @@ mod tests { use axum::http::StatusCode; use str0m::Rtc; + use tokio::sync::oneshot; use uuid::Uuid; use crate::{ @@ -324,8 +340,8 @@ mod tests { (Rooms::new(tx, 0), rx) } - fn reply() -> (mpsc::SyncSender>, mpsc::Receiver>) { - mpsc::sync_channel(1) + fn reply() -> (oneshot::Sender>, oneshot::Receiver>) { + oneshot::channel() } fn rtc() -> Rtc { @@ -333,9 +349,15 @@ mod tests { } fn create_room(rooms: &mut Rooms) -> (RoomId, StreamKey) { - let (tx, rx) = mpsc::sync_channel(1); + let (tx, rx) = oneshot::channel(); rooms.create(RoomId::new(0), tx); - rx.recv().unwrap() + rx.blocking_recv().unwrap() + } + + fn resolve(rooms: &Rooms, hashed_stream_key: [u8; 32]) -> Option { + let (tx, mut rx) = tokio::sync::mpsc::channel(1); + rooms.resolve_stream_key(hashed_stream_key, tx); + rx.blocking_recv().unwrap() } fn missing_id() -> RoomId { @@ -380,15 +402,14 @@ mod tests { let (tx, rx) = reply(); rooms.reissue_stream_key(room_id.clone(), tx); - let new_key = rx.recv().unwrap().expect("room exists"); + let new_key = rx.blocking_recv().unwrap().expect("room exists"); - let (tx, rx) = reply(); - rooms.resolve_stream_key(old_key.hashed(), tx); - assert_eq!(rx.recv().unwrap(), None, "the old key must stop resolving"); - - let (tx, rx) = reply(); - rooms.resolve_stream_key(new_key.hashed(), tx); - assert_eq!(rx.recv().unwrap(), Some(room_id)); + assert_eq!( + resolve(&rooms, old_key.hashed()), + None, + "the old key must stop resolving" + ); + assert_eq!(resolve(&rooms, new_key.hashed()), Some(room_id)); } #[test] @@ -397,7 +418,7 @@ mod tests { let (tx, rx) = reply(); rooms.reissue_stream_key(missing_id(), tx); - assert!(rx.recv().unwrap().is_none()); + assert!(rx.blocking_recv().unwrap().is_none()); } #[test] @@ -408,7 +429,7 @@ mod tests { let (tx, rx) = reply(); rooms.update_state(room_id.clone(), RoomState::Live, tx); - assert_eq!(rx.recv().unwrap(), Some(())); + assert_eq!(rx.blocking_recv().unwrap(), Some(())); assert!(matches!( rooms.get(&room_id).unwrap().state, RoomState::Live @@ -416,7 +437,7 @@ mod tests { let (tx, rx) = reply(); rooms.update_state(missing_id(), RoomState::Live, tx); - assert_eq!(rx.recv().unwrap(), None); + assert_eq!(rx.blocking_recv().unwrap(), None); } #[test] @@ -434,20 +455,20 @@ mod tests { let (tx, rx) = reply(); rooms.update_state(room_id.clone(), RoomState::Live, tx); - assert_eq!(rx.recv().unwrap(), Some(())); + assert_eq!(rx.blocking_recv().unwrap(), Some(())); let (tx, _rx) = reply::(); rooms.register_viewer(rtc(), room_id.clone(), Uuid::new_v4(), tx); let (tx, rx) = reply(); rooms.get_room(room_id, tx); - let (views, state) = rx.recv().unwrap().unwrap(); + let (views, state) = rx.blocking_recv().unwrap().unwrap(); assert_eq!(views, 1); assert!(matches!(state, RoomState::Live)); let (tx, rx) = reply::<(usize, RoomState)>(); rooms.get_room(missing_id(), tx); - assert!(rx.recv().unwrap().is_none()); + assert!(rx.blocking_recv().unwrap().is_none()); } #[test] @@ -458,12 +479,12 @@ mod tests { let (tx, rx) = reply(); rooms.delete(room_id.clone(), tx); - assert_eq!(rx.recv().unwrap(), Some(())); + assert_eq!(rx.blocking_recv().unwrap(), Some(())); assert!(!rooms.contains_key(&room_id)); let (tx, rx) = reply(); rooms.delete(room_id, tx); - assert_eq!(rx.recv().unwrap(), None); + assert_eq!(rx.blocking_recv().unwrap(), None); } #[test] diff --git a/src/types.rs b/src/types.rs index 9ac3586..31e7e48 100644 --- a/src/types.rs +++ b/src/types.rs @@ -6,7 +6,7 @@ use std::{ rc::{Rc, Weak}, sync::{ atomic::{AtomicU64, Ordering}, - mpsc::{Sender, SyncSender}, + mpsc::Sender, }, time::Instant, }; @@ -22,6 +22,7 @@ use str0m::{ }; use derive_more::{Debug, Display, Eq}; +use tokio::sync::oneshot; use tracing::error; use uuid::Uuid; @@ -209,43 +210,43 @@ pub enum SfuMessage { rtc: Box, room_id: RoomId, session_id: Uuid, - reply: SyncSender>, + reply: oneshot::Sender>, }, RegisterStreamer { rtc: Box, room_id: RoomId, session_id: Uuid, - reply: SyncSender>, + reply: oneshot::Sender>, }, CreateRoom { room_id: RoomId, - reply: SyncSender<(RoomId, StreamKey)>, + reply: oneshot::Sender<(RoomId, StreamKey)>, }, DeleteRoom { room_id: RoomId, - reply: SyncSender>, + reply: oneshot::Sender>, }, UpdateRoomState { room_id: RoomId, state: RoomState, - reply: SyncSender>, + reply: oneshot::Sender>, }, GetRoom { room_id: RoomId, - reply: SyncSender>, + reply: oneshot::Sender>, }, ResolveStreamKey { hashed_stream_key: [u8; 32], - reply: SyncSender>, + reply: tokio::sync::mpsc::Sender>, }, TerminateSession { room_id: RoomId, session_id: Uuid, - reply: SyncSender>, + reply: oneshot::Sender>, }, ReissueStreamKey { room_id: RoomId, - reply: SyncSender>, + reply: oneshot::Sender>, }, Datagram { data: Vec, From 5c24a14b0461e4dc3b0b3b979c75c71dc2747cc6 Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Wed, 14 Oct 2026 18:06:46 +0900 Subject: [PATCH 2/6] docs(adr): replace ADR 0007 with replies over oneshot --- README.ko.md | 1 - README.md | 1 - docs/architecture/overview.md | 1 - .../0007-defer-sync-mpsc-recv-blocking.md | 2 +- docs/decisions/0012-reply-over-oneshot.md | 31 +++++++++++++++++++ 5 files changed, 32 insertions(+), 4 deletions(-) create mode 100644 docs/decisions/0012-reply-over-oneshot.md diff --git a/README.ko.md b/README.ko.md index 3866150..0e5dd0d 100644 --- a/README.ko.md +++ b/README.ko.md @@ -164,7 +164,6 @@ LOAD_METRICS_PORT=9464 cargo test --release --test load -- --ignored --nocapture - TURN이 없습니다. UDP `:40000`에 직접 닿지 못하는 클라이언트는 연결할 수 없습니다. - 방 하나는 미디어 루프 하나에서만 돌기 때문에 코어 하나를 넘지 못합니다. [#73](https://github.com/kimsun721/stream/issues/73)에서 방이 여러 루프에 걸치도록 바꿉니다. - 방과 스트림 키는 메모리에만 있어서 재시작하면 사라집니다. -- HTTP 핸들러는 미디어 루프의 응답을 기다리는 동안 tokio 워커를 붙잡습니다. [ADR 0007](docs/decisions/0007-defer-sync-mpsc-recv-blocking.md) 참고. - 프로세스 하나로 동작하며, 여러 서버로 부하를 나누지 않습니다. ## 문서 diff --git a/README.md b/README.md index 641ecc1..1eb2292 100644 --- a/README.md +++ b/README.md @@ -164,7 +164,6 @@ LOAD_METRICS_PORT=9464 cargo test --release --test load -- --ignored --nocapture - No TURN. A client that cannot reach UDP `:40000` directly cannot connect. - A room lives on one media loop, so one room is bounded by one core. [#73](https://github.com/kimsun721/stream/issues/73) lets a room span loops. - Rooms and stream keys live in memory. A restart drops them. -- An HTTP handler holds a tokio worker while it waits for a media loop to answer. See [ADR 0007](docs/decisions/0007-defer-sync-mpsc-recv-blocking.md). - One process. Nothing spreads load across servers. ## Docs diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index d91f8b7..aef61c5 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -80,7 +80,6 @@ The client opens a DataChannel **before** `createOffer`, so the DC is part of th ## Known limits -- HTTP handlers block on `recv()` while a loop works through its rooms. Deferred, see [ADR 0007](../decisions/0007-defer-sync-mpsc-recv-blocking.md). - A room lives on one loop, so one room is bounded by one core. [#73](https://github.com/kimsun721/stream/issues/73) places clients instead. - One process. - A keyframe requested by one viewer goes to every viewer on that layer. See [ADR 0004](../decisions/0004-keyframe-broadcast-with-debounce.md). diff --git a/docs/decisions/0007-defer-sync-mpsc-recv-blocking.md b/docs/decisions/0007-defer-sync-mpsc-recv-blocking.md index d2ebd20..6c36f2a 100644 --- a/docs/decisions/0007-defer-sync-mpsc-recv-blocking.md +++ b/docs/decisions/0007-defer-sync-mpsc-recv-blocking.md @@ -2,7 +2,7 @@ ## Status -Accepted (2026-05-11) +Superseded by [ADR 0012](0012-reply-over-oneshot.md) (2026-10-08). Accepted (2026-05-11). ## Context diff --git a/docs/decisions/0012-reply-over-oneshot.md b/docs/decisions/0012-reply-over-oneshot.md new file mode 100644 index 0000000..a831d77 --- /dev/null +++ b/docs/decisions/0012-reply-over-oneshot.md @@ -0,0 +1,31 @@ +# 0012. Reply to HTTP handlers over `tokio::sync::oneshot` + +## Status + +Accepted (2026-10-08). Supersedes [ADR 0007](0007-defer-sync-mpsc-recv-blocking.md). + +## Context + +[ADR 0007](0007-defer-sync-mpsc-recv-blocking.md) left HTTP handlers blocking on `std::sync::mpsc::recv()` for a media loop's reply. It rejected `tokio::sync` channels because they would "couple the std-thread SFU to a tokio runtime or require bridging". + +That premise was wrong when it was written. The types in `tokio::sync` are runtime agnostic, and `oneshot::Sender::send` is a plain function that a std thread can call. Nothing changed since to make it possible. The alternative was misjudged. + +The deferral has a cost that is listed among the README's limits: a handler holds a tokio worker for as long as a loop takes to answer. + +## Decision + +- A reply travels over `tokio::sync::oneshot`. The media loop sends with a plain call and the handler awaits. +- Stream key resolution asks every loop, so its reply uses `tokio::sync::mpsc` with one sender per loop. A loop sends with `try_send`, which needs no runtime either, and the channel holds one reply per loop, so it never fills. +- The request channel stays `std::sync::mpsc::sync_channel`. A loop waits on it until str0m's next deadline, and tokio's `mpsc` has no blocking receive with a timeout outside a runtime. The mux feeds the same channel. + +## Alternatives considered + +- **`spawn_blocking` around the wait.** Frees the worker but parks a thread per waiting request. +- **Move the request channel to tokio as well.** No blocking receive with a timeout. +- **Keep deferring.** The reason given for deferring does not hold. + +## Consequences + +**Gain**: a handler no longer holds a worker while it waits; a handler can put a timeout on a reply; one limit leaves the README. + +**Lose**: `SfuMessage` now names tokio types, so the loop's message type depends on the tokio crate, though not on a runtime. From f99b33971a3474f5b0b7d0a00a888f429569f592 Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Wed, 14 Oct 2026 19:53:57 +0900 Subject: [PATCH 3/6] refactor(web): answer 503 when a media loop's channel is full --- src/servers/web.rs | 56 +++++++++++++++++++++------------------------- 1 file changed, 26 insertions(+), 30 deletions(-) diff --git a/src/servers/web.rs b/src/servers/web.rs index 8d2fa0a..c9db896 100644 --- a/src/servers/web.rs +++ b/src/servers/web.rs @@ -1,6 +1,9 @@ use std::{ net::{Ipv4Addr, SocketAddr}, - sync::{Arc, mpsc::SyncSender}, + sync::{ + Arc, + mpsc::{SyncSender, TrySendError}, + }, time::Instant, }; @@ -215,10 +218,8 @@ impl FromRequestParts for AuthStreamKey { hashed_stream_key, reply: tx.clone(), }; - sfu_tx.send(msg).map_err(|e| { - error!("send to sfu loop failed: {}", e); - StatusCode::INTERNAL_SERVER_ERROR.into_response() - })?; + + send_to_sfu(sfu_tx, msg).map_err(IntoResponse::into_response)?; } drop(tx); @@ -250,10 +251,7 @@ async fn create_room(State(state): State) -> Result>, sfu_tx: SyncSender, ) -> Result<(), StatusCode> { - sfu_tx.send(msg).map_err(|e| { - error!("send to sfu loop failed: {}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; + send_to_sfu(&sfu_tx, msg)?; if let Some(status_code) = rx.await.map_err(|e| { error!("send to sfu loop failed: {}", e); @@ -582,3 +565,16 @@ async fn register_client( Ok(()) } + +fn send_to_sfu(sfu_tx: &SyncSender, msg: SfuMessage) -> Result<(), StatusCode> { + sfu_tx.try_send(msg).map_err(|e| match e { + TrySendError::Disconnected(_) => { + error!("send to sfu loop failed: {}", e); + StatusCode::INTERNAL_SERVER_ERROR + } + TrySendError::Full(_) => { + error!("send to sfu loop failed: {}", e); + StatusCode::SERVICE_UNAVAILABLE + } + }) +} From 7551a057b9a363561e11765d058f0f92d84ccecc Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Wed, 14 Oct 2026 19:53:57 +0900 Subject: [PATCH 4/6] docs(adr): record try_send and 503 in ADR 0012 --- docs/architecture/overview.md | 2 +- docs/decisions/0012-reply-over-oneshot.md | 7 +++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index aef61c5..1ba7ff3 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -22,7 +22,7 @@ The external backend is treated as a black box that speaks HTTP. This server mak - mux: one std thread, blocked on the socket. - Media loops: `server.total_shards` std threads, the core count by default. Each is the sole owner of its `Rooms`, as [ADR 0001](../decisions/0001-actor-pattern.md) describes for the single loop it started with. -Every loop has one `sync_channel` that both the web threads and the mux write to, so a loop waits on one thing: its channel, until the next str0m deadline. The mux copies each datagram into an owned buffer to cross the thread boundary, and drops it rather than wait when a loop's channel is full. +Every loop has one `sync_channel` that both the web threads and the mux write to, so a loop waits on one thing: its channel, until the next str0m deadline. The mux copies each datagram into an owned buffer to cross the thread boundary, and drops it rather than wait when a loop's channel is full. The web threads do not wait either: a request to a full loop gets `503`. All loops send on the same socket, so clients see one address. diff --git a/docs/decisions/0012-reply-over-oneshot.md b/docs/decisions/0012-reply-over-oneshot.md index a831d77..e227284 100644 --- a/docs/decisions/0012-reply-over-oneshot.md +++ b/docs/decisions/0012-reply-over-oneshot.md @@ -17,15 +17,18 @@ The deferral has a cost that is listed among the README's limits: a handler hold - A reply travels over `tokio::sync::oneshot`. The media loop sends with a plain call and the handler awaits. - Stream key resolution asks every loop, so its reply uses `tokio::sync::mpsc` with one sender per loop. A loop sends with `try_send`, which needs no runtime either, and the channel holds one reply per loop, so it never fills. - The request channel stays `std::sync::mpsc::sync_channel`. A loop waits on it until str0m's next deadline, and tokio's `mpsc` has no blocking receive with a timeout outside a runtime. The mux feeds the same channel. +- A handler sends a request with `try_send` and answers `503` when the loop's channel is full. The channel also carries media from the mux, and media that waits reaches viewers late or not at all, while a refused request costs the client a retry. +- Stream key resolution answers `503` if any loop's channel is full, not `401`, since the key may belong to that loop. ## Alternatives considered - **`spawn_blocking` around the wait.** Frees the worker but parks a thread per waiting request. - **Move the request channel to tokio as well.** No blocking receive with a timeout. - **Keep deferring.** The reason given for deferring does not hold. +- **Blocking send, or an unbounded request channel.** Either keeps every request at the cost of media. A handler blocked on a full channel takes the next free slot ahead of the mux, and an unbounded channel queues datagrams behind requests without limit. ## Consequences -**Gain**: a handler no longer holds a worker while it waits; a handler can put a timeout on a reply; one limit leaves the README. +**Gain**: a handler no longer holds a worker while it waits or while a loop is full; a handler can put a timeout on a reply; one limit leaves the README. -**Lose**: `SfuMessage` now names tokio types, so the loop's message type depends on the tokio crate, though not on a runtime. +**Lose**: `SfuMessage` now names tokio types, so the loop's message type depends on the tokio crate, though not on a runtime. A request is refused while its loop's channel is full, and the client has to retry. From 21ee632a6b91cafe76743abad2c0400046a66544 Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Wed, 14 Oct 2026 19:58:57 +0900 Subject: [PATCH 5/6] feat(metrics): count requests refused on a full channel --- observability/grafana/dashboards/stream.json | 41 ++++++++++++++++++++ src/metrics.rs | 10 +++++ src/metrics/exposition.rs | 11 ++++-- src/servers/web.rs | 2 +- 4 files changed, 60 insertions(+), 4 deletions(-) diff --git a/observability/grafana/dashboards/stream.json b/observability/grafana/dashboards/stream.json index 257e592..ff7658b 100644 --- a/observability/grafana/dashboards/stream.json +++ b/observability/grafana/dashboards/stream.json @@ -438,6 +438,47 @@ "mode": "multi" } } + }, + { + "id": 10, + "type": "timeseries", + "title": "Refused requests", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 32 + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "refId": "A", + "expr": "rate(stream_requests_refused_total[$__rate_interval])", + "legendFormat": "503 on a full channel" + } + ], + "fieldConfig": { + "defaults": { + "unit": "reqps" + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom" + }, + "tooltip": { + "mode": "multi" + } + } } ], "templating": { diff --git a/src/metrics.rs b/src/metrics.rs index 58a66db..f936e2a 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -173,6 +173,7 @@ atomics!( CHANNEL_MESSAGES, CHANNEL_DRAIN_FULL, CHANNEL_DROPS, + REQUESTS_REFUSED, SOCKET_READS, SOCKET_READ_BYTES, ROOMS_VISITED, @@ -210,6 +211,13 @@ pub fn channel_dropped() { CHANNEL_DROPS.fetch_add(1, Relaxed); } +/// A web request answered `503` because its media loop's channel was full. +/// Counted rather than logged, since it arrives in floods. Stream key +/// resolution asks every loop but counts once. +pub fn request_refused() { + REQUESTS_REFUSED.fetch_add(1, Relaxed); +} + pub fn socket_read(bytes: usize) { SOCKET_READS.fetch_add(1, Relaxed); SOCKET_READ_BYTES.fetch_add(bytes as u64, Relaxed); @@ -352,6 +360,7 @@ pub struct Snapshot { pub channel_messages: u64, pub channel_drain_full: u64, pub channel_drops: u64, + pub requests_refused: u64, pub socket_reads: u64, pub socket_read_bytes: u64, @@ -394,6 +403,7 @@ pub fn snapshot() -> Snapshot { channel_messages: CHANNEL_MESSAGES.load(Relaxed), channel_drain_full: CHANNEL_DRAIN_FULL.load(Relaxed), channel_drops: CHANNEL_DROPS.load(Relaxed), + requests_refused: REQUESTS_REFUSED.load(Relaxed), socket_reads: SOCKET_READS.load(Relaxed), socket_read_bytes: SOCKET_READ_BYTES.load(Relaxed), diff --git a/src/metrics/exposition.rs b/src/metrics/exposition.rs index cad8162..56646cc 100644 --- a/src/metrics/exposition.rs +++ b/src/metrics/exposition.rs @@ -15,8 +15,8 @@ use super::{ BUCKET_COUNT, CHANNEL_DRAIN_FULL, CHANNEL_DROPS, CHANNEL_MESSAGES, CHANNEL_WAIT, CLIENT_TICK_ERRORS, CLIENTS_DISCONNECTED, DEMOTIONS, Durations, FORWARD_DELAY, Histogram, LAP_LATENESS, LAPS, MEDIA_DATAS, MEDIA_WRITE_BYTES, MEDIA_WRITES, PHASES, PROMOTIONS, Phase, - RELAY_SAVED_BYTES, RELAY_SAVED_FRAMES, ROOMS_SKIPPED, ROOMS_VISITED, SENT_BYTES, SENT_PACKETS, - SOCKET_READ_BYTES, SOCKET_READS, ShardGauges, shards, + RELAY_SAVED_BYTES, RELAY_SAVED_FRAMES, REQUESTS_REFUSED, ROOMS_SKIPPED, ROOMS_VISITED, + SENT_BYTES, SENT_PACKETS, SOCKET_READ_BYTES, SOCKET_READS, ShardGauges, shards, }; pub const CONTENT_TYPE: &str = "application/openmetrics-text; version=1.0.0; charset=utf-8"; @@ -82,7 +82,7 @@ impl Collector for Exposition { } } -fn counters() -> [(&'static str, &'static str, &'static AtomicU64); 19] { +fn counters() -> [(&'static str, &'static str, &'static AtomicU64); 20] { [ ("stream_laps", "Media loop iterations", &LAPS), ( @@ -100,6 +100,11 @@ fn counters() -> [(&'static str, &'static str, &'static AtomicU64); 19] { "Datagrams the mux read but dropped because a loop's channel was full", &CHANNEL_DROPS, ), + ( + "stream_requests_refused", + "Web requests answered 503 because a loop's channel was full", + &REQUESTS_REFUSED, + ), ( "stream_socket_reads", "Datagrams read from the media socket", diff --git a/src/servers/web.rs b/src/servers/web.rs index c9db896..fea3e93 100644 --- a/src/servers/web.rs +++ b/src/servers/web.rs @@ -573,7 +573,7 @@ fn send_to_sfu(sfu_tx: &SyncSender, msg: SfuMessage) -> Result<(), S StatusCode::INTERNAL_SERVER_ERROR } TrySendError::Full(_) => { - error!("send to sfu loop failed: {}", e); + metrics::request_refused(); StatusCode::SERVICE_UNAVAILABLE } }) From 84482ef8fe78846ea050c56a33c35987337d6e3a Mon Sep 17 00:00:00 2001 From: kimsun721 Date: Wed, 14 Oct 2026 19:58:57 +0900 Subject: [PATCH 6/6] docs(adr): note that refusals are counted, not logged --- docs/decisions/0012-reply-over-oneshot.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/decisions/0012-reply-over-oneshot.md b/docs/decisions/0012-reply-over-oneshot.md index e227284..6993984 100644 --- a/docs/decisions/0012-reply-over-oneshot.md +++ b/docs/decisions/0012-reply-over-oneshot.md @@ -17,7 +17,7 @@ The deferral has a cost that is listed among the README's limits: a handler hold - A reply travels over `tokio::sync::oneshot`. The media loop sends with a plain call and the handler awaits. - Stream key resolution asks every loop, so its reply uses `tokio::sync::mpsc` with one sender per loop. A loop sends with `try_send`, which needs no runtime either, and the channel holds one reply per loop, so it never fills. - The request channel stays `std::sync::mpsc::sync_channel`. A loop waits on it until str0m's next deadline, and tokio's `mpsc` has no blocking receive with a timeout outside a runtime. The mux feeds the same channel. -- A handler sends a request with `try_send` and answers `503` when the loop's channel is full. The channel also carries media from the mux, and media that waits reaches viewers late or not at all, while a refused request costs the client a retry. +- A handler sends a request with `try_send` and answers `503` when the loop's channel is full. The channel also carries media from the mux, and media that waits reaches viewers late or not at all, while a refused request costs the client a retry. A refusal is counted in `stream_requests_refused` rather than logged, since refusals come in floods. - Stream key resolution answers `503` if any loop's channel is full, not `401`, since the key may belong to that loop. ## Alternatives considered