fix(core): 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

@@ -38,10 +38,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 kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import stream.video.sfu.models.WebsocketReconnectStrategy
 
 /**
@@ -67,15 +71,85 @@
 ) {
     private val logger by taggedLogger("Call:JoinCoordinator:$type:$id")
 
+    /**
+     * Single-flight bookkeeping for [join] (same idea as
+     * [io.getstream.video.android.core.utils.StreamSingleFlightProcessorImpl], but the shared
+     * work runs on the **caller's** coroutine so a ViewModel/UI cancel still aborts the join).
+     *
+     * Concurrent [join] calls must share one attempt: each would otherwise build its own
+     * [RtcSession] while reusing [CallSessionManager.sessionId], leaving 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].
+     *
+     * Held only long enough to read or publish [joinFlight], never across the join itself.
+     */
+    private val joinMutex = Mutex()
+
+    /** The in-flight [join], if any. Completed flights are never reused. */
+    private var joinFlight: CompletableDeferred<Result<RtcSession>>? = null
+
     private fun isVideoEnabled(): Boolean = state.settings.value?.video?.enabled ?: false
 
+    /**
+     * Joins the call, coalescing concurrent callers into one in-flight execution (single-flight).
+     * Additional callers await the same [Result]. The winner runs on its own caller coroutine.
+     */
     suspend fun join(
         create: Boolean = false,
         createOptions: CreateCallOptions? = null,
         ring: Boolean = false,
         notify: Boolean = false,
         hintHighScaleLivestreamPublisher: Boolean? = null,
         callJoinInterceptor: CallJoinInterceptor? = null,
+    ): Result<RtcSession> {
+        val (flight, isWinner) = joinMutex.withLock {
+            val running = joinFlight?.takeUnless { it.isCompleted }
+            if (running != null) {
+                logger.i {
+                    "[join] Single-flight: join already in flight — awaiting its result"
+                }
+                running to false
+            } else {
+                CompletableDeferred<Result<RtcSession>>().also { joinFlight = it } to true
+            }
+        }
+
+        if (isWinner) {
+            // Winner executes on this caller's coroutine (e.g. viewModelScope). Followers
+            // only await [flight]; cancelling the winner cancels the shared join for them too.
+            try {
+                val result = executeJoin(
+                    create,
+                    createOptions,
+                    ring,
+                    notify,
+                    hintHighScaleLivestreamPublisher,
+                    callJoinInterceptor,
+                )
+                flight.complete(result)
+                return result
+            } catch (e: CancellationException) {
+                flight.cancel(e)
+                throw e
+            } catch (e: Throwable) {
+                flight.completeExceptionally(e)
+                throw e
+            }
+        }
+        return flight.await()
+    }
+
+    private suspend fun executeJoin(
+        create: Boolean,
+        createOptions: CreateCallOptions?,
+        ring: Boolean,
+        notify: Boolean,
+        hintHighScaleLivestreamPublisher: Boolean?,
+        callJoinInterceptor: CallJoinInterceptor?,
     ): Result<RtcSession> {
         callAnalytics.joinAnalytics.onJoinFunctionStart()
         callAnalytics.mediaPermissionObserver.mediaPermissionStatus()
@@ -194,7 +268,7 @@
     }
 
     fun isPermanentError(error: Any): Boolean {
         if (error is Error.ThrowableError) {
             if (error.message.contains("Unable to resolve host")) {
                 return false
             }
@@ -300,6 +374,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 +383,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",
@@ -335,6 +411,28 @@
         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

stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt

@@ -24,6 +24,7 @@ import io.getstream.android.video.generated.models.RingCallResponse
 import io.getstream.result.Error
 import io.getstream.result.Result.Failure
 import io.getstream.result.Result.Success
+import io.getstream.video.android.core.CallJoinInterceptor
 import io.getstream.video.android.core.CallLeaveReason
 import io.getstream.video.android.core.CallState
 import io.getstream.video.android.core.RealtimeConnection
@@ -41,6 +42,9 @@ import io.mockk.every
 import io.mockk.mockk
 import io.mockk.unmockkAll
 import io.mockk.verify
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.TestScope
@@ -232,6 +236,135 @@ class CallJoinCoordinatorTest {
         assertThat(coordinator.isPermanentError(permanent)).isTrue()
     }
 
+    @Test
+    fun `concurrent joins issue a single coordinator join and share one session`() = runTest(
+        testDispatcher,
+    ) {
+        stubJoinCall(Success(mockJoinResponse))
+        // Suspends until released, so all callers are inside join at the same time —
+        // which is exactly the window the old session.value check failed to cover.
+        val connectGate = CompletableDeferred<Unit>()
+        coEvery { mockSession.connectInternal() } coAnswers {
+            connectGate.await()
+            SfuConnectionResult.Success
+        }
+        val coordinator = coordinator()
+
+        val joins = (1..5).map {
+            async { coordinator.join() }
+        }
+        advanceUntilIdle()
+        connectGate.complete(Unit)
+        val results = joins.awaitAll()
+        advanceUntilIdle()
+
+        results.forEach { assertThat(it).isInstanceOf(Success::class.java) }
+        assertThat(results.map { (it as Success).value }.distinct()).hasSize(1)
+        // One join request and one SFU connect for five callers.
+        coVerify(exactly = 1) {
+            apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any())
+        }
+        coVerify(exactly = 1) { mockSession.connectInternal() }
+        verify(exactly = 1) { sessionManager.setActiveSession(mockSession) }
+    }
+
+    @Test
+    fun `concurrent joins run the join setup exactly once`() = runTest(testDispatcher) {
+        stubJoinCall(Success(mockJoinResponse))
+        val connectGate = CompletableDeferred<Unit>()
+        coEvery { mockSession.connectInternal() } coAnswers {
+            connectGate.await()
+            SfuConnectionResult.Success
+        }
+        val coordinator = coordinator()
+        val interceptor = mockk<CallJoinInterceptor>(relaxed = true)
+
+        // The interceptor-carrying caller goes first, then a bare join() like the auto-join in
+        // CallState — which used to overwrite the interceptor with null.
+        val first = async { coordinator.join(callJoinInterceptor = interceptor) }
+        advanceUntilIdle()
+        val second = async { coordinator.join() }
+        advanceUntilIdle()
+        connectGate.complete(Unit)
+        val results = listOf(first, second).awaitAll()
+        advanceUntilIdle()
+
+        results.forEach { assertThat(it).isInstanceOf(Success::class.java) }
+        verify(exactly = 1) { callAnalytics.joinAnalytics.onJoinFunctionStart() }
+        verify(exactly = 1) { callAnalytics.mediaPermissionObserver.mediaPermissionStatus() }
+        verify(exactly = 1) { lifecycle.resetLeaveGuard() }
+        verify(exactly = 1) { state.callJoinInterceptor = interceptor }
+        verify(exactly = 0) { state.callJoinInterceptor = null }
+    }
+
+    @Test
+    fun `a join after the previous one finished starts a fresh attempt`() = runTest(
+        testDispatcher,
+    ) {
+        stubJoinCall(Success(mockJoinResponse))
+        coEvery { mockSession.connectInternal() } returns SfuConnectionResult.Success
+        val coordinator = coordinator()
+
+        coordinator.join()
+        advanceUntilIdle()
+        // The completed in-flight join must not be reused, otherwise a later join() would
+        // replay a stale result instead of starting again.
+        sessionFlow.value = null
+        coordinator.join()
+        advanceUntilIdle()
+
+        coVerify(exactly = 2) {
+            apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any())
+        }
+    }
+
+    @Test
+    fun `a session that cannot connect is cleaned up rather than left running`() = runTest(
+        testDispatcher,
+    ) {
+        stubJoinCall(Success(mockJoinResponse))
+        coEvery { mockSession.connectInternal() } returns SfuConnectionResult.Failure(
+            Exception("permanent auth error"),
+            cause = SfuConnectFailureCause.TerminalSocketFailure,
+        )
+
+        val result = coordinator().joinInternal(
+            joinAnalyticsModel = JoinAnalyticsModel(0, JoinReason.FirstAttempt),
+        )
+        advanceUntilIdle()
+
+        assertThat(result).isInstanceOf(Failure::class.java)
+        verify { mockSession.cleanup() }
+        assertThat(sessionFlow.value).isNull()
+    }
+
+    @Test
+    fun `failed recovery tears down the join session and any reconnect replacement`() = runTest(
+        testDispatcher,
+    ) {
+        stubJoinCall(Success(mockJoinResponse))
+        val replacement = mockk<RtcSession>(relaxed = true)
+        coEvery { mockSession.connectInternal() } coAnswers {
+            // Reconnect swapped the active session before recovery settled as failed.
+            sessionFlow.value = replacement
+            connectionFlow.value = RealtimeConnection.ReconnectingFailed
+            SfuConnectionResult.Failure(
+                Exception("recoverable socket failure"),
+                cause = SfuConnectFailureCause.RecoverableSocketFailure,
+            )
+        }
+
+        val result = coordinator().joinInternal(
+            joinAnalyticsModel = JoinAnalyticsModel(0, JoinReason.FirstAttempt),
+        )
+        advanceUntilIdle()
+
+        assertThat(result).isInstanceOf(Failure::class.java)
+        verify { mockSession.cleanup() }
+        verify { replacement.cleanup() }
+        assertThat(sessionFlow.value).isNull()
+    }
+
     @Test
     fun `joinAndRing joins then rings the members`() = runTest(testDispatcher) {
         stubJoinCall(Success(mockJoinResponse))