From a2dfb9b6c4b64b5170734f94008d38ef11236dd5 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Wed, 8 Jul 2026 09:38:16 -0700 Subject: [PATCH 1/5] Handle SID reassignment --- livekit-datatrack/src/remote/manager.rs | 54 +++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/livekit-datatrack/src/remote/manager.rs b/livekit-datatrack/src/remote/manager.rs index ce1bb260f..55068c883 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,57 @@ 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, desc)) = 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 } = &*desc.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 descriptor = self.descriptors.remove(&old_sid).unwrap(); + *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); From e0745adba0bd8f12f64e084295ed8dcceb7cf3af Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Wed, 8 Jul 2026 09:38:47 -0700 Subject: [PATCH 2/5] Unit test SID reassignment --- livekit-datatrack/src/remote/manager.rs | 123 ++++++++++++++++++++++++ 1 file changed, 123 insertions(+) diff --git a/livekit-datatrack/src/remote/manager.rs b/livekit-datatrack/src/remote/manager.rs index 55068c883..1d88cb37a 100644 --- a/livekit-datatrack/src/remote/manager.rs +++ b/livekit-datatrack/src/remote/manager.rs @@ -767,6 +767,129 @@ 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, + sequence: 0, + frame_number: 0, + timestamp: Timestamp::from_ticks(0), + extensions: Extensions::default(), + }, + 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 }; From d711dafb71f062440c5c3a7cb1cfbfc24ce454a6 Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Wed, 8 Jul 2026 09:42:15 -0700 Subject: [PATCH 3/5] Changeset --- .changeset/handle_data_track_sid_reassignment.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 .changeset/handle_data_track_sid_reassignment.md 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 From 105c6eb4e35162149cf4c10234c564cb1324037b Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Wed, 8 Jul 2026 09:47:26 -0700 Subject: [PATCH 4/5] Spread fake fields in unit test --- livekit-datatrack/src/remote/manager.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/livekit-datatrack/src/remote/manager.rs b/livekit-datatrack/src/remote/manager.rs index 1d88cb37a..385e52b80 100644 --- a/livekit-datatrack/src/remote/manager.rs +++ b/livekit-datatrack/src/remote/manager.rs @@ -877,10 +877,8 @@ mod tests { header: Header { marker: FrameMarker::Single, track_handle: new_sub_handle, - sequence: 0, - frame_number: 0, - timestamp: Timestamp::from_ticks(0), extensions: Extensions::default(), + ..Faker.fake() }, payload: Bytes::from_static(&[1, 2, 3, 4, 5]), }; From 16742b4248992b8bd0cd3b0676e4633d12b375ba Mon Sep 17 00:00:00 2001 From: Jacob Gelman <3182119+ladvoc@users.noreply.github.com> Date: Wed, 8 Jul 2026 09:55:20 -0700 Subject: [PATCH 5/5] Avoid unwrap --- livekit-datatrack/src/remote/manager.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/livekit-datatrack/src/remote/manager.rs b/livekit-datatrack/src/remote/manager.rs index 385e52b80..f88fcfbbe 100644 --- a/livekit-datatrack/src/remote/manager.rs +++ b/livekit-datatrack/src/remote/manager.rs @@ -243,7 +243,7 @@ impl Manager { info: &DataTrackInfo, ) -> bool { // Publisher identity and pub handle are stable across republications. - let Some((old_sid, desc)) = self.descriptors.iter().find(|(_, desc)| { + 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 { @@ -252,7 +252,7 @@ impl Manager { // 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 } = &*desc.info; + 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; @@ -262,7 +262,9 @@ impl Manager { let new_sid = info.sid(); log::debug!("SID reassigned: {} -> {}", old_sid, new_sid); - let descriptor = self.descriptors.remove(&old_sid).unwrap(); + let Some(descriptor) = self.descriptors.remove(&old_sid) else { + return false; + }; *descriptor.info.sid.write().unwrap() = new_sid.clone(); match &descriptor.subscription {