Recover the connection state when a peer connection stays NEW after a reconnect

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt

@@ -574,8 +574,8 @@
     private val connectionConfiguration: PeerConnection.RTCConfiguration
         get() = buildConnectionConfiguration(iceServers)
 
     internal val subscriber: MutableStateFlow<Subscriber?> = MutableStateFlow(null)
     internal val publisher: MutableStateFlow<Publisher?> = MutableStateFlow(null)
 
     internal lateinit var sfuConnectionModule: SfuConnectionModule
 
@@ -798,45 +798,25 @@
         }
     }
 
+    /** Applies [iceHealthTransition] to the current state. Internal for direct testing. */
+    internal fun evaluateIceHealth() {
+        val pubIce = publisher.value?.iceState?.value
+        val subIce = subscriber.value?.iceState?.value
+        val next = iceHealthTransition(
+            connection = call.state.connection.value,
+            sfuSocketConnected = _sfuSfuSocketState.value is SfuSocketState.Connected,
+            publisherIce = pubIce,
+            subscriberIce = subIce,
+        )
+        if (next != null) {
+            logger.i { "[iceMonitor] pub=$pubIce, sub=$subIce — marking $next" }
+            call.state._connection.value = next
+        }
+    }
+
     private fun startIceMonitoring() {
         if (iceMonitoringJob?.isActive == true) return
         iceMonitoringJob = coroutineScope.launch {
-            val badIceStates = setOf(
-                PeerConnection.IceConnectionState.DISCONNECTED,
-                PeerConnection.IceConnectionState.FAILED,
-            )
-            val goodIceStates = setOf(
-                PeerConnection.IceConnectionState.CONNECTED,
-                PeerConnection.IceConnectionState.COMPLETED,
-            )
-
-            fun evaluateIceHealth() {
-                val conn = call.state.connection.value
-                val pubIce = publisher.value?.iceState?.value
-                val subIce = subscriber.value?.iceState?.value
-
-                val pubBad = pubIce != null && pubIce in badIceStates
-                val subBad = subIce != null && subIce in badIceStates
-
-                if ((pubBad || subBad) && conn is RealtimeConnection.Connected) {
-                    logger.w {
-                        "[iceMonitor] ICE degraded (pub=$pubIce, sub=$subIce) — marking Reconnecting"
-                    }
-                    call.state._connection.value = RealtimeConnection.Reconnecting
-                } else if (conn is RealtimeConnection.Reconnecting &&
-                    _sfuSfuSocketState.value is SfuSocketState.Connected
-                ) {
-                    val pubOk = pubIce == null || pubIce in goodIceStates
-                    val subOk = subIce == null || subIce in goodIceStates
-                    if (pubOk && subOk) {
-                        logger.i {
-                            "[iceMonitor] ICE recovered (pub=$pubIce, sub=$subIce) — marking Connected"
-                        }
-                        call.state._connection.value = RealtimeConnection.Connected
-                    }
-                }
-            }
-
             launch {
                 publisher.collect { pub ->
                     pub?.iceState?.collect { evaluateIceHealth() }
@@ -847,6 +827,12 @@
                     sub?.iceState?.collect { evaluateIceHealth() }
                 }
             }
+            // The evaluation is edge-triggered by ICE changes, but after a reconnect the ICE
+            // states can settle before the SFU socket reports Connected. Re-evaluate on socket
+            // state changes too, so recovery does not depend on a later ICE transition.
+            launch {
+                _sfuSfuSocketState.collect { evaluateIceHealth() }
+            }
         }
     }
 
