Skip to content

Commit f5c4f0e

Browse files
andremionandremion
committed
Recover the connection state when a peer connection stays NEW after a reconnect
RealtimeConnection.Connected has a single writer after a reconnect: the ICE health monitor. Its recovery required both peer connections to reach an established ICE state, but a peer connection with nothing to negotiate stays NEW forever (the subscriber right after a reconnect with no inbound tracks), so the recovery never fired and the UI showed 'Reconnecting..' indefinitely. The nightly testReconnectionDuringCallRecording kept exhausting its retries on exactly this: socket reconnected, publisher ICE CONNECTED, subscriber NEW. - Extract the decision into iceHealthTransition and recover when the SFU socket is connected and no ICE side is bad (DISCONNECTED or FAILED), instead of requiring both sides to be established. A side that fails later still flips the state back through the degraded branch. - Re-evaluate on SFU socket state changes too, so recovery does not depend on a later ICE edge. - evaluateIceHealth is an internal member so the wrapper is directly testable; pure unit tests pin the contract on both ICE sides, including the exact state combination from the CI logs, and the recovery test fails against the old predicate. - Drop the structurally dead 'cause != TerminalSocketFailure' condition in joinInternal: the terminal case already returned earlier in the same when, so only the recovery outcome is left to check.
1 parent e6c6ad9 commit f5c4f0e

3 files changed

Lines changed: 184 additions & 40 deletions

File tree

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

Lines changed: 59 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -798,45 +798,25 @@ public class RtcSession internal constructor(
798798
}
799799
}
800800

