Single-flight Call.join to stop concurrent-join race

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

@@ -16,6 +16,7 @@
 
 package io.getstream.video.android.core.call.components
 
+import io.getstream.android.video.generated.models.JoinCallResponse
 import io.getstream.android.video.generated.models.RingCallRequest
 import io.getstream.log.taggedLogger
 import io.getstream.result.Error
@@ -38,10 +39,14 @@
 import io.getstream.video.android.core.call.SfuConnectFailureCause
 import io.getstream.video.android.core.call.SfuConnectionResult
 import io.getstream.video.android.core.model.toIceServer
+import io.getstream.video.android.core.utils.StreamRefCountedSingleFlightProcessor
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
 import stream.video.sfu.models.WebsocketReconnectStrategy
 
 /**
@@ -65,10 +70,39 @@
     private val callRegistry: ClientCallRegistry,
     private val hasRequiredPermissions: () -> Boolean,
 ) {
+    private companion object {
+        const val JOIN_FLIGHT_KEY = "join"
+    }
+
     private val logger by taggedLogger("Call:JoinCoordinator:$type:$id")
 
+    /**
+     * Coalesces concurrent [join] calls into one attempt on the call [scope].
+     *
+     * Without this, overlapping joins each build an [RtcSession] while reusing
+     * [CallSessionManager.sessionId], which leaves SFU-evicted zombies that fail every RPC
+     * with PARTICIPANT_NOT_FOUND. Checking [CallSessionManager.session] is not enough — it
+     * is only set after the coordinator round-trip.
+     *
+     * Coalescing the whole [join] also keeps once-per-join work once-only: JoinInitiated /
+     * MediaDevicePermission analytics, installing [CallState.callJoinInterceptor], resetting
+     * the leave guard, and moving to [RealtimeConnection.InProgress].
+     *
+     * [StreamRefCountedSingleFlightProcessor] keeps the join alive when one waiter (e.g. an
+     * Activity) is cancelled while others still await, and cancels it when the last waiter
+     * leaves.
+     */
+    private val joinFlight = StreamRefCountedSingleFlightProcessor(scope)
+
     private fun isVideoEnabled(): Boolean = state.settings.value?.video?.enabled ?: false
 
