mirror/Meshtastic-Android
1
0
Fork 0
mirror of https://github.com/meshtastic/Meshtastic-Android.git synced 2026-07-08 04:15:37 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix(runtime): harden BLE profile, OTA setup, and DB access (#6126)

Browse Source
This commit is contained in:
Jeremiah K
2026-07-07 14:58:43 -05:00
committed by GitHub
parent 80bed50063
commit e12c4a579d
7 changed files with 306 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/BleConnection.kt
View File

@@ -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 <T> profile(
serviceUuid: Uuid,
timeout: Duration = 30.seconds,

35
core/ble/src/commonMain/kotlin/org/meshtastic/core/ble/KableBleConnection.kt
View File

@@ -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 <T> 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()

69
core/database/src/commonMain/kotlin/org/meshtastic/core/database/DatabaseManager.kt
View File

@@ -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 <T> withDb(block: suspend (MeshtasticDatabase) -> T): T? = withContext(limitedIo) {
val db = _currentDb.value ?: return@withContext null
override suspend fun <T> 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 <T> 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 <T> 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<Throwable>(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")
}

4
core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeBle.kt
View File

@@ -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")
}

177
feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransport.kt
View File

@@ -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<String>(Channel.UNLIMITED)
@Volatile private var otaService: BleService? = null
@Volatile private var responseChannel = Channel<String>(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<Unit> = safeCatching {
otaService = null
notificationJob?.cancel()
notificationJob = null
connectionStateJob?.cancel()
connectionStateJob = null
isConnected = false
responseChannel.close()
val connectResponseChannel = Channel<String>(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<Unit>()
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<Unit>()
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<String>()
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

7
feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/BleScanSupport.kt
View File

@@ -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 <T> Channel<T>.receiveWithin(timeout: Duration, onTimeout: () -> Throwable): T = try {
withTimeout(timeout) { receive() }
} catch (@Suppress("SwallowedException") e: TimeoutCancellationException) {
currentCoroutineContext().ensureActive()
throw onTimeout()
}

73
feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/BleOtaTransportTest.kt
View File

@@ -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<BleOtaTransport, FakeBleScanner, FakeBleConnection> {
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<OtaProtocolException.ConnectionFailed>(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<OtaProtocolException.ConnectionFailed>(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<OtaProtocolException.ConnectionFailed>(exception)
val cause = assertIs<IllegalStateException>(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()