fix(rtc_engine): escalate a PeerConnection that stays disconnected

.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.

livekit/specs/signalling-reconnection.allium

@@ -231,6 +231,10 @@ config {
     -- transport whose far end is gone is still claiming 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)

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,
     },
 };
@@ -59,6 +59,10 @@ pub struct PeerTransport {
     /// Distinguishes "still `Connected` because nothing ever broke" from "was `Connected`,
     /// broke, and came back" — a level read of the connection state cannot tell them apart.
     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 {
@@ -87,6 +91,7 @@ impl PeerTransport {
             })),
             negotiation_generation: AtomicU32::new(0),
             disconnect_generation: AtomicU32::new(0),
+            disconnect_grace_armed: AtomicBool::new(false),
         }
     }
 
@@ -104,6 +109,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);
+    }
+
     /// Mark this transport as awaiting a fresh ICE generation from the remote side.
     ///
     /// The subscriber never issues its own offer — the SFU does — so it has no
@@ -812,6 +831,40 @@ mod tests {
         assert_eq!(transport.disconnect_generation(), 2);
     }
 
+    /// 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

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";
@@ -1588,7 +1603,9 @@ impl SessionInner {
         Ok(())
     }
 
-    async fn on_rtc_event(&self, event: RtcEvent) -> EngineResult<()> {
+    // `&Arc<Self>` 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<Self>, event: RtcEvent) -> EngineResult<()> {
         match event {
             RtcEvent::IceCandidate { ice_candidate, target } => {
                 log::debug!("local ice_candidate {:?} {:?}", ice_candidate, target);
@@ -1630,6 +1647,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 } => {
@@ -1993,6 +2013,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<Self>, 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(
@@ -2674,6 +2755,26 @@ pub fn handle_remote_dt_packets(dc: &DataChannel, emitter: WeakUnboundedSender<S
     dc.on_message(on_message.into());
 }
 
+/// Whether a transport that has sat out [`PC_DISCONNECTED_GRACE`] should be escalated,
+/// judged from its state at the moment the countdown elapses.
+///
+/// Split out from the timer so it can be exercised without waiting out a real countdown.
+fn should_escalate_after_grace(state: PeerConnectionState) -> bool {
+    match state {
+        // Recovered on its own — the transient disturbance this grace period exists to
+        // absorb. Nothing to do.
+        PeerConnectionState::Connected => false,
+        // Torn down deliberately (close, or a full reconnect replacing this session).
+        PeerConnectionState::Closed => false,
+        // Already reported the moment it happened, by the `Failed` branch in the connection
+        // change handler; escalating again here would double-report the same failure.
+        PeerConnectionState::Failed => false,
+        // Still `Disconnected` — or never got back past `Connecting` — after a window that a
+        // healthy blip would have recovered inside. Treat it as a dead transport.
+        _ => true,
+    }
+}
+
 /// The resume-path recovery decision, split out from the transport so it can be exercised
 /// without standing up a PeerConnection. See
 /// [`SessionInner::wait_pc_reconnected_with_snapshot`] for the reasoning behind each branch.
@@ -2739,7 +2840,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));
+    }
 
     /// The generations sampled at the start of a resume, for readability below.
     const SNAPSHOT: Option<(u32, u32)> = Some((7, 3));