diff --git a/.changeset/refresh-subscribed-codecs-after-publisher-answer.md b/.changeset/refresh-subscribed-codecs-after-publisher-answer.md new file mode 100644 index 000000000..199af9570 --- /dev/null +++ b/.changeset/refresh-subscribed-codecs-after-publisher-answer.md @@ -0,0 +1,9 @@ +--- +"client-sdk-android": patch +--- + +Keep dynacast-paused simulcast layers paused across publisher renegotiations. Previously a paused layer started to upload again after any renegotiation of the publisher connection. It stayed on until the next subscribed-quality change. Publishing a track, unpublishing one, a codec change, and an ICE restart all caused this. + +The server's answer to a publisher offer no longer carries the RFC 8853 pause markers. Applying that answer re-enables every layer the client paused. If the subscribed set does not change, the server sends no subscribed-codec update, so the pause was never restated. Directly after it applies a publisher answer, the SDK now re-applies the last known subscribed codecs. + +One frame per layer can still escape, because the encoders restart during the apply. This matches the JavaScript and Rust SDKs. diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt index c041ac663..cb8b35aa9 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt @@ -1183,6 +1183,7 @@ internal constructor( fun onUserPacket(packet: LivekitModels.UserPacket, kind: LivekitModels.DataPacket.Kind, encryptionType: LivekitModels.Encryption.Type) fun onStreamStateUpdate(streamStates: List) fun onSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate) + fun onRequestSubscribedCodecRefresh() {} fun onSubscriptionPermissionUpdate(subscriptionPermissionUpdate: LivekitRtc.SubscriptionPermissionUpdate) fun onSubscriptionError(subscriptionResponse: LivekitRtc.SubscriptionResponse) suspend fun onSignalConnected(isResume: Boolean) @@ -1251,7 +1252,7 @@ internal constructor( coroutineScope.launch { when (val outcome = publisher?.setRemoteDescription(sessionDescription, offerId).nullSafe()) { is Either.Left -> { - // do nothing. + listener?.onRequestSubscribedCodecRefresh() } is Either.Right -> { diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt index 42622a36e..cb0882454 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt @@ -1484,6 +1484,10 @@ constructor( localParticipant.handleSubscribedQualityUpdate(subscribedQualityUpdate) } + override fun onRequestSubscribedCodecRefresh() { + localParticipant.handleSubscribedCodecRefresh() + } + /** * @suppress */ diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt index 9ec1d98f5..158309819 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt @@ -1400,6 +1400,16 @@ internal constructor( } } + internal fun handleSubscribedCodecRefresh() { + if (!dynacast) { + return + } + + videoTrackPublications.forEach { (_, track) -> + (track as? LocalVideoTrack)?.refreshSubscribedCodecs() + } + } + private fun publishAdditionalCodecForTrack(track: LocalVideoTrack, codec: VideoCodec, options: VideoTrackPublishOptions) { val existingPublication = trackPublications[track.sid] ?: run { LKLog.w { "attempting to publish additional codec for non-published track?!" } diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt index 074e2ce08..9a6394858 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalVideoTrack.kt @@ -39,6 +39,7 @@ import io.livekit.android.util.FlowObservable import io.livekit.android.util.LKLog import io.livekit.android.util.flowDelegate import io.livekit.android.webrtc.peerconnection.RTCThreadToken +import io.livekit.android.webrtc.peerconnection.executeBlockingOnRTCThread import livekit.LivekitRtc import livekit.LivekitRtc.SubscribedCodec import livekit.org.webrtc.CameraVideoCapturer @@ -334,54 +335,58 @@ constructor( LKLog.i { "attempted to set publishing layer for disposed video track." } return } - try { - val parameters = sender.parameters ?: return - val encodings = parameters.encodings ?: return - var hasChanged = false - - if (encodings.firstOrNull()?.scalabilityMode != null) { - val encoding = encodings.first() - var maxQuality = ProtoVideoQuality.OFF - for (quality in qualities) { - if (quality.enabled && (maxQuality == ProtoVideoQuality.OFF || quality.quality.number > maxQuality.number)) { - maxQuality = quality.quality + // Both the dynacast path and the publisher-answer refresh reach this from separate + // coroutine scopes on the same multi-threaded dispatcher, so the read-modify-write of + // sender.parameters is serialized here rather than at either call site. + executeBlockingOnRTCThread(rtcThreadToken) { + try { + val parameters = sender.parameters ?: return@executeBlockingOnRTCThread + val encodings = parameters.encodings ?: return@executeBlockingOnRTCThread + var hasChanged = false + + if (encodings.firstOrNull()?.scalabilityMode != null) { + val encoding = encodings.first() + var maxQuality = ProtoVideoQuality.OFF + for (quality in qualities) { + if (quality.enabled && (maxQuality == ProtoVideoQuality.OFF || quality.quality.number > maxQuality.number)) { + maxQuality = quality.quality + } } - } - if (maxQuality == ProtoVideoQuality.OFF) { - if (encoding.active) { - LKLog.v { "setting svc track to disabled" } - encoding.active = false + if (maxQuality == ProtoVideoQuality.OFF) { + if (encoding.active) { + LKLog.v { "setting svc track to disabled" } + encoding.active = false + hasChanged = true + } + } else if (!encoding.active) { + LKLog.v { "setting svc track to enabled" } + encoding.active = true hasChanged = true } - } else if (!encoding.active) { - LKLog.v { "setting svc track to enabled" } - encoding.active = true - hasChanged = true - } - } else { - // simulcast dynacast encodings - for (quality in qualities) { - val rid = EncodingUtils.ridForVideoQuality(quality.quality) ?: continue - val encoding = encodings.firstOrNull { it.rid == rid } - // use low quality layer settings for non-simulcasted streams - ?: encodings.takeIf { it.size == 1 && quality.quality == ProtoVideoQuality.LOW }?.first() - ?: continue - if (encoding.active != quality.enabled) { - hasChanged = true - encoding.active = quality.enabled - LKLog.v { "setting layer ${quality.quality} to ${quality.enabled}" } + } else { + // simulcast dynacast encodings + for (quality in qualities) { + val rid = EncodingUtils.ridForVideoQuality(quality.quality) ?: continue + val encoding = encodings.firstOrNull { it.rid == rid } + // use low quality layer settings for non-simulcasted streams + ?: encodings.takeIf { it.size == 1 && quality.quality == ProtoVideoQuality.LOW }?.first() + ?: continue + if (encoding.active != quality.enabled) { + hasChanged = true + encoding.active = quality.enabled + LKLog.v { "setting layer ${quality.quality} to ${quality.enabled}" } + } } } - } - if (hasChanged) { - // This refreshes the native code with the new information - sender.parameters = parameters + if (hasChanged) { + // This refreshes the native code with the new information + sender.parameters = parameters + } + } catch (e: Exception) { + LKLog.w(e) { "Exception caught while setting publishing layers." } } - } catch (e: Exception) { - LKLog.w(e) { "Exception caught while setting publishing layers." } - return } } @@ -430,6 +435,18 @@ constructor( return newCodecs } + /** + * Re-applies the most recently received subscribed codecs. + * + * Applying a publisher answer re-enables simulcast layers that dynacast had paused, and the + * server only sends a subscribed codec update when the set changes, so the pause is restated + * here instead. + */ + internal fun refreshSubscribedCodecs() { + val codecs = subscribedCodecs ?: return + setPublishingCodecs(codecs) + } + internal fun addSimulcastTrack(codec: VideoCodec, encodings: List): SimulcastTrackInfo? { if (this.simulcastCodecs.containsKey(codec)) { LKLog.w { "$codec already added, skipping." } diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt index f547f899f..18ada8460 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockRtpSender.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023-2025 LiveKit, Inc. + * Copyright 2023-2026 LiveKit, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,13 +25,16 @@ import org.mockito.kotlin.whenever import java.util.UUID object MockRtpSender { - fun create(id: String = "sender_id"): RtpSender { + fun create( + id: String = "sender_id", + encodings: List = emptyList(), + ): RtpSender { var rtpParameters: RtpParameters = MockRtpParameters( transactionId = UUID.randomUUID().toString(), degradationPreference = null, rtcp = MockRtpParameters.MockRtcp("", false), headerExtensions = mutableListOf(), - encodings = mutableListOf(), + encodings = encodings.toMutableList(), codecs = mutableListOf(), ) return Mockito.mock(RtpSender::class.java).apply { diff --git a/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt b/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt index c3c373424..d48991142 100644 --- a/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt +++ b/livekit-android-test/src/main/java/livekit/org/webrtc/MockRtpTransceiver.kt @@ -1,5 +1,5 @@ /* - * Copyright 2023-2025 LiveKit, Inc. + * Copyright 2023-2026 LiveKit, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -41,7 +41,7 @@ object MockRtpTransceiver { when (direction) { RtpTransceiverDirection.SEND_RECV, RtpTransceiverDirection.SEND_ONLY -> { - val sender = MockRtpSender.create(id = id) + val sender = MockRtpSender.create(id = id, encodings = init.sendEncodings) Mockito.`when`(mock.sender) .then { sender } } diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt index 43869ea83..244efa6fd 100644 --- a/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt +++ b/livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt @@ -479,6 +479,96 @@ class LocalParticipantMockE2ETest : MockE2ETest() { ) } + /** + * Disables the high layer for the track's publishing codec, leaving the lower layers enabled. + */ + private fun receiveSubscribedCodecUpdateDisablingHighLayer(trackSid: String) { + wsFactory.receiveMessage( + with(LivekitRtc.SignalResponse.newBuilder()) { + subscribedQualityUpdate = with(LivekitRtc.SubscribedQualityUpdate.newBuilder()) { + this.trackSid = trackSid + addAllSubscribedCodecs( + listOf( + with(SubscribedCodec.newBuilder()) { + codec = VideoCodec.VP8.codecName + addAllQualities( + listOf( + LivekitModels.VideoQuality.HIGH to false, + LivekitModels.VideoQuality.MEDIUM to true, + LivekitModels.VideoQuality.LOW to true, + ).map { (quality, enabled) -> + SubscribedQuality.newBuilder() + .setQuality(quality) + .setEnabled(enabled) + .build() + }, + ) + build() + }, + ), + ) + build() + } + build().toOkioByteString() + }, + ) + } + + /** + * Answers any publisher offer, echoing back the offer id so the answer is not discarded as stale. + */ + private fun answerPublisherOffers() { + wsFactory.registerSignalRequestHandler { request -> + if (request.hasOffer()) { + wsFactory.receiveMessage( + with(LivekitRtc.SignalResponse.newBuilder()) { + answer = with(LivekitRtc.SessionDescription.newBuilder()) { + sdp = "remote_answer" + type = "answer" + id = request.offer.id + build() + } + build() + }, + ) + true + } else { + false + } + } + } + + @Test + fun publisherAnswerRestatesPausedLayers() = runTest { + connect() + room.dynacast = true + answerPublisherOffers() + + room.localParticipant.publishVideoTrack(track = createLocalTrack(width = 1280, height = 720)) + testScheduler.advanceUntilIdle() + + receiveSubscribedCodecUpdateDisablingHighLayer( + room.localParticipant.videoTrackPublications.first().first.sid, + ) + testScheduler.advanceUntilIdle() + + val encodings = getPublisherPeerConnection().transceivers.first().sender.parameters.encodings + val highEncoding = encodings.first { it.rid == "f" } + val lowEncoding = encodings.first { it.rid == "q" } + assertFalse(highEncoding.active) + assertTrue(lowEncoding.active) + + // The SFU's answer no longer carries the RFC 8853 pause markers, so applying it re-enables + // the layer that dynacast paused. + highEncoding.active = true + + getPublisherPeerConnection().observer?.onRenegotiationNeeded() + testScheduler.advanceUntilIdle() + + assertFalse(highEncoding.active) + assertTrue(lowEncoding.active) + } + private fun createLocalTrack( width: Int = 1280, height: Int = 720,