Mark data track unpublished after republish fail

.changeset/mark_data_track_unpublished_after_republish_fail.md

@@ -0,0 +1,9 @@
+---
+livekit: patch
+livekit-capture: patch
+livekit-datatrack: patch
+livekit-ffi: patch
+livekit-uniffi: patch
+---
+
+Mark data track unpublished after republish fail - #1537 (@ladvoc)

livekit-datatrack/src/local/events.rs

@@ -35,6 +35,7 @@ pub enum InputEvent {
     ///
     /// This must be sent after a full reconnect in order for existing publications
     /// to be recognized by the SFU. Each republished track will be assigned a new SID.
+    /// Tracks the SFU rejects are unpublished.
     ///
     RepublishTracks,
     /// Shutdown the manager and all associated tracks.

livekit-datatrack/src/local/manager.rs

@@ -221,9 +221,13 @@ impl Manager {
                     log::warn!("Track {} already active", event.handle);
                     return;
                 }
-                let Ok(updated_info) = event.result else {
-                    log::warn!("Republish failed for track {}", event.handle);
-                    return;
+                let updated_info = match event.result {
+                    Ok(info) => info,
+                    Err(err) => {
+                        log::warn!("Republish failed for track {}: {}", event.handle, err);
+                        _ = state_tx.send(PublishState::Unpublished);
+                        return;
+                    }
                 };
 
                 log::debug!("Track {} republished", event.handle);
@@ -496,7 +500,7 @@ impl ManagerInput {
 mod tests {
     use super::*;
     use crate::{
-        api::DataTrackSid,
+        api::{DataTrackSid, PushFrameErrorReason},
         e2ee::{EncryptedPayload, EncryptionError, EncryptionProvider},
         packet::Packet,
         utils::testing::expect_event,
@@ -752,6 +756,46 @@ mod tests {
         assert!(track.try_push(vec![0xFF].into()).is_ok());
     }
 
+    #[tokio::test]
+    async fn test_republish_rejected() {
+        let options = ManagerOptions { encryption_provider: None };
+        let (manager, input, mut output) = Manager::new(options);
+        tokio::spawn(manager.run());
+
+        let (result_tx, result_rx) = oneshot::channel();
+        let event = PublishRequest { options: DataTrackOptions::new("track"), result_tx };
+        input.send(event.into()).unwrap();
+
+        let event = expect_event!(output, OutputEvent::SfuPublishRequest);
+        let handle = event.handle;
+
+        let info = DataTrackInfo {
+            sid: RwLock::new(Faker.fake()).into(),
+            pub_handle: handle,
+            name: "track".into(),
+            uses_e2ee: false,
+            schema: None,
+            frame_encoding: None,
+        };
+        let event = SfuPublishResponse { handle, result: Ok(info) };
+        input.send(event.into()).unwrap();
+        let track = result_rx.await.unwrap().unwrap();
+
+        // Simulate reconnect
+        input.send(InputEvent::RepublishTracks).unwrap();
+        expect_event!(output, OutputEvent::SfuPublishRequest);
+
+        // SFU rejects the republish
+        let event = SfuPublishResponse { handle, result: Err(PublishError::NotAllowed) };
+        input.send(event.into()).unwrap();
+
+        // Track is unpublished
+        timeout(Duration::from_millis(500), track.wait_for_unpublish()).await.unwrap();
+        assert!(!track.is_published());
+        let err = track.try_push(vec![0xFF].into()).unwrap_err();
+        assert!(matches!(err.reason(), PushFrameErrorReason::TrackUnpublished));
+    }
+
     #[tokio::test]
     async fn test_query_published() {
         let options = ManagerOptions { encryption_provider: None };