fix(settings): Apply manual channel writes in order (#6077)

feature/settings/src/commonMain/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModel.kt

@@ -19,7 +19,9 @@ package org.meshtastic.feature.settings.radio
 import androidx.lifecycle.ViewModel
 import androidx.lifecycle.viewModelScope
 import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -33,6 +35,9 @@ import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 import org.jetbrains.compose.resources.StringResource
 import org.koin.core.annotation.InjectedParam
 import org.koin.core.annotation.KoinViewModel
@@ -76,6 +81,7 @@ import org.meshtastic.core.resources.key_backup_restore_failed
 import org.meshtastic.core.resources.key_backup_restored
 import org.meshtastic.core.resources.key_backup_saved
 import org.meshtastic.core.resources.timeout
+import org.meshtastic.core.resources.unknown_error
 import org.meshtastic.core.ui.util.SnackbarManager
 import org.meshtastic.core.ui.util.getChannelList
 import org.meshtastic.core.ui.viewmodel.safeLaunch
@@ -98,8 +104,11 @@ import org.meshtastic.proto.LocalModuleConfig
 import org.meshtastic.proto.MeshPacket
 import org.meshtastic.proto.ModuleConfig
 import org.meshtastic.proto.User
+import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
 
+internal val MANUAL_CHANNEL_WRITE_DELAY: Duration = 1.seconds
+
 /** Data class that represents the current RadioConfig state. */
 data class RadioConfigState(
     val isLocal: Boolean = false,
@@ -218,6 +227,9 @@ open class RadioConfigViewModel(
     val mqttProbeStatus: StateFlow<MqttProbeStatus?> = _mqttProbeStatus.asStateFlow()
 
     private var probeJob: Job? = null
+    private val channelUpdateMutex = Mutex()
+    private var manualChannelBatchEnqueueing = false
+    private val manualChannelBatchRequestIds = mutableSetOf<Int>()
 
     /**
      * Run a one-shot reachability/credentials probe against an MQTT broker. Cancels any in-flight probe before starting
@@ -249,6 +261,7 @@ open class RadioConfigViewModel(
         get() = _destNode
 
     private val requestIds = MutableStateFlow(hashSetOf<Int>())
+    private val requestTimeoutJobs = mutableMapOf<Int, Job>()
     private val _radioConfigState = MutableStateFlow(RadioConfigState())
     val radioConfigState: StateFlow<RadioConfigState> = _radioConfigState
 
@@ -401,22 +414,83 @@ open class RadioConfigViewModel(
         }
     }
 
+    @Suppress("TooGenericExceptionCaught")
     fun updateChannels(new: List<ChannelSettings>, old: List<ChannelSettings>) {
         val destNum = destNum ?: destNode.value?.num ?: return
-        getChannelList(new, old).forEach { channel ->
-            safeLaunch(tag = "setRemoteChannel") {
-                val packetId = radioConfigUseCase.setRemoteChannel(destNum, channel)
-                registerRequestId(packetId)
-            }
-        }
 
-        if (destNum == myNodeNum) {
-            safeLaunch(tag = "migrateChannels") {
-                packetRepository.migrateChannelsByPSK(old, new)
-                radioConfigRepository.replaceAllSettings(new)
+        safeLaunch(tag = "setRemoteChannels") {
+            // Manual channel saves are an ordered batch: only update canonical local state after every write request is
+            // enqueued. Serialize batches so two ordered write plans cannot interleave on the radio link, and diff
+            // each queued save against the canonical list at the moment it starts.
+            channelUpdateMutex.withLock {
+                val current = radioConfigState.value.channelList.ifEmpty { old }
+                val updatePlan = getManualChannelUpdatePlan(new, current)
+                if (updatePlan.isEmpty()) return@withLock
+                if (!beginManualChannelBatch(updatePlan.size)) return@withLock
+                val batchRequestIds = mutableSetOf<Int>()
+
+                try {
+                    applyManualChannelUpdatePlan(
+                        updatePlan = updatePlan,
+                        currentSettings = current,
+                        finalSettings = new,
+                        writeChannel = { channel -> radioConfigUseCase.setRemoteChannel(destNum, channel) },
+                        registerRequestId = { packetId ->
+                            batchRequestIds.add(packetId)
+                            registerManualChannelBatchRequestId(packetId)
+                        },
+                        onInterrupted = { result ->
+                            reconcileInterruptedManualChannelUpdate(
+                                destNum = destNum,
+                                oldSettings = current,
+                                appliedSettings = result.appliedSettings,
+                            )
+                        },
+                    )
+                    commitManualChannelSettings(destNum = destNum, oldSettings = current, newSettings = new)
+                    finishManualChannelBatch()
+                } catch (e: CancellationException) {
+                    abortManualChannelBatch(batchRequestIds)
+                    throw e
+                } catch (e: Throwable) {
+                    abortManualChannelBatch(batchRequestIds)
+                    Logger.w(e) { "Manual channel update failed after enqueue" }
+                    e.message?.let(::sendError) ?: sendError(Res.string.unknown_error)
+                }
             }
         }
-        _radioConfigState.update { it.copy(channelList = new) }
+    }
+
+    private fun getManualChannelUpdatePlan(new: List<ChannelSettings>, old: List<ChannelSettings>): List<Channel> =
+        getChannelList(new, old).sortedBy { it.index }
+
+    private suspend fun commitManualChannelSettings(
+        destNum: Int,
+        oldSettings: List<ChannelSettings>,
+        newSettings: List<ChannelSettings>,
+    ) {
+        withContext(NonCancellable) {
+            if (destNum == myNodeNum) {
+                packetRepository.migrateChannelsByPSK(oldSettings, newSettings)
+                radioConfigRepository.replaceAllSettings(newSettings)
+            }
+            _radioConfigState.update { it.copy(channelList = newSettings) }
+        }
+    }
+
+    private suspend fun reconcileInterruptedManualChannelUpdate(
+        destNum: Int,
+        oldSettings: List<ChannelSettings>,
+        appliedSettings: List<ChannelSettings>,
+    ) {
+        withContext(NonCancellable) {
+            Logger.w { "Reconciling interrupted manual channel update appliedSettings=${appliedSettings.size}" }
+            if (destNum == myNodeNum) {
+                packetRepository.migrateChannelsByPSK(oldSettings, appliedSettings)
+                radioConfigRepository.replaceAllSettings(appliedSettings)
+            }
+            _radioConfigState.update { it.copy(channelList = appliedSettings) }
+        }
     }
 
     fun setConfig(config: Config) {
@@ -628,7 +702,7 @@ open class RadioConfigViewModel(
     }
 
     fun clearPacketResponse() {
-        requestIds.value = hashSetOf()
+        clearRequestIds()
         _radioConfigState.update { it.copy(responseState = ResponseState.Empty) }
     }
 
@@ -740,6 +814,38 @@ open class RadioConfigViewModel(
         }
     }
 
+    private fun beginManualChannelBatch(total: Int): Boolean {
+        if (hasUnrelatedPendingRequest() || hasPendingRequestRegistration()) {
+            Logger.w { "Manual channel update skipped while another radio request is pending" }
+            return false
+        }
+        manualChannelBatchEnqueueing = true
+        _radioConfigState.update { state ->
+            state.copy(route = "", responseState = ResponseState.Loading(total = total))
+        }
+        return true
+    }
+
+    private fun finishManualChannelBatch() {
+        manualChannelBatchEnqueueing = false
+        if (requestIds.value.isEmpty()) {
+            setResponseStateSuccess()
+        }
+    }
+
+    private fun abortManualChannelBatch(batchRequestIds: Set<Int>) {
+        manualChannelBatchEnqueueing = false
+        removeRequestIds(batchRequestIds)
+    }
+
+    private fun completeSetRequestOrProgressBatch() {
+        if (manualChannelBatchEnqueueing) {
+            incrementCompleted()
+        } else {
+            setResponseStateSuccess()
+        }
+    }
+
     protected fun setResponseStateSuccess() {
         _radioConfigState.update { state ->
             if (state.responseState is ResponseState.Loading) {
@@ -772,7 +878,8 @@ open class RadioConfigViewModel(
     }
 
     private fun registerRequestId(packetId: Int) {
-        requestIds.update { it.apply { add(packetId) } }
+        requestTimeoutJobs.remove(packetId)?.cancel()
+        requestIds.update { it.withPacketId(packetId) }
         _radioConfigState.update { state ->
             if (state.responseState is ResponseState.Loading) {
                 val total = maxOf(requestIds.value.size, state.responseState.total)
@@ -786,15 +893,46 @@ open class RadioConfigViewModel(
         }
 
         val requestTimeout = 30.seconds
-        safeLaunch(tag = "requestTimeout") {
-            delay(requestTimeout)
-            if (requestIds.value.contains(packetId)) {
-                requestIds.update { it.apply { remove(packetId) } }
-                if (requestIds.value.isEmpty()) {
-                    sendError(Res.string.timeout)
+        requestTimeoutJobs[packetId] =
+            safeLaunch(tag = "requestTimeout") {
+                delay(requestTimeout)
+                if (requestIds.value.contains(packetId)) {
+                    removeRequestId(packetId)
+                    if (requestIds.value.isEmpty()) {
+                        sendError(Res.string.timeout)
+                    }
                 }
             }
-        }
+    }
+
+    private fun registerManualChannelBatchRequestId(packetId: Int) {
+        manualChannelBatchRequestIds.add(packetId)
+        registerRequestId(packetId)
+    }
+
+    private fun hasUnrelatedPendingRequest(): Boolean = requestIds.value.any { it !in manualChannelBatchRequestIds }
+
+    private fun hasPendingRequestRegistration(): Boolean = requestIds.value.isEmpty() &&
+        manualChannelBatchRequestIds.isEmpty() &&
+        radioConfigState.value.responseState is ResponseState.Loading
+
+    private fun clearRequestIds() {
+        requestTimeoutJobs.values.forEach { it.cancel() }
+        requestTimeoutJobs.clear()
+        manualChannelBatchRequestIds.clear()
+        requestIds.value = hashSetOf()
+    }
+
+    private fun removeRequestId(packetId: Int) {
+        requestTimeoutJobs.remove(packetId)?.cancel()
+        manualChannelBatchRequestIds.remove(packetId)
+        requestIds.update { it.withoutPacketId(packetId) }
+    }
+
+    private fun removeRequestIds(packetIds: Set<Int>) {
+        packetIds.forEach { requestTimeoutJobs.remove(it)?.cancel() }
+        manualChannelBatchRequestIds.removeAll(packetIds)
+        requestIds.update { ids -> ids.withoutPacketIds(packetIds) }
     }
 
     private fun processPacketResponse(packet: MeshPacket) {
@@ -813,9 +951,9 @@ open class RadioConfigViewModel(
             is RadioResponseResult.Success -> {
                 if (route.isEmpty()) {
                     val data = packet.decoded!!
-                    requestIds.update { it.apply { remove(data.request_id) } }
+                    removeRequestId(data.request_id)
                     if (requestIds.value.isEmpty()) {
-                        setResponseStateSuccess()
+                        completeSetRequestOrProgressBatch()
                     } else {
                         incrementCompleted()
                     }
@@ -938,14 +1076,79 @@ open class RadioConfigViewModel(
         }
 
         val requestId = packet.decoded?.request_id ?: return
-        requestIds.update { it.apply { remove(requestId) } }
+        removeRequestId(requestId)
 
         if (requestIds.value.isEmpty()) {
             if (route.isNotEmpty() && !AdminRoute.entries.any { it.name == route }) {
                 clearPacketResponse()
             } else if (route.isEmpty()) {
-                setResponseStateSuccess()
+                completeSetRequestOrProgressBatch()
             }
         }
     }
 }
+
+internal data class ManualChannelUpdateResult(val packetIds: List<Int>, val finalSettings: List<ChannelSettings>)
+
+internal data class InterruptedManualChannelUpdate(
+    val appliedSettings: List<ChannelSettings>,
+    val appliedWriteCount: Int,
+)
+
+internal suspend fun applyManualChannelUpdatePlan(
+    updatePlan: List<Channel>,
+    currentSettings: List<ChannelSettings>,
+    finalSettings: List<ChannelSettings>,
+    writeChannel: suspend (Channel) -> Int,
+    registerRequestId: (Int) -> Unit,
+    onInterrupted: suspend (InterruptedManualChannelUpdate) -> Unit = {},
+    writeDelay: Duration = MANUAL_CHANNEL_WRITE_DELAY,
+    delayFn: suspend (Duration) -> Unit = { delay(it) },
+): ManualChannelUpdateResult {
+    val packetIds = mutableListOf<Int>()
+    val appliedSettings = currentSettings.toMutableList()
+    var appliedWriteCount = 0
+    var updateComplete = false
+    try {
+        for ((index, channel) in updatePlan.withIndex()) {
+            val packetId = writeChannel(channel)
+            packetIds.add(packetId)
+            registerRequestId(packetId)
+            appliedSettings.applyManualChannelWrite(channel)
+            appliedWriteCount++
+            if (index < updatePlan.lastIndex) {
+                delayFn(writeDelay)
+            }
+        }
+        updateComplete = true
+    } finally {
+        if (!updateComplete && appliedWriteCount > 0) {
+            onInterrupted(
+                InterruptedManualChannelUpdate(
+                    appliedSettings = appliedSettings.toList(),
+                    appliedWriteCount = appliedWriteCount,
+                ),
+            )
+        }
+    }
+    return ManualChannelUpdateResult(packetIds = packetIds, finalSettings = finalSettings)
+}
+
+private fun MutableList<ChannelSettings>.applyManualChannelWrite(channel: Channel) {
+    while (size <= channel.index) {
+        add(ChannelSettings())
+    }
+    this[channel.index] =
+        if (channel.role == Channel.Role.DISABLED) {
+            ChannelSettings()
+        } else {
+            channel.settings ?: ChannelSettings()
+        }
+}
+
+private fun HashSet<Int>.withPacketId(packetId: Int): HashSet<Int> = HashSet(this).apply { add(packetId) }
+
+private fun HashSet<Int>.withoutPacketId(packetId: Int): HashSet<Int> = HashSet(this).apply { remove(packetId) }
+
+private fun HashSet<Int>.withoutPacketIds(packetIds: Set<Int>): HashSet<Int> =
+    HashSet(this).apply { removeAll(packetIds) }

feature/settings/src/commonTest/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModelTest.kt

@@ -34,6 +34,7 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.resetMain
 import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
@@ -67,6 +68,7 @@ import org.meshtastic.core.testing.FakeLockdownCoordinator
 import org.meshtastic.core.testing.FakeNodeRepository
 import org.meshtastic.core.ui.util.SnackbarManager
 import org.meshtastic.feature.settings.navigation.ConfigRoute
+import org.meshtastic.proto.Channel
 import org.meshtastic.proto.ChannelSet
 import org.meshtastic.proto.ChannelSettings
 import org.meshtastic.proto.Config
@@ -84,8 +86,11 @@ import kotlin.test.AfterTest
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
+import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
+import kotlin.time.Duration
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class RadioConfigViewModelTest {
@@ -359,6 +364,315 @@ class RadioConfigViewModelTest {
         assertEquals(new, viewModel.radioConfigState.value.channelList)
     }
 
+    @Test
+    fun `updateChannels writes changed channels sequentially in index order`() = runTest {
+        val node = Node(num = 123, user = User(id = "!123"))
+        nodeRepository.setNodes(listOf(node))
+        viewModel = createViewModel()
+
+        val channelA = ChannelSettings(name = "A")
+        val channelB = ChannelSettings(name = "B")
+        val channelC = ChannelSettings(name = "C")
+        val old = listOf(channelA, channelB, channelC)
+        val new = listOf(channelA, channelC, channelB)
+        val writtenIndexes = mutableListOf<Int>()
+        var activeWrites = 0
+        var maxConcurrentWrites = 0
+
+        everySuspend { radioConfigUseCase.setRemoteChannel(any(), any()) } calls
+            {
+                val channel = it.args[1] as Channel
+                activeWrites++
+                maxConcurrentWrites = maxOf(maxConcurrentWrites, activeWrites)
+                writtenIndexes.add(channel.index)
+                delay(1)
+                activeWrites--
+                writtenIndexes.size
+            }
+
+        viewModel.updateChannels(new, old)
+        advanceUntilIdle()
+
+        assertEquals(listOf(1, 2), writtenIndexes)
+        assertEquals(1, maxConcurrentWrites)
+        assertEquals(new, viewModel.radioConfigState.value.channelList)
+        verifySuspend(exactly(2)) { radioConfigUseCase.setRemoteChannel(123, any()) }
+    }
+
+    @Test
+    fun `updateChannels waits for all ordered writes before success`() = runTest {
+        val node = Node(num = 123, user = User(id = "!123"))
+        val packetFlow = MutableSharedFlow<MeshPacket>()
+        every { serviceRepository.meshPacketFlow } returns packetFlow
+        every { processRadioResponseUseCase(any(), 123, any()) } returns RadioResponseResult.Success
+        nodeRepository.setNodes(listOf(node))
+        viewModel = createViewModel()
+
+        val channelA = ChannelSettings(name = "A")
+        val channelB = ChannelSettings(name = "B")
+        val channelC = ChannelSettings(name = "C")
+        val old = listOf(channelA, channelB, channelC)
+        val new = listOf(channelA, channelC, channelB)
+        var nextPacketId = 40
+
+        everySuspend { radioConfigUseCase.setRemoteChannel(any(), any()) } calls { ++nextPacketId }
+
+        viewModel.updateChannels(new, old)
+        runCurrent()
+
+        packetFlow.emit(MeshPacket(decoded = Data(request_id = 41)))
+        runCurrent()
+
+        val midBatchState = viewModel.radioConfigState.value.responseState
+        assertTrue(midBatchState is ResponseState.Loading)
+        assertEquals(2, midBatchState.total)
+        assertEquals(1, midBatchState.completed)
+
+        advanceTimeBy(MANUAL_CHANNEL_WRITE_DELAY.inWholeMilliseconds)
+        runCurrent()
+
+        packetFlow.emit(MeshPacket(decoded = Data(request_id = 42)))
+        runCurrent()
+
+        assertTrue(viewModel.radioConfigState.value.responseState is ResponseState.Success)
+    }
+
+    @Test
+    fun `updateChannels reconciles applied channel writes when ordered write fails`() = runTest {
+        val node = Node(num = 123, user = User(id = "!123"))
+        val old = fourChannelFixture()
+        val (channelA, _, _, channelD) = old
+        val new = listOf(channelA, channelD)
+        val partiallyApplied = listOf(channelA, channelD, old[2], old[3])
+        val writtenIndexes = mutableListOf<Int>()
+
+        every { radioConfigRepository.channelSetFlow } returns MutableStateFlow(ChannelSet(settings = old))
+        nodeRepository.setNodes(listOf(node))
+        nodeRepository.setMyNodeInfo(myNodeInfo(myNodeNum = 123))
+        viewModel = createViewModel()
+        runCurrent()
+
+        everySuspend { radioConfigUseCase.setRemoteChannel(any(), any()) } calls
+            {
+                val channel = it.args[1] as Channel
+                writtenIndexes.add(channel.index)
+                if (writtenIndexes.size == 2) {
+                    throw IllegalStateException("boom")
+                }
+                channel.index + 100
+            }
+
+        viewModel.updateChannels(new, old)
+        advanceUntilIdle()
+
+        assertEquals(listOf(1, 2), writtenIndexes)
+        assertEquals(partiallyApplied, viewModel.radioConfigState.value.channelList)
+        assertTrue(viewModel.radioConfigState.value.responseState is ResponseState.Error)
+        verifySuspend { packetRepository.migrateChannelsByPSK(old, partiallyApplied) }
+        verifySuspend { radioConfigRepository.replaceAllSettings(partiallyApplied) }
+    }
+
+    @Test
+    fun `updateChannels serializes overlapping channel saves`() = runTest {
+        val node = Node(num = 123, user = User(id = "!123"))
+        val old = fourChannelFixture()
+        val (channelA, _, channelC, channelD) = old
+        val firstNew = listOf(channelA, channelD)
+        val secondNew = listOf(channelA, channelC)
+        val writtenChannels = mutableListOf<String>()
+        var firstWrite = true
+
+        every { radioConfigRepository.channelSetFlow } returns MutableStateFlow(ChannelSet(settings = old))
+        nodeRepository.setNodes(listOf(node))
+        viewModel = createViewModel()
+        runCurrent()
+
+        everySuspend { radioConfigUseCase.setRemoteChannel(any(), any()) } calls
+            {
+                val channel = it.args[1] as Channel
+                writtenChannels.add("${channel.index}:${channel.role}:${channel.settings?.name.orEmpty()}")
+                if (firstWrite) {
+                    firstWrite = false
+                    delay(10_000)
+                }
+                writtenChannels.size
+            }
+
+        viewModel.updateChannels(firstNew, old)
+        runCurrent()
+        viewModel.updateChannels(secondNew, old)
+        advanceUntilIdle()
+
+        assertEquals(listOf("1:SECONDARY:D", "2:DISABLED:", "3:DISABLED:", "1:SECONDARY:C"), writtenChannels)
+        assertEquals(secondNew, viewModel.radioConfigState.value.channelList)
+    }
+
+    @Test
+    fun `updateChannels does not start manual batch while another radio request is pending`() = runTest {
+        val node = Node(num = 123, user = User(id = "!123"))
+        nodeRepository.setNodes(listOf(node))
+        viewModel = createViewModel()
+
+        val old = listOf(ChannelSettings(name = "Old"))
+        val new = listOf(ChannelSettings(name = "New"))
+
+        everySuspend { radioConfigUseCase.getOwner(any()) } calls
+            {
+                delay(10_000)
+                42
+            }
+        everySuspend { radioConfigUseCase.setRemoteChannel(any(), any()) } returns 100
+
+        viewModel.setResponseStateLoading(ConfigRoute.USER)
+        runCurrent()
+
+        viewModel.updateChannels(new, old)
+        runCurrent()
+
+        assertEquals(ConfigRoute.USER.name, viewModel.radioConfigState.value.route)
+        assertTrue(viewModel.radioConfigState.value.responseState is ResponseState.Loading)
+        verifySuspend(exactly(0)) { radioConfigUseCase.setRemoteChannel(any(), any()) }
+
+        advanceUntilIdle()
+    }
+
+    @Test
+    fun `aborted manual batch preserves unrelated pending request`() = runTest {
+        val node = Node(num = 123, user = User(id = "!123"))
+        val packetFlow = MutableSharedFlow<MeshPacket>()
+        val old = fourChannelFixture()
+        val (channelA, _, _, channelD) = old
+        val new = listOf(channelA, channelD)
+        val owner = User(id = "!123", long_name = "Updated")
+        var writeCount = 0
+
+        every { serviceRepository.meshPacketFlow } returns packetFlow
+        nodeRepository.setNodes(listOf(node))
+        viewModel = createViewModel()
+
+        everySuspend { radioConfigUseCase.getOwner(any()) } returns 42
+        everySuspend { radioConfigUseCase.setRemoteChannel(any(), any()) } calls
+            {
+                writeCount++
+                if (writeCount == 2) {
+                    viewModel.setResponseStateLoading(ConfigRoute.USER)
+                    throw IllegalStateException("boom")
+                }
+                100
+            }
+        every { processRadioResponseUseCase(any(), 123, any()) } calls
+            {
+                val pendingRequestIds = it.args[2] as Set<Int>
+                if (42 in pendingRequestIds) RadioResponseResult.Owner(owner) else null
+            }
+
+        viewModel.updateChannels(new, old)
+        runCurrent()
+        advanceTimeBy(MANUAL_CHANNEL_WRITE_DELAY.inWholeMilliseconds)
+        runCurrent()
+
+        packetFlow.emit(MeshPacket(decoded = Data(request_id = 42)))
+        runCurrent()
+
+        assertEquals(owner, viewModel.radioConfigState.value.userConfig)
+    }
+
+    @Test
+    fun `aborted manual batch cancels batch timeout before packet id reuse`() = runTest {
+        val node = Node(num = 123, user = User(id = "!123"))
+        val old = fourChannelFixture()
+        val (channelA, _, _, channelD) = old
+        val new = listOf(channelA, channelD)
+        var writeCount = 0
+
+        nodeRepository.setNodes(listOf(node))
+        viewModel = createViewModel()
+
+        everySuspend { radioConfigUseCase.setRemoteChannel(any(), any()) } calls
+            {
+                writeCount++
+                if (writeCount == 2) throw IllegalStateException("boom")
+                100
+            }
+        everySuspend { radioConfigUseCase.getOwner(any()) } returns 100
+
+        viewModel.updateChannels(new, old)
+        runCurrent()
+        advanceTimeBy(MANUAL_CHANNEL_WRITE_DELAY.inWholeMilliseconds)
+        runCurrent()
+
+        viewModel.setResponseStateLoading(ConfigRoute.USER)
+        runCurrent()
+
+        advanceTimeBy(29_500)
+        runCurrent()
+
+        assertTrue(viewModel.radioConfigState.value.responseState is ResponseState.Loading)
+    }
+
+    @Test
+    fun `applyManualChannelUpdatePlan paces writes except after final channel`() = runTest {
+        val channelA = Channel(index = 1, role = Channel.Role.SECONDARY, settings = ChannelSettings(name = "A"))
+        val channelB = Channel(index = 2, role = Channel.Role.DISABLED, settings = ChannelSettings())
+        val channelC = Channel(index = 3, role = Channel.Role.DISABLED, settings = ChannelSettings())
+        val writtenIndexes = mutableListOf<Int>()
+        val registeredRequestIds = mutableListOf<Int>()
+        val delays = mutableListOf<Duration>()
+
+        val result =
+            applyManualChannelUpdatePlan(
+                updatePlan = listOf(channelA, channelB, channelC),
+                currentSettings = listOf(ChannelSettings(name = "old")),
+                finalSettings = listOf(ChannelSettings(name = "new")),
+                writeChannel = { channel ->
+                    writtenIndexes.add(channel.index)
+                    channel.index + 100
+                },
+                registerRequestId = { registeredRequestIds.add(it) },
+                delayFn = { delays.add(it) },
+            )
+
+        assertEquals(listOf(1, 2, 3), writtenIndexes)
+        assertEquals(listOf(101, 102, 103), result.packetIds)
+        assertEquals(listOf(ChannelSettings(name = "new")), result.finalSettings)
+        assertEquals(listOf(101, 102, 103), registeredRequestIds)
+        assertEquals(listOf(MANUAL_CHANNEL_WRITE_DELAY, MANUAL_CHANNEL_WRITE_DELAY), delays)
+    }
+
+    @Test
+    fun `applyManualChannelUpdatePlan invokes onInterrupted when writeChannel fails mid-plan`() = runTest {
+        val channelA = Channel(index = 1, role = Channel.Role.SECONDARY, settings = ChannelSettings(name = "A"))
+        val channelB = Channel(index = 2, role = Channel.Role.SECONDARY, settings = ChannelSettings(name = "B"))
+        val channelC = Channel(index = 3, role = Channel.Role.SECONDARY, settings = ChannelSettings(name = "C"))
+        val writtenIndexes = mutableListOf<Int>()
+
+        var interrupted: InterruptedManualChannelUpdate? = null
+
+        val error =
+            assertFailsWith<IllegalStateException> {
+                applyManualChannelUpdatePlan(
+                    updatePlan = listOf(channelA, channelB, channelC),
+                    currentSettings = listOf(ChannelSettings(name = "old")),
+                    finalSettings = listOf(ChannelSettings(name = "new")),
+                    writeChannel = { channel ->
+                        writtenIndexes.add(channel.index)
+                        if (channel.index == 2) throw IllegalStateException("boom")
+                        channel.index + 100
+                    },
+                    registerRequestId = {},
+                    onInterrupted = { interrupted = it },
+                    delayFn = {},
+                )
+            }
+
+        assertEquals("boom", error.message)
+        // Channel A (index 1) completed before channel B (index 2) threw.
+        assertEquals(listOf(1, 2), writtenIndexes)
+        assertNotNull(interrupted)
+        assertEquals(1, interrupted!!.appliedWriteCount)
+        assertEquals("A", interrupted!!.appliedSettings[1].name)
+    }
+
     @Test
     fun `setResponseStateLoading for REBOOT calls useCase after config response`() = runTest {
         val node = Node(num = 123, user = User(id = "!123"))
@@ -919,6 +1233,13 @@ class RadioConfigViewModelTest {
         verifySuspend { radioConfigUseCase.setConfig(123, Config(security = decoded)) }
     }
 
+    private fun fourChannelFixture() = listOf(
+        ChannelSettings(name = "A"),
+        ChannelSettings(name = "B"),
+        ChannelSettings(name = "C"),
+        ChannelSettings(name = "D"),
+    )
+
     private fun myNodeInfo(myNodeNum: Int) = MyNodeInfo(
         myNodeNum = myNodeNum,
         hasGPS = false,