fix(core): report SFU WebSocket connection/join timeouts accurately

Splits the SFU socket connect into a transport-open phase (bounded by
OkHttp's connect timeout) and a distinct join-response phase (bounded by
a dedicated timer via a new WebSocketConnected state), so a silent SFU no
longer hangs the join. Surfaces real transport failure messages and HTTP
status codes on the resulting NetworkError, routes all recoverable errors
through a single DisconnectedTemporarily state carrying the exact
code/reason, and maps both timeout flavours to REQUEST_TIMEOUT in analytics
(join-response via error code, transport via SocketTimeoutException cause).

Also moves join-error analytics to the join flow only, adds a per-session
SFU WS retry counter, and wires connectionTimeoutInMs from the builder to
both the OkHttp client and the join-response deadline.

Co-authored-by: Cursor <[email protected]>

Author: PratimMallick

stream-video-android-core/api/stream-video-android-core.api

@@ -12038,6 +12038,7 @@ public final class io/getstream/video/android/core/errors/VideoErrorCode : java/
 	public static final field NETWORK_FAILED Lio/getstream/video/android/core/errors/VideoErrorCode;
 	public static final field NO_ERROR_BODY Lio/getstream/video/android/core/errors/VideoErrorCode;
 	public static final field PARSER_ERROR Lio/getstream/video/android/core/errors/VideoErrorCode;
+	public static final field SFU_JOIN_RESPONSE_TIMEOUT Lio/getstream/video/android/core/errors/VideoErrorCode;
 	public static final field SOCKET_CLOSED Lio/getstream/video/android/core/errors/VideoErrorCode;
 	public static final field SOCKET_FAILURE Lio/getstream/video/android/core/errors/VideoErrorCode;
 	public static final field TOKEN_DATE_INCORRECT Lio/getstream/video/android/core/errors/VideoErrorCode;
@@ -14408,6 +14409,11 @@ public final class io/getstream/video/android/core/socket/common/StreamWebSocket
 	public fun toString ()Ljava/lang/String;
 }
 
+public final class io/getstream/video/android/core/socket/common/StreamWebSocketEvent$Open : io/getstream/video/android/core/socket/common/StreamWebSocketEvent {
+	public static final field INSTANCE Lio/getstream/video/android/core/socket/common/StreamWebSocketEvent$Open;
+	public fun toString ()Ljava/lang/String;
+}
+
 public final class io/getstream/video/android/core/socket/common/StreamWebSocketEvent$SfuMessage : io/getstream/video/android/core/socket/common/StreamWebSocketEvent {
 	public fun <init> (Lio/getstream/video/android/core/events/SfuDataEvent;)V
 	public final fun component1 ()Lio/getstream/video/android/core/events/SfuDataEvent;
@@ -14858,6 +14864,11 @@ public final class io/getstream/video/android/core/socket/sfu/state/SfuSocketSta
 	public fun toString ()Ljava/lang/String;
 }
 
+public final class io/getstream/video/android/core/socket/sfu/state/SfuSocketState$WebSocketConnected : io/getstream/video/android/core/socket/sfu/state/SfuSocketState {
+	public static final field INSTANCE Lio/getstream/video/android/core/socket/sfu/state/SfuSocketState$WebSocketConnected;
+	public fun toString ()Ljava/lang/String;
+}
+
 public abstract class io/getstream/video/android/core/socket/sfu/state/SfuSocketStateEvent {
 }
 
@@ -14935,6 +14946,11 @@ public final class io/getstream/video/android/core/socket/sfu/state/SfuSocketSta
 	public fun toString ()Ljava/lang/String;
 }
 
