fix(rtc_engine): require evidence of PeerConnection recovery on resume

.changeset/resume_requires_evidence_of_pc_recovery.md

@@ -0,0 +1,12 @@
+---
+livekit: patch
+---
+
+Fix resume reporting success for a PeerConnection that had not recovered.
+
+A resume decided recovery from `PeerConnectionState`, which keeps reading `Connected` for tens
+of seconds after the far end goes away. A resume could therefore emit `Resumed` — and so
+`RoomEvent::Reconnected` with `ConnectionState::Connected` — for a session whose subscriber
+transport was dead, leaving applications with no signal that they had stopped receiving media.
+A resume now requires each transport to have entered `Connected` since the resume began, or to
+have held it throughout, rather than trusting the state it currently reports.

livekit/specs/signalling-reconnection.allium

@@ -224,9 +224,13 @@ config {
     reconnect_base_delay_ms: Integer = 300
     reconnect_backoff_multiplier: Integer = 2
     reconnect_max_delay_ms: Integer = 7000
-    -- Settle delay before re-checking PC state on the resume path
-    -- (PC_RECONNECT_SETTLE_DELAY). Resume-only; full reconnect builds new PCs.
-    pc_reconnect_settle_delay: Duration = 1.seconds
+    -- How long a resume waits before accepting "this transport never left
+    -- connected" as evidence it is still good (PC_RECONNECT_SETTLE_DELAY).
+    -- Bounds only that ambiguous case: a transport that re-enters connected
+    -- during the resume is accepted immediately. Must exceed ICE's receiving
+    -- 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
 }
 
 ------------------------------------------------------------
