Roll back the sender when publishing fails after AddTrack

.changes/publish-failure-rollback

@@ -0,0 +1 @@
+patch type="fixed" "Roll back the sender when publishing fails after the track was added on the server (e.g. audio frame-watcher timeout), so the SFU no longer keeps an orphaned track that the client cannot mute or unpublish"

Sources/LiveKit/Participant/LocalParticipant.swift

@@ -101,6 +101,7 @@ public class LocalParticipant: Participant, @unchecked Sendable {
             // Mark re-negotiation required...
             try await room.publisherShouldNegotiate()
         }
+        await track.set(transport: nil, rtpSender: nil)
 
         track._state.mutate { $0.rtpSenderForCodec.removeAll() }
 
@@ -506,13 +507,8 @@ extension LocalParticipant {
 
         log("[Publish/Backup] Additional video codec: \(videoCodec)...")
 
-        guard let track = localTrackPublication.track as? LocalVideoTrack else {
-            throw LiveKitError(.invalidState, message: "Track is nil")
-        }
-
-        if !videoCodec.isBackup {
-            throw LiveKitError(.invalidState, message: "Attempted to publish a non-backup video codec as backup")
-        }
+        guard let track = localTrackPublication.track as? LocalVideoTrack else { throw LiveKitError(.invalidState, message: "Track is nil") }
+        guard videoCodec.isBackup else { throw LiveKitError(.invalidState, message: "Attempted to publish a non-backup video codec as backup") }
 
         let publisher = try room.requirePublisher()
 
@@ -544,37 +540,42 @@ extension LocalParticipant {
         let sender = try await RTC.run {
             let transceiver = try publisher.addTransceiver(with: track.mediaTrack.raw, transceiverInit: transInit)
             try transceiver.set(preferredVideoCodec: videoCodec)
-            let sender = transceiver.sender
-            sender.set(degradationPreference: degradationPreference)
-            return RTCSender(sender)
+            transceiver.sender.set(degradationPreference: degradationPreference)
+            return RTCSender(transceiver.sender)
         }
         log("[Publish] Added transceiver...")
 
-        // Request a new track to the server
-        let trackInfo = try await room.signalClient.sendAddTrack(cid: sender.senderId,
-                                                                 name: track.name,
-                                                                 type: track.kind.toPBType(),
-                                                                 source: track.source.toPBType())
-        {
-            $0.sid = localTrackPublication.sid.stringValue
-            $0.simulcastCodecs = [
-                Livekit_SimulcastCodec.with { sc in
-                    sc.cid = sender.senderId
-                    sc.codec = videoCodec.name
-                },
-            ]
-
-            $0.layers = layers
-        }
+        do {
+            // Request a new track to the server
+            let trackInfo = try await room.signalClient.sendAddTrack(cid: sender.senderId,
+                                                                     name: track.name,
+                                                                     type: track.kind.toPBType(),
+                                                                     source: track.source.toPBType())
+            {
+                $0.sid = localTrackPublication.sid.stringValue
+                $0.simulcastCodecs = [
+                    Livekit_SimulcastCodec.with { sc in
+                        sc.cid = sender.senderId
+                        sc.codec = videoCodec.name
+                    },
+                ]
+
+                $0.layers = layers
+            }
 
-        log("[Publish] server responded trackInfo: \(trackInfo)")
+            log("[Publish] server responded trackInfo: \(trackInfo)")
 
-        await RTC.run { sender.raw._set(subscribedQualities: subscribedCodec.qualities) }
+            await RTC.run { sender.raw._set(subscribedQualities: subscribedCodec.qualities) }
 
-        // Attach multi-codec sender...
-        track._state.mutate { $0.rtpSenderForCodec[videoCodec] = sender }
+            // Attach multi-codec sender...
+            track._state.mutate { $0.rtpSenderForCodec[videoCodec] = sender }
 
-        try await room.publisherShouldNegotiate()
+            try await room.publisherShouldNegotiate()
+        } catch {
+            track._state.mutate { $0.rtpSenderForCodec[videoCodec] = nil }
+            await rollback(sender: sender, publisher: publisher, room: room)
+            throw error
+        }
     }
 }
 
