diff --git a/.changeset/mark_data_track_unpublished_after_republish_fail.md b/.changeset/mark_data_track_unpublished_after_republish_fail.md new file mode 100644 index 000000000..400df95b9 --- /dev/null +++ b/.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) diff --git a/livekit-datatrack/src/local/events.rs b/livekit-datatrack/src/local/events.rs index 9400e3f07..e8cfcc261 100644 --- a/livekit-datatrack/src/local/events.rs +++ b/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. diff --git a/livekit-datatrack/src/local/manager.rs b/livekit-datatrack/src/local/manager.rs index e485b82d7..6e74cd4a4 100644 --- a/livekit-datatrack/src/local/manager.rs +++ b/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 };