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..a933a12f8 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() } @@ -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,6 +812,10 @@ 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 @@ -818,6 +823,18 @@ 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 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") 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 }