#721 Fix publisher negotiation reliability by preventing overlapping renegotiations

.changeset/fix-publisher-negotiation-race.md

@@ -0,0 +1,5 @@
+---
+"client-sdk-android": patch
+---
+
+Fix publisher negotiation reliability by preventing overlapping renegotiations.

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

@@ -36,6 +36,7 @@ import io.livekit.android.room.util.MediaConstraintKeys
 import io.livekit.android.room.util.createAnswer
 import io.livekit.android.room.util.setLocalDescription
 import io.livekit.android.room.util.waitUntilConnected
+import io.livekit.android.room.util.waitUntilConnectedOrDisconnected
 import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.Either
 import io.livekit.android.util.FlowObservable
@@ -194,6 +195,7 @@ internal constructor(
     private var isSubscriberPrimary = false
     private var isClosed = true
 
+    @Volatile
     private var hasPublished = false
 
     private var coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher)
@@ -670,21 +672,24 @@ internal constructor(
     }
 
     internal fun negotiatePublisher() {
+        // Mark intent to publish before checking Signal state so reconnect() can
+        // re-trigger negotiation even if the first attempt occurs before Signal connects.
+        hasPublished = true
+
         if (!client.isConnected) {
             return
         }
 
-        hasPublished = true
-
         coroutineScope.launch {
-            if (negotiatePublisherMutex.tryLock()) {
-                try {
-                    publisher?.negotiate?.invoke(getPublisherOfferConstraints())
-                } finally {
-                    negotiatePublisherMutex.unlock()
+            // Ensure only one negotiation sequence runs at a time.
+            negotiatePublisherMutex.withLock {
+                publisher?.negotiate?.invoke(getPublisherOfferConstraints())
+
+                // Best-effort wait for publisher to connect before releasing mutex
+                // to avoid overlapping renegotiations.
+                withTimeoutOrNull(MAX_ICE_CONNECT_TIMEOUT_MS.toLong()) {
+                    publisherObserver.waitUntilConnectedOrDisconnected()
                 }
-            } else {
-                LKLog.v { "negotiatePublisher: skipping, negotiation already in progress" }
             }
         }
     }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 LiveKit, Inc.
+ * Copyright 2024-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.
@@ -19,6 +19,7 @@ package io.livekit.android.room.util
 import io.livekit.android.util.FlowObservable
 import io.livekit.android.util.flow
 import io.livekit.android.webrtc.isConnected
+import io.livekit.android.webrtc.isDisconnected
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.takeWhile
 import livekit.org.webrtc.PeerConnection.PeerConnectionState
@@ -39,3 +40,18 @@ internal suspend fun PeerConnectionStateObservable.waitUntilConnected() {
         }
         .collect()
 }
+
+/**
+ * Waits until the connection reaches a stable terminal state:
+ * [PeerConnectionState.isConnected] or [PeerConnectionState.isDisconnected].
+ *
+ * Note: [PeerConnectionState.isDisconnected] intentionally excludes the temporary
+ * [PeerConnectionState.DISCONNECTED] state and only treats FAILED/CLOSED as disconnected.
+ */
+internal suspend fun PeerConnectionStateObservable.waitUntilConnectedOrDisconnected() {
+    this::connectionState.flow
+        .takeWhile {
+            !it.isConnected() && !it.isDisconnected()
+        }
+        .collect()
+}