Detect silently dropped WebSocket connections and retry until subscriptions are restored

app/src/main/kotlin/io/homeassistant/companion/android/websocket/WebsocketManager.kt

@@ -137,7 +137,7 @@ class WebsocketManager(appContext: Context, workerParams: WorkerParameters) :
         // play ping pong to ensure we have a connection and server changes are handled.
         do {
             delay(30000)
-        } while (jobs.values.any { it.isActive } && isActive && shouldWeRun() && manageServerJobs(jobs, this))
+        } while (isActive && shouldWeRun() && manageServerJobs(jobs, this))
 
         jobs.forEach { it.value.cancel() }
         jobs.clear()
@@ -176,9 +176,9 @@ class WebsocketManager(appContext: Context, workerParams: WorkerParameters) :
     private suspend fun manageServerJobs(jobs: MutableMap<Int, Job>, coroutineScope: CoroutineScope): Boolean {
         val servers = serverManager.servers()
 
-        // Clean up...
-        jobs.filter { (serverId, _) ->
-            servers.none { it.id == serverId } || !shouldRunForServer(serverId)
+        // Clean up, including stopped jobs so they are started again below...
+        jobs.filter { (serverId, job) ->
+            servers.none { it.id == serverId } || !job.isActive || !shouldRunForServer(serverId)
         }
             .forEach { (serverId, job) ->
                 job.cancel()

app/src/main/res/xml/changelog_master.xml

@@ -3,14 +3,17 @@
     tools:ignore="MissingDefaultResource">
     <release version="2026.7.2 - Main" versioncode="3">
         <change>Android 17: added assistant volume level sensor + notification command for control</change>
+        <change>Local push notifications now keep retrying to reconnect while the server is unreachable, like during a server restart, instead of staying disconnected</change>
         <change>Bug fixes and dependency updates</change>
     </release>
     <release version="2026.7.2 - Wear" versioncode="2">
         <change>Android 17: added assistant volume level sensor</change>
+        <change>Local push notifications now keep retrying to reconnect while the server is unreachable, like during a server restart, instead of staying disconnected</change>
         <change>Bug fixes and dependency updates</change>
     </release>
     <release version="2026.7.2 - Automotive" versioncode="1">
         <change>Android 17: added assistant volume level sensor + notification command for control</change>
+        <change>Local push notifications now keep retrying to reconnect while the server is unreachable, like during a server restart, instead of staying disconnected</change>
         <change>Bug fixes and dependency updates</change>
     </release>
 </changelog>

common/src/main/kotlin/io/homeassistant/companion/android/common/data/websocket/impl/WebSocketCoreImpl.kt

@@ -182,6 +182,9 @@ internal class WebSocketCoreImpl(
     @Volatile
     private var pendingCloseReason: WebSocketState.Closed.Reason? = null
 
+    /** The running subscription restore loop, replaced on each closing so only one runs at a time. */
+    private var reconnectJob: Job? = null
+
     /**
      * A [CompletableDeferred] that signals the establishment and authentication of a WebSocket connection.
      *
@@ -1036,38 +1039,65 @@ internal class WebSocketCoreImpl(
             val shouldAttemptReconnect = hasSubscriptions && wasActive
 
             if (shouldAttemptReconnect && wsScope.isActive) {
-                // Delay before reconnect unless URL changed
-                if (closeReason != WebSocketState.Closed.Reason.CHANGED_URL) {
-                    delay(DELAY_BEFORE_RECONNECT)
+                // A new closing takes over the restore loop with a fresh snapshot
+                reconnectJob?.cancel()
+                reconnectJob = wsScope.launch {
+                    // Delay before reconnect unless URL changed
+                    if (closeReason != WebSocketState.Closed.Reason.CHANGED_URL) {
+                        delay(DELAY_BEFORE_RECONNECT)
+                    }
+                    reconnectSubscriptions()
                 }
-                reconnectSubscriptions()
             }
         }
     }
 
     private suspend fun reconnectSubscriptions() {
-        if (connect()) {
-            Timber.d("Resubscribing to active subscriptions...")
-            activeMessages.filterValues { it is ActiveMessage.Subscription }.entries
-                .forEach { (oldId, oldActiveMessage) ->
-                    oldActiveMessage as ActiveMessage.Subscription
-                    activeMessages.remove(oldId)
-
-                    val response = sendMessage(
-                        Command.WithAnswer.Subscription(
-                            request = oldActiveMessage.request,
-                            eventFlow = oldActiveMessage.eventFlow,
-                            onEvent = oldActiveMessage.onEvent,
-                        ),
-                    )
-                    if (response == null || response.success != true) {
-                        Timber.e("Issue re-registering subscription with ${oldActiveMessage.request}")
-                    }
+        var toRestore: Set<Long> = activeMessages.filterValues { it is ActiveMessage.Subscription }.keys
+        while (toRestore.isNotEmpty()) {
+            if (connect()) {
+                toRestore = resubscribeActiveSubscriptions(toRestore)
+                if (toRestore.isEmpty()) return
+                Timber.w("${toRestore.size} subscriptions not restored, retrying in $DELAY_BEFORE_RECONNECT")
+            } else {
+                if (getConnectionState() == WebSocketState.ClosedAuth) {
+                    Timber.e("Authentication failed, not retrying to resubscribe to active subscriptions")
+                    return
                 }
-        } else {
-            // TODO https://github.com/home-assistant/android/issues/5259 handle re-connection gracefully or terminates the flows
-            Timber.w("Unable to reconnect, cannot resubscribe to active subscriptions")
+                Timber.w(
+                    "Unable to reconnect, retrying in $DELAY_BEFORE_RECONNECT to resubscribe to active subscriptions",
+                )
+            }
+            delay(DELAY_BEFORE_RECONNECT)
+            // Subscriptions closed while waiting no longer need to be restored
+            toRestore = toRestore.filterTo(mutableSetOf()) { activeMessages[it] is ActiveMessage.Subscription }
+        }
+    }
+
+    /** @return the ids of the subscriptions that could not be restored and are kept for a retry */
+    private suspend fun resubscribeActiveSubscriptions(oldIds: Set<Long>): Set<Long> {
+        Timber.d("Resubscribing to active subscriptions...")
+        val failed = mutableSetOf<Long>()
+        oldIds.forEach { oldId ->
+            val oldActiveMessage = activeMessages[oldId] as? ActiveMessage.Subscription ?: return@forEach
+
+            val response = sendMessage(
+                Command.WithAnswer.Subscription(
+                    request = oldActiveMessage.request,
+                    eventFlow = oldActiveMessage.eventFlow,
+                    onEvent = oldActiveMessage.onEvent,
+                ),
+            )
+            if (response == null || response.success != true) {
+                failed += oldId
+                // Drop the rejected attempt, the kept original is retried instead
+                response?.id?.let { activeMessages.remove(it) }
+                Timber.e("Issue re-registering subscription with ${oldActiveMessage.request}")
+            } else {
+                activeMessages.remove(oldId)
+            }
         }
+        return failed
     }
 
     private fun URL.toWebSocketURL(): String {

common/src/test/kotlin/io/homeassistant/companion/android/common/data/websocket/impl/WebSocketCoreImplTest.kt

@@ -1445,6 +1445,191 @@ misc
         }
     }
 
+    @Test
+    fun `Given an Active subscription When the server stays unavailable Then it retries until the server is back and resubscribes`() = runTest {
+        setupServer(backgroundScope = backgroundScope)
+        prepareAuthenticationAnswer()
+        assertTrue(webSocketCore.connect())
+
+        mockResultSuccessForId(2)
+        val subscription = checkNotNull(
+            webSocketCore.subscribeTo<StateChangedEvent>(
+                SUBSCRIBE_TYPE_SUBSCRIBE_EVENTS,
+                mapOf("event_type" to "state_changed"),
+            ),
+        )
+
+        subscription.test {
+            // The server becomes unavailable: the auth message cannot be sent anymore
+            every { mockConnection.send(match<String> { it.contains(""""type":"auth"""") }) } returns false
+            var connectionAttemptCount = 0
+            every { mockOkHttpClient.newWebSocket(any(), any()) } answers {
+                connectionAttemptCount++
+                mockConnection
+            }
+            closeConnection()
+
+            // The first attempt after the delay fails, the subscription is kept and retries continue
+            advanceTimeBy(11.seconds)
+            runCurrent()
+            assertTrue(connectionAttemptCount >= 1, "Should have attempted to reconnect")
+            advanceTimeBy(30.seconds)
+            runCurrent()
+            assertTrue(connectionAttemptCount >= 3, "Should keep retrying while the server is unavailable")
+            assertTrue(
+                webSocketCore.activeMessages.any { it.value is ActiveMessage.Subscription },
+                "Subscription should still be tracked while retrying",
+            )
+
+            // The server is back, the next retry reconnects and resubscribes with a new ID
+            prepareAuthenticationAnswer()
+            var resubscribeId: Long? = null
+            every {
+                mockConnection.send(match<String> { it.contains(""""type":"$SUBSCRIBE_TYPE_SUBSCRIBE_EVENTS"""") })
+            } answers {
+                val id = checkNotNull(Regex(""""id":(\d+)""").find(firstArg<String>())?.groupValues?.get(1)?.toLong())
+                resubscribeId = id
+                webSocketListener.onMessage(
+                    mockConnection,
+                    """{"id":$id,"type":"result","success":true,"result":{}}""",
+                )
+                true
+            }
+            advanceTimeBy(11.seconds)
+            runCurrent()
+
+            val newId = checkNotNull(resubscribeId) { "Subscription should have been re-registered" }
+            assertNotEquals(2L, newId)
+            webSocketListener.onMessage(
+                mockConnection,
+                """{"id":$newId, "type":"event", "event":{"event_type":"state_changed", "time_fired":"2016-11-26T01:37:24.265429+00:00", "data": {"entity_id":"light.bed_light"}}}""",
+            )
+            assertEquals("light.bed_light", awaitItem().entityId)
+        }
+    }
+
+    @Test
+    fun `Given a resubscription rejected on a live connection Then it is retried until restored`() = runTest {
+        setupServer(backgroundScope = backgroundScope)
+        prepareAuthenticationAnswer()
+        assertTrue(webSocketCore.connect())
+
+        mockResultSuccessForId(2)
+        val subscription = checkNotNull(
+            webSocketCore.subscribeTo<StateChangedEvent>(
+                SUBSCRIBE_TYPE_SUBSCRIBE_EVENTS,
+                mapOf("event_type" to "state_changed"),
+            ),
+        )
+
+        subscription.test {
+            // The server rejects the first resubscription on the restored connection
+            var subscribeAttempts = 0
+            var lastSubscribeId: Long? = null
+            every {
+                mockConnection.send(match<String> { it.contains(""""type":"$SUBSCRIBE_TYPE_SUBSCRIBE_EVENTS"""") })
+            } answers {
+                val id = checkNotNull(Regex(""""id":(\d+)""").find(firstArg<String>())?.groupValues?.get(1)?.toLong())
+                subscribeAttempts++
+                lastSubscribeId = id
+                webSocketListener.onMessage(
+                    mockConnection,
+                    """{"id":$id,"type":"result","success":${subscribeAttempts > 1},"result":{}}""",
+                )
+                true
+            }
+            closeConnection()
+
+            advanceTimeBy(11.seconds)
+            runCurrent()
+            assertEquals(1, subscribeAttempts, "First resubscription should have been attempted")
+            assertEquals(
+                1,
+                webSocketCore.activeMessages.count { it.value is ActiveMessage.Subscription },
+                "The rejected attempt should be dropped and the original kept for a retry",
+            )
+
+            // The retry happens on the live connection without another disconnection
+            advanceTimeBy(11.seconds)
+            runCurrent()
+            assertEquals(2, subscribeAttempts, "The rejected subscription should have been retried")
+
+            val newId = checkNotNull(lastSubscribeId)
+            webSocketListener.onMessage(
+                mockConnection,
+                """{"id":$newId, "type":"event", "event":{"event_type":"state_changed", "time_fired":"2016-11-26T01:37:24.265429+00:00", "data": {"entity_id":"light.bed_light"}}}""",
+            )
+            assertEquals("light.bed_light", awaitItem().entityId)
+            assertEquals(
+                1,
+                webSocketCore.activeMessages.count { it.value is ActiveMessage.Subscription },
+                "Only the restored subscription should remain",
+            )
+        }
+    }
+
+    @Test
+    fun `Given a resubscription without acknowledgement When reconnecting again Then the subscription is restored`() = runTest {
+        setupServer(backgroundScope = backgroundScope)
+        prepareAuthenticationAnswer()
+        assertTrue(webSocketCore.connect())
+
+        mockResultSuccessForId(2)
+        val subscription = checkNotNull(
+            webSocketCore.subscribeTo<StateChangedEvent>(
+                SUBSCRIBE_TYPE_SUBSCRIBE_EVENTS,
+                mapOf("event_type" to "state_changed"),
+            ),
+        )
+
+        subscription.test {
+            // First reconnection: the subscribe request is sent but never acknowledged
+            every {
+                mockConnection.send(match<String> { it.contains(""""type":"$SUBSCRIBE_TYPE_SUBSCRIBE_EVENTS"""") })
+            } returns true
+            closeConnection()
+            advanceTimeBy(11.seconds)
+            runCurrent()
+
+            // Once the answer times out the subscription must be kept for the next reconnection
+            advanceTimeBy(31.seconds)
+            runCurrent()
+            assertTrue(
+                webSocketCore.activeMessages.any { it.value is ActiveMessage.Subscription },
+                "Subscription should be kept when the resubscription is not acknowledged",
+            )
+
+            // Second reconnection: the subscribe request is acknowledged and events flow again
+            var resubscribeId: Long? = null
+            every {
+                mockConnection.send(match<String> { it.contains(""""type":"$SUBSCRIBE_TYPE_SUBSCRIBE_EVENTS"""") })
+            } answers {
+                val id = checkNotNull(Regex(""""id":(\d+)""").find(firstArg<String>())?.groupValues?.get(1)?.toLong())
+                resubscribeId = id
+                webSocketListener.onMessage(
+                    mockConnection,
+                    """{"id":$id,"type":"result","success":true,"result":{}}""",
+                )
+                true
+            }
+            closeConnection()
+            advanceTimeBy(11.seconds)
+            runCurrent()
+
+            val newId = checkNotNull(resubscribeId) { "Subscription should have been re-registered" }
+            webSocketListener.onMessage(
+                mockConnection,
+                """{"id":$newId, "type":"event", "event":{"event_type":"state_changed", "time_fired":"2016-11-26T01:37:24.265429+00:00", "data": {"entity_id":"light.bed_light"}}}""",
+            )
+            assertEquals("light.bed_light", awaitItem().entityId)
+            assertEquals(
+                1,
+                webSocketCore.activeMessages.count { it.value is ActiveMessage.Subscription },
+                "Only the acknowledged subscription should remain",
+            )
+        }
+    }
+
     @Test
     fun `Given pending simple messages When connection closes during URL change Then they complete with exception`() = runTest {
         val urlFlow = MutableStateFlow<UrlState>(UrlState.HasUrl("https://io.ha".toHttpUrlOrNull()?.toUrl()))