diff --git a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt index 6e66f38ad68..6519e6bc743 100644 --- a/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt @@ -527,7 +527,8 @@ public class RtcSession internal constructor( * Creates and publishes an audio track for transmitting audio. * This is used both when microphone is enabled and when screen sharing starts with muted microphone. */ - private suspend fun createAndPublishAudioTrack() { + @VisibleForTesting + internal suspend fun createAndPublishAudioTrack() { val canUserSendAudio = call.state.ownCapabilities.value.contains( OwnCapability.SendAudio, ) @@ -537,20 +538,33 @@ public class RtcSession internal constructor( setMuteState(isEnabled = true, TrackType.TRACK_TYPE_AUDIO) val streamId = buildTrackId(TrackType.TRACK_TYPE_AUDIO) - val track = publisher.value?.publishStream( + val audio = publisher.value?.publishStream( streamId, TrackType.TRACK_TYPE_AUDIO, - ) + ).asPublishedOrNull(TrackType.TRACK_TYPE_AUDIO) ?: return setLocalTrack( TrackType.TRACK_TYPE_AUDIO, AudioTrack( streamId = streamId, - audio = track as org.webrtc.AudioTrack, + audio = audio, ), ) } + private inline fun MediaStreamTrack?.asPublishedOrNull( + trackType: TrackType, + ): T? { + val typed = this as? T + if (typed == null) { + logger.w { + "[trackPublishing] Skipping $trackType: no track from publisher " + + "(publisher missing, no publish options, or publish failed)" + } + } + return typed + } + /** * Connection and WebRTC. */ @@ -1123,17 +1137,18 @@ public class RtcSession internal constructor( setMuteState(isEnabled = true, TrackType.TRACK_TYPE_VIDEO) val streamId = buildTrackId(TrackType.TRACK_TYPE_VIDEO) - val track = publisher.value?.publishStream( + val video = publisher.value?.publishStream( streamId, TrackType.TRACK_TYPE_VIDEO, call.mediaManager.camera.resolution.value, - ) + ).asPublishedOrNull(TrackType.TRACK_TYPE_VIDEO) + ?: return@collectLatest setLocalTrack( TrackType.TRACK_TYPE_VIDEO, VideoTrack( streamId = streamId, - video = track as org.webrtc.VideoTrack, + video = video, ), ) } else { @@ -1169,16 +1184,18 @@ public class RtcSession internal constructor( if (canUserShareScreen) { setMuteState(true, TrackType.TRACK_TYPE_SCREEN_SHARE) val streamId = buildTrackId(TrackType.TRACK_TYPE_SCREEN_SHARE) - val track = publisher.value?.publishStream( + val video = publisher.value?.publishStream( streamId, TrackType.TRACK_TYPE_SCREEN_SHARE, - ) + ).asPublishedOrNull( + TrackType.TRACK_TYPE_SCREEN_SHARE, + ) ?: return@collectLatest setLocalTrack( TrackType.TRACK_TYPE_SCREEN_SHARE, VideoTrack( streamId = streamId, - video = track as org.webrtc.VideoTrack, + video = video, ), ) } 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..36de46f7bbe 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 @@ -16,6 +16,7 @@ package io.getstream.video.android.core.call.components +import io.getstream.android.video.generated.models.JoinCallResponse import io.getstream.android.video.generated.models.RingCallRequest import io.getstream.log.taggedLogger import io.getstream.result.Error @@ -38,10 +39,14 @@ import io.getstream.video.android.core.call.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.CancellationException import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.delay import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext import stream.video.sfu.models.WebsocketReconnectStrategy /** @@ -65,10 +70,42 @@ internal class CallJoinCoordinator( private val callRegistry: ClientCallRegistry, private val hasRequiredPermissions: () -> Boolean, ) { + private companion object { + const val JOIN_FLIGHT_KEY = "join" + } + private val logger by taggedLogger("Call:JoinCoordinator:$type:$id") + /** + * Coalesces concurrent [join] calls into one attempt on the call [scope]. + * + * Without this, overlapping joins each build an [RtcSession] while reusing + * [CallSessionManager.sessionId], which leaves SFU-evicted zombies that fail every RPC + * with PARTICIPANT_NOT_FOUND. Checking [CallSessionManager.session] is not enough — it + * is only set after the coordinator round-trip. + * + * Coalescing the whole [join] also keeps once-per-join work once-only: JoinInitiated / + * MediaDevicePermission analytics, installing [CallState.callJoinInterceptor], resetting + * the leave guard, and moving to [RealtimeConnection.InProgress]. + * + * [StreamRefCountedSingleFlightProcessor] keeps the join alive when one waiter (e.g. an + * Activity) is cancelled while others still await, and cancels it when the last waiter + * leaves. + */ + private val joinFlight = StreamRefCountedSingleFlightProcessor(scope) + private fun isVideoEnabled(): Boolean = state.settings.value?.video?.enabled ?: false + /** + * Joins the call, coalescing concurrent callers into one in-flight execution (single-flight). + * + * The shared work runs on the call [scope]. Each caller still awaits on its own coroutine, + * so destroying one UI scope only drops that waiter; remaining waiters keep the join. + * The shared job is **not** cancelled when the last waiter leaves — incoming accept can + * finish/recreate the Activity after the SFU session is already in, and aborting then + * leaves ringing Idle (Connecting…) forever. Leave / call cleanup still cancel [scope] + * and abort the join. + */ suspend fun join( create: Boolean = false, createOptions: CreateCallOptions? = null, @@ -77,6 +114,70 @@ internal class CallJoinCoordinator( hintHighScaleLivestreamPublisher: Boolean? = null, callJoinInterceptor: CallJoinInterceptor? = null, ): Result { + var coalesced = false + return joinFlight.run( + JOIN_FLIGHT_KEY, + onCoalesced = { + coalesced = true + logger.w { + "[join] Concurrent join coalesced into in-flight join " + + "(interceptorIgnored=${callJoinInterceptor != null && + callJoinInterceptor !== state.callJoinInterceptor})" + } + if (callJoinInterceptor != null && + callJoinInterceptor !== state.callJoinInterceptor + ) { + logger.w { + "[join] Coalesced caller interceptor dropped; in-flight interceptor kept" + } + } + }, + cancelIfLastWaiter = false, + ) { + executeJoin( + create, + createOptions, + ring, + notify, + hintHighScaleLivestreamPublisher, + callJoinInterceptor, + ) + }.also { + if (coalesced) { + sessionManager.session.value?.sfuTracer?.trace( + "join-coalesced", + "concurrent join awaited in-flight join", + ) + } + }.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 { + // Subsequent join() calls while a session is live return that session instead of + // building a second one. [joinInternal] repeats the same check for direct callers. + sessionManager.session.value?.let { existing -> + logger.w { "[join] Call already joined — returning existing session" } + existing.sfuTracer.trace("join-already-joined", "join() while session already live") + return Success(existing) + } + callAnalytics.joinAnalytics.onJoinFunctionStart() callAnalytics.mediaPermissionObserver.mediaPermissionStatus() logger.d { @@ -202,6 +303,12 @@ internal class CallJoinCoordinator( return true } + /** + * Performs one join attempt: coordinator round-trip, [RtcSession] creation and SFU connect. + * + * Direct callers (tests, retry loop) must not build a second session while one is live. + * The already-joined check here enforces that; [executeJoin] also gates before setup. + */ suspend fun joinInternal( create: Boolean = false, createOptions: CreateCallOptions? = null, @@ -210,12 +317,21 @@ internal class CallJoinCoordinator( hintHighScaleLivestreamPublisher: Boolean? = null, joinAnalyticsModel: JoinAnalyticsModel, ): Result { + // Gate before any teardown: cancelSfuObservers() would leave the live session without + // its SFU event subscription, and only monitorSession() (further down, on the new-session + // path) restores it. + sessionManager.session.value?.let { existing -> + logger.i { "[joinInternal] Call already joined — returning existing session" } + existing.sfuTracer.trace( + "join-already-joined", + "joinInternal() while session already live", + ) + return Success(existing) + } + sessionManager.nonFastReconnectAttempts = 0 sessionMonitor.cancelSfuObservers() - if (sessionManager.session.value != null) { - return Failure(Error.GenericError("Call $type:$id has already been joined")) - } logger.d { "[joinInternal] #track; create: $create, ring: $ring, notify: $notify, createOptions: $createOptions" } @@ -266,6 +382,29 @@ internal class CallJoinCoordinator( state._connection.value = RealtimeConnection.Joined(localSession) + // Last-/sole-waiter cancel aborts this call-scoped job. If that happens after the + // session is installed, clear it — otherwise the idempotent join() path returns + // Success(zombie) and we keep a half-joined participant (PARTICIPANT_NOT_FOUND). + try { + return completeJoinAfterSessionInstall(localSession, result.value) + } catch (ce: CancellationException) { + withContext(NonCancellable) { + logger.w { + "[joinInternal] Join cancelled after session install — discarding session" + } + discardFailedSession(localSession) + if (state._connection.value is RealtimeConnection.Joined) { + state._connection.value = RealtimeConnection.Disconnected + } + } + throw ce + } + } + + private suspend fun completeJoinAfterSessionInstall( + localSession: RtcSession, + joinResponse: JoinCallResponse, + ): Result { // This is the SFU ws connection val sfuConnectionResult = localSession.connectInternal() @@ -300,6 +439,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 +452,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", @@ -330,11 +471,33 @@ internal class CallJoinCoordinator( // (re)establish monitoring when the session is unchanged, using the response that // still matches it, so we neither double-register nor monitor with a stale response. if (connectedSession === localSession) { - sessionMonitor.monitorSession(result.value) + sessionMonitor.monitorSession(joinResponse) } return Success(value = connectedSession) } + /** + * Tears down every session left after a failed join connect. Clearing the reference + * alone is not enough: sockets and peer connections stay alive and keep issuing SFU + * RPCs for a participant that is gone, which the SFU answers with PARTICIPANT_NOT_FOUND. + * + * Recoverable failures may already have swapped in a replacement via [CallReconnector] + * before [didReconnectSucceed] settles as failed. That replacement is not useful once + * join is returning Failure — tear it down too so nothing live is left behind. + */ + private fun discardFailedSession(localSession: RtcSession) { + val active = sessionManager.session.value + logger.d { + "[joinInternal] Discarding session(s) after failed join connect " + + "(activeIsJoinSession=${active === localSession})" + } + sessionManager.setActiveSession(null) + if (active != null && active !== localSession) { + active.cleanup() + } + localSession.cleanup() + } + /** * Reports the SFU WebSocket join failure to analytics. Only called from the join * flow ([joinInternal]) so that reconnect-driven [RtcSession.connectInternal] failures 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..290ac445a4d --- /dev/null +++ b/stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessor.kt @@ -0,0 +1,237 @@ +/* + * 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.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.ConcurrentHashMap +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 cancels the shared job only when [run] is called with + * [cancelIfLastWaiter] (the default). Pass `false` when the work must outlive the last + * UI waiter (Activity finish / screen handoff) and be torn down only by [scope] cancel + * (leave / call cleanup). + * - [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). + * + * Candidate for Stream Android Core v2 alongside [StreamSingleFlightProcessorImpl]. + * + * High-level [run] algorithm: + * ``` + * run + * ├── acquireWaiter + * │ └── selectFlightLocked + * │ ├── createFlightLocked + * │ └── removeFlightIfCurrentLocked + * └── awaitSharedResult + * └── releaseWaiter + * ``` + * + * All map mutations and the [closed] flag go through [mutex]. Callers only reuse a flight + * while its deferred is still [Deferred.isActive] — a `Cancelling` job is not joinable. + */ +internal class StreamRefCountedSingleFlightProcessor( + private val scope: CoroutineScope, +) { + private class Flight( + val key: String, + val deferred: Deferred>, + var waiters: Int, + ) + + private class Acquired( + val flight: Flight, + val coalesced: Boolean, + ) + + private val mutex = Mutex() + private val flights = ConcurrentHashMap>() + 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 when [cancelIfLastWaiter] is true) is cancelled. + * + * [onCoalesced] runs on this waiter when it attaches to an already-running flight + * (before awaiting the shared result). + * + * When [cancelIfLastWaiter] is false, the last cancelled waiter leaves the shared job + * running and keeps the map entry so a later [run] can coalesce instead of starting a + * second execution. + */ + suspend fun run( + key: String, + onCoalesced: () -> Unit = {}, + cancelIfLastWaiter: Boolean = true, + block: suspend () -> T, + ): Result { + val acquired = acquireWaiter(key, block) + ?: return Result.failure( + ClosedSendChannelException("RefCountedSingleFlight is closed"), + ) + if (acquired.coalesced) onCoalesced() + return awaitSharedResult(acquired.flight, cancelIfLastWaiter) + } + + private suspend fun acquireWaiter( + key: String, + block: suspend () -> T, + ): Acquired? = mutex.withLock { + if (closed.get()) return@withLock null + selectFlightLocked(key, block) + } + + @Suppress("UNCHECKED_CAST") + private fun selectFlightLocked( + key: String, + block: suspend () -> T, + ): Acquired { + val running = flights[key]?.takeIf { it.deferred.isActive } as Flight? + if (running != null) { + running.waiters++ + return Acquired(running, coalesced = true) + } + return Acquired(createFlightLocked(key, block), coalesced = false) + } + + private fun createFlightLocked( + key: String, + block: suspend () -> T, + ): Flight { + lateinit var deferred: Deferred> + 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 { + removeFlightIfCurrentLocked(key, deferred) + } + } + } + return Flight(key = key, deferred = deferred, waiters = 1).also { flights[key] = it } + } + + private fun removeFlightIfCurrentLocked(key: String, deferred: Deferred<*>) { + if (flights[key]?.deferred === deferred) { + flights.remove(key) + } + } + + private fun cancelAndDetachLocked(flight: Flight<*>) { + removeFlightIfCurrentLocked(flight.key, flight.deferred) + flight.deferred.cancel() + } + + private suspend fun awaitSharedResult( + flight: Flight, + cancelIfLastWaiter: Boolean, + ): Result { + var released = false + suspend fun releaseWaiter(cancelIfLast: Boolean) { + if (released) return + released = true + // Decrement, map removal, and cancel must stay under one lock so a new run() cannot + // attach to a flight that is about to be cancelled (waiters already at 0). + mutex.withLock { + flight.waiters = (flight.waiters - 1).coerceAtLeast(0) + if (flight.waiters != 0) return@withLock + when { + cancelIfLast && flight.deferred.isActive -> cancelAndDetachLocked(flight) + // Job already dead (completed, failed, or scope cancelled): drop the map + // entry. A deferred created on an already-cancelled scope can die before + // its body/finally runs, which would otherwise leave `has(key) == true`. + !flight.deferred.isActive -> + removeFlightIfCurrentLocked(flight.key, flight.deferred) + // Last waiter left and the job is still running (cancelIfLastWaiter = + // false): keep the entry so a later run() coalesces instead of double-joining. + } + } + } + + return try { + flight.deferred.await() + } catch (ce: CancellationException) { + // NonCancellable: waiter bookkeeping must run while this coroutine is cancelling. + withContext(NonCancellable) { + releaseWaiter(cancelIfLast = cancelIfLastWaiter) + } + throw ce + } finally { + withContext(NonCancellable) { + releaseWaiter(cancelIfLast = false) + } + } + } + + fun has(key: String): Boolean = flights.containsKey(key) + + suspend fun cancel(key: String): Result = runCatching { + mutex.withLock { + val flight = flights[key] ?: return@withLock + cancelAndDetachLocked(flight) + } + } + + suspend fun clear(cancelRunning: Boolean): Result = runCatching { + mutex.withLock { + if (cancelRunning) { + val snapshot = flights.values.toList() + flights.clear() + snapshot.forEach { it.deferred.cancel() } + } else { + flights.clear() + } + } + } + + suspend fun stop(): Result = runCatching { + mutex.withLock { + if (closed.compareAndSet(false, true)) { + val snapshot = flights.values.toList() + flights.clear() + snapshot.forEach { it.deferred.cancel() } + } + } + } +} 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..13b9e1aa5e0 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,51 @@ 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()) + } + // The single outer gate must also stop a second RtcSession from being installed. + verify(exactly = 0) { sessionManager.setActiveSession(mockSession) } + verify { existing.sfuTracer.trace("join-already-joined", 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), ) + advanceUntilIdle() - assertThat(result).isInstanceOf(Failure::class.java) + assertThat(result).isInstanceOf(Success::class.java) + assertThat((result as Success).value).isSameInstanceAs(existing) + coVerify(exactly = 0) { + apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any()) + } + verify { existing.sfuTracer.trace("join-already-joined", any()) } + // The guard must not tear down the live session's SFU observers: nothing on this path + // re-registers them, so cancelling here would silently stop event monitoring. + verify(exactly = 0) { sessionMonitor.cancelSfuObservers() } } @Test @@ -232,6 +275,274 @@ 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) } + verify(exactly = 4) { mockSession.sfuTracer.trace("join-coalesced", any()) } + } + + @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 does not abort the shared 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 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() + + // Call-scoped join keeps running after the UI waiters drop (incoming accept can + // finish/recreate the Activity). Leave still aborts it by cancelling the call scope. + assertThat(sessionFlow.value).isSameInstanceAs(mockSession) + verify { sessionManager.setActiveSession(mockSession) } + + val retry = coordinator.join() + advanceUntilIdle() + assertThat(retry).isInstanceOf(Success::class.java) + assertThat((retry as Success).value).isSameInstanceAs(mockSession) + coVerify(exactly = 1) { + apiClient.joinRequest(any(), any(), any(), any(), any(), any(), any(), any()) + } + } + + @Test + fun `cancelling the sole waiter does not abort 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).isSameInstanceAs(mockSession) + verify { sessionManager.setActiveSession(mockSession) } + } + + @Test + fun `cancelling after setActiveSession lets the in-flight join finish`() = runTest( + testDispatcher, + ) { + stubJoinCall(Success(mockJoinResponse)) + val connectGate = CompletableDeferred() + coEvery { mockSession.connectInternal() } coAnswers { + connectGate.await() + SfuConnectionResult.Success + } + val coordinator = coordinator() + + val join = async { coordinator.join() } + advanceUntilIdle() + assertThat(sessionFlow.value).isSameInstanceAs(mockSession) + assertThat(connectionFlow.value).isInstanceOf(RealtimeConnection.Joined::class.java) + + join.cancel() + advanceUntilIdle() + assertFailsWith { join.await() } + + connectGate.complete(Unit) + advanceUntilIdle() + + assertThat(sessionFlow.value).isSameInstanceAs(mockSession) + assertThat(connectionFlow.value).isInstanceOf(RealtimeConnection.Joined::class.java) + verify(exactly = 0) { mockSession.cleanup() } + + val retry = coordinator.join() + advanceUntilIdle() + assertThat(retry).isInstanceOf(Success::class.java) + assertThat((retry as Success).value).isSameInstanceAs(mockSession) + coVerify(exactly = 1) { + 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)) diff --git a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt index ad2ecb8253d..900bf83ee1e 100644 --- a/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt @@ -775,6 +775,33 @@ class RtcSessionTest2 { verify(exactly = 0) { rtcSession["createPublisher"](any>()) } } + @Test + fun `createAndPublishAudioTrack does not crash when publishStream returns null`() = runTest( + testDispatcher, + ) { + ownCapabilitiesFlow.value = listOf(OwnCapability.SendAudio) + val (rtcSession, publisherMock) = createRtcSessionSpyWithMockSocket() + rtcSession.publisher.value = publisherMock + coEvery { + publisherMock.publishStream(any(), TrackType.TRACK_TYPE_AUDIO) + } returns null + + rtcSession.createAndPublishAudioTrack() + + coVerify { publisherMock.publishStream(any(), TrackType.TRACK_TYPE_AUDIO) } + } + + @Test + fun `createAndPublishAudioTrack does not crash when publisher is missing`() = runTest( + testDispatcher, + ) { + ownCapabilitiesFlow.value = listOf(OwnCapability.SendAudio) + val (rtcSession, _) = createRtcSessionSpyWithMockSocket() + rtcSession.publisher.value = null + + rtcSession.createAndPublishAudioTrack() + } + private fun RtcSession.fieldValue(name: String): T? { val field = RtcSession::class.java.getDeclaredField(name) field.isAccessible = true 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..7e829ff36a4 --- /dev/null +++ b/stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/utils/StreamRefCountedSingleFlightProcessorTest.kt @@ -0,0 +1,495 @@ +/* + * 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.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.ClosedSendChannelException +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.withContext +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 `onCoalesced runs only for waiters that attach to an in-flight job`() = runTest( + testDispatcher, + ) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val coalesced = AtomicInteger(0) + val gate = CompletableDeferred() + + val executions = AtomicInteger(0) + val jobs = (1..5).map { + async { + processor.run("key", onCoalesced = { coalesced.incrementAndGet() }) { + 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()) + assertEquals(4, coalesced.get()) + } + + @Test + fun `last waiter on a cancelled scope still removes the flight`() = runTest(testDispatcher) { + val cancelledScope = TestScope(testDispatcher) + cancelledScope.cancel() + val processor = StreamRefCountedSingleFlightProcessor(cancelledScope) + + val result = runCatching { + processor.run("key") { "should not complete" } + } + + assertTrue(result.exceptionOrNull() is CancellationException) + assertFalse(processor.has("key")) + } + + @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 `last waiter cancel with cancelIfLastWaiter false keeps the job and coalesces the next run`() = + runTest(testDispatcher) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val executions = AtomicInteger(0) + val gate = CompletableDeferred() + + val first = async { + processor.run("key", cancelIfLastWaiter = false) { + executions.incrementAndGet() + gate.await() + "ok" + } + } + advanceUntilIdle() + first.cancel() + advanceUntilIdle() + assertFailsWith { first.await() } + assertTrue(processor.has("key")) + assertEquals(1, executions.get()) + + val second = async { + processor.run("key", cancelIfLastWaiter = false) { + executions.incrementAndGet() + gate.await() + "should not run" + } + } + advanceUntilIdle() + gate.complete(Unit) + assertEquals("ok", second.await().getOrThrow()) + advanceUntilIdle() + assertEquals(1, executions.get()) + assertFalse(processor.has("key")) + } + + @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) + } + + @Test + fun `after last waiter cancel a new run starts a fresh execution`() = runTest( + testDispatcher, + ) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val gate = CompletableDeferred() + val executions = AtomicInteger(0) + + val first = async { + processor.run("key") { + executions.incrementAndGet() + gate.await() + "first" + } + } + advanceUntilIdle() + first.cancel() + advanceUntilIdle() + assertFailsWith { first.await() } + assertFalse(processor.has("key")) + + assertEquals( + "second", + processor.run("key") { + executions.incrementAndGet() + "second" + }.getOrThrow(), + ) + advanceUntilIdle() + assertEquals(2, executions.get()) + } + + /** + * Protects the race where last-waiter cancel unlocked before cancelling: a new run could + * attach to the dying flight and then get cancelled with it. Removal + cancel stay under + * one lock so that cannot happen. A new run may still share if it arrives before release + * (first-wins) — that must succeed, not throw CancellationException. + */ + @Test + fun `last-waiter cancel race does not cancel a newly started run`() = runBlocking { + repeat(100) { iteration -> + val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + val processor = StreamRefCountedSingleFlightProcessor(scope) + val blockEntered = CompletableDeferred() + val holdBlock = CompletableDeferred() + + val first = scope.async { + processor.run("key") { + blockEntered.complete(Unit) + holdBlock.await() + "first-$iteration" + } + } + blockEntered.await() + + first.cancel() + val second = scope.async { + processor.run("key") { + "second-$iteration" + } + } + holdBlock.complete(Unit) + + assertFailsWith { first.await() } + val secondResult = second.await() + assertTrue( + "iteration $iteration failed: ${secondResult.exceptionOrNull()}", + secondResult.isSuccess, + ) + scope.cancel() + } + } + + @Test + fun `cancel key during NonCancellable cleanup starts a fresh flight`() = runTest( + testDispatcher, + ) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val blockEntered = CompletableDeferred() + val cleanupStarted = CompletableDeferred() + val holdCleanup = CompletableDeferred() + val hang = CompletableDeferred() + val executions = AtomicInteger(0) + + val first = async { + processor.run("key") { + executions.incrementAndGet() + try { + blockEntered.complete(Unit) + hang.await() + "first" + } finally { + withContext(NonCancellable) { + cleanupStarted.complete(Unit) + holdCleanup.await() + } + } + } + } + advanceUntilIdle() + blockEntered.await() + + processor.cancel("key") + advanceUntilIdle() + cleanupStarted.await() + assertFalse(processor.has("key")) + + val second = processor.run("key") { + executions.incrementAndGet() + "second" + } + advanceUntilIdle() + + assertEquals("second", second.getOrThrow()) + assertEquals(2, executions.get()) + + holdCleanup.complete(Unit) + advanceUntilIdle() + assertFailsWith { first.await() } + } + + @Test + fun `stop rejects new runs while an old flight is still cleaning up`() = runTest( + testDispatcher, + ) { + val processor = StreamRefCountedSingleFlightProcessor(testScope) + val blockEntered = CompletableDeferred() + val hang = CompletableDeferred() + + val first = async { + processor.run("key") { + blockEntered.complete(Unit) + hang.await() + "first" + } + } + advanceUntilIdle() + blockEntered.await() + + assertTrue(processor.stop().isSuccess) + val second = processor.run("key") { error("should not run") } + assertTrue(second.isFailure) + assertTrue(second.exceptionOrNull() is ClosedSendChannelException) + + hang.complete(Unit) + advanceUntilIdle() + assertFailsWith { first.await() } + + val third = processor.run("key") { "after-stop" } + assertTrue(third.exceptionOrNull() is ClosedSendChannelException) + } +}