From 213407cf5e8f68e4bcd13cfcee5998c828fbb96a Mon Sep 17 00:00:00 2001 From: shijing xian Date: Sun, 16 Aug 2026 23:35:52 +0800 Subject: [PATCH] fix(rtc_engine): escalate a PeerConnection that stays disconnected Only `PeerConnectionState::Failed` drove recovery. libwebrtc does not reach `Failed` until ICE consent expires, tens of seconds after a transport actually stops working; `Disconnected`, which it reports within seconds, was ignored entirely. Nothing acted in between, so a session whose media plane had died stayed "connected" and silently deaf for the whole consent window before anything began recovering. This is most visible in steady state -- a NAT rebind or a network handover with no signalling failure at all -- where no resume is in flight to notice. `Disconnected` is not on its own a reason to reconnect: ordinary network disturbance produces brief disconnects that recover unaided, and tearing down a session for one would be worse than the disturbance. So a transport entering `Disconnected` now starts a grace period and is judged on its state when that elapses, rather than on the transition that started it. A connection that recovered needs no cancellation bookkeeping -- it simply reads as connected and the countdown lapses silently. Repeated transitions collapse onto the running countdown instead of each spawning their own. Escalation is left to the engine, which already interprets 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. For reference, client-sdk-js does not act on `disconnected` either -- its `PCTransportManager.updateState` has no branch for it, so the state silently retains its previous value and `verifyTransport` keeps accepting it. Its actual net for this case is the server-driven `ConnectionQuality::LOST` signal (`scheduleLostQualityReconnect`), which catches a publisher the server has stopped receiving but not a subscriber that has stopped receiving the server. Acting locally on `disconnected` covers both directions. Co-Authored-By: Claude Opus 5 (1M context) --- .../escalate_persistent_pc_disconnect.md | 12 ++ livekit/specs/signalling-reconnection.allium | 30 ++++ livekit/src/rtc_engine/peer_transport.rs | 58 ++++++- livekit/src/rtc_engine/rtc_session.rs | 141 +++++++++++++++++- 4 files changed, 238 insertions(+), 3 deletions(-) create mode 100644 .changeset/escalate_persistent_pc_disconnect.md 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));