From 3c5dcb2004a2409b3fba81992ea92a675d3bf9ef Mon Sep 17 00:00:00 2001 From: shijing xian Date: Sun, 16 Aug 2026 23:19:56 +0800 Subject: [PATCH 1/7] fix(rtc_engine): require evidence of PeerConnection recovery on resume A resume decided that a transport had recovered by reading `PeerConnectionState`. That state keeps reporting `Connected` for tens of seconds after the far end goes away -- ICE only leaves `Connected` after its receiving timeout, and only reaches `Failed` after consent expiry -- so the check could not distinguish a transport that recovered from one whose peer had vanished. When it read the stale value, the resume reported success and the engine emitted `Resumed`, and so `RoomEvent::Reconnected` with `ConnectionState::Connected`, for a session whose subscriber transport was dead. Applications had no signal that they had stopped receiving media. The transport's eventual `Failed` then started a fresh cycle -- as a resume again, since no escalation had been recorded -- which burned the full `ICE_CONNECT_TIMEOUT` before escalating. Track two per-transport generations instead, both bumped from existing seams: `negotiation_generation` on every applied remote description, and `disconnect_generation` on every transition away from `Connected`. A resume samples both before touching the signalling link, and then accepts a transport only when it is connected and either renegotiated since the resume began -- positive proof of a live path -- or never left `Connected` for the settle window. A transport that broke and has not renegotiated is rejected regardless of what it currently reports. Because a renegotiation is accepted immediately, genuine recovery no longer waits out a fixed delay; the settle window now bounds only the ambiguous case and is raised to 3s so it exceeds ICE's receiving timeout. Also mark the subscriber as restarting ICE for the duration of a resume. It never issues its own offer, so it had no `create_and_send_offer(ice_restart)` call to set the flag, and remote candidates for the new generation arriving before the SFU's offer were applied against the old remote description instead of being queued. Mirrors `PCTransportManager.triggerIceRestart` in client-sdk-js. Co-Authored-By: Claude Opus 5 (1M context) --- ...resume_requires_evidence_of_pc_recovery.md | 12 + livekit/specs/signalling-reconnection.allium | 20 +- livekit/src/rtc_engine/mod.rs | 34 ++- livekit/src/rtc_engine/peer_transport.rs | 162 +++++++++- livekit/src/rtc_engine/rtc_session.rs | 280 ++++++++++++++++-- 5 files changed, 473 insertions(+), 35 deletions(-) create mode 100644 .changeset/resume_requires_evidence_of_pc_recovery.md 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..f571e7cf4 --- /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. Resumes now require evidence of recovery: a negotiation completed since the +resume began, or a connection that never broke. diff --git a/livekit/specs/signalling-reconnection.allium b/livekit/specs/signalling-reconnection.allium index daa7977ee..ef65f86f2 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 will accept "this transport never left connected" as + -- sufficient evidence of recovery (PC_RECONNECT_SETTLE_DELAY). Bounds only + -- the ambiguous case: a transport that completes a renegotiation during the + -- resume is accepted immediately. Must exceed ICE's receiving timeout, or a + -- 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 } ------------------------------------------------------------ @@ -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* + -- recovered, measured against generations sampled before the resume + -- began (step 0). A transport counts as recovered when it is connected + -- and either renegotiated since the resume started, or never left + -- connected for the whole settle window. Its 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..629643643 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -55,17 +55,26 @@ 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 will accept "this transport never left `Connected`" as sufficient +/// evidence of recovery. /// -/// 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. +/// This bounds *only* the ambiguous case. A transport that completes a renegotiation during +/// the resume — the publisher's ICE-restart answer, or the subscriber offer sent by the node +/// we landed on — is positive proof of a live path and is accepted immediately, so a genuine +/// recovery never pays this cost. +/// +/// It has to be long enough for ICE to notice a path that has silently died: libwebrtc keeps +/// reporting `Connected` until its receiving timeout elapses. Too short and a dead transport +/// is still claiming `Connected` when we look, which is precisely how a resume used to report +/// success for a session that never received media again. +/// +/// The trade-off is resume latency for a signal-only blip where the media plane was fine +/// throughout: there is no renegotiation to short-circuit on, so it waits this long before +/// declaring success. /// /// 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); +pub const PC_RECONNECT_SETTLE_DELAY: Duration = Duration::from_secs(3); #[derive(Debug, Clone, Copy, Eq, PartialEq)] pub enum SimulateScenario { @@ -1098,9 +1107,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 generations, 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 +1139,12 @@ impl EngineInner { let session = self.running_handle.read().session.clone(); + // 0. Sample the per-transport generations BEFORE touching the signalling link, so + // that step 3 can require evidence of recovery (a completed renegotiation, or an + // unbroken connection) rather than trusting `PeerConnectionState`, which keeps + // reporting `Connected` for tens of seconds after the far end disappears. + 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?; @@ -1140,7 +1156,7 @@ impl EngineInner { // 3. Re-offer the publisher (strictly AFTER SyncState) and wait for the // PeerConnections to reconnect, applying the settle delay. 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..82829c188 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,21 @@ pub struct PeerTransport { peer_connection: PeerConnection, on_offer_handler: Mutex>, inner: Arc>, + + /// Incremented every time a remote description is applied to this transport, i.e. every + /// time a negotiation round-trip completes (publisher answers, subscriber offers). + /// + /// Together with [`Self::disconnect_generation`] this lets a resume decide whether a + /// transport reporting `Connected` did so *because it recovered* or merely because + /// libwebrtc has not yet noticed the far end is gone. See + /// [`super::rtc_session::SessionInner::wait_pc_reconnected_with_snapshot`]. + negotiation_generation: AtomicU32, + + /// Incremented every time the PeerConnection leaves `Connected`. + /// + /// 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, } impl Debug for PeerTransport { @@ -67,6 +85,8 @@ impl PeerTransport { max_send_bitrate_bps: None, pending_initial_offer: None, })), + negotiation_generation: AtomicU32::new(0), + disconnect_generation: AtomicU32::new(0), } } @@ -74,6 +94,39 @@ impl PeerTransport { self.peer_connection.connection_state() == PeerConnectionState::Connected } + /// See [`Self::negotiation_generation`]. + pub fn negotiation_generation(&self) -> u32 { + self.negotiation_generation.load(Ordering::Acquire) + } + + /// See [`Self::disconnect_generation`]. + pub fn disconnect_generation(&self) -> u32 { + self.disconnect_generation.load(Ordering::Acquire) + } + + /// 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 + /// `create_and_send_offer(ice_restart)` call to set this flag. Without it, remote + /// candidates for the *new* generation that arrive before the SFU's offer are applied + /// against the old, dead remote description instead of being queued and replayed once + /// the new offer lands. Mirrors `subscriber.restartingIce = true` in client-sdk-js's + /// `PCTransportManager.triggerIceRestart`. + pub async fn mark_restarting_ice(&self) { + self.inner.lock().await.restarting_ice = true; + } + + /// Record a PeerConnection state transition observed by the session. + /// + /// Called for every `RtcEvent::ConnectionChange` on this transport's target so that + /// leaving `Connected` is durably recorded, rather than having to be caught by whoever + /// happens to be polling at that instant. + pub fn note_connection_state(&self, state: PeerConnectionState) { + if state != PeerConnectionState::Connected { + self.disconnect_generation.fetch_add(1, Ordering::AcqRel); + } + } + pub fn peer_connection(&self) -> PeerConnection { self.peer_connection.clone() } @@ -119,6 +172,15 @@ impl PeerTransport { self.peer_connection.set_remote_description(remote_description).await?; + // A negotiation round-trip completed on this transport. Recorded *after* the + // description is applied, so the generation only advances on success. + // + // Note the rollback in `create_and_send_offer` deliberately calls + // `peer_connection.set_remote_description` directly rather than going through this + // method, so re-applying the existing remote description does not count as a fresh + // negotiation. + self.negotiation_generation.fetch_add(1, Ordering::AcqRel); + for ic in inner.pending_candidates.drain(..) { self.peer_connection.add_ice_candidate(ic).await?; } @@ -652,6 +714,104 @@ mod tests { assert_eq!(transport.peer_connection().signaling_state(), SignalingState::HaveLocalOffer); } + /// The negotiation generation must advance exactly when a remote description is applied, + /// because the resume path treats that advance as proof that a transport re-established + /// against the node it landed on. If it failed to bump, a genuine recovery would be + /// rejected and escalate to an unnecessary full reconnect; if it bumped without a real + /// negotiation, a dead transport would be accepted as recovered. + #[tokio::test] + async fn negotiation_generation_advances_on_applied_remote_description() { + use libwebrtc::prelude::*; + use livekit_protocol as proto; + + let factory = PeerConnectionFactory::default(); + let config = RtcConfiguration { + ice_servers: vec![], + continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, + ice_transport_type: IceTransportsType::All, + }; + + let alice_pc = factory.create_peer_connection(config.clone()).unwrap(); + let bob_pc = factory.create_peer_connection(config).unwrap(); + let _dc = alice_pc.create_data_channel("gen", DataChannelInit::default()).unwrap(); + + let transport = + PeerTransport::new(alice_pc, proto::SignalTarget::Publisher, /* single_pc_mode= */ true); + + let offers = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let emitted = offers.clone(); + transport.on_offer(Some(Box::new(move |offer| { + emitted.lock().expect("offers lock poisoned").push(offer); + }))); + + assert_eq!(transport.negotiation_generation(), 0); + + // Sending an offer is not by itself a completed negotiation — nothing has come back + // from the far end yet, so there is no evidence of a live path. + transport.create_and_send_offer(OfferOptions::default()).await.unwrap(); + assert_eq!( + transport.negotiation_generation(), + 0, + "an unanswered offer must not count as a completed negotiation" + ); + + let offer = offers + .lock() + .expect("offers lock poisoned") + .first() + .cloned() + .expect("offer was not emitted"); + + bob_pc.set_remote_description(offer).await.unwrap(); + let answer = bob_pc.create_answer(AnswerOptions::default()).await.unwrap(); + bob_pc.set_local_description(answer.clone()).await.unwrap(); + + // Applying the answer completes the round-trip. + transport.set_remote_description(answer).await.unwrap(); + assert_eq!(transport.negotiation_generation(), 1); + } + + /// The disconnect generation records leaving `Connected` durably, so a resume polling + /// afterwards can still tell that the transport broke — the failure it must not miss is + /// a transport that dropped and returned to `Connected` between two polls. + #[test] + fn disconnect_generation_records_leaving_connected() { + 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_eq!(transport.disconnect_generation(), 0); + + transport.note_connection_state(PeerConnectionState::Connected); + assert_eq!( + transport.disconnect_generation(), + 0, + "staying Connected is not a disconnect" + ); + + transport.note_connection_state(PeerConnectionState::Disconnected); + assert_eq!(transport.disconnect_generation(), 1); + + // Returning to Connected leaves the record standing: the resume must still be able + // to see that the transport broke. + transport.note_connection_state(PeerConnectionState::Connected); + assert_eq!(transport.disconnect_generation(), 1); + + transport.note_connection_state(PeerConnectionState::Failed); + assert_eq!(transport.disconnect_generation(), 2); + } + #[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..f5d5c6ac0 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, }; +/// Per-transport negotiation/disconnect generations sampled at the start of a resume. +/// +/// A resume compares against these to tell a transport that genuinely recovered from one +/// that merely has not noticed its far end is gone yet. See +/// [`SessionInner::wait_pc_reconnected_with_snapshot`]. +#[derive(Debug, Clone, Copy)] +pub struct PcGenerationSnapshot { + publisher_negotiation: u32, + publisher_disconnect: u32, + subscriber_negotiation: 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"; @@ -849,14 +862,27 @@ impl RtcSession { self.inner.wait_pc_connection().await } - /// Wait for PCs to be connected on the resume path. + /// Sample the per-transport negotiation/disconnect generations. /// - /// 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 + /// Must be taken *before* the resume touches the signalling link, so that any + /// renegotiation or disconnect caused by the resume is visible as a change against it. + pub fn pc_generation_snapshot(&self) -> PcGenerationSnapshot { + self.inner.pc_generation_snapshot() + } + + /// Wait for the PeerConnections to have demonstrably recovered on the resume path. + /// + /// Unlike [`Self::wait_pc_connection`], this requires evidence of recovery relative to + /// `snapshot` rather than trusting the current `PeerConnectionState`, which can still + /// read `Connected` for tens of seconds after the far end has gone away. `settle_delay` + /// bounds only the "nothing ever broke" case; a transport that renegotiates is accepted + /// as soon as it does. 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 +1608,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,7 +2252,29 @@ impl SessionInner { Ok(reconnect_response) } + fn pc_generation_snapshot(&self) -> PcGenerationSnapshot { + let (subscriber_negotiation, subscriber_disconnect) = self + .subscriber_pc + .as_ref() + .map(|pc| (pc.negotiation_generation(), pc.disconnect_generation())) + .unwrap_or((0, 0)); + + PcGenerationSnapshot { + publisher_negotiation: self.publisher_pc.negotiation_generation(), + publisher_disconnect: self.publisher_pc.disconnect_generation(), + subscriber_negotiation, + subscriber_disconnect, + } + } + async fn restart_publisher(&self) -> EngineResult<()> { + // The subscriber's ICE is restarted by the SFU, which will send us a fresh offer. + // Queue any remote candidates until it arrives rather than applying them to the + // outgoing generation. + if let Some(ref sub_pc) = self.subscriber_pc { + sub_pc.mark_restarting_ice().await; + } + // 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 // when we have something to keep alive on the publisher side. @@ -2227,17 +2287,49 @@ impl SessionInner { } /// Timeout after ['MAX_ICE_CONNECT_TIMEOUT'] + /// + /// Used for the initial connect, where the transports start from `New` and there is no + /// stale state to be fooled by: reaching `Connected` at all is proof of connection. 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; - } + /// Resume-path wait: requires *evidence* that each transport recovered, not merely that + /// it currently reports `Connected`. + /// + /// A level read of `PeerConnectionState` cannot distinguish a transport that recovered + /// from one whose far end has gone away but which libwebrtc has not yet timed out (ICE + /// only leaves `Connected` after its receiving timeout, and only reaches `Failed` after + /// consent expiry, tens of seconds later). Accepting the level is what let a resume + /// report success while the subscriber was in fact dead, emitting `Resumed` — and hence + /// `RoomEvent::Reconnected` — for a session that never received media again. + /// + /// Against the generation snapshot taken at the start of the resume, a transport counts + /// as recovered when it is `Connected` *and*: + /// - its negotiation generation advanced — a fresh offer/answer completed since the + /// resume began, which is positive proof of a live path (the publisher's ICE-restart + /// answer; the subscriber's offer from the node we landed on); or + /// - it never left `Connected` for the whole settle window — nothing broke, so the + /// pre-existing connection is still good (the signal-only blip case). + /// + /// A transport that left `Connected` and has not renegotiated since is explicitly *not* + /// recovered, however it reports right 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 +2337,29 @@ 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 { + // The settle window only gates the "nothing ever broke" branch below; a + // transport that proves recovery by renegotiating is accepted immediately. + let settled = started.elapsed() >= settle_delay; + + let publisher_ok = Self::transport_recovered( + &self.publisher_pc, + snapshot.as_ref().map(|s| (s.publisher_negotiation, s.publisher_disconnect)), + settled, + ); + + let subscriber_ok = 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) + match self.subscriber_pc.as_ref() { + None => true, + Some(pc) => Self::transport_recovered( + pc, + snapshot + .as_ref() + .map(|s| (s.subscriber_negotiation, s.subscriber_disconnect)), + settled, + ), + } }; // In single-PC mode the publisher is the only transport, so it must always @@ -2257,7 +2367,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 +2386,26 @@ impl SessionInner { } } + /// Decide whether `pc` has recovered. + /// + /// `generations` is `None` on the initial-connect path, where `Connected` is taken at + /// face value because there is no earlier state to confuse it with. On the resume path + /// it carries the `(negotiation, disconnect)` generations sampled before the resume + /// started. See [`Self::wait_pc_reconnected_with_snapshot`] for the rationale. + fn transport_recovered( + pc: &PeerTransport, + generations: Option<(u32, u32)>, + settled: bool, + ) -> bool { + recovery_decision( + pc.is_connected(), + pc.negotiation_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 +2674,42 @@ pub fn handle_remote_dt_packets(dc: &DataChannel, emitter: WeakUnboundedSender, + settled: bool, +) -> bool { + if !connected { + return false; + } + + let Some((snap_negotiation, snap_disconnect)) = generations else { + return true; // initial connect: reaching Connected is the whole contract + }; + + if negotiation != snap_negotiation { + return true; // renegotiated since the resume began — positive proof of a live path + } + + if disconnect != snap_disconnect { + return false; // it broke, and has not renegotiated since; `Connected` is not trustworthy + } + + // Never broke and never renegotiated. Most likely a signal-only failure where the media + // plane was fine throughout — but it is also what a not-yet-timed-out dead transport + // looks like, so only accept once the settle window has passed and ICE has had a chance + // to notice. + settled +} + macro_rules! make_rtc_config { ($fncname:ident, $proto:ty) => { fn $fncname(value: $proto, mut config: RtcConfiguration) -> RtcConfiguration { @@ -2573,7 +2739,81 @@ 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}; + + /// The generations sampled at the start of a resume, for readability below. + const SNAPSHOT: Option<(u32, u32)> = Some((7, 3)); + + /// The defect this fix exists for. + /// + /// After a node failure the subscriber's PeerConnection keeps reporting `Connected` + /// until ICE times out, so a resume that trusts the connection state declares success + /// for a transport that is in fact dead — emitting `Resumed`, and hence + /// `RoomEvent::Reconnected` plus `ConnectionState::Connected`, for a session that never + /// receives media again. The transport left `Connected` at some point (recorded in the + /// disconnect generation) and has not renegotiated since, so it must not be accepted no + /// matter what it currently reports, and no matter how long we have settled for. + #[test] + fn stale_connected_after_a_disconnect_is_not_recovery() { + assert!(!recovery_decision( + /* connected= */ true, + /* negotiation= */ 7, // unchanged: nothing renegotiated + /* disconnect= */ 4, // bumped: it left Connected at some point + SNAPSHOT, + /* settled= */ true, + )); + } + + /// A completed renegotiation is positive proof of a live path, so it is accepted + /// immediately — a genuine recovery never waits out the settle window. + #[test] + fn renegotiation_is_accepted_immediately() { + assert!(recovery_decision( + /* connected= */ true, + /* negotiation= */ 8, // the new node's offer/answer landed + /* disconnect= */ 4, // it did break — irrelevant, it has since renegotiated + SNAPSHOT, + /* settled= */ false, // still inside the settle window + )); + } + + /// The signal-only blip: the media plane never broke, so the pre-existing connection is + /// still good. Accepted, but only once the settle window has given ICE a chance to + /// notice a path that died silently. + #[test] + fn unbroken_connection_is_accepted_only_after_settling() { + let unbroken = |settled| { + recovery_decision( + /* connected= */ true, + /* negotiation= */ 7, + /* disconnect= */ 3, + SNAPSHOT, + settled, + ) + }; + + assert!(!unbroken(false), "must not accept before ICE could notice a dead path"); + assert!(unbroken(true)); + } + + /// A transport that is not currently connected is never recovered, whatever its + /// generations say. + #[test] + fn disconnected_transport_is_never_recovered() { + for negotiation in [7, 8] { + for settled in [false, true] { + assert!(!recovery_decision(false, negotiation, 3, SNAPSHOT, settled)); + } + } + } + + /// The initial-connect path has no earlier state to be confused by, so reaching + /// `Connected` is the whole contract and must not be gated on renegotiation. + #[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() { From b2f300aa54448063a9f95dea0ed79b7ab8de7d51 Mon Sep 17 00:00:00 2001 From: shijing xian Date: Mon, 17 Aug 2026 18:20:51 +0800 Subject: [PATCH 2/7] fix(rtc_engine): bound the subscriber's ICE-restart window to the resume `restart_publisher` marked the subscriber as awaiting a fresh ICE generation on every resume, but only `set_remote_description` cleared it -- and the SFU re-offers the subscriber only when the resume actually moved the participant to another node. After the far more common signal-only resume no offer arrives, so the flag stayed set for the lifetime of the transport and every subsequent remote candidate was buffered instead of applied, leaving the subscriber unable to adopt any new network path the server proposed. The publisher never had this problem because it sets the flag alongside an offer it will certainly receive an answer to. The subscriber's is speculative, so it needs an explicit close: `finish_restarting_ice` clears the flag and applies whatever queued behind it, and the resume calls it once it can no longer expect an offer. Where the SFU did re-offer, the description has already cleared the flag and this is a no-op. Called on the resume's failure path as well, so a transport we may yet keep is never left stranded. Candidates queued because no remote description exists yet stay queued: there is still nothing to apply them against. Co-Authored-By: Claude Opus 5 (1M context) --- livekit/src/rtc_engine/mod.rs | 10 +- livekit/src/rtc_engine/peer_transport.rs | 177 +++++++++++++++++++++-- livekit/src/rtc_engine/rtc_session.rs | 45 ++++-- 3 files changed, 208 insertions(+), 24 deletions(-) diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index 629643643..011066a56 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -1156,7 +1156,15 @@ impl EngineInner { // 3. Re-offer the publisher (strictly AFTER SyncState) and wait for the // PeerConnections to reconnect, applying the settle delay. session.restart_publisher().await?; - session.wait_pc_reconnected(pc_snapshot, PC_RECONNECT_SETTLE_DELAY).await?; + let reconnected = session.wait_pc_reconnected(pc_snapshot, PC_RECONNECT_SETTLE_DELAY).await; + + // The SFU re-offers the subscriber only when the resume moved us to a different + // node; on an ordinary signal-only resume no offer is coming, so close the window + // `restart_publisher` opened and apply whatever queued behind it. Done on the + // failure path too: leaving it open would strand every later remote candidate on a + // transport we may yet keep. + session.finish_subscriber_ice_restart().await; + reconnected?; // 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 82829c188..975c76d79 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -112,10 +112,48 @@ impl PeerTransport { /// against the old, dead remote description instead of being queued and replayed once /// the new offer lands. Mirrors `subscriber.restartingIce = true` in client-sdk-js's /// `PCTransportManager.triggerIceRestart`. + /// + /// This is speculative in a way the publisher's equivalent is not: the publisher sets the + /// flag alongside an offer it will certainly receive an answer to, whereas here we are + /// only *expecting* the SFU to re-offer. It therefore MUST be paired with + /// [`Self::finish_restarting_ice`] to bound the window, or a resume where the SFU never + /// re-offers would leave the flag set for the lifetime of the transport. pub async fn mark_restarting_ice(&self) { self.inner.lock().await.restarting_ice = true; } + /// End the window opened by [`Self::mark_restarting_ice`], applying anything that queued + /// while it was open. + /// + /// [`Self::set_remote_description`] already does this when the expected description + /// arrives, so in that case this is a no-op. It exists for the case where the expected + /// description never comes: the SFU only re-offers the subscriber when the resume + /// actually moved us to a new node, so after an ordinary signal-only resume the flag + /// would otherwise stay set forever and every subsequent remote candidate would be + /// buffered instead of applied — leaving the transport unable to adopt any new network + /// path the server proposes. + pub async fn finish_restarting_ice(&self) -> EngineResult<()> { + let mut inner = self.inner.lock().await; + if !inner.restarting_ice { + return Ok(()); + } + inner.restarting_ice = false; + + // No fresh description arrived, so anything queued belongs to the generation that is + // still current and can be applied exactly as it would have been before the resume. + // With no description to apply them against, leave them queued as + // `add_ice_candidate` itself would. + if self.peer_connection.current_remote_description().is_none() { + return Ok(()); + } + + for ic in inner.pending_candidates.drain(..) { + self.peer_connection.add_ice_candidate(ic).await?; + } + + Ok(()) + } + /// Record a PeerConnection state transition observed by the session. /// /// Called for every `RtcEvent::ConnectionChange` on this transport's target so that @@ -735,8 +773,11 @@ mod tests { let bob_pc = factory.create_peer_connection(config).unwrap(); let _dc = alice_pc.create_data_channel("gen", DataChannelInit::default()).unwrap(); - let transport = - PeerTransport::new(alice_pc, proto::SignalTarget::Publisher, /* single_pc_mode= */ true); + let transport = PeerTransport::new( + alice_pc, + proto::SignalTarget::Publisher, + /* single_pc_mode= */ true, + ); let offers = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); let emitted = offers.clone(); @@ -788,17 +829,16 @@ mod tests { }) .unwrap(); - let transport = - PeerTransport::new(pc, proto::SignalTarget::Subscriber, /* single_pc_mode= */ false); + let transport = PeerTransport::new( + pc, + proto::SignalTarget::Subscriber, + /* single_pc_mode= */ false, + ); assert_eq!(transport.disconnect_generation(), 0); transport.note_connection_state(PeerConnectionState::Connected); - assert_eq!( - transport.disconnect_generation(), - 0, - "staying Connected is not a disconnect" - ); + assert_eq!(transport.disconnect_generation(), 0, "staying Connected is not a disconnect"); transport.note_connection_state(PeerConnectionState::Disconnected); assert_eq!(transport.disconnect_generation(), 1); @@ -812,6 +852,125 @@ mod tests { assert_eq!(transport.disconnect_generation(), 2); } + /// A resume marks the subscriber as awaiting a fresh offer, but the SFU only re-offers + /// when the resume moved the participant to another node. After the far more common + /// signal-only resume no offer arrives, so nothing in `set_remote_description` ever + /// clears the flag — and every subsequent remote candidate is buffered instead of + /// applied, leaving the transport unable to adopt any new network path the server + /// proposes for the rest of the session. + /// + /// `finish_restarting_ice` closes that window explicitly. This test drives the + /// no-re-offer sequence: mark, queue a candidate, finish, and assert the queue drained + /// and later candidates go straight through. + #[tokio::test] + async fn finishing_ice_restart_without_a_new_offer_resumes_applying_candidates() { + use libwebrtc::prelude::*; + use livekit_protocol as proto; + + let factory = PeerConnectionFactory::default(); + let config = RtcConfiguration { + ice_servers: vec![], + continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, + ice_transport_type: IceTransportsType::All, + }; + + // Stand up a subscriber-shaped transport that has a remote description, which is the + // state a resume finds it in. + let alice_pc = factory.create_peer_connection(config.clone()).unwrap(); + let bob_pc = factory.create_peer_connection(config).unwrap(); + let _dc = bob_pc.create_data_channel("sub", DataChannelInit::default()).unwrap(); + + let transport = PeerTransport::new( + alice_pc, + proto::SignalTarget::Subscriber, + /* single_pc_mode= */ false, + ); + + // Bob (standing in for the SFU) offers; Alice answers. Alice now has a remote + // description, so candidates would normally apply immediately. + let offer = bob_pc.create_offer(OfferOptions::default()).await.unwrap(); + bob_pc.set_local_description(offer.clone()).await.unwrap(); + transport.create_anwser(offer, AnswerOptions::default()).await.unwrap(); + assert!(transport.peer_connection().current_remote_description().is_some()); + + let candidate = |port| { + IceCandidate::parse( + "0", + 0, + &format!("candidate:1 1 UDP 2130706431 192.168.1.1 {port} typ host"), + ) + .expect("test candidate should parse") + }; + + // A resume opens the window; candidates must now queue rather than be applied to a + // generation that may be on its way out. + transport.mark_restarting_ice().await; + transport.add_ice_candidate(candidate(50000)).await.unwrap(); + assert_eq!( + transport.inner.lock().await.pending_candidates.len(), + 1, + "candidates must queue while the transport awaits a fresh offer" + ); + + // The SFU never re-offers -- the signal-only resume case. Closing the window must + // drain what queued behind it. + transport.finish_restarting_ice().await.unwrap(); + assert!( + !transport.inner.lock().await.restarting_ice, + "the window must not outlive the resume that opened it" + ); + assert!( + transport.inner.lock().await.pending_candidates.is_empty(), + "queued candidates must be applied when the window closes" + ); + + // And the transport must be back to applying candidates as they arrive, so it can + // still adopt new network paths the server proposes. + transport.add_ice_candidate(candidate(50001)).await.unwrap(); + assert!( + transport.inner.lock().await.pending_candidates.is_empty(), + "later candidates must be applied directly, not buffered" + ); + } + + /// Closing a window that was never opened must not disturb a transport that is legitimately + /// queueing candidates because it has no remote description yet. + #[tokio::test] + async fn finishing_ice_restart_is_a_noop_when_not_restarting() { + 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, + ); + + // No remote description yet, so this candidate queues for that reason, not because of + // an ICE restart. + let candidate = + IceCandidate::parse("0", 0, "candidate:1 1 UDP 2130706431 192.168.1.1 50000 typ host") + .expect("test candidate should parse"); + transport.add_ice_candidate(candidate).await.unwrap(); + assert_eq!(transport.inner.lock().await.pending_candidates.len(), 1); + + transport.finish_restarting_ice().await.unwrap(); + assert_eq!( + transport.inner.lock().await.pending_candidates.len(), + 1, + "candidates awaiting a first remote description must stay queued" + ); + } + #[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 f5d5c6ac0..c5ce8ab8f 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -840,6 +840,16 @@ impl RtcSession { self.inner.restart_publisher().await } + /// Close the subscriber's "awaiting a fresh offer" window opened by + /// [`Self::restart_publisher`], applying any candidates that queued while it was open. + /// + /// Must be called once the resume can no longer expect an offer from the SFU — it only + /// re-offers the subscriber when the resume moved us to a different node, so on the + /// common signal-only resume nothing else would ever close the window. + pub async fn finish_subscriber_ice_restart(&self) { + self.inner.finish_subscriber_ice_restart().await + } + /// Ends the resume-time accumulation started by [`Self::restart`], /// returning the participant identities seen since. The post-resume /// snapshot arrived several round trips before the PeerConnections @@ -2267,10 +2277,21 @@ impl SessionInner { } } + async fn finish_subscriber_ice_restart(&self) { + if let Some(ref sub_pc) = self.subscriber_pc { + // A candidate we cannot apply is not worth failing the resume over — it would + // escalate to a full reconnect over one bad candidate. The rest still drain. + if let Err(err) = sub_pc.finish_restarting_ice().await { + log::warn!("failed to apply queued subscriber ice candidates: {:?}", err); + } + } + } + async fn restart_publisher(&self) -> EngineResult<()> { - // The subscriber's ICE is restarted by the SFU, which will send us a fresh offer. - // Queue any remote candidates until it arrives rather than applying them to the - // outgoing generation. + // The SFU restarts the subscriber's ICE and sends us a fresh offer *if* this resume + // moved us to another node. Queue remote candidates until we know, rather than + // applying them to a generation that may be on its way out. Closed by + // `finish_subscriber_ice_restart` once the resume can no longer expect that offer. if let Some(ref sub_pc) = self.subscriber_pc { sub_pc.mark_restarting_ice().await; } @@ -2758,9 +2779,8 @@ mod tests { assert!(!recovery_decision( /* connected= */ true, /* negotiation= */ 7, // unchanged: nothing renegotiated - /* disconnect= */ 4, // bumped: it left Connected at some point - SNAPSHOT, - /* settled= */ true, + /* disconnect= */ 4, // bumped: it left Connected at some point + SNAPSHOT, /* settled= */ true, )); } @@ -2771,9 +2791,9 @@ mod tests { assert!(recovery_decision( /* connected= */ true, /* negotiation= */ 8, // the new node's offer/answer landed - /* disconnect= */ 4, // it did break — irrelevant, it has since renegotiated - SNAPSHOT, - /* settled= */ false, // still inside the settle window + /* disconnect= */ + 4, // it did break — irrelevant, it has since renegotiated + SNAPSHOT, /* settled= */ false, // still inside the settle window )); } @@ -2784,11 +2804,8 @@ mod tests { fn unbroken_connection_is_accepted_only_after_settling() { let unbroken = |settled| { recovery_decision( - /* connected= */ true, - /* negotiation= */ 7, - /* disconnect= */ 3, - SNAPSHOT, - settled, + /* connected= */ true, /* negotiation= */ 7, /* disconnect= */ 3, + SNAPSHOT, settled, ) }; From 28a6d6e2408736b0465a583d8883d7ee40d1f7f3 Mon Sep 17 00:00:00 2001 From: shijing xian Date: Mon, 17 Aug 2026 20:31:54 +0800 Subject: [PATCH 3/7] fix(rtc_engine): close the subscriber ICE-restart window on every resume exit The previous commit closed the window after the PC wait, which covered the wait succeeding and the wait failing but not `restart_publisher` itself failing: it opened the window as its first act and could then fail sending the publisher offer, whose `?` in the caller skipped the close entirely. That leak was survivable only by a three-hop argument -- any resume failure sets `full_reconnect`, a full reconnect closes the old session before building a new one, so the stranded flag ends up on a dead transport. True today, but it makes a local invariant depend on distant escalation policy; if a failed resume were ever retried as a resume (as client-sdk-js does for signal-level errors) the leak would become live. Make the pairing structural instead. Opening the window moves out of `restart_publisher` into the resume, directly adjacent to the close, and both steps are wrapped so every exit passes through it. `restart_publisher` goes back to doing just the one thing its name claims. Co-Authored-By: Claude Opus 5 (1M context) --- livekit/src/rtc_engine/mod.rs | 29 ++++++++++----- livekit/src/rtc_engine/peer_transport.rs | 47 ++++++++++++++++++++++++ livekit/src/rtc_engine/rtc_session.rs | 27 +++++++++----- 3 files changed, 83 insertions(+), 20 deletions(-) diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index 011066a56..c397811f3 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -1154,17 +1154,26 @@ impl EngineInner { 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. - session.restart_publisher().await?; - let reconnected = session.wait_pc_reconnected(pc_snapshot, PC_RECONNECT_SETTLE_DELAY).await; - - // The SFU re-offers the subscriber only when the resume moved us to a different - // node; on an ordinary signal-only resume no offer is coming, so close the window - // `restart_publisher` opened and apply whatever queued behind it. Done on the - // failure path too: leaving it open would strand every later remote candidate on a - // transport we may yet keep. + // PeerConnections to demonstrably recover. + // + // The subscriber is held in "awaiting a fresh ICE generation" for the duration, so + // remote candidates queue rather than being applied to a generation that may be on + // its way out. The SFU only re-offers the subscriber when this resume moved us to + // another node, so on an ordinary signal-only resume no offer is coming and nothing + // else would ever close that window. + // + // Both steps are therefore wrapped so that EVERY exit — including a publisher + // offer that fails before we ever wait — passes through the close below. Leaving + // the window open would strand every later remote candidate on a transport we may + // yet keep, i.e. the subscriber would silently stop adopting new network paths. + session.begin_subscriber_ice_restart().await; + let recovered = async { + session.restart_publisher().await?; + session.wait_pc_reconnected(pc_snapshot, PC_RECONNECT_SETTLE_DELAY).await + } + .await; session.finish_subscriber_ice_restart().await; - reconnected?; + recovered?; // 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 975c76d79..cbad51857 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -971,6 +971,53 @@ mod tests { ); } + /// The window must close even when the resume did nothing else at all -- a publisher + /// offer that fails immediately after the window opens still has to leave the subscriber + /// applying candidates. Closing twice must also be harmless, since the SFU's offer may + /// already have closed the window before the resume gets round to it. + #[tokio::test] + async fn finishing_ice_restart_is_unconditional_and_idempotent() { + use libwebrtc::prelude::*; + use livekit_protocol as proto; + + let factory = PeerConnectionFactory::default(); + let config = RtcConfiguration { + ice_servers: vec![], + continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, + ice_transport_type: IceTransportsType::All, + }; + + let alice_pc = factory.create_peer_connection(config.clone()).unwrap(); + let bob_pc = factory.create_peer_connection(config).unwrap(); + let _dc = bob_pc.create_data_channel("sub", DataChannelInit::default()).unwrap(); + + let transport = PeerTransport::new( + alice_pc, + proto::SignalTarget::Subscriber, + /* single_pc_mode= */ false, + ); + + let offer = bob_pc.create_offer(OfferOptions::default()).await.unwrap(); + bob_pc.set_local_description(offer.clone()).await.unwrap(); + transport.create_anwser(offer, AnswerOptions::default()).await.unwrap(); + + // Open the window, then close it with nothing having happened in between. + transport.mark_restarting_ice().await; + transport.finish_restarting_ice().await.unwrap(); + assert!(!transport.inner.lock().await.restarting_ice); + + // Closing again is a no-op rather than an error. + transport.finish_restarting_ice().await.unwrap(); + assert!(!transport.inner.lock().await.restarting_ice); + + // The transport is still usable: candidates apply rather than queue. + let candidate = + IceCandidate::parse("0", 0, "candidate:1 1 UDP 2130706431 192.168.1.1 50000 typ host") + .expect("test candidate should parse"); + transport.add_ice_candidate(candidate).await.unwrap(); + assert!(transport.inner.lock().await.pending_candidates.is_empty()); + } + #[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 c5ce8ab8f..f5ffc03b1 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -840,12 +840,21 @@ impl RtcSession { self.inner.restart_publisher().await } - /// Close the subscriber's "awaiting a fresh offer" window opened by - /// [`Self::restart_publisher`], applying any candidates that queued while it was open. + /// Open the subscriber's "awaiting a fresh ICE generation" window for the duration of a + /// resume, so remote candidates queue instead of being applied to a generation that may + /// be on its way out. /// - /// Must be called once the resume can no longer expect an offer from the SFU — it only - /// re-offers the subscriber when the resume moved us to a different node, so on the - /// common signal-only resume nothing else would ever close the window. + /// MUST be paired with [`Self::finish_subscriber_ice_restart`] on every exit from the + /// resume, including failures: the SFU re-offers the subscriber only when the resume + /// moved us to a different node, so on the common signal-only resume nothing else ever + /// closes the window and the transport would stop applying candidates for good. + pub async fn begin_subscriber_ice_restart(&self) { + self.inner.begin_subscriber_ice_restart().await + } + + /// Close the window opened by [`Self::begin_subscriber_ice_restart`], applying any + /// candidates that queued while it was open. Idempotent, and a no-op when the SFU's + /// offer already closed the window. pub async fn finish_subscriber_ice_restart(&self) { self.inner.finish_subscriber_ice_restart().await } @@ -2287,15 +2296,13 @@ impl SessionInner { } } - async fn restart_publisher(&self) -> EngineResult<()> { - // The SFU restarts the subscriber's ICE and sends us a fresh offer *if* this resume - // moved us to another node. Queue remote candidates until we know, rather than - // applying them to a generation that may be on its way out. Closed by - // `finish_subscriber_ice_restart` once the resume can no longer expect that offer. + async fn begin_subscriber_ice_restart(&self) { if let Some(ref sub_pc) = self.subscriber_pc { sub_pc.mark_restarting_ice().await; } + } + 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 // when we have something to keep alive on the publisher side. From a467fe4565993ef64080085d1f04eb2a90188ddb Mon Sep 17 00:00:00 2001 From: shijing xian Date: Mon, 17 Aug 2026 20:37:21 +0800 Subject: [PATCH 4/7] docs(rtc_engine): record the one resume exit that skips the ICE-restart close Cancellation is the single path that does not reach `finish_subscriber_ice_restart`: `reconnect_task` runs in a `select!` against `close_notifier`, so an engine close while the resume is awaiting drops the future outright and leaves the subscriber's window open. That is sound rather than merely tolerated, and the note says why: an engine close sets the terminal `closed` flag, so no further resume runs and the flag is stranded on a transport nothing will touch again. It also records why there is no guard -- `Drop` cannot await and the flag lives behind an async mutex -- and the two changes that would invalidate the reasoning, so whoever makes one of them finds out here instead of rediscovering the hole. Co-Authored-By: Claude Opus 5 (1M context) --- livekit/src/rtc_engine/mod.rs | 13 +++++++++++++ livekit/src/rtc_engine/peer_transport.rs | 5 +++++ 2 files changed, 18 insertions(+) diff --git a/livekit/src/rtc_engine/mod.rs b/livekit/src/rtc_engine/mod.rs index c397811f3..1db9eb7fb 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -1166,6 +1166,19 @@ impl EngineInner { // offer that fails before we ever wait — passes through the close below. Leaving // the window open would strand every later remote candidate on a transport we may // yet keep, i.e. the subscriber would silently stop adopting new network paths. + // + // NOTE: one path does not reach the close — cancellation. `reconnect_task` runs in + // a `select!` against `close_notifier` (see `reconnection_needed`), so an engine + // close while we are awaiting here drops this future outright and the window stays + // open. That is sound rather than merely tolerable: `close()` sets `closed`, which + // is terminal — `wait_reconnection` refuses afterwards and no further resume runs — + // so the flag is stranded on a transport nothing will use again. + // + // It is also not fixable with a guard: `Drop` cannot await, and the flag lives + // behind an async mutex. If the engine ever became reusable after close, or this + // window came to gate anything beyond candidate buffering, the invariant would + // need re-establishing on the close path (or the flag moving to an atomic that a + // `Drop` guard can clear synchronously). session.begin_subscriber_ice_restart().await; let recovered = async { session.restart_publisher().await?; diff --git a/livekit/src/rtc_engine/peer_transport.rs b/livekit/src/rtc_engine/peer_transport.rs index cbad51857..55e928f30 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -132,6 +132,11 @@ impl PeerTransport { /// would otherwise stay set forever and every subsequent remote candidate would be /// buffered instead of applied — leaving the transport unable to adopt any new network /// path the server proposes. + /// + /// NOTE: a resume cancelled mid-flight by an engine close never reaches this, leaving the + /// window open. Sound today because an engine close is terminal, so the transport is never + /// used again; see the note in `EngineInner::try_resume_connection` for what would have to + /// change for that to matter. pub async fn finish_restarting_ice(&self) -> EngineResult<()> { let mut inner = self.inner.lock().await; if !inner.restarting_ice { From 5655154a1213be7ad5b84abee5951e4773e5cfe6 Mon Sep 17 00:00:00 2001 From: shijing xian Date: Mon, 17 Aug 2026 20:57:30 +0800 Subject: [PATCH 5/7] chore(changeset): note the subscriber ICE-restart window fix The existing changeset covers only the stale-`Connected` resume verdict. The subscriber ICE-restart window fix added later in this PR is a separate user-facing behaviour change and needs its own changelog entry. Co-Authored-By: Claude Opus 5 (1M context) --- .changeset/subscriber_ice_restart_window_bounded.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .changeset/subscriber_ice_restart_window_bounded.md diff --git a/.changeset/subscriber_ice_restart_window_bounded.md b/.changeset/subscriber_ice_restart_window_bounded.md new file mode 100644 index 000000000..cf44b9607 --- /dev/null +++ b/.changeset/subscriber_ice_restart_window_bounded.md @@ -0,0 +1,13 @@ +--- +livekit: patch +--- + +Fix the subscriber buffering remote ICE candidates for the rest of the session after a resume. + +A resume marked the subscriber as awaiting a fresh ICE generation so that remote candidates +queue rather than being applied to a generation on its way out, but only an arriving remote +description closed that window — and the server re-offers the subscriber only when the resume +moved the participant to a different node. After an ordinary signal-only resume no offer +arrives, so the window stayed open and every later remote candidate was queued instead of +applied, leaving the subscriber unable to adopt any new network path the server proposed. The +window is now closed on every exit from the resume, applying anything queued behind it. From 5c5e34b80581434e6b2d0524b5f1a3547cc21333 Mon Sep 17 00:00:00 2001 From: shijing xian Date: Mon, 17 Aug 2026 21:20:12 +0800 Subject: [PATCH 6/7] refactor(rtc_engine): drop the unreachable no-subscriber default from the PC wait The subscriber branch of the resume wait carried a `None => true` default, inherited from the `.map(..).unwrap_or(true)` it replaced. It was unreachable: `subscriber_pc` is `Some` exactly when `single_pc_mode` is false (established once in `connect`, never reassigned), so the branch guarded by `!single_pc_mode` could only ever see `Some`. Unreachable and also fail-open in the wrong direction -- it would have reported a non-existent subscriber as recovered, which is the very thing this PR removes elsewhere. Match on the `Option` instead, which is itself the source of truth for whether a second transport exists. `None` now honestly means single-PC mode rather than standing in for an impossible state, there is no default to fall open through, and the redundant `single_pc_mode` test disappears from the condition. Also record on the field that the equivalence is load-bearing, since this wait now reads absence as "nothing more to wait for". Co-Authored-By: Claude Opus 5 (1M context) --- livekit/src/rtc_engine/rtc_session.rs | 37 +++++++++++++++++---------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index f5ffc03b1..d47b022b1 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -370,7 +370,12 @@ 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; in single peer connection mode + /// this is None and publisher_pc handles both send/receive. + /// + /// The equivalence is load-bearing, not incidental: `wait_pc_connection_inner` reads + /// absence as "there is no second transport to wait for". Anything that could leave this + /// `None` in dual-PC mode would make a resume stop waiting on the subscriber. subscriber_pc: Option, /// Whether single peer connection mode is active single_pc_mode: bool, @@ -2375,19 +2380,23 @@ impl SessionInner { settled, ); - let subscriber_ok = if self.single_pc_mode || !self.subscriber_primary { - true // No subscriber in single PC mode or if PC is publisher primary - } else { - match self.subscriber_pc.as_ref() { - None => true, - Some(pc) => Self::transport_recovered( - pc, - snapshot - .as_ref() - .map(|s| (s.subscriber_negotiation, s.subscriber_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_negotiation, s.subscriber_disconnect)), + settled, + ), }; // In single-PC mode the publisher is the only transport, so it must always From 6049dd6ece408771229923e8a71b00d2f6d3d4fb Mon Sep 17 00:00:00 2001 From: shijing xian Date: Mon, 17 Aug 2026 22:02:51 +0800 Subject: [PATCH 7/7] fix(rtc_engine): judge resume recovery on reconnection, not renegotiation Two corrections to this PR, both found by tracing what the server actually does on a resume. 1. The recovery test accepted a completed negotiation as proof of a live path. It is not: applying the server's subscriber offer shows the signalling link works and the server is rebuilding, nothing about media. Cloud emits that offer in response to SyncState, so it lands before or during the wait, while a subscriber orphaned by a dead node still reads `Connected` for ICE's receiving timeout. The fast path would then accept immediately -- exactly the premature `Resumed` this PR set out to prevent. Count entries into `Connected` instead of applied remote descriptions. That is evidence ICE and DTLS completed on this attempt, and it drops the `set_remote_description` bump: both counters now come from the one `ConnectionChange` handler. 2. Drop the subscriber ICE-restart window. It guarded against remote candidates arriving before the server's offer and being applied to the outgoing generation, but the server cannot produce that order. On a same-node resume `ResumeParticipant` -> `ICERestart` -> `clearLocalDescriptionSent` buffers candidates until after the offer. On a reconnect landing elsewhere, Cloud's migration-in path drops subscriber candidates entirely while `MigrateStateInit`, and only leaves that state immediately before creating the offer, at which point gathering has not started. So it protected nothing while costing two defects -- a flag that outlived the resume, and an ordering hazard that withheld candidates during the very window judging recovery -- plus needless queueing on every signal-only resume. Also trims the comments this PR added, so they describe what the code guarantees rather than narrating the change. Co-Authored-By: Claude Opus 5 (1M context) --- ...resume_requires_evidence_of_pc_recovery.md | 12 +- .../subscriber_ice_restart_window_bounded.md | 13 - livekit/specs/signalling-reconnection.allium | 28 +- livekit/src/rtc_engine/mod.rs | 74 +--- livekit/src/rtc_engine/peer_transport.rs | 355 ++---------------- livekit/src/rtc_engine/rtc_session.rs | 234 +++++------- 6 files changed, 157 insertions(+), 559 deletions(-) delete mode 100644 .changeset/subscriber_ice_restart_window_bounded.md diff --git a/.changeset/resume_requires_evidence_of_pc_recovery.md b/.changeset/resume_requires_evidence_of_pc_recovery.md index f571e7cf4..b366c4dda 100644 --- a/.changeset/resume_requires_evidence_of_pc_recovery.md +++ b/.changeset/resume_requires_evidence_of_pc_recovery.md @@ -4,9 +4,9 @@ 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. Resumes now require evidence of recovery: a negotiation completed since the -resume began, or a connection that never broke. +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/.changeset/subscriber_ice_restart_window_bounded.md b/.changeset/subscriber_ice_restart_window_bounded.md deleted file mode 100644 index cf44b9607..000000000 --- a/.changeset/subscriber_ice_restart_window_bounded.md +++ /dev/null @@ -1,13 +0,0 @@ ---- -livekit: patch ---- - -Fix the subscriber buffering remote ICE candidates for the rest of the session after a resume. - -A resume marked the subscriber as awaiting a fresh ICE generation so that remote candidates -queue rather than being applied to a generation on its way out, but only an arriving remote -description closed that window — and the server re-offers the subscriber only when the resume -moved the participant to a different node. After an ordinary signal-only resume no offer -arrives, so the window stayed open and every later remote candidate was queued instead of -applied, leaving the subscriber unable to adopt any new network path the server proposed. The -window is now closed on every exit from the resume, applying anything queued behind it. diff --git a/livekit/specs/signalling-reconnection.allium b/livekit/specs/signalling-reconnection.allium index ef65f86f2..0b5d88a86 100644 --- a/livekit/specs/signalling-reconnection.allium +++ b/livekit/specs/signalling-reconnection.allium @@ -224,12 +224,12 @@ config { reconnect_base_delay_ms: Integer = 300 reconnect_backoff_multiplier: Integer = 2 reconnect_max_delay_ms: Integer = 7000 - -- How long a resume will accept "this transport never left connected" as - -- sufficient evidence of recovery (PC_RECONNECT_SETTLE_DELAY). Bounds only - -- the ambiguous case: a transport that completes a renegotiation during the - -- resume is accepted immediately. Must exceed ICE's receiving timeout, or a - -- transport whose far end is gone is still claiming connected when checked. - -- Resume-only; full reconnect builds new PCs. + -- 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 } @@ -542,14 +542,14 @@ rule ResumeAwaitsPeerConnections { requires: engine.status = reconnecting ensures: AwaitPeerConnectionsRequested(engine, settle: config.pc_reconnect_settle_delay) @guidance - -- Step 4: wait until each required PeerConnection has *demonstrably* - -- recovered, measured against generations sampled before the resume - -- began (step 0). A transport counts as recovered when it is connected - -- and either renegotiated since the resume started, or never left - -- connected for the whole settle window. Its 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. + -- 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 1db9eb7fb..cff8a28aa 100644 --- a/livekit/src/rtc_engine/mod.rs +++ b/livekit/src/rtc_engine/mod.rs @@ -55,25 +55,18 @@ pub(crate) type EngineEmitter = mpsc::UnboundedSender; pub(crate) type EngineEvents = mpsc::UnboundedReceiver; pub(crate) type EngineResult = Result; -/// How long a resume will accept "this transport never left `Connected`" as sufficient -/// evidence of recovery. +/// How long a resume waits before accepting "this transport never left `Connected`" as +/// evidence that it is still good. /// -/// This bounds *only* the ambiguous case. A transport that completes a renegotiation during -/// the resume — the publisher's ICE-restart answer, or the subscriber offer sent by the node -/// we landed on — is positive proof of a live path and is accepted immediately, so a genuine -/// recovery never pays this cost. +/// 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. /// -/// It has to be long enough for ICE to notice a path that has silently died: libwebrtc keeps -/// reporting `Connected` until its receiving timeout elapses. Too short and a dead transport -/// is still claiming `Connected` when we look, which is precisely how a resume used to report -/// success for a session that never received media again. -/// -/// The trade-off is resume latency for a signal-only blip where the media plane was fine -/// throughout: there is no renegotiation to short-circuit on, so it waits this long before -/// declaring success. -/// -/// Only applied to the resume path. Full reconnect builds brand-new PCs which -/// don't suffer from the "looks-Connected-but-isn't" race. +/// 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)] @@ -1107,7 +1100,7 @@ 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 generations, before anything can perturb them; + /// 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 *demonstrated* PC recovery; @@ -1139,10 +1132,9 @@ impl EngineInner { let session = self.running_handle.read().session.clone(); - // 0. Sample the per-transport generations BEFORE touching the signalling link, so - // that step 3 can require evidence of recovery (a completed renegotiation, or an - // unbroken connection) rather than trusting `PeerConnectionState`, which keeps - // reporting `Connected` for tens of seconds after the far end disappears. + // 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 @@ -1153,40 +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 demonstrably recover. - // - // The subscriber is held in "awaiting a fresh ICE generation" for the duration, so - // remote candidates queue rather than being applied to a generation that may be on - // its way out. The SFU only re-offers the subscriber when this resume moved us to - // another node, so on an ordinary signal-only resume no offer is coming and nothing - // else would ever close that window. - // - // Both steps are therefore wrapped so that EVERY exit — including a publisher - // offer that fails before we ever wait — passes through the close below. Leaving - // the window open would strand every later remote candidate on a transport we may - // yet keep, i.e. the subscriber would silently stop adopting new network paths. - // - // NOTE: one path does not reach the close — cancellation. `reconnect_task` runs in - // a `select!` against `close_notifier` (see `reconnection_needed`), so an engine - // close while we are awaiting here drops this future outright and the window stays - // open. That is sound rather than merely tolerable: `close()` sets `closed`, which - // is terminal — `wait_reconnection` refuses afterwards and no further resume runs — - // so the flag is stranded on a transport nothing will use again. - // - // It is also not fixable with a guard: `Drop` cannot await, and the flag lives - // behind an async mutex. If the engine ever became reusable after close, or this - // window came to gate anything beyond candidate buffering, the invariant would - // need re-establishing on the close path (or the flag moving to an atomic that a - // `Drop` guard can clear synchronously). - session.begin_subscriber_ice_restart().await; - let recovered = async { - session.restart_publisher().await?; - session.wait_pc_reconnected(pc_snapshot, PC_RECONNECT_SETTLE_DELAY).await - } - .await; - session.finish_subscriber_ice_restart().await; - recovered?; + // 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_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 55e928f30..634190e82 100644 --- a/livekit/src/rtc_engine/peer_transport.rs +++ b/livekit/src/rtc_engine/peer_transport.rs @@ -45,19 +45,10 @@ pub struct PeerTransport { on_offer_handler: Mutex>, inner: Arc>, - /// Incremented every time a remote description is applied to this transport, i.e. every - /// time a negotiation round-trip completes (publisher answers, subscriber offers). - /// - /// Together with [`Self::disconnect_generation`] this lets a resume decide whether a - /// transport reporting `Connected` did so *because it recovered* or merely because - /// libwebrtc has not yet noticed the far end is gone. See - /// [`super::rtc_session::SessionInner::wait_pc_reconnected_with_snapshot`]. - negotiation_generation: AtomicU32, + /// Counts entries into `Connected`; see [`Self::note_connection_state`]. + connected_generation: AtomicU32, - /// Incremented every time the PeerConnection leaves `Connected`. - /// - /// 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. + /// Counts exits from `Connected`; see [`Self::note_connection_state`]. disconnect_generation: AtomicU32, } @@ -85,7 +76,7 @@ impl PeerTransport { max_send_bitrate_bps: None, pending_initial_offer: None, })), - negotiation_generation: AtomicU32::new(0), + connected_generation: AtomicU32::new(0), disconnect_generation: AtomicU32::new(0), } } @@ -94,78 +85,27 @@ impl PeerTransport { self.peer_connection.connection_state() == PeerConnectionState::Connected } - /// See [`Self::negotiation_generation`]. - pub fn negotiation_generation(&self) -> u32 { - self.negotiation_generation.load(Ordering::Acquire) + pub fn connected_generation(&self) -> u32 { + self.connected_generation.load(Ordering::Acquire) } - /// See [`Self::disconnect_generation`]. pub fn disconnect_generation(&self) -> u32 { self.disconnect_generation.load(Ordering::Acquire) } - /// 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 - /// `create_and_send_offer(ice_restart)` call to set this flag. Without it, remote - /// candidates for the *new* generation that arrive before the SFU's offer are applied - /// against the old, dead remote description instead of being queued and replayed once - /// the new offer lands. Mirrors `subscriber.restartingIce = true` in client-sdk-js's - /// `PCTransportManager.triggerIceRestart`. - /// - /// This is speculative in a way the publisher's equivalent is not: the publisher sets the - /// flag alongside an offer it will certainly receive an answer to, whereas here we are - /// only *expecting* the SFU to re-offer. It therefore MUST be paired with - /// [`Self::finish_restarting_ice`] to bound the window, or a resume where the SFU never - /// re-offers would leave the flag set for the lifetime of the transport. - pub async fn mark_restarting_ice(&self) { - self.inner.lock().await.restarting_ice = true; - } - - /// End the window opened by [`Self::mark_restarting_ice`], applying anything that queued - /// while it was open. - /// - /// [`Self::set_remote_description`] already does this when the expected description - /// arrives, so in that case this is a no-op. It exists for the case where the expected - /// description never comes: the SFU only re-offers the subscriber when the resume - /// actually moved us to a new node, so after an ordinary signal-only resume the flag - /// would otherwise stay set forever and every subsequent remote candidate would be - /// buffered instead of applied — leaving the transport unable to adopt any new network - /// path the server proposes. + /// Record a connection-state transition, so a later observer can tell what the transport + /// has *done* rather than only what it currently reports. /// - /// NOTE: a resume cancelled mid-flight by an engine close never reaches this, leaving the - /// window open. Sound today because an engine close is terminal, so the transport is never - /// used again; see the note in `EngineInner::try_resume_connection` for what would have to - /// change for that to matter. - pub async fn finish_restarting_ice(&self) -> EngineResult<()> { - let mut inner = self.inner.lock().await; - if !inner.restarting_ice { - return Ok(()); - } - inner.restarting_ice = false; - - // No fresh description arrived, so anything queued belongs to the generation that is - // still current and can be applied exactly as it would have been before the resume. - // With no description to apply them against, leave them queued as - // `add_ice_candidate` itself would. - if self.peer_connection.current_remote_description().is_none() { - return Ok(()); - } - - for ic in inner.pending_candidates.drain(..) { - self.peer_connection.add_ice_candidate(ic).await?; - } - - Ok(()) - } - - /// Record a PeerConnection state transition observed by the session. - /// - /// Called for every `RtcEvent::ConnectionChange` on this transport's target so that - /// leaving `Connected` is durably recorded, rather than having to be caught by whoever - /// happens to be polling at that instant. + /// 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 { + if state == PeerConnectionState::Connected { + self.connected_generation.fetch_add(1, Ordering::AcqRel); + } else { self.disconnect_generation.fetch_add(1, Ordering::AcqRel); } } @@ -215,15 +155,6 @@ impl PeerTransport { self.peer_connection.set_remote_description(remote_description).await?; - // A negotiation round-trip completed on this transport. Recorded *after* the - // description is applied, so the generation only advances on success. - // - // Note the rollback in `create_and_send_offer` deliberately calls - // `peer_connection.set_remote_description` directly rather than going through this - // method, so re-applying the existing remote description does not count as a fresh - // negotiation. - self.negotiation_generation.fetch_add(1, Ordering::AcqRel); - for ic in inner.pending_candidates.drain(..) { self.peer_connection.add_ice_candidate(ic).await?; } @@ -757,71 +688,11 @@ mod tests { assert_eq!(transport.peer_connection().signaling_state(), SignalingState::HaveLocalOffer); } - /// The negotiation generation must advance exactly when a remote description is applied, - /// because the resume path treats that advance as proof that a transport re-established - /// against the node it landed on. If it failed to bump, a genuine recovery would be - /// rejected and escalate to an unnecessary full reconnect; if it bumped without a real - /// negotiation, a dead transport would be accepted as recovered. - #[tokio::test] - async fn negotiation_generation_advances_on_applied_remote_description() { - use libwebrtc::prelude::*; - use livekit_protocol as proto; - - let factory = PeerConnectionFactory::default(); - let config = RtcConfiguration { - ice_servers: vec![], - continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, - ice_transport_type: IceTransportsType::All, - }; - - let alice_pc = factory.create_peer_connection(config.clone()).unwrap(); - let bob_pc = factory.create_peer_connection(config).unwrap(); - let _dc = alice_pc.create_data_channel("gen", DataChannelInit::default()).unwrap(); - - let transport = PeerTransport::new( - alice_pc, - proto::SignalTarget::Publisher, - /* single_pc_mode= */ true, - ); - - let offers = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); - let emitted = offers.clone(); - transport.on_offer(Some(Box::new(move |offer| { - emitted.lock().expect("offers lock poisoned").push(offer); - }))); - - assert_eq!(transport.negotiation_generation(), 0); - - // Sending an offer is not by itself a completed negotiation — nothing has come back - // from the far end yet, so there is no evidence of a live path. - transport.create_and_send_offer(OfferOptions::default()).await.unwrap(); - assert_eq!( - transport.negotiation_generation(), - 0, - "an unanswered offer must not count as a completed negotiation" - ); - - let offer = offers - .lock() - .expect("offers lock poisoned") - .first() - .cloned() - .expect("offer was not emitted"); - - bob_pc.set_remote_description(offer).await.unwrap(); - let answer = bob_pc.create_answer(AnswerOptions::default()).await.unwrap(); - bob_pc.set_local_description(answer.clone()).await.unwrap(); - - // Applying the answer completes the round-trip. - transport.set_remote_description(answer).await.unwrap(); - assert_eq!(transport.negotiation_generation(), 1); - } - - /// The disconnect generation records leaving `Connected` durably, so a resume polling - /// afterwards can still tell that the transport broke — the failure it must not miss is - /// a transport that dropped and returned to `Connected` between two polls. + /// 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 disconnect_generation_records_leaving_connected() { + fn connection_state_transitions_are_counted_separately() { use libwebrtc::prelude::*; use livekit_protocol as proto; @@ -839,188 +710,24 @@ mod tests { proto::SignalTarget::Subscriber, /* single_pc_mode= */ false, ); + let counters = || (transport.connected_generation(), transport.disconnect_generation()); + + assert_eq!(counters(), (0, 0)); - assert_eq!(transport.disconnect_generation(), 0); + transport.note_connection_state(PeerConnectionState::Connecting); + assert_eq!(counters(), (0, 1)); transport.note_connection_state(PeerConnectionState::Connected); - assert_eq!(transport.disconnect_generation(), 0, "staying Connected is not a disconnect"); + 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); - assert_eq!(transport.disconnect_generation(), 1); - - // Returning to Connected leaves the record standing: the resume must still be able - // to see that the transport broke. transport.note_connection_state(PeerConnectionState::Connected); - assert_eq!(transport.disconnect_generation(), 1); + assert_eq!(counters(), (2, 2)); transport.note_connection_state(PeerConnectionState::Failed); - assert_eq!(transport.disconnect_generation(), 2); - } - - /// A resume marks the subscriber as awaiting a fresh offer, but the SFU only re-offers - /// when the resume moved the participant to another node. After the far more common - /// signal-only resume no offer arrives, so nothing in `set_remote_description` ever - /// clears the flag — and every subsequent remote candidate is buffered instead of - /// applied, leaving the transport unable to adopt any new network path the server - /// proposes for the rest of the session. - /// - /// `finish_restarting_ice` closes that window explicitly. This test drives the - /// no-re-offer sequence: mark, queue a candidate, finish, and assert the queue drained - /// and later candidates go straight through. - #[tokio::test] - async fn finishing_ice_restart_without_a_new_offer_resumes_applying_candidates() { - use libwebrtc::prelude::*; - use livekit_protocol as proto; - - let factory = PeerConnectionFactory::default(); - let config = RtcConfiguration { - ice_servers: vec![], - continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, - ice_transport_type: IceTransportsType::All, - }; - - // Stand up a subscriber-shaped transport that has a remote description, which is the - // state a resume finds it in. - let alice_pc = factory.create_peer_connection(config.clone()).unwrap(); - let bob_pc = factory.create_peer_connection(config).unwrap(); - let _dc = bob_pc.create_data_channel("sub", DataChannelInit::default()).unwrap(); - - let transport = PeerTransport::new( - alice_pc, - proto::SignalTarget::Subscriber, - /* single_pc_mode= */ false, - ); - - // Bob (standing in for the SFU) offers; Alice answers. Alice now has a remote - // description, so candidates would normally apply immediately. - let offer = bob_pc.create_offer(OfferOptions::default()).await.unwrap(); - bob_pc.set_local_description(offer.clone()).await.unwrap(); - transport.create_anwser(offer, AnswerOptions::default()).await.unwrap(); - assert!(transport.peer_connection().current_remote_description().is_some()); - - let candidate = |port| { - IceCandidate::parse( - "0", - 0, - &format!("candidate:1 1 UDP 2130706431 192.168.1.1 {port} typ host"), - ) - .expect("test candidate should parse") - }; - - // A resume opens the window; candidates must now queue rather than be applied to a - // generation that may be on its way out. - transport.mark_restarting_ice().await; - transport.add_ice_candidate(candidate(50000)).await.unwrap(); - assert_eq!( - transport.inner.lock().await.pending_candidates.len(), - 1, - "candidates must queue while the transport awaits a fresh offer" - ); - - // The SFU never re-offers -- the signal-only resume case. Closing the window must - // drain what queued behind it. - transport.finish_restarting_ice().await.unwrap(); - assert!( - !transport.inner.lock().await.restarting_ice, - "the window must not outlive the resume that opened it" - ); - assert!( - transport.inner.lock().await.pending_candidates.is_empty(), - "queued candidates must be applied when the window closes" - ); - - // And the transport must be back to applying candidates as they arrive, so it can - // still adopt new network paths the server proposes. - transport.add_ice_candidate(candidate(50001)).await.unwrap(); - assert!( - transport.inner.lock().await.pending_candidates.is_empty(), - "later candidates must be applied directly, not buffered" - ); - } - - /// Closing a window that was never opened must not disturb a transport that is legitimately - /// queueing candidates because it has no remote description yet. - #[tokio::test] - async fn finishing_ice_restart_is_a_noop_when_not_restarting() { - 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, - ); - - // No remote description yet, so this candidate queues for that reason, not because of - // an ICE restart. - let candidate = - IceCandidate::parse("0", 0, "candidate:1 1 UDP 2130706431 192.168.1.1 50000 typ host") - .expect("test candidate should parse"); - transport.add_ice_candidate(candidate).await.unwrap(); - assert_eq!(transport.inner.lock().await.pending_candidates.len(), 1); - - transport.finish_restarting_ice().await.unwrap(); - assert_eq!( - transport.inner.lock().await.pending_candidates.len(), - 1, - "candidates awaiting a first remote description must stay queued" - ); - } - - /// The window must close even when the resume did nothing else at all -- a publisher - /// offer that fails immediately after the window opens still has to leave the subscriber - /// applying candidates. Closing twice must also be harmless, since the SFU's offer may - /// already have closed the window before the resume gets round to it. - #[tokio::test] - async fn finishing_ice_restart_is_unconditional_and_idempotent() { - use libwebrtc::prelude::*; - use livekit_protocol as proto; - - let factory = PeerConnectionFactory::default(); - let config = RtcConfiguration { - ice_servers: vec![], - continual_gathering_policy: ContinualGatheringPolicy::GatherOnce, - ice_transport_type: IceTransportsType::All, - }; - - let alice_pc = factory.create_peer_connection(config.clone()).unwrap(); - let bob_pc = factory.create_peer_connection(config).unwrap(); - let _dc = bob_pc.create_data_channel("sub", DataChannelInit::default()).unwrap(); - - let transport = PeerTransport::new( - alice_pc, - proto::SignalTarget::Subscriber, - /* single_pc_mode= */ false, - ); - - let offer = bob_pc.create_offer(OfferOptions::default()).await.unwrap(); - bob_pc.set_local_description(offer.clone()).await.unwrap(); - transport.create_anwser(offer, AnswerOptions::default()).await.unwrap(); - - // Open the window, then close it with nothing having happened in between. - transport.mark_restarting_ice().await; - transport.finish_restarting_ice().await.unwrap(); - assert!(!transport.inner.lock().await.restarting_ice); - - // Closing again is a no-op rather than an error. - transport.finish_restarting_ice().await.unwrap(); - assert!(!transport.inner.lock().await.restarting_ice); - - // The transport is still usable: candidates apply rather than queue. - let candidate = - IceCandidate::parse("0", 0, "candidate:1 1 UDP 2130706431 192.168.1.1 50000 typ host") - .expect("test candidate should parse"); - transport.add_ice_candidate(candidate).await.unwrap(); - assert!(transport.inner.lock().await.pending_candidates.is_empty()); + assert_eq!(counters(), (2, 3)); } #[test] diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs index d47b022b1..a47d0a5e1 100644 --- a/livekit/src/rtc_engine/rtc_session.rs +++ b/livekit/src/rtc_engine/rtc_session.rs @@ -63,16 +63,16 @@ use crate::{ DataPacketKind, }; -/// Per-transport negotiation/disconnect generations sampled at the start of a resume. +/// Connection-state transition counts per transport, sampled before a resume begins. /// -/// A resume compares against these to tell a transport that genuinely recovered from one -/// that merely has not noticed its far end is gone yet. See +/// 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_negotiation: u32, + publisher_connected: u32, publisher_disconnect: u32, - subscriber_negotiation: u32, + subscriber_connected: u32, subscriber_disconnect: u32, } @@ -370,12 +370,9 @@ struct SessionInner { fast_publish: AtomicBool, publisher_pc: PeerTransport, - /// `Some` exactly when [`Self::single_pc_mode`] is false; in single peer connection mode - /// this is None and publisher_pc handles both send/receive. - /// - /// The equivalence is load-bearing, not incidental: `wait_pc_connection_inner` reads - /// absence as "there is no second transport to wait for". Anything that could leave this - /// `None` in dual-PC mode would make a resume stop waiting on the subscriber. + /// `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, @@ -845,25 +842,6 @@ impl RtcSession { self.inner.restart_publisher().await } - /// Open the subscriber's "awaiting a fresh ICE generation" window for the duration of a - /// resume, so remote candidates queue instead of being applied to a generation that may - /// be on its way out. - /// - /// MUST be paired with [`Self::finish_subscriber_ice_restart`] on every exit from the - /// resume, including failures: the SFU re-offers the subscriber only when the resume - /// moved us to a different node, so on the common signal-only resume nothing else ever - /// closes the window and the transport would stop applying candidates for good. - pub async fn begin_subscriber_ice_restart(&self) { - self.inner.begin_subscriber_ice_restart().await - } - - /// Close the window opened by [`Self::begin_subscriber_ice_restart`], applying any - /// candidates that queued while it was open. Idempotent, and a no-op when the SFU's - /// offer already closed the window. - pub async fn finish_subscriber_ice_restart(&self) { - self.inner.finish_subscriber_ice_restart().await - } - /// Ends the resume-time accumulation started by [`Self::restart`], /// returning the participant identities seen since. The post-resume /// snapshot arrived several round trips before the PeerConnections @@ -886,21 +864,16 @@ impl RtcSession { self.inner.wait_pc_connection().await } - /// Sample the per-transport negotiation/disconnect generations. - /// - /// Must be taken *before* the resume touches the signalling link, so that any - /// renegotiation or disconnect caused by the resume is visible as a change against it. + /// 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 demonstrably recovered on the resume path. - /// - /// Unlike [`Self::wait_pc_connection`], this requires evidence of recovery relative to - /// `snapshot` rather than trusting the current `PeerConnectionState`, which can still - /// read `Connected` for tens of seconds after the far end has gone away. `settle_delay` - /// bounds only the "nothing ever broke" case; a transport that renegotiates is accepted - /// as soon as it does. See [`SessionInner::wait_pc_reconnected_with_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, @@ -2277,36 +2250,20 @@ impl SessionInner { } fn pc_generation_snapshot(&self) -> PcGenerationSnapshot { - let (subscriber_negotiation, subscriber_disconnect) = self + let (subscriber_connected, subscriber_disconnect) = self .subscriber_pc .as_ref() - .map(|pc| (pc.negotiation_generation(), pc.disconnect_generation())) + .map(|pc| (pc.connected_generation(), pc.disconnect_generation())) .unwrap_or((0, 0)); PcGenerationSnapshot { - publisher_negotiation: self.publisher_pc.negotiation_generation(), + publisher_connected: self.publisher_pc.connected_generation(), publisher_disconnect: self.publisher_pc.disconnect_generation(), - subscriber_negotiation, + subscriber_connected, subscriber_disconnect, } } - async fn finish_subscriber_ice_restart(&self) { - if let Some(ref sub_pc) = self.subscriber_pc { - // A candidate we cannot apply is not worth failing the resume over — it would - // escalate to a full reconnect over one bad candidate. The rest still drain. - if let Err(err) = sub_pc.finish_restarting_ice().await { - log::warn!("failed to apply queued subscriber ice candidates: {:?}", err); - } - } - } - - async fn begin_subscriber_ice_restart(&self) { - if let Some(ref sub_pc) = self.subscriber_pc { - sub_pc.mark_restarting_ice().await; - } - } - 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 @@ -2319,34 +2276,32 @@ impl SessionInner { Ok(()) } - /// Timeout after ['MAX_ICE_CONNECT_TIMEOUT'] + /// Wait for the transports to connect, timing out after [`ICE_CONNECT_TIMEOUT`]. /// - /// Used for the initial connect, where the transports start from `New` and there is no - /// stale state to be fooled by: reaching `Connected` at all is proof of connection. + /// 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_inner(None, Duration::ZERO).await } - /// Resume-path wait: requires *evidence* that each transport recovered, not merely that - /// it currently reports `Connected`. + /// Wait for the transports to have *reconnected*, which a resume cannot establish by + /// reading `PeerConnectionState`. /// - /// A level read of `PeerConnectionState` cannot distinguish a transport that recovered - /// from one whose far end has gone away but which libwebrtc has not yet timed out (ICE - /// only leaves `Connected` after its receiving timeout, and only reaches `Failed` after - /// consent expiry, tens of seconds later). Accepting the level is what let a resume - /// report success while the subscriber was in fact dead, emitting `Resumed` — and hence - /// `RoomEvent::Reconnected` — for a session that never received media again. + /// 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. /// - /// Against the generation snapshot taken at the start of the resume, a transport counts - /// as recovered when it is `Connected` *and*: - /// - its negotiation generation advanced — a fresh offer/answer completed since the - /// resume began, which is positive proof of a live path (the publisher's ICE-restart - /// answer; the subscriber's offer from the node we landed on); or - /// - it never left `Connected` for the whole settle window — nothing broke, so the - /// pre-existing connection is still good (the signal-only blip case). + /// 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 left `Connected` and has not renegotiated since is explicitly *not* - /// recovered, however it reports right now. + /// 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, @@ -2370,13 +2325,13 @@ impl SessionInner { return Err(EngineError::Connection("closed".into())); } - // The settle window only gates the "nothing ever broke" branch below; a - // transport that proves recovery by renegotiating is accepted immediately. + // 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_negotiation, s.publisher_disconnect)), + snapshot.as_ref().map(|s| (s.publisher_connected, s.publisher_disconnect)), settled, ); @@ -2394,7 +2349,7 @@ impl SessionInner { pc, snapshot .as_ref() - .map(|s| (s.subscriber_negotiation, s.subscriber_disconnect)), + .map(|s| (s.subscriber_connected, s.subscriber_disconnect)), settled, ), }; @@ -2423,12 +2378,8 @@ impl SessionInner { } } - /// Decide whether `pc` has recovered. - /// - /// `generations` is `None` on the initial-connect path, where `Connected` is taken at - /// face value because there is no earlier state to confuse it with. On the resume path - /// it carries the `(negotiation, disconnect)` generations sampled before the resume - /// started. See [`Self::wait_pc_reconnected_with_snapshot`] for the rationale. + /// `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)>, @@ -2436,7 +2387,7 @@ impl SessionInner { ) -> bool { recovery_decision( pc.is_connected(), - pc.negotiation_generation(), + pc.connected_generation(), pc.disconnect_generation(), generations, settled, @@ -2711,16 +2662,16 @@ pub fn handle_remote_dt_packets(dc: &DataChannel, emitter: WeakUnboundedSender, settled: bool, ) -> bool { @@ -2728,22 +2679,21 @@ fn recovery_decision( return false; } - let Some((snap_negotiation, snap_disconnect)) = generations else { + let Some((snap_connected, snap_disconnect)) = generations else { return true; // initial connect: reaching Connected is the whole contract }; - if negotiation != snap_negotiation { - return true; // renegotiated since the resume began — positive proof of a live path + if connected_generation != snap_connected { + return true; // entered Connected on this attempt, so ICE and DTLS completed } - if disconnect != snap_disconnect { - return false; // it broke, and has not renegotiated since; `Connected` is not trustworthy + if disconnect_generation != snap_disconnect { + return false; // dropped and not back since; the current state is stale } - // Never broke and never renegotiated. Most likely a signal-only failure where the media - // plane was fine throughout — but it is also what a not-yet-timed-out dead transport - // looks like, so only accept once the settle window has passed and ICE has had a chance - // to notice. + // 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 } @@ -2778,70 +2728,62 @@ make_rtc_config!(make_rtc_config_reconnect, proto::ReconnectResponse); mod tests { use super::{parse_sdp_max_message_size, recovery_decision, DEFAULT_MAX_MESSAGE_SIZE}; - /// The generations sampled at the start of a resume, for readability below. + /// `(connected, disconnect)` counts as sampled before a resume, for readability below. const SNAPSHOT: Option<(u32, u32)> = Some((7, 3)); - /// The defect this fix exists for. - /// - /// After a node failure the subscriber's PeerConnection keeps reporting `Connected` - /// until ICE times out, so a resume that trusts the connection state declares success - /// for a transport that is in fact dead — emitting `Resumed`, and hence - /// `RoomEvent::Reconnected` plus `ConnectionState::Connected`, for a session that never - /// receives media again. The transport left `Connected` at some point (recorded in the - /// disconnect generation) and has not renegotiated since, so it must not be accepted no - /// matter what it currently reports, and no matter how long we have settled for. + /// 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_disconnect_is_not_recovery() { + fn stale_connected_after_a_drop_is_not_recovery() { assert!(!recovery_decision( /* connected= */ true, - /* negotiation= */ 7, // unchanged: nothing renegotiated - /* disconnect= */ 4, // bumped: it left Connected at some point + /* connected_generation= */ 7, // unchanged: never re-entered Connected + /* disconnect_generation= */ 4, // bumped: it dropped at some point SNAPSHOT, /* settled= */ true, )); } - /// A completed renegotiation is positive proof of a live path, so it is accepted - /// immediately — a genuine recovery never waits out the settle window. + /// 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 renegotiation_is_accepted_immediately() { + fn reconnection_is_accepted_immediately() { assert!(recovery_decision( /* connected= */ true, - /* negotiation= */ 8, // the new node's offer/answer landed - /* disconnect= */ - 4, // it did break — irrelevant, it has since renegotiated + /* 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 )); } - /// The signal-only blip: the media plane never broke, so the pre-existing connection is - /// still good. Accepted, but only once the settle window has given ICE a chance to - /// notice a path that died silently. + /// 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 unbroken_connection_is_accepted_only_after_settling() { - let unbroken = |settled| { - recovery_decision( - /* connected= */ true, /* negotiation= */ 7, /* disconnect= */ 3, - SNAPSHOT, settled, - ) - }; + fn untouched_connection_is_accepted_only_after_settling() { + let untouched = |settled| recovery_decision(true, 7, 3, SNAPSHOT, settled); - assert!(!unbroken(false), "must not accept before ICE could notice a dead path"); - assert!(unbroken(true)); + assert!(!untouched(false), "must not accept before ICE could notice a dead path"); + assert!(untouched(true)); } - /// A transport that is not currently connected is never recovered, whatever its - /// generations say. #[test] - fn disconnected_transport_is_never_recovered() { - for negotiation in [7, 8] { + fn transport_not_currently_connected_is_never_recovered() { + for connected_generation in [7, 8] { for settled in [false, true] { - assert!(!recovery_decision(false, negotiation, 3, SNAPSHOT, settled)); + assert!(!recovery_decision(false, connected_generation, 3, SNAPSHOT, settled)); } } } - /// The initial-connect path has no earlier state to be confused by, so reaching - /// `Connected` is the whole contract and must not be gated on renegotiation. + /// 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));