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

fix(dfu): harden Legacy nRF52 recovery after BLE stalls (#6201)

Browse Source
This commit is contained in:
Jeremiah K
2026-07-10 15:02:54 -05:00
committed by GitHub
parent 0476dee1e7
commit 3c8683ea8b
7 changed files with 1516 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuUploadTransport.kt
View File

@@ -43,7 +43,10 @@ interface DfuUploadTransport {
*/
suspend fun transferFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit): Result<Unit>
/** Best-effort abort of any in-flight transfer (for cancellation / error recovery). Never throws. */
/**
* Best-effort abort. Operational transport exceptions are swallowed; structured-concurrency cancellation and Error
* subtypes propagate.
*/
suspend fun abort()
/** Disconnect and release resources. */

67
feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuProtocol.kt
View File

@@ -166,6 +166,43 @@ internal fun legacyPrnRequestPayload(packets: Int): ByteArray = byteArrayOf(
((packets ushr 8) and 0xFF).toByte(),
)
// ---------------------------------------------------------------------------
// Stream profiles
// ---------------------------------------------------------------------------
/**
* Default packet-receipt-notification interval (packets between flow-control ACKs). Higher values mean fewer
* notification round-trips per byte and therefore faster throughput, at the cost of a slightly longer recovery window
* if a packet is dropped.
*
* Capped at 10 to match Nordic's own Legacy DFU implementation, which force-limits legacy PRN to ≤10 with the comment:
* "DFU bootloaders from SDK 6.0.0 or older were unable to save incoming data to flash as fast as they are being sent …
* PRN = 10 may be the highest supported value" and treats status 6 (OPERATION_FAILED) as "data sent too fast — reduce
* PRN to 10 or less." The stock Adafruit bootloader shares this SDK11 flash-write path, so a higher value risks
* OPERATION_FAILED mid-stream on stock bootloaders. 10 is the safe ceiling that still batches flow-control ACKs.
*/
internal const val PRN_INTERVAL_PACKETS = 10
/**
* Tighter PRN interval used only on a recovery upload that follows a mid-stream disconnect. Halving the interval
* reduces the burst depth between flow-control checkpoints. Used exclusively by [LegacyDfuStreamProfile.RECOVERY] as a
* recovery heuristic; it does not by itself prevent a bootloader-side stall.
*/
internal const val RECOVERY_PRN_INTERVAL_PACKETS = 5
/**
* Legacy upload profile. Selects the PRN interval used in BOTH the `PACKET_RECEIPT_NOTIF_REQ` request and the
* receipt-await threshold inside [org.meshtastic.feature.firmware.ota.dfu.LegacyDfuTransport.streamFirmware].
* - [NORMAL]: the healthy first-attempt transfer profile — fewer PRN round-trips, higher throughput.
* - [RECOVERY]: selected after a [LegacyDfuException.MidStreamDisconnect] so subsequent attempts on the same run use
* the tighter interval. A [LegacyDfuException.StaleSessionReset] alone does NOT switch the profile, since the link
* did not actually drop mid-upload.
*/
internal enum class LegacyDfuStreamProfile(val prnIntervalPackets: Int) {
NORMAL(PRN_INTERVAL_PACKETS),
RECOVERY(RECOVERY_PRN_INTERVAL_PACKETS),
}
// ---------------------------------------------------------------------------
// Exceptions
// ---------------------------------------------------------------------------
@@ -200,12 +237,36 @@ sealed class LegacyDfuException(message: String, cause: Throwable? = null) : Dfu
LegacyDfuException("Packet receipt mismatch: expected $expected bytes received, device reports $actual")
/**
* START_DFU was rejected with `INVALID_STATE` because the bootloader still holds a previous, interrupted DFU
* session. The transport has issued a RESET; retrying the whole session (fresh reconnect) will start over on the
* now-clean bootloader. Retryable — distinct from a hard [ProtocolError].
* START_DFU was rejected with INVALID_STATE because the bootloader still holds state from an interrupted session.
* The current connection is closed without attempting RESET; SecureDfuHandler performs a bounded reset-prime over a
* fresh connection before retrying.
*/
class StaleSessionReset :
LegacyDfuException(
"Bootloader rejected START with INVALID_STATE (leftover from an interrupted flash); reset and retrying.",
)
/**
* The BLE link dropped while firmware bytes were in flight. Carries the host in-flight offset (the end of the write
* the host had dispatched or was in flight at the moment of the drop) so the retry coordinator can distinguish a
* mid-stream failure (which switches subsequent Legacy uploads to [LegacyDfuStreamProfile.RECOVERY]) from a
* handshake/connect failure (which does not).
*
* [bytesSent] is the end offset of the host write currently dispatched or in flight. It may include the current
* chunk when the disconnect occurs during `service.write()`. It does NOT prove the host stack accepted the chunk,
* and it does NOT prove the bootloader received or committed it. [lastConfirmedBytes] is the authoritative
* PRN-confirmed checkpoint (the last byte the bootloader explicitly acknowledged); `-1` means no PRN was received
* before the drop. Do not treat [bytesSent] as confirmed device progress.
*
* Typed — callers must NOT parse the message to detect a mid-stream drop.
*/
class MidStreamDisconnect(
val bytesSent: Int,
val totalBytes: Int,
val connectionState: String,
val lastConfirmedBytes: Int,
) : LegacyDfuException(
"BLE link dropped mid-upload at host in-flight offset $bytesSent/$totalBytes " +
"(lastConfirmed=$lastConfirmedBytes, state=$connectionState)",
)
}

242
feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransport.kt
View File

