From b89405afcb62cec55d9a867e48aa4dc6575b08ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Fri, 28 Aug 2026 11:28:37 +0200 Subject: [PATCH 1/3] Roll back the sender when publishing fails after AddTrack Co-Authored-By: Claude Fable 5 --- .changes/publish-failure-rollback | 1 + .../Participant/LocalParticipant.swift | 73 ++++++++++++------- .../LiveKitCoreTests/PublishTrackTests.swift | 23 ++++++ Tests/LiveKitTestSupport/Tracks.swift | 11 +++ 4 files changed, 80 insertions(+), 28 deletions(-) create mode 100644 .changes/publish-failure-rollback diff --git a/.changes/publish-failure-rollback b/.changes/publish-failure-rollback new file mode 100644 index 000000000..1d4b33e9d --- /dev/null +++ b/.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" diff --git a/Sources/LiveKit/Participant/LocalParticipant.swift b/Sources/LiveKit/Participant/LocalParticipant.swift index e29599a12..df195115b 100644 --- a/Sources/LiveKit/Participant/LocalParticipant.swift +++ b/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() } @@ -509,10 +510,7 @@ extension LocalParticipant { 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 videoCodec.isBackup else { throw LiveKitError(.invalidState, message: "Attempted to publish a non-backup video codec as backup") } let publisher = try room.requirePublisher() @@ -544,37 +542,41 @@ 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 { + try? await publisher.remove(track: sender) + throw error + } } } @@ -811,6 +813,7 @@ extension LocalParticipant { return publication } catch { log("[publish] failed \(track), error: \(error)", .error) + await rollbackPublish(of: track, publisher: publisher, room: room) // Stop track when publish fails try await track.stop() // Rethrow @@ -818,6 +821,20 @@ extension LocalParticipant { } } + // 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 rollbackPublish(of track: LocalTrack, publisher: Transport, room: Room) async { + guard let sender = track._state.read({ $0.transport === publisher ? $0.rtpSender : nil }) else { return } + do { + try await publisher.remove(track: sender) + try await room.publisherShouldNegotiate() + } catch { + log("[publish] failed to roll back sender, error: \(error)", .warning) + } + await track.set(transport: nil, rtpSender: nil) + } + private func checkPermissions(toPublish track: LocalTrack) throws { guard permissions.canPublish else { throw LiveKitError(.insufficientPermissions, message: "Participant does not have permission to publish") diff --git a/Tests/LiveKitCoreTests/PublishTrackTests.swift b/Tests/LiveKitCoreTests/PublishTrackTests.swift index e3b602f84..bbb3f03b9 100644 --- a/Tests/LiveKitCoreTests/PublishTrackTests.swift +++ b/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() diff --git a/Tests/LiveKitTestSupport/Tracks.swift b/Tests/LiveKitTestSupport/Tracks.swift index 2a4ae9ccc..124f8ac30 100644 --- a/Tests/LiveKitTestSupport/Tracks.swift +++ b/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 } From 0c1921377cb55a6501f03ba94a6f03ff6a16d5d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Fri, 28 Aug 2026 11:44:46 +0200 Subject: [PATCH 2/3] Clear the backup-codec sender map entry when its publish rolls back Co-Authored-By: Claude Fable 5 --- Sources/LiveKit/Participant/LocalParticipant.swift | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/Sources/LiveKit/Participant/LocalParticipant.swift b/Sources/LiveKit/Participant/LocalParticipant.swift index df195115b..b5edf8939 100644 --- a/Sources/LiveKit/Participant/LocalParticipant.swift +++ b/Sources/LiveKit/Participant/LocalParticipant.swift @@ -507,9 +507,7 @@ extension LocalParticipant { log("[Publish/Backup] Additional video codec: \(videoCodec)...") - guard let track = localTrackPublication.track as? LocalVideoTrack else { - throw LiveKitError(.invalidState, message: "Track is nil") - } + 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() @@ -574,6 +572,7 @@ extension LocalParticipant { try await room.publisherShouldNegotiate() } catch { + track._state.mutate { $0.rtpSenderForCodec[videoCodec] = nil } try? await publisher.remove(track: sender) throw error } From a751c6e18c004cd2ae2038b2bba2de4be19b4df6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:20:56 +0200 Subject: [PATCH 3/3] Renegotiate when a backup-codec publish rolls back its sender Co-Authored-By: Claude Fable 5 --- Sources/LiveKit/Participant/LocalParticipant.swift | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/Sources/LiveKit/Participant/LocalParticipant.swift b/Sources/LiveKit/Participant/LocalParticipant.swift index b5edf8939..a933a12f8 100644 --- a/Sources/LiveKit/Participant/LocalParticipant.swift +++ b/Sources/LiveKit/Participant/LocalParticipant.swift @@ -573,7 +573,7 @@ extension LocalParticipant { try await room.publisherShouldNegotiate() } catch { track._state.mutate { $0.rtpSenderForCodec[videoCodec] = nil } - try? await publisher.remove(track: sender) + await rollback(sender: sender, publisher: publisher, room: room) throw error } } @@ -812,7 +812,10 @@ extension LocalParticipant { return publication } catch { log("[publish] failed \(track), error: \(error)", .error) - await rollbackPublish(of: track, publisher: publisher, room: room) + 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 @@ -823,15 +826,13 @@ extension LocalParticipant { // 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 rollbackPublish(of track: LocalTrack, publisher: Transport, room: Room) async { - guard let sender = track._state.read({ $0.transport === publisher ? $0.rtpSender : nil }) else { return } + 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) } - await track.set(transport: nil, rtpSender: nil) } private func checkPermissions(toPublish track: LocalTrack) throws {