fix: Replace relay waiters without reconnect storms (v2.5.20)

@@ -16,8 +16,8 @@1616         applicationId "com.peachpasswords"1717         minSdk 261818         targetSdk 3519-        versionCode 24011920-        versionName "2.5.19"19+        versionCode 24012020+        versionName "2.5.20"2121 2222         testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"2323         vectorDrawables {

@@ -9,6 +9,7 @@99 import kotlinx.coroutines.Job1010 import kotlinx.coroutines.TimeoutCancellationException1111 import kotlinx.coroutines.channels.Channel12+import kotlinx.coroutines.cancelAndJoin1213 import kotlinx.coroutines.delay1314 import kotlinx.coroutines.isActive1415 import kotlinx.coroutines.launch@@ -31,6 +32,7 @@3132 3233 private const val RECONCILE_INTERVAL_MS = 60_000L3334 private const val RETRY_DELAY_MS = 1_000L35+private const val MAX_RETRY_DELAY_MS = 30_000L3436 /**3537  * Headroom over [SESSION_TIMEOUT_MS] for the part of a dial the coordinator3638  * cannot see: the selector, KK and the two identity frames. Half again the 10s@@ -116,6 +118,11 @@116118     )117119 }118120 121+internal fun nextRelayRetryDelay(current: Long, succeeded: Boolean, dialer: Boolean): Long {122+    val initial = if (dialer) DIAL_RETRY_DELAY_MS else RETRY_DELAY_MS123+    return if (succeeded) initial else (current * 2).coerceAtMost(MAX_RETRY_DELAY_MS)124+}125+119126 internal fun logRelayReconcileInventory(120127     deviceStore: PairedDeviceRegistry,121128 ): RelayReconcileInventory {@@ -164,8 +171,8 @@164171         ourStaticPublicKey: ByteArray,165172         localIdentity: LocalDeviceIdentity,166173         vaultFactory: () -> SyncVaultAdapter?,167-    ) {168-        if (supervisor?.isActive == true) return174+    ): Boolean {175+        if (supervisor?.isActive == true) return false169176         val privateKey = ourStaticPrivateKey.copyOf()170177         val publicKey = ourStaticPublicKey.copyOf()171178         retainedPrivateKey = privateKey@@ -182,8 +189,15 @@182189                         // the relay or network path has stopped routing it.183190                         // Manual sync must replace that socket, not merely184191                         // observe the still-running coroutine and skip it.185-                        watchers.values.forEach(Job::cancel)192+                        val staleWatchers = watchers.values.toList()186193                         watchers.clear()194+                        staleWatchers.forEach { it.cancelAndJoin() }195+                        if (staleWatchers.isNotEmpty()) {196+                            // The relay releases a room after observing the197+                            // disconnect. Do not race the replacement socket198+                            // into the still-occupied capability.199+                            delay(RETRY_DELAY_MS)200+                        }187201                         PeachSyncLog.i("cloud-relay: refreshing relay waiters")188202                     }189203                     reconcile(@@ -200,6 +214,7 @@200214                 watchers.clear()201215             }202216         }217+        return true203218     }204219 205220     /** Re-read the device list at once instead of on the next interval. */@@ -317,8 +332,10 @@317332             if (watchers[key]?.isActive == true) continue318333             val deviceDescription = relayDeviceDescription(target)319334             watchers[key] = scope.launch {335+                var retryDelayMs =336+                    if (target.weDial) DIAL_RETRY_DELAY_MS else RETRY_DELAY_MS320337                 while (isActive) {321-                    runCatching {338+                    val result = runCatching {322339                         val known = deviceStore.get(target.deviceId)323340                             ?: throw IllegalStateException("relay peer was unpaired")324341                         PeachSyncLog.i(@@ -401,7 +418,15 @@401418                             }402419                         }403420                     }404-                    delay(if (target.weDial) DIAL_RETRY_DELAY_MS else RETRY_DELAY_MS)421+                    retryDelayMs = nextRelayRetryDelay(422+                        current = retryDelayMs,423+                        succeeded = result.isSuccess,424+                        dialer = target.weDial,425+                    )426+                    // A relay conflict or rate limit must not become a427+                    // connection storm. Successful sessions reset this to the428+                    // ordinary role-specific retry interval.429+                    delay(retryDelayMs)405430                 }406431             }407432         }@@ -604,7 +629,11 @@604629             override suspend fun receive(): ByteArray = incoming.receive()605630 606631             override fun close(code: Int, reason: String) {632+                // Relay rooms carry no close payload or application state.633+                // Abort after requesting a normal close so a replaced waiter634+                // releases its single-occupant room immediately.607635                 connected.close(code, reason)636+                connected.cancel()608637                 incoming.close()609638             }610639         }

@@ -70,7 +70,7 @@7070             stopSelf()7171             return START_NOT_STICKY7272         }73-        startReconnectListener()73+        val listenerStarted = startReconnectListener()7474         // A start command means something changed — a new pairing, a manual7575         // "sync now", the app coming forward. Re-read the device list now7676         // rather than on the next interval; the listener is already running@@ -78,7 +78,9 @@7878         // Replace existing waiters as well as re-reading the device list. A7979         // locally active coroutine does not prove its idle WebSocket still8080         // routes through the relay, and "sync now" must recover that state.81-        cloudRelayListener.requestReconcile(refreshExisting = true)81+        cloudRelayListener.requestReconcile(82+            refreshExisting = shouldRefreshExistingWaiters(listenerStarted),83+        )8284         // START_STICKY: if the system reclaims us under pressure, come back —8385         // a listener that silently stays down looks identical to a peer that8486         // has simply gone offline, and the user gets no signal either way.@@ -104,9 +106,9 @@104106      * `SyncViewModel.syncNow`, and from a session that reaches a locked vault105107      * mid-handshake.106108      */107-    private fun startReconnectListener() {108-        val master = vault.syncMasterKey() ?: return109-        try {109+    private fun startReconnectListener(): Boolean {110+        val master = vault.syncMasterKey() ?: return false111+        return try {110112             val keys = deriveLocalSyncKeypair(master, getOrCreateDeviceId(this))111113             val identity = buildLocalIdentity(this, master)112114             cloudRelayListener.start(scope, keys.secretKey, keys.publicKey, identity) {@@ -140,6 +142,8 @@140142         }141143     }142144 }145+146+internal fun shouldRefreshExistingWaiters(listenerStarted: Boolean): Boolean = !listenerStarted143147 144148 internal const val SYNC_NOTIFICATION_CHANNEL_ID = "peach_sync"145149 internal const val SYNC_NOTIFICATION_ID = 4201

@@ -70,6 +70,14 @@7070     }7171 7272     @Test73+    fun `relay retry backs off after failures and resets after success`() {74+        assertEquals(2_000L, nextRelayRetryDelay(1_000L, succeeded = false, dialer = false))75+        assertEquals(30_000L, nextRelayRetryDelay(20_000L, succeeded = false, dialer = false))76+        assertEquals(1_000L, nextRelayRetryDelay(30_000L, succeeded = true, dialer = false))77+        assertEquals(5_000L, nextRelayRetryDelay(30_000L, succeeded = true, dialer = true))78+    }79+80+    @Test7381     fun `reconcile logs paired and eligible counts when no peer passes the filter`() = runBlocking {7482         val devices = listOf(7583             PairedDevice(

@@ -60,4 +60,9 @@6060         assertTrue(ran)6161         assertTrue(reported.isEmpty())6262     }63+64+    @Test fun `only a nudge to an existing listener replaces its waiters`() {65+        assertEquals(false, shouldRefreshExistingWaiters(listenerStarted = true))66+        assertEquals(true, shouldRefreshExistingWaiters(listenerStarted = false))67+    }6368 }