From f22a18c30766012f3f36a70a1b0c7dac969668bd Mon Sep 17 00:00:00 2001 From: Adrian Niculescu <15037449+adrian-niculescu@users.noreply.github.com> Date: Fri, 31 Jul 2026 14:43:22 +0300 Subject: [PATCH] Fixed full reconnect republish racing concurrent publishes of the same track --- .changeset/violet-crabs-brake.md | 5 + .../detekt-baseline-release.xml | 2 +- .../room/participant/LocalParticipant.kt | 207 ++++++- .../room/track/LocalScreencastVideoTrack.kt | 30 +- .../android/room/track/LocalVideoTrack.kt | 46 +- .../io/livekit/android/room/track/Track.kt | 43 +- .../room/ConcurrentPublishMockE2ETest.kt | 581 ++++++++++++++++++ 7 files changed, 879 insertions(+), 35 deletions(-) create mode 100644 .changeset/violet-crabs-brake.md create mode 100644 livekit-android-test/src/test/java/io/livekit/android/room/ConcurrentPublishMockE2ETest.kt diff --git a/.changeset/violet-crabs-brake.md b/.changeset/violet-crabs-brake.md new file mode 100644 index 000000000..a8002317d --- /dev/null +++ b/.changeset/violet-crabs-brake.md @@ -0,0 +1,5 @@ +--- +"client-sdk-android": patch +--- + +Fixed full reconnect republish racing concurrent publishes of the same track, which could leave the mic published but silent or silently unpublished. diff --git a/livekit-android-sdk/detekt-baseline-release.xml b/livekit-android-sdk/detekt-baseline-release.xml index 54ebf273d..5300deaed 100644 --- a/livekit-android-sdk/detekt-baseline-release.xml +++ b/livekit-android-sdk/detekt-baseline-release.xml @@ -13,7 +13,7 @@ CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$@Throws(TrackException.PublishException::class) private suspend fun publishTrackImpl( track: Track, options: TrackPublishOptions, requestConfig: AddTrackRequest.Builder.() -> Unit, encodings: List<RtpParameters.Encoding> = emptyList(), publishListener: PublishListener? = null, ): LocalTrackPublication? CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$private fun computeVideoEncodings( isScreenShare: Boolean, dimensions: Track.Dimensions, options: VideoTrackPublishOptions, ): List<RtpParameters.Encoding> CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$private suspend fun setTrackEnabled( source: Track.Source, enabled: Boolean, screenCaptureParams: ScreenCaptureParams? = null, ): Boolean - CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$suspend fun publishVideoTrack( track: LocalVideoTrack, options: VideoTrackPublishOptions = VideoTrackPublishOptions( null, if (track.options.isScreencast) screenShareTrackPublishDefaults else videoTrackPublishDefaults, ), publishListener: PublishListener? = null, ): Boolean + CyclomaticComplexMethod:LocalParticipant.kt$LocalParticipant$private suspend fun publishVideoTrackImpl( track: LocalVideoTrack, options: VideoTrackPublishOptions = VideoTrackPublishOptions( null, if (track.options.isScreencast) screenShareTrackPublishDefaults else videoTrackPublishDefaults, ), publishListener: PublishListener? = null, ): Boolean CyclomaticComplexMethod:LocalVideoTrack.kt$LocalVideoTrack$private fun setPublishingLayersForSender( sender: RtpSender, qualities: List<LivekitRtc.SubscribedQuality>, ) CyclomaticComplexMethod:NetworkInfo.kt$AndroidNetworkInfo$override fun getNetworkType(): NetworkType CyclomaticComplexMethod:PeerConnectionTransport.kt$@VisibleForTesting fun ensureCodecBitrates( media: MediaDescription, trackBitrates: Map<TrackBitrateInfoKey, TrackBitrateInfo>, ) 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 5abbc922c..60f9cbb7f 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 @@ -86,6 +86,7 @@ import livekit.org.webrtc.SurfaceTextureHelper import livekit.org.webrtc.VideoCapturer import livekit.org.webrtc.VideoProcessor import java.util.Collections +import java.util.concurrent.ConcurrentHashMap import javax.inject.Named import kotlin.math.max import kotlin.math.min @@ -130,11 +131,20 @@ internal constructor( private val jobs = mutableMapOf() - // For ensuring that only one caller can execute setTrackEnabled at a time. - // Without it, there's a potential to create multiple of the same source, + // Serializes all publishes and setTrackEnabled calls for a source. Without it, + // concurrent publishes of the same track can stop a track another path is + // publishing, there's a potential to create multiple of the same source, and // Camera has deadlock issues with multiple CameraCapturers trying to activate/stop. private val sourcePubLocks = Track.Source.entries.associateWith { Mutex() } + // Tracks the SDK stopped after a failed publish, keyed to the enabled-state + // revision of that stop. A successful publish clears only a marker that + // predates it, and a restart applies only while the revision is unchanged, so + // any consumer enabled-state transition invalidates the marker. Only marked + // tracks may be restarted when republishing; a track the consumer stopped + // stays stopped. + private val tracksStoppedByFailedPublish = ConcurrentHashMap() + internal val enabledPublishVideoCodecs = Collections.synchronizedList(mutableListOf()) private var defaultAudioTrack: LocalAudioTrack? = null @@ -353,6 +363,7 @@ internal constructor( if (source == Track.Source.CAMERA && pub.track is LocalVideoTrack) { (pub.track as? LocalVideoTrack)?.startCapture() } + pub.track?.let { tracksStoppedByFailedPublish.remove(it) } success = true } else { // Not published yet, create the default track and publish. @@ -361,15 +372,15 @@ internal constructor( val track = getOrCreateDefaultVideoTrack() track.start() track.startCapture() - if (publishVideoTrack(track)) { + tracksStoppedByFailedPublish.remove(track) + if (publishVideoTrackImpl(track)) { success = true } else if (isTrackPublished(track)) { - // A concurrent publish outside the pub lock won the race; + // A concurrent publish won the race; // the track is live, so leave it alone. success = true } else { - track.stopCapture() - track.stop() + stopTrackForFailedPublish(track) } } @@ -377,15 +388,15 @@ internal constructor( val track = getOrCreateDefaultAudioTrack() track.prewarm() track.start() - if (publishAudioTrack(track)) { + tracksStoppedByFailedPublish.remove(track) + if (publishAudioTrackImpl(track)) { success = true } else if (isTrackPublished(track)) { - // A concurrent publish outside the pub lock won the race; + // A concurrent publish won the race; // the track is live, so leave it alone. success = true } else { - track.stop() - track.stopPrewarm() + stopTrackForFailedPublish(track) } } @@ -400,13 +411,18 @@ internal constructor( } track.startForegroundService(screenCaptureParams.notificationId, screenCaptureParams.notification) track.startCapture() - if (!publishVideoTrack(track, options = VideoTrackPublishOptions(null, screenShareTrackPublishDefaults))) { + if (!publishVideoTrackImpl(track, options = VideoTrackPublishOptions(null, screenShareTrackPublishDefaults))) { screenCaptureParams.onStop?.invoke() track.apply { stopCapture() stop() dispose() } + } else if (!track.enabled) { + // The MediaProjection ended while the publish was in + // flight, so the onStop unpublish ran before the + // publication existed. Run it now that it does. + unpublishTrack(track, stopOnUnpublish = false) } else { success = true } @@ -442,6 +458,11 @@ internal constructor( /** * Publishes an audio track. * + * Publishes for the same [Track.Source] are serialized: this call suspends while + * another publish or [setMicrophoneEnabled] call for the source is in progress. + * [publishListener] callbacks run while that serialization is held, so they must + * not block or synchronously start another publish or enable call for the source. + * * @param track The track to publish. * @param options The publish options to use, or [Room.audioTrackPublishDefaults] if none is passed. * @return true if the track published successfully @@ -453,6 +474,20 @@ internal constructor( audioTrackPublishDefaults, ).copy(preconnect = defaultsManager.isPrerecording), publishListener: PublishListener? = null, + ): Boolean { + val source = options.source ?: Track.Source.MICROPHONE + return sourcePubLocks.getValue(source).withLock { + publishAudioTrackImpl(track, options, publishListener) + } + } + + private suspend fun publishAudioTrackImpl( + track: LocalAudioTrack, + options: AudioTrackPublishOptions = AudioTrackPublishOptions( + null, + audioTrackPublishDefaults, + ).copy(preconnect = defaultsManager.isPrerecording), + publishListener: PublishListener? = null, ): Boolean { if (track.isDisposed) { LKLog.w { "Attempting to publish a disposed track, ignoring." } @@ -499,6 +534,12 @@ internal constructor( /** * Publishes an video track. * + * Publishes for the same [Track.Source] are serialized: this call suspends while + * another publish or [setCameraEnabled]/[setScreenShareEnabled] call for the + * source is in progress. [publishListener] callbacks run while that serialization + * is held, so they must not block or synchronously start another publish or + * enable call for the source. + * * @param track The track to publish. * @param options The publish options to use, or [Room.videoTrackPublishDefaults] if none is passed. * @return true if the track published successfully @@ -510,6 +551,21 @@ internal constructor( if (track.options.isScreencast) screenShareTrackPublishDefaults else videoTrackPublishDefaults, ), publishListener: PublishListener? = null, + ): Boolean { + val source = options.source + ?: if (track.options.isScreencast) Track.Source.SCREEN_SHARE else Track.Source.CAMERA + return sourcePubLocks.getValue(source).withLock { + publishVideoTrackImpl(track, options, publishListener) + } + } + + private suspend fun publishVideoTrackImpl( + track: LocalVideoTrack, + options: VideoTrackPublishOptions = VideoTrackPublishOptions( + null, + if (track.options.isScreencast) screenShareTrackPublishDefaults else videoTrackPublishDefaults, + ), + publishListener: PublishListener? = null, ): Boolean { @Suppress("NAME_SHADOWING") var options = options @@ -640,6 +696,10 @@ internal constructor( return null } + // On success, only a failure marker that predates this publish is cleared; + // one added mid-flight by a newer failure survives. + val stopMarkerGeneration = tracksStoppedByFailedPublish[track] + fun onPublishFailure(e: TrackException.PublishException, triggerEvent: Boolean = true) { publishListener?.onPublishFailure(e) if (triggerEvent) { @@ -804,6 +864,9 @@ internal constructor( options = options, ) addTrackPublication(publication) + if (stopMarkerGeneration != null) { + tracksStoppedByFailedPublish.remove(track, stopMarkerGeneration) + } LKLog.v { "add track publication $publication" } publishListener?.onPublishSuccess(publication) @@ -1286,10 +1349,9 @@ internal constructor( internal fun prepareForFullReconnect() { val pubs = localTrackPublications.toList() // creates a copy, so is safe from the following removal. - // Only set the first time we start a full reconnect. - if (republishes == null) { - republishes = pubs - } + // Accumulate across attempts: publications created between failed attempts + // must be restored too. Consumed by republishTracks on success. + republishes = republishes.orEmpty() + pubs trackPublications = trackPublications.toMutableMap().apply { clear() } @@ -1301,25 +1363,125 @@ internal constructor( internal suspend fun republishTracks() { val publish = republishes?.toList() ?: emptyList() + + // The accumulated snapshot can hold several publications for one track; the + // last entry carries the newest state (e.g. a mute during the reconnect), so + // it wins. + val latestPubs = publish.filter { it.track != null }.associateBy { it.track }.values + for (pub in latestPubs) { + try { + republishTrack(pub) + } catch (e: Exception) { + e.rethrowIfCancellationSignal() + LKLog.w(e) { "Failed to republish track ${pub.sid}" } + } + } republishes = null + } - for (pub in publish) { - val track = pub.track ?: continue - unpublishTrack(track, false) + private suspend fun republishTrack(pub: LocalTrackPublication) { + val track = pub.track ?: return + sourcePubLocks.getValue(republishLockSource(pub, track)).withLock { + // The snapshot publication belongs to the dead session; drop it if it + // survived the reconnect clear by racing a concurrent publish. + if (trackPublications[pub.sid] === pub) { + trackPublications = trackPublications.toMutableMap().apply { remove(pub.sid) } + } + // A concurrent publish (e.g. setMicrophoneEnabled during the reconnect) + // may have already landed this track on the new session; it is live and + // supersedes the snapshot, so leave it alone. + if (isTrackPublished(track)) { + return@withLock + } // Cannot publish muted tracks. if (!pub.muted) { + // A stopped screencast's MediaProjection is single-use: the share has + // ended and cannot be restored. + if (track is LocalScreencastVideoTrack && !track.enabled) { + return@withLock + } + restartTrackIfStopped(track) val success = when (track) { - is LocalAudioTrack -> publishAudioTrack(track, pub.options as AudioTrackPublishOptions, null) - is LocalVideoTrack -> publishVideoTrack(track, pub.options as VideoTrackPublishOptions, null) + is LocalAudioTrack -> publishAudioTrackImpl(track, pub.options as AudioTrackPublishOptions, null) + is LocalVideoTrack -> publishVideoTrackImpl(track, pub.options as VideoTrackPublishOptions, null) else -> throw IllegalStateException("LocalParticipant has a non local track publish?") } - if (!success) { - track.stop() + handleRepublishResult(track, success) + } + } + } + + private fun handleRepublishResult(track: Track, success: Boolean) { + if (!success && !isTrackPublished(track)) { + stopTrackForFailedPublish(track) + } else if (success && track is LocalScreencastVideoTrack && !track.enabled) { + // The MediaProjection ended while the republish was in flight, so the + // onStop unpublish ran before the publication existed. Run it now that + // it does. + unpublishTrack(track, stopOnUnpublish = false) + } + } + + // The lock source is derived from the publish options rather than the server's + // TrackInfo, matching the source a concurrent publish of this track would lock. + private fun republishLockSource(pub: LocalTrackPublication, track: Track): Track.Source { + return pub.options.source + ?: when (track) { + is LocalAudioTrack -> Track.Source.MICROPHONE + is LocalVideoTrack -> if (track.options.isScreencast) Track.Source.SCREEN_SHARE else Track.Source.CAMERA + else -> pub.source + } + } + + // A concurrent publish that failed during the reconnect may have stopped the + // track; a track republished as unmuted must be live again. Only tracks the + // SDK itself stopped are restarted, and only while no other enabled-state + // transition has happened since that stop: a track the consumer stopped stays + // stopped. Screencasts are never restarted: a stopped MediaProjection cannot + // be reused. + private fun restartTrackIfStopped(track: Track) { + if (track.isDisposed || track.enabled) { + return + } + val markerRevision = tracksStoppedByFailedPublish.remove(track) ?: return + when (track) { + is LocalAudioTrack -> { + if (track.setEnabledIfRevisionUnchanged(markerRevision, true) != null) { + track.prewarm() + } + } + + is LocalScreencastVideoTrack -> {} + + is LocalVideoTrack -> { + val appliedRevision = track.setEnabledIfRevisionUnchanged(markerRevision, true) + if (appliedRevision != null) { + track.startCaptureIfRevisionUnchanged(appliedRevision) } } } } + // Stops a track after its publish failed, atomically recording the stop so + // republishing can distinguish it from a stop the consumer made. Screencasts + // are not recorded; they are never restarted. + private fun stopTrackForFailedPublish(track: Track) { + when (track) { + is LocalScreencastVideoTrack -> track.stop() + + is LocalVideoTrack -> { + tracksStoppedByFailedPublish[track] = track.stopReturningRevision() + } + + is LocalAudioTrack -> { + tracksStoppedByFailedPublish[track] = track.stopReturningRevision() + track.stopPrewarm() + } + + else -> track.stop() + } + } + internal fun onLocalTrackSubscribed(publication: LocalTrackPublication) { if (!trackPublications.containsKey(publication.sid)) { LKLog.w { "Could not find local track publication for subscribed event " } @@ -1401,6 +1563,7 @@ internal constructor( } defaultAudioTrack = null defaultVideoTrack = null + tracksStoppedByFailedPublish.clear() } /** 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..f6072faf8 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 @@ -189,6 +189,23 @@ constructor( serviceConnection.startForeground(notificationId, notification) } + // MediaProjection stop callbacks arrive on the capture handler thread, and + // ScreenCapturerAndroid.stopCapture round-trips through that same handler, so + // hopping to the RTC thread here would deadlock the two. Screencasts take no + // part in the enabled-state revision scheme, making the direct call safe. + override fun stopCapture() { + if (isDisposed) { + return + } + try { + capturer.stopCapture() + } catch (e: IllegalStateException) { + // A late MediaProjection stop can race disposal; a disposed capturer + // has nothing left to stop. + LKLog.d(e) { "stopCapture on a disposed screen capturer, ignoring." } + } + } + override fun stop() { super.stop() serviceConnection.stop() @@ -239,9 +256,7 @@ constructor( ): LocalScreencastVideoTrack { val source = peerConnectionFactory.createVideoSource(options.isScreencast) source.setVideoProcessor(videoProcessor) - val callback = MediaProjectionCallback().apply { - addOnStopCallback(onStop) - } + val callback = MediaProjectionCallback() val capturer = createScreenCapturer(mediaProjectionPermissionResultData, callback) capturer.initialize( SurfaceTextureHelper.create("ScreenVideoCaptureThread", rootEglBase.eglBaseContext), @@ -250,7 +265,7 @@ constructor( ) val track = peerConnectionFactory.createVideoTrack(UUID.randomUUID().toString(), source) - return screencastVideoTrackFactory.create( + val screencastTrack = screencastVideoTrackFactory.create( capturer = capturer, source = source, options = options, @@ -258,6 +273,13 @@ constructor( rtcTrack = track, mediaProjectionCallback = callback, ) + + // The track's own stop callback is registered first (in its init block), + // so the track is already stopped and its ended state observable by the + // time this callback runs. + callback.addOnStopCallback(onStop) + + return screencastTrack } 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..25c600075 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 @@ -125,25 +125,57 @@ constructor( * Starts the [capturer] with the capture params contained in [options]. */ open fun startCapture() { - capturer.startCapture( - options.captureParams.width, - options.captureParams.height, - options.captureParams.maxFps, - ) + withRTCTrack { + enabledStateRevision.incrementAndGet() + capturer.startCapture( + options.captureParams.width, + options.captureParams.height, + options.captureParams.maxFps, + ) + } } /** * Stops the [capturer]. */ open fun stopCapture() { - capturer.stopCapture() + withRTCTrack { + enabledStateRevision.incrementAndGet() + capturer.stopCapture() + } + } + + // Starts capture only if no enabled-state mutation happened since + // [expectedRevision]; returns whether capture was started. The check and + // [startCapture] run as one unit on the RTC thread. + internal fun startCaptureIfRevisionUnchanged(expectedRevision: Long): Boolean { + return withRTCTrack(defaultValue = false) { + if (enabledStateRevision.get() != expectedRevision) { + false + } else { + startCapture() + true + } + } } override fun stop() { - capturer.stopCapture() + stopCapture() super.stop() } + // Capture stop and disable run as one unit on the RTC thread so no + // concurrent state change can land between them; [stopCapture] is the + // overridable seam for subclasses. + internal override fun stopReturningRevision(): Long { + return withRTCTrack(defaultValue = enabledStateRevision.get()) { + stopCapture() + val revision = enabledStateRevision.incrementAndGet() + rtcTrack.setEnabled(false) + revision + } + } + override fun dispose() { super.dispose() capturer.dispose() diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/Track.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/Track.kt index 35c55edb7..6390a38c8 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/Track.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/Track.kt @@ -28,6 +28,7 @@ import livekit.LivekitRtc import livekit.org.webrtc.MediaStreamTrack import livekit.org.webrtc.RTCStatsCollectorCallback import livekit.org.webrtc.RTCStatsReport +import java.util.concurrent.atomic.AtomicLong import kotlin.contracts.ExperimentalContracts import kotlin.contracts.InvocationKind import kotlin.contracts.contract @@ -54,9 +55,49 @@ abstract class Track( } internal set + // Advances on every enabled-state mutation, letting owners detect state + // transitions made through the public API behind their back. Mutated only on + // the RTC thread, which serializes it with the actual state change. + internal val enabledStateRevision = AtomicLong() + var enabled: Boolean get() = withRTCTrack(defaultValue = false) { rtcTrack.enabled() } - set(value) = withRTCTrack { rtcTrack.setEnabled(value) } + set(value) { + withRTCTrack { + enabledStateRevision.incrementAndGet() + rtcTrack.setEnabled(value) + } + } + + // Sets [enabled] and returns the revision of this mutation. + internal fun setEnabledReturningRevision(value: Boolean): Long { + return withRTCTrack(defaultValue = enabledStateRevision.get()) { + val revision = enabledStateRevision.incrementAndGet() + rtcTrack.setEnabled(value) + revision + } + } + + // Stops the track as a single enabled-state transition and returns the + // revision of the mutation. + internal open fun stopReturningRevision(): Long { + return setEnabledReturningRevision(false) + } + + // Sets [enabled] only if no enabled-state mutation happened since + // [expectedRevision]; returns the revision of the applied mutation, or null + // if the state had already moved on. + internal fun setEnabledIfRevisionUnchanged(expectedRevision: Long, value: Boolean): Long? { + return withRTCTrack(defaultValue = null) { + if (enabledStateRevision.get() != expectedRevision) { + null + } else { + val revision = enabledStateRevision.incrementAndGet() + rtcTrack.setEnabled(value) + revision + } + } + } var statsGetter: RTCStatsGetter? = null diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/ConcurrentPublishMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/ConcurrentPublishMockE2ETest.kt new file mode 100644 index 000000000..802362d32 --- /dev/null +++ b/livekit-android-test/src/test/java/io/livekit/android/room/ConcurrentPublishMockE2ETest.kt @@ -0,0 +1,581 @@ +/* + * 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 + +import android.Manifest +import android.app.Application +import android.content.Context +import androidx.test.core.app.ApplicationProvider +import io.livekit.android.room.participant.AudioTrackPublishOptions +import io.livekit.android.room.track.Track +import io.livekit.android.test.MockE2ETest +import io.livekit.android.test.mock.TestData +import io.livekit.android.test.util.toPBByteString +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runCurrent +import livekit.LivekitRtc +import livekit.org.webrtc.PeerConnection +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.Shadows + +/** + * Tests for concurrent publishes of the same track instance, e.g. the full + * reconnect republish overlapping setMicrophoneEnabled, or a direct + * publishAudioTrack overlapping setMicrophoneEnabled. + */ +@ExperimentalCoroutinesApi +@RunWith(RobolectricTestRunner::class) +class ConcurrentPublishMockE2ETest : MockE2ETest() { + + private fun grantAudioPermission() { + val context = ApplicationProvider.getApplicationContext() + Shadows.shadowOf(context as Application).grantPermissions(Manifest.permission.RECORD_AUDIO) + } + + private fun reconnectWebsocket() { + wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request)) + val softReconnectParam = wsFactory.request.url + .queryParameter(SignalClient.CONNECT_QUERY_RECONNECT) + ?.toIntOrNull() + ?: 0 + + if (softReconnectParam == 0) { + simulateMessageFromServer(TestData.JOIN) + } else { + simulateMessageFromServer(TestData.RECONNECT) + } + } + + private fun connectPublisherPeerConnection() { + getPublisherPeerConnection().moveToIceConnectionState(PeerConnection.IceConnectionState.CONNECTED) + } + + private fun deferNextAddTrackResponse(): () -> LivekitRtc.AddTrackRequest? { + var deferredAddTrack: LivekitRtc.AddTrackRequest? = null + wsFactory.registerSignalRequestHandler { request -> + if (request.hasAddTrack() && deferredAddTrack == null) { + deferredAddTrack = request.addTrack + true + } else { + false + } + } + return { deferredAddTrack } + } + + private fun respondToAddTrack(addTrack: LivekitRtc.AddTrackRequest) { + wsFactory.receiveMessage( + with(LivekitRtc.SignalResponse.newBuilder()) { + trackPublished = with(LivekitRtc.TrackPublishedResponse.newBuilder()) { + cid = addTrack.cid + track = TestData.LOCAL_AUDIO_TRACK + build() + } + build() + }, + ) + } + + private fun sentAddTrackCount(): Int { + return wsFactory.ws.sentRequests.count { requestString -> + LivekitRtc.SignalRequest.newBuilder() + .mergeFrom(requestString.toPBByteString()) + .build() + .hasAddTrack() + } + } + + @Test + fun micEnabledDuringReconnectSurvivesRepublish() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + assertTrue(room.localParticipant.setMicrophoneEnabled(false)) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + + // App enables the mic mid-reconnect; the publish completes against the new session. + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val pub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(pub) + val micTrack = pub!!.track!! + + // Reconnect completes; republishTracks() runs against the old (muted) snapshot. + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The mid-reconnect publication supersedes the snapshot and must survive. + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertTrue(micTrack.enabled) + } + + @Test + fun micEnableWaitsForInFlightRepublish() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val micTrack = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)!!.track!! + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + + val deferredAddTrack = deferNextAddTrackResponse() + + connectPeerConnection() + connectPublisherPeerConnection() + runCurrent() + assertNotNull("republish addTrack should be in flight", deferredAddTrack()) + + // The app's enable must wait for the in-flight republish instead of + // failing as a duplicate and stopping the shared track. + var micResult: Boolean? = null + val micJob = launch { + micResult = room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + assertNull(micResult) + + respondToAddTrack(deferredAddTrack()!!) + runCurrent() + micJob.join() + + assertEquals(true, micResult) + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertTrue(micTrack.enabled) + } + + @Test + fun republishSkipsTrackPublishedByConcurrentEnable() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val micTrack = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)!!.track!! + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + + val deferredAddTrack = deferNextAddTrackResponse() + + var micResult: Boolean? = null + val micJob = launch { + micResult = room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + assertNotNull("app addTrack should be in flight", deferredAddTrack()) + + // Reconnect completes while the app's publish is in flight. + connectPeerConnection() + connectPublisherPeerConnection() + runCurrent() + + respondToAddTrack(deferredAddTrack()!!) + runCurrent() + micJob.join() + advanceUntilIdle() + + assertEquals(true, micResult) + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertTrue(micTrack.enabled) + + // The republish must skip the superseded snapshot entry rather than + // republishing or stopping the app's live track. + assertEquals(1, sentAddTrackCount()) + } + + @Test + fun micEnabledBetweenReconnectAttemptsSurvives() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + assertTrue(room.localParticipant.setMicrophoneEnabled(false)) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + + // App enables the mic during reconnect attempt 1; the publish lands on + // attempt 1's session. + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val micTrack = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)!!.track!! + + // Attempt 1 fails at the ICE wait; attempt 2 starts and joins. + testScheduler.advanceTimeBy(25_000) + reconnectWebsocket() + runCurrent() + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The publication created between attempts must be restored on attempt 2. + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertTrue(micTrack.enabled) + } + + @Test + fun micMutedDuringReconnectStaysMutedAcrossAttempts() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val micTrack = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)!!.track!! + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + + // App re-enables during attempt 1, then mutes; the mute is the newest state. + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + assertTrue(room.localParticipant.setMicrophoneEnabled(false)) + + // Attempt 1 fails at the ICE wait; attempt 2 joins and republishes. + testScheduler.advanceTimeBy(25_000) + reconnectWebsocket() + runCurrent() + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The mute must win over the older unmuted snapshot entry: nothing is + // republished and the track stays disabled. + assertNull(room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)) + assertFalse(micTrack.enabled) + + // Enabling afterwards publishes fresh and reactivates the track. + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertTrue(micTrack.enabled) + } + + @Test + fun republishRestartsTrackStoppedByFailedEnable() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val micTrack = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)!!.track!! + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + + // The app's enable gets no response and dies on the add track deadline, + // stopping the shared track on its failure path. + val deferredAddTrack = deferNextAddTrackResponse() + testScheduler.advanceTimeBy(500) + var micResult: Boolean? = null + val micJob = launch { + micResult = room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + assertNotNull(deferredAddTrack()) + + testScheduler.advanceTimeBy(25_000) + micJob.join() + assertEquals(false, micResult) + assertFalse(micTrack.enabled) + + // Attempt 1 failed at the ICE wait meanwhile; attempt 2 joins and republishes. + reconnectWebsocket() + runCurrent() + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The republished track must be live, not stopped. + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertTrue(micTrack.enabled) + } + + @Test + fun consumerStoppedTrackStaysStoppedAcrossReconnect() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val micTrack = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)!!.track!! + + // App stops the track through the public Track API without muting. + micTrack.stop() + assertFalse(micTrack.enabled) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The republish must restore the app's exact state: publication present + // and unmuted, track left stopped. + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertFalse(micTrack.enabled) + } + + @Test + fun consumerStopAfterRecoveredEnableFailureStaysStopped() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + // The first enable dies on the add track deadline, stopping the track and + // marking it as SDK-stopped. + val deferredAddTrack = deferNextAddTrackResponse() + var firstEnable: Boolean? = null + val firstEnableJob = launch { + firstEnable = room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + assertNotNull(deferredAddTrack()) + testScheduler.advanceTimeBy(21_000) + firstEnableJob.join() + assertEquals(false, firstEnable) + + // The retry succeeds; the marker no longer applies. + assertTrue(room.localParticipant.setMicrophoneEnabled(true)) + val micTrack = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE)!!.track!! + assertTrue(micTrack.enabled) + + // App stops the track through the public Track API without muting. + micTrack.stop() + assertFalse(micTrack.enabled) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The stale failure marker must not re-enable the consumer-stopped track. + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertFalse(micTrack.enabled) + } + + @Test + fun consumerStopAfterDirectPublishRecoveryStaysStopped() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + // The enable dies on the add track deadline, stopping the track and + // marking it as SDK-stopped. + val deferredAddTrack = deferNextAddTrackResponse() + var enableResult: Boolean? = null + val enableJob = launch { + enableResult = room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + assertNotNull(deferredAddTrack()) + testScheduler.advanceTimeBy(21_000) + enableJob.join() + assertEquals(false, enableResult) + + // App recovers by starting the track and publishing it directly; the + // marker no longer applies. + val micTrack = room.localParticipant.getOrCreateDefaultAudioTrack() + micTrack.start() + assertTrue(room.localParticipant.publishAudioTrack(micTrack)) + assertTrue(micTrack.enabled) + + // App stops the track through the public Track API without muting. + micTrack.stop() + assertFalse(micTrack.enabled) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The stale failure marker must not re-enable the consumer-stopped track. + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertFalse(micTrack.enabled) + } + + @Test + fun consumerStopAfterCrossSourceFailureMarkerStaysStopped() = runTest { + grantAudioPermission() + room.setReconnectionType(ReconnectType.FORCE_FULL_RECONNECT) + connect() + + // Publish the default track under a different source; withhold the response. + val deferredAddTrack = deferNextAddTrackResponse() + val micTrack = room.localParticipant.getOrCreateDefaultAudioTrack() + micTrack.start() + var publishResult: Boolean? = null + val publishJob = launch { + publishResult = room.localParticipant.publishAudioTrack( + micTrack, + AudioTrackPublishOptions(source = Track.Source.SCREEN_SHARE_AUDIO), + ) + } + runCurrent() + assertNotNull(deferredAddTrack()) + + // A concurrent enable under the microphone lock fails on the duplicate cid, + // stopping the track and recording a marker newer than the publish. + var micResult: Boolean? = null + val micJob = launch { + micResult = room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + micJob.join() + assertEquals(false, micResult) + assertFalse(micTrack.enabled) + + // The older publish completes; its clear must not erase the newer marker. + respondToAddTrack(deferredAddTrack()!!) + runCurrent() + publishJob.join() + assertEquals(true, publishResult) + + // Consumer recovers through the public Track API, then intentionally stops. + micTrack.start() + micTrack.stop() + assertFalse(micTrack.enabled) + + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + runCurrent() + connectPeerConnection() + connectPublisherPeerConnection() + advanceUntilIdle() + + // The marker predates the consumer's transitions and must not re-enable + // the track. + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertFalse(micTrack.enabled) + } + + @Test + fun enabledStateRevisionGuardsStaleTransitions() = runTest { + grantAudioPermission() + connect() + + val micTrack = room.localParticipant.getOrCreateDefaultAudioTrack() + micTrack.start() + val staleRevision = micTrack.enabledStateRevision.get() + + // A transition after the snapshot invalidates it. + micTrack.stop() + assertNull(micTrack.setEnabledIfRevisionUnchanged(staleRevision, true)) + assertFalse(micTrack.enabled) + + // An unchanged revision allows the transition. + val currentRevision = micTrack.enabledStateRevision.get() + assertNotNull(micTrack.setEnabledIfRevisionUnchanged(currentRevision, true)) + assertTrue(micTrack.enabled) + + // A stop's returned revision is that of the stop mutation itself, and any + // later transition advances past it. + val stopRevision = micTrack.stopReturningRevision() + assertEquals(stopRevision, micTrack.enabledStateRevision.get()) + micTrack.start() + assertTrue(micTrack.enabledStateRevision.get() > stopRevision) + } + + @Test + fun micEnableSerializesWithDirectPublish() = runTest { + grantAudioPermission() + connect() + + val deferredAddTrack = deferNextAddTrackResponse() + + val micTrack = room.localParticipant.getOrCreateDefaultAudioTrack() + var publishResult: Boolean? = null + val publishJob = launch { + publishResult = room.localParticipant.publishAudioTrack(micTrack) + } + runCurrent() + assertNotNull("direct publish addTrack should be in flight", deferredAddTrack()) + + var micResult: Boolean? = null + val micJob = launch { + micResult = room.localParticipant.setMicrophoneEnabled(true) + } + runCurrent() + assertNull(micResult) + + respondToAddTrack(deferredAddTrack()!!) + runCurrent() + publishJob.join() + micJob.join() + + assertEquals(true, publishResult) + assertEquals(true, micResult) + val finalPub = room.localParticipant.getTrackPublication(Track.Source.MICROPHONE) + assertNotNull(finalPub) + assertFalse(finalPub!!.muted) + assertTrue(micTrack.enabled) + } +}