Skip to content

Replace sfu connect exception with sfu connect failure domain model #1745

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rahul-lohra wants to merge 3 commits into from
Closed

Replace sfu connect exception with sfu connect failure domain model #1745

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7aa184c
fix: Refactor with SfuConnectFailureCause code
rahul-lohra Jul 10, 2026
d29edb4
fix: Add kdoc
rahul-lohra Jul 10, 2026
af22c42
fix: Replace SfuConnectException with SfuConnectFailure which clearly
rahul-lohra Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 26 additions & 27 deletions stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -780,50 +780,49 @@ 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
// PARTICIPANT_NOT_FOUND and burn attempts before the loop escalates.
// 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(
WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN,
"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.",
),
)
}
}
}
Comment on lines +783 to 828

@coderabbitai coderabbitai Bot Jul 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win

SocketStateObservationTimeout branch never awaits the REJOIN it triggers — _join() falls through to Success.

Unlike the RecoverableSocketFailure branch, this branch fires scope.launch { reconnect(REJOIN, ...) } and then falls out of the when with no return, straight into val connectedSession = session.value ?: return Failure(...) (Line 829). Since session.value is already non-null (set at Line 773 before connectInternal() ran), execution proceeds to monitorSession(...) and return Success(...) immediately — without ever observing whether the REJOIN succeeded.

This contradicts JoinRecoverableFailureTest.socket state observation timeout starts a REJOIN itself, which expects _join() to stay suspended after triggering REJOIN and resolve to Failure only once the connection settles as ReconnectingFailed. As written, _join() returns Success synchronously, so that test would fail, and in production the join flow would falsely report success while the SFU connection actually timed out.

🐛 Proposed fix to await the triggered REJOIN, mirroring the RecoverableSocketFailure branch 
                     is SfuConnectFailure.SocketStateObservationTimeout -> {
                         logger.w {
                             "[_join] SFU socket state observation timed out with no recovery started — triggering REJOIN"
                         }
                         scope.launch {
                             reconnect(
                                 WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN,
                                 "join-recoverable-connect-failure",
                             )
                         }
+                        if (!didReconnectSucceed()) {
+                            logger.e { "[_join] Could not recover from socket state observation timeout. Error : $sfuConnectionResult" }
+                            sendJoinErrorAnalytics(sfuConnectionResult)
+                            return Failure(
+                                Error.GenericError(
+                                    sfuConnectionResult.cause.error.message ?: "SFU connection timed out",
+                                ),
+                            )
+                        }
                     }
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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
// PARTICIPANT_NOT_FOUND and burn attempts before the loop escalates.
// 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(
WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN,
"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.",
),
)
}
}
}
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
// PARTICIPANT_NOT_FOUND and burn attempts before the loop escalates.
// A full REJOIN re-fetches credentials and starts a clean join, which
// is the only thing that can actually succeed here.
logger.w {
"[_join] SFU socket state observation timed out with no recovery started — triggering REJOIN"
}
scope.launch {
reconnect(
WebsocketReconnectStrategy.WEBSOCKET_RECONNECT_STRATEGY_REJOIN,
"join-recoverable-connect-failure",
)
}
if (!didReconnectSucceed()) {
logger.e { "[_join] Could not recover from socket state observation timeout. Error : $sfuConnectionResult" }
sendJoinErrorAnalytics(sfuConnectionResult)
return Failure(
Error.GenericError(
sfuConnectionResult.cause.error.message ?: "SFU connection timed out",
),
)
}
}
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.cause.error.message ?: "RtcSession error occurred.",
),
)
}
}
}
}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/Call.kt`
around lines 783 - 828, Update the SocketStateObservationTimeout branch in
_join() to await the REJOIN result instead of launching it asynchronously and
falling through. Mirror the RecoverableSocketFailure handling: invoke reconnect
with REJOIN, await didReconnectSucceed(), and return an appropriate Failure
after recovery settles unsuccessfully, ensuring _join() does not proceed to
monitorSession or return Success prematurely.

Expand Down Expand Up @@ -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,
)
}
Expand Down Expand Up @@ -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)
}
}

Expand Down Expand Up @@ -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()
Expand Down
56 changes: 12 additions & 44 deletions stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
}
}

Expand Down Expand Up @@ -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,
)
}
Expand All @@ -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(),
)
}
Expand Down Expand Up @@ -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 =
Expand Down
45 changes: 45 additions & 0 deletions ...eo-android-core/src/main/kotlin/io/getstream/video/android/core/call/SfuConnectFailure.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
}
5 changes: 4 additions & 1 deletion ...d-core/src/test/kotlin/io/getstream/video/android/core/rtc/FastReconnectIceRestartTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand Down
42 changes: 22 additions & 20 deletions ...id-core/src/test/kotlin/io/getstream/video/android/core/rtc/JoinRecoverableFailureTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 {

Expand Down Expand Up @@ -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 {
Expand All @@ -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
Expand All @@ -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))

Expand Down
Loading
Loading