diff --git a/.changeset/escalate_persistent_pc_disconnect.md b/.changeset/escalate_persistent_pc_disconnect.md new file mode 100644 index 000000000..9c05da50e --- /dev/null +++ b/.changeset/escalate_persistent_pc_disconnect.md @@ -0,0 +1,12 @@ +--- +livekit: patch +--- + +Escalate a PeerConnection that stays disconnected, instead of waiting for `Failed`. + +Only `PeerConnectionState::Failed` triggered recovery, and libwebrtc does not reach it until +ICE consent expires — tens of seconds after the transport actually stopped working. A session +whose media plane died stayed "connected" and silently deaf for that entire window before +anything began recovering. A transport entering `Disconnected` now starts a grace period, and +is treated as failed if it has not recovered when that elapses; brief disconnects during +ordinary network disturbance still resolve on their own and are ignored. diff --git a/livekit/specs/signalling-reconnection.allium b/livekit/specs/signalling-reconnection.allium index 0b5d88a86..a82656610 100644 --- a/livekit/specs/signalling-reconnection.allium +++ b/livekit/specs/signalling-reconnection.allium @@ -231,6 +231,10 @@ config { -- timeout, or a transport whose far end is gone still reads connected when -- checked. Resume-only; full reconnect builds new PCs. pc_reconnect_settle_delay: Duration = 3.seconds + -- How long a PeerConnection may sit in `disconnected` before it is treated + -- as failed (PC_DISCONNECTED_GRACE). Long enough to absorb a transient + -- disturbance, far short of ICE consent expiry. + pc_disconnected_grace: Duration = 5.seconds } ------------------------------------------------------------ @@ -413,6 +417,31 @@ rule ReconnectOnPeerConnectionFailure { retry_now: false, cause: peer_connection_failed ) + @guidance + -- `failed` is not the earliest signal that a transport has died: ICE + -- reports `disconnected` within seconds, but does not escalate to + -- `failed` until consent expires, tens of seconds later. Reacting only + -- to `failed` left a session "connected" and deaf for that whole + -- window. See PeerConnectionDisconnectedBeyondGrace. +} + +rule PeerConnectionDisconnectedBeyondGrace { + when: PeerConnectionDisconnected(engine, for: config.pc_disconnected_grace) + requires: engine.reconnect_permission = permitted + ensures: ReconnectNeeded( + engine, + full_reconnect: false, + retry_now: false, + cause: peer_connection_failed + ) + @guidance + -- `disconnected` alone is NOT actionable: ordinary network disturbance + -- produces brief disconnects that recover on their own, and tearing + -- down a session for one would be worse than the disturbance. So the + -- transport is given a grace period and judged on its state when that + -- elapses, not on the transition that started it -- a connection that + -- recovered needs no cancellation, it simply reads as connected. + -- Concurrent transitions collapse onto the running countdown. } rule ServerRequestsReconnect { @@ -746,6 +775,7 @@ surface MediaRecovery { provides: EngineSessionConnected(media) PeerConnectionFailed(media) + PeerConnectionDisconnected(media) PublisherRepublished(media) PeerConnectionsReconnected(media) RestartAttemptSucceeded(media) diff --git a/livekit/src/rtc_engine/peer_transport.rs b/livekit/src/rtc_engine/peer_transport.rs index 634190e82..c13d31d6a 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -15,7 +15,7 @@ use std::{ fmt::{Debug, Formatter}, sync::{ - atomic::{AtomicU32, Ordering}, + atomic::{AtomicBool, AtomicU32, Ordering}, Arc, }, }; @@ -50,6 +50,10 @@ pub struct PeerTransport { /// Counts exits from `Connected`; see [`Self::note_connection_state`]. disconnect_generation: AtomicU32, + + /// Whether a `Disconnected` countdown is currently running for this transport, so that a + /// flapping connection collapses onto one timer instead of spawning one per transition. + disconnect_grace_armed: AtomicBool, } impl Debug for PeerTransport { @@ -78,6 +82,7 @@ impl PeerTransport { })), connected_generation: AtomicU32::new(0), disconnect_generation: AtomicU32::new(0), + disconnect_grace_armed: AtomicBool::new(false), } } @@ -93,6 +98,20 @@ impl PeerTransport { self.disconnect_generation.load(Ordering::Acquire) } + /// Claim the right to start a `Disconnected` countdown for this transport. + /// + /// Returns `true` for the caller that armed it and `false` if a countdown is already + /// running, so repeated `Disconnected` transitions collapse onto a single timer rather + /// than spawning one per event. + pub fn try_arm_disconnect_grace(&self) -> bool { + !self.disconnect_grace_armed.swap(true, Ordering::AcqRel) + } + + /// Release the countdown claim taken by [`Self::try_arm_disconnect_grace`]. + pub fn disarm_disconnect_grace(&self) { + self.disconnect_grace_armed.store(false, Ordering::Release); + } + /// Record a connection-state transition, so a later observer can tell what the transport /// has *done* rather than only what it currently reports. /// @@ -730,6 +749,43 @@ mod tests { assert_eq!(counters(), (2, 3)); } + /// A flapping transport can emit many `Disconnected` transitions in a row. Each must + /// collapse onto the countdown already running rather than spawning its own, or a + /// connection bouncing under load would accumulate timers that all fire and re-report + /// the same failure. + #[test] + fn disconnect_grace_admits_one_countdown_at_a_time() { + use libwebrtc::prelude::*; + use livekit_protocol as proto; + + let factory = PeerConnectionFactory::default(); + let pc = factory + .create_peer_connection(RtcConfiguration { + ice_servers: vec![], + continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, + ice_transport_type: IceTransportsType::All, + }) + .unwrap(); + + let transport = PeerTransport::new( + pc, + proto::SignalTarget::Subscriber, + /* single_pc_mode= */ false, + ); + + assert!(transport.try_arm_disconnect_grace(), "first caller arms the countdown"); + assert!( + !transport.try_arm_disconnect_grace(), + "a second transition must join the running countdown, not start another" + ); + assert!(!transport.try_arm_disconnect_grace()); + + // Once the countdown has run, a later disconnect must be able to arm a fresh one -- + // otherwise a transport that recovers and dies again is never escalated. + transport.disarm_disconnect_grace(); + assert!(transport.try_arm_disconnect_grace()); + } + #[test] fn no_video_codec_is_noop() { // Audio-only SDP should not be modified diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index a47d0a5e1..f2a096f04 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -77,6 +77,21 @@ pub struct PcGenerationSnapshot { } pub const ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// How long a PeerConnection may sit in `Disconnected` before it is treated as a failure. +/// +/// `Disconnected` is not by itself a reason to reconnect: ICE reports it after a couple of +/// seconds without incoming packets, and ordinary network disturbances — a brief handover, a +/// congested link — resolve on their own well inside that. But it is also the *only* early +/// signal that a transport has died: libwebrtc will not escalate to `Failed` until consent +/// expires, tens of seconds later, and until this existed nothing acted in between. A +/// session whose media plane had silently gone away therefore stayed "connected" and deaf +/// for the whole consent window before anything began recovering. +/// +/// So we start a countdown instead of reacting immediately, and act only if the transport is +/// still not connected when it elapses. Long enough to ride out a transient blip, far short +/// of consent expiry. +pub const PC_DISCONNECTED_GRACE: Duration = Duration::from_secs(5); pub const TRACK_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10); pub const LOSSY_DC_LABEL: &str = "_lossy"; pub const RELIABLE_DC_LABEL: &str = "_reliable"; @@ -1585,7 +1600,9 @@ impl SessionInner { Ok(()) } - async fn on_rtc_event(&self, event: RtcEvent) -> EngineResult<()> { + // `&Arc` rather than `&self`: a `Disconnected` transition arms a countdown that + // outlives this call and needs to keep the session alive while it runs. + async fn on_rtc_event(self: &Arc, event: RtcEvent) -> EngineResult<()> { match event { RtcEvent::IceCandidate { ice_candidate, target } => { log::debug!("local ice_candidate {:?} {:?}", ice_candidate, target); @@ -1627,6 +1644,9 @@ impl SessionInner { proto::leave_request::Action::Resume, false, ); + } else if state == PeerConnectionState::Disconnected { + // Not actionable yet — see `PC_DISCONNECTED_GRACE`. + self.arm_disconnected_grace(target); } } RtcEvent::DataChannel { data_channel, target } => { @@ -1990,6 +2010,67 @@ impl SessionInner { Ok(transceiver) } + /// The transport for `target`, or `None` when this session has none (there is no + /// separate subscriber in single-PC mode). + fn transport_for(&self, target: SignalTarget) -> Option<&PeerTransport> { + match target { + SignalTarget::Publisher => Some(&self.publisher_pc), + SignalTarget::Subscriber => self.subscriber_pc.as_ref(), + } + } + + /// Start the [`PC_DISCONNECTED_GRACE`] countdown for a transport that has just entered + /// `Disconnected`, and escalate if it has not recovered by the time it elapses. + /// + /// The decision is deliberately made from the transport's state *when the countdown + /// elapses* rather than from the transition that started it: a connection that recovered + /// on its own — the common case for a transient disturbance — needs no cancellation + /// bookkeeping, it simply reads as connected and the countdown lapses silently. A + /// connection that flapped and is down again at that point is escalated, which is the + /// right call regardless of how many times it bounced in between. + /// + /// Escalation is left to the engine, which already knows how to interpret a transport + /// failure in context: outside a reconnect it starts one, and during a reconnect it + /// sticks a full-reconnect escalation onto the cycle rather than looping on resume. + fn arm_disconnected_grace(self: &Arc, target: SignalTarget) { + let Some(transport) = self.transport_for(target) else { + return; + }; + if !transport.try_arm_disconnect_grace() { + return; // a countdown is already running for this transport + } + + let session = self.clone(); + livekit_runtime::spawn(async move { + livekit_runtime::sleep(PC_DISCONNECTED_GRACE).await; + + let Some(transport) = session.transport_for(target) else { + return; + }; + transport.disarm_disconnect_grace(); + + if session.closed.load(Ordering::Acquire) { + return; + } + + let state = transport.peer_connection().connection_state(); + if should_escalate_after_grace(state) { + log::warn!( + "{:?} pc stayed in {:?} for {:?}; treating as a failed transport", + target, + state, + PC_DISCONNECTED_GRACE, + ); + session.on_session_disconnected( + "pc_state disconnected beyond grace period", + DisconnectReason::UnknownReason, + proto::leave_request::Action::Resume, + false, + ); + } + }); + } + /// Called when the SignalClient or one of the PeerConnection has lost the connection /// The RTCEngine may try a reconnect. fn on_session_disconnected( @@ -2662,6 +2743,25 @@ pub fn handle_remote_dt_packets(dc: &DataChannel, emitter: WeakUnboundedSender bool { + match state { + // Recovered on its own — the transient disturbance this grace period absorbs. + PeerConnectionState::Connected => false, + // Torn down deliberately (close, or a full reconnect replacing this session). + PeerConnectionState::Closed => false, + // Already reported by the `Failed` branch in the connection-change handler; escalating + // here too would report the same failure twice. + PeerConnectionState::Failed => false, + // Still `Disconnected`, or never got back past `Connecting`, after a window a healthy + // blip would have recovered inside. Treat it as dead. + _ => true, + } +} + /// Whether a transport counts as recovered, split out from the transport so the rule can be /// exercised without a PeerConnection. See /// [`SessionInner::wait_pc_reconnected_with_snapshot`] for why the current state is not enough. @@ -2726,7 +2826,44 @@ make_rtc_config!(make_rtc_config_reconnect, proto::ReconnectResponse); #[cfg(test)] mod tests { - use super::{parse_sdp_max_message_size, recovery_decision, DEFAULT_MAX_MESSAGE_SIZE}; + use libwebrtc::prelude::PeerConnectionState; + + use super::{ + parse_sdp_max_message_size, recovery_decision, should_escalate_after_grace, + DEFAULT_MAX_MESSAGE_SIZE, + }; + + /// A transport still down when the grace period elapses is a dead transport, and must be + /// escalated rather than left until libwebrtc gets round to declaring it `Failed` after + /// consent expiry — which is tens of seconds of silently unusable media. + #[test] + fn transport_still_down_after_grace_is_escalated() { + assert!(should_escalate_after_grace(PeerConnectionState::Disconnected)); + // Never made it back past Connecting within the window — equally dead. + assert!(should_escalate_after_grace(PeerConnectionState::Connecting)); + } + + /// The reason this is a countdown and not an immediate reaction: a brief `Disconnected` + /// during ordinary network disturbance is normal and self-healing, and tearing down the + /// session for one would be far worse than the disturbance. + #[test] + fn transport_that_recovered_during_grace_is_left_alone() { + assert!(!should_escalate_after_grace(PeerConnectionState::Connected)); + } + + /// `Failed` is reported the moment it happens, so escalating it here as well would + /// report the same failure twice. + #[test] + fn failed_is_not_double_reported_after_grace() { + assert!(!should_escalate_after_grace(PeerConnectionState::Failed)); + } + + /// A closed transport is a deliberate teardown — session close, or a full reconnect that + /// has replaced this session. Escalating would fight the shutdown it is observing. + #[test] + fn closed_transport_is_not_escalated_after_grace() { + assert!(!should_escalate_after_grace(PeerConnectionState::Closed)); + } /// `(connected, disconnect)` counts as sampled before a resume, for readability below. const SNAPSHOT: Option<(u32, u32)> = Some((7, 3));