diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt index 31a28bf572..97d0358f03 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt @@ -67,7 +67,7 @@ import io.getstream.video.android.core.analytics.reporting.model.AnalyticsCallAb import io.getstream.video.android.core.audio.StreamAudioDevice import io.getstream.video.android.core.call.FastReconnectResult import io.getstream.video.android.core.call.RtcSession -import io.getstream.video.android.core.call.SfuConnectException +import io.getstream.video.android.core.call.SfuConnectFailure import io.getstream.video.android.core.call.SfuConnectionResult import io.getstream.video.android.core.call.audio.InputAudioFilter import io.getstream.video.android.core.call.connection.StreamPeerConnectionFactory @@ -780,13 +780,8 @@ public class Call( when (sfuConnectionResult) { is SfuConnectionResult.Success -> Unit is SfuConnectionResult.Failure -> { - if (sfuConnectionResult.recoverable) { - // A recoverable socket disconnect means stateJob is already launching - // Call.reconnect, so we just await its outcome. A connect safety-timeout - // instead tore down a stuck socket into a state stateJob ignores, so - // nothing would start recovery and didReconnectSucceed() would block - // forever — trigger a REJOIN ourselves for that case. - if (sfuConnectionResult.error is SfuConnectException.Timeout) { + when (sfuConnectionResult.cause) { + is SfuConnectFailure.SocketStateObservationTimeout -> { // REJOIN (not FAST) on purpose: a connect timeout means the initial // join never completed. There is no established SFU session or // negotiated media path to resume, so a FAST resume would likely hit @@ -794,7 +789,7 @@ public class Call( // A full REJOIN re-fetches credentials and starts a clean join, which // is the only thing that can actually succeed here. logger.w { - "[_join] Recoverable SFU connect timeout with no recovery started — triggering REJOIN" + "[_join] SFU socket state observation timed out with no recovery started — triggering REJOIN" } scope.launch { reconnect( @@ -802,28 +797,32 @@ public class Call( "join-recoverable-connect-failure", ) } - } else { - logger.w { "[_join] Recoverable SFU connection failure — awaiting recovery outcome" } } - if (!didReconnectSucceed()) { - logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" } + + is SfuConnectFailure.RecoverableSocketFailure -> { + logger.w { "[_join] Recoverable SFU socket failure — awaiting recovery outcome" } + if (!didReconnectSucceed()) { + logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" } + sendJoinErrorAnalytics(sfuConnectionResult) + return Failure( + Error.GenericError( + sfuConnectionResult.cause.error.message ?: "SFU connection failed", + ), + ) + } + } + + is SfuConnectFailure.TerminalSocketFailure -> { + logger.e { + "[_join] Got terminal error while connecting to SFU. Error : $sfuConnectionResult" + } sendJoinErrorAnalytics(sfuConnectionResult) return Failure( Error.GenericError( - sfuConnectionResult.error.message ?: "SFU connection failed", + sfuConnectionResult.cause.error.message ?: "RtcSession error occurred.", ), ) } - } else { - logger.e { - "[_join] Got non recoverable error while connecting to SFU. Error : $sfuConnectionResult" - } - sendJoinErrorAnalytics(sfuConnectionResult) - return Failure( - Error.GenericError( - sfuConnectionResult.error.message ?: "RtcSession error occurred.", - ), - ) } } } @@ -851,7 +850,7 @@ public class Call( callAnalytics.sfuAnalytics.onSfuWsCompleted( success = false, retryCount = session.value?.sfuWsRetryCount?.get() ?: 0, - failureReason = failure.error.message, + failureReason = failure.cause.error.message, failureCode = (failure.abortReason ?: AnalyticsCallAbortReason.SFU_ERROR).name, ) } @@ -1298,7 +1297,7 @@ public class Call( monitorSession(joinResponse.value) ReconnectOutcome.Success } - is SfuConnectionResult.Failure -> ReconnectOutcome.Failed(result.error) + is SfuConnectionResult.Failure -> ReconnectOutcome.Failed(result.cause.error) } } @@ -1380,7 +1379,7 @@ public class Call( monitorSession(joinResponse.value) ReconnectOutcome.Success } - is SfuConnectionResult.Failure -> ReconnectOutcome.Failed(result.error) + is SfuConnectionResult.Failure -> ReconnectOutcome.Failed(result.cause.error) } } finally { oldSession.finalizeMigration() diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt index d855a77d42..70c8dc6bca 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt @@ -223,48 +223,18 @@ internal sealed class SfuConnectionResult { /** * The connection attempt failed. * - * @param error the typed cause of the failure (see [SfuConnectException]). - * Callers branch on its concrete type to decide how to recover — e.g. a - * [SfuConnectException.Timeout] means the safety-timeout tore down a stuck - * socket and no recovery loop was started, so the caller must start one. - * @param recoverable `true` when the failure can be recovered from, so the - * initial-join flow can await a recovery loop instead of treating it as - * permanent. `false` for terminal failures with no recovery (auth errors, - * permanent disconnects, etc.). + * @param cause the typed SFU connect failure, including its underlying error. * @param abortReason the analytics abort reason derived from the disconnect * state's error code, or `null` when the terminal state carried no error. * The caller (the join flow) decides whether/when to report it, so that * analytics are not emitted for reconnect-driven `connectInternal` calls. */ data class Failure( - val error: Exception, - val recoverable: Boolean = false, + val cause: SfuConnectFailure, val abortReason: AnalyticsCallAbortReason? = null, ) : SfuConnectionResult() } -/** - * Typed causes carried by [SfuConnectionResult.Failure], so callers of - * [RtcSession.connectInternal] can branch on the failure kind instead of parsing - * the message text. - */ -internal sealed class SfuConnectException(message: String) : Exception(message) { - /** - * The connect attempt never reached a terminal socket state within the safety - * timeout, so [RtcSession.connectInternal] tore the in-flight socket down. - * That leaves the socket in a state `stateJob` ignores, so no recovery loop is - * started on its own — the initial-join flow must start one itself. - */ - class Timeout(message: String) : SfuConnectException(message) - - /** - * The SFU socket reached a terminal [SfuSocketState.Disconnected] state. When - * the failure is recoverable, `stateJob` is already launching `Call.reconnect`, - * so the initial-join flow only needs to await the outcome. - */ - class Disconnected(message: String) : SfuConnectException(message) -} - /** * Outcome of [RtcSession.fastReconnect] — extends [SfuConnectionResult] * semantics with a [PeerConnectionStale] case for when the socket @@ -934,7 +904,7 @@ public class RtcSession internal constructor( ) { when (val result = connectInternal(reconnectDetails, options)) { is SfuConnectionResult.Success -> Unit - is SfuConnectionResult.Failure -> throw result.error + is SfuConnectionResult.Failure -> throw result.cause.error } } @@ -989,11 +959,8 @@ public class RtcSession internal constructor( // can't resurface through stateJob and resurrect this abandoned session. sfuConnectionModule.socketConnection.disconnect() sendCallStats() - // Typed as Timeout: we disconnected the socket into a state stateJob ignores, - // so no recovery loop starts on its own — the caller must start it. return SfuConnectionResult.Failure( - error = SfuConnectException.Timeout(msg), - recoverable = true, + cause = SfuConnectFailure.SocketStateObservationTimeout(Exception(msg)), abortReason = AnalyticsCallAbortReason.REQUEST_TIMEOUT, ) } @@ -1011,13 +978,14 @@ public class RtcSession internal constructor( logger.w { "[connectInternal] $msg" } sfuTracer.trace("connect-failed", msg) sendCallStats() - // Recoverable only when the terminal state is one stateJob reacts to; in that - // case a Call.reconnect is already being launched, so the join flow just awaits. - val recoverable = - (sfuSocketState as? SfuSocketState.Disconnected)?.isRecoverable() ?: false + val error = Exception(msg) + val failure = if ((sfuSocketState as? SfuSocketState.Disconnected)?.isRecoverable() == true) { + SfuConnectFailure.RecoverableSocketFailure(error) + } else { + SfuConnectFailure.TerminalSocketFailure(error) + } SfuConnectionResult.Failure( - error = SfuConnectException.Disconnected(msg), - recoverable = recoverable, + cause = failure, abortReason = networkError?.toAbortReason(), ) } @@ -2057,7 +2025,7 @@ public class RtcSession internal constructor( publisher.value?.currentOptions(), ) if (connectResult is SfuConnectionResult.Failure) { - return FastReconnectResult.Failed(connectResult.error) + return FastReconnectResult.Failed(connectResult.cause.error) } val peerConnectionClosed = diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/SfuConnectFailure.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/SfuConnectFailure.kt new file mode 100644 index 0000000000..676f7df9dc --- /dev/null +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/SfuConnectFailure.kt @@ -0,0 +1,45 @@ +/* + * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved. + * + * Licensed under the Stream License; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://github.com/GetStream/stream-video-android/blob/main/LICENSE + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.getstream.video.android.core.call + +/** + * Typed SFU connect failure produced by [RtcSession.connectInternal]. + */ +internal sealed interface SfuConnectFailure { + val error: Exception + + /** + * [RtcSession.connectInternal] stopped waiting for the socket to reach a terminal state. + */ + data class SocketStateObservationTimeout( + override val error: Exception, + ) : SfuConnectFailure + + /** + * The socket failure is already being recovered by [RtcSession.stateJob]. + */ + data class RecoverableSocketFailure( + override val error: Exception, + ) : SfuConnectFailure + + /** + * The socket failure should fail immediately without recovery. + */ + data class TerminalSocketFailure( + override val error: Exception, + ) : SfuConnectFailure +} diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/FastReconnectIceRestartTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/FastReconnectIceRestartTest.kt index 369eedfaf1..22e3ade14c 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/FastReconnectIceRestartTest.kt +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/FastReconnectIceRestartTest.kt @@ -29,6 +29,7 @@ import io.getstream.video.android.core.StreamVideoClient import io.getstream.video.android.core.analytics.call.observer.SfuAnalytics import io.getstream.video.android.core.call.FastReconnectResult import io.getstream.video.android.core.call.RtcSession +import io.getstream.video.android.core.call.SfuConnectFailure import io.getstream.video.android.core.call.SfuConnectionResult import io.getstream.video.android.core.call.connection.Publisher import io.getstream.video.android.core.call.connection.Subscriber @@ -243,7 +244,9 @@ class FastReconnectIceRestartTest { session.subscriber.value = sub val error = Exception("SFU connection timed out") - coEvery { session.connectInternal(any(), any()) } returns SfuConnectionResult.Failure(error) + coEvery { session.connectInternal(any(), any()) } returns SfuConnectionResult.Failure( + SfuConnectFailure.TerminalSocketFailure(error), + ) val result = session.fastReconnect(null) diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/JoinRecoverableFailureTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/JoinRecoverableFailureTest.kt index 10a115146e..66ced74306 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/JoinRecoverableFailureTest.kt +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/JoinRecoverableFailureTest.kt @@ -28,7 +28,7 @@ import io.getstream.video.android.core.analytics.call.observer.model.JoinAnalyti import io.getstream.video.android.core.analytics.call.observer.model.JoinReason import io.getstream.video.android.core.base.DispatcherRule import io.getstream.video.android.core.call.RtcSession -import io.getstream.video.android.core.call.SfuConnectException +import io.getstream.video.android.core.call.SfuConnectFailure import io.getstream.video.android.core.call.SfuConnectionResult import io.getstream.video.android.core.internal.module.CoordinatorConnectionModule import io.getstream.video.android.core.internal.network.NetworkStateProvider @@ -54,18 +54,14 @@ import org.junit.Test import stream.video.sfu.models.WebsocketReconnectStrategy /** - * Tests the initial-join handling of a recoverable SFU connection failure - * (e.g. a connection timeout) in [Call._join]. + * Tests the initial-join handling of failed SFU connect attempts in [Call._join]. * - * A recoverable failure's [SfuConnectionResult.Failure.error] type decides the path: - * - [SfuConnectException.Disconnected] — the RtcSession's stateJob is already launching - * [Call.reconnect], so `_join` defers to that single recovery loop and awaits its - * terminal outcome. - * - [SfuConnectException.Timeout] — nothing will start recovery on its own (the connect - * safety-timeout tore the socket down into a state stateJob ignores), so `_join` must - * start the reconnect itself before awaiting. - * - * A non-recoverable failure must fail immediately. + * The failure cause decides how `_join` orchestrates recovery: + * - [SfuConnectFailure.SocketStateObservationTimeout] starts a REJOIN + * because no reconnect loop was started by stateJob. + * - [SfuConnectFailure.RecoverableSocketFailure] waits for the reconnect + * loop already started by stateJob. + * - [SfuConnectFailure.TerminalSocketFailure] fails immediately. */ class JoinRecoverableFailureTest { @@ -131,13 +127,14 @@ class JoinRecoverableFailureTest { } @Test - fun `recoverable socket disconnect awaits the existing reconnect loop`() = runTest( + fun `recoverable socket failure awaits the existing reconnect loop`() = runTest( testDispatcher, ) { coEvery { mockSession.connectInternal(any(), any()) } returns SfuConnectionResult.Failure( - SfuConnectException.Disconnected("SFU socket disconnected"), - recoverable = true, + SfuConnectFailure.RecoverableSocketFailure( + Exception("SFU socket disconnected"), + ), ) val deferred = async { @@ -157,13 +154,14 @@ class JoinRecoverableFailureTest { } @Test - fun `recoverable connect timeout starts a REJOIN itself`() = runTest( + fun `socket state observation timeout starts a REJOIN itself`() = runTest( testDispatcher, ) { coEvery { mockSession.connectInternal(any(), any()) } returns SfuConnectionResult.Failure( - SfuConnectException.Timeout("SFU connection timed out"), - recoverable = true, + SfuConnectFailure.SocketStateObservationTimeout( + Exception("SFU connection timed out"), + ), ) // Stub the loop so we only assert it is invoked, not run it for real. coEvery { call.reconnect(any(), any()) } returns Unit @@ -189,11 +187,15 @@ class JoinRecoverableFailureTest { } @Test - fun `non-recoverable failure fails immediately without awaiting reconnect`() = runTest( + fun `terminal socket failure fails immediately without awaiting reconnect`() = runTest( testDispatcher, ) { coEvery { mockSession.connectInternal(any(), any()) } returns - SfuConnectionResult.Failure(Exception("permanent auth error"), recoverable = false) + SfuConnectionResult.Failure( + SfuConnectFailure.TerminalSocketFailure( + Exception("permanent auth error"), + ), + ) val result = call._join(joinAnalyticsModel = JoinAnalyticsModel(0, JoinReason.FirstAttempt)) diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt index 45b03c8ce1..ec3dacf43e 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt @@ -29,7 +29,7 @@ import io.getstream.video.android.core.StreamVideoClient import io.getstream.video.android.core.analytics.call.observer.SfuAnalytics import io.getstream.video.android.core.analytics.reporting.model.AnalyticsCallAbortReason import io.getstream.video.android.core.call.RtcSession -import io.getstream.video.android.core.call.SfuConnectException +import io.getstream.video.android.core.call.SfuConnectFailure import io.getstream.video.android.core.call.SfuConnectionResult import io.getstream.video.android.core.call.connection.Publisher import io.getstream.video.android.core.errors.VideoErrorCode @@ -53,7 +53,6 @@ import io.mockk.spyk import io.mockk.unmockkAll import io.mockk.verify import junit.framework.TestCase.assertEquals -import junit.framework.TestCase.assertFalse import junit.framework.TestCase.assertNotNull import junit.framework.TestCase.assertNull import junit.framework.TestCase.assertTrue @@ -295,20 +294,23 @@ class RtcSessionTest2 { ) assertTrue( "Expected timeout message", - (result as SfuConnectionResult.Failure).error.message!!.contains("timed out"), + (result as SfuConnectionResult.Failure).cause.error.message!!.contains( + "timed out", + ), + ) + assertTrue( + "Expected RecoverableSocketFailure", + result.cause is SfuConnectFailure.RecoverableSocketFailure, ) - // A timeout is recoverable — the stateJob triggers a reconnect, so the - // join flow should await it rather than failing permanently. - assertTrue("Expected recoverable=true for a timeout", result.recoverable) assertEquals( AnalyticsCallAbortReason.REQUEST_TIMEOUT, - (result as SfuConnectionResult.Failure).abortReason, + result.abortReason, ) } @Suppress("DEPRECATION") @Test - fun `connectInternal safety timeout fires when socket stays non-terminal`() = + fun `connectInternal socket state observation timeout fires when socket stays non-terminal`() = runTest(testDispatcher) { val sfuSocketStateFlow = MutableStateFlow( SfuSocketState.Disconnected.Stopped, @@ -346,7 +348,7 @@ class RtcSessionTest2 { coJustRun { rtcSession.sendCallStats(any(), any(), any()) } val resultDeferred = async { rtcSession.connectInternal() } - // Safety timeout = 2 * 50ms + 1000ms grace + // Socket state observation timeout = 2 * 50ms + 1000ms grace advanceTimeBy(1_101L) val result = resultDeferred.await() @@ -355,15 +357,14 @@ class RtcSessionTest2 { result is SfuConnectionResult.Failure, ) assertTrue( - "Expected safety-timeout message", - (result as SfuConnectionResult.Failure).error.message!!.contains("timed out"), + "Expected socket state observation timeout message", + (result as SfuConnectionResult.Failure).cause.error.message!!.contains( + "timed out", + ), ) - assertTrue("Expected recoverable=true for a safety timeout", result.recoverable) - // A Timeout-typed error is what makes _join self-trigger a REJOIN for this path, - // since stateJob ignores the state the stuck socket was torn down into. assertTrue( - "Expected a SfuConnectException.Timeout error for a safety timeout", - result.error is SfuConnectException.Timeout, + "Expected SocketStateObservationTimeout", + result.cause is SfuConnectFailure.SocketStateObservationTimeout, ) assertEquals( AnalyticsCallAbortReason.REQUEST_TIMEOUT, @@ -429,7 +430,10 @@ class RtcSessionTest2 { AnalyticsCallAbortReason.REQUEST_TIMEOUT, (result as SfuConnectionResult.Failure).abortReason, ) - assertTrue("Expected recoverable=true for a transport timeout", result.recoverable) + assertTrue( + "Expected RecoverableSocketFailure", + result.cause is SfuConnectFailure.RecoverableSocketFailure, + ) } @Suppress("DEPRECATION") @@ -480,9 +484,9 @@ class RtcSessionTest2 { "Expected SfuConnectionResult.Failure but got $result", result is SfuConnectionResult.Failure, ) - assertFalse( - "Expected recoverable=false for a permanent disconnect", - (result as SfuConnectionResult.Failure).recoverable, + assertTrue( + "Expected TerminalSocketFailure", + (result as SfuConnectionResult.Failure).cause is SfuConnectFailure.TerminalSocketFailure, ) }