@@ -811,13 +812,29 @@ extension LocalParticipant {
             return publication
         } catch {
             log("[publish] failed \(track), error: \(error)", .error)
+            if let sender = track._state.read({ $0.transport === publisher ? $0.rtpSender : nil }) {
+                await rollback(sender: sender, publisher: publisher, room: room)
+                await track.set(transport: nil, rtpSender: nil)
+            }
             // Stop track when publish fails
             try await track.stop()
             // Rethrow
             throw error
         }
     }
 
+    // The server only learns about a track going away through renegotiation, so a publish that
+    // fails after the sender was attached has to detach it or the SFU keeps the track alive
+    // with no local publication to mute or unpublish it.
+    private func rollback(sender: RTCSender, publisher: Transport, room: Room) async {
+        do {
+            try await publisher.remove(track: sender)
+            try await room.publisherShouldNegotiate()
+        } catch {
+            log("[publish] failed to roll back sender, error: \(error)", .warning)
+        }
+    }
+
     private func checkPermissions(toPublish track: LocalTrack) throws {
         guard permissions.canPublish else {
             throw LiveKitError(.insufficientPermissions, message: "Participant does not have permission to publish")

Tests/LiveKitCoreTests/PublishTrackTests.swift

@@ -23,6 +23,29 @@ import LiveKitTestSupport
 
 @Suite(.serialized, .tags(.media, .e2e))
 struct PublishTrackTests {
+    @Test func publishFailureAfterAddTrackRollsBack() async throws {
+        try await TestEnvironment.withRooms([RoomTestingOptions(canPublish: true), RoomTestingOptions(canSubscribe: true)]) { rooms in
+            let publisher = rooms[0].localParticipant
+            let publisherIdentity = try #require(publisher.identity)
+            let remote = try #require(rooms[1].remoteParticipants[publisherIdentity])
+
+            let failedTrack = FrameStarvedAudioTrack()
+            await #expect(throws: LiveKitError.self) {
+                try await publisher.publish(audioTrack: failedTrack)
+            }
+            #expect(publisher.audioTracks.isEmpty)
+            #expect(failedTrack._state.rtpSender == nil)
+
+            // A retry must leave exactly one track on the server, not a second one next to an orphan.
+            let publication = try await publisher.publish(audioTrack: TestAudioTrack())
+            let deadline = Date().addingTimeInterval(15)
+            while Date() < deadline, Set(remote.audioTracks.map(\.sid)) != [publication.sid] {
+                try await Task.sleep(nanoseconds: 200_000_000)
+            }
+            #expect(Set(remote.audioTracks.map(\.sid)) == [publication.sid])
+        }
+    }
+
     @Test func publishWithoutPermissions() async throws {
         try await TestEnvironment.withRoom(RoomTestingOptions(canPublish: false)) { room in
             let audioTrack = await LocalAudioTrack.createTrack()

Tests/LiveKitTestSupport/Tracks.swift

@@ -205,6 +205,17 @@ public class TestAudioTrack: LocalAudioTrack, @unchecked Sendable {
     }
 }
 
+/// Fails where a real audio track fails when its engine never delivers a frame:
+/// after `AddTrack` and negotiation, before the publication exists.
+public final class FrameStarvedAudioTrack: TestAudioTrack, @unchecked Sendable {
+    override public func startWaitingForFrames() async throws {
+        // Give negotiation time to reach the server so it holds a live publication when
+        // the failure hits, as it does with the real 5s frame-watcher timeout.
+        try await Task.sleep(nanoseconds: 2_000_000_000)
+        throw LiveKitError(.timedOut, message: "No audio frames")
+    }
+}
+
 public class AudioTrackWatcher: AudioRenderer, @unchecked Sendable {
     public let id: String
     public var didRenderFirstFrame: Bool { _state.didRenderFirstFrame }