+public final class io/getstream/video/android/core/socket/sfu/state/SfuSocketStateEvent$WebSocketConnected : io/getstream/video/android/core/socket/sfu/state/SfuSocketStateEvent {
+	public static final field INSTANCE Lio/getstream/video/android/core/socket/sfu/state/SfuSocketStateEvent$WebSocketConnected;
+	public fun toString ()Ljava/lang/String;
+}
+
 public final class io/getstream/video/android/core/socket/sfu/state/SfuSocketStateEvent$WebSocketEventLost : io/getstream/video/android/core/socket/sfu/state/SfuSocketStateEvent {
 	public static final field INSTANCE Lio/getstream/video/android/core/socket/sfu/state/SfuSocketStateEvent$WebSocketEventLost;
 	public fun toString ()Ljava/lang/String;

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

@@ -112,6 +112,7 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
@@ -770,24 +771,81 @@ public class Call(
         }
         session.value = localSession
 
+        if (session.value == null) {
+            return Failure(Error.GenericError("RtcSession was null during connection to sfu"))
+        }
+
         session.value?.let {
             state._connection.value = RealtimeConnection.Joined(it)
         }
 
-        when (val result = session.value?.connectInternal()) {
-            is SfuConnectionResult.Connected -> Unit
-            is SfuConnectionResult.Failed ->
-                return Failure(
-                    Error.GenericError(result.error.message ?: "RtcSession error occurred."),
-                )
-            null ->
-                return Failure(Error.GenericError("RtcSession was null during connect"))
+        // This is the SFU ws connection
+        val sfuConnectionResult = session.value!!.connectInternal()
+
+        when (sfuConnectionResult) {
+            is SfuConnectionResult.Success -> Unit
+            is SfuConnectionResult.Failure -> {
+                if (sfuConnectionResult.recoverable) {
+                    logger.w { "[_join] Recoverable SFU connection failure — awaiting reconnect outcome" }
+                    if (!didReconnectSucceed()) {
+                        logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" }
+                        sendJoinErrorAnalytics(sfuConnectionResult)
+                        return Failure(
+                            Error.GenericError(
+                                sfuConnectionResult.error.message ?: "SFU connection failed",
+                            ),
+                        )
+                    }
+                } 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.",
+                        ),
+                    )
+                }
+            }
         }
         client.state.setActiveCall(this)
         monitorSession(result.value)
         return Success(value = session.value!!)
     }
 
