From 6013e689dbe9823e9f5eb6ed0d30e9a8beae36c8 Mon Sep 17 00:00:00 2001 From: pratimmallick Date: Mon, 10 Aug 2026 16:47:55 +0530 Subject: [PATCH 1/3] fix(core): single-flight Call.join to stop concurrent-join race Coalesce overlapping join() callers onto one in-flight attempt and clean up sessions that fail to connect, preventing SFU-evicted zombie publishers. Co-authored-by: Cursor --- .../call/components/CallJoinCoordinator.kt | 91 +++++++++++++++ .../components/CallJoinCoordinatorTest.kt | 106 ++++++++++++++++++ 2 files changed, 197 insertions(+) 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..a2c9e8340b 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,21 @@ internal class CallJoinCoordinator( return Success(value = connectedSession) } + /** + * Tears down a session this join created but could not connect. Clearing the reference + * alone is not enough: the socket and peer connections stay alive and keep issuing SFU + * RPCs for a participant that is gone, which the SFU answers with PARTICIPANT_NOT_FOUND. + * + * Skipped when recovery has already swapped a different session in — that one belongs to + * the reconnect flow, which owns the disposal of the session it replaced. + */ + private fun discardFailedSession(localSession: RtcSession) { + if (sessionManager.session.value !== localSession) return + logger.d { "[joinInternal] Discarding the session this join could not connect" } + sessionManager.setActiveSession(null) + 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..dac37b1c12 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,108 @@ 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 `joinAndRing joins then rings the members`() = runTest(testDispatcher) { stubJoinCall(Success(mockJoinResponse)) From 3ba89a49226c277cebcc9974360b8c0158203c10 Mon Sep 17 00:00:00 2001 From: pratimmallick Date: Mon, 10 Aug 2026 18:35:25 +0530 Subject: [PATCH 2/3] fix(core): always tear down sessions after failed join connect Remove the discardFailedSession ownership guard. Once join is returning Failure (including after failed join-time recovery), clear the active slot and cleanup both the join session and any reconnect replacement. Co-authored-by: Cursor --- .../call/components/CallJoinCoordinator.kt | 19 ++++++++----- .../components/CallJoinCoordinatorTest.kt | 27 +++++++++++++++++++ 2 files changed, 40 insertions(+), 6 deletions(-) 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 a2c9e8340b..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 @@ -412,17 +412,24 @@ internal class CallJoinCoordinator( } /** - * Tears down a session this join created but could not connect. Clearing the reference - * alone is not enough: the socket and peer connections stay alive and keep issuing SFU + * 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. * - * Skipped when recovery has already swapped a different session in — that one belongs to - * the reconnect flow, which owns the disposal of the session it replaced. + * 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) { - if (sessionManager.session.value !== localSession) return - logger.d { "[joinInternal] Discarding the session this join could not connect" } + 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() } 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 dac37b1c12..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 @@ -338,6 +338,33 @@ class CallJoinCoordinatorTest { 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)) From e3a5c0888840bab74330d4abe4bfa144a5d0367e Mon Sep 17 00:00:00 2001 From: pratimmallick Date: Mon, 17 Aug 2026 14:03:04 +0530 Subject: [PATCH 3/3] fix(core): call-scoped refcounted single-flight for Call.join Move join coalescing to StreamRefCountedSingleFlightProcessor so work runs on the call scope, survives individual waiter cancellation, and cancels only when the last waiter leaves. Subsequent join() on an already-joined call returns the existing session instead of failing and tearing down the live call. Co-authored-by: Cursor --- .../call/components/CallJoinCoordinator.kt | 99 ++++--- .../StreamRefCountedSingleFlightProcessor.kt | 149 ++++++++++ .../utils/StreamSingleFlightProcessorImpl.kt | 3 + .../components/CallJoinCoordinatorTest.kt | 135 ++++++++- ...reamRefCountedSingleFlightProcessorTest.kt | 258 ++++++++++++++++++ 5 files changed, 588 insertions(+), 56 deletions(-) create mode 100644 stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessor.kt create mode 100644 stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessorTest.kt 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 f15e297a1d..53437131e0 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,14 +38,11 @@ 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 io.getstream.video.android.core.utils.StreamRefCountedSingleFlightProcessor 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 /** @@ -72,31 +69,31 @@ 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). + * Coalesces concurrent [join] calls into one attempt on the call [scope]. * - * 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. + * 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]. * - * Held only long enough to read or publish [joinFlight], never across the join itself. + * [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 joinMutex = Mutex() - - /** The in-flight [join], if any. Completed flights are never reused. */ - private var joinFlight: CompletableDeferred>? = null + 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). - * Additional callers await the same [Result]. The winner runs on its own caller coroutine. + * + * 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, @@ -106,41 +103,26 @@ internal class CallJoinCoordinator( 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, + 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, + ), ) - 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( @@ -151,6 +133,12 @@ internal class CallJoinCoordinator( hintHighScaleLivestreamPublisher: Boolean?, callJoinInterceptor: CallJoinInterceptor?, ): Result { + // Idempotent: subsequent join() while a session is already live returns that session + 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 { @@ -287,8 +275,9 @@ internal class CallJoinCoordinator( sessionManager.nonFastReconnectAttempts = 0 sessionMonitor.cancelSfuObservers() - if (sessionManager.session.value != null) { - return Failure(Error.GenericError("Call $type:$id has already been joined")) + sessionManager.session.value?.let { existing -> + logger.i { "[joinInternal] Call already joined — returning existing session" } + return Success(existing) } logger.d { "[joinInternal] #track; create: $create, ring: $ring, notify: $notify, createOptions: $createOptions" @@ -464,4 +453,8 @@ internal class CallJoinCoordinator( logger.d { "[_join] Reconnect after recoverable connection failure settled on $terminal" } return terminal is RealtimeConnection.Connected } + + private companion object { + const val JOIN_FLIGHT_KEY = "join" + } } diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessor.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessor.kt new file mode 100644 index 0000000000..9e63950d8f --- /dev/null +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessor.kt @@ -0,0 +1,149 @@ +/* + * 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.utils + +// package io.getstream.android.core.internal.processing + +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Deferred +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.async +import kotlinx.coroutines.channels.ClosedSendChannelException +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext +import java.util.concurrent.atomic.AtomicBoolean + +/** + * Single-flight that coalesces concurrent calls by key, runs the shared work on [scope], + * and tracks how many callers are still awaiting. + * + * Compared to [StreamSingleFlightProcessorImpl]: + * - Cancelling **one** waiter does **not** cancel the shared job (work is owned by [scope]). + * - Cancelling the **last** waiter **does** cancel the shared job (sole-caller cancel still + * aborts the operation). + * - [CancellationException] is rethrown to the cancelled waiter instead of being wrapped in + * [Result.failure]. + * + * Use this when the operation should survive Activity/ViewModel teardown of some waiters + * (e.g. screen handoff, UI join + call-scoped auto-join) but should not keep running after + * nobody is waiting — unless [scope] itself is cancelled (leave / call cleanup). + * + * Candidate for Stream Android Core v2 alongside [StreamSingleFlightProcessorImpl]. + */ +internal class StreamRefCountedSingleFlightProcessor( + private val scope: CoroutineScope, +) { + private class Flight( + val deferred: Deferred>, + var waiters: Int, + ) + + private val mutex = Mutex() + private val flights = mutableMapOf>() + private val closed = AtomicBoolean(false) + + /** + * Runs [block] once for [key] while concurrent callers await the same result. + * + * Returns [Result.failure] with a [ClosedSendChannelException] if [stop] has already + * been called. [CancellationException] is still rethrown when this waiter (or the shared + * job, including last-waiter cancel) is cancelled. + */ + @Suppress("UNCHECKED_CAST") + suspend fun run(key: String, block: suspend () -> T): Result { + if (closed.get()) { + return Result.failure(ClosedSendChannelException("RefCountedSingleFlight is closed")) + } + + val flight = mutex.withLock { + val running = flights[key]?.takeUnless { it.deferred.isCompleted } as Flight? + if (running != null) { + running.waiters++ + running + } else { + val deferred = scope.async { + try { + // Complete normally even when [block] fails so the scope does not see an + // uncaught child exception; waiters receive Result.failure after await. + try { + Result.success(block()) + } catch (ce: CancellationException) { + throw ce + } catch (t: Throwable) { + Result.failure(t) + } + } finally { + mutex.withLock { + if (flights[key]?.deferred === this@async) { + flights.remove(key) + } + } + } + } + Flight(deferred = deferred, waiters = 1).also { flights[key] = it } + } + } + + var released = false + suspend fun releaseWaiter(cancelIfLast: Boolean) { + if (released) return + released = true + val shouldCancelShared = mutex.withLock { + flight.waiters = (flight.waiters - 1).coerceAtLeast(0) + cancelIfLast && flight.waiters == 0 && flight.deferred.isActive + } + if (shouldCancelShared) { + flight.deferred.cancel() + } + } + + return try { + flight.deferred.await() + } catch (ce: CancellationException) { + // NonCancellable: waiter bookkeeping must run while this coroutine is cancelling. + withContext(NonCancellable) { + releaseWaiter(cancelIfLast = true) + } + throw ce + } finally { + withContext(NonCancellable) { + releaseWaiter(cancelIfLast = false) + } + } + } + + fun has(key: String): Boolean = flights.containsKey(key) + + fun cancel(key: String): Result = runCatching { + flights[key]?.deferred?.cancel() + } + + fun clear(cancelRunning: Boolean): Result = runCatching { + if (cancelRunning) { + flights.values.forEach { it.deferred.cancel() } + } + flights.clear() + } + + fun stop(): Result = runCatching { + if (closed.compareAndSet(false, true)) { + clear(cancelRunning = true).getOrThrow() + } + } +} diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamSingleFlightProcessorImpl.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamSingleFlightProcessorImpl.kt index 6918e50390..bcdd8b4e96 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamSingleFlightProcessorImpl.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamSingleFlightProcessorImpl.kt @@ -33,6 +33,9 @@ import java.util.concurrent.atomic.AtomicBoolean * * The shared work runs in [scope] (recommend a `CoroutineScope(SupervisorJob() + Dispatchers.IO)`), * so cancelling one awaiting caller does not cancel the shared execution. + * + * For the variant that cancels the shared job when the **last** waiter is cancelled, see + * [StreamRefCountedSingleFlightProcessor]. */ internal class StreamSingleFlightProcessorImpl( private val scope: CoroutineScope, 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 fd513f70cd..9bc2554acb 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 @@ -42,6 +42,7 @@ import io.mockk.every import io.mockk.mockk import io.mockk.unmockkAll import io.mockk.verify +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll @@ -54,6 +55,7 @@ import org.junit.After import org.junit.Before import org.junit.Rule import org.junit.Test +import kotlin.test.assertFailsWith /** * Tests the join orchestration in [CallJoinCoordinator]: the [join] retry loop, join-and-ring, @@ -202,14 +204,40 @@ class CallJoinCoordinatorTest { } @Test - fun `join fails when the call is already joined`() = runTest(testDispatcher) { - sessionFlow.value = mockk(relaxed = true) + fun `join returns the existing session when the call is already joined`() = runTest( + testDispatcher, + ) { + val existing = mockk(relaxed = true) + sessionFlow.value = existing + connectionFlow.value = RealtimeConnection.Connected + + val result = coordinator().join() + advanceUntilIdle() + + assertThat(result).isInstanceOf(Success::class.java) + assertThat((result as Success).value).isSameInstanceAs(existing) + assertThat(sessionFlow.value).isSameInstanceAs(existing) + assertThat(connectionFlow.value).isEqualTo(RealtimeConnection.Connected) + verify(exactly = 0) { sessionManager.setActiveSession(null) } + verify(exactly = 0) { callAnalytics.joinAnalytics.onJoinFunctionStart() } + coVerify(exactly = 0) { + apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any()) + } + } + + @Test + fun `joinInternal returns the existing session when the call is already joined`() = runTest( + testDispatcher, + ) { + val existing = mockk(relaxed = true) + sessionFlow.value = existing val result = coordinator().joinInternal( joinAnalyticsModel = JoinAnalyticsModel(0, JoinReason.FirstAttempt), ) - assertThat(result).isInstanceOf(Failure::class.java) + assertThat(result).isInstanceOf(Success::class.java) + assertThat((result as Success).value).isSameInstanceAs(existing) } @Test @@ -318,6 +346,107 @@ class CallJoinCoordinatorTest { } } + @Test + fun `cancelling one waiter leaves the shared join running for others`() = runTest( + testDispatcher, + ) { + stubJoinCall(Success(mockJoinResponse)) + val connectGate = CompletableDeferred() + coEvery { mockSession.connectInternal() } coAnswers { + connectGate.await() + SfuConnectionResult.Success + } + val coordinator = coordinator() + + // Two different caller jobs (e.g. Activity A and Activity B / CallState auto-join). + val first = async { coordinator.join() } + advanceUntilIdle() + val second = async { coordinator.join() } + advanceUntilIdle() + + first.cancel() + advanceUntilIdle() + assertFailsWith { first.await() } + + // Shared call-scoped join must still be alive for the second waiter. + connectGate.complete(Unit) + val secondResult = second.await() + advanceUntilIdle() + + assertThat(secondResult).isInstanceOf(Success::class.java) + coVerify(exactly = 1) { + apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any()) + } + coVerify(exactly = 1) { mockSession.connectInternal() } + } + + @Test + fun `cancelling the last waiter cancels the shared join`() = runTest(testDispatcher) { + // Gate before the session is installed so cancel cannot leave a half-joined session + // behind in this test. + val joinRequestGate = CompletableDeferred() + coEvery { + apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any()) + } coAnswers { + joinRequestGate.await() + Success(mockJoinResponse) + } + coEvery { mockSession.connectInternal() } returns SfuConnectionResult.Success + val coordinator = coordinator() + + val first = async { coordinator.join() } + advanceUntilIdle() + val second = async { coordinator.join() } + advanceUntilIdle() + + first.cancel() + second.cancel() + advanceUntilIdle() + assertFailsWith { first.await() } + assertFailsWith { second.await() } + + joinRequestGate.complete(Unit) + advanceUntilIdle() + + assertThat(sessionFlow.value).isNull() + coVerify(exactly = 0) { sessionManager.setActiveSession(mockSession) } + + // A later join must start a fresh attempt, not reuse the cancelled flight. + stubJoinCall(Success(mockJoinResponse)) + val retry = coordinator.join() + advanceUntilIdle() + assertThat(retry).isInstanceOf(Success::class.java) + // First (cancelled) flight entered joinRequest once; the retry is a second attempt. + coVerify(exactly = 2) { + apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any()) + } + } + + @Test + fun `cancelling the sole waiter cancels the call-scoped join`() = runTest(testDispatcher) { + val joinRequestGate = CompletableDeferred() + coEvery { + apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any()) + } coAnswers { + joinRequestGate.await() + Success(mockJoinResponse) + } + coEvery { mockSession.connectInternal() } returns SfuConnectionResult.Success + val coordinator = coordinator() + + val join = async { coordinator.join() } + advanceUntilIdle() + join.cancel() + advanceUntilIdle() + assertFailsWith { join.await() } + + joinRequestGate.complete(Unit) + advanceUntilIdle() + + assertThat(sessionFlow.value).isNull() + coVerify(exactly = 0) { sessionManager.setActiveSession(mockSession) } + } + @Test fun `a session that cannot connect is cleaned up rather than left running`() = runTest( testDispatcher, diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessorTest.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessorTest.kt new file mode 100644 index 0000000000..5bb028b1c9 --- /dev/null +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessorTest.kt @@ -0,0 +1,258 @@ +/* + * 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.utils + +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.channels.ClosedSendChannelException +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.assertFailsWith + +class StreamRefCountedSingleFlightProcessorTest { + + private val testDispatcher = StandardTestDispatcher() + private val testScope = TestScope(testDispatcher) + + @Test + fun `concurrent callers share one execution`() = runTest(testDispatcher) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val executions = AtomicInteger(0) + val gate = CompletableDeferred() + + val jobs = (1..5).map { + async { + processor.run("key") { + executions.incrementAndGet() + gate.await() + "ok" + } + } + } + advanceUntilIdle() + gate.complete(Unit) + val results = jobs.awaitAll() + advanceUntilIdle() + + assertEquals(listOf("ok", "ok", "ok", "ok", "ok"), results.map { it.getOrThrow() }) + assertEquals(1, executions.get()) + } + + @Test + fun `cancelling one waiter leaves the shared job running`() = runTest(testDispatcher) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val executions = AtomicInteger(0) + val gate = CompletableDeferred() + + val first = async { + processor.run("key") { + executions.incrementAndGet() + gate.await() + "ok" + } + } + advanceUntilIdle() + val second = async { + processor.run("key") { + executions.incrementAndGet() + gate.await() + "ok" + } + } + advanceUntilIdle() + + first.cancel() + advanceUntilIdle() + assertFailsWith { first.await() } + + gate.complete(Unit) + assertEquals("ok", second.await().getOrThrow()) + advanceUntilIdle() + assertEquals(1, executions.get()) + } + + @Test + fun `cancelling the last waiter cancels the shared job`() = runTest(testDispatcher) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val started = CompletableDeferred() + val gate = CompletableDeferred() + var completed = false + + val first = async { + processor.run("key") { + started.complete(Unit) + gate.await() + completed = true + "ok" + } + } + advanceUntilIdle() + started.await() + + val second = async { + processor.run("key") { + gate.await() + completed = true + "ok" + } + } + advanceUntilIdle() + + first.cancel() + second.cancel() + advanceUntilIdle() + assertFailsWith { first.await() } + assertFailsWith { second.await() } + + gate.complete(Unit) + advanceUntilIdle() + assertFalse(completed) + assertFalse(processor.has("key")) + } + + @Test + fun `cancelling the sole waiter cancels the shared job`() = runTest(testDispatcher) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val gate = CompletableDeferred() + var completed = false + + val job = async { + processor.run("key") { + gate.await() + completed = true + "ok" + } + } + advanceUntilIdle() + job.cancel() + advanceUntilIdle() + assertFailsWith { job.await() } + + gate.complete(Unit) + advanceUntilIdle() + assertFalse(completed) + } + + @Test + fun `a run after the previous one finished starts a fresh attempt`() = runTest( + testDispatcher, + ) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val executions = AtomicInteger(0) + + assertEquals( + "a", + processor.run("key") { + executions.incrementAndGet() + "a" + }.getOrThrow(), + ) + advanceUntilIdle() + assertEquals( + "b", + processor.run("key") { + executions.incrementAndGet() + "b" + }.getOrThrow(), + ) + advanceUntilIdle() + + assertEquals(2, executions.get()) + } + + @Test + fun `different keys do not coalesce`() = runTest(testDispatcher) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val gate = CompletableDeferred() + val executions = AtomicInteger(0) + + val a = async { + processor.run("a") { + executions.incrementAndGet() + gate.await() + "a" + } + } + val b = async { + processor.run("b") { + executions.incrementAndGet() + gate.await() + "b" + } + } + advanceUntilIdle() + gate.complete(Unit) + assertEquals( + listOf("a", "b"), + listOf(a.await().getOrThrow(), b.await().getOrThrow()), + ) + assertEquals(2, executions.get()) + } + + @Test + fun `block exceptions propagate to all waiters as Result failure`() = runTest( + testDispatcher, + ) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val gate = CompletableDeferred() + + val first = async { + processor.run("key") { + gate.await() + throw IllegalStateException("boom") + } + } + advanceUntilIdle() + val second = async { + processor.run("key") { + gate.await() + throw IllegalStateException("boom") + } + } + advanceUntilIdle() + gate.complete(Unit) + advanceUntilIdle() + + val firstResult = first.await() + val secondResult = second.await() + assertTrue(firstResult.isFailure) + assertTrue(secondResult.isFailure) + assertTrue(firstResult.exceptionOrNull() is IllegalStateException) + assertTrue(secondResult.exceptionOrNull() is IllegalStateException) + assertEquals("boom", firstResult.exceptionOrNull()?.message) + assertEquals("boom", secondResult.exceptionOrNull()?.message) + } + + @Test + fun `stop rejects new runs with Result failure`() = runTest(testDispatcher) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + assertTrue(processor.stop().isSuccess) + + val result = processor.run("key") { error("should not run") } + assertTrue(result.isFailure) + assertTrue(result.exceptionOrNull() is ClosedSendChannelException) + } +}