o_sfu_protocol/core/
connection_lifecycle.rs

1//! socket lifecycle transitions for [`ProtocolCore`]
2//!
3//! [`connect`] starts a user attempt and clears replayable intent
4//! [`disconnect`] ends that attempt and suppresses recovery
5//! [`on_ws_close`] maps terminal codes to [`ConnectionState::Closed`]
6//! while transient closes preserve the connect context for [`handle_recovery_timer`]
7//!
8//! welcome messages enter through [`ProtocolCore::on_ws_message`]
9//! transport readiness enters through [`ProtocolCore::on_transport_ready`]
10//! each transition returns ordered [`Command`] values for the host
11
12use super::{
13    Command, Commands, ConnectContext, ConnectionState, INITIAL_RECOVERY_DELAY_MS, ProtocolCore,
14    RECOVERY_TIMER_ID, empty_features, next_recovery_delay,
15};
16use crate::{shared::RecordingState, signaling::WebSocketCloseCode};
17
18/// host-visible reason attached to a terminal lifecycle state change
19///
20/// values are derived from terminal websocket close codes and rendered as
21/// compatibility labels by production command translation
22#[derive(Clone, Copy)]
23enum LifecycleCloseCause {
24    /// authentication was rejected by the server
25    AuthFailed,
26    /// the server removed the user from the room
27    Kicked,
28    /// the room refused the connection because capacity was exhausted
29    RoomFull,
30}
31
32/// maps terminal websocket close codes to host-visible lifecycle causes
33fn terminal_close_cause(close_code: WebSocketCloseCode) -> Option<LifecycleCloseCause> {
34    match close_code {
35        WebSocketCloseCode::AuthFailed => Some(LifecycleCloseCause::AuthFailed),
36        WebSocketCloseCode::Kicked => Some(LifecycleCloseCause::Kicked),
37        WebSocketCloseCode::RoomFull => Some(LifecycleCloseCause::RoomFull),
38        _ => None,
39    }
40}
41
42/// returns the compatibility label exposed through `EmitStateChange`
43fn lifecycle_close_cause_label(cause: LifecycleCloseCause) -> &'static str {
44    match cause {
45        LifecycleCloseCause::AuthFailed => "auth_failed",
46        LifecycleCloseCause::Kicked => "kicked",
47        LifecycleCloseCause::RoomFull => "full",
48    }
49}
50
51fn reset_public_state(commands: &mut Commands) {
52    commands.extend([
53        Command::SetAvailableFeatures {
54            features: empty_features(),
55        },
56        Command::SetRecordingState {
57            state: RecordingState::default(),
58        },
59    ]);
60}
61
62fn state_change(state: ConnectionState, cause: Option<LifecycleCloseCause>) -> Command {
63    Command::EmitStateChange {
64        state,
65        cause: cause.map(lifecycle_close_cause_label).map(str::to_owned),
66    }
67}
68
69/// starts a fresh connection attempt from an inactive or recovering state
70///
71/// this is the only lifecycle entry point that wipes both
72/// runtime state and sticky replay state before reconnecting
73/// a brand-new [`connect`] means "start over with this endpoint and auth context",
74/// not "resume whatever the previous user was trying to do"
75///
76/// calls from live admission states are ignored so the host cannot accidentally
77/// stack overlapping connection attempts on top of an already-live user
78/// a call from [`ConnectionState::Recovering`] also cancels the stale recovery timer before the
79/// new socket attempt starts
80///
81/// ```text
82/// Disconnected --connect(url, jwt, room)--> Connecting
83/// Closed       --connect(url, jwt, room)--> Connecting
84/// Recovering  --connect(url, jwt, room)--> Connecting
85/// ```
86pub(super) fn connect(
87    core: &mut ProtocolCore,
88    url: String,
89    jwt: String,
90    room: Option<String>,
91) -> Commands {
92    let mut commands = match core.state() {
93        ConnectionState::Disconnected | ConnectionState::Closed => Vec::new(),
94        ConnectionState::Recovering => vec![Command::CancelTimer {
95            id: RECOVERY_TIMER_ID,
96        }],
97        _ => return Vec::new(),
98    };
99    let connect_url = url.clone();
100    core.connect_context = Some(ConnectContext { url, jwt, room });
101    core.recovery_delay_ms = INITIAL_RECOVERY_DELAY_MS;
102    core.phase
103        .apply_lifecycle_state(ConnectionState::Connecting);
104    core.clear_runtime_state();
105    core.clear_sticky_state();
106    reset_public_state(&mut commands);
107    commands.push(state_change(core.state(), None));
108    commands.push(Command::Connect { url: connect_url });
109    commands
110}
111
112/// ends the current user attempt on purpose
113///
114/// unlike [`on_ws_close`], this is not a recovery path
115/// it clears the saved connect context, runtime state and sticky replay state,
116/// then closes the websocket and peer connection
117/// any later recovery-timer delivery becomes a no-op because the caller
118/// explicitly asked to stop
119pub(super) fn disconnect(core: &mut ProtocolCore) -> Commands {
120    if matches!(
121        core.state(),
122        ConnectionState::Disconnected | ConnectionState::Closed
123    ) {
124        return Vec::new();
125    }
126    core.phase
127        .apply_lifecycle_state(ConnectionState::Disconnected);
128    core.connect_context = None;
129    core.recovery_delay_ms = INITIAL_RECOVERY_DELAY_MS;
130    let mut commands = vec![Command::CancelTimer {
131        id: RECOVERY_TIMER_ID,
132    }];
133    commands.extend(core.teardown_runtime_state());
134    core.clear_sticky_state();
135    commands.push(Command::CloseWebSocket {
136        code: u16::from(WebSocketCloseCode::Clean),
137    });
138    commands.push(Command::ClosePeerConnection);
139    reset_public_state(&mut commands);
140    commands.push(state_change(core.state(), None));
141    commands
142}
143
144/// handles websocket closure after a user was already in flight
145///
146/// there are three different cases here and mixing them up is the main way to
147/// break reconnect behavior:
148///
149/// - terminal close codes move to [`ConnectionState::Closed`], clear the saved connect context,
150///   and suppress recovery
151/// - non-terminal closes with saved connect context move to [`ConnectionState::Recovering`] and
152///   schedule the recovery timer
153/// - non-terminal closes without saved connect context fall back to
154///   [`ConnectionState::Disconnected`], because there is nothing safe to reconnect to
155///
156/// example:
157///
158/// ```text
159/// Connected --on_ws_close(AuthFailed)--> Closed
160/// Connected --on_ws_close(1011)--> Recovering
161/// ```
162pub(super) fn on_ws_close(core: &mut ProtocolCore, close_code: u16) -> Commands {
163    if matches!(
164        core.state(),
165        ConnectionState::Disconnected | ConnectionState::Closed
166    ) {
167        return Vec::new();
168    }
169
170    if let Some(
171        terminal_code @ (WebSocketCloseCode::ProtocolError
172        | WebSocketCloseCode::AuthFailed
173        | WebSocketCloseCode::Kicked
174        | WebSocketCloseCode::RoomFull),
175    ) = WebSocketCloseCode::from_u16(close_code)
176    {
177        core.phase.apply_lifecycle_state(ConnectionState::Closed);
178        core.connect_context = None;
179        core.recovery_delay_ms = INITIAL_RECOVERY_DELAY_MS;
180        let mut commands = core.teardown_runtime_state();
181        commands.push(Command::CancelTimer {
182            id: RECOVERY_TIMER_ID,
183        });
184        commands.push(Command::ClosePeerConnection);
185        reset_public_state(&mut commands);
186        commands.push(state_change(
187            core.state(),
188            terminal_close_cause(terminal_code),
189        ));
190        return commands;
191    }
192
193    if core.connect_context.is_none() {
194        core.phase
195            .apply_lifecycle_state(ConnectionState::Disconnected);
196        let mut commands = core.teardown_runtime_state();
197        reset_public_state(&mut commands);
198        commands.push(state_change(core.state(), None));
199        return commands;
200    }
201
202    let scheduled_delay_ms = core.recovery_delay_ms;
203    core.recovery_delay_ms = next_recovery_delay(scheduled_delay_ms);
204    core.phase
205        .apply_lifecycle_state(ConnectionState::Recovering);
206    let mut commands = core.teardown_runtime_state();
207    commands.push(Command::ClosePeerConnection);
208    commands.push(state_change(core.state(), None));
209    commands.push(Command::ScheduleTimer {
210        id: RECOVERY_TIMER_ID,
211        ms: scheduled_delay_ms,
212    });
213    commands
214}
215
216/// retries the saved websocket connection after a recovery delay
217///
218/// this is narrow
219/// only [`ConnectionState::Recovering`] may consume the recovery timer
220/// a stale timer firing after a successful reconnect or explicit
221/// disconnect must do nothing, otherwise old scheduled work can restart an
222/// inactive attempt
223///
224/// example:
225///
226/// ```text
227/// Connected --on_ws_close(1011)--> Recovering
228/// Recovering --handle_recovery_timer()--> Connecting
229/// Connected --handle_recovery_timer()--> no-op
230/// ```
231pub(super) fn handle_recovery_timer(core: &mut ProtocolCore) -> Commands {
232    if core.state() != ConnectionState::Recovering {
233        return Vec::new();
234    }
235    let Some(connect_context) = core.connect_context.as_ref() else {
236        return Vec::new();
237    };
238    let connect_url = connect_context.url.clone();
239    core.phase
240        .apply_lifecycle_state(ConnectionState::Connecting);
241    let mut commands = vec![state_change(core.state(), None)];
242    commands.push(Command::Connect { url: connect_url });
243    commands
244}
o_sfu_protocol/core/connection_lifecycle.rs

o_sfu_protocol/core/
connection_lifecycle.rs
