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 4ea7451beb3..53437131e06 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,6 +38,7 @@ 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 io.getstream.video.android.core.utils.StreamRefCountedSingleFlightProcessor import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.delay import kotlinx.coroutines.flow.first @@ -67,8 +68,33 @@ internal class CallJoinCoordinator( ) { 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 +103,42 @@ internal class CallJoinCoordinator( hintHighScaleLivestreamPublisher: Boolean? = null, callJoinInterceptor: CallJoinInterceptor? = null, ): Result { + 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 { + // 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 { @@ -213,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" @@ -300,6 +363,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 +376,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 +400,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 @@ -366,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 00000000000..9e63950d8f8 --- /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 6918e503901..bcdd8b4e960 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 b1fec5161ac..9bc2554acb3 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,10 @@ 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 import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.test.StandardTestDispatcher import kotlinx.coroutines.test.TestScope @@ -50,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, @@ -198,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 @@ -232,6 +264,236 @@ 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 `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, + ) { + 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)) 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 00000000000..5bb028b1c9d --- /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) + } +}