Skip to content

Commit 64cb210

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.
1 parent 354ed25 commit 64cb210

2 files changed

Lines changed: 214 additions & 37 deletions

File tree

  • stream-video-android-core/src

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

Lines changed: 65 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,49 @@ 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+
// CLOSED must also block a recovery: a closed peer connection never emits another
2224+
// ICE event, so recovering past it would lock in a wrong Connected state. It is
2225+
// deliberately not a degrade trigger, because peer connections close during
2226+
// legitimate teardowns and the closing flow owns the connection state there.
2227+
val pubBlocked = pubBad || publisherIce == PeerConnection.IceConnectionState.CLOSED
2228+
val subBlocked = subBad || subscriberIce == PeerConnection.IceConnectionState.CLOSED
2229+
return when {
2230+
(pubBad || subBad) && connection is RealtimeConnection.Connected ->
2231+
RealtimeConnection.Reconnecting
2232+
2233+
connection is RealtimeConnection.Reconnecting && sfuSocketConnected &&
2234+
!pubBlocked && !subBlocked ->
2235+
RealtimeConnection.Connected
2236+
2237+
else -> null
2238+
}
2239+
}
2240+
22132241
private const val CONNECT_INTERNAL_SAFETY_GRACE_MS = 1_000L
22142242
}
22152243
}

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

Lines changed: 149 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,153 @@ 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+
// A closed peer connection never emits another ICE event, so it must block recovery.
920+
assertNull(
921+
RtcSession.iceHealthTransition(
922+
connection = RealtimeConnection.Reconnecting,
923+
sfuSocketConnected = true,
924+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
925+
subscriberIce = PeerConnection.IceConnectionState.CLOSED,
926+
),
927+
)
928+
assertNull(
929+
RtcSession.iceHealthTransition(
930+
connection = RealtimeConnection.Reconnecting,
931+
sfuSocketConnected = true,
932+
publisherIce = PeerConnection.IceConnectionState.CLOSED,
933+
subscriberIce = null,
934+
),
935+
)
936+
}
937+
938+
@Test
939+
fun `iceHealthTransition degrades a connected call when an ICE side goes bad`() {
940+
assertEquals(
941+
RealtimeConnection.Reconnecting,
942+
RtcSession.iceHealthTransition(
943+
connection = RealtimeConnection.Connected,
944+
sfuSocketConnected = true,
945+
publisherIce = PeerConnection.IceConnectionState.FAILED,
946+
subscriberIce = PeerConnection.IceConnectionState.NEW,
947+
),
948+
)
949+
assertEquals(
950+
RealtimeConnection.Reconnecting,
951+
RtcSession.iceHealthTransition(
952+
connection = RealtimeConnection.Connected,
953+
sfuSocketConnected = true,
954+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
955+
subscriberIce = PeerConnection.IceConnectionState.FAILED,
956+
),
957+
)
958+
assertNull(
959+
RtcSession.iceHealthTransition(
960+
connection = RealtimeConnection.Connected,
961+
sfuSocketConnected = true,
962+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
963+
subscriberIce = PeerConnection.IceConnectionState.NEW,
964+
),
965+
)
966+
// CLOSED does not degrade: peer connections close during legitimate teardowns and
967+
// the closing flow owns the connection state there.
968+
assertNull(
969+
RtcSession.iceHealthTransition(
970+
connection = RealtimeConnection.Connected,
971+
sfuSocketConnected = true,
972+
publisherIce = PeerConnection.IceConnectionState.CONNECTED,
973+
subscriberIce = PeerConnection.IceConnectionState.CLOSED,
974+
),
975+
)
976+
}
977+
978+
@Test
979+
fun `evaluateIceHealth applies the transition to the connection state`() = runTest(
980+
testDispatcher,
981+
) {
982+
every { mockCallState.connection } returns
983+
MutableStateFlow<RealtimeConnection>(RealtimeConnection.Connected)
984+
val internalConnection = mockk<MutableStateFlow<RealtimeConnection>>(relaxed = true)
985+
every { mockCallState._connection } returns internalConnection
986+
987+
val rtcSession = RtcSession(
988+
client = mockStreamVideo,
989+
powerManager = mockPowerManager,
990+
call = mockCall,
991+
sessionManager = CallSessionManager(),
992+
sessionId = "test-session-id",
993+
apiKey = "test-api-key",
994+
lifecycle = mockLifecycle,
995+
sfuUrl = "https://test-sfu.stream.com",
996+
sfuWsUrl = "wss://test-sfu.stream.com",
997+
sfuToken = "fake-sfu-token",
998+
sfuName = "test-sfu-edge",
999+
clientImpl = mockVideoClient,
1000+
coroutineScope = testScope,
1001+
remoteIceServers = emptyList(),
1002+
sfuConnectionModuleProvider = { mockk(relaxed = true) },
1003+
sfuAnalytics = SfuAnalytics.getFakeSfuAnalytics(),
1004+
)
1005+
every { rtcSession.subscriber.value!!.iceState } returns
1006+
MutableStateFlow<PeerConnection.IceConnectionState?>(
1007+
PeerConnection.IceConnectionState.FAILED,
1008+
)
1009+
1010+
rtcSession.evaluateIceHealth()
1011+
1012+
verify { internalConnection.value = RealtimeConnection.Reconnecting }
1013+
}
1014+
8661015
private fun createRtcSessionSpyWithMockSocket(): Pair<RtcSession, Publisher> {
8671016
val mockSocket = mockk<SfuSocketConnection>()
8681017
val mockConnectedEvent = mockk<JoinCallResponseEvent>(relaxed = true)

0 commit comments

Comments
 (0)