fix(core): pause mute SFU sync during reconnect and migration

Keep recording local mute bits while the session is torn down, and
flush UpdateMuteStates only after the active SFU is ready. Also
clear TrackKeyedJobs in cleanup.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: PratimMallick

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/call/RtcSession.kt

@@ -173,6 +173,7 @@ import stream.video.sfu.signal.UpdateSubscriptionsResponse
 import java.io.InterruptedIOException
 import java.net.SocketTimeoutException
 import java.util.Collections
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 /**
  * Keeps track of which track is being rendered at what resolution.
@@ -328,6 +329,13 @@ public class RtcSession internal constructor(
     },
 ) {
     private val muteStateSyncJobs = TrackKeyedJobs()
+
+    /**
+     * When false, [setMuteState] still records the latest desired mute bits locally but does
+     * not POST [UpdateMuteStatesRequest] — used while this session's SFU is being torn down
+     * for reconnect/migration so a media collector cannot target the old connection.
+     */
+    private val muteSyncEnabled = AtomicBoolean(true)
     private val oneBasedSessionCounter = sessionCounter + 1
 
     /**
@@ -1210,6 +1218,7 @@ public class RtcSession internal constructor(
 
     private suspend fun connectRtc() {
         logger.d { "[connectRtc] #sfu; #track; no args" }
+        resumeMuteSync()
         // step 6 - onNegotiationNeeded will trigger and complete the setup using SetPublisherRequest
         publisher.value?.let {
             listenToMediaChanges()
@@ -1304,6 +1313,8 @@ public class RtcSession internal constructor(
         }
 
         mediaScope.cancel()
+        muteSyncEnabled.set(false)
+        muteStateSyncJobs.cancelAll()
 
         // cleanup all non-local tracks
         supervisorJob.cancel()
@@ -1349,15 +1360,30 @@ public class RtcSession internal constructor(
      *
      * The local mute map is updated with [MutableStateFlow.update] so concurrent collectors
      * 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. Sync resumes
+     * from [connectRtc] once the active SFU is ready.
      */
     private fun setMuteState(isEnabled: Boolean, trackType: TrackType) {
         logger.d { "[setPublishState] #sfu; $trackType isEnabled: $isEnabled" }
 
         muteState.update { it + (trackType to isEnabled) }
+        syncMuteStateToSfu(trackType, isEnabled)
+    }
+
+    private fun syncMuteStateToSfu(trackType: TrackType, isEnabled: Boolean) {
+        if (!muteSyncEnabled.get()) {
+            logger.d {
+                "[syncMuteStateToSfu] deferred until SFU is ready; $trackType isEnabled: $isEnabled"
+            }
+            return
+        }
 
         val currentSfu = sfuUrl
         // Coalesce retries for this track only. Other tracks keep their in-flight RPCs.
         muteStateSyncJobs.launch(coroutineScope, trackType) {
+            if (!muteSyncEnabled.get()) return@launch
             flow {
                 val request = UpdateMuteStatesRequest(
                     session_id = sessionId,
@@ -1372,6 +1398,7 @@ public class RtcSession internal constructor(
                 emit(response)
             }.flowOn(DispatcherProvider.IO).retryWhen { cause, attempt ->
                 if (cause is SessionFatalException) return@retryWhen false
+                if (!muteSyncEnabled.get()) return@retryWhen false
                 val sameValue = muteState.value[trackType] == isEnabled
                 val sameSfu = currentSfu == sfuUrl
                 val isPermanent = isPermanentError(cause)
@@ -1386,6 +1413,14 @@ public class RtcSession internal constructor(
         }
     }
 
+    private fun resumeMuteSync() {
+        if (!muteSyncEnabled.compareAndSet(false, true)) return
+        logger.d { "[resumeMuteSync] flushing latest mute state to the active SFU" }
+        muteState.value.forEach { (trackType, isEnabled) ->
+            syncMuteStateToSfu(trackType, isEnabled)
+        }
+    }
+
     private fun isPermanentError(cause: Throwable): Boolean {
         return false
     }
@@ -2121,6 +2156,7 @@ public class RtcSession internal constructor(
         if (cancelEventJob) eventJob?.cancel()
         iceMonitoringJob?.cancel()
         iceMonitoringJob = null
+        muteSyncEnabled.set(false)
         muteStateSyncJobs.cancelAll()
         participantsMonitoringJob?.cancel()
         participantsMonitoringJob = null

stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/call/TrackKeyedJobsTest.kt

@@ -79,4 +79,34 @@ class TrackKeyedJobsTest {
             scope.coroutineContext[Job]?.cancel()
         }
     }
+
+    @Test
+    fun `cancelAll cancels every track job and clears the map`() = runTest {
+        val scope = CoroutineScope(UnconfinedTestDispatcher(testScheduler) + Job())
+        val jobs = TrackKeyedJobs()
+        val audioHold = CompletableDeferred<Unit>()
+        val videoHold = CompletableDeferred<Unit>()
+        val audioDone = AtomicBoolean(false)
+        val videoDone = AtomicBoolean(false)
+
+        try {
+            jobs.launch(scope, TrackType.TRACK_TYPE_AUDIO) {
+                audioHold.await()
+                audioDone.set(true)
+            }
+            jobs.launch(scope, TrackType.TRACK_TYPE_VIDEO) {
+                videoHold.await()
+                videoDone.set(true)
+            }
+
+            jobs.cancelAll()
+            audioHold.complete(Unit)
+            videoHold.complete(Unit)
+
+            assertThat(audioDone.get()).isFalse()
+            assertThat(videoDone.get()).isFalse()
+        } finally {
+            scope.coroutineContext[Job]?.cancel()
+        }
+    }
 }

stream-video-android-core/src/test/kotlin/io/getstream/video/android/core/rtc/RtcSessionTest2.kt

@@ -80,6 +80,7 @@ import stream.video.sfu.models.WebsocketReconnectStrategy
 import stream.video.sfu.signal.StartNoiseCancellationRequest
 import stream.video.sfu.signal.StopNoiseCancellationRequest
 import java.io.InterruptedIOException
+import java.util.concurrent.atomic.AtomicBoolean
 
 class RtcSessionTest2 {
 
@@ -926,6 +927,53 @@ class RtcSessionTest2 {
         }
     }
 
+    @Test
+    fun `mute collectors during migration keep local state and do not call the old SFU`() =
+        runTest(testDispatcher) {
+            val signalService = mockk<SignalServerService>(relaxed = true)
+            ownCapabilitiesFlow.value = listOf(OwnCapability.SendAudio)
+            val (rtcSession, publisherMock) = muteSyncSession(signalService)
+            rtcSession.publisher.value = publisherMock
+            val audioTrack = mockk<org.webrtc.AudioTrack>(relaxed = true)
+            coEvery {
+                publisherMock.publishStream(any(), TrackType.TRACK_TYPE_AUDIO)
+            } returns audioTrack
+
+            rtcSession.enterMigration()
+            rtcSession.createAndPublishAudioTrack()
+            testScheduler.advanceUntilIdle()
+
+            assertEquals(true, rtcSession.muteState.value[TrackType.TRACK_TYPE_AUDIO])
+            coVerify(exactly = 0) { signalService.updateMuteStates(any()) }
+        }
+
+    @Test
+    fun `mute sync resumes and flushes local state once the SFU is ready`() = runTest(
+        testDispatcher,
+    ) {
+        val signalService = mockk<SignalServerService>(relaxed = true)
+        ownCapabilitiesFlow.value = listOf(OwnCapability.SendAudio)
+        val (rtcSession, publisherMock) = muteSyncSession(signalService)
+        rtcSession.publisher.value = publisherMock
+        val audioTrack = mockk<org.webrtc.AudioTrack>(relaxed = true)
+        coEvery {
+            publisherMock.publishStream(any(), TrackType.TRACK_TYPE_AUDIO)
+        } returns audioTrack
+
+        rtcSession.enterMigration()
+        rtcSession.createAndPublishAudioTrack()
+        testScheduler.advanceUntilIdle()
+        coVerify(exactly = 0) { signalService.updateMuteStates(any()) }
+        assertEquals(false, muteSyncEnabled(rtcSession).get())
+
+        RtcSession::class.java.getDeclaredMethod("resumeMuteSync").apply {
+            isAccessible = true
+            invoke(rtcSession)
+        }
+
+        assertEquals(true, muteSyncEnabled(rtcSession).get())
+    }
+
     @Test
     fun `stopNoiseCancellation sends the request to the SFU with the session id`() = runTest {
         // Given
@@ -943,6 +991,44 @@ class RtcSessionTest2 {
         }
     }
 
+    private fun muteSyncEnabled(rtcSession: RtcSession): AtomicBoolean {
+        val field = RtcSession::class.java.getDeclaredField("muteSyncEnabled")
+        field.isAccessible = true
+        return field.get(rtcSession) as AtomicBoolean
+    }
+
+    private fun muteSyncSession(
+        signalService: SignalServerService,
+    ): Pair<RtcSession, Publisher> {
+        val mockModule = mockk<SfuConnectionModule>(relaxed = true) {
+            every { api } returns signalService
+        }
+        val rtcSession = spyk(
+            RtcSession(
+                client = mockStreamVideo,
+                powerManager = mockPowerManager,
+                call = mockCall,
+                sessionManager = CallSessionManager(),
+                sessionId = "session-id",
+                apiKey = "api-key",
+                lifecycle = mockLifecycle,
+                sfuUrl = "https://test-sfu.stream.com",
+                sfuWsUrl = "wss://test-sfu.stream.com",
+                sfuToken = "fake-sfu-token",
+                sfuName = "test-sfu-edge",
+                clientImpl = mockVideoClient,
+                coroutineScope = testScope,
+                rtcSessionScope = testScope,
+                remoteIceServers = emptyList(),
+                sfuConnectionModuleProvider = { mockModule },
+                sfuAnalytics = SfuAnalytics.getFakeSfuAnalytics(),
+            ),
+            recordPrivateCalls = true,
+        )
+        val publisherMock = mockk<Publisher>(relaxed = true)
+        return rtcSession to publisherMock
+    }
+
     private fun noiseCancellationSession(
         signalService: SignalServerService,
     ): Pair<RtcSession, SfuConnectionModule> {