diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt index 4ea7451beb..f15e297a1d 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinator.kt +++ b/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.RtcSession 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,8 +71,33 @@ internal class CallJoinCoordinator( ) { 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>? = 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, @@ -76,6 +105,51 @@ internal class CallJoinCoordinator( notify: Boolean = false, hintHighScaleLivestreamPublisher: Boolean? = null, callJoinInterceptor: CallJoinInterceptor? = null, + ): Result { + 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>().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 { callAnalytics.joinAnalytics.onJoinFunctionStart() callAnalytics.mediaPermissionObserver.mediaPermissionStatus() @@ -300,6 +374,7 @@ internal class CallJoinCoordinator( "[_join] Got terminal error while connecting to SFU. Error : $sfuConnectionResult" } sendJoinErrorAnalytics(sfuConnectionResult) + discardFailedSession(localSession) return Failure( Error.GenericError( sfuConnectionResult.error.message ?: "RtcSession error occurred.", @@ -312,6 +387,7 @@ internal class CallJoinCoordinator( 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 @@ internal class CallJoinCoordinator( 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 diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt index b1fec5161a..fd513f70cd 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/components/CallJoinCoordinatorTest.kt +++ b/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() + 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() + coEvery { mockSession.connectInternal() } coAnswers { + connectGate.await() + SfuConnectionResult.Success + } + val coordinator = coordinator() + val interceptor = mockk(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(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))