Fix publish deadlock when no response from server

Author: davidliu

.changeset/chilly-kiwis-travel.md

@@ -0,0 +1,5 @@
+---
+"client-sdk-android": patch
+---
+
+Fix publish deadlock when no response from server

livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt

@@ -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,
-            )
         }
     }
 

livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt

@@ -1390,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}" }
+            }
         }
     }
 

livekit-android-test/src/test/java/io/livekit/android/room/participant/LocalParticipantMockE2ETest.kt

@@ -47,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
@@ -67,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)
@@ -123,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()
@@ -145,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 {
@@ -562,4 +589,22 @@ class LocalParticipantMockE2ETest : MockE2ETest() {
 
         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)
+    }
 }