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 1a8d152818..350ed89a93 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 @@ -174,7 +174,6 @@ import stream.video.sfu.signal.UpdateSubscriptionsResponse import java.io.InterruptedIOException import java.net.SocketTimeoutException import java.util.Collections -import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger /** @@ -339,13 +338,6 @@ public class RtcSession internal constructor( */ private val muteSyncEnabled = AtomicBoolean(true) - /** - * Tracks whose SFU sync was skipped while [muteSyncEnabled] was false. [resumeMuteSync] - * flushes only these so we do not POST for unpublished tracks (the map is pre-seeded - * with audio/video/screen-share) or race [listenToMediaChanges] collectLatest replay. - */ - private val pendingMuteSyncTracks: MutableSet = - ConcurrentHashMap.newKeySet() private val oneBasedSessionCounter = sessionCounter + 1 /** @@ -1402,7 +1394,6 @@ public class RtcSession internal constructor( mediaScope.cancel() muteSyncEnabled.set(false) - pendingMuteSyncTracks.clear() muteStateSyncJobs.cancelAll() // cleanup all non-local tracks @@ -1451,11 +1442,10 @@ public class RtcSession internal constructor( * cannot lose another track's bit via a stale read–copy–write. * * During reconnect/migration, [cancelActiveWork] pauses SFU sync so collectors can keep - * recording the latest desired bits without posting to a stale connection. Tracks that - * would have posted are recorded in [pendingMuteSyncTracks]. Sync resumes from - * [connectRtc] once the active SFU is ready and flushes only those pending tracks — - * not the whole pre-seeded map — so unpublished tracks stay silent and - * [listenToMediaChanges] collectLatest replay does not cancel a redundant flush. + * recording the latest desired bits without posting to a stale connection. [resumeMuteSync] + * re-opens the gate from [connectRtc] once the active SFU is ready; it does not replay the + * skipped tracks itself, because [listenToMediaChanges] restarts in the same call and its + * collectLatest replay re-signals every track anyway. */ private fun setMuteState(isEnabled: Boolean, trackType: TrackType) { logger.d { "[setPublishState] #sfu; $trackType isEnabled: $isEnabled" } @@ -1466,15 +1456,10 @@ public class RtcSession internal constructor( private fun syncMuteStateToSfu(trackType: TrackType, isEnabled: Boolean) { if (!muteSyncEnabled.get()) { - pendingMuteSyncTracks.add(trackType) - // Resume may have flipped the flag between the check and the add. Fall through - // and post; same-track launch coalesces with any flush already in flight. - if (!muteSyncEnabled.get()) { - logger.d { - "[syncMuteStateToSfu] pending until SFU is ready; $trackType isEnabled: $isEnabled" - } - return + logger.d { + "[syncMuteStateToSfu] skipped, SFU not ready; $trackType isEnabled: $isEnabled" } + return } val currentSfu = sfuUrl @@ -1510,15 +1495,18 @@ public class RtcSession internal constructor( } } + /** + * Re-opens the SFU mute sync gate closed by [cancelActiveWork]. + * + * Deliberately does not replay the tracks that were skipped while paused. [connectRtc] + * restarts [listenToMediaChanges] immediately after this, and its collectLatest replay + * re-signals camera, microphone and screen-share from their current state. Flushing here + * as well produced a second RPC for the same track that the replay then cancelled + * (`IOException: Canceled`), which is what this gate exists to avoid. + */ private fun resumeMuteSync() { if (!muteSyncEnabled.compareAndSet(false, true)) return - val tracksToFlush = pendingMuteSyncTracks.toSet() - pendingMuteSyncTracks.clear() - logger.d { "[resumeMuteSync] flushing pending mute state for $tracksToFlush" } - tracksToFlush.forEach { trackType -> - val isEnabled = muteState.value[trackType] ?: return@forEach - syncMuteStateToSfu(trackType, isEnabled) - } + logger.d { "[resumeMuteSync] SFU ready, mute sync re-enabled" } } private fun isPermanentError(cause: Throwable): Boolean { @@ -2260,11 +2248,6 @@ public class RtcSession internal constructor( iceMonitoringJob?.cancel() iceMonitoringJob = null muteSyncEnabled.set(false) - // Nothing is deferred at the moment we pause, so anything still in the set is - // a leftover from a collector that raced the previous [resumeMuteSync] and - // posted anyway. Dropping it here keeps the next resume from flushing a track - // this pause never deferred. - pendingMuteSyncTracks.clear() muteStateSyncJobs.cancelAll() participantsMonitoringJob?.cancel() participantsMonitoringJob = null 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 479475f588..f745611ac0 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 @@ -1111,7 +1111,7 @@ class RtcSessionTest2 { } @Test - fun `mute sync resumes and flushes local state once the SFU is ready`() = runTest( + fun `mute sync resumes without replaying the tracks it skipped`() = runTest( testDispatcher, ) { val signalService = mockk(relaxed = true) @@ -1129,9 +1129,8 @@ class RtcSessionTest2 { testScheduler.advanceUntilIdle() assertEquals(true, rtcSession.muteState.value[TrackType.TRACK_TYPE_AUDIO]) - assertTrue(pendingMuteSyncTracks(rtcSession).contains(TrackType.TRACK_TYPE_AUDIO)) - coVerify(exactly = 0) { signalService.updateMuteStates(any()) } assertEquals(false, muteSyncEnabled(rtcSession).get()) + coVerify(exactly = 0) { signalService.updateMuteStates(any()) } RtcSession::class.java.getDeclaredMethod("resumeMuteSync").apply { isAccessible = true @@ -1139,32 +1138,16 @@ class RtcSessionTest2 { } testScheduler.advanceUntilIdle() - coVerify(exactly = 1) { signalService.updateMuteStates(any()) } - assertTrue(pendingMuteSyncTracks(rtcSession).isEmpty()) + // Resume only re-opens the gate. listenToMediaChanges' collectLatest replay is what + // re-signals each track, so resuming must not POST on its own — a flush here races + // that replay and gets cancelled mid-request. assertEquals(true, muteSyncEnabled(rtcSession).get()) - } - - @Test - fun `resumeMuteSync does not flush tracks that were never pending`() = runTest( - testDispatcher, - ) { - val signalService = mockk(relaxed = true) - val (rtcSession, _) = muteSyncSession(signalService) - - rtcSession.enterMigration() - testScheduler.advanceUntilIdle() coVerify(exactly = 0) { signalService.updateMuteStates(any()) } - RtcSession::class.java.getDeclaredMethod("resumeMuteSync").apply { - isAccessible = true - invoke(rtcSession) - } + // The gate really is open: the next change reaches the SFU. + rtcSession.createAndPublishAudioTrack() testScheduler.advanceUntilIdle() - - // muteState is pre-seeded with audio/video/screen-share; none of those were - // recorded while paused, so resume must not POST UpdateMuteStates. - coVerify(exactly = 0) { signalService.updateMuteStates(any()) } - assertEquals(true, muteSyncEnabled(rtcSession).get()) + coVerify(exactly = 1) { signalService.updateMuteStates(any()) } } @Test @@ -1190,13 +1173,6 @@ class RtcSessionTest2 { return field.get(rtcSession) as AtomicBoolean } - @Suppress("UNCHECKED_CAST") - private fun pendingMuteSyncTracks(rtcSession: RtcSession): MutableSet { - val field = RtcSession::class.java.getDeclaredField("pendingMuteSyncTracks") - field.isAccessible = true - return field.get(rtcSession) as MutableSet - } - private fun muteSyncSession( signalService: SignalServerService, ): Pair {