From 9b5e3bb5b49dc4c3a7f2dcbfab9f3b8e369abc6e Mon Sep 17 00:00:00 2001 From: Jeremiah K <17190268+jeremiah-k@users.noreply.github.com> Date: Mon, 6 Jul 2026 12:07:20 -0500 Subject: [PATCH] fix(settings): Apply manual channel writes in order (#6077) --- .../settings/radio/RadioConfigViewModel.kt | 251 ++++++++++++-- .../radio/RadioConfigViewModelTest.kt | 321 ++++++++++++++++++ 2 files changed, 548 insertions(+), 24 deletions(-) diff --git a/feature/settings/src/commonMain/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModel.kt b/feature/settings/src/commonMain/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModel.kt index 9de0552b7..fa8a7d199 100644 --- a/feature/settings/src/commonMain/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModel.kt +++ b/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.asStateFlow() private var probeJob: Job? = null + private val channelUpdateMutex = Mutex() + private var manualChannelBatchEnqueueing = false + private val manualChannelBatchRequestIds = mutableSetOf() /** * 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()) + private val requestTimeoutJobs = mutableMapOf() private val _radioConfigState = MutableStateFlow(RadioConfigState()) val radioConfigState: StateFlow = _radioConfigState @@ -401,22 +414,83 @@ open class RadioConfigViewModel( } } + @Suppress("TooGenericExceptionCaught") fun updateChannels(new: List, old: List) { 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() + + 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, old: List): List = + getChannelList(new, old).sortedBy { it.index } + + private suspend fun commitManualChannelSettings( + destNum: Int, + oldSettings: List, + newSettings: List, + ) { + 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, + appliedSettings: List, + ) { + 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) { + 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) { + 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, val finalSettings: List) + +internal data class InterruptedManualChannelUpdate( + val appliedSettings: List, + val appliedWriteCount: Int, +) + +internal suspend fun applyManualChannelUpdatePlan( + updatePlan: List, + currentSettings: List, + finalSettings: List, + 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() + 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.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.withPacketId(packetId: Int): HashSet = HashSet(this).apply { add(packetId) } + +private fun HashSet.withoutPacketId(packetId: Int): HashSet = HashSet(this).apply { remove(packetId) } + +private fun HashSet.withoutPacketIds(packetIds: Set): HashSet = + HashSet(this).apply { removeAll(packetIds) } diff --git a/feature/settings/src/commonTest/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModelTest.kt b/feature/settings/src/commonTest/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModelTest.kt index 055f206b3..e3f95d03f 100644 --- a/feature/settings/src/commonTest/kotlin/org/meshtastic/feature/settings/radio/RadioConfigViewModelTest.kt +++ b/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() + 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() + 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() + + 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() + 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() + 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 + 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() + val registeredRequestIds = mutableListOf() + val delays = mutableListOf() + + 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() + + var interrupted: InterruptedManualChannelUpdate? = null + + val error = + assertFailsWith { + 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,