@@ -922,7 +908,7 @@
         message = "Use connectInternal() which returns SfuConnectionResult instead of throwing.",
         replaceWith = ReplaceWith("connectInternal(reconnectDetails, options)"),
     )
     suspend fun connect(
         reconnectDetails: ReconnectDetails? = null,
         options: List<PublishOption>? = null,
     ) {
@@ -1620,7 +1606,7 @@
                                 paused = false,
                             )
 
                             if (event.trackType == TrackType.TRACK_TYPE_AUDIO) {
                                 if (event.sessionId == sessionId) {
                                     val isMicDisabled = !call.mediaManager.microphone.isEnabled.value
                                     if (isMicDisabled) {
@@ -2209,7 +2195,49 @@
     private fun connectInternalSafetyTimeoutMs(): Long =
         clientImpl.connectionTimeoutInMs * 2 + CONNECT_INTERNAL_SAFETY_GRACE_MS
 
-    private companion object {
+    internal companion object {
+        private val badIceStates = setOf(
+            PeerConnection.IceConnectionState.DISCONNECTED,
+            PeerConnection.IceConnectionState.FAILED,
+        )
+
+        /**
+         * Decides the ICE health transition for the realtime connection, or null for no change.
+         *
+         * Degrades a Connected call when either peer connection reports a bad ICE state.
+         * Recovers a Reconnecting call once the SFU socket is connected and no side is bad.
+         * NEW and CHECKING count as healthy for the recovery: a peer connection with nothing
+         * to negotiate stays NEW forever (e.g. the subscriber right after a reconnect with no
+         * inbound tracks), so requiring an established state on both sides deadlocks the
+         * recovery and the UI shows "Reconnecting" indefinitely. If a side later fails, the
+         * degraded branch marks Reconnecting again.
+         */
+        internal fun iceHealthTransition(
+            connection: RealtimeConnection,
+            sfuSocketConnected: Boolean,
+            publisherIce: PeerConnection.IceConnectionState?,
+            subscriberIce: PeerConnection.IceConnectionState?,
+        ): RealtimeConnection? {
+            val pubBad = publisherIce != null && publisherIce in badIceStates
+            val subBad = subscriberIce != null && subscriberIce in badIceStates
+            // CLOSED must also block a recovery: a closed peer connection never emits another
+            // ICE event, so recovering past it would lock in a wrong Connected state. It is
+            // deliberately not a degrade trigger, because peer connections close during
+            // legitimate teardowns and the closing flow owns the connection state there.
+            val pubBlocked = pubBad || publisherIce == PeerConnection.IceConnectionState.CLOSED
+            val subBlocked = subBad || subscriberIce == PeerConnection.IceConnectionState.CLOSED
+            return when {
+                (pubBad || subBad) && connection is RealtimeConnection.Connected ->
+                    RealtimeConnection.Reconnecting
+
+                connection is RealtimeConnection.Reconnecting && sfuSocketConnected &&
+                    !pubBlocked && !subBlocked ->
+                    RealtimeConnection.Connected
+
+                else -> null
+            }
+        }
+
         private const val CONNECT_INTERNAL_SAFETY_GRACE_MS = 1_000L
     }
 }

stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt

@@ -24,6 +24,7 @@ import io.getstream.video.android.core.Call
 import io.getstream.video.android.core.CallState
 import io.getstream.video.android.core.MediaManagerImpl
 import io.getstream.video.android.core.ParticipantState
+import io.getstream.video.android.core.RealtimeConnection
 import io.getstream.video.android.core.StreamVideo
 import io.getstream.video.android.core.StreamVideoClient
 import io.getstream.video.android.core.analytics.call.observer.SfuAnalytics
@@ -70,6 +71,7 @@ import kotlinx.coroutines.test.runTest
 import org.junit.After
 import org.junit.Before
 import org.junit.Test
+import org.webrtc.PeerConnection
 import org.webrtc.SessionDescription
 import stream.video.sfu.event.ReconnectDetails
 import stream.video.sfu.models.PeerType
@@ -863,6 +865,153 @@ class RtcSessionTest2 {
         }
     }
 
+    @Test
+    fun `iceHealthTransition recovers a reconnecting call when no ICE side is bad`() {
+        // The subscriber has nothing to negotiate after a reconnect and stays NEW; that must
+        // not block the recovery (it deadlocked the connection state as Reconnecting forever).
+        assertEquals(
+            RealtimeConnection.Connected,
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Reconnecting,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.CONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.NEW,
+            ),
+        )
+        // No peer connections at all: the connected socket is the only transport signal.
+        assertEquals(
+            RealtimeConnection.Connected,
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Reconnecting,
+                sfuSocketConnected = true,
+                publisherIce = null,
+                subscriberIce = null,
+            ),
+        )
+    }
+
+    @Test
+    fun `iceHealthTransition does not recover while an ICE side is bad or the socket is down`() {
+        assertNull(
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Reconnecting,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.DISCONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.NEW,
+            ),
+        )
+        assertNull(
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Reconnecting,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.CONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.DISCONNECTED,
+            ),
+        )
+        assertNull(
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Reconnecting,
+                sfuSocketConnected = false,
+                publisherIce = PeerConnection.IceConnectionState.CONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.CONNECTED,
+            ),
+        )
+        // A closed peer connection never emits another ICE event, so it must block recovery.
+        assertNull(
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Reconnecting,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.CONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.CLOSED,
+            ),
+        )
+        assertNull(
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Reconnecting,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.CLOSED,
+                subscriberIce = null,
+            ),
+        )
+    }
+
+    @Test
+    fun `iceHealthTransition degrades a connected call when an ICE side goes bad`() {
+        assertEquals(
+            RealtimeConnection.Reconnecting,
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Connected,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.FAILED,
+                subscriberIce = PeerConnection.IceConnectionState.NEW,
+            ),
+        )
+        assertEquals(
+            RealtimeConnection.Reconnecting,
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Connected,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.CONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.FAILED,
+            ),
+        )
+        assertNull(
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Connected,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.CONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.NEW,
+            ),
+        )
+        // CLOSED does not degrade: peer connections close during legitimate teardowns and
+        // the closing flow owns the connection state there.
+        assertNull(
+            RtcSession.iceHealthTransition(
+                connection = RealtimeConnection.Connected,
+                sfuSocketConnected = true,
+                publisherIce = PeerConnection.IceConnectionState.CONNECTED,
+                subscriberIce = PeerConnection.IceConnectionState.CLOSED,
+            ),
+        )
+    }
+
+    @Test
+    fun `evaluateIceHealth applies the transition to the connection state`() = runTest(
+        testDispatcher,
+    ) {
+        every { mockCallState.connection } returns
+            MutableStateFlow<RealtimeConnection>(RealtimeConnection.Connected)
+        val internalConnection = mockk<MutableStateFlow<RealtimeConnection>>(relaxed = true)
+        every { mockCallState._connection } returns internalConnection
+
+        val rtcSession = RtcSession(
+            client = mockStreamVideo,
+            powerManager = mockPowerManager,
+            call = mockCall,
+            sessionManager = CallSessionManager(),
+            sessionId = "test-session-id",
+            apiKey = "test-api-key",
+            lifecycle = mockLifecycle,
+            sfuUrl = "https://test-sfu.stream.com",
+            sfuWsUrl = "wss://test-sfu.stream.com",
+            sfuToken = "fake-sfu-token",
+            sfuName = "test-sfu-edge",
+            clientImpl = mockVideoClient,
+            coroutineScope = testScope,
+            remoteIceServers = emptyList(),
+            sfuConnectionModuleProvider = { mockk(relaxed = true) },
+            sfuAnalytics = SfuAnalytics.getFakeSfuAnalytics(),
+        )
+        every { rtcSession.subscriber.value!!.iceState } returns
+            MutableStateFlow<PeerConnection.IceConnectionState?>(
+                PeerConnection.IceConnectionState.FAILED,
+            )
+
+        rtcSession.evaluateIceHealth()
+
+        verify { internalConnection.value = RealtimeConnection.Reconnecting }
+    }
+
     private fun createRtcSessionSpyWithMockSocket(): Pair<RtcSession, Publisher> {
         val mockSocket = mockk<SfuSocketConnection>()
         val mockConnectedEvent = mockk<JoinCallResponseEvent>(relaxed = true)