diff --git a/app/src/main/java/eu/darken/capod/monitor/core/aap/AapAutoConnect.kt b/app/src/main/java/eu/darken/capod/monitor/core/aap/AapAutoConnect.kt index be203176..8c75b002 100644 --- a/app/src/main/java/eu/darken/capod/monitor/core/aap/AapAutoConnect.kt +++ b/app/src/main/java/eu/darken/capod/monitor/core/aap/AapAutoConnect.kt @@ -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()) private val processedModelCorrections = java.util.Collections.synchronizedSet(mutableSetOf()) + /** + * 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() + + /** 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 = 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 = 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 = channelFlow { + aapManager.disconnectEvents.collect { event -> + launch { handleReconnect(event) } } - .map { } // SharedFlow → Flow - .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 = 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) } } diff --git a/app/src/main/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManager.kt b/app/src/main/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManager.kt index 8f1ecc19..db8a52f2 100644 --- a/app/src/main/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManager.kt +++ b/app/src/main/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManager.kt @@ -54,8 +54,8 @@ class AapConnectionManager @Inject constructor( private val _allStates = MutableStateFlow>(emptyMap()) val allStates: StateFlow> = _allStates.asStateFlow() - private val _disconnectEvents = MutableSharedFlow(extraBufferCapacity = 16) - val disconnectEvents: SharedFlow = _disconnectEvents.asSharedFlow() + private val _disconnectEvents = MutableSharedFlow(extraBufferCapacity = 16) + val disconnectEvents: SharedFlow = _disconnectEvents.asSharedFlow() /** Emits when a connection receives private keys (IRK/ENC) from the device. */ private val _keysReceived = MutableSharedFlow>(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, +) diff --git a/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapConnection.kt b/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapConnection.kt index 6e2d73eb..c38c68e2 100644 --- a/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapConnection.kt +++ b/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapConnection.kt @@ -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 get() = engine.state val keysReceived: SharedFlow get() = engine.keysReceived val stemPressEvents: SharedFlow 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") } } diff --git a/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngine.kt b/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngine.kt index 8ff4482a..beb7641f 100644 --- a/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngine.kt +++ b/app/src/main/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngine.kt @@ -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" } } diff --git a/app/src/test/java/eu/darken/capod/monitor/core/aap/AapAutoConnectTest.kt b/app/src/test/java/eu/darken/capod/monitor/core/aap/AapAutoConnectTest.kt index 46c654c1..06b71d71 100644 --- a/app/src/test/java/eu/darken/capod/monitor/core/aap/AapAutoConnectTest.kt +++ b/app/src/test/java/eu/darken/capod/monitor/core/aap/AapAutoConnectTest.kt @@ -5,6 +5,7 @@ import eu.darken.capod.common.bluetooth.BluetoothManager2 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.pods.core.apple.aap.protocol.AapDeviceInfo import eu.darken.capod.pods.core.apple.ble.BlePodSnapshot @@ -22,7 +23,9 @@ import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.flow.toList import kotlinx.coroutines.launch import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.advanceTimeBy import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Nested @@ -40,7 +43,7 @@ class AapAutoConnectTest : BaseTest() { private lateinit var profilesFlow: MutableStateFlow> private lateinit var connectedDevicesFlow: MutableStateFlow> - private lateinit var disconnectEventsFlow: MutableSharedFlow + private lateinit var disconnectEventsFlow: MutableSharedFlow private lateinit var allStatesFlow: MutableStateFlow> private val testAddress = "AA:BB:CC:DD:EE:FF" @@ -351,7 +354,7 @@ class AapAutoConnectTest : BaseTest() { // Clear profiles, then disconnect profilesFlow.value = emptyList() allStatesFlow.value = emptyMap() - disconnectEventsFlow.tryEmit(testAddress) + disconnectEventsFlow.tryEmit(AapDisconnectEvent(testAddress, wasEverReady = true)) advanceUntilIdle() // The reconnect loop should not call connect since profile is gone @@ -369,7 +372,7 @@ class AapAutoConnectTest : BaseTest() { // Remove bonded device, then disconnect every { bluetoothManager.bondedDevices() } returns flowOf(emptySet()) allStatesFlow.value = emptyMap() - disconnectEventsFlow.tryEmit(testAddress) + disconnectEventsFlow.tryEmit(AapDisconnectEvent(testAddress, wasEverReady = true)) advanceUntilIdle() // Reconnect should not call connect since not bonded @@ -387,7 +390,7 @@ class AapAutoConnectTest : BaseTest() { // Remove classic BT connection, then disconnect connectedDevicesFlow.value = emptyList() allStatesFlow.value = emptyMap() - disconnectEventsFlow.tryEmit(testAddress) + disconnectEventsFlow.tryEmit(AapDisconnectEvent(testAddress, wasEverReady = true)) advanceUntilIdle() // Reconnect should not call connect since not classically connected @@ -405,7 +408,7 @@ class AapAutoConnectTest : BaseTest() { // Remove from BLE scans, then disconnect every { blePodMonitor.devices } returns flowOf(emptyList()) allStatesFlow.value = emptyMap() - disconnectEventsFlow.tryEmit(testAddress) + disconnectEventsFlow.tryEmit(AapDisconnectEvent(testAddress, wasEverReady = true)) advanceUntilIdle() // Reconnect should not call connect since not visible in BLE @@ -422,7 +425,7 @@ class AapAutoConnectTest : BaseTest() { // Keep as READY, emit disconnect event // allStatesFlow still shows READY → reconnect should skip - disconnectEventsFlow.tryEmit(testAddress) + disconnectEventsFlow.tryEmit(AapDisconnectEvent(testAddress, wasEverReady = true)) advanceUntilIdle() // Should not attempt connect — already connected @@ -430,6 +433,44 @@ class AapAutoConnectTest : BaseTest() { job.cancel() } + + @Test + fun `was-ready disconnect reconnects without extra cooldown`() = runTest(testDispatcher) { + val autoConnect = createAutoConnect() + val job = setupForReconnect(autoConnect) + advanceUntilIdle() + allStatesFlow.value = emptyMap() + + // A session that reached READY drops: only the normal first retry delay (3s), no backoff cooldown. + disconnectEventsFlow.tryEmit(AapDisconnectEvent(testAddress, wasEverReady = true)) + advanceTimeBy(3_100) + runCurrent() + + coVerify(exactly = 1) { aapManager.connect(testAddress, any(), any()) } + + job.cancel() + } + + @Test + fun `never-ready disconnect backs off before reconnecting`() = runTest(testDispatcher) { + val autoConnect = createAutoConnect() + val job = setupForReconnect(autoConnect) + advanceUntilIdle() + allStatesFlow.value = emptyMap() + + // A session that never reached READY (failed handshake): first failure adds a 5s cooldown + // on top of the 3s retry delay, so nothing should connect within the 3.1s a was-ready drop would. + disconnectEventsFlow.tryEmit(AapDisconnectEvent(testAddress, wasEverReady = false)) + advanceTimeBy(3_100) + runCurrent() + coVerify(exactly = 0) { aapManager.connect(testAddress, any(), any()) } + + // After the 5s cooldown + 3s retry delay elapse, the reconnect fires. + advanceUntilIdle() + coVerify(exactly = 1) { aapManager.connect(testAddress, any(), any()) } + + job.cancel() + } } @Nested diff --git a/app/src/test/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManagerTest.kt b/app/src/test/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManagerTest.kt index 5d071acb..6920f5af 100644 --- a/app/src/test/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManagerTest.kt +++ b/app/src/test/java/eu/darken/capod/pods/core/apple/aap/AapConnectionManagerTest.kt @@ -12,7 +12,9 @@ import eu.darken.capod.pods.core.apple.aap.protocol.AapSetting import io.kotest.assertions.throwables.shouldThrow import io.kotest.matchers.maps.shouldBeEmpty import io.kotest.matchers.shouldBe +import io.mockk.Runs import io.mockk.every +import io.mockk.just import io.mockk.mockk import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.TimeoutCancellationException @@ -27,6 +29,7 @@ import testhelpers.TestTimeSource import java.io.ByteArrayInputStream import java.io.ByteArrayOutputStream import java.io.IOException +import java.io.InputStream import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger @@ -135,6 +138,53 @@ class AapConnectionManagerTest : BaseTest() { advanceUntilIdle() } + @Test + fun `handshake timeout disconnects a silent session`() = testScope.runTest { + // Socket connects and the handshake is sent, but the peer never replies: read() blocks + // forever. The handshake watchdog must time out and tear the session down. + val readBlocked = CountDownLatch(1) + val closeCalls = AtomicInteger(0) + val blockingInput = object : InputStream() { + override fun read(): Int { + readBlocked.await(2, TimeUnit.SECONDS) + return -1 + } + + override fun read(b: ByteArray): Int { + readBlocked.await(2, TimeUnit.SECONDS) + return -1 + } + } + val silentSocket = mockk(relaxed = true) { + every { connect() } just Runs + every { outputStream } returns ByteArrayOutputStream() + every { inputStream } returns blockingInput + every { close() } answers { + closeCalls.incrementAndGet() + readBlocked.countDown() + } + } + every { socketFactory.createSocket(any(), any()) } returns silentSocket + val connection = AapConnection( + device = testDevice, + profile = AapDeviceProfile.forModel(PodModel.AIRPODS_PRO3), + socketFactory = socketFactory, + timeSource = timeSource, + connectTimeout = 50.milliseconds, + handshakeTimeout = 100.milliseconds, + ) + + connection.connect(testScope) + // Fire the 100ms handshake watchdog. + advanceUntilIdle() + + // disconnect() runs engine.reset() before closing the socket, so awaiting the close latch + // guarantees the state has already flipped to DISCONNECTED. + readBlocked.await(2, TimeUnit.SECONDS) shouldBe true + connection.state.value.connectionState shouldBe AapPodState.ConnectionState.DISCONNECTED + closeCalls.get() shouldBe 1 + } + @Test fun `remote disconnect cleans up allStates`() = testScope.runTest { // Empty inputStream → readLoop gets -1 immediately → DISCONNECTED diff --git a/app/src/test/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngineTest.kt b/app/src/test/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngineTest.kt index 279be8ac..0e8e83f4 100644 --- a/app/src/test/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngineTest.kt +++ b/app/src/test/java/eu/darken/capod/pods/core/apple/aap/engine/AapSessionEngineTest.kt @@ -114,6 +114,38 @@ class AapSessionEngineTest : BaseTest() { engine.state.value.connectionState shouldBe AapPodState.ConnectionState.DISCONNECTED } + @Test + fun `wasEverReady is false until READY then true`() { + val engine = createEngine() + val scope = TestScope(UnconfinedTestDispatcher()) + + engine.start(scope) + engine.wasEverReady shouldBe false + engine.onHandshakeSent() + engine.wasEverReady shouldBe false + + // First non-CONTROL message during HANDSHAKING → READY + engine.processMessage(dummyMessage(commandType = 0x0002)) + engine.wasEverReady shouldBe true + } + + @Test + fun `wasEverReady survives reset`() { + val engine = createEngine() + val scope = TestScope(UnconfinedTestDispatcher()) + + engine.start(scope) + engine.onHandshakeSent() + engine.processMessage(dummyMessage(commandType = 0x0002)) + engine.wasEverReady shouldBe true + + engine.reset() + + engine.state.value.connectionState shouldBe AapPodState.ConnectionState.DISCONNECTED + // Consumers read this at disconnect time — reset must not clear it. + engine.wasEverReady shouldBe true + } + @Test fun `reset is idempotent`() { val engine = createEngine() @@ -220,12 +252,13 @@ class AapSessionEngineTest : BaseTest() { } @Test - fun `Connect Response with non-zero status does not store features`() { + fun `Connect Response with non-zero status tears down the session`() { val engine = createEngine() val scope = TestScope(UnconfinedTestDispatcher()) engine.start(scope) engine.onHandshakeSent() + engine.state.value.connectionState shouldBe AapPodState.ConnectionState.HANDSHAKING val packet = AapPacket.ConnectResponse( raw = ByteArray(18), @@ -237,9 +270,9 @@ class AapSessionEngineTest : BaseTest() { ) engine.processConnectResponse(packet) - engine.state.value.connectResponseStatus shouldBe 0x0001 + // Hard protocol rejection: fast-fail to DISCONNECTED instead of waiting out the watchdog. + engine.state.value.connectionState shouldBe AapPodState.ConnectionState.DISCONNECTED engine.state.value.negotiatedFeatures.shouldBeNull() - engine.state.value.connectionState shouldBe AapPodState.ConnectionState.HANDSHAKING } }