fix(aap): Bound handshake with watchdog and escalate reconnect backoff

2026-09-16 11:16:12 -04:00 · 2026-06-14 21:28:42 +02:00
parent e406fd4cc1
commit fd4a43de96
7 changed files with 337 additions and 81 deletions
@@ -9,20 +9,22 @@ import eu.darken.capod.common.flow.setupCommonEventHandlers
 import eu.darken.capod.monitor.core.ble.BlePodMonitor
 import eu.darken.capod.pods.core.apple.PodModel
 import eu.darken.capod.pods.core.apple.aap.AapConnectionManager
+import eu.darken.capod.pods.core.apple.aap.AapDisconnectEvent
 import eu.darken.capod.pods.core.apple.aap.AapPodState
 import eu.darken.capod.profiles.core.AppleDeviceProfile
 import eu.darken.capod.profiles.core.DeviceProfilesRepo
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.mapLatest
 import kotlinx.coroutines.flow.merge
-import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
+import java.util.concurrent.ConcurrentHashMap
 import javax.inject.Inject
 import javax.inject.Singleton
 import kotlin.time.Duration.Companion.seconds
@@ -37,6 +39,21 @@ class AapAutoConnect @Inject constructor(
    private val activeReconnects = java.util.Collections.synchronizedSet(mutableSetOf<String>())
    private val processedModelCorrections = java.util.Collections.synchronizedSet(mutableSetOf<String>())

+    /**
+     * Per-address count of consecutive connection attempts that never reached READY (failed socket
+     * connects and handshake-timeout disconnects). Drives the escalating reconnect cooldown so a
+     * device that keeps stalling its handshake in a crowded RF environment doesn't churn a tight
+     * reconnect loop. Reset to 0 once a session works (a was-ready disconnect) or the device leaves.
+     */
+    private val preReadyFailures = ConcurrentHashMap<String, Int>()
+
+    /** Cooldown (ms) to wait before the next attempt, given how many consecutive pre-READY failures we've seen. */
+    private fun cooldownFor(address: String): Long {
+        val failures = preReadyFailures[address] ?: 0
+        if (failures <= 0) return 0L
+        return HANDSHAKE_BACKOFF[minOf(failures - 1, HANDSHAKE_BACKOFF.lastIndex)]
+    }
+
    fun monitor(): Flow<Unit> = merge(
        initialConnect(),
        reconnectOnDisconnect(),
@@ -74,104 +91,150 @@ class AapAutoConnect @Inject constructor(
            return
        }

+        // Escalating cooldown carried over from prior failed attempts (socket failures or handshake
+        // stalls). 0 on the first attempt so a healthy device connects without delay.
+        val cooldown = cooldownFor(address)
+        if (cooldown > 0) {
+            log(TAG) { "AAP initial connect cooldown ${cooldown}ms for $address (failures=${preReadyFailures[address]})" }
+            delay(cooldown)
+        }
+
        log(TAG) { "AAP connecting to $address (${profile.label})" }
        try {
            aapManager.connect(address, bonded.internal!!, profile.model)
            log(TAG) { "AAP connected to $address" }
+            return
        } catch (e: Exception) {
            log(TAG, WARN) { "AAP initial connect failed for $address: ${e.message}" }

            for ((attempt, delayMs) in RETRY_DELAYS.withIndex()) {
                delay(delayMs)

-                // Bail out if classic BT disconnected
+                // Bail out if classic BT disconnected — device left, clear the penalty
                val currentConnected = bluetoothManager.connectedDevices.first().map { it.address }.toSet()
                if (address !in currentConnected) {
                    log(TAG) { "AAP initial retry: $address no longer classically connected, stopping" }
-                    break
+                    preReadyFailures.remove(address)
+                    return
                }

                val retryState = aapManager.allStates.value[address]
                if (retryState != null && retryState.connectionState != AapPodState.ConnectionState.DISCONNECTED) {
                    log(TAG) { "AAP initial retry: $address already reconnected, stopping" }
-                    break
+                    return
                }

                try {
                    log(TAG) { "AAP initial retry ${attempt + 1} for $address after ${delayMs}ms" }
                    aapManager.connect(address, bonded.internal!!, profile.model)
                    log(TAG) { "AAP connected to $address on retry ${attempt + 1}" }
-                    break
+                    return
                } catch (retryException: Exception) {
                    log(TAG, WARN) { "AAP initial retry ${attempt + 1} failed for $address: ${retryException.message}" }
                }
            }
+
+            // Every attempt this episode failed at the socket level — escalate the next round's cooldown.
+            preReadyFailures.merge(address, 1, Int::plus)
        }
    }

-    private fun reconnectOnDisconnect(): Flow<Unit> = aapManager.disconnectEvents
-        .onEach { address ->
-            if (!activeReconnects.add(address)) {
-                log(TAG, VERBOSE) { "AAP reconnect already in progress for $address, skipping" }
-                return@onEach
-            }
-
-            try {
-                for ((attempt, delayMs) in RETRY_DELAYS.withIndex()) {
-                    delay(delayMs)
-
-                    // Check if still profiled
-                    val profile = profilesRepo.profiles.first()
-                        .firstOrNull { it.address == address }
-                    if (profile == null) {
-                        log(TAG) { "AAP reconnect: $address no longer profiled, stopping" }
-                        break
-                    }
-
-                    // Check if still bonded
-                    val bonded = bluetoothManager.bondedDevices().first()
-                        .firstOrNull { it.address == address }
-                    if (bonded == null) {
-                        log(TAG) { "AAP reconnect: $address no longer bonded, stopping" }
-                        break
-                    }
-
-                    // Check if still classically connected
-                    val currentConnected = bluetoothManager.connectedDevices.first().map { it.address }.toSet()
-                    if (address !in currentConnected) {
-                        log(TAG) { "AAP reconnect: $address no longer classically connected, stopping" }
-                        break
-                    }
-
-                    // Check if still visible in BLE
-                    val bleDevices = blePodMonitor.devices.first()
-                    if (bleDevices.none { it.meta?.profile?.address == address }) {
-                        log(TAG) { "AAP reconnect: $address no longer visible in BLE, stopping" }
-                        break
-                    }
-
-                    // Check if already reconnected (e.g., by initialConnect)
-                    val currentState = aapManager.allStates.value[address]
-                    if (currentState != null && currentState.connectionState != AapPodState.ConnectionState.DISCONNECTED) {
-                        log(TAG) { "AAP reconnect: $address already reconnected" }
-                        break
-                    }
-
-                    try {
-                        log(TAG) { "AAP reconnect attempt ${attempt + 1} for $address in ${delayMs}ms" }
-                        aapManager.connect(address, bonded.internal!!, profile.model)
-                        log(TAG) { "AAP reconnected to $address" }
-                        break
-                    } catch (e: Exception) {
-                        log(TAG, WARN) { "AAP reconnect attempt ${attempt + 1} failed for $address: ${e.message}" }
-                    }
-                }
-            } finally {
-                activeReconnects.remove(address)
-            }
+    // Process each disconnect on its own child coroutine so a long per-address backoff never blocks
+    // another device's reconnect (the collector would otherwise serialise events). channelFlow gives
+    // us a scope to launch into; it intentionally emits nothing — it stays subscribed for its lifetime.
+    private fun reconnectOnDisconnect(): Flow<Unit> = channelFlow<Unit> {
+        aapManager.disconnectEvents.collect { event ->
+            launch { handleReconnect(event) }
        }
-        .map { } // SharedFlow<BluetoothAddress> → Flow<Unit>
-        .setupCommonEventHandlers(TAG) { "reconnect" }
+    }.setupCommonEventHandlers(TAG) { "reconnect" }
+
+    private suspend fun handleReconnect(event: AapDisconnectEvent) {
+        val address = event.address
+
+        // A session that actually worked resets the penalty (prompt reconnect). One that never reached
+        // READY is a failed handshake/short session — escalate so a persistent stall backs off.
+        if (event.wasEverReady) {
+            preReadyFailures.remove(address)
+        } else {
+            preReadyFailures.merge(address, 1, Int::plus)
+        }
+
+        if (!activeReconnects.add(address)) {
+            log(TAG, VERBOSE) { "AAP reconnect already in progress for $address, skipping" }
+            return
+        }
+
+        try {
+            // Escalating cooldown before the normal retry cadence when handshakes keep stalling.
+            val cooldown = cooldownFor(address)
+            if (cooldown > 0) {
+                log(TAG) { "AAP reconnect cooldown ${cooldown}ms for $address (failures=${preReadyFailures[address]})" }
+                delay(cooldown)
+            }
+
+            for ((attempt, delayMs) in RETRY_DELAYS.withIndex()) {
+                delay(delayMs)
+
+                // Check if still profiled
+                val profile = profilesRepo.profiles.first()
+                    .firstOrNull { it.address == address }
+                if (profile == null) {
+                    log(TAG) { "AAP reconnect: $address no longer profiled, stopping" }
+                    preReadyFailures.remove(address)
+                    return
+                }
+
+                // Check if still bonded
+                val bonded = bluetoothManager.bondedDevices().first()
+                    .firstOrNull { it.address == address }
+                if (bonded == null) {
+                    log(TAG) { "AAP reconnect: $address no longer bonded, stopping" }
+                    preReadyFailures.remove(address)
+                    return
+                }
+
+                // Check if still classically connected
+                val currentConnected = bluetoothManager.connectedDevices.first().map { it.address }.toSet()
+                if (address !in currentConnected) {
+                    log(TAG) { "AAP reconnect: $address no longer classically connected, stopping" }
+                    preReadyFailures.remove(address)
+                    return
+                }
+
+                // Check if still visible in BLE
+                val bleDevices = blePodMonitor.devices.first()
+                if (bleDevices.none { it.meta?.profile?.address == address }) {
+                    log(TAG) { "AAP reconnect: $address no longer visible in BLE, stopping" }
+                    preReadyFailures.remove(address)
+                    return
+                }
+
+                // Check if already reconnected (e.g., by initialConnect)
+                val currentState = aapManager.allStates.value[address]
+                if (currentState != null && currentState.connectionState != AapPodState.ConnectionState.DISCONNECTED) {
+                    log(TAG) { "AAP reconnect: $address already reconnected" }
+                    return
+                }
+
+                try {
+                    log(TAG) { "AAP reconnect attempt ${attempt + 1} for $address in ${delayMs}ms" }
+                    aapManager.connect(address, bonded.internal!!, profile.model)
+                    log(TAG) { "AAP reconnected to $address" }
+                    return
+                } catch (e: Exception) {
+                    log(TAG, WARN) { "AAP reconnect attempt ${attempt + 1} failed for $address: ${e.message}" }
+                }
+            }
+
+            // Every attempt threw at the socket level (the device is still present but won't accept a
+            // socket) — escalate so we don't hammer it. Mirrors connectWithRetries. Handshake stalls
+            // don't reach here: their reconnect socket succeeds and we return above; they escalate via
+            // the never-ready disconnect event instead.
+            preReadyFailures.merge(address, 1, Int::plus)
+        } finally {
+            activeReconnects.remove(address)
+        }
+    }

    private fun correctModelOnDeviceInfo(): Flow<Unit> = aapManager.allStates
        .map { states -> states.mapValues { (_, state) -> state.deviceInfo?.modelNumber } }
@@ -224,5 +287,12 @@ class AapAutoConnect @Inject constructor(
    companion object {
        private val TAG = logTag("Monitor", "AapAutoConnect")
        internal val RETRY_DELAYS = longArrayOf(3_000, 3_000, 3_000, 5_000, 5_000, 10_000, 10_000)
+
+        /**
+         * Escalating cooldown (ms) applied before a reconnect/connect attempt, indexed by
+         * consecutive pre-READY failures minus one. Caps at 60s so a device that perpetually stalls
+         * its handshake in crowded RF settles into a slow ~minute cadence instead of a tight loop.
+         */
+        internal val HANDSHAKE_BACKOFF = longArrayOf(5_000, 15_000, 30_000, 60_000)
    }
 }
@@ -54,8 +54,8 @@ class AapConnectionManager @Inject constructor(
    private val _allStates = MutableStateFlow<Map<BluetoothAddress, AapPodState>>(emptyMap())
    val allStates: StateFlow<Map<BluetoothAddress, AapPodState>> = _allStates.asStateFlow()

-    private val _disconnectEvents = MutableSharedFlow<BluetoothAddress>(extraBufferCapacity = 16)
-    val disconnectEvents: SharedFlow<BluetoothAddress> = _disconnectEvents.asSharedFlow()
+    private val _disconnectEvents = MutableSharedFlow<AapDisconnectEvent>(extraBufferCapacity = 16)
+    val disconnectEvents: SharedFlow<AapDisconnectEvent> = _disconnectEvents.asSharedFlow()

    /** Emits when a connection receives private keys (IRK/ENC) from the device. */
    private val _keysReceived = MutableSharedFlow<Pair<BluetoothAddress, KeyExchangeResult>>(extraBufferCapacity = 16)
@@ -188,7 +188,7 @@ class AapConnectionManager @Inject constructor(
                    }

                    if (!wasIntentional) {
-                        _disconnectEvents.tryEmit(address)
+                        _disconnectEvents.tryEmit(AapDisconnectEvent(address, connection.wasEverReady))
                    }

                    // End this collector coroutine (also cancels child key-forwarding coroutine)
@@ -217,3 +217,13 @@ class AapConnectionManager @Inject constructor(
        connection.send(command)
    }
 }
+
+/**
+ * Emitted on an unintentional AAP disconnect. [wasEverReady] tells the reconnect layer whether this
+ * session ever completed the handshake (reached READY) — a never-ready disconnect is a failed
+ * handshake/short session and feeds the escalating reconnect backoff; a was-ready drop reconnects promptly.
+ */
+data class AapDisconnectEvent(
+    val address: BluetoothAddress,
+    val wasEverReady: Boolean,
+)
@@ -23,12 +23,14 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
 import java.io.IOException
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.time.Duration
@@ -46,10 +48,14 @@ internal class AapConnection(
    private val socketFactory: L2capSocketFactory,
    timeSource: TimeSource,
    private val connectTimeout: Duration = DEFAULT_CONNECT_TIMEOUT,
+    private val handshakeTimeout: Duration = DEFAULT_HANDSHAKE_TIMEOUT,
 ) {

    private val engine = AapSessionEngine(profile, timeSource)

+    /** True once this connection's session reached READY at least once. Survives the engine reset. */
+    val wasEverReady: Boolean get() = engine.wasEverReady
+
    val state: StateFlow<AapPodState> get() = engine.state
    val keysReceived: SharedFlow<KeyExchangeResult> get() = engine.keysReceived
    val stemPressEvents: SharedFlow<StemPressEvent> get() = engine.stemPressEvents
@@ -60,6 +66,7 @@ internal class AapConnection(

    private var socket: BluetoothSocket? = null
    private var readerJob: Job? = null
+    private val disconnected = AtomicBoolean(false)
    private val writeMutex = Mutex()
    private val framer = AapFramer()

@@ -109,6 +116,27 @@ internal class AapConnection(

            // Launch read loop in the provided scope — connect() returns immediately
            readerJob = scope.launch(Dispatchers.IO) { readLoop(sock) }
+
+            // Handshake watchdog: the socket can connect and the handshake be sent, but if the peer
+            // never replies the engine sits in HANDSHAKING forever (blocking read, no deadline).
+            // Bound it: if neither READY nor DISCONNECTED is reached in time, tear the socket down so
+            // the reconnect path can recover. READY / an earlier DISCONNECTED end the wait with no action.
+            scope.launch {
+                val settled = withTimeoutOrNull(handshakeTimeout) {
+                    state.first {
+                        it.connectionState == AapPodState.ConnectionState.READY ||
+                                it.connectionState == AapPodState.ConnectionState.DISCONNECTED
+                    }
+                }
+                // Re-check after the timeout: READY may have landed in the boundary race between
+                // withTimeoutOrNull cancelling and us getting here — don't tear down a live session.
+                if (settled == null && state.value.connectionState != AapPodState.ConnectionState.READY) {
+                    log(TAG, Logging.Priority.WARN) {
+                        "Handshake timed out after $handshakeTimeout for ${device.address} — disconnecting"
+                    }
+                    disconnect()
+                }
+            }
        } catch (e: Exception) {
            log(TAG, Logging.Priority.ERROR) { "Connection failed: $e" }
            cleanupSocket()
@@ -118,6 +146,9 @@ internal class AapConnection(
    }

    suspend fun disconnect() = withContext(Dispatchers.IO) {
+        // Idempotent: the handshake watchdog and the manager's DISCONNECTED observer can both reach
+        // here for the same dying session. Run the teardown exactly once.
+        if (!disconnected.compareAndSet(false, true)) return@withContext
        log(TAG, Logging.Priority.INFO) { "Disconnecting" }
        readerJob?.cancel()
        readerJob = null
@@ -261,6 +292,13 @@ internal class AapConnection(
    companion object {
        private const val PSM = 0x1001
        internal val DEFAULT_CONNECT_TIMEOUT = 5.seconds
+
+        /**
+         * Upper bound on the post-handshake wait for the first sign of life (READY). Normal
+         * handshakes complete in well under 2s; 10s tolerates a congested 2.4 GHz band before
+         * giving up so the reconnect path can recover instead of wedging in HANDSHAKING forever.
+         */
+        internal val DEFAULT_HANDSHAKE_TIMEOUT = 10.seconds
        private val TAG = logTag("AAP", "Connection")
    }
 }
@@ -86,6 +86,15 @@ internal class AapSessionEngine(
    private var activeSendRaw: (suspend (AapCommand) -> Unit)? = null
    private var runtimeState = EngineRuntimeState()

+    /**
+     * True once this session has reached [AapPodState.ConnectionState.READY] at least once.
+     * Deliberately NOT cleared by [reset] — consumers read it at disconnect time to tell a
+     * working session that dropped apart from one that never completed the handshake (drives
+     * the reconnect backoff). Cleared only when a fresh session starts.
+     */
+    private var _wasEverReady: Boolean = false
+    val wasEverReady: Boolean get() = _wasEverReady
+
    fun start(scope: CoroutineScope) {
        dispatch(AapEngineEvent.SessionStarted(scope))
    }
@@ -130,6 +139,7 @@ internal class AapSessionEngine(
        when (event) {
            is AapEngineEvent.SessionStarted -> {
                scope = event.scope
+                _wasEverReady = false
                runtimeState = runtimeState.copy(handshakeResponseReceived = false)
                _state.value = _state.value.copy(connectionState = AapPodState.ConnectionState.CONNECTING)
            }
@@ -159,11 +169,14 @@ internal class AapSessionEngine(
    private fun handleConnectResponse(packet: AapPacket.ConnectResponse) {
        if (packet.status != 0) {
            log(TAG, ERROR) {
-                "ConnectResponse failed: status=0x${"%04X".format(packet.status)} " +
+                "ConnectResponse rejected (status=0x${"%04X".format(packet.status)}) — tearing down: " +
                        "major=${packet.major} minor=${packet.minor} " +
                        "features=0x${"%016X".format(packet.features.toLong())}"
            }
-            _state.value = _state.value.copy(connectResponseStatus = packet.status)
+            // Hard protocol rejection: don't wait out the handshake watchdog — disconnect now so
+            // the reconnect path can recover (or back off). Nested dispatch mirrors the existing
+            // inbound-update re-dispatch pattern.
+            dispatch(AapEngineEvent.ResetRequested)
            return
        }

@@ -192,6 +205,7 @@ internal class AapSessionEngine(
            _state.value.connectionState == AapPodState.ConnectionState.HANDSHAKING
        ) {
            runtimeState = runtimeState.copy(handshakeResponseReceived = true)
+            _wasEverReady = true
            _state.value = _state.value.copy(connectionState = AapPodState.ConnectionState.READY)
            log(TAG) { "Connection READY" }
        }