@@ -538,8 +542,14 @@ rule ResumeAwaitsPeerConnections {
     requires: engine.status = reconnecting
     ensures: AwaitPeerConnectionsRequested(engine, settle: config.pc_reconnect_settle_delay)
     @guidance
-        -- Step 4: wait for the PeerConnections to reconnect, then apply the
-        -- settle delay before trusting their state.
+        -- Step 4: wait until each required PeerConnection has demonstrably
+        -- reconnected, measured against the transition counts sampled before
+        -- the resume began (step 0). A transport qualifies when it is connected
+        -- and either entered connected since then, or never left it for the
+        -- whole settle window. The connection state alone is not sufficient:
+        -- ICE keeps reporting connected for tens of seconds after the far end
+        -- disappears, so trusting it let a resume report success for a
+        -- transport that was already dead.
 }
 
 rule ResumeRechecksLink {

livekit/src/rtc_engine/mod.rs

@@ -55,17 +55,19 @@ pub(crate) type EngineEmitter = mpsc::UnboundedSender<EngineEvent>;
 pub(crate) type EngineEvents = mpsc::UnboundedReceiver<EngineEvent>;
 pub(crate) type EngineResult<T> = Result<T, EngineError>;
 
-/// Settling delay before checking PeerConnection state on the resume path.
+/// How long a resume waits before accepting "this transport never left `Connected`" as
+/// evidence that it is still good.
 ///
-/// Lets a freshly issued ICE-restart offer/answer round-trip take effect when the
-/// underlying PC was still in `Connected` at the moment we started the reconnect
-/// (e.g. signal-only failure). Without this, the resume can return success
-/// immediately and the next failure detector then trips the engine into a real
-/// disconnect.
+/// Only the ambiguous case waits: a transport that re-entered `Connected` during the resume
+/// has demonstrably reconnected and is accepted immediately. But a transport whose far end
+/// has silently gone away also still reports `Connected`, because ICE holds that state until
+/// its receiving timeout — so this has to outlast that timeout, or a dead transport is
+/// accepted as healthy.
 ///
-/// Only applied to the resume path. Full reconnect builds brand-new PCs which
-/// don't suffer from the "looks-Connected-but-isn't" race.
-pub const PC_RECONNECT_SETTLE_DELAY: Duration = Duration::from_secs(1);
+/// The cost is resume latency for a signal-only failure, where the media plane was fine and
+/// there is no reconnection to short-circuit on. Resume-only; a full reconnect builds new
+/// PeerConnections and has no stale state to misread.
+pub const PC_RECONNECT_SETTLE_DELAY: Duration = Duration::from_secs(3);
 
 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
 pub enum SimulateScenario {
@@ -1098,9 +1100,10 @@ impl EngineInner {
     /// each non-trivial seam is its own method so the sequence — and the reason
     /// for the ordering — is explicit rather than implied by statement order.
     /// Mirrors the resume chain in `livekit/specs/signalling-reconnection.allium`:
+    ///   0. sample the PC transition counts, before anything can perturb them;
     ///   1. reopen the signalling link (queue gate stays on until step 4);
     ///   2. SyncState before the publisher re-offer;
-    ///   3. re-offer the publisher, then await PC reconnection + settle;
+    ///   3. re-offer the publisher, then await *demonstrated* PC recovery;
     ///   4. re-check link liveness, then drain the queue.
     async fn try_resume_connection(self: &Arc<Self>) -> EngineResult<()> {
         // Test-only: force the configured number of resume attempts to fail so tests
@@ -1129,6 +1132,11 @@ impl EngineInner {
 
         let session = self.running_handle.read().session.clone();
 
+        // 0. Sample the transports' connection-state transition counts BEFORE anything can
+        //    perturb them, so step 3 can tell a transport that reconnected from one still
+        //    reporting a `Connected` that predates the failure.
+        let pc_snapshot = session.pc_generation_snapshot();
+
         // 1. Reopen the signalling link. The SignalClient stays gated
         //    (`reconnecting=true`) so queueable mutations buffer until step 4.
         let reconnect_response = session.restart().await?;
@@ -1137,10 +1145,10 @@ impl EngineInner {
         //    SyncState, which must precede the publisher re-offer.
         self.resume_sync_state(reconnect_response).await;
 
-        // 3. Re-offer the publisher (strictly AFTER SyncState) and wait for the
-        //    PeerConnections to reconnect, applying the settle delay.
+        // 3. Re-offer the publisher (strictly AFTER SyncState), then wait for the transports
+        //    to have demonstrably reconnected rather than merely to report `Connected`.
         session.restart_publisher().await?;
-        session.wait_pc_reconnected(PC_RECONNECT_SETTLE_DELAY).await?;
+        session.wait_pc_reconnected(pc_snapshot, PC_RECONNECT_SETTLE_DELAY).await?;
 
         // 4. Re-check link liveness and drain the queued mutations.
         self.resume_finalize(&session).await

livekit/src/rtc_engine/peer_transport.rs

@@ -14,7 +14,10 @@
 
 use std::{
     fmt::{Debug, Formatter},
-    sync::Arc,
+    sync::{
+        atomic::{AtomicU32, Ordering},
+        Arc,
+    },
 };
 
 use libwebrtc::prelude::*;
@@ -41,6 +44,12 @@ pub struct PeerTransport {
     peer_connection: PeerConnection,
     on_offer_handler: Mutex<Option<OnOfferCreated>>,
     inner: Arc<AsyncMutex<TransportInner>>,
+
+    /// Counts entries into `Connected`; see [`Self::note_connection_state`].
+    connected_generation: AtomicU32,
+
+    /// Counts exits from `Connected`; see [`Self::note_connection_state`].
+    disconnect_generation: AtomicU32,
 }
 
 impl Debug for PeerTransport {
@@ -67,13 +76,40 @@ impl PeerTransport {
                 max_send_bitrate_bps: None,
                 pending_initial_offer: None,
             })),
+            connected_generation: AtomicU32::new(0),
+            disconnect_generation: AtomicU32::new(0),
         }
     }
 
     pub fn is_connected(&self) -> bool {
         self.peer_connection.connection_state() == PeerConnectionState::Connected
     }
 
+    pub fn connected_generation(&self) -> u32 {
+        self.connected_generation.load(Ordering::Acquire)
+    }
+
+    pub fn disconnect_generation(&self) -> u32 {
+        self.disconnect_generation.load(Ordering::Acquire)
+    }
+
+    /// Record a connection-state transition, so a later observer can tell what the transport
+    /// has *done* rather than only what it currently reports.
+    ///
+    /// Both counters exist because `PeerConnectionState` is a level, and a level cannot
+    /// distinguish a transport that recovered from one whose far end vanished: ICE keeps
+    /// reporting `Connected` until its receiving timeout, and only reaches `Failed` after
+    /// consent expiry tens of seconds later. Comparing these against a snapshot taken before
+    /// a resume gives that history. Called for every `RtcEvent::ConnectionChange`, so a
+    /// transport that drops and returns between two polls is still visible as having dropped.
+    pub fn note_connection_state(&self, state: PeerConnectionState) {
+        if state == PeerConnectionState::Connected {
+            self.connected_generation.fetch_add(1, Ordering::AcqRel);
+        } else {
+            self.disconnect_generation.fetch_add(1, Ordering::AcqRel);
+        }
+    }
+
     pub fn peer_connection(&self) -> PeerConnection {
         self.peer_connection.clone()
     }
@@ -652,6 +688,48 @@ mod tests {
         assert_eq!(transport.peer_connection().signaling_state(), SignalingState::HaveLocalOffer);
     }
 
+    /// The two counters are the whole basis on which a resume decides a transport recovered,
+    /// so each transition must land on exactly one of them, and a transport that drops and
+    /// returns must leave both marks rather than looking untouched.
+    #[test]
+    fn connection_state_transitions_are_counted_separately() {
+        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,
+        );
+        let counters = || (transport.connected_generation(), transport.disconnect_generation());
+
+        assert_eq!(counters(), (0, 0));
+
+        transport.note_connection_state(PeerConnectionState::Connecting);
+        assert_eq!(counters(), (0, 1));
+
+        transport.note_connection_state(PeerConnectionState::Connected);
+        assert_eq!(counters(), (1, 1));
+
+        // Dropping and returning must move both counters: a resume needs to see that it broke
+        // *and* that it came back, and a poll sampling only the current state sees neither.
+        transport.note_connection_state(PeerConnectionState::Disconnected);
+        transport.note_connection_state(PeerConnectionState::Connected);
+        assert_eq!(counters(), (2, 2));
+
+        transport.note_connection_state(PeerConnectionState::Failed);
+        assert_eq!(counters(), (2, 3));
+    }
+
     #[test]
     fn no_video_codec_is_noop() {
         // Audio-only SDP should not be modified