diff --git a/.changeset/resume_requires_evidence_of_pc_recovery.md b/.changeset/resume_requires_evidence_of_pc_recovery.md new file mode 100644 index 000000000..b366c4dda --- /dev/null +++ b/.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. diff --git a/livekit/specs/signalling-reconnection.allium b/livekit/specs/signalling-reconnection.allium index daa7977ee..0b5d88a86 100644 --- a/livekit/specs/signalling-reconnection.allium +++ b/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 { diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index 2895cfab5..cff8a28aa 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -55,17 +55,19 @@ pub(crate) type EngineEmitter = mpsc::UnboundedSender; pub(crate) type EngineEvents = mpsc::UnboundedReceiver; pub(crate) type EngineResult = Result; -/// 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) -> 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 diff --git a/livekit/src/rtc_engine/peer_transport.rs b/livekit/src/rtc_engine/peer_transport.rs index 7eccd2efe..634190e82 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/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>, inner: Arc>, + + /// 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,6 +76,8 @@ impl PeerTransport { max_send_bitrate_bps: None, pending_initial_offer: None, })), + connected_generation: AtomicU32::new(0), + disconnect_generation: AtomicU32::new(0), } } @@ -74,6 +85,31 @@ impl PeerTransport { 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 diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index bb6c45f38..a47d0a5e1 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -20,7 +20,7 @@ use std::{ atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}, Arc, }, - time::Duration, + time::{Duration, Instant}, }; use bytes::Bytes; @@ -63,6 +63,19 @@ use crate::{ DataPacketKind, }; +/// Connection-state transition counts per transport, sampled before a resume begins. +/// +/// A resume compares against these to tell a transport that actually reconnected from one +/// still reporting a `Connected` that predates the failure. See +/// [`SessionInner::wait_pc_reconnected_with_snapshot`]. +#[derive(Debug, Clone, Copy)] +pub struct PcGenerationSnapshot { + publisher_connected: u32, + publisher_disconnect: u32, + subscriber_connected: u32, + subscriber_disconnect: u32, +} + pub const ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); pub const TRACK_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10); pub const LOSSY_DC_LABEL: &str = "_lossy"; @@ -357,7 +370,9 @@ struct SessionInner { fast_publish: AtomicBool, publisher_pc: PeerTransport, - /// In single peer connection mode, this is None and publisher_pc handles both send/receive + /// `Some` exactly when [`Self::single_pc_mode`] is false, where publisher_pc handles both + /// send and receive. `wait_pc_connection_inner` relies on that equivalence, reading absence + /// as "no second transport to wait for". subscriber_pc: Option, /// Whether single peer connection mode is active single_pc_mode: bool, @@ -849,14 +864,22 @@ impl RtcSession { self.inner.wait_pc_connection().await } - /// Wait for PCs to be connected on the resume path. - /// - /// Sleeps `settle_delay` before polling, giving the just-issued ICE - /// restart offer/answer round-trip a chance to take effect when the - /// failure was signal-only (PCs may still report `Connected` immediately - /// after a WS hiccup, even though the new ufrag/pwd hasn't propagated yet). - pub async fn wait_pc_reconnected(&self, settle_delay: Duration) -> EngineResult<()> { - self.inner.wait_pc_connection_with_delay(settle_delay).await + /// Sample each transport's connection-state transition counts. Must be taken *before* the + /// resume touches the signalling link, so anything the resume causes shows up as a change. + pub fn pc_generation_snapshot(&self) -> PcGenerationSnapshot { + self.inner.pc_generation_snapshot() + } + + /// Wait for the PeerConnections to have reconnected, judged against `snapshot` rather + /// than against the current `PeerConnectionState`, which stays `Connected` for tens of + /// seconds after a far end goes away. `settle_delay` bounds only the ambiguous + /// never-dropped case. See [`SessionInner::wait_pc_reconnected_with_snapshot`]. + pub async fn wait_pc_reconnected( + &self, + snapshot: PcGenerationSnapshot, + settle_delay: Duration, + ) -> EngineResult<()> { + self.inner.wait_pc_reconnected_with_snapshot(snapshot, settle_delay).await } /// Ensure the publisher peer connection is connected and the data channel is open. @@ -1582,6 +1605,18 @@ impl SessionInner { RtcEvent::ConnectionChange { state, target } => { log::debug!("connection change, {:?} {:?}", state, target); + // Durably record leaving `Connected` before waking waiters: a resume that + // polls after this point must be able to see that the transport broke, even + // though it may since have returned to `Connected`. + if target == SignalTarget::Publisher { + self.publisher_pc.note_connection_state(state); + } + if target == SignalTarget::Subscriber { + if let Some(ref sub_pc) = self.subscriber_pc { + sub_pc.note_connection_state(state); + } + } + self.pc_state_notify.notify_waiters(); if state == PeerConnectionState::Failed { @@ -2214,6 +2249,21 @@ impl SessionInner { Ok(reconnect_response) } + fn pc_generation_snapshot(&self) -> PcGenerationSnapshot { + let (subscriber_connected, subscriber_disconnect) = self + .subscriber_pc + .as_ref() + .map(|pc| (pc.connected_generation(), pc.disconnect_generation())) + .unwrap_or((0, 0)); + + PcGenerationSnapshot { + publisher_connected: self.publisher_pc.connected_generation(), + publisher_disconnect: self.publisher_pc.disconnect_generation(), + subscriber_connected, + subscriber_disconnect, + } + } + async fn restart_publisher(&self) -> EngineResult<()> { // In single-PC mode the publisher is the only transport, so always restart its ICE // even if the user hasn't explicitly published a track yet. Otherwise only restart @@ -2226,18 +2276,48 @@ impl SessionInner { Ok(()) } - /// Timeout after ['MAX_ICE_CONNECT_TIMEOUT'] + /// Wait for the transports to connect, timing out after [`ICE_CONNECT_TIMEOUT`]. + /// + /// For the initial connect, where transports start from `New`: reaching `Connected` is + /// itself the proof, since there is no earlier connection to confuse it with. async fn wait_pc_connection(&self) -> EngineResult<()> { - self.wait_pc_connection_with_delay(Duration::ZERO).await + self.wait_pc_connection_inner(None, Duration::ZERO).await } - /// Like [`Self::wait_pc_connection`] but sleeps `settle_delay` before polling. - async fn wait_pc_connection_with_delay(&self, settle_delay: Duration) -> EngineResult<()> { - let wait_connected = async move { - if !settle_delay.is_zero() { - livekit_runtime::sleep(settle_delay).await; - } + /// Wait for the transports to have *reconnected*, which a resume cannot establish by + /// reading `PeerConnectionState`. + /// + /// That state is a level, and the level is identical for a transport that reconnected and + /// one whose far end has vanished: ICE holds `Connected` until its receiving timeout, and + /// only reaches `Failed` after consent expiry tens of seconds later. Trusting it let a + /// resume declare success while the subscriber was dead, emitting `Resumed` — and so + /// `RoomEvent::Reconnected` — for a session that received no further media. + /// + /// So compare against the transition counts taken before the resume started. A transport + /// counts as reconnected when it is `Connected` *and* either: + /// - it entered `Connected` since the snapshot, meaning ICE and DTLS completed on this + /// attempt; or + /// - it never left `Connected` for the whole settle window, so the pre-existing + /// connection is still good (the signal-only failure, where media never broke). + /// + /// A transport that dropped and has not reconnected since is not recovered, whatever it + /// reports now. + async fn wait_pc_reconnected_with_snapshot( + &self, + snapshot: PcGenerationSnapshot, + settle_delay: Duration, + ) -> EngineResult<()> { + self.wait_pc_connection_inner(Some(snapshot), settle_delay).await + } + + async fn wait_pc_connection_inner( + &self, + snapshot: Option, + settle_delay: Duration, + ) -> EngineResult<()> { + let started = Instant::now(); + let wait_connected = async move { loop { let notified = self.pc_state_notify.notified(); @@ -2245,11 +2325,33 @@ impl SessionInner { return Err(EngineError::Connection("closed".into())); } - let publisher_connected = self.publisher_pc.is_connected(); - let subscriber_connected = if self.single_pc_mode || !self.subscriber_primary { - true // No subscriber in single PC mode or if PC is publisher primary - } else { - self.subscriber_pc.as_ref().map(|pc| pc.is_connected()).unwrap_or(true) + // The settle window gates only the "nothing ever broke" branch; a transport + // that reconnected is accepted as soon as it does. + let settled = started.elapsed() >= settle_delay; + + let publisher_ok = Self::transport_recovered( + &self.publisher_pc, + snapshot.as_ref().map(|s| (s.publisher_connected, s.publisher_disconnect)), + settled, + ); + + // The subscriber transport exists exactly when this is not single-PC mode + // (see `RtcSession::connect`), so matching on the `Option` itself states the + // cases without a fail-open default for a combination that cannot occur: + // absent means single-PC mode, where the publisher is the only transport and + // there is nothing else to wait for. + let subscriber_ok = match self.subscriber_pc.as_ref() { + None => true, + // Present, but a publisher-primary connection does not carry media on it, + // so it is not on the critical path for this wait. + Some(_) if !self.subscriber_primary => true, + Some(pc) => Self::transport_recovered( + pc, + snapshot + .as_ref() + .map(|s| (s.subscriber_connected, s.subscriber_disconnect)), + settled, + ), }; // In single-PC mode the publisher is the only transport, so it must always @@ -2257,7 +2359,7 @@ impl SessionInner { let need_publisher = self.single_pc_mode || self.has_published.load(Ordering::Acquire); - if subscriber_connected && (!need_publisher || publisher_connected) { + if subscriber_ok && (!need_publisher || publisher_ok) { break; } @@ -2276,6 +2378,22 @@ impl SessionInner { } } + /// `generations` carries the `(connected, disconnect)` counts sampled before the resume, + /// or `None` on the initial connect. See [`Self::wait_pc_reconnected_with_snapshot`]. + fn transport_recovered( + pc: &PeerTransport, + generations: Option<(u32, u32)>, + settled: bool, + ) -> bool { + recovery_decision( + pc.is_connected(), + pc.connected_generation(), + pc.disconnect_generation(), + generations, + settled, + ) + } + /// Start publisher negotiation fn publisher_negotiation_needed(self: &Arc) { let fast_publish = self.fast_publish.load(Ordering::Acquire); @@ -2544,6 +2662,41 @@ pub fn handle_remote_dt_packets(dc: &DataChannel, emitter: WeakUnboundedSender, + settled: bool, +) -> bool { + if !connected { + return false; + } + + let Some((snap_connected, snap_disconnect)) = generations else { + return true; // initial connect: reaching Connected is the whole contract + }; + + if connected_generation != snap_connected { + return true; // entered Connected on this attempt, so ICE and DTLS completed + } + + if disconnect_generation != snap_disconnect { + return false; // dropped and not back since; the current state is stale + } + + // Never dropped, never re-entered: either the media plane was fine throughout (a + // signal-only failure) or the far end is gone and ICE has not noticed. Indistinguishable + // until the settle window gives ICE time to time out. + settled +} + macro_rules! make_rtc_config { ($fncname:ident, $proto:ty) => { fn $fncname(value: $proto, mut config: RtcConfiguration) -> RtcConfiguration { @@ -2573,7 +2726,69 @@ make_rtc_config!(make_rtc_config_reconnect, proto::ReconnectResponse); #[cfg(test)] mod tests { - use super::{parse_sdp_max_message_size, DEFAULT_MAX_MESSAGE_SIZE}; + use super::{parse_sdp_max_message_size, recovery_decision, DEFAULT_MAX_MESSAGE_SIZE}; + + /// `(connected, disconnect)` counts as sampled before a resume, for readability below. + const SNAPSHOT: Option<(u32, u32)> = Some((7, 3)); + + /// A subscriber left dead by a node failure keeps reporting `Connected` until ICE times + /// out. Accepting that reports a recovery that did not happen — `Resumed`, and so + /// `RoomEvent::Reconnected` with `ConnectionState::Connected`, for a session receiving no + /// media. The dropped-and-not-back-since history has to win over the current state, and + /// no amount of settling may override it. + #[test] + fn stale_connected_after_a_drop_is_not_recovery() { + assert!(!recovery_decision( + /* connected= */ true, + /* connected_generation= */ 7, // unchanged: never re-entered Connected + /* disconnect_generation= */ 4, // bumped: it dropped at some point + SNAPSHOT, /* settled= */ true, + )); + } + + /// Re-entering `Connected` means ICE and DTLS completed on this attempt, so it is accepted + /// straight away and a real reconnection never pays the settle delay. + /// + /// Note this must be evidence of *connecting*, not of negotiating: the server sends its + /// subscriber offer while the old transport can still read `Connected`, so accepting a + /// completed negotiation here would reproduce the defect above. + #[test] + fn reconnection_is_accepted_immediately() { + assert!(recovery_decision( + /* connected= */ true, + /* connected_generation= */ 8, // entered Connected since the snapshot + /* disconnect_generation= */ 4, // it did drop — irrelevant, it is back + SNAPSHOT, /* settled= */ false, // still inside the settle window + )); + } + + /// A signal-only failure leaves the media plane untouched, so the existing connection is + /// still good — but that is indistinguishable from a far end that vanished until ICE has + /// had time to notice. + #[test] + fn untouched_connection_is_accepted_only_after_settling() { + let untouched = |settled| recovery_decision(true, 7, 3, SNAPSHOT, settled); + + assert!(!untouched(false), "must not accept before ICE could notice a dead path"); + assert!(untouched(true)); + } + + #[test] + fn transport_not_currently_connected_is_never_recovered() { + for connected_generation in [7, 8] { + for settled in [false, true] { + assert!(!recovery_decision(false, connected_generation, 3, SNAPSHOT, settled)); + } + } + } + + /// The initial connect has no earlier connection to confuse `Connected` with, so it is + /// taken at face value rather than gated on a transition. + #[test] + fn initial_connect_takes_connected_at_face_value() { + assert!(recovery_decision(true, 0, 0, None, false)); + assert!(!recovery_decision(false, 0, 0, None, true)); + } #[test] fn parses_max_message_size_from_application_section() {