diff --git a/.changeset/handle_data_track_sid_reassignment.md b/.changeset/handle_data_track_sid_reassignment.md new file mode 100644 index 000000000..3f2134d8a --- /dev/null +++ b/.changeset/handle_data_track_sid_reassignment.md @@ -0,0 +1,7 @@ +--- +livekit-ffi: patch +livekit-datatrack: patch +livekit: patch +--- + +# Handle data track SID reassignment diff --git a/livekit-datatrack/src/remote/manager.rs b/livekit-datatrack/src/remote/manager.rs index ce1bb260f..f88fcfbbe 100644 --- a/livekit-datatrack/src/remote/manager.rs +++ b/livekit-datatrack/src/remote/manager.rs @@ -179,6 +179,9 @@ impl Manager { if self.descriptors.contains_key(&sid) { continue; } + if self.handle_sid_reassigned(&publisher_identity, &info).await { + continue; + } self.handle_track_published(publisher_identity.clone(), info).await; } } @@ -229,6 +232,59 @@ impl Manager { _ = self.event_out_tx.send(TrackPublished { track }.into()).await; } + /// Detects and handles SID reassignment, which occurs when the publisher + /// republishes its tracks after a full reconnect. + /// + /// Returns `true` if an SID reassignment occurred, `false` otherwise. + /// + async fn handle_sid_reassigned( + &mut self, + publisher_identity: &str, + info: &DataTrackInfo, + ) -> bool { + // Publisher identity and pub handle are stable across republications. + let Some((old_sid, descriptor)) = self.descriptors.iter().find(|(_, desc)| { + desc.publisher_identity.as_ref() == publisher_identity + && desc.info.pub_handle == info.pub_handle + }) else { + return false; + }; + + // Invariant: other than SID, info should not have changed. + // TODO: consider refactoring to move SID out of info to allow for direct comparison. + let DataTrackInfo { sid: _, pub_handle: _, name, uses_e2ee } = &*descriptor.info; + if *name != info.name || *uses_e2ee != info.uses_e2ee { + log::warn!("Info mismatch for {}, treating as new publication", old_sid); + return false; + } + let old_sid = old_sid.clone(); + + let new_sid = info.sid(); + log::debug!("SID reassigned: {} -> {}", old_sid, new_sid); + + let Some(descriptor) = self.descriptors.remove(&old_sid) else { + return false; + }; + *descriptor.info.sid.write().unwrap() = new_sid.clone(); + + match &descriptor.subscription { + SubscriptionState::None => {} + SubscriptionState::Pending { .. } | SubscriptionState::Active { .. } => { + // The SFU does not carry subscriptions across a publisher's full + // reconnect; re-request the subscription under the new SID. + let event = SfuUpdateSubscription { sid: new_sid.clone(), subscribe: true }; + _ = self.event_out_tx.send(event.into()).await; + } + } + if let SubscriptionState::Active { sub_handle, .. } = &descriptor.subscription { + // Keep the routing index consistent until the SFU assigns a new handle + // (see `register_subscriber_handle`). + self.sub_handles.insert(*sub_handle, new_sid.clone()); + } + self.descriptors.insert(new_sid, descriptor); + true + } + fn on_set_pipeline_options(&mut self, event: SetPipelineOptions) { let Some(descriptor) = self.descriptors.get(&event.sid) else { log::warn!("Unknown track {}, cannot set pipeline options", event.sid); @@ -713,6 +769,127 @@ mod tests { } } + #[tokio::test] + async fn test_sid_reassignment_does_not_republish() { + let options = ManagerOptions { decryption_provider: None }; + let (manager, input, mut output) = Manager::new(options); + livekit_runtime::spawn(manager.run()); + + let pub_handle: Handle = Faker.fake(); + let old_sid: DataTrackSid = Faker.fake(); + let new_sid: DataTrackSid = Faker.fake(); + + // Simulate track published + let info = DataTrackInfo { + sid: RwLock::new(old_sid.clone()).into(), + pub_handle, + name: "test".into(), + uses_e2ee: false, + }; + let event = SfuPublicationUpdates { updates: HashMap::from([("id".into(), vec![info])]) }; + input.send(event.into()).unwrap(); + + let track = expect_event!(output, OutputEvent::TrackPublished).track; + assert_eq!(track.info().sid(), old_sid); + + // Simulate publisher full reconnect: same track, new SID + let info = DataTrackInfo { + sid: RwLock::new(new_sid.clone()).into(), + pub_handle, + name: "test".into(), + uses_e2ee: false, + }; + let event = SfuPublicationUpdates { updates: HashMap::from([("id".into(), vec![info])]) }; + input.send(event.into()).unwrap(); + + // Drain remaining events; no publish/unpublish should appear + input.send(InputEvent::Shutdown).unwrap(); + while let Some(event) = output.next().await { + assert!(!matches!( + event, + OutputEvent::TrackPublished(_) | OutputEvent::TrackUnpublished(_) + )); + } + assert_eq!(track.info().sid(), new_sid); + } + + #[tokio::test] + async fn test_sid_reassignment_resubscribes_active_subscription() { + let options = ManagerOptions { decryption_provider: None }; + let (manager, input, mut output) = Manager::new(options); + livekit_runtime::spawn(manager.run()); + + let pub_handle: Handle = Faker.fake(); + let old_sid: DataTrackSid = Faker.fake(); + let new_sid: DataTrackSid = Faker.fake(); + let old_sub_handle: Handle = Faker.fake(); + let new_sub_handle: Handle = Faker.fake(); + + // Simulate track published + let info = DataTrackInfo { + sid: RwLock::new(old_sid.clone()).into(), + pub_handle, + name: "test".into(), + uses_e2ee: false, + }; + let event = SfuPublicationUpdates { updates: HashMap::from([("id".into(), vec![info])]) }; + input.send(event.into()).unwrap(); + let track = expect_event!(output, OutputEvent::TrackPublished).track; + + // Subscribe to the track + let (result_tx, result_rx) = oneshot::channel(); + let event = SubscribeRequest { + sid: old_sid.clone(), + options: DataTrackSubscribeOptions::default(), + result_tx, + }; + input.send(event.into()).unwrap(); + expect_event!(output, OutputEvent::SfuUpdateSubscription); + + // Simulate SFU assigning subscriber handle + let event = SfuSubscriberHandles { mapping: HashMap::from([(old_sub_handle, old_sid)]) }; + input.send(event.into()).unwrap(); + + let mut frame_rx = + time::timeout(Duration::from_secs(1), result_rx).await.unwrap().unwrap().unwrap(); + + // Simulate publisher full reconnect: same track, new SID + let info = DataTrackInfo { + sid: RwLock::new(new_sid.clone()).into(), + pub_handle, + name: "test".into(), + uses_e2ee: false, + }; + let event = SfuPublicationUpdates { updates: HashMap::from([("id".into(), vec![info])]) }; + input.send(event.into()).unwrap(); + + // Manager should re-subscribe under the new SID + let event = expect_event!(output, OutputEvent::SfuUpdateSubscription); + assert!(event.subscribe); + assert_eq!(event.sid, new_sid); + assert_eq!(track.info().sid(), new_sid); + assert!(track.is_published()); + + // Simulate SFU assigning a new subscriber handle + let event = SfuSubscriberHandles { mapping: HashMap::from([(new_sub_handle, new_sid)]) }; + input.send(event.into()).unwrap(); + + // Frames received on the new handle reach the existing subscriber + let packet = Packet { + header: Header { + marker: FrameMarker::Single, + track_handle: new_sub_handle, + extensions: Extensions::default(), + ..Faker.fake() + }, + payload: Bytes::from_static(&[1, 2, 3, 4, 5]), + }; + input.send(InputEvent::PacketReceived(packet.serialize())).unwrap(); + + let frame = time::timeout(Duration::from_secs(1), frame_rx.recv()).await.unwrap().unwrap(); + assert_eq!(frame.payload.as_ref(), &[1, 2, 3, 4, 5]); + } + #[tokio::test] async fn test_subscribe_receives_frame() { let options = ManagerOptions { decryption_provider: None };