o_sfu_core/engine/room/effects/
batch.rs

1use super::{output::RoomOutputPlan, transport::RoomTransportPlan};
2use crate::engine::{
3    media_transport::MediaTransport,
4    room::{
5        Room, SourcePolicyGuard,
6        media_graph::{
7            ConsumerSetupOrigin, ProducerActivityCommit, PublishCommit, ReceiverRouteCommit,
8            ReceiverRouteWork,
9        },
10        source_policy::SourcePolicyTurn,
11        state::{
12            ConnectionCloseCommit, DisconnectCommit, LifecycleEffects, PresenceCommit,
13            UserJoinedFanout,
14        },
15    },
16};
17
18#[derive(Debug, Clone, Copy)]
19pub struct RoomEffectContext<'a> {
20    media_transport: Option<&'a MediaTransport>,
21    route_effects: bool,
22    joined_fanout: UserJoinedFanout,
23}
24
25impl<'a> RoomEffectContext<'a> {
26    pub const fn runtime(media_transport: &'a MediaTransport) -> Self {
27        Self {
28            media_transport: Some(media_transport),
29            route_effects: true,
30            joined_fanout: UserJoinedFanout::Emit,
31        }
32    }
33
34    #[cfg(any(test, feature = "testing-transport"))]
35    pub const fn state_only(media_transport: Option<&'a MediaTransport>) -> Self {
36        Self {
37            media_transport,
38            route_effects: false,
39            joined_fanout: UserJoinedFanout::Suppress,
40        }
41    }
42
43    pub(in crate::engine::room) const fn user_joined_fanout(self) -> UserJoinedFanout {
44        self.joined_fanout
45    }
46
47    fn media_transport(self) -> Option<&'a MediaTransport> {
48        self.media_transport
49    }
50
51    fn route_transport(self) -> Option<&'a MediaTransport> {
52        self.route_effects.then_some(self.media_transport).flatten()
53    }
54}
55
56/// batches post-lock transport, signaling and policy side-effects for room state transitions
57///
58/// ```text
59/// room state mutation (holds RoomState write lock)
60///   - mutate in-memory graph
61///   - return commit data
62///             |
63///             v  drop RoomState write lock
64/// RoomEffects::from_*(commit) -> execute
65///             |
66///             v  step 1: transport execution
67///   +-----------------------------------------------------------+
68///   | - create/remove local consumer routes on workers          |
69///   | - register/remove cross-worker relay route targets        |
70///   | - dispatch session teardowns                              |
71///   +-----------------------------------------------------------+
72///             |
73///             v  step 2: RoomOutputPlan pre-policy fanout
74///   +-----------------------------------------------------------+
75///   | - send track snapshots and presence user-info fanout      |
76///   +-----------------------------------------------------------+
77///             |
78///             v  step 3: source policy turn
79///   +-----------------------------------------------------------+
80///   | - re-evaluate audio admission and video bandwidth solver  |
81///   | - commit packet gate and BWE target updates               |
82///   +-----------------------------------------------------------+
83///             |
84///             v  step 4: RoomOutputPlan post-policy fanout
85///   +-----------------------------------------------------------+
86///   | - send user-info and lifecycle close/track/fanout output  |
87///   +-----------------------------------------------------------+
88/// ```
89///
90/// Publication commits return [`PublicationEffects`] so their execution requires
91/// the source-policy guard held since the state commit.
92#[derive(Debug, Default)]
93#[must_use = "room effect batches must be executed after the state transition commits"]
94pub struct RoomEffects {
95    transport: RoomTransportPlan,
96    output: RoomOutputPlan,
97    source_policy: SourcePolicyTurn,
98}
99
100/// Requires the source-policy guard held continuously from the publication commit.
101/// Activation emits user-info and applies policy before transport opens packet gates.
102#[derive(Debug)]
103#[must_use = "publication effects must execute with the guard held since their state commit"]
104pub(in crate::engine::room) struct PublicationEffects {
105    batch: RoomEffects,
106    policy_before_transport: bool,
107}
108
109impl RoomEffects {
110    pub(in crate::engine::room) fn from_join(
111        effects: LifecycleEffects,
112        transport_plan: RoomTransportPlan,
113    ) -> Self {
114        let mut batch = Self {
115            transport: transport_plan,
116            ..Self::default()
117        };
118        batch.source_policy.request();
119        batch.output.lifecycle = effects;
120        batch
121    }
122
123    pub(in crate::engine::room) fn from_connection_close(commit: ConnectionCloseCommit) -> Self {
124        let mut batch = Self::default();
125        match commit {
126            ConnectionCloseCommit::Current {
127                session_teardown,
128                effects,
129                transport_plan,
130                ..
131            } => {
132                batch.transport = transport_plan;
133                batch.output.lifecycle = effects;
134                batch.source_policy.request();
135                batch.transport.extend_teardown(session_teardown);
136            }
137            ConnectionCloseCommit::StalePlacement { session_teardown } => {
138                batch.transport.extend_teardown([session_teardown]);
139            }
140        }
141        batch
142    }
143
144    pub(in crate::engine::room) fn from_disconnect(commit: DisconnectCommit) -> Self {
145        let mut batch = Self {
146            transport: commit.transport_plan,
147            ..Self::default()
148        };
149        batch.source_policy.request();
150        batch.output.lifecycle = commit.effects;
151        batch.transport.extend_teardown(commit.session_teardowns);
152        batch
153    }
154
155    pub(in crate::engine::room) fn from_presence(commit: PresenceCommit) -> Self {
156        let mut batch = Self::default();
157        batch.output.user_info = Some(commit.fanout);
158        batch.source_policy.request();
159        batch
160    }
161
162    pub(in crate::engine::room) fn from_publish(commit: PublishCommit) -> PublicationEffects {
163        let mut batch = Self::default();
164        batch
165            .transport
166            .push_receiver_work(commit.receiver_route_work, ConsumerSetupOrigin::Publish);
167        batch.output.user_info_before_policy = commit.presence.map(|presence| presence.fanout);
168        batch.source_policy.request();
169        PublicationEffects {
170            batch,
171            policy_before_transport: false,
172        }
173    }
174
175    pub(in crate::engine::room) fn from_publication_activity(
176        commit: ProducerActivityCommit,
177    ) -> PublicationEffects {
178        let ProducerActivityCommit {
179            source,
180            stream_id,
181            update,
182            remote_activity_effects,
183            track_snapshots,
184            presence,
185        } = commit;
186        let policy_before_transport = update.activity().is_active();
187        let mut batch = Self::default();
188        batch
189            .transport
190            .extend_remote_source_activity(remote_activity_effects);
191        batch.transport.push_producer(source, stream_id, update);
192        batch.output.track_snapshots = track_snapshots;
193        batch.output.user_info_before_policy = presence.map(|presence| presence.fanout);
194        batch.source_policy.request();
195        PublicationEffects {
196            batch,
197            policy_before_transport,
198        }
199    }
200
201    pub(in crate::engine::room) fn from_receiver_intent(commit: ReceiverRouteCommit) -> Self {
202        let mut batch = Self::from_receiver_route(commit.work, ConsumerSetupOrigin::Subscribe);
203        batch.source_policy.request();
204        batch
205    }
206
207    pub(in crate::engine::room) fn from_consumer_readiness(commit: ReceiverRouteCommit) -> Self {
208        let ReceiverRouteCommit {
209            work,
210            track_snapshots,
211        } = commit;
212        let mut batch = Self::from_receiver_route(work, ConsumerSetupOrigin::Readiness);
213        batch.output.track_snapshots = track_snapshots;
214        batch.source_policy.request();
215        batch
216    }
217
218    fn from_receiver_route(work: ReceiverRouteWork, origin: ConsumerSetupOrigin) -> Self {
219        let mut batch = Self::default();
220        batch.transport.push_receiver_work(work, origin);
221        batch
222    }
223
224    /// preserves the room-wide side-effect order across transport and policy work
225    pub async fn execute(self, room: &Room, context: RoomEffectContext<'_>) {
226        let mut output = self.output;
227        self.transport
228            .execute(room, context.route_transport())
229            .await;
230        output.emit_before_policy();
231        self.source_policy
232            .execute(room, context.media_transport(), None)
233            .await;
234        output.emit_after_policy();
235    }
236}
237
238impl PublicationEffects {
239    pub async fn execute(self, guard: &SourcePolicyGuard<'_>, context: RoomEffectContext<'_>) {
240        let room = guard.room();
241        let mut output = self.batch.output;
242        let mut source_policy = self.batch.source_policy;
243        if self.policy_before_transport {
244            output.emit_user_info_before_policy();
245            source_policy
246                .execute_guarded(guard, context.media_transport(), None)
247                .await;
248            source_policy = SourcePolicyTurn::default();
249        }
250        self.batch
251            .transport
252            .execute(room, context.route_transport())
253            .await;
254        output.emit_before_policy();
255        source_policy
256            .execute_guarded(guard, context.media_transport(), None)
257            .await;
258        output.emit_after_policy();
259    }
260}
o_sfu_core/engine/room/effects/batch.rs

o_sfu_core/engine/room/effects/
batch.rs
