Isolate per-track mute RPCs and atomic mute map

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

@@ -326,7 +326,7 @@ public class RtcSession internal constructor(
         )
     },
 ) {
-    private var muteStateSyncJob: Job? = null
+    private val muteStateSyncJobs = TrackKeyedJobs()
     private val oneBasedSessionCounter = sessionCounter + 1
 
     /**
@@ -1341,6 +1341,10 @@ public class RtcSession internal constructor(
      * matches the web SDK, which only signals the track(s) that actually changed. On SFU
      * (re)connect/migration each enabled track is re-signalled individually via
      * [listenToMediaChanges], so the full state is still restored.
+     *
+     * Sync jobs are keyed by [trackType]. Cancelling a shared job used to drop an in-flight
+     * audio unmute when video (or screen-share) published a moment later — reconnect restarts
+     * [listenToMediaChanges] and fires those collectors together.
      */
     private fun setMuteState(isEnabled: Boolean, trackType: TrackType) {
         logger.d { "[setPublishState] #sfu; $trackType isEnabled: $isEnabled" }
@@ -1352,12 +1356,8 @@ public class RtcSession internal constructor(
         muteState.value = new
 
         val currentSfu = sfuUrl
-        // prevent running multiple of these at the same time
-        // if there's already a job active. cancel it
-        muteStateSyncJob?.cancel()
-        // start a new job
-        // this code is a bit more complicated due to the retry behaviour
-        muteStateSyncJob = coroutineScope.launch {
+        // Coalesce retries for this track only. Other tracks keep their in-flight RPCs.
+        muteStateSyncJobs.launch(coroutineScope, trackType) {
             flow {
                 val request = UpdateMuteStatesRequest(
                     session_id = sessionId,
@@ -1372,7 +1372,7 @@ public class RtcSession internal constructor(
                 emit(response)
             }.flowOn(DispatcherProvider.IO).retryWhen { cause, attempt ->
                 if (cause is SessionFatalException) return@retryWhen false
-                val sameValue = new == muteState.value
+                val sameValue = muteState.value[trackType] == isEnabled
                 val sameSfu = currentSfu == sfuUrl
                 val isPermanent = isPermanentError(cause)
                 val willRetry = !isPermanent && sameValue && sameSfu && attempt < 30
@@ -2121,8 +2121,7 @@ public class RtcSession internal constructor(
         if (cancelEventJob) eventJob?.cancel()
         iceMonitoringJob?.cancel()
         iceMonitoringJob = null
-        muteStateSyncJob?.cancel()
-        muteStateSyncJob = null
+        muteStateSyncJobs.cancelAll()
         participantsMonitoringJob?.cancel()
         participantsMonitoringJob = null
         serialProcessor.stop()

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

@@ -0,0 +1,47 @@
+/*
+ * 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.call
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+import stream.video.sfu.models.TrackType
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * One job per [TrackType]. Replacing a job cancels only that track's previous work, so an
+ * in-flight audio mute RPC is not dropped when video or screen-share sync starts.
+ */
+internal class TrackKeyedJobs {
+    private val jobs = ConcurrentHashMap<TrackType, Job>()
+
+    fun launch(
+        scope: CoroutineScope,
+        trackType: TrackType,
+        block: suspend CoroutineScope.() -> Unit,
+    ) {
+        val job = scope.launch(start = CoroutineStart.LAZY, block = block)
+        jobs.put(trackType, job)?.cancel()
+        job.start()
+    }
+
+    fun cancelAll() {
+        jobs.values.forEach { it.cancel() }
+        jobs.clear()
+    }
+}

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

@@ -0,0 +1,82 @@
+/*
+ * 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.call
+
+import com.google.common.truth.Truth.assertThat
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+import stream.video.sfu.models.TrackType
+import java.util.concurrent.atomic.AtomicBoolean
+
+class TrackKeyedJobsTest {
+
+    @Test
+    fun `launching a second track does not cancel the first track's job`() = runTest {
+        val scope = CoroutineScope(UnconfinedTestDispatcher(testScheduler) + Job())
+        val jobs = TrackKeyedJobs()
+        val release = CompletableDeferred<Unit>()
+        val audioDone = AtomicBoolean(false)
+        val videoDone = AtomicBoolean(false)
+
+        try {
+            jobs.launch(scope, TrackType.TRACK_TYPE_AUDIO) {
+                release.await()
+                audioDone.set(true)
+            }
+            jobs.launch(scope, TrackType.TRACK_TYPE_VIDEO) {
+                release.await()
+                videoDone.set(true)
+            }
+
+            release.complete(Unit)
+
+            assertThat(audioDone.get()).isTrue()
+            assertThat(videoDone.get()).isTrue()
+        } finally {
+            scope.coroutineContext[Job]?.cancel()
+        }
+    }
+
+    @Test
+    fun `launching the same track cancels the previous job`() = runTest {
+        val scope = CoroutineScope(UnconfinedTestDispatcher(testScheduler) + Job())
+        val jobs = TrackKeyedJobs()
+        val firstHold = CompletableDeferred<Unit>()
+        val firstDone = AtomicBoolean(false)
+        val secondDone = AtomicBoolean(false)
+
+        try {
+            jobs.launch(scope, TrackType.TRACK_TYPE_AUDIO) {
+                firstHold.await()
+                firstDone.set(true)
+            }
+            jobs.launch(scope, TrackType.TRACK_TYPE_AUDIO) {
+                secondDone.set(true)
+            }
+            firstHold.complete(Unit)
+
+            assertThat(firstDone.get()).isFalse()
+            assertThat(secondDone.get()).isTrue()
+        } finally {
+            scope.coroutineContext[Job]?.cancel()
+        }
+    }
+}