From f5218d0cebd982534693b7c9c0d8813c0e00f9bb Mon Sep 17 00:00:00 2001 From: Adrian Niculescu <15037449+adrian-niculescu@users.noreply.github.com> Date: Mon, 27 Jul 2026 20:25:47 +0300 Subject: [PATCH] Fixed setTrackEnabled orphaning track resources when cancelled before publish completes A cancellation landing in startForegroundService or publishVideoTrack unwound out of setTrackEnabled without any cleanup, abandoning the newly created screencast track and, when the bind had already succeeded, leaving ScreenCaptureService bound for the lifetime of the context. Each enable path now tears the unpublished track down in a finally block, so cancellation and publish failures release capture, track resources, and the service binding, and invoke the onStop callback exactly once, even when the platform ended the projection while publishing was in flight. A projection stop already dispatched on its callback thread can reach the track after that cleanup disposed it, where ScreenCapturerAndroid throws from stopCapture. LocalScreencastVideoTrack.stop() now tolerates losing that race instead of crashing the callback thread. publishTrackImpl negotiates the sender transceiver concurrently with the add track request, so a failed or cancelled request could leave a negotiated transceiver holding its sender and SDP m-section with no publication to unpublish. The sender is now detached whenever no publication is created. The rollback targets exactly the sender this attempt created on the transport that created it, so it cannot detach a concurrent publish of the same track, and the handle is recorded non-cancellably so a cancellation cannot discard the RTC-thread result. It runs only while that publisher is connected with no reconnect in flight, since a publish usually fails because the signal connection died and that same failure starts the reconnect which renegotiates the publisher. Neither connectionState nor the reconnect job has moved at the point a dying session fails its pending publishes, and a resume leave fails them without starting a reconnect at all, so a session identity is replaced at every boundary and each sender handle keeps the one that owned its transceiver. Matching that identity is what makes the decision independent of when a failed publish resumes: a soft reconnect keeps the publisher, so cleanup arriving after the next session is serving would otherwise mutate it. The check runs on the RTC thread immediately ahead of the mutation it guards. Stopping the transceiver and releasing its m-section is reserved for video, matching unpublishTrack. LocalScreencastVideoTrack also never released its SurfaceTextureHelper, so its capture thread outlived dispose() even on the normal teardown path. The helper is now registered with the track's CloseableManager, matching LocalVideoTrack. Fixes #985 --- .../set-track-enabled-cancellation-cleanup.md | 5 + .../java/io/livekit/android/room/RTCEngine.kt | 78 +++- .../room/participant/LocalParticipant.kt | 228 +++++++---- .../room/track/LocalScreencastVideoTrack.kt | 26 +- .../android/room/track/LocalVideoTrack.kt | 2 +- .../android/test/mock/MockPeerConnection.kt | 4 +- .../SetTrackEnabledCleanupMockE2ETest.kt | 356 ++++++++++++++++++ 7 files changed, 608 insertions(+), 91 deletions(-) create mode 100644 .changeset/set-track-enabled-cancellation-cleanup.md create mode 100644 livekit-android-test/src/test/java/io/livekit/android/room/participant/SetTrackEnabledCleanupMockE2ETest.kt diff --git a/.changeset/set-track-enabled-cancellation-cleanup.md b/.changeset/set-track-enabled-cancellation-cleanup.md new file mode 100644 index 000000000..e0ef688aa --- /dev/null +++ b/.changeset/set-track-enabled-cancellation-cleanup.md @@ -0,0 +1,5 @@ +--- +"client-sdk-android": patch +--- + +Fixed setTrackEnabled leaking track resources when cancelled before the track is published, including the sender negotiated for a failed publish, and LocalScreencastVideoTrack leaking its SurfaceTextureHelper on dispose. 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 19d3464f9..b67a4b5e2 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 @@ -133,6 +133,7 @@ internal constructor( } when (newVal) { ConnectionState.CONNECTED -> { + signalSessionState = SignalSessionState(ended = false) if (oldVal == ConnectionState.DISCONNECTED || oldVal == ConnectionState.CONNECTING) { LKLog.d { "primary ICE connected" } listener?.onEngineConnected() @@ -158,8 +159,17 @@ internal constructor( @Volatile internal var reconnectType: ReconnectType = ReconnectType.DEFAULT + + @Volatile private var reconnectingJob: Job? = null + /** + * Replaced at each signal session boundary. Sender handles retain the state that owned their + * transceiver, so cleanup cannot mutate a publisher after that session ends. + */ + @Volatile + private var signalSessionState = SignalSessionState(ended = true) + @Volatile private var fullReconnectOnNext = false @@ -424,10 +434,13 @@ internal constructor( internal suspend fun createSenderTransceiver( rtcTrack: MediaStreamTrack, transInit: RtpTransceiverInit, - ): RtpTransceiver? { - return publisher?.withPeerConnection { + ): SenderTransceiverHandle? { + val sessionState = signalSessionState + val publisher = publisher ?: return null + val transceiver = publisher.withPeerConnection { addTransceiver(rtcTrack, transInit) - } + } ?: return null + return SenderTransceiverHandle(publisher, transceiver, sessionState) } fun updateSubscriptionPermissions( @@ -505,6 +518,9 @@ internal constructor( } private fun abortPendingPublishTracks() { + // Ordered ahead of the resumes: each one unwinds a publish into cleanup that reads this, + // and on some paths the reconnect is only triggered later, by the socket closing. + signalSessionState = SignalSessionState(ended = true) synchronized(pendingTrackResolvers) { pendingTrackResolvers.values.forEach { it.resumeWithException(TrackException.PublishException("pending track aborted")) @@ -535,6 +551,7 @@ internal constructor( } val forceFullReconnect = fullReconnectOnNext fullReconnectOnNext = false + signalSessionState = SignalSessionState(ended = true) val job = coroutineScope.launch { var hasResumedOnce = false var hasReconnectedOnce = false @@ -1515,6 +1532,45 @@ internal constructor( } } + /** + * Detaches a sender transceiver whose publish attempt produced no publication. + * + * A publish fails most often because the signal connection died, and that same failure + * starts a reconnect which renegotiates this peer connection. Mutating its senders and + * transceivers alongside that negotiation races it, so the rollback is skipped unless the + * publisher is connected and idle. The sender retains its signal session state so an aborted + * publish cannot mutate the same publisher after a soft reconnect. + * + * @param stopTransceiver releases the transceiver and its SDP m-section for reuse. Reserved + * for video, matching the teardown an unpublish performs. + * @return whether the sender was detached, so the track can drop its reference to it. + */ + internal fun rollbackSenderTransceiver( + senderTransceiver: SenderTransceiverHandle, + stopTransceiver: Boolean, + ): Boolean { + val publisher = senderTransceiver.publisher + return runBlocking { + publisher.withPeerConnection { + val sessionState = signalSessionState + val publisherIsIdle = this@RTCEngine.publisher === publisher && + connectionState == ConnectionState.CONNECTED && + reconnectingJob?.isActive != true && + senderTransceiver.signalSessionState === sessionState && + !sessionState.ended + if (!publisherIsIdle) { + return@withPeerConnection false + } + val transceiver = senderTransceiver.transceiver + removeTrack(transceiver.sender) + if (stopTransceiver && !transceiver.isStopped) { + transceiver.stopInternal() + } + true + } ?: false + } + } + @VisibleForTesting fun getPublisherPeerConnection() = publisher!!.peerConnection @@ -1524,6 +1580,22 @@ internal constructor( subscriber!!.peerConnection } +/** + * Identity of one signal session, replaced at every session boundary. + * + * [ended] distinguishes a session that is finishing from one that is serving, for work that + * starts after the boundary and would otherwise match the current identity. + */ +internal class SignalSessionState( + val ended: Boolean, +) + +internal class SenderTransceiverHandle( + internal val publisher: PeerConnectionTransport, + internal val transceiver: RtpTransceiver, + internal val signalSessionState: SignalSessionState, +) + /** * @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 a1fe88614..882535be1 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 @@ -35,6 +35,7 @@ import io.livekit.android.room.ConnectionState import io.livekit.android.room.DefaultsManager import io.livekit.android.room.RTCEngine import io.livekit.android.room.Room +import io.livekit.android.room.SenderTransceiverHandle import io.livekit.android.room.TrackBitrateInfo import io.livekit.android.room.datastream.outgoing.OutgoingDataStreamManager import io.livekit.android.room.isSVCCodec @@ -64,11 +65,14 @@ import io.livekit.android.util.rethrowIfCancellationSignal import io.livekit.android.webrtc.sortVideoCodecPreferences import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext import livekit.LivekitModels import livekit.LivekitModels.AudioTrackFeature import livekit.LivekitModels.Codec @@ -86,7 +90,9 @@ import livekit.org.webrtc.SurfaceTextureHelper import livekit.org.webrtc.VideoCapturer import livekit.org.webrtc.VideoProcessor import java.util.Collections +import java.util.concurrent.atomic.AtomicBoolean import javax.inject.Named +import kotlin.coroutines.coroutineContext import kotlin.math.max import kotlin.math.min import kotlin.time.Duration @@ -316,6 +322,9 @@ internal constructor( * * For screenshare audio, a [ScreenAudioCapturer] can be used. * + * If a screen capture track is created but enabling fails or is cancelled, the capture is + * stopped and released, and [ScreenCaptureParams.onStop] is invoked. + * * @param screenCaptureParams When enabling the screenshare, this must be provided with * [ScreenCaptureParams.mediaProjectionPermissionResultData] containing resultData returned from launching * [MediaProjectionManager.createScreenCaptureIntent()](https://developer.android.com/reference/android/media/projection/MediaProjectionManager#createScreenCaptureIntent()). @@ -355,25 +364,29 @@ internal constructor( when (source) { Track.Source.CAMERA -> { val track = getOrCreateDefaultVideoTrack() - track.start() - track.startCapture() - if (!publishVideoTrack(track)) { - track.stopCapture() - track.stop() - } else { - success = true + try { + track.start() + track.startCapture() + success = publishVideoTrack(track) + } finally { + if (!success) { + track.stopCapture() + track.stop() + } } } Track.Source.MICROPHONE -> { val track = getOrCreateDefaultAudioTrack() - track.prewarm() - track.start() - if (!publishAudioTrack(track)) { - track.stop() - track.stopPrewarm() - } else { - success = true + try { + track.prewarm() + track.start() + success = publishAudioTrack(track) + } finally { + if (!success) { + track.stop() + track.stopPrewarm() + } } } @@ -381,22 +394,32 @@ internal constructor( if (screenCaptureParams == null) { throw IllegalArgumentException("Media Projection params is required to create a screen share track.") } + // MediaProjection termination and setup failure may race, but onStop + // is a single terminal signal for each enable attempt. + val stopNotified = AtomicBoolean(false) + fun notifyStopped() { + if (stopNotified.compareAndSet(false, true)) { + screenCaptureParams.onStop?.invoke() + } + } + val track = createScreencastTrack(mediaProjectionPermissionResultData = screenCaptureParams.mediaProjectionPermissionResultData) { unpublishTrack(it) - screenCaptureParams.onStop?.invoke() + notifyStopped() } - track.startForegroundService(screenCaptureParams.notificationId, screenCaptureParams.notification) - track.startCapture() - if (!publishVideoTrack(track, options = VideoTrackPublishOptions(null, screenShareTrackPublishDefaults))) { - screenCaptureParams.onStop?.invoke() - track.apply { - stopCapture() - stop() - dispose() + try { + track.startForegroundService(screenCaptureParams.notificationId, screenCaptureParams.notification) + track.startCapture() + success = publishVideoTrack(track, options = VideoTrackPublishOptions(null, screenShareTrackPublishDefaults)) + } finally { + if (!success) { + // stop() releases the screen capture service binding as well. + track.stopCapture() + track.stop() + track.dispose() + notifyStopped() } - } else { - success = true } } @@ -672,6 +695,8 @@ internal constructor( } // For fast publish, we can negotiate PC and request add track at the same time + var senderTransceiver: SenderTransceiverHandle? = null + var previousTrackTransceiver: RtpTransceiver? = null suspend fun negotiate() { if (this.engine.publisher == null) { throw IllegalStateException("publisher is not configured yet!") @@ -682,15 +707,14 @@ internal constructor( listOf(this.sid.value), encodings, ) - val transceiver = engine.createSenderTransceiver(track.rtcTrack, transInit) - - when (track) { - is LocalVideoTrack -> track.transceiver = transceiver - is LocalAudioTrack -> track.transceiver = transceiver - else -> { - throw IllegalArgumentException("Trying to publish a non local track of type ${track.javaClass}") + // Ownership must be recorded before cancellation can discard the RTC-thread result. + val createdTransceiver = withContext(NonCancellable) { + engine.createSenderTransceiver(track.rtcTrack, transInit).also { + senderTransceiver = it } } + coroutineContext.ensureActive() + val transceiver = createdTransceiver?.transceiver if (transceiver == null) { val exception = TrackException.PublishException("null sender returned from peer connection") @@ -698,6 +722,22 @@ internal constructor( throw exception } + when (track) { + is LocalVideoTrack -> { + previousTrackTransceiver = track.transceiver + track.transceiver = transceiver + } + + is LocalAudioTrack -> { + previousTrackTransceiver = track.transceiver + track.transceiver = transceiver + } + + else -> { + throw IllegalArgumentException("Trying to publish a non local track of type ${track.javaClass}") + } + } + track.statsGetter = engine.createStatsGetter(transceiver.sender) val finalOptions = options @@ -744,62 +784,90 @@ internal constructor( } } - val trackInfo: TrackInfo? - if (enabledPublishVideoCodecs.isNotEmpty()) { - // Can simultaneous publish and negotiate. - // codec is pre-verified in publishVideoTrack - trackInfo = coroutineScope { - val negotiateJob = launch { negotiate() } - val publishJob = async { requestAddTrack() } + var publication: LocalTrackPublication? = null + try { + val trackInfo: TrackInfo? + if (enabledPublishVideoCodecs.isNotEmpty()) { + // Can simultaneous publish and negotiate. + // codec is pre-verified in publishVideoTrack + trackInfo = coroutineScope { + val negotiateJob = launch { negotiate() } + val publishJob = async { requestAddTrack() } + + negotiateJob.join() + return@coroutineScope publishJob.await() + } + } else { + // legacy path. + trackInfo = requestAddTrack() + if (trackInfo != null) { + if (options is VideoTrackPublishOptions) { + // server might not support the codec the client has requested, in that case, fallback + // to a supported codec + val primaryCodecMime = trackInfo.codecsList.firstOrNull()?.mimeType + + if (primaryCodecMime != null) { + val updatedCodec = primaryCodecMime.mimeTypeToVideoCodec() + if (updatedCodec != null && updatedCodec != options.videoCodec) { + LKLog.d { "falling back to server selected codec: $updatedCodec" } + options = options.copy(videoCodec = updatedCodec) + + // recompute encodings since bitrates/etc could have changed + val videoTrack = track as LocalVideoTrack + + encodings = computeVideoEncodings(videoTrack.options.isScreencast, videoTrack.dimensions, options) + encodings // encodings is used in negotiate, this suppresses unused lint + } + } + } - negotiateJob.join() - return@coroutineScope publishJob.await() + negotiate() + } } - } else { - // legacy path. - trackInfo = requestAddTrack() + if (trackInfo != null) { - if (options is VideoTrackPublishOptions) { - // server might not support the codec the client has requested, in that case, fallback - // to a supported codec - val primaryCodecMime = trackInfo.codecsList.firstOrNull()?.mimeType - - if (primaryCodecMime != null) { - val updatedCodec = primaryCodecMime.mimeTypeToVideoCodec() - if (updatedCodec != null && updatedCodec != options.videoCodec) { - LKLog.d { "falling back to server selected codec: $updatedCodec" } - options = options.copy(videoCodec = updatedCodec) - - // recompute encodings since bitrates/etc could have changed - val videoTrack = track as LocalVideoTrack - - encodings = computeVideoEncodings(videoTrack.options.isScreencast, videoTrack.dimensions, options) - encodings // encodings is used in negotiate, this suppresses unused lint + publication = LocalTrackPublication( + info = trackInfo, + track = track, + participant = this, + options = options, + ) + addTrackPublication(publication) + LKLog.v { "add track publication $publication" } + + publishListener?.onPublishSuccess(publication) + internalListener?.onTrackPublished(publication, this) + eventBus.postEvent(ParticipantEvent.LocalTrackPublished(this, publication), scope) + } + } finally { + if (publication == null) { + // Negotiation can win the race against a failed or cancelled add track request. + // Without a publication there is no unpublish to stop the transceiver, so it + // would hold its sender and SDP m-section until the connection closes. + senderTransceiver?.let { createdTransceiver -> + val canRestorePreviousTransceiver = engine.rollbackSenderTransceiver( + senderTransceiver = createdTransceiver, + stopTransceiver = track is LocalVideoTrack, + ) + val transceiver = createdTransceiver.transceiver + when (track) { + is LocalVideoTrack -> { + if (track.transceiver === transceiver) { + track.transceiver = previousTrackTransceiver.takeIf { canRestorePreviousTransceiver } + } + } + + is LocalAudioTrack -> { + if (track.transceiver === transceiver) { + track.transceiver = previousTrackTransceiver.takeIf { canRestorePreviousTransceiver } + } } } } - - negotiate() } } - return if (trackInfo != null) { - val publication = LocalTrackPublication( - info = trackInfo, - track = track, - participant = this, - options = options, - ) - addTrackPublication(publication) - LKLog.v { "add track publication $publication" } - - publishListener?.onPublishSuccess(publication) - internalListener?.onTrackPublished(publication, this) - eventBus.postEvent(ParticipantEvent.LocalTrackPublished(this, publication), scope) - publication - } else { - null - } + return publication } private fun computeVideoEncodings( @@ -1207,7 +1275,7 @@ internal constructor( ) scope.launch { - val transceiver = engine.createSenderTransceiver(track.rtcTrack, transceiverInit) + val transceiver = engine.createSenderTransceiver(track.rtcTrack, transceiverInit)?.transceiver if (transceiver == null) { LKLog.w { "couldn't create new transceiver! $codec" } return@launch diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt index b5dcaa573..563c7dc5e 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/LocalScreencastVideoTrack.kt @@ -28,6 +28,7 @@ import android.view.WindowManager import dagger.assisted.Assisted import dagger.assisted.AssistedFactory import dagger.assisted.AssistedInject +import io.livekit.android.memory.SurfaceTextureHelperCloser import io.livekit.android.room.DefaultsManager import io.livekit.android.room.participant.LocalParticipant import io.livekit.android.room.track.screencapture.ScreenCaptureConnection @@ -190,7 +191,15 @@ constructor( } override fun stop() { - super.stop() + // The projection's onStop arrives on its own callback thread and stops this track, so + // it can lose a race against dispose(), after which the capturer throws when stopped. + if (!isDisposed) { + try { + super.stop() + } catch (e: Exception) { + LKLog.w(e) { "Exception when stopping the screen share track." } + } + } serviceConnection.stop() orientationEventListener.disable() } @@ -243,21 +252,28 @@ constructor( addOnStopCallback(onStop) } val capturer = createScreenCapturer(mediaProjectionPermissionResultData, callback) + val surfaceTextureHelper = SurfaceTextureHelper.create("ScreenVideoCaptureThread", rootEglBase.eglBaseContext) capturer.initialize( - SurfaceTextureHelper.create("ScreenVideoCaptureThread", rootEglBase.eglBaseContext), + surfaceTextureHelper, context, source.capturerObserver, ) - val track = peerConnectionFactory.createVideoTrack(UUID.randomUUID().toString(), source) + val rtcTrack = peerConnectionFactory.createVideoTrack(UUID.randomUUID().toString(), source) - return screencastVideoTrackFactory.create( + val track = screencastVideoTrackFactory.create( capturer = capturer, source = source, options = options, name = name, - rtcTrack = track, + rtcTrack = rtcTrack, mediaProjectionCallback = callback, ) + // The track owns the SurfaceTextureHelper; dispose() releases its capture thread. + track.closeableManager.registerResource( + rtcTrack, + SurfaceTextureHelperCloser(surfaceTextureHelper), + ) + return track } private fun createScreenCapturer( 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 925e46e35..358ed6fb9 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 @@ -119,7 +119,7 @@ constructor( internal val simulcastTransceivers: List get() = simulcastCodecs.values.mapNotNull { it.transceiver } - private val closeableManager = CloseableManager() + internal val closeableManager = CloseableManager() /** * Starts the [capturer] with the capture params contained in [options]. diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt index e29924242..6a51cfebe 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.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. @@ -170,7 +170,7 @@ class MockPeerConnection( } override fun removeTrack(sender: RtpSender?): Boolean { - return super.removeTrack(sender) + return true } override fun addTransceiver(track: MediaStreamTrack): RtpTransceiver { diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/participant/SetTrackEnabledCleanupMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/participant/SetTrackEnabledCleanupMockE2ETest.kt new file mode 100644 index 000000000..a0410fdda --- /dev/null +++ b/livekit-android-test/src/test/java/io/livekit/android/room/participant/SetTrackEnabledCleanupMockE2ETest.kt @@ -0,0 +1,356 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.livekit.android.room.participant + +import android.Manifest +import android.app.Application +import android.app.Notification +import android.content.ComponentName +import android.content.Intent +import io.livekit.android.room.ConnectionState +import io.livekit.android.room.track.LocalScreencastVideoTrack +import io.livekit.android.room.track.screencapture.ScreenCaptureParams +import io.livekit.android.room.track.screencapture.ScreenCaptureService +import io.livekit.android.test.MockE2ETest +import io.livekit.android.test.mock.camera.MockCameraProvider +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runCurrent +import livekit.LivekitRtc +import livekit.org.webrtc.ScreenCapturerAndroid +import livekit.org.webrtc.SurfaceTextureHelper +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.MockedConstruction +import org.mockito.Mockito.atLeastOnce +import org.mockito.Mockito.doThrow +import org.mockito.Mockito.mock +import org.mockito.Mockito.mockConstruction +import org.mockito.Mockito.mockStatic +import org.mockito.Mockito.never +import org.mockito.Mockito.verify +import org.mockito.kotlin.any +import org.mockito.kotlin.anyOrNull +import org.robolectric.RobolectricTestRunner +import org.robolectric.Shadows.shadowOf + +/** + * Enabling a track suspends while publishing, so a cancellation can land after the track and its + * resources were created. These tests pin that setTrackEnabled always tears down a track it could + * not publish, at every suspension point. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class SetTrackEnabledCleanupMockE2ETest : MockE2ETest() { + + private lateinit var screenCapturers: MockedConstruction + private var projectionCallback: LocalScreencastVideoTrack.MediaProjectionCallback? = null + + @Before + fun mockScreenCapturerConstruction() { + screenCapturers = mockConstruction(ScreenCapturerAndroid::class.java) { _, creation -> + projectionCallback = creation.arguments()[1] as LocalScreencastVideoTrack.MediaProjectionCallback + } + } + + @After + fun releaseScreenCapturerConstruction() { + screenCapturers.close() + } + + private val resumeLeave = with(LivekitRtc.SignalResponse.newBuilder()) { + leave = with(LivekitRtc.LeaveRequest.newBuilder()) { + action = LivekitRtc.LeaveRequest.Action.RESUME + build() + } + build() + } + + private fun screenCaptureParams(onStop: (() -> Unit)? = null) = ScreenCaptureParams( + mediaProjectionPermissionResultData = Intent(), + // An explicit notification, so the service does not reach for a notification manager. + notificationId = 42, + notification = Notification(), + onStop = onStop, + ) + + private fun connectScreenCaptureService() { + shadowOf(context as Application).boundServiceConnections.single().onServiceConnected( + ComponentName(context, ScreenCaptureService::class.java), + ScreenCaptureService().onBind(null), + ) + } + + @Test + fun cancelDuringScreenShareServiceBindCleansUpTrack() = runTest { + connect() + + var onStopCalled = false + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCalled = true }) + } + // Suspended in startForegroundService, waiting for the service to connect. + runCurrent() + assertEquals(1, shadowOf(context as Application).boundServiceConnections.size) + + job.cancelAndJoin() + + val capturer = screenCapturers.constructed().single() + verify(capturer, atLeastOnce()).stopCapture() + verify(capturer).dispose() + assertTrue(onStopCalled) + assertEquals(0, shadowOf(context as Application).boundServiceConnections.size) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + } + + @Test + fun cancelDuringScreenSharePublishCleansUpTrackAndUnbindsService() = runTest { + connect() + // Never answer the add track request, so publishing suspends until cancelled. + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + var onStopCalled = false + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCalled = true }) + } + runCurrent() + connectScreenCaptureService() + // Suspended in publishVideoTrack, with capture running and the service bound. + runCurrent() + val capturer = screenCapturers.constructed().single() + verify(capturer, never()).stopCapture() + + job.cancelAndJoin() + + verify(capturer, atLeastOnce()).stopCapture() + verify(capturer).dispose() + assertTrue(onStopCalled) + assertEquals(0, shadowOf(context as Application).boundServiceConnections.size) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + + // Negotiation ran concurrently with the add track request, so the transceiver it + // created must be rolled back; nothing else stops it once the track is abandoned. + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single()).stopInternal() + } + + @Test + fun projectionStopDuringScreenSharePublishDeliversOnStopOnce() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + var onStopCount = 0 + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCount++ }) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + // The user ends the projection from the system UI while publishing is still in flight. + projectionCallback!!.onStop() + assertEquals(1, onStopCount) + + job.cancelAndJoin() + + assertEquals(1, onStopCount) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + } + + @Test + fun projectionStopArrivingAfterCancelledEnableDoesNotThrow() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + job.cancelAndJoin() + + // A projection stop that was already dispatched can arrive after cleanup disposed the + // track, and the capturer throws once it is disposed. The callback must not propagate. + val capturer = screenCapturers.constructed().single() + doThrow(RuntimeException("capturer is disposed.")).`when`(capturer).stopCapture() + projectionCallback!!.onStop() + } + + @Test + fun successfulScreenShareKeepsTrackAndServiceBinding() = runTest { + connect() + + var onStopCalled = false + val result = async(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams { onStopCalled = true }) + } + runCurrent() + connectScreenCaptureService() + advanceUntilIdle() + + assertTrue(result.await()) + val capturer = screenCapturers.constructed().single() + verify(capturer, never()).stopCapture() + verify(capturer, never()).dispose() + assertFalse(onStopCalled) + assertEquals(1, shadowOf(context as Application).boundServiceConnections.size) + assertEquals(1, room.localParticipant.trackPublications.size) + } + + @Test + fun cancelDuringMicrophonePublishStopsTrackAndKeepsTransceiver() = runTest { + connect() + shadowOf(context as Application).grantPermissions(Manifest.permission.RECORD_AUDIO) + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + + job.cancelAndJoin() + + assertFalse(room.localParticipant.getOrCreateDefaultAudioTrack().enabled) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + + // Detaching the sender is the whole rollback for audio: a single m-section is a cheap + // thing to leave behind, and stopping it renegotiates the publisher for no gain. + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun cancelDuringScreenSharePublishWhileReconnectingKeepsTransceiver() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + // Losing the socket with the add track request outstanding is what fails the publish, + // and it starts the reconnect that renegotiates this publisher. + wsFactory.ws.cancel() + runCurrent() + + job.cancelAndJoin() + + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun resumeLeaveDuringScreenSharePublishKeepsTransceiver() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + // A resume leave fails the pending publish and leaves the reconnect to the socket close, + // so the publisher still reads as connected while the failure unwinds. + simulateMessageFromServer(resumeLeave) + runCurrent() + job.join() + + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun resumeLeaveKeepsTransceiverWhenCleanupTrailsTheNextSession() = runTest { + connect() + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setScreenShareEnabled(true, screenCaptureParams()) + } + runCurrent() + connectScreenCaptureService() + runCurrent() + + simulateMessageFromServer(resumeLeave) + // The failed publish stays queued until the next session is serving. A soft reconnect + // keeps the publisher, so the transceiver this attempt created is the resumed session's + // to negotiate, not this cleanup's to remove. + component.rtcEngine().connectionState = ConnectionState.RESUMING + component.rtcEngine().connectionState = ConnectionState.CONNECTED + runCurrent() + job.join() + + val transceivers = getPublisherPeerConnection().transceivers + assertEquals(1, transceivers.size) + verify(transceivers.single(), never()).stopInternal() + } + + @Test + fun cancelDuringCameraPublishStopsTrack() = runTest { + connect() + MockCameraProvider.register() + shadowOf(context as Application).grantPermissions(Manifest.permission.CAMERA) + wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler) + + val job = launch(StandardTestDispatcher(testScheduler)) { + room.localParticipant.setCameraEnabled(true) + } + runCurrent() + + job.cancelAndJoin() + + assertFalse(room.localParticipant.getOrCreateDefaultVideoTrack().enabled) + assertTrue(room.localParticipant.trackPublications.isEmpty()) + } + + @Test + fun disposeReleasesScreencastSurfaceTextureHelper() { + val helper = mock(SurfaceTextureHelper::class.java) + mockStatic(SurfaceTextureHelper::class.java).use { helperFactory -> + helperFactory.`when` { + SurfaceTextureHelper.create(any(), anyOrNull()) + }.thenReturn(helper) + + val track = room.localParticipant.createScreencastTrack( + mediaProjectionPermissionResultData = Intent(), + ) {} + track.dispose() + } + + verify(helper).stopListening() + verify(helper).dispose() + } +}