+    /**
+     * Joins the call, coalescing concurrent callers into one in-flight execution (single-flight).
+     *
+     * The shared work runs on the call [scope]. Each caller still awaits on its own coroutine,
+     * so destroying one UI scope only drops that waiter; remaining waiters keep the join.
+     * Cancelling the last waiter cancels the shared job.
+     */
     suspend fun join(
         create: Boolean = false,
         createOptions: CreateCallOptions? = null,
@@ -77,6 +111,44 @@
         hintHighScaleLivestreamPublisher: Boolean? = null,
         callJoinInterceptor: CallJoinInterceptor? = null,
     ): Result<RtcSession> {
+        return joinFlight.run(JOIN_FLIGHT_KEY) {
+            executeJoin(
+                create,
+                createOptions,
+                ring,
+                notify,
+                hintHighScaleLivestreamPublisher,
+                callJoinInterceptor,
+            )
+        }.fold(
+            onSuccess = { it },
+            onFailure = { error ->
+                Failure(
+                    Error.ThrowableError(
+                        message = error.message ?: "Join single-flight failed",
+                        cause = error,
+                    ),
+                )
+            },
+        )
+    }
+
+    private suspend fun executeJoin(
+        create: Boolean,
+        createOptions: CreateCallOptions?,
+        ring: Boolean,
+        notify: Boolean,
+        hintHighScaleLivestreamPublisher: Boolean?,
+        callJoinInterceptor: CallJoinInterceptor?,
+    ): Result<RtcSession> {
+        // Single already-joined gate for the whole join flow: subsequent join() calls while a
+        // session is live return that session instead of building a second one. Every retry
+        // below clears the session first, so [joinInternal] always starts without one.
+        sessionManager.session.value?.let { existing ->
+            logger.i { "[join] Call already joined — returning existing session" }
+            return Success(existing)
+        }
+
         callAnalytics.joinAnalytics.onJoinFunctionStart()
         callAnalytics.mediaPermissionObserver.mediaPermissionStatus()
         logger.d {
@@ -194,7 +266,7 @@
     }
 
     fun isPermanentError(error: Any): Boolean {
         if (error is Error.ThrowableError) {
             if (error.message.contains("Unable to resolve host")) {
                 return false
             }
@@ -202,6 +274,14 @@
         return true
     }
 
+    /**
+     * Performs one join attempt: coordinator round-trip, [RtcSession] creation and SFU connect.
+     *
+     * Assumes no session is active — [executeJoin] owns the already-joined check and clears the
+     * session before every retry. Calling this with a live session would build a second
+     * [RtcSession] for the same `sessionId` and produce the SFU-evicted zombie this coordinator
+     * exists to prevent.
+     */
     suspend fun joinInternal(
         create: Boolean = false,
         createOptions: CreateCallOptions? = null,
@@ -213,9 +293,6 @@
         sessionManager.nonFastReconnectAttempts = 0
         sessionMonitor.cancelSfuObservers()
 
-        if (sessionManager.session.value != null) {
-            return Failure(Error.GenericError("Call $type:$id has already been joined"))
-        }
         logger.d {
             "[joinInternal] #track; create: $create, ring: $ring, notify: $notify, createOptions: $createOptions"
         }
@@ -266,6 +343,29 @@
 
         state._connection.value = RealtimeConnection.Joined(localSession)
 
+        // Last-/sole-waiter cancel aborts this call-scoped job. If that happens after the
+        // session is installed, clear it — otherwise the idempotent join() path returns
+        // Success(zombie) and we keep a half-joined participant (PARTICIPANT_NOT_FOUND).
+        try {
+            return completeJoinAfterSessionInstall(localSession, result.value)
+        } catch (ce: CancellationException) {
+            withContext(NonCancellable) {
+                logger.w {
+                    "[joinInternal] Join cancelled after session install — discarding session"
+                }
+                discardFailedSession(localSession)
+                if (state._connection.value is RealtimeConnection.Joined) {
+                    state._connection.value = RealtimeConnection.Disconnected
+                }
+            }
+            throw ce
+        }
+    }
+
+    private suspend fun completeJoinAfterSessionInstall(
+        localSession: RtcSession,
+        joinResponse: JoinCallResponse,
+    ): Result<RtcSession> {
         // This is the SFU ws connection
         val sfuConnectionResult = localSession.connectInternal()
 
@@ -300,6 +400,7 @@
                             "[_join] Got terminal error while connecting to SFU. Error : $sfuConnectionResult"
                         }
                         sendJoinErrorAnalytics(sfuConnectionResult)
+                        discardFailedSession(localSession)
                         return Failure(
                             Error.GenericError(
                                 sfuConnectionResult.error.message ?: "RtcSession error occurred.",
@@ -308,10 +409,11 @@
                     }
                 }
 
                 if (sfuConnectionResult.cause != SfuConnectFailureCause.TerminalSocketFailure) {
                     if (!didReconnectSucceed()) {
                         logger.e { "[_join] Could not recover. Error : $sfuConnectionResult" }
                         sendJoinErrorAnalytics(sfuConnectionResult)
+                        discardFailedSession(localSession)
                         return Failure(
                             Error.GenericError(
                                 sfuConnectionResult.error.message ?: "SFU connection failed",
@@ -330,11 +432,33 @@
         // (re)establish monitoring when the session is unchanged, using the response that
         // still matches it, so we neither double-register nor monitor with a stale response.
         if (connectedSession === localSession) {
-            sessionMonitor.monitorSession(result.value)
+            sessionMonitor.monitorSession(joinResponse)
         }
         return Success(value = connectedSession)
     }
 
+    /**
+     * Tears down every session left after a failed join connect. Clearing the reference
+     * alone is not enough: sockets and peer connections stay alive and keep issuing SFU
+     * RPCs for a participant that is gone, which the SFU answers with PARTICIPANT_NOT_FOUND.
+     *
+     * Recoverable failures may already have swapped in a replacement via [CallReconnector]
+     * before [didReconnectSucceed] settles as failed. That replacement is not useful once
+     * join is returning Failure — tear it down too so nothing live is left behind.
+     */
+    private fun discardFailedSession(localSession: RtcSession) {
+        val active = sessionManager.session.value
+        logger.d {
+            "[joinInternal] Discarding session(s) after failed join connect " +
+                "(activeIsJoinSession=${active === localSession})"
+        }
+        sessionManager.setActiveSession(null)
+        if (active != null && active !== localSession) {
+            active.cleanup()
+        }
+        localSession.cleanup()
+    }
+
     /**
      * Reports the SFU WebSocket join failure to analytics. Only called from the join
      * flow ([joinInternal]) so that reconnect-driven [RtcSession.connectInternal] failures