+    /**
+     * Reports the SFU WebSocket join failure to analytics. Only called from the join
+     * flow ([_join]) so that reconnect-driven [RtcSession.connectInternal] failures are
+     * not counted as join errors. The retry count comes from the session's
+     * [RtcSession.sfuWsRetryCount]; the failure reason and abort code come straight from
+     * the [SfuConnectionResult.Failure] the connect attempt produced.
+     */
+    private fun sendJoinErrorAnalytics(failure: SfuConnectionResult.Failure) {
+        callAnalytics.sfuAnalytics.onSfuWsCompleted(
+            success = false,
+            retryCount = session.value?.sfuWsRetryCount?.get() ?: 0,
+            failureReason = failure.error.message,
+            failureCode = (failure.abortReason ?: AnalyticsCallAbortReason.SFU_ERROR).name,
+        )
+    }
+
+    /**
+     * Suspends until the reconnect loop triggered by a recoverable connection failure
+     * reaches a terminal state, returning `true` if the call recovered (became
+     * [RealtimeConnection.Connected]) and `false` otherwise
+     * ([RealtimeConnection.ReconnectingFailed] / [RealtimeConnection.Disconnected]).
+     */
+    private suspend fun didReconnectSucceed(): Boolean {
+        val terminal = state.connection.first {
+            it is RealtimeConnection.Connected ||
+                it is RealtimeConnection.ReconnectingFailed ||
+                it is RealtimeConnection.Disconnected
+        }
+        logger.d { "[_join] Reconnect after recoverable connection failure settled on $terminal" }
+        return terminal is RealtimeConnection.Connected
+    }
+
     private fun Call.monitorSession(result: JoinCallResponse) {
         sfuEvents?.cancel()
         sfuListener?.cancel()
@@ -1207,15 +1265,14 @@ public class Call(
             val result = newSession.connectInternal(
                 reconnectDetails,
                 currentOptions,
-                JoinAnalyticsModel(joinAnalyticsModel.retryAttempt),
             )
         ) {
-            is SfuConnectionResult.Connected -> {
+            is SfuConnectionResult.Success -> {
                 newSession.sfuTracer.trace("rejoin", reason)
                 monitorSession(joinResponse.value)
                 ReconnectOutcome.Success
             }
-            is SfuConnectionResult.Failed -> ReconnectOutcome.Failed(result.error)
+            is SfuConnectionResult.Failure -> ReconnectOutcome.Failed(result.error)
         }
     }
 
@@ -1291,14 +1348,13 @@ public class Call(
             val result = newSession.connectInternal(
                 reconnectDetails,
                 currentOptions,
-                JoinAnalyticsModel(joinAnalyticsModel.retryAttempt),
             )
             when (result) {
-                is SfuConnectionResult.Connected -> {
+                is SfuConnectionResult.Success -> {
                     monitorSession(joinResponse.value)
                     ReconnectOutcome.Success
                 }
-                is SfuConnectionResult.Failed -> ReconnectOutcome.Failed(result.error)
+                is SfuConnectionResult.Failure -> ReconnectOutcome.Failed(result.error)
             }
         } finally {
             oldSession.finalizeMigration()
@@ -2218,8 +2274,9 @@ public class Call(
     companion object {
         /** How many consecutive FAST reconnect failures are allowed before
          *  escalating to a full REJOIN. Kept small because each failed FAST
-         *  attempt can cost up to DEFAULT_SOCKET_TIMEOUT (10 s) waiting for
-         *  the WebSocket handshake to time out. */
+         *  attempt can cost up to the socket connection deadline before it gives
+         *  up: OkHttp's WebSocket-upgrade timeout, followed by the join-response
+         *  wait — both driven by StreamVideoBuilder.connectionTimeoutInMs (default 10s). */
         private const val MAX_FAST_RECONNECT_ATTEMPTS = 3
 
         /** Absolute upper bound on loop iterations across all strategies

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

@@ -84,7 +84,10 @@ import java.net.ConnectException
  * @property loggingLevel Represents and wraps SDK logging levels for Stream logger, HTTP interceptor and native WebRTC logging.
  * @property notificationConfig The configurations for handling push notification.
  * @property ringNotification Overwrite the default notification logic for incoming calls.
- * @property connectionTimeoutInMs Connection timeout in seconds.
+ * @property connectionTimeoutInMs Connection timeout in milliseconds. Applies both to the
+ * coordinator/SFU OkHttp clients (HTTP and WebSocket upgrade) and to the SFU join-response
+ * wait (time allowed for the SFU to deliver its JoinCallResponse after the socket opens).
+ * Defaults to 10s.
  * @property ensureSingleInstance Verify that only 1 version of the video client exists. Prevents integration mistakes.
  * @property videoDomain URL overwrite to allow for testing against a local instance of video.
  * @property callServiceConfig Configuration for the call foreground service. See [CallServiceConfig]. (Deprecated) Use `callServiceConfigRegistry` instead.
@@ -127,7 +130,7 @@ public class StreamVideoBuilder @JvmOverloads constructor(
     private val loggingLevel: LoggingLevel = LoggingLevel(),
     private val notificationConfig: NotificationConfig = NotificationConfig(),
     private val ringNotification: ((call: Call) -> Notification?)? = null,
-    private val connectionTimeoutInMs: Long = 10_000,
+    private val connectionTimeoutInMs: Long = 5_000,
     private var ensureSingleInstance: Boolean = true,
     private val videoDomain: String = "video.stream-io-api.com",
     @Deprecated(
@@ -297,6 +300,7 @@ public class StreamVideoBuilder @JvmOverloads constructor(
             appName = appName,
             audioProcessing = audioProcessing,
             loggingLevel = loggingLevel,
+            connectionTimeoutInMs = connectionTimeoutInMs,
             leaveAfterDisconnectSeconds = leaveAfterDisconnectSeconds,
             enableCallUpdatesAfterLeave = callUpdatesAfterLeave,
             enableStatsCollection = enableStatsReporting,

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

@@ -188,6 +188,7 @@ internal class StreamVideoClient internal constructor(
     internal val appName: String? = null,
     internal val audioProcessing: ManagedAudioProcessingFactory? = null,
     internal val loggingLevel: LoggingLevel = LoggingLevel(),
+    internal val connectionTimeoutInMs: Long = 10_000,
     internal val leaveAfterDisconnectSeconds: Long = 30,
     internal val appVersion: String? = null,
     internal val enableCallUpdatesAfterLeave: Boolean = false,