From d1dac57ccb427bd52d65918fc5811f55a9e7f254 Mon Sep 17 00:00:00 2001 From: AlexDev_ <56083016+alexdev03@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:09:22 +0200 Subject: [PATCH] fix(ws): reconnect after network handover (#808) --- .../data/remote/NetworkMonitor.kt | 74 +++++- .../hermescontrol/data/ws/HermesWsClient.kt | 195 +++++++++----- .../data/remote/NetworkMonitorTest.kt | 97 +++++++ .../data/ws/HermesWsClientTest.kt | 243 ++++++++++++++++++ 4 files changed, 536 insertions(+), 73 deletions(-) create mode 100644 app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt diff --git a/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt b/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt index 714a0f05..c0326e7b 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/remote/NetworkMonitor.kt @@ -3,27 +3,79 @@ package com.m57.hermescontrol.data.remote import android.content.Context import android.net.ConnectivityManager import android.net.Network +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.asStateFlow +import java.util.concurrent.atomic.AtomicInteger object NetworkMonitor { private val _isConnected = MutableStateFlow(true) // optimistic default val isConnected: StateFlow = _isConnected.asStateFlow() + // Unlike StateFlow, this queues every break-before-make old -> null -> new transition. + private val networkChangeEvents = MutableSharedFlow(extraBufferCapacity = Int.MAX_VALUE) + internal val networkChanges: SharedFlow = networkChangeEvents.asSharedFlow() + + private val callbackLock = Any() + private val callbackGeneration = AtomicInteger(0) + private var callbackRegistration: Pair? = null + + @Volatile + private var defaultNetwork: Network? = null + fun init(context: Context) { val cm = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager - _isConnected.value = cm.activeNetwork != null - cm.registerDefaultNetworkCallback( - object : ConnectivityManager.NetworkCallback() { - override fun onAvailable(network: Network) { - _isConnected.value = true - } + val callback = createCallback(cm) + var networkChanged = false + var networkAvailable = false + val previous = + synchronized(callbackLock) { + val activeNetwork = cm.activeNetwork + networkChanged = defaultNetwork != activeNetwork + defaultNetwork = activeNetwork + networkAvailable = activeNetwork != null + _isConnected.value = networkAvailable + callbackRegistration.also { callbackRegistration = cm to callback } + } + if (networkChanged) networkChangeEvents.tryEmit(networkAvailable) + previous?.let { (previousCm, previousCallback) -> + runCatching { previousCm.unregisterNetworkCallback(previousCallback) } + } + cm.registerDefaultNetworkCallback(callback) + } - override fun onLost(network: Network) { - _isConnected.value = false - } - }, - ) + internal fun createCallback(cm: ConnectivityManager): ConnectivityManager.NetworkCallback { + val generation = callbackGeneration.incrementAndGet() + return object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) { + val changed = + synchronized(callbackLock) { + if (callbackGeneration.get() != generation) return + (defaultNetwork != network).also { + defaultNetwork = network + _isConnected.value = true + } + } + if (changed) networkChangeEvents.tryEmit(true) + } + + override fun onLost(network: Network) { + var networkAvailable = false + val changed = + synchronized(callbackLock) { + if (callbackGeneration.get() != generation || defaultNetwork != network) return + val activeNetwork = cm.activeNetwork?.takeUnless { it == network } + (defaultNetwork != activeNetwork).also { + defaultNetwork = activeNetwork + networkAvailable = activeNetwork != null + _isConnected.value = networkAvailable + } + } + if (changed) networkChangeEvents.tryEmit(networkAvailable) + } + } } } diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt index d202a8f8..5662aedc 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/HermesWsClient.kt @@ -181,24 +181,9 @@ object HermesWsClient { } init { - // Monitor network state to trigger immediate reconnect when network is restored wsScope.launch { - NetworkMonitor.isConnected.collect { connected -> - val shouldOpen = - synchronized(outboundLock) { - if (connected && !isConnected && !intentionalClose.get() && AuthManager.isAutoReconnect()) { - Log.d(TAG, "Network restored — triggering immediate reconnect") - currentBackoff = INITIAL_BACKOFF_MS - reconnectJob?.cancel() - reconnectJob = null - true - } else { - false - } - } - if (shouldOpen) { - openSocket() - } + NetworkMonitor.networkChanges.collect { networkAvailable -> + reconnectForNetworkChange(networkAvailable) } } // Extract credential_warning from gateway.ready / session.info payloads. @@ -244,36 +229,80 @@ object HermesWsClient { /** Open a WebSocket connection using settings from [AuthManager]. */ fun connect() { - acceptQueuedMessages.set(true) - if (connected.get()) { - Log.d(TAG, "Already connected — skipping") - return + var socketToCancel: WebSocket? = null + val shouldOpen = + synchronized(outboundLock) { + acceptQueuedMessages.set(true) + if (_connectionStatus.value == ConnectionStatus.AUTH_EXPIRED) { + connectionGeneration.incrementAndGet() + reconnectJob?.cancel() + reconnectJob = null + outboundDrainJob?.cancel() + outboundDrainJob = null + socketToCancel = webSocket + webSocket = null + closingSocket = null + connected.set(false) + stopHealthTracking() + } + if (connected.get()) { + Log.d(TAG, "Already connected — skipping") + return@synchronized false + } + if (_connectionStatus.value == ConnectionStatus.CONNECTING || + _connectionStatus.value == ConnectionStatus.RECONNECTING + ) { + Log.d(TAG, "Connection already in flight (${_connectionStatus.value}) — skipping") + return@synchronized false + } + // An explicit connect follows successful authentication and is + // therefore allowed to leave the terminal AUTH_EXPIRED state. + intentionalClose.set(false) + _connectionStatus.value = ConnectionStatus.CONNECTING + currentBackoff = INITIAL_BACKOFF_MS + true + } + if (socketToCancel != null) { + rejectAllPending() + socketToCancel?.cancel() } - // Guard against re-entrant connect() while a connection is already in - // flight. The singleton may be CONNECTING (mid handshake) or RECONNECTING - // (a scheduled reconnect is pending). Opening a second socket on the same - // `webSocket` field races the in-flight one and can leave the status - // stuck on RECONNECTING (e.g. the chat tab calls connect() on every open - // while the app-level reconnect is already running). Only start a fresh - // socket from a terminal state. - if (_connectionStatus.value == ConnectionStatus.CONNECTING || - _connectionStatus.value == ConnectionStatus.RECONNECTING - ) { - Log.d(TAG, "Connection already in flight (${_connectionStatus.value}) — skipping") - return - } - // AUTH_EXPIRED cannot be resolved by reconnecting alone — caller must - // re-authenticate. Leave the status as-is so the UI can surface sign-in. - if (_connectionStatus.value == ConnectionStatus.AUTH_EXPIRED) { - Log.d(TAG, "Connection is AUTH_EXPIRED — skipping reconnect; re-auth required") - return - } - intentionalClose.set(false) + if (shouldOpen) openSocket() + } + + @VisibleForTesting + internal fun reconnectForNetworkChange( + networkAvailable: Boolean = NetworkMonitor.isConnected.value, + openReplacement: () -> Unit = ::openSocket, + ) { + val socketToCancel: WebSocket? + val shouldOpen: Boolean synchronized(outboundLock) { + if (intentionalClose.get() || _connectionStatus.value == ConnectionStatus.AUTH_EXPIRED) { + return + } + Log.d(TAG, "Default network changed — reconnecting WebSocket") currentBackoff = INITIAL_BACKOFF_MS + reconnectJob?.cancel() + reconnectJob = null + outboundDrainJob?.cancel() + outboundDrainJob = null + connectionGeneration.incrementAndGet() + socketToCancel = webSocket + webSocket = null + closingSocket = null + connected.set(false) + stopHealthTracking() + shouldOpen = networkAvailable && AuthManager.isAutoReconnect() + _connectionStatus.value = + when { + !networkAvailable -> ConnectionStatus.NO_NETWORK + shouldOpen -> ConnectionStatus.RECONNECTING + else -> ConnectionStatus.DISCONNECTED + } + rejectAllPending() } - _connectionStatus.value = ConnectionStatus.CONNECTING - openSocket() + socketToCancel?.cancel() + if (shouldOpen) openReplacement() } /** @@ -288,7 +317,7 @@ object HermesWsClient { * or because ticket refresh succeeded). Returns false if we are in gated mode * and ticket refresh failed or cannot be performed. */ - internal fun refreshWsTicketIfNeeded(): Boolean { + internal fun refreshWsTicketIfNeeded(generation: Int): Boolean { val isGated = try { AuthManager.serverStore.getLatestState().wsAuthParam == "ticket" @@ -303,13 +332,26 @@ object HermesWsClient { // The loopback dashboard token is regenerated on every server // restart. Refresh it before each WebSocket handshake so automatic // reconnect does not get stuck in AUTH_EXPIRED with a stale token. - DashboardSessionTokenRefresher.refresh() + val token = + synchronized(DashboardSessionTokenRefresher) { + runCatching { + DashboardSessionTokenRefresher.fetch(AuthManager.baseUrl(), OkHttpProvider.probe) + }.getOrNull() + } + synchronized(outboundLock) { + if (isCurrentGeneration(generation) && token != null) { + runCatching { AuthManager.setToken(token) } + } + } return true } val ticketResult = requestWsTicket() if (!ticketResult.ticket.isNullOrBlank()) { - AuthManager.setToken(ticketResult.ticket) + synchronized(outboundLock) { + if (!isCurrentGeneration(generation)) return false + AuthManager.setToken(ticketResult.ticket) + } if (BuildConfig.DEBUG) Log.d(TAG, "WS ticket refreshed") return true } @@ -321,7 +363,7 @@ object HermesWsClient { ticketResult.httpCode != null && ticketResult.httpCode >= 500 -> ConnectionStatus.RECONNECTING else -> ConnectionStatus.DISCONNECTED } - return handleWsTicketRefreshFailure(status) + return handleWsTicketRefreshFailure(status, generation) } /** @@ -396,9 +438,15 @@ object HermesWsClient { return TicketRequestResult(null, null) } - private fun handleWsTicketRefreshFailure(status: ConnectionStatus): Boolean { - _connectionStatus.value = status - if (status == ConnectionStatus.RECONNECTING) scheduleReconnect() + private fun handleWsTicketRefreshFailure( + status: ConnectionStatus, + generation: Int, + ): Boolean { + synchronized(outboundLock) { + if (!isCurrentGeneration(generation)) return false + _connectionStatus.value = status + if (status == ConnectionStatus.RECONNECTING) scheduleReconnect() + } return false } @@ -530,13 +578,14 @@ object HermesWsClient { onSent: ((String) -> Unit)? = null, ): String { val id = requestId.incrementAndGet().toString() - onSent?.invoke(id) val request = JsonRpcRequest(id = id, method = method, params = params.mapValues { it.value.toJsonElement() }) val json = OkHttpProvider.json.encodeToString(request) if (BuildConfig.DEBUG) Log.d(TAG, "→ $json") synchronized(outboundLock) { + onSent?.invoke(id) val ws = webSocket if (ws != null && connected.get()) { + // Never replay send(true): the server may have executed it even if its response is lost. if (!ws.send(json)) { if (webSocket !== ws || !acceptQueuedMessages.get()) return@synchronized if (isRetryableMessage(json)) { @@ -569,8 +618,7 @@ object HermesWsClient { id: String, json: String, ) { - messageQueue.add(json) - queuedMessagesById[id] = json + if (queuedMessagesById.putIfAbsent(id, json) == null) messageQueue.add(json) } private fun removeQueuedMessage(id: String) { @@ -644,16 +692,21 @@ object HermesWsClient { // ── Internal ───────────────────────────────────────────────────────── private fun openSocket() { - if (!refreshWsTicketIfNeeded()) { + val generation = + synchronized(outboundLock) { + if (intentionalClose.get()) return + connectionGeneration.incrementAndGet() + } + if (!refreshWsTicketIfNeeded(generation)) { Log.w(TAG, "Aborting openSocket: WS ticket refresh failed") return } + if (!isCurrentGeneration(generation)) return val url = AuthManager.wsUrl() val safeUrl = url.replace(Regex("token=[^&]+"), "token=REDACTED") if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl") val request = Request.Builder().url(url).build() - val generation = connectionGeneration.incrementAndGet() val newSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl(generation)) synchronized(outboundLock) { if (connectionGeneration.get() == generation && !intentionalClose.get()) { @@ -664,9 +717,17 @@ object HermesWsClient { } } + private fun isCurrentGeneration(generation: Int): Boolean = + connectionGeneration.get() == generation && !intentionalClose.get() + private fun scheduleReconnect() { synchronized(outboundLock) { - if (intentionalClose.get() || reconnectJob?.isActive == true) return + if (intentionalClose.get() || + _connectionStatus.value == ConnectionStatus.AUTH_EXPIRED || + reconnectJob?.isActive == true + ) { + return + } if (!AuthManager.isAutoReconnect()) { if (BuildConfig.DEBUG) Log.d(TAG, "Auto-reconnect disabled") _connectionStatus.value = ConnectionStatus.DISCONNECTED @@ -694,7 +755,9 @@ object HermesWsClient { false } else { reconnectJob = null - !intentionalClose.get() && !connected.get() + !intentionalClose.get() && + !connected.get() && + _connectionStatus.value != ConnectionStatus.AUTH_EXPIRED } } if (shouldOpen) openSocket() @@ -707,7 +770,7 @@ object HermesWsClient { private class WsListenerImpl( private val generation: Int, ) : WebSocketListener() { - private fun isCurrent(): Boolean = connectionGeneration.get() == generation && !intentionalClose.get() + private fun isCurrent(): Boolean = isCurrentGeneration(generation) override fun onOpen( webSocket: WebSocket, @@ -765,8 +828,16 @@ object HermesWsClient { WsEvent.Unknown(text) } when (event) { - is WsEvent.RpcResult -> resolvePending(event.id, event.result, null) - is WsEvent.RpcError -> resolvePending(event.id, null, event.error) + is WsEvent.RpcResult -> { + removeQueuedMessage(event.id) + resolvePending(event.id, event.result, null) + } + + is WsEvent.RpcError -> { + removeQueuedMessage(event.id) + resolvePending(event.id, null, event.error) + } + else -> Unit } val emitted = rawMessages.tryEmit(text) @@ -820,7 +891,7 @@ object HermesWsClient { reason.startsWith("auth:", ignoreCase = true) ) { _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED - } else { + } else if (_connectionStatus.value != ConnectionStatus.AUTH_EXPIRED) { _connectionStatus.value = ConnectionStatus.RECONNECTING scheduleReconnect() } @@ -851,7 +922,7 @@ object HermesWsClient { ) == true || t.message?.contains("unauthorized", ignoreCase = true) == true ) { _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED - } else { + } else if (_connectionStatus.value != ConnectionStatus.AUTH_EXPIRED) { _connectionStatus.value = ConnectionStatus.RECONNECTING scheduleReconnect() } diff --git a/app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt b/app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt new file mode 100644 index 00000000..18a4b7a8 --- /dev/null +++ b/app/src/test/java/com/m57/hermescontrol/data/remote/NetworkMonitorTest.kt @@ -0,0 +1,97 @@ +package com.m57.hermescontrol.data.remote + +import android.net.ConnectivityManager +import android.net.Network +import app.cash.turbine.test +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +class NetworkMonitorTest { + @Before + fun setUp() { + val defaultNetworkField = NetworkMonitor::class.java.getDeclaredField("defaultNetwork") + defaultNetworkField.isAccessible = true + defaultNetworkField.set(NetworkMonitor, null) + val connectedField = NetworkMonitor::class.java.getDeclaredField("_isConnected") + connectedField.isAccessible = true + @Suppress("UNCHECKED_CAST") + (connectedField.get(NetworkMonitor) as MutableStateFlow).value = true + } + + @Test + fun `losing the old default network keeps connectivity when replacement is active`() { + val connectivityManager = mockk() + val oldNetwork = mockk() + val replacement = mockk() + every { connectivityManager.activeNetwork } returns replacement + val callback = NetworkMonitor.createCallback(connectivityManager) + + callback.onAvailable(oldNetwork) + callback.onLost(oldNetwork) + + assertTrue(NetworkMonitor.isConnected.value) + } + + @Test + fun `old network loss after replacement availability is ignored`() = + runTest { + val connectivityManager = mockk() + val oldNetwork = mockk() + val replacement = mockk() + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.networkChanges.test { + callback.onAvailable(oldNetwork) + assertTrue(awaitItem()) + callback.onAvailable(replacement) + assertTrue(awaitItem()) + callback.onLost(oldNetwork) + + assertTrue(NetworkMonitor.isConnected.value) + expectNoEvents() + } + } + + @Test + fun `callback from previous registration is ignored after restart`() { + val connectivityManager = mockk() + val oldNetwork = mockk() + val replacement = mockk() + every { connectivityManager.activeNetwork } returns null + val staleCallback = NetworkMonitor.createCallback(connectivityManager) + staleCallback.onAvailable(oldNetwork) + val currentCallback = NetworkMonitor.createCallback(connectivityManager) + currentCallback.onAvailable(replacement) + + staleCallback.onLost(oldNetwork) + + assertTrue(NetworkMonitor.isConnected.value) + } + + @Test + fun `break before make emits loss and replacement identity transitions in order`() = + runTest { + val connectivityManager = mockk() + val oldNetwork = mockk() + val replacement = mockk() + every { connectivityManager.activeNetwork } returns null + val callback = NetworkMonitor.createCallback(connectivityManager) + + NetworkMonitor.networkChanges.test { + callback.onAvailable(oldNetwork) + callback.onLost(oldNetwork) + callback.onAvailable(replacement) + + assertEquals(true, awaitItem()) + assertEquals(false, awaitItem()) + assertEquals(true, awaitItem()) + expectNoEvents() + } + } +} diff --git a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt index 8a889c7f..f07be3d5 100644 --- a/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt +++ b/app/src/test/java/com/m57/hermescontrol/data/ws/HermesWsClientTest.kt @@ -4,6 +4,7 @@ import android.util.Log import com.m57.hermescontrol.data.local.AuthManager import com.m57.hermescontrol.data.remote.CleartextPolicy import com.m57.hermescontrol.data.remote.CookieManager +import com.m57.hermescontrol.data.remote.DashboardSessionTokenRefresher import com.m57.hermescontrol.data.remote.NetworkMonitor import com.m57.hermescontrol.data.remote.ServerEndpoint import com.m57.hermescontrol.data.remote.buildFakePersistentCookieJar @@ -12,6 +13,7 @@ import io.mockk.mockk import io.mockk.mockkObject import io.mockk.mockkStatic import io.mockk.unmockkAll +import io.mockk.verify import kotlinx.coroutines.Job import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first @@ -33,6 +35,8 @@ import org.junit.Test import java.io.IOException import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.concurrent.thread class HermesWsClientTest { private lateinit var mockWebServer: MockWebServer @@ -181,6 +185,144 @@ class HermesWsClientTest { assertEquals(1, queue.size) } + @Test + fun testSendRegistersRequestIdWhileOutboundLockHeld() { + val lockField = HermesWsClient::class.java.getDeclaredField("outboundLock").apply { isAccessible = true } + val outboundLock = lockField.get(HermesWsClient) + var callbackHeldLock = false + + HermesWsClient.send("test.method", onSent = { callbackHeldLock = Thread.holdsLock(outboundLock) }) + + assertTrue(callbackHeldLock) + } + + @Test + fun testNetworkChangeInvalidatesOldSocketAndOpensReplacementImmediately() { + every { AuthManager.isAutoReconnect() } returns true + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) + val socket = mockk(relaxed = true) + every { socket.send(any()) } returns true + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, socket) + val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration") + generationField.isAccessible = true + val generation = generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger + val oldGeneration = generation.get() + val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl") + val constructor = listenerClass.declaredConstructors.single() + constructor.isAccessible = true + val staleListener = constructor.newInstance(oldGeneration) as WebSocketListener + var replacementOpens = 0 + val deferred = HermesWsClient.request(WsMethods.PROCESS_LIST, mapOf("session_id" to "s1")) + + HermesWsClient.reconnectForNetworkChange { replacementOpens++ } + staleListener.onClosing(socket, 4401, "expired") + staleListener.onFailure(socket, IOException("old network lost"), null) + + verify(exactly = 1) { socket.cancel() } + assertEquals(1, replacementOpens) + assertEquals(oldGeneration + 1, generation.get()) + assertFalse(HermesWsClient.isConnected) + assertEquals(ConnectionStatus.RECONNECTING, HermesWsClient.connectionStatus.value) + assertEquals(null, socketField.get(HermesWsClient)) + assertTrue(deferred.isCompleted) + } + + @Test + fun testNetworkLossPreservesAuthExpiredStatus() = + runBlocking { + every { AuthManager.isAutoReconnect() } returns true + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + val statusField = HermesWsClient::class.java.getDeclaredField("_connectionStatus") + statusField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val status = statusField.get(HermesWsClient) as MutableStateFlow + status.value = ConnectionStatus.AUTH_EXPIRED + val socket = mockk(relaxed = true) + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, socket) + + HermesWsClient.reconnectForNetworkChange(false) + HermesWsClient.reconnectForNetworkChange(true) + + assertEquals(ConnectionStatus.AUTH_EXPIRED, HermesWsClient.connectionStatus.value) + verify(exactly = 0) { socket.cancel() } + } + + @Test + fun testBreakBeforeMakeDoesNotReplayAcceptedRequest() { + every { AuthManager.isAutoReconnect() } returns true + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) + val oldSocket = mockk(relaxed = true) + every { oldSocket.send(any()) } returns true + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, oldSocket) + HermesWsClient.send(WsMethods.SUBSCRIPTION_CHANGE, mapOf("cancel" to true)) + val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue") + queueField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue + assertTrue(queue.isEmpty()) + + HermesWsClient.reconnectForNetworkChange(false) + assertTrue(queue.isEmpty()) + + var replacementOpens = 0 + HermesWsClient.reconnectForNetworkChange(true) { replacementOpens++ } + val replacementSocket = mockk(relaxed = true) + every { replacementSocket.send(any()) } returns true + val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration") + generationField.isAccessible = true + val generation = generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger + val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl") + val constructor = listenerClass.declaredConstructors.single() + constructor.isAccessible = true + val replacementListener = constructor.newInstance(generation.get()) as WebSocketListener + replacementListener.onOpen(replacementSocket, mockk(relaxed = true)) + + assertEquals(1, replacementOpens) + assertTrue(queue.isEmpty()) + verify(exactly = 1) { oldSocket.cancel() } + verify(exactly = 0) { replacementSocket.send(any()) } + } + + @Test + fun testNetworkChangePreservesAuthExpiredSocket() { + every { AuthManager.isAutoReconnect() } returns true + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + val statusField = HermesWsClient::class.java.getDeclaredField("_connectionStatus") + statusField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val status = statusField.get(HermesWsClient) as MutableStateFlow + status.value = ConnectionStatus.AUTH_EXPIRED + val socket = mockk(relaxed = true) + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, socket) + + HermesWsClient.reconnectForNetworkChange() + + assertEquals(ConnectionStatus.AUTH_EXPIRED, HermesWsClient.connectionStatus.value) + verify(exactly = 0) { socket.cancel() } + } + @Test fun testReconnectSchedulingIsIdempotentWhilePending() { mockkObject(NetworkMonitor) @@ -537,6 +679,103 @@ class HermesWsClientTest { io.mockk.verify(exactly = 0) { socket.cancel() } } + @Test + fun testSocketFailureDoesNotOverwriteAuthExpired() { + val socket = mockk(relaxed = true) + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration") + generationField.isAccessible = true + (generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger).set(1) + val statusField = HermesWsClient::class.java.getDeclaredField("_connectionStatus") + statusField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val status = statusField.get(HermesWsClient) as MutableStateFlow + status.value = ConnectionStatus.AUTH_EXPIRED + val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl") + val constructor = listenerClass.declaredConstructors.single() + constructor.isAccessible = true + val listener = constructor.newInstance(1) as WebSocketListener + + listener.onFailure(socket, IOException("network changed"), null) + + assertEquals(ConnectionStatus.AUTH_EXPIRED, HermesWsClient.connectionStatus.value) + } + + @Test + fun testExplicitConnectRetriesAfterAuthExpired() = + runBlocking { + val statusField = HermesWsClient::class.java.getDeclaredField("_connectionStatus") + statusField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val status = statusField.get(HermesWsClient) as MutableStateFlow + status.value = ConnectionStatus.AUTH_EXPIRED + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) + val oldSocket = mockk(relaxed = true) + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, oldSocket) + mockWebServer.enqueue(MockResponse().withWebSocketUpgrade(object : WebSocketListener() {})) + + HermesWsClient.connect() + + withTimeout(5_000) { + HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } + } + verify(exactly = 1) { oldSocket.cancel() } + } + + @Test + fun testDisconnectThenConnectSupersedesBlockedSocketOpen() = + runBlocking { + mockkObject(DashboardSessionTokenRefresher) + every { AuthManager.baseUrl() } returns mockWebServer.url("/").toString() + val refreshStarted = CountDownLatch(1) + val releaseRefresh = CountDownLatch(1) + val refreshCalls = AtomicInteger(0) + every { DashboardSessionTokenRefresher.fetch(any(), any()) } answers { + if (refreshCalls.getAndIncrement() == 0) { + refreshStarted.countDown() + releaseRefresh.await(5, TimeUnit.SECONDS) + "stale-token" + } else { + null + } + } + val staleConnect = thread { HermesWsClient.connect() } + assertTrue(refreshStarted.await(5, TimeUnit.SECONDS)) + + HermesWsClient.disconnect(clearPendingMessages = true) + mockWebServer.enqueue(MockResponse().withWebSocketUpgrade(object : WebSocketListener() {})) + val generationField = HermesWsClient::class.java.getDeclaredField("connectionGeneration") + generationField.isAccessible = true + val generation = generationField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicInteger + val disconnectedGeneration = generation.get() + val replacementConnect = thread { HermesWsClient.connect() } + val generationDeadline = System.currentTimeMillis() + 5_000 + while (generation.get() == disconnectedGeneration && System.currentTimeMillis() < generationDeadline) { + Thread.sleep(10) + } + assertTrue(generation.get() > disconnectedGeneration) + val currentGeneration = generation.get() + + releaseRefresh.countDown() + staleConnect.join(5_000) + replacementConnect.join(5_000) + withTimeout(5_000) { + HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } + } + + assertFalse(staleConnect.isAlive) + assertFalse(replacementConnect.isAlive) + assertEquals(currentGeneration, generation.get()) + assertEquals(ConnectionStatus.CONNECTED, HermesWsClient.connectionStatus.value) + verify(exactly = 0) { AuthManager.setToken("stale-token") } + } + @Test fun testReceiveMessage() { var serverWebSocket: WebSocket? = null @@ -780,9 +1019,13 @@ class HermesWsClientTest { // Disconnect — this sets intentionalClose = true and cancels reconnect HermesWsClient.disconnect(clearPendingMessages = true) + var replacementOpens = 0 + HermesWsClient.reconnectForNetworkChange(false) + HermesWsClient.reconnectForNetworkChange(true) { replacementOpens++ } assertFalse(HermesWsClient.isConnected) assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value) + assertEquals(0, replacementOpens) } @Test