From e12c4a579d97dab1e8646fcd7bcbe710406d1076 Mon Sep 17 00:00:00 2001 From: Jeremiah K <17190268+jeremiah-k@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:58:43 -0500 Subject: [PATCH] fix(runtime): harden BLE profile, OTA setup, and DB access (#6126) --- .../org/meshtastic/core/ble/BleConnection.kt | 6 +- .../meshtastic/core/ble/KableBleConnection.kt | 35 +++- .../core/database/DatabaseManager.kt | 69 +++++-- .../org/meshtastic/core/testing/FakeBle.kt | 4 + .../feature/firmware/ota/BleOtaTransport.kt | 177 +++++++++++++----- .../feature/firmware/ota/BleScanSupport.kt | 7 +- .../firmware/ota/BleOtaTransportTest.kt | 73 ++++++++ 7 files changed, 306 insertions(+), 65 deletions(-) diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt index 35cb3bee3..c38f38e7d 100644 --- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt +++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt @@ -62,7 +62,11 @@ interface BleConnection { /** Disconnects from the current device. */ suspend fun disconnect() - /** Executes a block within a discovered profile. */ + /** + * Executes [setup] after the requested BLE profile is available. This is shared by radio, OTA, and DFU flows, so + * implementations should keep profile-entry setup bounded and avoid transfer-hot-path work here when a caller can + * reuse an already discovered [BleService]. + */ suspend fun profile( serviceUuid: Uuid, timeout: Duration = 30.seconds, diff --git a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt index 506c8be82..e143eb267 100644 --- a/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt +++ b/core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt @@ -29,15 +29,20 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.TimeoutCancellationException +import kotlinx.coroutines.async +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.job import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeout +import org.meshtastic.core.common.util.ioDispatcher import kotlin.concurrent.Volatile import kotlin.time.Duration import kotlin.time.Duration.Companion.seconds @@ -245,6 +250,7 @@ class KableBleConnection(private val scope: CoroutineScope, private val loggingC _deviceFlow.emit(null) } + @Suppress("ThrowsCount") override suspend fun profile( serviceUuid: Uuid, timeout: Duration, @@ -253,7 +259,34 @@ class KableBleConnection(private val scope: CoroutineScope, private val loggingC val p = peripheral ?: error("Not connected") val cScope = connectionScope ?: error("No active connection scope") val service = KableBleService(p, serviceUuid) - return withTimeout(timeout) { cScope.setup(service) } + return withTimeout(timeout) { + // Shared BLE profile guard: wait for Kable service discovery before handing out the service, and map a + // connection-scope shutdown during caller setup to NotConnectedException instead of waiting for timeout. + withContext(ioDispatcher) { + val profileExecution = async { + p.services.first { it != null } + cScope.setup(service) + } + + val disconnectHandle = + cScope.coroutineContext.job.invokeOnCompletion { + profileExecution.cancel(CancellationException("Connection lost during BLE profile execution")) + } + + try { + profileExecution.await() + } catch (e: CancellationException) { + currentCoroutineContext().ensureActive() + if (!cScope.coroutineContext.job.isActive) { + throw NotConnectedException("Connection lost during BLE profile execution") + } + throw e + } finally { + disconnectHandle.dispose() + profileExecution.cancel() + } + } + } } override fun maximumWriteValueLength(writeType: BleWriteType): Int? = peripheral?.negotiatedMaxWriteLength() diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt index bbfe75cbe..f098ad462 100644 --- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt +++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt @@ -31,7 +31,9 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.StateFlow @@ -279,36 +281,74 @@ open class DatabaseManager( Logger.d { "Closed inactive database ${anonymizeDbName(dbName)} to free connections" } } - private val limitedIo = dispatchers.io.limitedParallelism(4) + // Short-term runtime containment: route withDb entry through a single-lane dispatcher to narrow the Room/SQLite + // connection-pool churn window seen during device/firmware update flows. Room suspend DAOs may continue on Room's + // own executor after suspension, so this is not a strict global DB-I/O serialization guarantee. Preserve bounded + // one-shot DB-critical blocks through cancellation, then re-check cancellation so stale callers do not continue + // after the DB releases. Long-lived Flow/Paging reads must stay out of withDb; revisit after direct currentDb.value + // callers are audited and safe DB concurrency can be restored. + private val limitedIo = dispatchers.io.limitedParallelism(1) /** Execute [block] with the current DB instance. Retries once if the pool closes during a DB switch. */ @Suppress("TooGenericExceptionCaught") - override suspend fun withDb(block: suspend (MeshtasticDatabase) -> T): T? = withContext(limitedIo) { - val db = _currentDb.value ?: return@withContext null + override suspend fun withDb(block: suspend (MeshtasticDatabase) -> T): T? { + val queuedAt = nowMillis + return withContext(limitedIo) { + val queuedMillis = nowMillis - queuedAt + if (queuedMillis >= WITH_DB_SLOW_OPERATION_MS) { + Logger.w { "withDb waited ${queuedMillis}ms for the temporary DB containment lane" } + } + + val startedAt = nowMillis + try { + withCurrentDb(block) + } finally { + val elapsedMillis = nowMillis - startedAt + if (elapsedMillis >= WITH_DB_SLOW_OPERATION_MS) { + Logger.w { + "withDb callback took ${elapsedMillis}ms on the temporary DB containment lane; persistent " + + "slow logs indicate DB access path should be revisited" + } + } + } + } + } + + @Suppress("ReturnCount", "ThrowsCount", "TooGenericExceptionCaught") + private suspend fun withCurrentDb(block: suspend (MeshtasticDatabase) -> T): T? { + val db = _currentDb.value ?: return null val active = currentDbName markLastUsed(active) try { - block(db) + return runCancellableDbBlock(db, block) } catch (e: CancellationException) { throw e // Preserve structured concurrency cancellation propagation. } catch (e: Exception) { // If the active database switched while we held a reference to the old one, // and the exception indicates a closed pool/connection, retry with the new DB. val retryDb = _currentDb.value - if (retryDb != null && retryDb !== db && isDbClosedException(e)) { - Logger.w { "withDb: database closed during switch (${e.message}), retrying with current DB" } - try { - block(retryDb) - } catch (retryEx: Exception) { - retryEx.addSuppressed(e) - throw retryEx - } - } else { - throw e + if (retryDb == null || retryDb === db || !isDbClosedException(e)) throw e + + Logger.w { "withDb: database closed during switch (${e.message}), retrying with current DB" } + return try { + runCancellableDbBlock(retryDb, block) + } catch (retryCancel: CancellationException) { + throw retryCancel + } catch (retryEx: Exception) { + retryEx.addSuppressed(e) + throw retryEx } } } + private suspend fun runCancellableDbBlock(db: MeshtasticDatabase, block: suspend (MeshtasticDatabase) -> T): T { + // Keep withDb callbacks bounded and one-shot: NonCancellable can hold the containment lane until this returns. + currentCoroutineContext().ensureActive() + val result = withContext(NonCancellable) { block(db) } + currentCoroutineContext().ensureActive() + return result + } + private fun isDbClosedException(e: Exception): Boolean = generateSequence(e) { it.cause } .any { throwable -> val msg = throwable.message?.lowercase() ?: return@any false @@ -317,6 +357,7 @@ open class DatabaseManager( private companion object { private const val BACKFILL_COLD_START_DELAY_MS = 2_000L + private const val WITH_DB_SLOW_OPERATION_MS = 1_000L val DB_TERMS = listOf("pool", "database", "connection", "sqlite") } diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt index b792ee9b1..040a9e075 100644 --- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt +++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt @@ -120,6 +120,9 @@ class FakeBleConnection : /** Number of times [connectAndAwait] has been invoked (including failures). */ var connectAndAwaitCalls: Int = 0 + /** Number of times [profile] has been invoked. */ + var profileCalls: Int = 0 + /** Externally simulate a remote disconnect (e.g. node power-cycle) for tests that exercise reconnect. */ fun simulateRemoteDisconnect(reason: DisconnectReason = DisconnectReason.Timeout) { _connectionState.value = BleConnectionState.Disconnected(reason) @@ -168,6 +171,7 @@ class FakeBleConnection : timeout: Duration, setup: suspend CoroutineScope.(BleService) -> T, ): T { + profileCalls++ if (serviceUuid in missingServices) { throw NoSuchElementException("Service $serviceUuid not found") } diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt index cd9d7b781..c893f5525 100644 --- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt +++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt @@ -20,18 +20,27 @@ import co.touchlab.kermit.Logger import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.catch +import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull import org.meshtastic.core.ble.BleCharacteristic import org.meshtastic.core.ble.BleConnectionFactory import org.meshtastic.core.ble.BleConnectionState import org.meshtastic.core.ble.BleDevice import org.meshtastic.core.ble.BleScanner +import org.meshtastic.core.ble.BleService import org.meshtastic.core.ble.BleWriteType import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_NOTIFY_CHARACTERISTIC import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_SERVICE_UUID @@ -56,89 +65,139 @@ class BleOtaTransport( private val otaChar = BleCharacteristic(OTA_WRITE_CHARACTERISTIC) private val txChar = BleCharacteristic(OTA_NOTIFY_CHARACTERISTIC) - private val responseChannel = Channel(Channel.UNLIMITED) + @Volatile private var otaService: BleService? = null + + @Volatile private var responseChannel = Channel(Channel.UNLIMITED) // Written from the connectionState collector (Dispatchers.Default) and read by the streaming loop's // connection-loss guard (Dispatchers.IO); @Volatile ensures the guard sees a mid-transfer disconnect. @Volatile private var isConnected = false + @Volatile private var notificationJob: Job? = null + + @Volatile private var connectionStateJob: Job? = null + /** Scan for the device by MAC address (or MAC+1 for OTA mode) with retries. */ private suspend fun scanForOtaDevice(): BleDevice? { val otaAddress = calculateMacPlusOne(address) val targetAddresses = setOf(address, otaAddress) - Logger.i { "BLE OTA: Will match addresses: $targetAddresses" } + Logger.i { "BLE OTA: Will match target OTA device addresses" } return scanForBleDevice(scanner = scanner, tag = "BLE OTA", serviceUuid = OTA_SERVICE_UUID) { it.address in targetAddresses } } - @Suppress("MagicNumber") + @Suppress("MagicNumber", "LongMethod") override suspend fun connect(): Result = safeCatching { + otaService = null + notificationJob?.cancel() + notificationJob = null + connectionStateJob?.cancel() + connectionStateJob = null + isConnected = false + responseChannel.close() + val connectResponseChannel = Channel(Channel.UNLIMITED) + responseChannel = connectResponseChannel + Logger.i { "BLE OTA: Waiting $REBOOT_DELAY for device to reboot into OTA mode..." } delay(REBOOT_DELAY) - Logger.i { "BLE OTA: Connecting to $address using Kable..." } + Logger.i { "BLE OTA: Connecting to OTA device using Kable..." } val device = scanForOtaDevice() ?: throw OtaProtocolException.ConnectionFailed( - "Device not found at address $address. " + - "Ensure the device has rebooted into OTA mode and is advertising.", + "Device not found. Ensure the device has rebooted into OTA mode and is advertising.", ) - bleConnection.connectionState - .onEach { state -> - Logger.d { "BLE OTA: Connection state changed to $state" } - isConnected = state is BleConnectionState.Connected - } - .launchIn(transportScope) + connectionStateJob = + bleConnection.connectionState + .onEach { state -> + Logger.d { "BLE OTA: Connection state changed to $state" } + isConnected = state is BleConnectionState.Connected + } + .launchIn(transportScope) try { val finalState = bleConnection.connectAndAwait(device, CONNECTION_TIMEOUT) if (finalState is BleConnectionState.Disconnected) { - Logger.w { "BLE OTA: Failed to connect to ${device.address} (state=$finalState)" } - throw OtaProtocolException.ConnectionFailed("Failed to connect to device at address ${device.address}") + Logger.w { "BLE OTA: Failed to connect to OTA device (state=$finalState)" } + throw OtaProtocolException.ConnectionFailed("Failed to connect to OTA device") } } catch (@Suppress("SwallowedException") e: kotlinx.coroutines.TimeoutCancellationException) { - Logger.w { "BLE OTA: Timed out waiting to connect to ${device.address}. Error: ${e.message}" } - throw OtaProtocolException.Timeout("Timed out connecting to device at address ${device.address}") + currentCoroutineContext().ensureActive() + Logger.w { "BLE OTA: Timed out waiting to connect to OTA device. Error: ${e.message}" } + throw OtaProtocolException.Timeout("Timed out connecting to OTA device") } - Logger.i { "BLE OTA: Connected to ${device.address}, discovering services..." } + Logger.i { "BLE OTA: Connected to OTA device, discovering services..." } - bleConnection.profile(OTA_SERVICE_UUID) { service -> - // Log negotiated MTU for diagnostics - val maxLen = bleConnection.maximumWriteValueLength(BleWriteType.WITHOUT_RESPONSE) - Logger.i { "BLE OTA: Service ready. Max write value length: $maxLen bytes" } + try { + bleConnection.profile(OTA_SERVICE_UUID) { service -> + service.requireOtaCharacteristics() - // Collect responses. onSubscription fires when the CCCD write completes — a precise readiness - // signal; the settle below is a conservative cushion. - val subscribed = CompletableDeferred() - service - .observe(txChar) { - Logger.d { "BLE OTA: TX characteristic subscribed" } - subscribed.complete(Unit) - } - .onEach { notifyBytes -> - try { - val response = notifyBytes.decodeToString() - Logger.d { "BLE OTA: Received response: $response" } - responseChannel.trySend(response) - } catch (@Suppress("TooGenericExceptionCaught") e: Exception) { - Logger.e(e) { "BLE OTA: Failed to decode response bytes" } + // Log negotiated MTU for diagnostics + val maxLen = bleConnection.maximumWriteValueLength(BleWriteType.WITHOUT_RESPONSE) + Logger.i { "BLE OTA: Service ready. Max write value length: $maxLen bytes" } + + // Collect notification responses. Kable writes the CCCD descriptor as part of flow collection + // startup. Prefer the onSubscription readiness callback, but bound the wait because some OTA loaders + // never report the CCCD write completion even though notifications still flow. + val notificationsReady = CompletableDeferred() + notificationJob = + launch(start = CoroutineStart.UNDISPATCHED) { + service + .observe(txChar) { + Logger.d { "BLE OTA: TX characteristic subscribed" } + notificationsReady.complete(Unit) + } + .onEach { notifyBytes -> + try { + val response = notifyBytes.decodeToString() + Logger.d { "BLE OTA: Received response: $response" } + connectResponseChannel.trySend(response) + } catch (@Suppress("TooGenericExceptionCaught") e: Exception) { + Logger.e(e) { "BLE OTA: Failed to decode response bytes" } + } + } + .catch { e -> + Logger.e(e) { "BLE OTA: Error in TX characteristic notification flow" } + if (!notificationsReady.isCompleted) { + notificationsReady.completeExceptionally(e) + } + connectResponseChannel.close(e) + } + .collect() + } + + val confirmed = withTimeoutOrNull(SUBSCRIPTION_SETTLE) { notificationsReady.await() } != null + if (confirmed) { + Logger.i { "BLE OTA: TX notifications subscribed" } + } else { + Logger.w { + "BLE OTA: TX notification subscription not confirmed after $SUBSCRIPTION_SETTLE; " + + "continuing with bounded settle fallback" } } - .catch { e -> - if (!subscribed.isCompleted) subscribed.completeExceptionally(e) - Logger.e(e) { "BLE OTA: Error in TX characteristic subscription" } - } - .launchIn(this) - - subscribed.await() - // Conservative settle after CCCD confirmation before issuing commands. - delay(SUBSCRIPTION_SETTLE) - Logger.i { "BLE OTA: Service discovered and ready" } + otaService = service + Logger.i { "BLE OTA: Service discovered and ready" } + } + } catch (e: TimeoutCancellationException) { + otaService = null + currentCoroutineContext().ensureActive() + Logger.w { "BLE OTA: Timed out waiting for OTA service discovery. Error: ${e.message}" } + throw OtaProtocolException.Timeout("Timed out waiting for BLE OTA service discovery") + } catch (e: OtaProtocolException) { + otaService = null + throw e + } catch (e: kotlinx.coroutines.CancellationException) { + otaService = null + throw e + } catch (@Suppress("TooGenericExceptionCaught") e: Exception) { + otaService = null + Logger.w(e) { "BLE OTA: Failed to prepare OTA service" } + throw OtaProtocolException.ConnectionFailed("Failed to prepare BLE OTA service", e) } } @@ -264,6 +323,12 @@ class BleOtaTransport( } override suspend fun close() { + notificationJob?.cancel() + notificationJob = null + connectionStateJob?.cancel() + connectionStateJob = null + otaService = null + responseChannel.close() bleConnection.disconnect() isConnected = false transportScope.cancel() @@ -275,6 +340,7 @@ class BleOtaTransport( } private suspend fun writeData(data: ByteArray, writeType: BleWriteType): Int { + val service = otaService ?: throw OtaProtocolException.TransferFailed("BLE OTA service is not ready") // takeIf { it > 0 }: a non-positive negotiated length would stall the loop (offset never advances); fall // back to a single whole-buffer write instead of looping forever. val maxLen = bleConnection.maximumWriteValueLength(writeType)?.takeIf { it > 0 } ?: data.size @@ -286,7 +352,7 @@ class BleOtaTransport( val chunkSize = minOf(data.size - offset, maxLen) val packet = data.copyOfRange(offset, offset + chunkSize) - bleConnection.profile(OTA_SERVICE_UUID) { service -> service.write(otaChar, packet, writeType) } + service.write(otaChar, packet, writeType) offset += chunkSize packetsSent++ @@ -301,6 +367,23 @@ class BleOtaTransport( OtaProtocolException.Timeout("Timeout waiting for response after $timeout") } + private fun BleService.requireOtaCharacteristics() { + val missing = mutableListOf() + if (!hasCharacteristic(txChar)) { + missing.add("TX notify characteristic $txChar") + } + if (!hasCharacteristic(otaChar)) { + missing.add("OTA write characteristic $otaChar") + } + + if (missing.isNotEmpty()) { + throw OtaProtocolException.ConnectionFailed( + "ESP32 OTA service was missing required characteristics after BLE service discovery: " + + missing.joinToString(separator = "; "), + ) + } + } + companion object { private val CONNECTION_TIMEOUT = 15.seconds private val SUBSCRIPTION_SETTLE = 500.milliseconds diff --git a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt index 6fe8fd8b9..6d3608ad6 100644 --- a/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt +++ b/feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt @@ -20,7 +20,9 @@ import co.touchlab.kermit.Logger import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.flow.onEach @@ -79,12 +81,12 @@ internal suspend fun scanForBleDevice( .scan(timeout = scanTimeout, serviceUuid = serviceUuid) .onEach { d -> if (foundDevices.add(d.address)) { - Logger.d { "$tag: Scan found device: ${d.address} (name=${d.name})" } + Logger.d { "$tag: Scan found candidate device (name=${d.name})" } } } .firstOrNull(predicate) if (device != null) { - Logger.i { "$tag: Found target device at ${device.address}" } + Logger.i { "$tag: Found target device" } return device } Logger.w { "$tag: Target not in ${foundDevices.size} devices found" } @@ -104,6 +106,7 @@ internal suspend fun scanForBleDevice( internal suspend fun Channel.receiveWithin(timeout: Duration, onTimeout: () -> Throwable): T = try { withTimeout(timeout) { receive() } } catch (@Suppress("SwallowedException") e: TimeoutCancellationException) { + currentCoroutineContext().ensureActive() throw onTimeout() } diff --git a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt index 73e487402..beca5d8d4 100644 --- a/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt +++ b/feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt @@ -22,6 +22,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.test.runTest import org.meshtastic.core.ble.BleWriteType import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_NOTIFY_CHARACTERISTIC +import org.meshtastic.core.ble.MeshtasticBleConstants.OTA_WRITE_CHARACTERISTIC import org.meshtastic.core.testing.FakeBleConnection import org.meshtastic.core.testing.FakeBleConnectionFactory import org.meshtastic.core.testing.FakeBleDevice @@ -39,7 +40,15 @@ class BleOtaTransportTest { private fun createTransport( scanner: FakeBleScanner = FakeBleScanner(), connection: FakeBleConnection = FakeBleConnection(), + seedOtaCharacteristics: Boolean = true, ): Triple { + if (seedOtaCharacteristics) { + // Seed at the choke point instead of every connect() site — the new + // service.requireOtaCharacteristics() validation in BleOtaTransport.connect() rejects + // services missing these, so default to present; negative tests opt out. + connection.service.addCharacteristic(OTA_NOTIFY_CHARACTERISTIC) + connection.service.addCharacteristic(OTA_WRITE_CHARACTERISTIC) + } val transport = BleOtaTransport( scanner = scanner, @@ -118,6 +127,42 @@ class BleOtaTransportTest { assertIs(result.exceptionOrNull()) } + @Test + fun `connect fails when OTA characteristics are missing`() = runTest { + val (transport, scanner) = createTransport(seedOtaCharacteristics = false) + + scanner.emitDevice(FakeBleDevice(address)) + + val result = transport.connect() + + assertTrue(result.isFailure) + val exception = result.exceptionOrNull() + assertIs(exception) + val message = exception.message.orEmpty() + assertTrue(message.contains("OTA service")) + assertTrue(message.contains("TX notify characteristic")) + assertTrue(message.contains("OTA write characteristic")) + } + + @Test + fun `connect fails when notification observation fails before subscription`() = runTest { + val scanner = FakeBleScanner() + val connection = FakeBleConnection() + val failure = IllegalStateException("observe failed before CCCD") + connection.service.observeBeforeSubscriptionExceptionByCharacteristic[OTA_NOTIFY_CHARACTERISTIC] = failure + val (transport) = createTransport(scanner, connection) + + scanner.emitDevice(FakeBleDevice(address)) + + val result = transport.connect() + + assertTrue(result.isFailure) + val exception = result.exceptionOrNull() + assertIs(exception) + val cause = assertIs(exception.cause) + assertEquals(failure.message, cause.message) + } + // ----------------------------------------------------------------------- // startOta() // ----------------------------------------------------------------------- @@ -249,6 +294,34 @@ class BleOtaTransportTest { assertEquals(1.0f, progressValues.last()) } + @Test + fun `streamFirmware reuses discovered OTA service for chunk writes`() = runTest { + val scanner = FakeBleScanner() + val connection = FakeBleConnection() + val (transport) = createTransport(scanner, connection) + connection.maxWriteValueLength = 200 + scanner.emitDevice(FakeBleDevice(address)) + transport.connect().getOrThrow() + val profileCallsAfterConnect = connection.profileCalls + + emitResponse(connection, "OK") + transport.startOta(600L, "hash").getOrThrow() + + emitResponse(connection, "ACK") + emitResponse(connection, "ACK") + emitResponse(connection, "OK") + + val result = transport.streamFirmware(ByteArray(600) { it.toByte() }, 512) {} + + assertTrue(result.isSuccess, "streamFirmware failed: ${result.exceptionOrNull()}") + assertEquals(1, profileCallsAfterConnect, "connect should discover the OTA profile once") + assertEquals( + profileCallsAfterConnect, + connection.profileCalls, + "writes should reuse the discovered OTA service", + ) + } + @Test fun `streamFirmware fails on connection lost`() = runTest { val scanner = FakeBleScanner()