801+
/** Applies [iceHealthTransition] to the current state. Internal for direct testing. */
802+
internal fun evaluateIceHealth() {
803+
val pubIce = publisher.value?.iceState?.value
804+
val subIce = subscriber.value?.iceState?.value
805+
val next = iceHealthTransition(
806+
connection = call.state.connection.value,
807+
sfuSocketConnected = _sfuSfuSocketState.value is SfuSocketState.Connected,
808+
publisherIce = pubIce,
809+
subscriberIce = subIce,
810+
)
811+
if (next != null) {
812+
logger.i { "[iceMonitor] pub=$pubIce, sub=$subIce — marking $next" }
813+
call.state._connection.value = next
814+
}
815+
}
816+
801817
private fun startIceMonitoring() {
802818
if (iceMonitoringJob?.isActive == true) return
803819
iceMonitoringJob = coroutineScope.launch {
804-
val badIceStates = setOf(
805-
PeerConnection.IceConnectionState.DISCONNECTED,
806-
PeerConnection.IceConnectionState.FAILED,
807-
)
808-
val goodIceStates = setOf(
809-
PeerConnection.IceConnectionState.CONNECTED,
810-
PeerConnection.IceConnectionState.COMPLETED,
811-
)
812-
813-
fun evaluateIceHealth() {
814-
val conn = call.state.connection.value
815-
val pubIce = publisher.value?.iceState?.value
816-
val subIce = subscriber.value?.iceState?.value
817-
818-
val pubBad = pubIce != null && pubIce in badIceStates
819-
val subBad = subIce != null && subIce in badIceStates
820-
821-
if ((pubBad || subBad) && conn is RealtimeConnection.Connected) {
822-
logger.w {
823-
"[iceMonitor] ICE degraded (pub=$pubIce, sub=$subIce) — marking Reconnecting"
824-
}
825-
call.state._connection.value = RealtimeConnection.Reconnecting
826-
} else if (conn is RealtimeConnection.Reconnecting &&
827-
_sfuSfuSocketState.value is SfuSocketState.Connected
828-
) {
829-
val pubOk = pubIce == null || pubIce in goodIceStates
830-
val subOk = subIce == null || subIce in goodIceStates
831-
if (pubOk && subOk) {
832-
logger.i {
833-
"[iceMonitor] ICE recovered (pub=$pubIce, sub=$subIce) — marking Connected"
834-
}
835-
call.state._connection.value = RealtimeConnection.Connected
836-
}
837-
}
838-
}
839-
840820
launch {
841821
publisher.collect { pub ->
842822
pub?.iceState?.collect { evaluateIceHealth() }
@@ -847,6 +827,12 @@ public class RtcSession internal constructor(
847827
sub?.iceState?.collect { evaluateIceHealth() }
848828
}
849829
}
830+
// The evaluation is edge-triggered by ICE changes, but after a reconnect the ICE
831+
// states can settle before the SFU socket reports Connected. Re-evaluate on socket
832+
// state changes too, so recovery does not depend on a later ICE transition.
833+
launch {
834+
_sfuSfuSocketState.collect { evaluateIceHealth() }
835+
}
850836
}
851837
}
852838

@@ -2209,7 +2195,43 @@ public class RtcSession internal constructor(
22092195
private fun connectInternalSafetyTimeoutMs(): Long =
22102196
clientImpl.connectionTimeoutInMs * 2 + CONNECT_INTERNAL_SAFETY_GRACE_MS
22112197

2212-
private companion object {
2198+
internal companion object {
2199+
private val badIceStates = setOf(
2200+
PeerConnection.IceConnectionState.DISCONNECTED,
2201+
PeerConnection.IceConnectionState.FAILED,
2202+
)
2203+
2204+
/**
2205+
* Decides the ICE health transition for the realtime connection, or null for no change.
2206+
*
2207+
* Degrades a Connected call when either peer connection reports a bad ICE state.
2208+
* Recovers a Reconnecting call once the SFU socket is connected and no side is bad.
2209+
* NEW and CHECKING count as healthy for the recovery: a peer connection with nothing
2210+
* to negotiate stays NEW forever (e.g. the subscriber right after a reconnect with no
2211+
* inbound tracks), so requiring an established state on both sides deadlocks the
2212+
* recovery and the UI shows "Reconnecting" indefinitely. If a side later fails, the
2213+
* degraded branch marks Reconnecting again.
2214+
*/
2215+
internal fun iceHealthTransition(
2216+
connection: RealtimeConnection,
2217+
sfuSocketConnected: Boolean,
2218+
publisherIce: PeerConnection.IceConnectionState?,
2219+
subscriberIce: PeerConnection.IceConnectionState?,
2220+
): RealtimeConnection? {
2221+
val pubBad = publisherIce != null && publisherIce in badIceStates
2222+
val subBad = subscriberIce != null && subscriberIce in badIceStates
2223+
return when {
2224+
(pubBad || subBad) && connection is RealtimeConnection.Connected ->
2225+
RealtimeConnection.Reconnecting
2226+
2227+
connection is RealtimeConnection.Reconnecting && sfuSocketConnected &&
2228+
!pubBad && !subBad ->
2229+
RealtimeConnection.Connected
2230+
2231+
else -> null
2232+
}
2233+
}
2234+
22132235
private const val CONNECT_INTERNAL_SAFETY_GRACE_MS = 1_000L
22142236
}
22152237
}

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -503,9 +503,9 @@ internal class CallJoinCoordinator(
503503
}
504504
}
505505

506-
if (sfuConnectionResult.cause != SfuConnectFailureCause.TerminalSocketFailure &&
507-
!didReconnectSucceed()
508-
) {
506+
// A terminal failure already returned above, so only recoverable causes
507+
// reach this point and the recovery outcome is the only condition left.
508+
if (!didReconnectSucceed()) {
509509
logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" }
510510
sendJoinErrorAnalytics(sfuConnectionResult)
511511
discardFailedSession(localSession)

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

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import io.getstream.video.android.core.Call
2424
import io.getstream.video.android.core.CallState
2525
import io.getstream.video.android.core.MediaManagerImpl
2626
import io.getstream.video.android.core.ParticipantState
27+
import io.getstream.video.android.core.RealtimeConnection
2728
import io.getstream.video.android.core.StreamVideo
2829
import io.getstream.video.android.core.StreamVideoClient
2930
import io.getstream.video.android.core.analytics.call.observer.SfuAnalytics
@@ -70,6 +71,7 @@ import kotlinx.coroutines.test.runTest
7071
import org.junit.After
7172
import org.junit.Before
7273
import org.junit.Test
74+
import org.webrtc.PeerConnection
7375
import org.webrtc.SessionDescription
7476
import stream.video.sfu.event.ReconnectDetails
7577
import stream.video.sfu.models.PeerType
@@ -863,6 +865,126 @@ class RtcSessionTest2 {
863865
}
864866
}
865867

868+
@Test
869+
fun `iceHealthTransition recovers a reconnecting call when no ICE side is bad`() {
870+
// The subscriber has nothing to negotiate after a reconnect and stays NEW; that must
871+
// not block the recovery (it deadlocked the connection state as Reconnecting forever).
872+
assertEquals(
873+
RealtimeConnection.Connected,
874+
RtcSession.iceHealthTransition(
875+
connection = RealtimeConnection.Reconnecting,
876+
sfuSocketConnected = true,
877+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
878+
subscriberIce = PeerConnection.IceConnectionState.NEW,
879+
),
880+
)
881+
// No peer connections at all: the connected socket is the only transport signal.
882+
assertEquals(
883+
RealtimeConnection.Connected,
884+
RtcSession.iceHealthTransition(
885+
connection = RealtimeConnection.Reconnecting,
886+
sfuSocketConnected = true,
887+
publisherIce = null,
888+
subscriberIce = null,
889+
),
890+
)
891+
}
892+
893+
@Test
894+
fun `iceHealthTransition does not recover while an ICE side is bad or the socket is down`() {
895+
assertNull(
896+
RtcSession.iceHealthTransition(
897+
connection = RealtimeConnection.Reconnecting,
898+
sfuSocketConnected = true,
899+
publisherIce = PeerConnection.IceConnectionState.DISCONNECTED,
900+
subscriberIce = PeerConnection.IceConnectionState.NEW,
901+
),
902+
)
903+
assertNull(
904+
RtcSession.iceHealthTransition(
905+
connection = RealtimeConnection.Reconnecting,
906+
sfuSocketConnected = true,
907+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
908+
subscriberIce = PeerConnection.IceConnectionState.DISCONNECTED,
909+
),
910+
)
911+
assertNull(
912+
RtcSession.iceHealthTransition(
913+
connection = RealtimeConnection.Reconnecting,
914+
sfuSocketConnected = false,
915+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
916+
subscriberIce = PeerConnection.IceConnectionState.CONNECTED,
917+
),
918+
)
919+
}
920+
921+
@Test
922+
fun `iceHealthTransition degrades a connected call when an ICE side goes bad`() {
923+
assertEquals(
924+
RealtimeConnection.Reconnecting,
925+
RtcSession.iceHealthTransition(
926+
connection = RealtimeConnection.Connected,
927+
sfuSocketConnected = true,
928+
publisherIce = PeerConnection.IceConnectionState.FAILED,
929+
subscriberIce = PeerConnection.IceConnectionState.NEW,
930+
),
931+
)
932+
assertEquals(
933+
RealtimeConnection.Reconnecting,
934+
RtcSession.iceHealthTransition(
935+
connection = RealtimeConnection.Connected,
936+
sfuSocketConnected = true,
937+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
938+
subscriberIce = PeerConnection.IceConnectionState.FAILED,
939+
),
940+
)
941+
assertNull(
942+
RtcSession.iceHealthTransition(
943+
connection = RealtimeConnection.Connected,
944+
sfuSocketConnected = true,
945+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
946+
subscriberIce = PeerConnection.IceConnectionState.NEW,
947+
),
948+
)
949+
}
950+
951+
@Test
952+
fun `evaluateIceHealth applies the transition to the connection state`() = runTest(
953+
testDispatcher,
954+
) {
955+
every { mockCallState.connection } returns
956+
MutableStateFlow<RealtimeConnection>(RealtimeConnection.Connected)
957+
val internalConnection = mockk<MutableStateFlow<RealtimeConnection>>(relaxed = true)
958+
every { mockCallState._connection } returns internalConnection
959+
960+
val rtcSession = RtcSession(
961+
client = mockStreamVideo,
962+
powerManager = mockPowerManager,
963+
call = mockCall,
964+
sessionManager = CallSessionManager(),
965+
sessionId = "test-session-id",
966+
apiKey = "test-api-key",
967+
lifecycle = mockLifecycle,
968+
sfuUrl = "https://test-sfu.stream.com",
969+
sfuWsUrl = "wss://test-sfu.stream.com",
970+
sfuToken = "fake-sfu-token",
971+
sfuName = "test-sfu-edge",
972+
clientImpl = mockVideoClient,
973+
coroutineScope = testScope,
974+
remoteIceServers = emptyList(),
975+
sfuConnectionModuleProvider = { mockk(relaxed = true) },
976+
sfuAnalytics = SfuAnalytics.getFakeSfuAnalytics(),
977+
)
978+
every { rtcSession.subscriber.value!!.iceState } returns
979+
MutableStateFlow<PeerConnection.IceConnectionState?>(
980+
PeerConnection.IceConnectionState.FAILED,
981+
)
982+
983+
rtcSession.evaluateIceHealth()
984+
985+
verify { internalConnection.value = RealtimeConnection.Reconnecting }
986+
}
987+
866988
private fun createRtcSessionSpyWithMockSocket(): Pair<RtcSession, Publisher> {
867989
val mockSocket = mockk<SfuSocketConnection>()
868990
val mockConnectedEvent = mockk<JoinCallResponseEvent>(relaxed = true)

0 commit comments

Comments
 (0)