Fix publish deadlocks (#618)

* Fast fail attempts to publish without permissions

* Fix publish deadlock when no response from server

+---
+"client-sdk-android": patch
+---
+
+Fix publish deadlock when no response from server

+---
+"client-sdk-android": patch
+---
+
+Fast fail attempts to publish without permissions

@@ -54,8 +54,10 @@ import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.joinAll
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.withTimeoutOrNull
 import kotlinx.coroutines.yield
 import livekit.LivekitModels
@@ -87,7 +89,7 @@ import javax.inject.Singleton
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
-import kotlin.coroutines.suspendCoroutine
+import kotlin.time.Duration.Companion.seconds
 
 /**
  * @suppress
@@ -332,17 +334,19 @@ internal constructor(
         }
 
         // Suspend until signal client receives message confirming track publication.
-        return suspendCoroutine { cont ->
-            synchronized(pendingTrackResolvers) {
-                pendingTrackResolvers[cid] = cont
+        return withTimeout(20.seconds) {
+            suspendCancellableCoroutine { cont ->
+                synchronized(pendingTrackResolvers) {
+                    pendingTrackResolvers[cid] = cont
+                }
+                client.sendAddTrack(
+                    cid = cid,
+                    name = name,
+                    type = kind,
+                    stream = stream,
+                    builder = builder,
+                )
             }
-            client.sendAddTrack(
-                cid = cid,
-                name = name,
-                type = kind,
-                stream = stream,
-                builder = builder,
-            )
         }
     }
 

@@ -252,6 +252,7 @@ internal constructor(
      * @see Room.videoTrackCaptureDefaults
      * @see Room.videoTrackPublishDefaults
      */
+    @Throws(TrackException.PublishException::class)
     suspend fun setCameraEnabled(enabled: Boolean) {
         setTrackEnabled(Track.Source.CAMERA, enabled)
     }
@@ -266,6 +267,7 @@ internal constructor(
      * @see Room.audioTrackCaptureDefaults
      * @see Room.audioTrackPublishDefaults
      */
+    @Throws(TrackException.PublishException::class)
     suspend fun setMicrophoneEnabled(enabled: Boolean) {
         setTrackEnabled(Track.Source.MICROPHONE, enabled)
     }
@@ -286,6 +288,7 @@ internal constructor(
      * @see Room.screenShareTrackPublishDefaults
      * @see ScreenAudioCapturer
      */
+    @Throws(TrackException.PublishException::class)
     suspend fun setScreenShareEnabled(
         enabled: Boolean,
         mediaProjectionPermissionResultData: Intent? = null,
@@ -481,7 +484,27 @@ internal constructor(
         )
     }
 
+    private fun hasPermissionsToPublish(source: Track.Source): Boolean {
+        val permissions = this.permissions
+        if (permissions == null) {
+            LKLog.w { "No permissions present for publishing track." }
+            return false
+        }
+        val canPublish = permissions.canPublish
+        val canPublishSources = permissions.canPublishSources
+
+        val sourceAllowed = canPublishSources.contains(source)
+
+        if (canPublish && (canPublishSources.isEmpty() || sourceAllowed)) {
+            return true
+        }
+
+        LKLog.w { "insufficient permissions to publish" }
+        return false
+    }
+
     /**
+     * @throws TrackException.PublishException thrown when the publish fails. see [TrackException.PublishException.message] for details.
      * @return true if the track publish was successful.
      */
     private suspend fun publishTrackImpl(
@@ -491,6 +514,15 @@ internal constructor(
         encodings: List<RtpParameters.Encoding> = emptyList(),
         publishListener: PublishListener? = null,
     ): LocalTrackPublication? {
+        val addTrackRequestBuilder = AddTrackRequest.newBuilder().apply {
+            this.requestConfig()
+        }
+
+        val trackSource = Track.Source.fromProto(addTrackRequestBuilder.source ?: LivekitModels.TrackSource.UNRECOGNIZED)
+        if (!hasPermissionsToPublish(trackSource)) {
+            throw TrackException.PublishException("Failed to publish track, insufficient permissions")
+        }
+
         @Suppress("NAME_SHADOWING") var options = options
 
         @Suppress("NAME_SHADOWING") var encodings = encodings
@@ -564,17 +596,13 @@ internal constructor(
         }
 
         suspend fun requestAddTrack(): TrackInfo {
-            val builder = AddTrackRequest.newBuilder().apply {
-                this.requestConfig()
-            }
-
             return try {
                 engine.addTrack(
                     cid = cid,
                     name = options.name ?: track.name,
                     kind = track.kind.toProto(),
                     stream = options.stream,
-                    builder = builder,
+                    builder = addTrackRequestBuilder,
                 )
             } catch (e: Exception) {
                 val exception = TrackException.PublishException("Failed to publish track", e)
@@ -1362,8 +1390,12 @@ internal constructor(
                 )
             }
             negotiateJob.join()
-            val trackInfo = publishJob.await()
-            LKLog.d { "published $codec for track ${track.sid}, $trackInfo" }
+            try {
+                val trackInfo = publishJob.await()
+                LKLog.d { "published $codec for track ${track.sid}, $trackInfo" }
+            } catch (e: Exception) {
+                LKLog.w(e) { "exception when publishing $codec for track ${track.sid}" }
+            }
         }
     }
 

@@ -591,6 +591,9 @@ data class ParticipantPermission(
     val canPublishData: Boolean,
     val hidden: Boolean,
     val recorder: Boolean,
+    /**
+     * The list of allowed sources. If this is empty, then all sources are allowed.
+     */
     val canPublishSources: List<Track.Source>,
     val canUpdateMetadata: Boolean,
     val canSubscribeMetrics: Boolean,

@@ -56,13 +56,15 @@ object TestData {
         identity = "local_participant_identity"
         state = LivekitModels.ParticipantInfo.State.ACTIVE
         metadata = "local_metadata"
-        permission = LivekitModels.ParticipantPermission.newBuilder()
-            .setCanPublish(true)
-            .setCanSubscribe(true)
-            .setCanPublishData(true)
-            .setHidden(true)
-            .setRecorder(false)
-            .build()
+        permission = with(LivekitModels.ParticipantPermission.newBuilder()) {
+            canPublish = true
+            canSubscribe = true
+            canPublishData = true
+
+            hidden = false
+            recorder = false
+            build()
+        }
         putAttributes("attribute", "value")
         build()
     }

 /*
- * Copyright 2023-2024 LiveKit, Inc.
+ * Copyright 2023-2025 LiveKit, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,6 +27,7 @@ import io.livekit.android.room.DefaultsManager
 import io.livekit.android.room.track.LocalVideoTrack
 import io.livekit.android.room.track.LocalVideoTrackOptions
 import io.livekit.android.room.track.Track
+import io.livekit.android.room.track.TrackException
 import io.livekit.android.room.track.VideoCaptureParameter
 import io.livekit.android.room.track.VideoCodec
 import io.livekit.android.test.MockE2ETest
@@ -46,6 +47,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.advanceUntilIdle
 import livekit.LivekitModels
 import livekit.LivekitModels.AudioTrackFeature
@@ -66,6 +68,7 @@ import org.mockito.kotlin.argThat
 import org.robolectric.RobolectricTestRunner
 import org.robolectric.Shadows
 import java.nio.ByteBuffer
+import kotlin.time.Duration.Companion.seconds
 
 @ExperimentalCoroutinesApi
 @RunWith(RobolectricTestRunner::class)
@@ -122,11 +125,23 @@ class LocalParticipantMockE2ETest : MockE2ETest() {
         wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler)
         wsFactory.ws.clearRequests()
 
-        val backgroundScope = CoroutineScope(coroutineContext + Job())
+        val standardTestDispatcher = StandardTestDispatcher()
+        val backgroundScope = CoroutineScope(coroutineContext + Job() + standardTestDispatcher)
         try {
-            backgroundScope.launch { room.localParticipant.setMicrophoneEnabled(true) }
-            backgroundScope.launch { room.localParticipant.setMicrophoneEnabled(true) }
+            backgroundScope.launch {
+                try {
+                    room.localParticipant.setMicrophoneEnabled(true)
+                } catch (_: Exception) {
+                }
+            }
+            backgroundScope.launch {
+                try {
+                    room.localParticipant.setMicrophoneEnabled(true)
+                } catch (_: Exception) {
+                }
+            }
 
+            standardTestDispatcher.scheduler.advanceTimeBy(1.seconds.inWholeMilliseconds)
             assertEquals(1, wsFactory.ws.sentRequests.size)
         } finally {
             backgroundScope.cancel()
@@ -144,10 +159,23 @@ class LocalParticipantMockE2ETest : MockE2ETest() {
         wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler)
         wsFactory.ws.clearRequests()
 
-        val backgroundScope = CoroutineScope(coroutineContext + Job())
+        val standardTestDispatcher = StandardTestDispatcher()
+        val backgroundScope = CoroutineScope(coroutineContext + Job() + standardTestDispatcher)
         try {
-            backgroundScope.launch { room.localParticipant.setMicrophoneEnabled(true) }
-            backgroundScope.launch { room.localParticipant.setCameraEnabled(true) }
+            backgroundScope.launch {
+                try {
+                    room.localParticipant.setMicrophoneEnabled(true)
+                } catch (_: Exception) {
+                }
+            }
+            backgroundScope.launch {
+                try {
+                    room.localParticipant.setCameraEnabled(true)
+                } catch (_: Exception) {
+                }
+            }
+
+            standardTestDispatcher.scheduler.advanceTimeBy(1.seconds.inWholeMilliseconds)
 
             assertEquals(2, wsFactory.ws.sentRequests.size)
         } finally {
@@ -534,4 +562,49 @@ class LocalParticipantMockE2ETest : MockE2ETest() {
         assertTrue(features.contains(AudioTrackFeature.TF_AUTO_GAIN_CONTROL))
         assertFalse(features.contains(AudioTrackFeature.TF_ENHANCED_NOISE_CANCELLATION))
     }
+
+    @Test
+    fun lackOfPublishPermissionCausesException() = runTest {
+        val noCanPublishJoin = with(TestData.JOIN.toBuilder()) {
+            join = with(join.toBuilder()) {
+                participant = with(participant.toBuilder()) {
+                    permission = with(permission.toBuilder()) {
+                        canPublish = false
+                        build()
+                    }
+                    build()
+                }
+                build()
+            }
+            build()
+        }
+        connect(noCanPublishJoin)
+
+        var didThrow = false
+        try {
+            room.localParticipant.publishVideoTrack(createLocalTrack())
+        } catch (e: TrackException.PublishException) {
+            didThrow = true
+        }
+
+        assertTrue(didThrow)
+    }
+
+    @Test
+    fun publishWithNoResponseCausesException() = runTest {
+        connect()
+
+        wsFactory.unregisterSignalRequestHandler(wsFactory.defaultSignalRequestHandler)
+        var didThrow = false
+        launch {
+            try {
+                room.localParticipant.publishVideoTrack(createLocalTrack())
+            } catch (e: TrackException.PublishException) {
+                didThrow = true
+            }
+        }
+
+        coroutineRule.dispatcher.scheduler.advanceUntilIdle()
+        assertTrue(didThrow)
+    }
 }