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 detect silently dropped connections and 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 detect silently dropped connections and 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 detect silently dropped connections and 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/WebSocketCore.kt

@@ -51,6 +51,17 @@ internal interface WebSocketCore {
      */
     suspend fun sendBytes(data: ByteArray): Boolean
 
+    /**
+     * Sends an application-level ping and verifies that the connection is alive.
+     *
+     * If no pong is received while the connection is believed to be established, the connection is
+     * cancelled: this surfaces a silently dropped socket (e.g. the server restarted without the
+     * TCP connection being reset) to the close handling, which restores active subscriptions.
+     *
+     * @return `true` if a pong was received, `false` otherwise.
+     */
+    suspend fun ping(): Boolean
+
     /**
      * Start a subscription for events on the websocket connection and get a Flow for listening to
      * new messages. When there are no more listeners, the subscription will automatically be cancelled

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

@@ -57,6 +57,7 @@ import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.cancellation.CancellationException
 import kotlin.time.Duration
+import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineExceptionHandler
@@ -99,6 +100,9 @@ import timber.log.Timber
 
 private val DELAY_BEFORE_RECONNECT = 10.seconds
 
+/** Upper bound for the backoff between reconnection attempts while the server stays unreachable. */
+private val MAX_DELAY_BEFORE_RECONNECT = 2.minutes
+
 /**
  * Implementation of the [WebSocketCore] interface for managing WebSocket connections to a Home Assistant server.
  *
@@ -182,6 +186,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.
      *
@@ -622,6 +629,26 @@ internal class WebSocketCoreImpl(
         }
     }
 
+    override suspend fun ping(): Boolean {
+        // Capture the connection this ping verifies, so a connection (re)established while
+        // waiting for the pong is never cancelled by mistake.
+        val holder = connectionHolder.get()
+        val response = sendMessage(mapOf("type" to "ping"))
+        if (response is PongSocketResponse) return true
+
+        if (holder != null && connectionHolder.get() === holder) {
+            // No pong on a connection that is believed to be established means the socket is
+            // half-open: the server went away without the TCP connection being reset, for
+            // example when it restarted and its address moved. OkHttp keeps buffering writes
+            // into such a socket without ever failing, so no close callback would fire on its
+            // own. Cancelling forces onFailure, which runs handleClosingSocket and lets the
+            // restore loop bring the connection and its subscriptions back.
+            Timber.w("No pong received on the established connection, cancelling it to trigger reconnection")
+            holder.webSocket.cancel()
+        }
+        return false
+    }
+
     override suspend fun <T : Any> subscribeTo(
         type: String,
         data: Map<String, Any?>,
@@ -765,16 +792,18 @@ internal class WebSocketCoreImpl(
             awaitClose {
                 wsScope.launch {
                     eventSubscriptionMutex.withLock {
-                        findSubscription(subscribeMessage)
-                            ?.let {
-                                val subscription = it.key
-                                Timber.d("Unsubscribing from $subscribeMessage")
-                                // Unsubscribe must happen before removing from activeMessages to ensure
-                                // the server acknowledges before we stop handling events for this subscription
-                                unsubscribeEvents(subscription)
-                                channel.close()
-                                activeMessages.remove(subscription)
-                            }
+                        // Resubscribing briefly tracks two entries for the same message, remove
+                        // them all so no orphan keeps the connection alive
+                        var subscription = findSubscription(subscribeMessage)?.key
+                        while (subscription != null) {
+                            Timber.d("Unsubscribing from $subscribeMessage")
+                            // Unsubscribe must happen before removing from activeMessages to ensure
+                            // the server acknowledges before we stop handling events for this subscription
+                            unsubscribeEvents(subscription)
+                            activeMessages.remove(subscription)
+                            subscription = findSubscription(subscribeMessage)?.key
+                        }
+                        channel.close()
                     }
                     if (activeMessages.isEmpty()) {
                         Timber.i("No more subscriptions, closing connection.")
@@ -1022,52 +1051,82 @@ internal class WebSocketCoreImpl(
         if (connectionHolder.get() == null) return
 
         wsScope.launch {
-            val hasSubscriptions: Boolean
-
             connectedMutex.withLock {
-                val holder = connectionHolder.getAndSet(null)
+                // Another callback of the same closing already handled it
+                val holder = connectionHolder.getAndSet(null) ?: return@launch
                 // Cancel URL observer - connect() will recreate it if needed
-                holder?.urlObserverJob?.cancel()
+                holder.urlObserverJob.cancel()
 
                 cleanupClosingSocket()
-                hasSubscriptions = activeMessages.any { it.value is ActiveMessage.Subscription }
-            }
-
-            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)
+                val hasSubscriptions = activeMessages.any { it.value is ActiveMessage.Subscription }
+
+                if (hasSubscriptions && wasActive && wsScope.isActive) {
+                    // A new closing takes over the restore loop with a fresh snapshot, replacing
+                    // the job under the lock so concurrent closings cannot start two loops
+                    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
+        var retryDelay = DELAY_BEFORE_RECONNECT
+        while (toRestore.isNotEmpty()) {
+            if (connect()) {
+                // The server is reachable again, restart the backoff
+                retryDelay = DELAY_BEFORE_RECONNECT
+                toRestore = resubscribeActiveSubscriptions(toRestore)
+                if (toRestore.isEmpty()) return
+                Timber.w("${toRestore.size} subscriptions not restored, retrying in $retryDelay")
+            } 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 $retryDelay to resubscribe to active subscriptions",
+                )
+            }
+            delay(retryDelay)
+            // Back off while the server stays unreachable so a long outage is not hammered
+            retryDelay = (retryDelay * 2).coerceAtMost(MAX_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/main/kotlin/io/homeassistant/companion/android/common/data/websocket/impl/WebSocketRepositoryImpl.kt

@@ -34,7 +34,6 @@ import io.homeassistant.companion.android.common.data.websocket.impl.entities.En
 import io.homeassistant.companion.android.common.data.websocket.impl.entities.GetConfigResponse
 import io.homeassistant.companion.android.common.data.websocket.impl.entities.GetTodosResponse
 import io.homeassistant.companion.android.common.data.websocket.impl.entities.MatterCommissionResponse
-import io.homeassistant.companion.android.common.data.websocket.impl.entities.PongSocketResponse
 import io.homeassistant.companion.android.common.data.websocket.impl.entities.RawMessageSocketResponse
 import io.homeassistant.companion.android.common.data.websocket.impl.entities.StateChangedEvent
 import io.homeassistant.companion.android.common.data.websocket.impl.entities.TemplateUpdatedEvent
@@ -69,14 +68,7 @@ class WebSocketRepositoryImpl internal constructor(
         return webSocketCore.shutdown()
     }
 
-    override suspend fun sendPing(): Boolean {
-        val socketResponse = webSocketCore.sendMessage(
-            mapOf(
-                "type" to "ping",
-            ),
-        )
-        return socketResponse is PongSocketResponse
-    }
+    override suspend fun sendPing(): Boolean = webSocketCore.ping()
 
     override suspend fun getConfig(): GetConfigResponse? {
         val socketResponse = webSocketCore.sendMessage(