@@ -34,11 +34,14 @@ 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.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.withTimeout
import kotlinx.coroutines.withTimeoutOrNull
import org.meshtastic.core.ble.BleConnectionFactory
import org.meshtastic.core.ble.BleConnectionState
import org.meshtastic.core.ble.BleDevice
@@ -48,10 +51,12 @@ import org.meshtastic.core.common.util.safeCatching
import org.meshtastic.feature.firmware.ota.calculateMacPlusOne
import org.meshtastic.feature.firmware.ota.scanForBleDevice
import org.meshtastic.feature.firmware.ota.withDisconnectTripwire
import kotlin.concurrent.Volatile
import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.minutes
import kotlin.time.Duration.Companion.seconds
import kotlin.time.TimeSource
/**
* Kable-based transport for the Nordic **Legacy DFU** protocol (Nordic SDK 11/12 / Adafruit `BLEDfu`).
@@ -66,21 +71,41 @@ import kotlin.time.Duration.Companion.seconds
*
* Phase-1 buttonless trigger is shared with [SecureDfuTransport] (see `triggerButtonlessDfu` there).
*/
class LegacyDfuTransport(
class LegacyDfuTransport
internal constructor(
private val scanner: BleScanner,
connectionFactory: BleConnectionFactory,
private val address: String,
dispatcher: CoroutineDispatcher,
private val streamProfile: LegacyDfuStreamProfile = LegacyDfuStreamProfile.NORMAL,
) : DfuUploadTransport {
private val transportScope = CoroutineScope(SupervisorJob() + dispatcher)
private val bleConnection = connectionFactory.create(transportScope, "Legacy DFU")
/**
* Stream progress captured by the disconnect-tripwire callback. The disconnect watcher runs in a child coroutine on
* the caller's dispatcher, which is IO in production but may differ in tests. These fields provide visible
* diagnostic snapshots between the streaming and watcher coroutines. A fresh transport is created per upload
* attempt, so these never leak across sessions.
*/
@Volatile private var streamOffset: Int = 0
@Volatile private var streamLastPrnOffset: Int = -1
@Volatile private var streamLastPrnLatencyMs: Long = -1
/** Receives parsed responses from the Control Point characteristic. */
private val notificationChannel = Channel<LegacyDfuResponse>(Channel.UNLIMITED)
/** Name advertised by the device in DFU mode (e.g. `4631_DFU`). Captured in [connectToDfuMode]. */
private var dfuAdvertisedName: String? = null
/**
* DFU Version reported by the optional DFU Version characteristic during [connectToDfuMode]. `-1` when the
* characteristic is absent or unreadable. Captured here so the stream-start log can include it without re-reading.
*/
private var dfuVersion: Int = -1
// ---------------------------------------------------------------------------
// Phase 2: Connect to device in DFU mode
// ---------------------------------------------------------------------------
@@ -144,6 +169,7 @@ class LegacyDfuTransport(
if (bytes.size >= 2) (bytes[0].toInt() and 0xFF) or ((bytes[1].toInt() and 0xFF) shl 8) else -1
}
.getOrElse { -1 }
dfuVersion = version
Logger.i { "Legacy DFU: DFU Version characteristic = $version (-1 ⇒ absent / unreadable)" }
if (version in 1..MIN_SUPPORTED_DFU_VERSION - 1) {
throw LegacyDfuException.UnsupportedBootloader(version)
@@ -204,7 +230,9 @@ class LegacyDfuTransport(
* 7. Stream firmware in MTU-sized chunks. Every PRN packets, await `PacketReceipt(bytesReceived)` and verify count.
* 8. After last byte, await final response for `RECEIVE_FIRMWARE_IMAGE`.
* 9. `VALIDATE`, await response.
* 10. `ACTIVATE_AND_RESET` — device reboots; write may fail with disconnect, treat as success.
* 10. `ACTIVATE_AND_RESET` — an ordinary disconnect/write Exception may occur because the device resets before the
* acknowledgement; treat that operational Exception as expected success (structured cancellation and Error
* subtypes still propagate).
*/
@Suppress("LongMethod")
override suspend fun transferFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit): Result<Unit> =
@@ -232,7 +260,7 @@ class LegacyDfuTransport(
Logger.i { "Legacy DFU: requestHighConnectionPriority -> $highPriorityRequested" }
// ── 3. PRN setup ────────────────────────────────────────────────────
writeControlPoint(legacyPrnRequestPayload(PRN_INTERVAL_PACKETS))
writeControlPoint(legacyPrnRequestPayload(streamProfile.prnIntervalPackets))
// ── 4. RECEIVE_FIRMWARE_IMAGE ──────────────────────────────────────
writeControlPoint(byteArrayOf(LegacyDfuOpcode.RECEIVE_FIRMWARE_IMAGE))
@@ -248,24 +276,17 @@ class LegacyDfuTransport(
requireSuccess(LegacyDfuOpcode.VALIDATE, awaitResponse(VALIDATE_TIMEOUT))
// ── 8. ACTIVATE_AND_RESET ──────────────────────────────────────────
// Device may reset before the GATT write ACK lands — treat any write failure / disconnect as success.
// The device may reset before the GATT write ACK lands; an ordinary disconnect/write Exception is expected
// because of that reset — safeCatching treats it as success. Structured cancellation and Error subtypes
// still propagate.
Logger.i { "Legacy DFU: Sending ACTIVATE_AND_RESET (disconnect during write is expected)" }
runCatching { writeControlPoint(byteArrayOf(LegacyDfuOpcode.ACTIVATE_AND_RESET)) }
safeCatching { writeControlPoint(byteArrayOf(LegacyDfuOpcode.ACTIVATE_AND_RESET)) }
.onFailure { Logger.i(it) { "Legacy DFU: ACTIVATE write reported failure (expected on reset)" } }
onProgress(1f)
Logger.i { "Legacy DFU: Upload complete, device rebooting into new firmware." }
}
/**
* Stream [firmware] to the Packet characteristic, awaiting a [LegacyDfuResponse.PacketReceipt] every
* [PRN_INTERVAL_PACKETS] packets and verifying the bytes-received count.
*
* Watches the connection state in parallel with the write loop; if the link drops mid-stream we cancel the write
* coroutine and surface a [DfuException.ConnectionFailed] immediately rather than waiting indefinitely for a write
* that will never complete.
*/
/**
* Low-speed when the bootloader did not negotiate a larger MTU, leaving us on the 20-byte packet floor. Valid once
* [connectToDfuMode] has established the connection (the MTU is known by then).
@@ -291,66 +312,113 @@ class LegacyDfuTransport(
return sized - (sized % DFU_PACKET_WORD_ALIGNMENT)
}
/**
* Stream [firmware] to the Packet characteristic under a single outer disconnect tripwire. The outer watcher is the
* sole streaming disconnect classifier — PRN waits inside the loop intentionally do NOT install another tripwire,
* so a link drop during a PRN wait surfaces as a typed [LegacyDfuException.MidStreamDisconnect] carrying host-side
* and confirmed-progress diagnostics, not a generic handshake-style [DfuException.ConnectionFailed].
*
* Every [streamProfile.prnIntervalPackets] packets the loop awaits a [LegacyDfuResponse.PacketReceipt] and verifies
* the bootloader's bytes-received count. The [streamOffset], [streamLastPrnOffset], and [streamLastPrnLatencyMs]
* snapshots give the watcher's onDrop callback visible diagnostic values.
*/
@Suppress("CyclomaticComplexMethod", "NestedBlockDepth", "LongMethod")
private suspend fun streamFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit) {
// Packet size = negotiated ATT MTU − 3, word-aligned and capped at 244 (see computeStreamPacketSize). Falls
// back to 20 bytes when the bootloader did not negotiate a larger MTU, which is the self-gating safety against
// bootloaders that can't accept large DFU writes — the 20-byte path is the slow but universally-safe default.
val mtu = computeStreamPacketSize()
var offset = 0
val prnInterval = streamProfile.prnIntervalPackets
// Reset stream progress for this attempt. These are @Volatile instance properties so the disconnect-tripwire
// callback (running in a child coroutine on the caller's dispatcher) has a visible snapshot.
streamOffset = 0
streamLastPrnOffset = -1
streamLastPrnLatencyMs = -1
Logger.i {
"Legacy DFU: Streaming ${firmware.size} bytes with packet size $mtu " +
"(advertised='${dfuAdvertisedName ?: "?"}')"
"(advertised='${dfuAdvertisedName ?: "?"}', dfuVersion=$dfuVersion, " +
"profile=$streamProfile, prnInterval=$prnInterval)"
}
bleConnection.withDisconnectTripwire(
onDrop = { state ->
Logger.w { "Legacy DFU: Link dropped mid-stream at offset $offset/${firmware.size} (state=$state)" }
DfuException.ConnectionFailed("BLE link dropped mid-upload at byte $offset/${firmware.size}")
Logger.w {
"Legacy DFU: Link dropped mid-stream at host in-flight offset $streamOffset/${firmware.size} " +
"(state=$state, lastConfirmedPrn=$streamLastPrnOffset, lastPrnLatencyMs=$streamLastPrnLatencyMs)"
}
LegacyDfuException.MidStreamDisconnect(
bytesSent = streamOffset,
totalBytes = firmware.size,
connectionState = state.toString(),
lastConfirmedBytes = streamLastPrnOffset,
)
},
) {
var packetsSincePrn = 0
var bytesAtLastPrn = 0L
bleConnection.profile(LegacyDfuUuids.SERVICE, timeout = STREAM_TIMEOUT) { service ->
val packetChar = service.characteristic(LEGACY_DFU_PACKET_UUID)
while (offset < firmware.size) {
val end = minOf(offset + mtu, firmware.size)
while (streamOffset < firmware.size) {
val end = minOf(streamOffset + mtu, firmware.size)
// Publish the in-flight boundary BEFORE invoking service.write() so the disconnect watcher
// observes the current attempted chunk when a disconnect fires synchronously inside the write
// (e.g. a test harness or a real bootloader that drops on receive). The write may not complete
// and the host stack may not accept it; this boundary is the dispatched boundary, not a
// confirmed one — the authoritative checkpoint is the PRN-confirmed offset (streamLastPrnOffset).
val chunk = firmware.copyOfRange(streamOffset, end)
streamOffset = end
try {
service.write(packetChar, firmware.copyOfRange(offset, end), BleWriteType.WITHOUT_RESPONSE)
service.write(packetChar, chunk, BleWriteType.WITHOUT_RESPONSE)
} catch (e: CancellationException) {
Logger.w(e) {
"Legacy DFU: Write CANCELLED at offset $offset/${firmware.size} cause=${e.cause}"
"Legacy DFU: Write CANCELLED at offset $streamOffset/${firmware.size} cause=${e.cause}"
}
throw e
} catch (@Suppress("TooGenericExceptionCaught") e: Throwable) {
Logger.w(e) { "Legacy DFU: Write FAILED at offset $offset/${firmware.size}: ${e.message}" }
Logger.w(e) {
"Legacy DFU: Write FAILED at offset $streamOffset/${firmware.size}: ${e.message}"
}
throw e
}
offset = end
packetsSincePrn++
if (packetsSincePrn >= PRN_INTERVAL_PACKETS && offset < firmware.size) {
Logger.d { "Legacy DFU: Awaiting PRN at offset $offset" }
if (packetsSincePrn >= prnInterval && streamOffset < firmware.size) {
val awaitMark = TimeSource.Monotonic.markNow()
Logger.d { "Legacy DFU: Awaiting PRN at offset $streamOffset" }
val receipt =
try {
awaitPacketReceipt()
awaitPacketReceiptDuringStream()
} catch (e: CancellationException) {
Logger.w(e) {
"Legacy DFU: awaitPacketReceipt CANCELLED at offset $offset cause=${e.cause}"
"Legacy DFU: awaitPacketReceiptDuringStream CANCELLED at offset $streamOffset " +
"cause=${e.cause}"
}
throw e
}
val expected = offset.toLong()
val latencyMs = awaitMark.elapsedNow().inWholeMilliseconds
if (latencyMs >= PRN_LATENCY_WARN_THRESHOLD_MS) {
Logger.w {
"Legacy DFU: PRN receipt latency ${latencyMs}ms at offset $streamOffset " +
"(>= ${PRN_LATENCY_WARN_THRESHOLD_MS}ms — possible bootloader backpressure or BLE scheduling delay)"
}
} else {
Logger.d { "Legacy DFU: PRN receipt at offset $streamOffset latency=${latencyMs}ms" }
}
val expected = streamOffset.toLong()
if (receipt.bytesReceived != expected) {
throw LegacyDfuException.PacketReceiptMismatch(expected, receipt.bytesReceived)
}
// Record the checkpoint only after the receipt validates — a mismatched PRN must
// not be reported as the last successful checkpoint in the link-drop log.
streamLastPrnOffset = streamOffset
streamLastPrnLatencyMs = latencyMs
bytesAtLastPrn = receipt.bytesReceived
packetsSincePrn = 0
onProgress(offset.toFloat() / firmware.size)
onProgress(streamOffset.toFloat() / firmware.size)
}
}
}
Logger.d { "Legacy DFU: Streamed $offset/${firmware.size} bytes (lastPRN=$bytesAtLastPrn)" }
Logger.d { "Legacy DFU: Streamed $streamOffset/${firmware.size} bytes (lastPRN=$bytesAtLastPrn)" }
}
}
@@ -361,13 +429,33 @@ class LegacyDfuTransport(
/**
* Send `RESET` to the device, instructing it to discard any in-progress transfer and reboot. Best-effort — the
* device may disconnect before the write ACK lands; that's expected.
*
* The RESET op remains a `WITH_RESPONSE` write (the bootloader is contractually allowed to act on it before the
* GATT ACK lands, but we still request one so the link-layer queues it reliably), but we bound the wait with a
* short timeout. A missing acknowledgement is ambiguous: the device may have accepted RESET and rebooted, become
* unresponsive, disconnected, or simply failed to receive or complete the write. We report the outcome
* (acknowledged, unacknowledged, or operational failure) without claiming the RESET landed. Operational Exceptions
* are best-effort and non-fatal; structured-concurrency cancellation and Error subtypes propagate. The caller
* (`SecureDfuHandler`) tears the connection down afterwards regardless.
*
* Parent cancellation is preserved: a [CancellationException] that escapes `withTimeoutOrNull` is propagated.
*/
override suspend fun abort() {
safeCatching {
writeControlPoint(byteArrayOf(LegacyDfuOpcode.RESET))
Logger.i { "Legacy DFU: RESET sent." }
val write =
try {
withTimeoutOrNull(RESET_WRITE_TIMEOUT) { writeControlPoint(byteArrayOf(LegacyDfuOpcode.RESET)) }
} catch (e: CancellationException) {
Logger.w(e) { "Legacy DFU: RESET write cancelled; disconnecting" }
throw e
} catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
Logger.w(e) { "Legacy DFU: RESET write failed; disconnecting" }
return
}
if (write != null) {
Logger.i { "Legacy DFU: RESET write acknowledged; disconnecting" }
} else {
Logger.w { "Legacy DFU: RESET write unacknowledged within the timeout; disconnecting" }
}
.onFailure { Logger.w(it) { "Legacy DFU: Failed to send RESET (device may already be disconnected)" } }
}
override suspend fun close() {
@@ -408,12 +496,11 @@ class LegacyDfuTransport(
private suspend fun awaitResponse(timeout: Duration): LegacyDfuResponse = try {
withTimeout(timeout) {
// Fail fast + accurately if the bootloader drops the link mid-handshake instead of answering. The stock
// same-MAC nRF Legacy bootloader (e.g. AdaDFU) commonly disconnects on the first Control Point command
// (stale-bond encryption mismatch); without this the receive() below just blocks until `timeout`, so
// the
// user waited the full 30s and saw a misleading "No response from Control Point" for what was really an
// immediate disconnect.
// Fail fast + accurately if the bootloader drops the link mid-handshake instead of answering.
// Some Legacy bootloader variants disconnect on the first Control Point command
// (e.g. stale-bond encryption mismatch); without this the receive() below just blocks until
// `timeout`, so the user waited for the full command timeout and saw a misleading
// "No response from Control Point" for what was really an immediate disconnect.
bleConnection.withDisconnectTripwire(onDrop = ::handshakeDropError) {
// Drain any stray PRNs that arrive before the response we want.
while (true) {
@@ -425,32 +512,43 @@ class LegacyDfuTransport(
}
}
} catch (_: TimeoutCancellationException) {
currentCoroutineContext().ensureActive()
throw DfuException.Timeout("No response from Legacy DFU Control Point after $timeout")
}
private suspend fun awaitPacketReceipt(): LegacyDfuResponse.PacketReceipt = try {
/**
* PRN wait used inside [streamFirmware]. Deliberately does NOT install its own [withDisconnectTripwire]:
* [streamFirmware] owns the sole disconnect watcher during PRN waits, and a nested tripwire here would race it and
* could surface a generic [DfuException.ConnectionFailed] from [handshakeDropError] instead of the typed
* [LegacyDfuException.MidStreamDisconnect] that drives the recovery profile. The handshake-level disconnect watcher
* lives in [awaitResponse] (its [withDisconnectTripwire] classifies link drops during Control Point responses).
*
* A [TimeoutCancellationException] may be raised by an outer [withTimeout] (structured cancellation) rather than
* the local 30 s missing-PRN timeout. We re-check [ensureActive] first so parent/outer cancellation is rethrown,
* and only then surface the local [DfuException.Timeout].
*/
private suspend fun awaitPacketReceiptDuringStream(): LegacyDfuResponse.PacketReceipt = try {
withTimeout(COMMAND_TIMEOUT) {
bleConnection.withDisconnectTripwire(onDrop = ::handshakeDropError) {
while (true) {
val r = notificationChannel.receive()
if (r is LegacyDfuResponse.PacketReceipt) return@withDisconnectTripwire r
if (r is LegacyDfuResponse.Failure) {
throw LegacyDfuException.ProtocolError(r.requestOpcode, r.status)
}
// Stray Success or Unknown → ignore.
while (true) {
val r = notificationChannel.receive()
if (r is LegacyDfuResponse.PacketReceipt) return@withTimeout r
if (r is LegacyDfuResponse.Failure) {
throw LegacyDfuException.ProtocolError(r.requestOpcode, r.status)
}
@Suppress("UNREACHABLE_CODE")
error("unreachable")
// Stray Success or Unknown → ignore.
}
@Suppress("UNREACHABLE_CODE")
error("unreachable")
}
} catch (_: TimeoutCancellationException) {
currentCoroutineContext().ensureActive()
throw DfuException.Timeout("No packet receipt notification after $COMMAND_TIMEOUT")
}
/** Error for a link drop while awaiting a Control Point response — distinguishes a disconnect from a true stall. */
private fun handshakeDropError(state: BleConnectionState): Throwable = DfuException.ConnectionFailed(
"BLE link dropped during DFU handshake (state=$state). The device disconnected before answering a DFU " +
"command — most often the stock Adafruit bootloader rebooting to the app. Retry, or use USB recovery.",
"BLE link dropped during DFU handshake (state=$state). The device disconnected before answering a Legacy DFU " +
"command; some bootloader variants reboot to the app or drop the link in this state.",
)
/**
@@ -460,11 +558,11 @@ class LegacyDfuTransport(
* transfer state and rejects a fresh `START_DFU` with `INVALID_STATE` until it is reset. This is the common case
* when *recovering* a device stranded in the bootloader.
*
* We do NOT try to RESET on this connection: the stock Adafruit bootloader goes unresponsive immediately after
* emitting INVALID_STATE (the link dies by supervision timeout ~5 s later), so a write here never lands. Instead we
* fast-fail with [LegacyDfuException.StaleSessionReset]; [SecureDfuHandler] then resets the bootloader over a
* *fresh* connection (which is responsive up until START) before retrying. Mirrors the intent of Nordic
* `LegacyDfuImpl.resetAndRestart()`.
* We do NOT try to RESET on this connection: some Legacy bootloader variants become unresponsive after returning
* INVALID_STATE. Skip a potentially blocking same-connection RESET and use the bounded fresh-connection reset-prime
* path instead. We fast-fail with [LegacyDfuException.StaleSessionReset]; [SecureDfuHandler] then resets the
* bootloader over a *fresh* connection (which is responsive up until START) before retrying. Mirrors the intent of
* Nordic `LegacyDfuImpl.resetAndRestart()`.
*/
private fun handleStartResponse(response: LegacyDfuResponse) {
if (response is LegacyDfuResponse.Failure && response.status == LegacyDfuStatus.INVALID_STATE) {
@@ -514,6 +612,15 @@ class LegacyDfuTransport(
private val CONNECT_TIMEOUT = 15.seconds
private val COMMAND_TIMEOUT = 30.seconds
/**
* Best-effort timeout around the Legacy `RESET` (`0x06`) Control-Point write. Legacy bootloaders may disconnect
* before the RESET write acknowledgement returns. A missing acknowledgement is ambiguous: the device may have
* accepted RESET and rebooted, become unresponsive, disconnected, or failed to receive/complete the write.
* Bound the acknowledgement wait to one second, report it as unacknowledged, then rely on connection teardown
* and the subsequent retry/re-advertisement flow.
*/
internal val RESET_WRITE_TIMEOUT = 1.seconds
/**
* Time to wait for the START_DFU response notification.
*
@@ -537,18 +644,13 @@ class LegacyDfuTransport(
private val STREAM_TIMEOUT = 15.minutes
/**
* Packet-receipt-notification interval (packets between flow-control ACKs). Higher values mean fewer
* notification round-trips per byte and therefore faster throughput, at the cost of a slightly longer recovery
* window if a packet is dropped (we have to wait until the next PRN boundary to detect the gap).
*
* Capped at 10 to match Nordic's own Legacy DFU implementation, which force-limits legacy PRN to ≤10 with the
* comment: "DFU bootloaders from SDK 6.0.0 or older were unable to save incoming data to flash as fast as they
* are being sent … PRN = 10 may be the highest supported value" and treats status 6 (OPERATION_FAILED) as "data
* sent too fast — reduce PRN to 10 or less." The stock Adafruit bootloader shares this SDK11 flash-write path,
* so a higher value (we previously used 30, tuned for the faster OTAFIX fork) risks OPERATION_FAILED mid-stream
* on stock bootloaders. 10 is the safe ceiling that still batches flow-control ACKs.
* Per-receipt latency threshold above which a PRN wait is logged at WARN. The Legacy bootloader normally ACKs
* each PRN window inside a few tens of milliseconds. A latency at or above this threshold is diagnostic only —
* it may indicate bootloader backpressure, flash activity, BLE scheduling delay, or degraded link conditions.
* It is not by itself a definite precursor to a supervision-timeout drop. Below this threshold receipts log at
* DEBUG.
*/
internal const val PRN_INTERVAL_PACKETS = 10
internal const val PRN_LATENCY_WARN_THRESHOLD_MS = 1_000L
/**
* Universally-safe Legacy DFU packet size (20 bytes — the original ATT_MTU minus the 3-byte ATT header). Used

367
feature/firmware/src/commonMain/kotlin/org/meshtastic/feature/firmware/ota/dfu/SecureDfuHandler.kt
View File

@@ -64,11 +64,42 @@ private const val DFU_REBOOT_WAIT_MS = 3_000L
private const val RETRY_DELAY_MS = 2_000L
private const val CONNECT_ATTEMPTS = 4
/** Legacy DFU can't resume, so retry the whole session this many times before giving up. */
private const val LEGACY_SESSION_ATTEMPTS = 3
/**
* Real upload attempts a confirmed Legacy primary is allowed before declaring failure. Legacy DFU has no resume, so a
* failed upload requires a whole fresh session (new transport + reconnect + re-handshake). Stale-session cleanup cycles
* do NOT consume this budget (see [MAX_LEGACY_STALE_RESETS]).
*/
internal const val LEGACY_SESSION_ATTEMPTS = 3
/**
* Maximum number of stale-session cleanup cycles a Legacy run will perform. A [LegacyDfuException.StaleSessionReset]
* surfaces when the bootloader rejects START_DFU with INVALID_STATE (a half-finished previous session is wedged in the
* bootloader). The cleanup connects a fresh transport, RESETs, and waits for the bootloader to reboot into a clean OTA
* session. This budget is separate from [LEGACY_SESSION_ATTEMPTS] so a run that legitimately needs to reset-prime more
* than once (e.g. a mid-stream drop that left the bootloader wedged twice in a row) is not punished by losing an upload
* attempt for a cleanup that did not actually try to upload.
*/
internal const val MAX_LEGACY_STALE_RESETS = 2
/** Limited probe/fallback budget for inconclusive detection or alternate-protocol attempts. */
private const val LIMITED_SESSION_ATTEMPTS = 1
internal const val LIMITED_SESSION_ATTEMPTS = 1
/**
* Two-stage session budget.
*
* [preEngagementAttempts] limits speculative connection attempts before the selected DFU service has been confirmed.
* [engagedAttempts] applies after a connection to that protocol's DFU service succeeds. Engagement means the selected
* protocol's DFU service was connected to; it does not by itself universally prevent fallback (see
* [DfuFallbackCoordinator] for the fallback policy). Secure always carries [preEngagementAttempts] == [engagedAttempts]
* so engagement never expands its budget.
*/
internal data class DfuAttemptBudget(val preEngagementAttempts: Int, val engagedAttempts: Int) {
init {
require(preEngagementAttempts in 1..engagedAttempts) {
"preEngagementAttempts ($preEngagementAttempts) must be in 1..engagedAttempts ($engagedAttempts)"
}
}
}
/**
* Transport-dispatch key: which [DfuUploadTransport] implementation ([SecureDfuTransport] or [LegacyDfuTransport]) to
@@ -105,12 +136,12 @@ internal sealed class DfuUploadResult {
}
/**
* Owns fallback ordering and retry budget policy. Fallback is intentionally limited to failures that happen before the
* selected transport has connected and engaged its DFU protocol session.
* Owns fallback ordering and retry budget policy. Whether an alternate protocol is tried after a primary failure
* depends on the [detection] outcome and engagement state — see [shouldTryAlternateAfterFailure].
*/
internal class DfuFallbackCoordinator(private val detection: BootloaderDetection) {
suspend fun execute(uploadWithRetry: suspend (DfuProtocolKind, Int) -> DfuUploadResult) {
suspend fun execute(uploadWithRetry: suspend (DfuProtocolKind, DfuAttemptBudget) -> DfuUploadResult) {
val orderedProtocols = detection.orderedProtocols()
Logger.i { "DFU: detection=$detection → protocols=$orderedProtocols" }
@@ -118,17 +149,17 @@ internal class DfuFallbackCoordinator(private val detection: BootloaderDetection
for ((index, protocol) in orderedProtocols.withIndex()) {
val isPrimary = index == 0
val hasAlternateProtocol = index < orderedProtocols.lastIndex
val sessionAttempts = sessionAttemptsFor(protocol, isPrimary)
val budget = budgetFor(protocol, isPrimary)
if (isPrimary) {
Logger.i { "DFU: primary protocol=$protocol ($sessionAttempts session attempt(s))" }
} else {
Logger.w {
"DFU: falling back to alternate protocol=$protocol before protocol engagement " +
"(detection=$detection, sessionAttempts=$sessionAttempts)"
Logger.i {
"DFU: primary protocol=$protocol " +
"(upload attempts pre=${budget.preEngagementAttempts}, engaged=${budget.engagedAttempts})"
}
} else {
Logger.w { "DFU: trying alternate protocol=$protocol (detection=$detection, budget=$budget)" }
}
when (val result = uploadWithRetry(protocol, sessionAttempts)) {
when (val result = uploadWithRetry(protocol, budget)) {
DfuUploadResult.Success -> return
is DfuUploadResult.Failure -> {
@@ -150,7 +181,7 @@ internal class DfuFallbackCoordinator(private val detection: BootloaderDetection
/**
* The ordered list of protocols to attempt for this detection outcome. The first element is the primary; the second
* is the alternate, attempted only if the primary fails before protocol engagement.
* is the alternate. Whether the alternate is reached depends on [shouldTryAlternateAfterFailure], not ordering.
*/
private fun BootloaderDetection.orderedProtocols(): List<DfuProtocolKind> = when (this) {
BootloaderDetection.LegacyObserved -> listOf(DfuProtocolKind.LEGACY, DfuProtocolKind.SECURE)
@@ -159,29 +190,45 @@ internal class DfuFallbackCoordinator(private val detection: BootloaderDetection
}
/**
* Session retry budget for the given protocol in this detection context.
* - Confirmed Legacy primary ([BootloaderDetection.LegacyObserved]): [LEGACY_SESSION_ATTEMPTS] (3) to cover the
* stale-session reset-prime recovery path.
* - Unknown Legacy primary and alternate protocols: [LIMITED_SESSION_ATTEMPTS] (1) to keep speculative probes short
* enough that a Secure bootloader still has time to accept its fallback attempt.
* Two-stage upload-attempt budget for the given protocol in this detection context.
* - Confirmed Legacy primary ([BootloaderDetection.LegacyObserved]): full [LEGACY_SESSION_ATTEMPTS] (3) budget
* before AND after engagement — the protocol was conclusively observed.
* - Unknown Legacy primary: [LIMITED_SESSION_ATTEMPTS] (1) pre-engagement probe so Secure fallback stays timely,
* promoted to [LEGACY_SESSION_ATTEMPTS] (3) once Legacy engages and is conclusively identified.
* - Legacy alternate (e.g. SecureObserved → Legacy fallback): same Unknown-style two-stage budget — pre-engagement
* probe of 1, full Legacy budget of 3 once engaged.
* - Secure: 1/1 — never expanded by engagement, since the Secure recovery path is single-shot.
*/
private fun sessionAttemptsFor(protocol: DfuProtocolKind, isPrimary: Boolean): Int = when {
protocol == DfuProtocolKind.LEGACY && isPrimary && detection == BootloaderDetection.LegacyObserved ->
LEGACY_SESSION_ATTEMPTS
private fun budgetFor(protocol: DfuProtocolKind, isPrimary: Boolean): DfuAttemptBudget = when {
protocol == DfuProtocolKind.SECURE ->
DfuAttemptBudget(
preEngagementAttempts = LIMITED_SESSION_ATTEMPTS,
engagedAttempts = LIMITED_SESSION_ATTEMPTS,
)
else -> LIMITED_SESSION_ATTEMPTS
protocol == DfuProtocolKind.LEGACY && isPrimary && detection == BootloaderDetection.LegacyObserved ->
DfuAttemptBudget(
preEngagementAttempts = LEGACY_SESSION_ATTEMPTS,
engagedAttempts = LEGACY_SESSION_ATTEMPTS,
)
// Unknown Legacy primary and Legacy alternate: speculative probe that promotes on engagement.
else ->
DfuAttemptBudget(
preEngagementAttempts = LIMITED_SESSION_ATTEMPTS,
engagedAttempts = LEGACY_SESSION_ATTEMPTS,
)
}
private fun shouldTryAlternateAfterFailure(isPrimary: Boolean, protocolEngaged: Boolean): Boolean {
if (!isPrimary) return false
return when (detection) {
// Unknown can mean the primary Legacy advertisement was missed, so a pre-engagement failure is the useful
// fallback signal.
// Unknown detection: Legacy is tried first, Secure fallback only if Legacy fails before engagement.
BootloaderDetection.Unknown -> !protocolEngaged
// Conclusive detections already observed the selected protocol's service. If it never engages, retrying the
// opposite service is usually a long doomed scan. Keep fallback only as insurance for a stale/wrong
// conclusive signal after the observed service was actually reached.
// Conclusive detection: a pre-engagement failure of the observed protocol does NOT immediately try the
// opposite protocol. Alternate fallback is retained only after engagement failure — insurance against a
// stale or misleading conclusive detection.
BootloaderDetection.LegacyObserved,
BootloaderDetection.SecureObserved,
-> protocolEngaged
@@ -203,6 +250,183 @@ private fun DfuProtocolKind.serviceUuid(): Uuid = when (this) {
DfuProtocolKind.SECURE -> SecureDfuUuids.SERVICE
}
/**
* Drives the bounded Legacy/Secure DFU upload retry loop. Extracted from [SecureDfuHandler.runDfuUploadWithRetry] so
* the retry policy can be unit-tested without bringing up the full BLE stack — callers supply a [runUploadSession]
* lambda that returns the next [DfuUploadResult] and a [resetStaleBootloader] lambda that performs the fresh-connection
* RESET-prime cycle.
*
* Policy:
* - Active attempt limit starts at [DfuAttemptBudget.preEngagementAttempts]; the first [DfuUploadResult.Failure] whose
* `protocolEngaged=true` promotes the active limit to [DfuAttemptBudget.engagedAttempts] BEFORE the budget-exhaustion
* check runs, so an engaged Legacy probe receives its full retry budget.
* - Non-[LegacyDfuException.StaleSessionReset] failures consume one upload attempt.
* - A [LegacyDfuException.StaleSessionReset] consumes one of [maxStaleResets], NOT an upload attempt — the cleanup
* cycle never tried to upload.
* - After the first [LegacyDfuException.MidStreamDisconnect] on Legacy, every subsequent Legacy attempt uses
* [LegacyDfuStreamProfile.RECOVERY]; a stale-session alone never switches the profile.
* - All exit paths are bounded: the loop stops when the upload-attempt budget is exhausted or the stale-response budget
* is exceeded. Up to [maxStaleResets] reset-prime cleanups run; the next stale response terminates without another
* cleanup.
* - [DfuProtocolKind.SECURE] carries pre==engaged budgets, so engagement promotion is a no-op there.
*
* If a mid-stream drop preceded the terminal failure, the first [LegacyDfuException.MidStreamDisconnect] is attached as
* a suppressed exception on the surfaced error so the underlying cause stays visible.
*/
@Suppress("CyclomaticComplexMethod", "NestedBlockDepth", "LongMethod")
internal suspend fun runDfuRetryLoop(
protocol: DfuProtocolKind,
budget: DfuAttemptBudget,
maxStaleResets: Int,
runUploadSession: suspend (LegacyDfuStreamProfile) -> DfuUploadResult,
resetStaleBootloader: suspend () -> Unit,
interAttemptDelay: suspend () -> Unit,
): DfuUploadResult {
var uploadAttempts = 0
var staleResets = 0
var midStreamSeen = false
var firstMidStreamError: LegacyDfuException.MidStreamDisconnect? = null
var lastError: Throwable? = null
var protocolEngaged = false
var activeAttempts = budget.preEngagementAttempts
while (uploadAttempts < activeAttempts) {
var skipInterAttemptDelay = false
val profile =
if (protocol == DfuProtocolKind.LEGACY && midStreamSeen) {
LegacyDfuStreamProfile.RECOVERY
} else {
LegacyDfuStreamProfile.NORMAL
}
uploadAttempts++
Logger.i { "DFU: upload attempt $uploadAttempts/$activeAttempts ($protocol, profile=$profile)" }
when (val result = runUploadSession(profile)) {
DfuUploadResult.Success -> return result
is DfuUploadResult.Failure -> {
lastError = result.error
val wasEngaged = protocolEngaged
protocolEngaged = protocolEngaged || result.protocolEngaged
// Promote the upload budget the moment a session first reports protocolEngaged. Promotion happens
// BEFORE the budget-exhaustion check (the while condition on the next iteration) so an engaged
// Legacy probe receives its full retry budget. For Secure (pre==engaged) this is a no-op.
if (!wasEngaged && protocolEngaged && activeAttempts < budget.engagedAttempts) {
activeAttempts = budget.engagedAttempts
Logger.i {
"DFU: protocol engaged; promoting upload budget " +
"(${budget.preEngagementAttempts} → ${budget.engagedAttempts}) ($protocol)"
}
}
// Legacy-only stale-session handling. StaleSessionReset consumes a stale-reset cycle, NOT an upload
// attempt — the cycle never tried to upload.
if (protocol == DfuProtocolKind.LEGACY && result.error is LegacyDfuException.StaleSessionReset) {
uploadAttempts--
staleResets++
if (staleResets > maxStaleResets) {
Logger.w {
"DFU: stale cleanup exhausted ($staleResets > $maxStaleResets) — giving up on $protocol"
}
break
}
if (midStreamSeen) {
// Expected: a stale session is the natural residue of a mid-stream drop, so an informational
// log keeps the operator calm.
Logger.i {
"DFU: stale cleanup cycle $staleResets/$maxStaleResets after mid-stream drop " +
"(expected) ($protocol)"
}
} else {
// Unexpected: the very first session came up stale with no preceding drop, which usually
// means the device was stranded in the bootloader before this update began.
Logger.w {
"DFU: stale cleanup cycle $staleResets/$maxStaleResets — unexpected initial stale " +
"session ($protocol)"
}
}
resetStaleBootloader()
skipInterAttemptDelay = true
}
if (!skipInterAttemptDelay) {
// Mid-stream disconnect: switch subsequent Legacy uploads to the recovery profile and surface the
// host in-flight offset in the log so it is not hidden by later cleanup failures.
val error = result.error
if (protocol == DfuProtocolKind.LEGACY && error is LegacyDfuException.MidStreamDisconnect) {
if (!midStreamSeen) {
midStreamSeen = true
firstMidStreamError = error
}
Logger.w(error) {
"DFU: mid-stream disconnect at host in-flight offset " +
"${error.bytesSent}/${error.totalBytes} " +
"(state=${error.connectionState}); subsequent $protocol attempts will use " +
"${LegacyDfuStreamProfile.RECOVERY}"
}
} else {
Logger.w(error) {
"DFU: upload attempt $uploadAttempts/$activeAttempts failed ($protocol): " +
"${error::class.simpleName}"
}
}
if (uploadAttempts < activeAttempts) interAttemptDelay()
}
}
}
}
val error = lastError ?: DfuException.TransferFailed("DFU upload failed after $activeAttempts attempts ($protocol)")
// Keep the original mid-stream failure visible even when the terminal failure is a later cleanup/transfer error.
firstMidStreamError?.let { drop ->
if (error !== drop) {
error.addSuppressed(drop)
}
}
return DfuUploadResult.Failure(error, protocolEngaged)
}
/**
* Cleanup policy for a DFU upload session's transport. Extracted from [SecureDfuHandler.runUploadSession]'s finally
* block so the skip-abort / always-close contract can be tested directly.
*
* Semantics:
* - Executes in [NonCancellable] so close always runs even if the caller was cancelled.
* - Skips [DfuUploadTransport.abort] after a completed session — nothing to abort.
* - Skips [DfuUploadTransport.abort] after a [LegacyDfuException.StaleSessionReset] — some Legacy variants become
* unresponsive after INVALID_STATE, so the same-connection RESET may block or be ineffective; the fresh-connection
* reset-prime path owns recovery.
* - Skips [DfuUploadTransport.abort] after a [LegacyDfuException.MidStreamDisconnect] — the link has already dropped,
* so there is no live connection over which to deliver an abort.
* - Attempts [DfuUploadTransport.abort] after a [DfuException.ConnectionFailed] — [DfuException.ConnectionFailed] is
* produced by [connectWithRetry] for any exhausted connection or setup failure, including cases where a GATT
* connection was established but a subsequent step (e.g. unsupported bootloader version, profile setup) failed. A
* usable connection may still exist, so the bounded best-effort abort is attempted before close.
* - Otherwise calls [DfuUploadTransport.abort], then [DfuUploadTransport.close] in a nested finally.
* - [DfuUploadTransport.abort]'s own cancellation/Error propagation contract is preserved: the [NonCancellable] context
* prevents the outer coroutine cancellation from interrupting close, but an [Error] thrown by abort still propagates
* after close runs.
*/
internal suspend fun cleanupDfuSessionTransport(
transport: DfuUploadTransport,
completed: Boolean,
sessionFailure: Throwable?,
) {
withContext(NonCancellable) {
val skipAbort =
sessionFailure is LegacyDfuException.StaleSessionReset ||
sessionFailure is LegacyDfuException.MidStreamDisconnect
try {
if (!completed && !skipAbort) transport.abort()
} finally {
transport.close()
}
}
}
/**
* KMP [FirmwareUpdateHandler] for nRF52 devices.
*
@@ -285,10 +509,10 @@ class SecureDfuHandler(
// reconnect + re-handshake), mirroring Nordic's DFU library ("the Legacy DFU will start again"). A
// stock bootloader that leaves the first session's control-point handshake unanswered often responds
// after a clean reconnect. Secure DFU resumes in place, so it runs a single session.
// DfuFallbackCoordinator resolves the detection into an ordered protocol list and may try the
// alternate protocol only before the selected protocol has connected and engaged its DFU session.
DfuFallbackCoordinator(detection).execute { protocol, sessionAttempts ->
runDfuUploadWithRetry(protocol, target, pkg, sessionAttempts, updateState)
// DfuFallbackCoordinator resolves the detection into an ordered protocol list; whether the alternate
// is tried depends on detection-specific engagement rules (shouldTryAlternateAfterFailure).
DfuFallbackCoordinator(detection).execute { protocol, budget ->
runDfuUploadWithRetry(protocol, target, pkg, budget, updateState)
}
zipFile
}
@@ -370,59 +594,44 @@ class SecureDfuHandler(
}
/**
* Run the connect + init + firmware upload, retrying the whole session up to [attempts] times. Each attempt uses a
* fresh [DfuUploadTransport] (new GATT connection + re-handshake) since Legacy DFU can't resume mid-stream.
*
* Returns the last failure with whether this protocol ever connected and started its DFU session, allowing fallback
* orchestration to switch protocols only for pre-engagement failures.
* Run the connect + init + firmware upload, retrying within the [DfuFallbackCoordinator] budget. Delegates the
* bounded retry logic to [runDfuRetryLoop] (testable without bringing up the full BLE stack); the lambdas here
* supply the real [runUploadSession] and [resetStaleBootloader] implementations.
*/
private suspend fun runDfuUploadWithRetry(
protocol: DfuProtocolKind,
target: String,
pkg: DfuZipPackage,
attempts: Int,
budget: DfuAttemptBudget,
updateState: (FirmwareUpdateState) -> Unit,
): DfuUploadResult {
var lastError: Throwable? = null
var protocolEngaged = false
repeat(attempts) { i ->
val attempt = i + 1
Logger.i { "DFU: upload session attempt $attempt/$attempts ($protocol)" }
when (val result = runUploadSession(protocol, target, pkg, updateState)) {
DfuUploadResult.Success -> return result
): DfuUploadResult = runDfuRetryLoop(
protocol = protocol,
budget = budget,
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile -> runUploadSession(protocol, target, pkg, profile, updateState) },
resetStaleBootloader = { resetStaleBootloader(protocol, target) },
interAttemptDelay = { delay(SESSION_RETRY_DELAY_MS) },
)
is DfuUploadResult.Failure -> {
lastError = result.error
protocolEngaged = protocolEngaged || result.protocolEngaged
Logger.w(result.error) {
"DFU: upload session $attempt/$attempts failed ($protocol): ${result.error::class.simpleName}"
}
// A stock bootloader holding a wedged session from an interrupted flash rejects START with
// INVALID_STATE, then goes unresponsive (the RESET can't land on that connection). A *fresh*
// connection is responsive up until START, so reset it there — the device reboots into a clean OTA
// session (GPREGRET OTA flag is retained) that the next attempt can flash normally.
if (result.error is LegacyDfuException.StaleSessionReset && attempt < attempts) {
resetStaleBootloader(protocol, target)
}
if (attempt < attempts) delay(SESSION_RETRY_DELAY_MS)
}
}
}
return DfuUploadResult.Failure(
lastError ?: DfuException.TransferFailed("DFU upload failed after $attempts attempts ($protocol)"),
protocolEngaged,
)
}
private fun createTransport(
protocol: DfuProtocolKind,
target: String,
legacyProfile: LegacyDfuStreamProfile = LegacyDfuStreamProfile.NORMAL,
): DfuUploadTransport = when (protocol) {
// Legacy transport receives the selected stream profile; default NORMAL keeps prior behavior for any caller
// that does not pass an explicit profile (e.g. reset-prime, which is a control-point operation only).
DfuProtocolKind.LEGACY ->
LegacyDfuTransport(bleScanner, bleConnectionFactory, target, dispatchers.default, legacyProfile)
private fun createTransport(protocol: DfuProtocolKind, target: String): DfuUploadTransport = when (protocol) {
DfuProtocolKind.LEGACY -> LegacyDfuTransport(bleScanner, bleConnectionFactory, target, dispatchers.default)
DfuProtocolKind.SECURE -> SecureDfuTransport(bleScanner, bleConnectionFactory, target, dispatchers.default)
}
/**
* Reboot a bootloader wedged in a stale DFU session. Connects a fresh transport (which is responsive before any
* START) and issues RESET (0x06) via [DfuUploadTransport.abort], then waits for the reboot + re-advertise. Best
* effort: any failure here is non-fatal — the caller retries the upload regardless.
* START) and issues RESET (0x06) via the bounded [DfuUploadTransport.abort], then closes the connection and waits
* for the configured [RESET_PRIME_REBOOT_WAIT_MS] reboot/re-advertisement interval. Operational [Exception]s during
* abort are best-effort (swallowed by [DfuUploadTransport.abort]); structured-concurrency cancellation and [Error]
* subtypes propagate per the abort contract.
*/
private suspend fun resetStaleBootloader(protocol: DfuProtocolKind, target: String) {
Logger.i { "DFU: reset-priming stale bootloader before retry" }
@@ -432,7 +641,7 @@ class SecureDfuHandler(
.connectToDfuMode()
.onSuccess {
transport.abort()
Logger.i { "DFU: reset-prime RESET sent; waiting for clean reboot" }
Logger.i { "DFU: reset-prime RESET attempted; disconnecting and waiting for re-advertisement" }
}
.onFailure { Logger.w(it) { "DFU: reset-prime connect failed: ${it.message}" } }
} finally {
@@ -446,11 +655,13 @@ class SecureDfuHandler(
protocol: DfuProtocolKind,
target: String,
pkg: DfuZipPackage,
legacyProfile: LegacyDfuStreamProfile,
updateState: (FirmwareUpdateState) -> Unit,
): DfuUploadResult {
val transport: DfuUploadTransport = createTransport(protocol, target)
val transport: DfuUploadTransport = createTransport(protocol, target, legacyProfile)
var completed = false
var protocolEngaged = false
var sessionFailure: Throwable? = null
try {
connectWithRetry(transport, protocol, updateState)
protocolEngaged = true
@@ -495,13 +706,11 @@ class SecureDfuHandler(
return DfuUploadResult.Success
} catch (e: CancellationException) {
throw e
} catch (@Suppress("TooGenericExceptionCaught") e: Throwable) {
} catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
sessionFailure = e
return DfuUploadResult.Failure(e, protocolEngaged)
} finally {
withContext(NonCancellable) {
if (!completed) transport.abort()
transport.close()
}
cleanupDfuSessionTransport(transport, completed, sessionFailure)
}
}

50
feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/DfuFallbackCoordinatorTest.kt
View File

@@ -63,11 +63,11 @@ class DfuFallbackCoordinatorTest {
fun `Unknown falls back from Legacy to Secure on pre-engagement failure`() = runTest {
val coordinator = DfuFallbackCoordinator(BootloaderDetection.Unknown)
val protocols = mutableListOf<DfuProtocolKind>()
val attemptCounts = mutableListOf<Int>()
val budgets = mutableListOf<DfuAttemptBudget>()
assertFailsWith<RuntimeException> {
coordinator.execute { protocol, attempts ->
coordinator.execute { protocol, budget ->
protocols.add(protocol)
attemptCounts.add(attempts)
budgets.add(budget)
if (protocol == DfuProtocolKind.LEGACY) {
DfuUploadResult.Failure(RuntimeException("connect failed"), protocolEngaged = false)
} else {
@@ -81,7 +81,9 @@ class DfuFallbackCoordinatorTest {
assertEquals("also failed", thrown.suppressedExceptions.first().message)
}
assertEquals(listOf(DfuProtocolKind.LEGACY, DfuProtocolKind.SECURE), protocols)
assertEquals(listOf(1, 1), attemptCounts)
// Unknown Legacy primary budget shape: pre=1 probe, engaged=3; promotion is not exercised because this test
// fails before engagement. Secure remains 1/1.
assertEquals(listOf(DfuAttemptBudget(1, 3), DfuAttemptBudget(1, 1)), budgets)
}
@Test
@@ -98,14 +100,17 @@ class DfuFallbackCoordinatorTest {
}
@Test
fun `LegacyObserved gives Legacy 3 session attempts`() = runTest {
fun `LegacyObserved gives Legacy 3 upload attempts`() = runTest {
val coordinator = DfuFallbackCoordinator(BootloaderDetection.LegacyObserved)
val attemptCounts = mutableListOf<Int>()
coordinator.execute { _, attempts ->
attemptCounts.add(attempts)
val budgets = mutableListOf<DfuAttemptBudget>()
coordinator.execute { _, budget ->
budgets.add(budget)
DfuUploadResult.Success
}
assertEquals(listOf(3), attemptCounts) // LEGACY_SESSION_ATTEMPTS
assertEquals(
listOf(DfuAttemptBudget(3, 3)),
budgets,
) // LEGACY_SESSION_ATTEMPTS — upload-attempt budget; stale cleanup is separate
}
@Test
@@ -123,11 +128,11 @@ class DfuFallbackCoordinatorTest {
}
@Test
fun `LegacyObserved gives Secure fallback 1 session attempt after Legacy engages`() = runTest {
fun `LegacyObserved gives Secure fallback budget of 1 upload attempt after Legacy engages`() = runTest {
val coordinator = DfuFallbackCoordinator(BootloaderDetection.LegacyObserved)
val attemptCounts = mutableListOf<Int>()
coordinator.execute { protocol, attempts ->
attemptCounts.add(attempts)
val budgets = mutableListOf<DfuAttemptBudget>()
coordinator.execute { protocol, budget ->
budgets.add(budget)
if (protocol == DfuProtocolKind.LEGACY) {
DfuUploadResult.Failure(RuntimeException("legacy protocol failed"), protocolEngaged = true)
} else {
@@ -135,20 +140,23 @@ class DfuFallbackCoordinatorTest {
}
}
assertEquals(
listOf(3, 1),
attemptCounts,
) // Legacy primary=LEGACY_SESSION_ATTEMPTS, Secure fallback=LIMITED_SESSION_ATTEMPTS
listOf(DfuAttemptBudget(3, 3), DfuAttemptBudget(1, 1)),
budgets,
) // Legacy primary=LEGACY_SESSION_ATTEMPTS (upload attempts), Secure fallback=LIMITED_SESSION_ATTEMPTS
}
@Test
fun `Unknown gives Legacy primary 1 session attempt`() = runTest {
fun `Unknown gives Legacy primary a pre-engagement probe budget of 1 with engaged promotion to 3`() = runTest {
val coordinator = DfuFallbackCoordinator(BootloaderDetection.Unknown)
val attemptCounts = mutableListOf<Int>()
coordinator.execute { _, attempts ->
attemptCounts.add(attempts)
val budgets = mutableListOf<DfuAttemptBudget>()
coordinator.execute { _, budget ->
budgets.add(budget)
DfuUploadResult.Success
}
assertEquals(listOf(1), attemptCounts) // LIMITED_SESSION_ATTEMPTS keeps speculative Unknown probes bounded
assertEquals(
listOf(DfuAttemptBudget(1, 3)),
budgets,
) // LIMITED_SESSION_ATTEMPTS keeps speculative Unknown probes bounded; engagement promotes to 3
}
@Test

635
feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuRetryPolicyTest.kt Normal file
View File

@@ -0,0 +1,635 @@
/*
* Copyright (c) 2026 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
@file:Suppress("MagicNumber")
package org.meshtastic.feature.firmware.ota.dfu
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith
import kotlin.test.assertFalse
import kotlin.test.assertIs
import kotlin.test.assertTrue
/**
* Tests the Legacy DFU retry policy extracted into [runDfuRetryLoop]. The retry loop is driven entirely by lambdas — a
* session-result supplier and a stale-reset cleanup hook — so the policy can be exercised without bringing up the full
* BLE stack. Covers the drop/stale/success sequences that the mid-stream-disconnect recovery design depends on.
*/
class LegacyDfuRetryPolicyTest {
/**
* Sequence A: NORMAL upload drops mid-stream → stale cleanup → RECOVERY upload succeeds.
*
* Verifies:
* - only 1 upload attempt is consumed before success (stale cleanup does NOT consume an upload attempt)
* - RECOVERY profile is selected for the post-drop attempts
* - exactly one stale-reset cleanup cycle runs
* - the surfaced result is Success
*/
@Test
fun `drop then stale then success consumes one upload attempt and switches to RECOVERY`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
var staleResets = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
// Confirmed-Legacy style budget: full budget before and after engagement.
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
when (sessions.size) {
1 ->
DfuUploadResult.Failure(
LegacyDfuException.MidStreamDisconnect(
20,
200,
"Disconnected",
lastConfirmedBytes = -1,
),
false,
)
2 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
else -> DfuUploadResult.Success
}
},
resetStaleBootloader = { staleResets++ },
interAttemptDelay = {},
)
assertEquals(DfuUploadResult.Success, outcome)
assertEquals(
listOf(LegacyDfuStreamProfile.NORMAL, LegacyDfuStreamProfile.RECOVERY, LegacyDfuStreamProfile.RECOVERY),
sessions,
)
assertEquals(1, staleResets)
}
/**
* Sequence B: drop → stale → RECOVERY drop → stale → RECOVERY success.
*
* Verifies the run fits within 3 upload attempts + 2 stale resets (the budget), each stale-reset cycle runs exactly
* once per drop, and the final RECOVERY attempt succeeds.
*/
@Test
fun `drop stale drop stale success fits within 3 uploads and 2 stale resets`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
var staleResets = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
// Alternating: attempts 1, 3 drop mid-stream; attempts 2, 4 are stale; attempt 5 succeeds.
when (sessions.size) {
1,
3,
->
DfuUploadResult.Failure(
LegacyDfuException.MidStreamDisconnect(
20,
200,
"Disconnected",
lastConfirmedBytes = -1,
),
false,
)
2,
4,
-> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
else -> DfuUploadResult.Success
}
},
resetStaleBootloader = { staleResets++ },
interAttemptDelay = {},
)
assertEquals(DfuUploadResult.Success, outcome)
// 5 sessions total, all but the first are RECOVERY.
assertEquals(5, sessions.size)
assertEquals(LegacyDfuStreamProfile.NORMAL, sessions.first())
sessions.drop(1).forEach { assertEquals(LegacyDfuStreamProfile.RECOVERY, it) }
assertEquals(MAX_LEGACY_STALE_RESETS, staleResets)
}
/**
* Repeated stale responses must stop at MAX_LEGACY_STALE_RESETS even though none of them consume an upload attempt
* — otherwise a wedged bootloader would loop forever.
*/
@Test
fun `repeated stale responses stop at MAX_LEGACY_STALE_RESETS`() = runTest {
var sessions = 0
var staleResets = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = {
sessions++
DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
},
resetStaleBootloader = { staleResets++ },
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
assertIs<LegacyDfuException.StaleSessionReset>(outcome.error)
// The loop tolerates MAX_LEGACY_STALE_RESETS stale responses that trigger cleanup; the next stale response
// terminates the loop.
assertEquals(MAX_LEGACY_STALE_RESETS + 1, sessions)
assertEquals(MAX_LEGACY_STALE_RESETS, staleResets)
}
/**
* Ordinary failures (not MidStreamDisconnect, not StaleSessionReset) consume the upload-attempt budget. After
* `attempts` consecutive ordinary failures, the loop exits with the last failure.
*/
@Test
fun `ordinary failures consume the three upload attempts`() = runTest {
var sessions = 0
var staleResets = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = {
sessions++
DfuUploadResult.Failure(DfuException.TransferFailed("ordinary failure"), false)
},
resetStaleBootloader = { staleResets++ },
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
assertIs<DfuException.TransferFailed>(outcome.error)
assertEquals(LEGACY_SESSION_ATTEMPTS, sessions)
assertEquals(0, staleResets, "ordinary failures must NOT trigger stale cleanup")
}
/**
* A StaleSessionReset must NOT consume an upload attempt — the cleanup cycle never tried to upload. After the stale
* cleanup, the same upload-attempt budget remains, and a subsequent successful attempt succeeds.
*/
@Test
fun `stale cleanup does not consume an upload attempt`() = runTest {
var sessions = 0
var staleResets = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = {
sessions++
if (sessions == 1) {
DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
} else {
DfuUploadResult.Success
}
},
resetStaleBootloader = { staleResets++ },
interAttemptDelay = {},
)
assertEquals(DfuUploadResult.Success, outcome)
assertEquals(2, sessions, "stale cleanup must not consume an upload attempt — 2 sessions for 1 success")
assertEquals(1, staleResets)
}
/**
* The profile switches from NORMAL to RECOVERY only after a MidStreamDisconnect. A StaleSessionReset by itself must
* NOT switch the profile — the link did not actually drop mid-upload.
*/
@Test
fun `NORMAL switches to RECOVERY only after MidStreamDisconnect`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
// Initial stale session, then mid-stream drop, then success — only the post-drop attempt(s) should be RECOVERY.
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
when (sessions.size) {
1 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
2 ->
DfuUploadResult.Failure(
LegacyDfuException.MidStreamDisconnect(20, 200, "Disconnected", lastConfirmedBytes = -1),
false,
)
else -> DfuUploadResult.Success
}
},
resetStaleBootloader = {},
interAttemptDelay = {},
)
// Session 1 (stale) stayed NORMAL; session 2 (drop) was still NORMAL; session 3 (post-drop) switched to
// RECOVERY.
assertEquals(
listOf(LegacyDfuStreamProfile.NORMAL, LegacyDfuStreamProfile.NORMAL, LegacyDfuStreamProfile.RECOVERY),
sessions,
)
}
/**
* An initial StaleSessionReset (no prior drop) keeps NORMAL — the most common case of a device stranded in the
* bootloader from a previous interrupted flash.
*/
@Test
fun `initial StaleSessionReset keeps NORMAL profile`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
when (sessions.size) {
1 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
else -> DfuUploadResult.Success
}
},
resetStaleBootloader = {},
interAttemptDelay = {},
)
sessions.forEach { assertEquals(LegacyDfuStreamProfile.NORMAL, it) }
}
/**
* The original mid-stream failure stays visible even when the terminal failure is a later cleanup/transfer error.
* The retry loop attaches the first MidStreamDisconnect as a suppressed exception on the surfaced error.
*/
@Test
fun `original MidStreamDisconnect stays visible when terminal failure differs`() = runTest {
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LEGACY_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = {
when (it) {
LegacyDfuStreamProfile.NORMAL ->
DfuUploadResult.Failure(
LegacyDfuException.MidStreamDisconnect(
50,
200,
"Disconnected",
lastConfirmedBytes = -1,
),
false,
)
// All RECOVERY attempts fail with ordinary errors until the budget is exhausted.
else -> DfuUploadResult.Failure(DfuException.TransferFailed("recovery transfer failed"), false)
}
},
resetStaleBootloader = {},
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
val terminal = outcome.error
assertIs<DfuException.TransferFailed>(terminal)
assertEquals(1, terminal.suppressedExceptions.size)
val suppressed = terminal.suppressedExceptions.single()
assertIs<LegacyDfuException.MidStreamDisconnect>(suppressed)
assertEquals(50, suppressed.bytesSent)
assertEquals(200, suppressed.totalBytes)
}
/**
* Secure DFU never throws the Legacy-only exceptions, so it just runs the `attempts` loop — no stale cleanup, no
* profile switching.
*/
@Test
fun `Secure protocol runs the attempts loop without stale handling`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
var staleResets = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.SECURE,
// Secure budget is always 1/1 — pre==engaged, so engagement promotion is a no-op.
budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LIMITED_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
DfuUploadResult.Failure(DfuException.TransferFailed("secure transfer failed"), false)
},
resetStaleBootloader = { staleResets++ },
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
assertIs<DfuException.TransferFailed>(outcome.error)
assertEquals(LIMITED_SESSION_ATTEMPTS, sessions.size)
// Secure never switches profile even if a MidStreamDisconnect-shaped failure surfaced (it cannot —
// MidStreamDisconnect is a LegacyDfuException subtype).
sessions.forEach { assertEquals(LegacyDfuStreamProfile.NORMAL, it) }
assertEquals(0, staleResets)
}
/** Secure DFU succeeds as soon as the first session succeeds — no fallback, no profile switching. */
@Test
fun `Secure protocol returns Success on first successful session`() = runTest {
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.SECURE,
budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LIMITED_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { DfuUploadResult.Success },
resetStaleBootloader = {},
interAttemptDelay = {},
)
assertEquals(DfuUploadResult.Success, outcome)
}
// -----------------------------------------------------------------------
// Two-stage budget: pre-engagement probe → engaged promotion
// -----------------------------------------------------------------------
/**
* Unknown Legacy pre-engagement connect failure: budget stays at the speculative 1-attempt cap so Secure fallback
* stays timely. The probe never engaged, so no promotion occurs.
*/
@Test
fun `Unknown Legacy pre-engagement connect failure stops after 1 attempt for timely Secure fallback`() = runTest {
var sessions = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = {
sessions++
DfuUploadResult.Failure(DfuException.ConnectionFailed("connect failed"), false)
},
resetStaleBootloader = {},
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
assertEquals(1, sessions, "pre-engagement budget must be 1 so Secure can fall back in time")
}
/**
* Unknown Legacy engages then MidStreamDisconnect: budget promotes from 1 → 3 inside the loop, and subsequent
* attempts use RECOVERY.
*/
@Test
fun `Unknown Legacy engagement during MidStreamDisconnect promotes budget to 3 and uses RECOVERY`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
if (sessions.size == 1) {
// First attempt: engages, then drops mid-stream.
DfuUploadResult.Failure(
LegacyDfuException.MidStreamDisconnect(50, 200, "Disconnected", lastConfirmedBytes = -1),
true,
)
} else {
// Subsequent RECOVERY attempts fail ordinarily to exhaust the promoted budget.
DfuUploadResult.Failure(DfuException.TransferFailed("recovery failed"), true)
}
},
resetStaleBootloader = {},
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
assertEquals(3, sessions.size, "promoted engaged budget must allow 3 attempts")
assertEquals(LegacyDfuStreamProfile.NORMAL, sessions.first())
sessions.drop(1).forEach { assertEquals(LegacyDfuStreamProfile.RECOVERY, it) }
}
/**
* Unknown Legacy engages, drop → stale → success: the stale cleanup after engagement does NOT consume the promoted
* budget. Three upload sessions total (drop, stale, success) but only one upload-attempt budget slot consumed
* before success.
*/
@Test
fun `Unknown Legacy stale cleanup after engagement does not consume promoted budget`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
var staleResets = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
when (sessions.size) {
1 ->
DfuUploadResult.Failure(
LegacyDfuException.MidStreamDisconnect(
50,
200,
"Disconnected",
lastConfirmedBytes = -1,
),
true,
)
2 -> DfuUploadResult.Failure(LegacyDfuException.StaleSessionReset(), true)
else -> DfuUploadResult.Success
}
},
resetStaleBootloader = { staleResets++ },
interAttemptDelay = {},
)
assertEquals(DfuUploadResult.Success, outcome)
assertEquals(3, sessions.size, "drop+stale+success = 3 sessions; stale cleanup does not consume budget")
assertEquals(1, staleResets)
}
/**
* A Legacy alternate that engages (e.g. SecureObserved primary fails after engagement → Legacy fallback) receives
* the full Legacy engaged budget of 3.
*/
@Test
fun `Legacy alternate that engages receives full engaged Legacy budget`() = runTest {
val sessions = mutableListOf<LegacyDfuStreamProfile>()
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.LEGACY,
// Legacy alternate: pre=1 (probe), engaged=3 (full Legacy budget).
budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LEGACY_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = { profile ->
sessions.add(profile)
// Engages immediately, then fails until the budget is exhausted.
DfuUploadResult.Failure(DfuException.TransferFailed("legacy alt failure"), true)
},
resetStaleBootloader = {},
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
assertEquals(3, sessions.size, "Legacy alternate that engages must receive full engaged budget")
}
/**
* Secure never receives a Legacy-style engaged promotion: its budget carries pre==engaged==1, so even when a Secure
* session reports protocolEngaged=true the active attempt cap stays at 1.
*/
@Test
fun `Secure budget stays at 1 even when Secure session reports protocolEngaged true`() = runTest {
var sessions = 0
val outcome =
runDfuRetryLoop(
protocol = DfuProtocolKind.SECURE,
budget = DfuAttemptBudget(LIMITED_SESSION_ATTEMPTS, LIMITED_SESSION_ATTEMPTS),
maxStaleResets = MAX_LEGACY_STALE_RESETS,
runUploadSession = {
sessions++
// Even with protocolEngaged=true, the Secure budget cannot grow (pre==engaged==1).
DfuUploadResult.Failure(DfuException.TransferFailed("secure engages then fails"), true)
},
resetStaleBootloader = {},
interAttemptDelay = {},
)
assertIs<DfuUploadResult.Failure>(outcome)
assertEquals(1, sessions, "Secure budget must stay at 1 even when protocolEngaged=true (no Legacy promotion)")
}
// -----------------------------------------------------------------------
// Cleanup policy (cleanupDfuSessionTransport)
// -----------------------------------------------------------------------
/** Fake transport for cleanup-policy tests — records abort/close calls and can inject failures. */
private class FakeCleanupTransport(val abortAction: () -> Unit = {}, val closeAction: () -> Unit = {}) :
DfuUploadTransport {
var abortCalled = false
var closeCalled = false
override val isLowSpeedTransfer: Boolean = false
override suspend fun connectToDfuMode(): Result<Unit> = Result.success(Unit)
override suspend fun transferInitPacket(initPacket: ByteArray): Result<Unit> = Result.success(Unit)
override suspend fun transferFirmware(firmware: ByteArray, onProgress: suspend (Float) -> Unit): Result<Unit> =
Result.success(Unit)
override suspend fun abort() {
abortCalled = true
abortAction()
}
override suspend fun close() {
closeCalled = true
closeAction()
}
}
@Test
fun `cleanup closes transport after abort throws Exception`() = runTest {
// Even if abort throws an operational Exception, close must still be called. The exception propagates after
// close because the helper's finally guarantees close.
val transport = FakeCleanupTransport(abortAction = { throw RuntimeException("abort failed") })
assertFailsWith<RuntimeException> {
cleanupDfuSessionTransport(transport, completed = false, sessionFailure = null)
}
assertTrue(transport.abortCalled)
assertTrue(transport.closeCalled)
}
@Test
fun `cleanup closes transport after abort throws Error`() = runTest {
// Error subtypes propagate through the cleanup helper's finally — close still runs.
val transport = FakeCleanupTransport(abortAction = { throw AssertionError("abort error") })
assertFailsWith<AssertionError> {
cleanupDfuSessionTransport(transport, completed = false, sessionFailure = null)
}
assertTrue(transport.abortCalled)
assertTrue(transport.closeCalled)
}
@Test
fun `cleanup skips abort after StaleSessionReset but always closes`() = runTest {
val transport = FakeCleanupTransport()
cleanupDfuSessionTransport(
transport,
completed = false,
sessionFailure = LegacyDfuException.StaleSessionReset(),
)
assertFalse(transport.abortCalled, "abort must be skipped after StaleSessionReset")
assertTrue(transport.closeCalled, "close must always run")
}
@Test
fun `cleanup skips abort after MidStreamDisconnect but always closes`() = runTest {
val transport = FakeCleanupTransport()
cleanupDfuSessionTransport(
transport,
completed = false,
sessionFailure = LegacyDfuException.MidStreamDisconnect(20, 200, "Disconnected", lastConfirmedBytes = -1),
)
assertFalse(transport.abortCalled, "abort must be skipped after MidStreamDisconnect")
assertTrue(transport.closeCalled, "close must always run")
}
@Test
fun `cleanup attempts abort after ambiguous ConnectionFailed and always closes`() = runTest {
val transport = FakeCleanupTransport()
cleanupDfuSessionTransport(
transport,
completed = false,
sessionFailure = DfuException.ConnectionFailed("connect failed"),
)
assertTrue(transport.abortCalled, "abort must be attempted after ConnectionFailed (link may still be usable)")
assertTrue(transport.closeCalled, "close must always run")
}
@Test
fun `cleanup skips abort after completed session but always closes`() = runTest {
val transport = FakeCleanupTransport()
cleanupDfuSessionTransport(transport, completed = true, sessionFailure = null)
assertFalse(transport.abortCalled, "abort must be skipped after a completed session")
assertTrue(transport.closeCalled, "close must always run")
}
}

341
feature/firmware/src/commonTest/kotlin/org/meshtastic/feature/firmware/ota/dfu/LegacyDfuTransportTest.kt
View File

@@ -18,11 +18,18 @@
package org.meshtastic.feature.firmware.ota.dfu
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitCancellation
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.test.advanceUntilIdle
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import org.meshtastic.core.ble.BleCharacteristic
import org.meshtastic.core.ble.BleConnection
@@ -40,6 +47,7 @@ import org.meshtastic.core.testing.FakeBleWrite
import kotlin.test.Test
import kotlin.test.assertContentEquals
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith
import kotlin.test.assertIs
import kotlin.test.assertTrue
import kotlin.time.Duration
@@ -308,7 +316,7 @@ class LegacyDfuTransportTest {
env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
// Send (PRN_INTERVAL_PACKETS + 1) packets of 20 bytes — guarantees a PRN window fires
// before the firmware completes, so the under-reported byte count surfaces as a mismatch.
val firmwareSize = (LegacyDfuTransport.PRN_INTERVAL_PACKETS + 1) * 20
val firmwareSize = (PRN_INTERVAL_PACKETS + 1) * 20
val result = env.transport.transferFirmware(ByteArray(firmwareSize)) {}
assertTrue(result.isFailure)
@@ -353,7 +361,11 @@ class LegacyDfuTransportTest {
// Test infrastructure
// -----------------------------------------------------------------------
private class TestEnv(val transport: LegacyDfuTransport, val service: AutoRespondingLegacyService) {
private class TestEnv(
val transport: LegacyDfuTransport,
val service: AutoRespondingLegacyService,
val connection: FakeBleConnection,
) {
val responder = service.responder
fun controlPointWrites(): List<FakeBleWrite> =
@@ -363,7 +375,10 @@ class LegacyDfuTransportTest {
service.delegate.writes.filter { it.characteristic.uuid == LEGACY_DFU_PACKET_UUID }
}
private suspend fun createConnectedTransport(mtu: Int? = null): TestEnv {
private suspend fun createConnectedTransport(
mtu: Int? = null,
profile: LegacyDfuStreamProfile = LegacyDfuStreamProfile.NORMAL,
): TestEnv {
val scanner = FakeBleScanner()
val fakeConnection = FakeBleConnection().apply { maxWriteValueLength = mtu }
val autoService = AutoRespondingLegacyService(fakeConnection.service)
@@ -372,21 +387,41 @@ class LegacyDfuTransportTest {
object : BleConnectionFactory {
override fun create(scope: CoroutineScope, tag: String): BleConnection = wrappedConnection
}
val transport = LegacyDfuTransport(scanner, factory, address, Dispatchers.Unconfined)
val transport = LegacyDfuTransport(scanner, factory, address, Dispatchers.Unconfined, profile)
scanner.emitDevice(FakeBleDevice(dfuAddress))
transport.connectToDfuMode().getOrThrow()
return TestEnv(transport, autoService)
return TestEnv(transport, autoService, fakeConnection)
}
/**
* Drives the simulated bootloader response stream. After each `write()` to control point or packet, this service
* synthesises the appropriate notification(s) on the control-point characteristic so the transport's pending
* `awaitResponse` / `awaitPacketReceipt` calls unblock.
* `awaitResponse` / `awaitPacketReceiptDuringStream` calls unblock.
*/
private class AutoRespondingLegacyService(val delegate: FakeBleService) : BleService {
val responder = LegacyResponder()
/**
* When `true`, the next `write()` to the Legacy Control Point characteristic suspends forever (via
* [awaitCancellation]) instead of completing. Used to prove the bounded abort returns within
* [LegacyDfuTransport.RESET_WRITE_TIMEOUT] when the device never ACKs the RESET.
*/
var hangOnControlPointWrites: Boolean = false
/**
* When `true`, the next `write()` to the Legacy Control Point characteristic throws an [AssertionError]. Used
* to prove [LegacyDfuTransport.abort] (which catches [Exception], not [Throwable]) does not swallow [Error]
* subtypes.
*/
var throwErrorOnControlPointWrites: Boolean = false
/**
* When `true`, the next `write()` to the Legacy Control Point characteristic throws a [RuntimeException]. Used
* to prove [LegacyDfuTransport.abort] swallows operational [Exception] subtypes (best-effort / non-fatal).
*/
var throwExceptionOnControlPointWrites: Boolean = false
override fun hasCharacteristic(c: BleCharacteristic) = delegate.hasCharacteristic(c)
override fun observe(c: BleCharacteristic): Flow<ByteArray> = delegate.observe(c)
@@ -396,6 +431,15 @@ class LegacyDfuTransportTest {
override fun preferredWriteType(c: BleCharacteristic): BleWriteType = delegate.preferredWriteType(c)
override suspend fun write(c: BleCharacteristic, data: ByteArray, writeType: BleWriteType) {
if (throwErrorOnControlPointWrites && c.uuid == LegacyDfuUuids.CONTROL_POINT) {
throw AssertionError("Simulated assertion failure during control point write")
}
if (throwExceptionOnControlPointWrites && c.uuid == LegacyDfuUuids.CONTROL_POINT) {
throw RuntimeException("Simulated link failure during control point write")
}
if (hangOnControlPointWrites && c.uuid == LegacyDfuUuids.CONTROL_POINT) {
awaitCancellation()
}
delegate.write(c, data, writeType)
val response = responder.onWrite(c.uuid, data) ?: return
response.forEach { delegate.emitNotification(LegacyDfuUuids.CONTROL_POINT, it) }
@@ -426,13 +470,33 @@ class LegacyDfuTransportTest {
var scheme: LegacyResponderScheme = LegacyResponderScheme.HappyPath
var failOnActivateWrite: Boolean = false
/**
* Optional suspend hook invoked after each firmware-packet write with the cumulative byte count, while
* [AutoRespondingLegacyService.write] is still in progress. Tests use this to hold a write in flight (e.g. via
* [awaitCancellation]) while another coroutine changes connection state to simulate a mid-stream link drop at a
* deterministic offset.
*/
var onFirmwarePacketWrite: (suspend (bytesReceived: Long) -> Unit)? = null
/**
* When `true`, packet-receipt notifications are NOT emitted by the simulated bootloader. Used to suspend the
* transport inside the PRN wait so a test can drive a disconnect while the wait is in flight.
*/
var suppressPrnNotifications: Boolean = false
private var packetBytesReceived = 0L
private var packetsSinceLastPrn = 0
private var firmwareTransferStarted = false
private var imageSizesWritten = false
private var expectedFirmwareSize: Int = 0
fun onWrite(uuid: kotlin.uuid.Uuid, data: ByteArray): List<ByteArray>? = when (uuid) {
/**
* Effective PRN cadence. Defaults to the NORMAL profile value; updated from the `PACKET_RECEIPT_NOTIF_REQ`
* control write so the responder matches whatever interval the transport requested (NORMAL=10, RECOVERY=5).
*/
private var prnInterval: Int = PRN_INTERVAL_PACKETS
suspend fun onWrite(uuid: kotlin.uuid.Uuid, data: ByteArray): List<ByteArray>? = when (uuid) {
LegacyDfuUuids.CONTROL_POINT -> handleControlWrite(data)
LEGACY_DFU_PACKET_UUID -> handlePacketWrite(data)
else -> null
@@ -454,7 +518,13 @@ class LegacyDfuTransportTest {
}
}
LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ -> null
LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ -> {
if (data.size >= 3) {
// Parse uint16-LE PRN value the transport just wrote and follow it for receipt cadence.
prnInterval = (data[1].toInt() and 0xFF) or ((data[2].toInt() and 0xFF) shl 8)
}
null
}
LegacyDfuOpcode.RECEIVE_FIRMWARE_IMAGE -> {
firmwareTransferStarted = true
@@ -476,7 +546,7 @@ class LegacyDfuTransportTest {
}
}
private fun handlePacketWrite(data: ByteArray): List<ByteArray>? {
private suspend fun handlePacketWrite(data: ByteArray): List<ByteArray>? {
// First packet write is the 12-byte image sizes payload (after START_DFU).
if (!imageSizesWritten) {
imageSizesWritten = true
@@ -494,17 +564,22 @@ class LegacyDfuTransportTest {
if (firmwareTransferStarted) {
packetBytesReceived += data.size
packetsSinceLastPrn++
onFirmwarePacketWrite?.invoke(packetBytesReceived)
val responses = mutableListOf<ByteArray>()
val firmwareDone = packetBytesReceived >= expectedFirmwareSize
if (packetsSinceLastPrn >= LegacyDfuTransport.PRN_INTERVAL_PACKETS && !firmwareDone) {
// Use the cadence the transport actually requested via PACKET_RECEIPT_NOTIF_REQ so a RECOVERY
// (prnInterval=5) upload receives receipts at the smaller interval.
if (packetsSinceLastPrn >= prnInterval && !firmwareDone) {
packetsSinceLastPrn = 0
val reported =
if (scheme == LegacyResponderScheme.PrnUnderReport) {
packetBytesReceived - 1
} else {
packetBytesReceived
}
responses += packetReceipt(reported)
if (!suppressPrnNotifications) {
val reported =
if (scheme == LegacyResponderScheme.PrnUnderReport) {
packetBytesReceived - 1
} else {
packetBytesReceived
}
responses += packetReceipt(reported)
}
}
if (firmwareDone) {
responses += success(LegacyDfuOpcode.RECEIVE_FIRMWARE_IMAGE)
@@ -585,4 +660,236 @@ class LegacyDfuTransportTest {
override fun maximumWriteValueLength(writeType: BleWriteType): Int? =
delegate.maximumWriteValueLength(writeType)
}
// -----------------------------------------------------------------------
// Stream profile selection
// -----------------------------------------------------------------------
@Test
fun `NORMAL profile writes PRN request value 10`() = runTest {
val env = createConnectedTransport(profile = LegacyDfuStreamProfile.NORMAL)
env.responder.scheme = LegacyResponderScheme.HappyPath
env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
env.transport.transferFirmware(ByteArray(40)) {}
val prnWrite = env.controlPointWrites().single { it.data[0] == LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ }
val requested = (prnWrite.data[1].toInt() and 0xFF) or ((prnWrite.data[2].toInt() and 0xFF) shl 8)
assertEquals(PRN_INTERVAL_PACKETS, requested)
}
@Test
fun `RECOVERY profile writes PRN request value 5`() = runTest {
val env = createConnectedTransport(profile = LegacyDfuStreamProfile.RECOVERY)
env.responder.scheme = LegacyResponderScheme.HappyPath
env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
env.transport.transferFirmware(ByteArray(40)) {}
val prnWrite = env.controlPointWrites().single { it.data[0] == LegacyDfuOpcode.PACKET_RECEIPT_NOTIF_REQ }
val requested = (prnWrite.data[1].toInt() and 0xFF) or ((prnWrite.data[2].toInt() and 0xFF) shl 8)
assertEquals(RECOVERY_PRN_INTERVAL_PACKETS, requested)
}
@Test
fun `RECOVERY profile awaits packet receipts at the smaller interval`() = runTest {
// RECOVERY prnInterval = 5, so a 100-byte firmware at 20-byte packets fires a PRN exactly once at byte 100
// (5 packets × 20B) — but offset == firmware.size at that point means the `offset < firmware.size` gate
// suppresses the wait. Use 110 bytes (would need 6 packets at NORMAL=10 to hit the first PRN) so the
// first PRN at 100B fires under RECOVERY but never fires under NORMAL within the 110B upload.
val env = createConnectedTransport(profile = LegacyDfuStreamProfile.RECOVERY)
env.responder.scheme = LegacyResponderScheme.HappyPath
env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
val progress = mutableListOf<Float>()
val result = env.transport.transferFirmware(ByteArray(110)) { progress += it }
assertTrue(result.isSuccess, "transferFirmware failed: ${result.exceptionOrNull()}")
// A PRN fires at offset 100 (5 packets × 20B under RECOVERY prnInterval=5); progress must record it during
// streaming, then the final 1f at completion. This proves the PRN was consumed mid-stream, not just drained
// by the final response wait.
assertEquals(listOf(100f / 110f, 1f), progress)
}
@Test
fun `NORMAL profile does not await a PRN for a sub-interval upload`() = runTest {
// 110-byte firmware at 20-byte packets = 6 packets. NORMAL prnInterval = 10, so no PRN is due before the
// upload completes — the auto-responder must NOT emit a PRN, and the transport must NOT call
// awaitPacketReceiptDuringStream.
val env = createConnectedTransport(profile = LegacyDfuStreamProfile.NORMAL)
env.responder.scheme = LegacyResponderScheme.HappyPath
env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
val result = env.transport.transferFirmware(ByteArray(110)) {}
assertTrue(result.isSuccess, "transferFirmware failed: ${result.exceptionOrNull()}")
}
// -----------------------------------------------------------------------
// Mid-stream disconnect
// -----------------------------------------------------------------------
@Test
fun `mid-stream disconnect returns typed MidStreamDisconnect with correct byte counts`() = runTest {
val env = createConnectedTransport()
env.responder.scheme = LegacyResponderScheme.HappyPath
// Hold the first firmware-packet write in flight so the transfer cannot run through all 200 bytes before
// the test injects the disconnect. Production streamOffset has already published 20 before the write, so
// the tripwire observer will read bytesSent == 20 when the link drops.
val firstPacketWriteInFlight = CompletableDeferred<Unit>()
env.responder.onFirmwarePacketWrite = { bytes ->
if (bytes >= 20L) {
firstPacketWriteInFlight.complete(Unit)
awaitCancellation()
}
}
env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
val transferDeferred = async { env.transport.transferFirmware(ByteArray(200)) {} }
firstPacketWriteInFlight.await()
runCurrent()
assertTrue(
!transferDeferred.isCompleted,
"transfer must remain suspended inside the first firmware-packet write",
)
env.connection.simulateRemoteDisconnect()
runCurrent()
val result = transferDeferred.await()
val error =
assertIs<LegacyDfuException.MidStreamDisconnect>(
result.exceptionOrNull(),
"expected the outer stream tripwire to classify the disconnect while the first write was in flight",
)
// First packet write was in flight (streamOffset published as 20 before service.write), so bytesSent == 20.
assertEquals(20, error.bytesSent)
assertEquals(200, error.totalBytes)
// No PRN was received before the drop (first PRN would be at byte 200), so lastConfirmedBytes is -1.
assertEquals(-1, error.lastConfirmedBytes)
assertTrue(error.connectionState.isNotBlank())
}
/**
* A drop during a stream PRN wait must be classified by the OUTER stream-level tripwire as
* [LegacyDfuException.MidStreamDisconnect] — NOT as a generic handshake-style [DfuException.ConnectionFailed]. The
* stream PRN wait uses [awaitPacketReceiptDuringStream], which deliberately does NOT install its own disconnect
* tripwire, so the outer watcher is the sole classifier.
*/
@Test
fun `stream-level tripwire classifies a drop during a PRN wait as MidStreamDisconnect`() = runTest {
val env = createConnectedTransport()
env.responder.scheme = LegacyResponderScheme.HappyPath
// Suppress PRNs so the transport suspends inside awaitPacketReceiptDuringStream at the first PRN boundary.
env.responder.suppressPrnNotifications = true
env.transport.transferInitPacket(ByteArray(14)).getOrThrow()
// First NORMAL PRN boundary = prnInterval (10) × default packet size (20B, MTU not negotiated) = byte 200.
// The responder fires this hook synchronously after accounting the 10th packet write, so completing the
// deferred proves the stream has written bytes 0..200 and is about to enter awaitPacketReceiptDuringStream —
// long before virtual time can reach the 30s COMMAND_TIMEOUT that would otherwise mask the tripwire path.
val reachedPrnBoundary = CompletableDeferred<Unit>()
env.responder.onFirmwarePacketWrite = { bytes -> if (bytes >= 200L) reachedPrnBoundary.complete(Unit) }
// Run the transfer asynchronously so the test body can synchronize the disconnect to the suspended PRN wait.
val transferDeferred = async { env.transport.transferFirmware(ByteArray(220)) {} }
// Wait for the 10th packet write to land, then flush dispatched continuations so the stream coroutine reaches
// the suspended receive() inside awaitPacketReceiptDuringStream — without advancing virtual time.
reachedPrnBoundary.await()
runCurrent()
// Simulate remote disconnect while the transport is suspended in awaitPacketReceiptDuringStream.
env.connection.simulateRemoteDisconnect()
runCurrent()
val result = transferDeferred.await()
assertTrue(result.isFailure)
val ex = assertIs<LegacyDfuException.MidStreamDisconnect>(result.exceptionOrNull())
// Must be the typed MidStreamDisconnect from the outer stream-level tripwire — NOT the generic
// ConnectionFailed that an inner handshake-style tripwire would have produced, and NOT a local Timeout.
assertEquals(200, ex.bytesSent)
assertEquals(220, ex.totalBytes)
}
// -----------------------------------------------------------------------
// Bounded RESET (abort)
// -----------------------------------------------------------------------
@Test
fun `RESET is written WITH_RESPONSE`() = runTest {
val env = createConnectedTransport()
env.transport.abort()
val resetWrite = env.controlPointWrites().single { it.data.isNotEmpty() && it.data[0] == LegacyDfuOpcode.RESET }
assertEquals(BleWriteType.WITH_RESPONSE, resetWrite.writeType)
}
@Test
fun `acknowledged RESET completes and logs acknowledgement`() = runTest {
val env = createConnectedTransport()
env.transport.abort()
// An acknowledged RESET means the write completed within the timeout; abort must return normally.
val resetWrite = env.controlPointWrites().single { it.data.isNotEmpty() && it.data[0] == LegacyDfuOpcode.RESET }
assertTrue(resetWrite.data.size == 1, "RESET payload is a single opcode byte")
}
@Test
fun `abort returns within RESET_WRITE_TIMEOUT when the RESET write hangs`() = runTest {
val env = createConnectedTransport()
env.service.hangOnControlPointWrites = true
// Run abort on the test dispatcher so virtual time advances past RESET_WRITE_TIMEOUT. The hanging write
// never completes; withTimeoutOrNull inside abort must return null and abort must return normally.
val abortJob = async { env.transport.abort() }
advanceUntilIdle()
abortJob.await()
// No RESET write should have been recorded on the delegate — the hang prevented the write from reaching
// FakeBleService.write, so an "acknowledged" log would be a lie.
val resetWrites =
env.controlPointWrites().filter { it.data.isNotEmpty() && it.data[0] == LegacyDfuOpcode.RESET }
assertTrue(resetWrites.isEmpty(), "RESET write should not have been recorded when the link hung")
}
@Test
fun `abort propagates parent cancellation rather than reporting success`() = runTest {
val env = createConnectedTransport()
env.service.hangOnControlPointWrites = true
// parentJob stands in for the caller's scope; cancelling it must propagate through withTimeoutOrNull
// (which only swallows its own TimeoutCancellationException, not parent cancellation) and out of abort.
val parentJob = Job()
val abortDeferred = async(parentJob) { env.transport.abort() }
// Let abort reach the hanging RESET write.
runCurrent()
parentJob.cancel()
assertFailsWith<CancellationException> { abortDeferred.await() }
}
@Test
fun `abort does not swallow Error subtypes from the RESET write`() = runTest {
val env = createConnectedTransport()
env.service.throwErrorOnControlPointWrites = true
// abort catches Exception (operational link failures) but lets Error subtypes propagate.
assertFailsWith<AssertionError> { env.transport.abort() }
}
@Test
fun `abort swallows operational Exception from the RESET write`() = runTest {
val env = createConnectedTransport()
env.service.throwExceptionOnControlPointWrites = true
// A RuntimeException from the control-point write is an operational Exception — abort must catch it
// (best-effort, non-fatal) and return normally without rethrowing.
env.transport.abort()
}
}