Handle data track SID reassignment

.changeset/handle_data_track_sid_reassignment.md

@@ -0,0 +1,7 @@
+---
+livekit-ffi: patch
+livekit-datatrack: patch
+livekit: patch
+---
+
+# Handle data track SID reassignment

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 };