Ensure participant disconnects are synthesized after connection resume (#1250)

### Before you submit your PR

Make sure the following is true before submitting your PR:

- [ ] I have read the [contributing
guidelines](https://github.com/livekit/rust-sdks/blob/main/CONTRIBUTING.md)
and validated that this PR will be accepted.
- [ ] I have read and followed the principles regarding breaking
changes, testing, and code quality.

### PR description

Describe the changes in this PR. Explain what the PR is meant to solve
and how to reproduce the issue in the first place.

### Breaking changes

If this PR introduces breaking changes, list them here and document the
rationale for introducing such a change.

### MSRV

If the PR modifies the crate's MSRV (Minimum Supported Rust Version),
document it here.

### Testing

Ideally, unit test the code you add, but ensure you're not repeating
existing test cases. Use as many already written scaffolding, utilities
as possible; write your own, when needed. If external services, APIs,
tokens are required (e.g., running an LK server instance), provide the
necessary information. Make sure your tests perform useful,
context-aware assertions and do not simply emulate "happy paths".

### Async

We want the project to be runtime-agnostic, so please reuse what's
already in
[livekit-runtime](https://github.com/livekit/rust-sdks/blob/main/livekit-runtime/)
and feel free to add anything missing. It's ok to use Tokio directly,
when writing unit tests, if necessary. When testing, do not use
artificial delays for the state to "catch up"; instead, respect the
event flow and subscribe properly using channels or other mechanisms.

Author: lukasIO

.changeset/ensure_participant_disconnects_are_synthesized_after_connect.md

@@ -0,0 +1,6 @@
+---
+livekit: patch
+livekit-ffi: patch
+---
+
+Ensure participant disconnects are synthesized after connection resume - #1250 (@lukasIO)

livekit/specs/signalling-reconnection.allium

@@ -565,6 +565,12 @@ rule ResumeDrainsQueue {
     ensures: ResumeAttemptSucceeded(engine)
     @guidance
         -- Step 6: the resume has fully recovered; drain the signal queue.
+        -- spec<->code: `resume_finalize` also hands the Room (external
+        -- boundary) the participant identities seen since the resume began —
+        -- a superset of the server's post-resume snapshot — and the Room
+        -- synthesizes disconnects for known participants absent from it,
+        -- whose DISCONNECTED update died with the previous connection.
+        -- Participant state itself is below this spec's altitude.
 }
 
 -- === Engine: attempt outcomes ===============================================

livekit/src/room/mod.rs

@@ -37,7 +37,12 @@ use livekit_runtime::JoinHandle;
 use parking_lot::RwLock;
 pub use proto::DisconnectReason;
 use proto::SignalTarget;
-use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration};
+use std::{
+    collections::{HashMap, HashSet},
+    fmt::Debug,
+    sync::Arc,
+    time::Duration,
+};
 use thiserror::Error;
 use tokio::sync::{
     broadcast,
@@ -847,6 +852,15 @@ impl Room {
         self.inner.rtc_engine.fail_transport_during_next_resume();
     }
 
+    /// Test-only: drop incoming DISCONNECTED participant entries, simulating an
+    /// SFU that fails to (re)deliver disconnect updates (e.g. a resume served
+    /// without the previous connection's state). Exercises the resume-time
+    /// participant reconciliation.
+    #[cfg(feature = "__lk-e2e-test")]
+    pub fn drop_disconnected_updates(&self, enabled: bool) {
+        self.inner.rtc_engine.drop_disconnected_updates(enabled);
+    }
+
     pub async fn get_stats(&self) -> EngineResult<SessionStats> {
         self.inner.rtc_engine.get_stats().await
     }
@@ -975,6 +989,9 @@ impl RoomSession {
     async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
         match event {
             EngineEvent::ParticipantUpdate { updates } => self.handle_participant_update(updates),
+            EngineEvent::ParticipantReconcile { seen_identities } => {
+                self.reconcile_absent_participants(seen_identities.into_iter())
+            }
             EngineEvent::MediaTrack { track, stream, transceiver } => {
                 self.handle_media_track(track, stream, transceiver)
             }
@@ -1132,6 +1149,32 @@ impl RoomSession {
         true
     }
 
+    /// Synthesize the disconnection of every known remote participant missing
+    /// from `present`, for syncs where the server's participant list is
+    /// authoritative (post-resume reconcile, room move): a participant who
+    /// left while our signal link was down never got its DISCONNECTED update
+    /// delivered to us, and would otherwise stay in the room forever.
+    fn reconcile_absent_participants(
+        self: &Arc<Self>,
+        present: impl Iterator<Item = ParticipantIdentity>,
+    ) {
+        let present: HashSet<ParticipantIdentity> = present.collect();
+        let missing: Vec<RemoteParticipant> = self
+            .remote_participants
+            .read()
+            .values()
+            .filter(|p| !present.contains(&p.identity()))
+            .cloned()
+            .collect();
+        for participant in missing {
+            log::info!(
+                "synthesizing disconnect for absent participant: {}",
+                participant.identity()
+            );
+            self.clone().handle_participant_disconnect(participant);
+        }
+    }
+
     /// Update the participants inside a Room.
     /// It'll create, update or remove a participant
     /// It also update the participant tracks.
@@ -1536,6 +1579,11 @@ impl RoomSession {
                 participants: vec![Participant::Local(self.local_participant.clone())],
             });
         }
+        // Participants we knew from the old room that are absent from the
+        // moved-to room have left it.
+        self.reconcile_absent_participants(
+            moved.other_participants.iter().map(|pi| pi.identity.clone().into()),
+        );
         self.handle_participant_update(moved.other_participants);
         if let Some(room) = moved.room {
             self.handle_room_update(room);

livekit/src/rtc_engine/mod.rs

@@ -18,7 +18,7 @@ use livekit_datatrack::backend as dt;
 use livekit_protocol as proto;
 use livekit_runtime::JoinHandle;
 use parking_lot::{RwLock, RwLockReadGuard};
-use std::{borrow::Cow, fmt::Debug, sync::Arc, time::Duration};
+use std::{borrow::Cow, collections::HashSet, fmt::Debug, sync::Arc, time::Duration};
 use thiserror::Error;
 use tokio::sync::{
     mpsc, oneshot, Notify, RwLock as AsyncRwLock, RwLockReadGuard as AsyncRwLockReadGuard,
@@ -114,6 +114,12 @@ pub enum EngineEvent {
     ParticipantUpdate {
         updates: Vec<proto::ParticipantInfo>,
     },
+    /// A signal resume fully recovered; any known participant whose identity
+    /// is not in `seen_identities` left the room while the signal link was
+    /// down and its disconnection must be synthesized.
+    ParticipantReconcile {
+        seen_identities: HashSet<ParticipantIdentity>,
+    },
     MediaTrack {
         track: MediaStreamTrack,
         stream: MediaStream,
@@ -437,6 +443,13 @@ impl RtcEngine {
             .fail_transport_during_next_resume
             .store(true, std::sync::atomic::Ordering::Release);
     }
+
+    /// Test-only: drop incoming DISCONNECTED participant entries on the current
+    /// session, simulating an SFU that fails to (re)deliver disconnect updates.
+    #[cfg(feature = "__lk-e2e-test")]
+    pub fn drop_disconnected_updates(&self, enabled: bool) {
+        self.session().drop_disconnected_updates(enabled);
+    }
 }
 
 impl EngineInner {
@@ -1156,6 +1169,15 @@ impl EngineInner {
         // has fully recovered, so deferred subscription updates / mutes / etc.
         // should now reach the server. Mirrors `client.setReconnected()`.
         session.signal_client().set_reconnected().await;
+
+        // Anyone who left while the signal link was down never got their
+        // DISCONNECTED update delivered to us; the room synthesizes those
+        // disconnects from the identities seen since the resume began. Sent
+        // from this task — the same producer that sends `Resumed` next — so
+        // they reach the application before `Reconnected`.
+        if let Some(seen_identities) = session.finish_resume() {
+            let _ = self.engine_tx.send(EngineEvent::ParticipantReconcile { seen_identities });
+        }
         Ok(())
     }
 }

livekit/src/rtc_engine/rtc_session.rs

@@ -401,6 +401,21 @@ struct SessionInner {
 
     participant_info: SessionParticipantInfo,
 
+    /// `Some` while a signal resume is in flight: accumulates the identity of
+    /// every participant mentioned in `Update`s since the resume began. The
+    /// server sends a full participant snapshot right after the
+    /// `ReconnectResponse` but may interleave delayed/batched updates around
+    /// it, so no single `Update` is identifiable as the snapshot — the union
+    /// is what's guaranteed to cover everyone still in the room once the
+    /// resume settles (see [`RtcSession::finish_resume`]).
+    resume_seen_identities: Mutex<Option<HashSet<ParticipantIdentity>>>,
+
+    /// Test-only: drop incoming DISCONNECTED participant entries, simulating
+    /// an SFU that fails to (re)deliver disconnect updates. Lets tests
+    /// exercise the resume-time participant reconciliation deterministically.
+    #[cfg(feature = "__lk-e2e-test")]
+    drop_disconnected_updates: AtomicBool,
+
     dc_emitter: mpsc::UnboundedSender<DataChannelEvent>,
 
     // Keep a strong reference to the subscriber datachannels,
@@ -620,6 +635,9 @@ impl RtcSession {
             next_packet_sequence: 1.into(),
             packet_rx_state: Mutex::new(TtlMap::new(RELIABLE_RECEIVED_STATE_TTL)),
             participant_info,
+            resume_seen_identities: Mutex::new(None),
+            #[cfg(feature = "__lk-e2e-test")]
+            drop_disconnected_updates: Default::default(),
             dc_emitter,
             sub_lossy_dc: Mutex::new(None),
             sub_reliable_dc: Mutex::new(None),
@@ -809,6 +827,24 @@ impl RtcSession {
         self.inner.restart_publisher().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
+    /// finished reconnecting, so this is a superset of the room's current
+    /// participants: any known participant absent from it left while the
+    /// signal link was down and its disconnection must be synthesized.
+    /// Returns `None` if no resume was in flight.
+    pub fn finish_resume(&self) -> Option<HashSet<ParticipantIdentity>> {
+        self.inner.resume_seen_identities.lock().take()
+    }
+
+    /// Test-only: drop incoming DISCONNECTED participant entries, simulating
+    /// an SFU that fails to (re)deliver disconnect updates.
+    #[cfg(feature = "__lk-e2e-test")]
+    pub fn drop_disconnected_updates(&self, enabled: bool) {
+        self.inner.drop_disconnected_updates.store(enabled, Ordering::Release);
+    }
+
     pub async fn wait_pc_connection(&self) -> EngineResult<()> {
         self.inner.wait_pc_connection().await
     }
@@ -1387,13 +1423,26 @@ impl SessionInner {
                 );
             }
             proto::signal_response::Message::Update(mut update) => {
+                #[cfg(feature = "__lk-e2e-test")]
+                // injecting faulty behaviour during a signal disconnect to ensure we can mimic
+                // losing/missing participant disconnect events after a resume
+                // for test_resume_synthesizes_disconnect_for_participant_that_left
+                if self.drop_disconnected_updates.load(Ordering::Acquire) {
+                    update.participants.retain(|pi| {
+                        pi.state != proto::participant_info::State::Disconnected as i32
+                    });
+                }
+
                 let local_participant_identity = self.participant_info.identity.as_str().into();
                 if let Ok(event) = dt::remote::event_from_participant_update(
                     &mut update,
                     local_participant_identity,
                 ) {
                     _ = self.emitter.send(SessionEvent::RemoteDataTrackInput(event.into()));
                 }
+                if let Some(seen) = self.resume_seen_identities.lock().as_mut() {
+                    seen.extend(update.participants.iter().map(|pi| pi.identity.clone().into()));
+                }
                 let _ = self
                     .emitter
                     .send(SessionEvent::ParticipantUpdate { updates: update.participants });
@@ -2140,6 +2189,10 @@ impl SessionInner {
     /// This reconnection if more seemless compared to the full reconnection implemented in
     /// ['RTCEngine']
     async fn restart(&self) -> EngineResult<proto::ReconnectResponse> {
+        // Start accumulating before the signal client reconnects: once
+        // `restart` returns, the resumed stream immediately delivers events
+        // (including the post-resume participant snapshot) on a concurrent task.
+        *self.resume_seen_identities.lock() = Some(HashSet::new());
         let reconnect_response = self.signal_client.restart().await?;
         log::debug!("received reconnect response: {:?}", reconnect_response);