o_sfu_core/engine/media_transport/rtc/
commands.rs

1//! mailbox command contract for the RTC worker engine
2//!
3//! `MediaTransport` production paths and cfg-gated worker harnesses translate
4//! transport intent into these values before the packet-loop task dispatches
5//! them while it owns mutable rtc state
6//! request commands carry a oneshot response
7//! fire-and-forget route controls are best-effort because they may target a
8//! worker that has already torn down the corresponding relay or session
9
10use std::sync::Arc;
11
12use o_sfu_rfc::webrtc::sdp;
13use o_sfu_router::rtp::MediaStream as RouterRtpParameters;
14use str0m::{
15    change::{SdpAnswer, SdpOffer},
16    media::{KeyframeRequestKind, MediaKind, Rid},
17};
18use tokio::sync::{mpsc, oneshot};
19
20use super::{
21    codec::{ParsedAnswerRids, RepairSummary, validate_answer_sdp},
22    relay_registry::{RelayPacketMailbox, RelayTargetId},
23    route_control::PacketLayerGate,
24};
25use crate::engine::{
26    media_transport::{
27        ActiveSpeakerSource, AppliedSessionAnswer, ConsumerRouteControl,
28        ConsumerRouteControlOutcome, ProducerRouteControl, ReceiverBweTargetUpdate, SessionOffer,
29        SessionUploadSlot, SourceActivityUpdate, TransportAdapterError, TransportConsumerRoute,
30        TransportMediaId, TransportResult, TransportSessionKey, TransportSourceDiagnosticsSnapshot,
31        TransportSourceKey,
32    },
33    metrics::{RtcMetricsRecorder, RtcRemoteControlDropKind, RtcRemotePacketGateConvergence},
34};
35
36/// command handle used by remote consumers to push control back to a source worker
37///
38/// a route that consumes media from another worker keeps this handle beside the
39/// remote-source registration
40/// later keyframe or layer-gate requests can then reach the worker that owns
41/// the producer without exposing the source worker internals
42///
43/// sends are deliberately best-effort
44/// stale remote routes, closed workers and full mailboxes are normal during
45/// teardown or topology churn
46#[derive(Debug, Clone)]
47pub struct RemoteSourceControl {
48    tx: mpsc::Sender<RtcWorkerCommand>,
49    target_id: RelayTargetId,
50    rtc_metrics: Arc<RtcMetricsRecorder>,
51}
52
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub(super) enum RemoteControlSendOutcome {
55    Forwarded,
56    Full,
57    Closed,
58}
59
60impl RemoteSourceControl {
61    /// creates a source-control handle for one relay target on a worker mailbox
62    pub(super) fn new(
63        tx: mpsc::Sender<RtcWorkerCommand>,
64        target_id: RelayTargetId,
65        rtc_metrics: Arc<RtcMetricsRecorder>,
66    ) -> Self {
67        Self {
68            tx,
69            target_id,
70            rtc_metrics,
71        }
72    }
73
74    /// asks the source worker to request a keyframe for a remote consumer
75    ///
76    /// this never waits for the source worker
77    /// `Full` retains caller-side retry authority while `Closed` ends it
78    pub(super) fn request_kf(
79        &self,
80        source: &TransportSourceKey,
81        rid: Option<Rid>,
82        kind: KeyframeRequestKind,
83    ) -> RemoteControlSendOutcome {
84        self.send_command(
85            RouteControlRequest::RequestRemoteKeyframe {
86                source: source.clone(),
87                target_id: self.target_id,
88                rid,
89                kind,
90            },
91            RtcRemoteControlDropKind::Keyframe,
92        )
93    }
94
95    /// publishes the effective remote-source packet gate to the source worker
96    pub(super) fn set_pkt_gate(
97        &self,
98        source: &TransportSourceKey,
99        packet_gate: PacketLayerGate,
100    ) -> bool {
101        self.send_command(
102            RouteControlRequest::SetRemoteSourcePacketGate {
103                source: source.clone(),
104                target_id: self.target_id,
105                packet_gate,
106            },
107            RtcRemoteControlDropKind::PacketGate,
108        ) == RemoteControlSendOutcome::Forwarded
109    }
110
111    pub(super) fn record_pkt_gate_retry(&self) {
112        self.rtc_metrics
113            .record_rtc_remote_packet_gate_convergence(RtcRemotePacketGateConvergence::Retry);
114    }
115
116    pub(super) fn record_pkt_gate_flushed(&self) {
117        self.rtc_metrics
118            .record_rtc_remote_packet_gate_convergence(RtcRemotePacketGateConvergence::Flushed);
119    }
120
121    #[inline]
122    fn send_command(
123        &self,
124        request: RouteControlRequest,
125        drop_kind: RtcRemoteControlDropKind,
126    ) -> RemoteControlSendOutcome {
127        match self.tx.try_send(RtcWorkerCommand::RouteControl {
128            request,
129            response: None,
130        }) {
131            Ok(()) => RemoteControlSendOutcome::Forwarded,
132            Err(mpsc::error::TrySendError::Full(_)) => {
133                self.rtc_metrics.record_rtc_remote_control_drop(drop_kind);
134                RemoteControlSendOutcome::Full
135            }
136            Err(mpsc::error::TrySendError::Closed(_)) => {
137                self.rtc_metrics.record_rtc_remote_control_drop(drop_kind);
138                RemoteControlSendOutcome::Closed
139            }
140        }
141    }
142}
143
144/// Response channel for a command executed by the packet loop.
145///
146/// Dropping the receiver cancels only the API wait and does not retract an
147/// enqueued command. Terminal worker cancellation may still win before dispatch.
148pub type RtcWorkerResponse<T> = oneshot::Sender<TransportResult<T>>;
149
150/// SDP answer parsed before packet-loop command delivery.
151pub struct ParsedSessionAnswer {
152    pub(super) answer: SdpAnswer,
153    pub(super) rids: ParsedAnswerRids,
154    pub(super) repair: RepairSummary,
155}
156
157impl ParsedSessionAnswer {
158    /// Validates and parses a remote SDP answer.
159    ///
160    /// # Errors
161    ///
162    /// Returns [`TransportAdapterError::InvalidInput`] for malformed repair
163    /// topology or SDP that str0m cannot parse.
164    pub(in crate::engine::media_transport) fn parse(answer_sdp: &str) -> TransportResult<Self> {
165        let repair = validate_answer_sdp(answer_sdp)?;
166        let answer = SdpAnswer::from_sdp_string(answer_sdp)
167            .map_err(|_error| TransportAdapterError::InvalidInput)?;
168        let rids = ParsedAnswerRids::parse(answer_sdp, &answer);
169        Ok(Self {
170            answer,
171            rids,
172            repair,
173        })
174    }
175}
176
177pub struct RtcSessionOffer {
178    offer: SdpOffer,
179    upload_slots: Vec<SessionUploadSlot>,
180}
181
182impl RtcSessionOffer {
183    pub(super) fn new(offer: SdpOffer, upload_slots: Vec<SessionUploadSlot>) -> Self {
184        Self {
185            offer,
186            upload_slots,
187        }
188    }
189
190    pub(in crate::engine::media_transport) fn into_session_offer(self) -> SessionOffer {
191        let mut sdp = self.offer.to_sdp_string();
192        let media_line_start = sdp.find("\r\nm=").map_or(sdp.len(), |index| index + 2);
193        sdp.insert_str(media_line_start, sdp::EOC_LINE);
194        SessionOffer::new(sdp).with_upload_slots(self.upload_slots)
195    }
196}
197
198#[derive(Debug)]
199pub enum WorkerMediaControlBatch {
200    ReceiverBwe(Vec<(usize, ReceiverBweTargetUpdate)>),
201    ProducerActivity(Vec<(usize, ProducerRouteControl)>),
202    ConsumerGates {
203        source: TransportSourceKey,
204        updates: Vec<(usize, TransportConsumerRoute, PacketLayerGate)>,
205    },
206    ConsumerFollowUp(Vec<(usize, ConsumerRouteControl)>),
207}
208
209#[derive(Debug)]
210pub enum WorkerMediaControlBatchOutcome {
211    Applied(Vec<TransportResult<()>>),
212    Consumers(Vec<ConsumerRouteControlOutcome>),
213}
214
215pub enum RouteControlRequest {
216    AddRelayTarget {
217        source: TransportSourceKey,
218        target_id: RelayTargetId,
219        target: RelayPacketMailbox,
220    },
221    RemoveRelayTarget {
222        source: TransportSourceKey,
223        target_id: RelayTargetId,
224    },
225    SetRelayTargetActive {
226        source: TransportSourceKey,
227        target_id: RelayTargetId,
228        active: bool,
229    },
230    SetRemoteSourceActivity {
231        source: TransportSourceKey,
232        update: SourceActivityUpdate,
233    },
234    RequestRemoteKeyframe {
235        source: TransportSourceKey,
236        target_id: RelayTargetId,
237        rid: Option<Rid>,
238        kind: KeyframeRequestKind,
239    },
240    SetRemoteSourcePacketGate {
241        source: TransportSourceKey,
242        target_id: RelayTargetId,
243        packet_gate: PacketLayerGate,
244    },
245}
246
247/// production command handled by the rtc packet-loop task
248///
249/// variants are grouped by ownership boundary: negotiation mutates str0m SDP
250/// state, media commands mutate producer or consumer registrations, relay
251/// commands mutate cross-worker fanout and observability commands read
252/// worker-local snapshots
253pub enum RtcWorkerCommand {
254    /// create the initial capability-probe offer before media registration
255    ///
256    /// worker startup binds the shared UDP socket, then this command lazily creates
257    /// session RTC state from its candidate address
258    /// it rejects a pending offer, a committed initial answer or registered media
259    CreateInitialSessionOffer {
260        room_id: Arc<str>,
261        session_key: TransportSessionKey,
262        response: RtcWorkerResponse<RtcSessionOffer>,
263    },
264    /// drain a staged follow-up offer after media topology changed
265    ///
266    /// media add and remove commands stage the SDP work before this command
267    /// runs
268    /// this command hands the staged offer to the worker API and preserves the
269    /// one-outstanding-offer rule owned by the worker
270    CreateSessionRenegotiationOffer {
271        session_key: TransportSessionKey,
272        response: RtcWorkerResponse<RtcSessionOffer>,
273    },
274    /// read active-speaker sources from worker-local route-control state
275    ///
276    /// the result is a cold-path observation for room policy
277    /// it does not mutate route state or packet-loop scheduling
278    ActiveSpeakerSourceSnapshot {
279        response: RtcWorkerResponse<Vec<ActiveSpeakerSource>>,
280    },
281    /// Read packet activity and active-speaker facts for selected sources.
282    SourceDiagnosticsSnapshot {
283        transport_media_ids: Vec<TransportMediaId>,
284        response: RtcWorkerResponse<TransportSourceDiagnosticsSnapshot>,
285    },
286    /// accept the answer for the current pending local offer
287    ///
288    /// this commits str0m SDP state, marks the session dirty, refreshes
289    /// negotiated producer parameters, registers remote candidate recovery hints
290    /// and returns the producer details that became usable after the answer
291    ApplySessionAnswer {
292        session_key: TransportSessionKey,
293        answer: ParsedSessionAnswer,
294        response: RtcWorkerResponse<AppliedSessionAnswer>,
295    },
296    /// remove a session from worker state
297    ///
298    /// teardown removes rtc state, media handles, route destinations, demux
299    /// indexes, bitrate counters and snapshot entries owned by the session
300    CloseSession {
301        session_key: TransportSessionKey,
302        response: RtcWorkerResponse<()>,
303    },
304    /// remove one producer or consumer media registration owned by a session
305    ///
306    /// producer removal drops incoming bitrate tracking and the source route
307    /// consumer removal drops local rewrite state and the destination route
308    /// negotiated media removal may stage the next SDP offer before the handle
309    /// leaves the public registry
310    RemoveMedia {
311        session_key: TransportSessionKey,
312        transport_media_id: TransportMediaId,
313        response: RtcWorkerResponse<()>,
314    },
315    /// resolve negotiated producer parameters for adapter tests
316    ///
317    /// this test-only command reads the answer-derived producer state after
318    /// negotiation so adapter tests can assert the transport boundary without
319    /// reaching into private registries
320    #[cfg(test)]
321    ResolveNegotiatedProducerParameters {
322        session_key: TransportSessionKey,
323        transport_media_id: TransportMediaId,
324        response: RtcWorkerResponse<RouterRtpParameters>,
325    },
326    /// resolve the MID stored by a registered media handle
327    ///
328    /// returns `None` when this worker has no current handle for the id, including
329    /// after media removal
330    ResolveMediaMid {
331        transport_media_id: TransportMediaId,
332        response: RtcWorkerResponse<Option<String>>,
333    },
334    /// register one browser upload as worker-local producer media
335    ///
336    /// before the initial answer this can declare receive state directly in
337    /// str0m
338    /// after negotiation it stages a recv-only m-section plus pending
339    /// receive identities, then registers bitrate counters and the producer
340    /// media handle
341    AddRecvMedia {
342        session_key: TransportSessionKey,
343        media_kind: MediaKind,
344        rtp_parameters: RouterRtpParameters,
345        response: RtcWorkerResponse<TransportMediaId>,
346    },
347    /// register one browser download as consumer media for a source
348    ///
349    /// the worker validates local or remote source ownership, stages or declares
350    /// send-only media, registers the consumer handle and creates the packet-loop
351    /// route destination
352    /// remote sources install rollback-protected control so failed consumer
353    /// setup does not leave stale relay state behind
354    AddSendMedia {
355        consumer_key: TransportSessionKey,
356        media_kind: MediaKind,
357        source: TransportSourceKey,
358        remote_source_control: Option<RemoteSourceControl>,
359        consumer_rtp_parameters: RouterRtpParameters,
360        active: bool,
361        response: RtcWorkerResponse<TransportMediaId>,
362    },
363    ApplyMediaControlBatch {
364        batch: WorkerMediaControlBatch,
365        response: RtcWorkerResponse<WorkerMediaControlBatchOutcome>,
366    },
367    RouteControl {
368        request: RouteControlRequest,
369        response: Option<RtcWorkerResponse<()>>,
370    },
371}
o_sfu_core/engine/media_transport/rtc/commands.rs

o_sfu_core/engine/media_transport/rtc/
commands.rs
