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 3de67808..7a930e28 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 @@ -14,6 +14,7 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow @@ -33,6 +34,7 @@ import okhttp3.RequestBody.Companion.toRequestBody import okhttp3.Response import okhttp3.WebSocket import okhttp3.WebSocketListener +import okio.utf8Size import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentLinkedQueue import java.util.concurrent.atomic.AtomicBoolean @@ -65,17 +67,26 @@ object HermesWsClient { private const val INITIAL_BACKOFF_MS = 1_000L private const val MAX_BACKOFF_MS = 30_000L private const val BACKOFF_MULTIPLIER = 2.0 + private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024 + private const val OUTBOUND_DRAIN_TIMEOUT_MS = 60_000L // ── Internal state (all access through synchronized / atomic) ──────── private val requestId = AtomicInteger(0) + private val connectionGeneration = AtomicInteger(0) private val connected = AtomicBoolean(false) private val intentionalClose = AtomicBoolean(false) + private val acceptQueuedMessages = AtomicBoolean(true) private val messageQueue = ConcurrentLinkedQueue() + private val queuedMessagesById = ConcurrentHashMap() + private val outboundLock = Any() @Volatile private var webSocket: WebSocket? = null + @Volatile + private var closingSocket: WebSocket? = null + @Volatile private var currentBackoff = INITIAL_BACKOFF_MS @@ -84,6 +95,9 @@ object HermesWsClient { @Volatile private var reconnectJob: Job? = null + @Volatile + private var outboundDrainJob: Job? = null + // ── Health and Ping/Pong tracking ──────────────────────────────────── @Volatile @@ -161,10 +175,19 @@ object HermesWsClient { // Monitor network state to trigger immediate reconnect when network is restored wsScope.launch { NetworkMonitor.isConnected.collect { connected -> - if (connected && !isConnected && !intentionalClose.get() && AuthManager.isAutoReconnect()) { - Log.d(TAG, "Network restored — triggering immediate reconnect") - currentBackoff = INITIAL_BACKOFF_MS - reconnectJob?.cancel() + 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() } } @@ -193,6 +216,7 @@ 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 @@ -217,7 +241,9 @@ object HermesWsClient { return } intentionalClose.set(false) - currentBackoff = INITIAL_BACKOFF_MS + synchronized(outboundLock) { + currentBackoff = INITIAL_BACKOFF_MS + } _connectionStatus.value = ConnectionStatus.CONNECTING openSocket() } @@ -268,41 +294,64 @@ object HermesWsClient { client.newCall(request).execute() } - if (response.isSuccessful) { - val body = response.body.string() - val ticketMatch = Regex("""\"ticket\":\"([^\"]+)\"""").find(body) - val ticket = ticketMatch?.groupValues?.getOrNull(1) - if (!ticket.isNullOrBlank()) { - AuthManager.setToken(ticket) - if (BuildConfig.DEBUG) Log.d(TAG, "WS ticket refreshed") - return true + response.use { + if (response.isSuccessful) { + val body = response.body.string() + val ticketMatch = Regex("""\"ticket\":\"([^\"]+)\"""").find(body) + val ticket = ticketMatch?.groupValues?.getOrNull(1) + if (!ticket.isNullOrBlank()) { + AuthManager.setToken(ticket) + if (BuildConfig.DEBUG) Log.d(TAG, "WS ticket refreshed") + return true + } else { + Log.w(TAG, "WS ticket refresh failed: response body did not contain ticket") + return handleWsTicketRefreshFailure(ConnectionStatus.DISCONNECTED) + } } else { - Log.w(TAG, "WS ticket refresh failed: response body did not contain ticket") - _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED - return false + Log.w(TAG, "WS ticket refresh failed: HTTP ${response.code}") + val status = + when { + response.code == 401 || response.code == 403 -> ConnectionStatus.AUTH_EXPIRED + response.code == 408 || response.code == 429 || response.code >= 500 -> + ConnectionStatus.RECONNECTING + else -> ConnectionStatus.DISCONNECTED + } + return handleWsTicketRefreshFailure(status) } - } else { - Log.w(TAG, "WS ticket refresh failed: HTTP ${response.code}") - _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED - return false } } catch (e: Exception) { Log.w(TAG, "WS ticket refresh failed: ${e.javaClass.simpleName}") - _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED - return false + return handleWsTicketRefreshFailure(ConnectionStatus.RECONNECTING) } } + private fun handleWsTicketRefreshFailure(status: ConnectionStatus): Boolean { + _connectionStatus.value = status + if (status == ConnectionStatus.RECONNECTING) scheduleReconnect() + return false + } + /** Cleanly close the WebSocket and stop auto-reconnect. */ - fun disconnect() { - intentionalClose.set(true) - reconnectJob?.cancel() - reconnectJob = null - stopHealthTracking() - webSocket?.close(1000, "Client closed") - webSocket = null - connected.set(false) - _connectionStatus.value = ConnectionStatus.DISCONNECTED + fun disconnect(clearPendingMessages: Boolean = false) { + synchronized(outboundLock) { + intentionalClose.set(true) + acceptQueuedMessages.set(!clearPendingMessages) + connectionGeneration.incrementAndGet() + reconnectJob?.cancel() + reconnectJob = null + outboundDrainJob?.cancel() + outboundDrainJob = null + stopHealthTracking() + webSocket?.close(1000, "Client closed") + webSocket = null + closingSocket = null + if (clearPendingMessages) { + messageQueue.clear() + queuedMessagesById.clear() + } + connected.set(false) + _connectionStatus.value = ConnectionStatus.DISCONNECTED + } } // ── Awaited RPC request layer (issue #526) ───────────────────────── @@ -368,6 +417,7 @@ object HermesWsClient { error: JsonRpcError?, ) { val call = pendingCalls.remove(id) ?: return + removeQueuedMessage(id) call.timeoutJob?.cancel() if (error != null) { call.deferred.completeExceptionally(HermesRpcException(error.message)) @@ -390,6 +440,7 @@ object HermesWsClient { val snapshot = pendingCalls.toList() pendingCalls.clear() for ((id, call) in snapshot) { + removeQueuedMessage(id) Log.w(TAG, "Rejecting pending request on disconnect: ${call.method} (id=$id)") call.timeoutJob?.cancel() call.deferred.completeExceptionally(error) @@ -412,16 +463,79 @@ object HermesWsClient { 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") - val ws = webSocket - if (ws != null && connected.get()) { - ws.send(json) - } else { - Log.d(TAG, "WS disconnected — queuing message") - messageQueue.add(json) + synchronized(outboundLock) { + val ws = webSocket + if (ws != null && connected.get()) { + if (!ws.send(json)) { + if (webSocket !== ws || !acceptQueuedMessages.get()) return@synchronized + if (isRetryableMessage(json)) { + Log.w(TAG, "WS rejected outgoing message — queuing for reconnect") + queueMessage(id, json) + recoverRejectedSocket(ws) + } else { + Log.w(TAG, "WS rejected oversized outgoing message — not retrying") + } + } + } else if (acceptQueuedMessages.get()) { + if (isRetryableMessage(json)) { + Log.d(TAG, "WS disconnected — queuing message") + queueMessage(id, json) + } else { + Log.w(TAG, "WS disconnected with oversized outgoing message — not queueing") + } + } } return id } + private fun isRetryableMessage(json: String): Boolean { + if (json.length > MAX_OUTBOUND_MESSAGE_BYTES) return false + if (json.length <= MAX_OUTBOUND_MESSAGE_BYTES / 4) return true + return json.utf8Size() <= MAX_OUTBOUND_MESSAGE_BYTES.toLong() + } + + private fun queueMessage( + id: String, + json: String, + ) { + messageQueue.add(json) + queuedMessagesById[id] = json + } + + private fun removeQueuedMessage(id: String) { + synchronized(outboundLock) { + queuedMessagesById.remove(id)?.let(messageQueue::remove) + } + } + + private fun markQueuedMessageSent(json: String) { + queuedMessagesById.entries.removeIf { it.value == json } + } + + private fun recoverRejectedSocket(ws: WebSocket) { + connected.set(false) + if (closingSocket === ws) return + _connectionStatus.value = ConnectionStatus.RECONNECTING + if (ws.queueSize() == 0L) { + ws.cancel() + scheduleReconnect() + return + } + if (intentionalClose.get()) return + outboundDrainJob?.cancel() + outboundDrainJob = + wsScope.launch { + delay(OUTBOUND_DRAIN_TIMEOUT_MS) + synchronized(outboundLock) { + if (!connected.get() && !intentionalClose.get() && webSocket === ws) { + ws.cancel() + outboundDrainJob = null + scheduleReconnect() + } + } + } + } + /** Convenience: submit a user prompt to an existing session. */ fun sendMessage( sessionId: String, @@ -463,56 +577,96 @@ object HermesWsClient { if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl") val request = Request.Builder().url(url).build() - webSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl()) + val generation = connectionGeneration.incrementAndGet() + val newSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl(generation)) + synchronized(outboundLock) { + if (connectionGeneration.get() == generation && !intentionalClose.get()) { + webSocket = newSocket + } else { + newSocket.cancel() + } + } } private fun scheduleReconnect() { - if (intentionalClose.get()) return - if (!AuthManager.isAutoReconnect()) { - if (BuildConfig.DEBUG) Log.d(TAG, "Auto-reconnect disabled") - _connectionStatus.value = ConnectionStatus.DISCONNECTED - return - } - if (!NetworkMonitor.isConnected.value) { - Log.d(TAG, "No network available — delaying reconnect scheduling") - _connectionStatus.value = ConnectionStatus.NO_NETWORK - return - } - val delay = currentBackoff - currentBackoff = - (currentBackoff * BACKOFF_MULTIPLIER) - .toLong() - .coerceAtMost(MAX_BACKOFF_MS) - if (BuildConfig.DEBUG) Log.d(TAG, "Reconnecting in ${delay}ms …") - - reconnectJob?.cancel() - reconnectJob = - wsScope.launch { - delay(delay) - if (!intentionalClose.get() && !connected.get()) { - openSocket() - } + synchronized(outboundLock) { + if (intentionalClose.get() || reconnectJob?.isActive == true) return + if (!AuthManager.isAutoReconnect()) { + if (BuildConfig.DEBUG) Log.d(TAG, "Auto-reconnect disabled") + _connectionStatus.value = ConnectionStatus.DISCONNECTED + return } + if (!NetworkMonitor.isConnected.value) { + Log.d(TAG, "No network available — delaying reconnect scheduling") + _connectionStatus.value = ConnectionStatus.NO_NETWORK + return + } + val reconnectDelay = currentBackoff + currentBackoff = + (currentBackoff * BACKOFF_MULTIPLIER) + .toLong() + .coerceAtMost(MAX_BACKOFF_MS) + if (BuildConfig.DEBUG) Log.d(TAG, "Reconnecting in ${reconnectDelay}ms …") + + reconnectJob = + wsScope.launch { + delay(reconnectDelay) + val owner = currentCoroutineContext()[Job] + val shouldOpen = + synchronized(outboundLock) { + if (reconnectJob !== owner) { + false + } else { + reconnectJob = null + !intentionalClose.get() && !connected.get() + } + } + if (shouldOpen) openSocket() + } + } } // ── Listener ───────────────────────────────────────────────────────── - private class WsListenerImpl : WebSocketListener() { + private class WsListenerImpl( + private val generation: Int, + ) : WebSocketListener() { + private fun isCurrent(): Boolean = connectionGeneration.get() == generation && !intentionalClose.get() + override fun onOpen( webSocket: WebSocket, response: Response, ) { - Log.i(TAG, "WebSocket opened") - connected.set(true) - _connectionStatus.value = ConnectionStatus.CONNECTED - currentBackoff = INITIAL_BACKOFF_MS - startHealthTracking() + synchronized(outboundLock) { + if (!isCurrent()) { + webSocket.close(1000, "Superseded") + return + } + HermesWsClient.webSocket = webSocket + closingSocket = null + outboundDrainJob?.cancel() + outboundDrainJob = null + Log.i(TAG, "WebSocket opened") + connected.set(true) + _connectionStatus.value = ConnectionStatus.CONNECTED + currentBackoff = INITIAL_BACKOFF_MS + startHealthTracking() - while (messageQueue.isNotEmpty()) { - val msg = messageQueue.poll() - if (msg != null) { + while (true) { + val msg = messageQueue.peek() ?: break + if (!isRetryableMessage(msg)) { + Log.w(TAG, "Dropping oversized queued message") + messageQueue.poll() + markQueuedMessageSent(msg) + continue + } if (BuildConfig.DEBUG) Log.d(TAG, "→ (queued) $msg") - webSocket.send(msg) + if (!webSocket.send(msg)) { + recoverRejectedSocket(webSocket) + break + } + messageQueue.poll() + markQueuedMessageSent(msg) } } } @@ -521,6 +675,7 @@ object HermesWsClient { webSocket: WebSocket, text: String, ) { + if (!isCurrent() || HermesWsClient.webSocket !== webSocket) return if (BuildConfig.DEBUG) Log.d(TAG, "← $text") lastPongTimestamp = System.currentTimeMillis() // Resolve any in-flight `request()` awaiting this RPC result/error @@ -549,9 +704,22 @@ object HermesWsClient { code: Int, reason: String, ) { - // Do NOT log [reason] — it may carry server-side context. - Log.d(TAG, "WebSocket closing: $code") - webSocket.close(code, reason) + synchronized(outboundLock) { + if (!isCurrent()) { + webSocket.close(code, reason) + return + } + closingSocket = webSocket + // Do NOT log [reason] — it may carry server-side context. + Log.d(TAG, "WebSocket closing: $code") + if (code == 4001 || code == 4401 || + reason.contains("unauthorized", ignoreCase = true) || + reason.startsWith("auth:", ignoreCase = true) + ) { + _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED + } + webSocket.close(code, reason) + } } override fun onClosed( @@ -559,19 +727,27 @@ object HermesWsClient { code: Int, reason: String, ) { - // Do NOT log [reason] — it may carry server-side context. The - // reason is still inspected internally to detect auth failures. - Log.i(TAG, "WebSocket closed: $code") - connected.set(false) - stopHealthTracking() - if (code == 4001 || code == 4401 || - reason.contains("unauthorized", ignoreCase = true) || - reason.startsWith("auth:", ignoreCase = true) - ) { - _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED - } else { - _connectionStatus.value = ConnectionStatus.RECONNECTING - scheduleReconnect() + synchronized(outboundLock) { + if (!isCurrent()) return + connectionGeneration.incrementAndGet() + closingSocket = null + outboundDrainJob?.cancel() + outboundDrainJob = null + if (HermesWsClient.webSocket === webSocket) HermesWsClient.webSocket = null + // Do NOT log [reason] — it may carry server-side context. The + // reason is still inspected internally to detect auth failures. + Log.i(TAG, "WebSocket closed: $code") + connected.set(false) + stopHealthTracking() + if (code == 4001 || code == 4401 || + reason.contains("unauthorized", ignoreCase = true) || + reason.startsWith("auth:", ignoreCase = true) + ) { + _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED + } else { + _connectionStatus.value = ConnectionStatus.RECONNECTING + scheduleReconnect() + } } } @@ -580,21 +756,29 @@ object HermesWsClient { t: Throwable, response: Response?, ) { - // Log the exception class only — [Throwable.message] can leak URLs - // or headers. The message is still inspected internally for auth - // detection. - Log.e(TAG, "WebSocket failure: ${t.javaClass.simpleName}", t) - connected.set(false) - stopHealthTracking() - val code = response?.code ?: 0 - if (code == 401 || t.message?.contains( - "401", - ) == true || t.message?.contains("unauthorized", ignoreCase = true) == true - ) { - _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED - } else { - _connectionStatus.value = ConnectionStatus.RECONNECTING - scheduleReconnect() + synchronized(outboundLock) { + if (!isCurrent()) return + connectionGeneration.incrementAndGet() + closingSocket = null + outboundDrainJob?.cancel() + outboundDrainJob = null + if (HermesWsClient.webSocket === webSocket) HermesWsClient.webSocket = null + // Log the exception class only — [Throwable.message] can leak URLs + // or headers. The message is still inspected internally for auth + // detection. + Log.e(TAG, "WebSocket failure: ${t.javaClass.simpleName}", t) + connected.set(false) + stopHealthTracking() + val code = response?.code ?: 0 + if (code == 401 || t.message?.contains( + "401", + ) == true || t.message?.contains("unauthorized", ignoreCase = true) == true + ) { + _connectionStatus.value = ConnectionStatus.AUTH_EXPIRED + } else { + _connectionStatus.value = ConnectionStatus.RECONNECTING + scheduleReconnect() + } } } } diff --git a/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatConnectionBanner.kt b/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatConnectionBanner.kt index 69e6fa42..c6cf431f 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatConnectionBanner.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/chat/components/ChatConnectionBanner.kt @@ -121,9 +121,7 @@ fun ChatConnectionBanner( } } - ConnectionStatus.RECONNECTING, - ConnectionStatus.AUTH_EXPIRED, - -> { + ConnectionStatus.AUTH_EXPIRED -> { TextButton( onClick = onReloginClick, colors = @@ -135,6 +133,8 @@ fun ChatConnectionBanner( } } + ConnectionStatus.RECONNECTING -> {} + else -> {} } } diff --git a/app/src/main/java/com/m57/hermescontrol/ui/settings/SettingsViewModel.kt b/app/src/main/java/com/m57/hermescontrol/ui/settings/SettingsViewModel.kt index 59fdedfd..14b5eeda 100644 --- a/app/src/main/java/com/m57/hermescontrol/ui/settings/SettingsViewModel.kt +++ b/app/src/main/java/com/m57/hermescontrol/ui/settings/SettingsViewModel.kt @@ -299,7 +299,7 @@ class SettingsViewModel( AuthManager.setToken(null) AuthManager.setSessionCookie(null) AuthManager.setWsAuthParam("token") - HermesWsClient.disconnect() + HermesWsClient.disconnect(clearPendingMessages = true) // Don't rebuild ApiClient here — let the navigation complete first } 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 43b87ac8..8a889c7f 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.NetworkMonitor import com.m57.hermescontrol.data.remote.ServerEndpoint import com.m57.hermescontrol.data.remote.buildFakePersistentCookieJar import io.mockk.every @@ -11,6 +12,7 @@ import io.mockk.mockk import io.mockk.mockkObject import io.mockk.mockkStatic import io.mockk.unmockkAll +import kotlinx.coroutines.Job import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch @@ -24,9 +26,11 @@ import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNotNull +import org.junit.Assert.assertSame import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test +import java.io.IOException import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit @@ -72,12 +76,20 @@ class HermesWsClientTest { @Suppress("UNCHECKED_CAST") (statusField.get(HermesWsClient) as MutableStateFlow).value = ConnectionStatus.DISCONNECTED - HermesWsClient.disconnect() // Ensure it starts clean + val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue") + queueField.isAccessible = true + @Suppress("UNCHECKED_CAST") + (queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue).clear() + + HermesWsClient.disconnect(clearPendingMessages = true) // Ensure it starts clean + val acceptQueuedMessagesField = HermesWsClient::class.java.getDeclaredField("acceptQueuedMessages") + acceptQueuedMessagesField.isAccessible = true + (acceptQueuedMessagesField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) } @After fun tearDown() { - HermesWsClient.disconnect() + HermesWsClient.disconnect(clearPendingMessages = true) // Wait a bit to allow internal OkHttp coroutines to clean up before shutting down MockWebServer // Increased from 100ms for OkHttp 5.x — needs more time for the WS close handshake Thread.sleep(500) @@ -136,6 +148,395 @@ class HermesWsClientTest { assertTrue(msg.contains(id)) } + @Test + fun testFailedSocketSendQueuesMessageForReconnect() { + val staleSocket = mockk(relaxed = true) + every { staleSocket.send(any()) } returns false + + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, staleSocket) + + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(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") + (statusField.get(HermesWsClient) as MutableStateFlow).value = ConnectionStatus.CONNECTED + + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello")) + + val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue") + queueField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue + assertFalse(HermesWsClient.isConnected) + assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value) + assertEquals(1, queue.size) + } + + @Test + fun testReconnectSchedulingIsIdempotentWhilePending() { + mockkObject(NetworkMonitor) + every { AuthManager.isAutoReconnect() } returns true + every { NetworkMonitor.isConnected } returns MutableStateFlow(true) + + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + + val backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff") + backoffField.isAccessible = true + backoffField.setLong(HermesWsClient, 1_000L) + + val reconnectMethod = HermesWsClient::class.java.getDeclaredMethod("scheduleReconnect") + reconnectMethod.isAccessible = true + reconnectMethod.invoke(HermesWsClient) + reconnectMethod.invoke(HermesWsClient) + + assertEquals(2_000L, backoffField.getLong(HermesWsClient)) + } + + @Test + fun testCompletedReconnectDoesNotClearReplacementJob() { + mockkObject(NetworkMonitor) + every { AuthManager.isAutoReconnect() } returns true + every { NetworkMonitor.isConnected } returns MutableStateFlow(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(false) + + val backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff") + backoffField.isAccessible = true + backoffField.setLong(HermesWsClient, 0L) + + val reconnectJobField = HermesWsClient::class.java.getDeclaredField("reconnectJob") + reconnectJobField.isAccessible = true + val replacementJob = mockk(relaxed = true) + + val lockField = HermesWsClient::class.java.getDeclaredField("outboundLock") + lockField.isAccessible = true + val lock = lockField.get(HermesWsClient) + val reconnectMethod = HermesWsClient::class.java.getDeclaredMethod("scheduleReconnect") + reconnectMethod.isAccessible = true + + synchronized(lock) { + reconnectMethod.invoke(HermesWsClient) + reconnectJobField.set(HermesWsClient, replacementJob) + } + + Thread.sleep(100) + assertSame(replacementJob, reconnectJobField.get(HermesWsClient)) + } + + @Test + fun testOversizedRejectedMessageIsNotQueued() { + val staleSocket = mockk(relaxed = true) + every { staleSocket.send(any()) } returns false + + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, staleSocket) + + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) + + HermesWsClient.send( + WsMethods.PROMPT_SUBMIT, + mapOf("session_id" to "s1", "text" to "x".repeat(16 * 1024 * 1024 + 1)), + ) + + 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()) + io.mockk.verify(exactly = 0) { staleSocket.cancel() } + } + + @Test + fun testDisconnectedOversizedMessageIsNotQueued() { + HermesWsClient.send( + WsMethods.PROMPT_SUBMIT, + mapOf("session_id" to "s1", "text" to "x".repeat(16 * 1024 * 1024 + 1)), + ) + + 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()) + } + + @Test + fun testQueuePressureDoesNotCancelAcceptedFrames() { + val pressuredSocket = mockk(relaxed = true) + every { pressuredSocket.send(any()) } returns false + every { pressuredSocket.queueSize() } returns 1024L + + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, pressuredSocket) + + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) + + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello")) + + io.mockk.verify(exactly = 0) { pressuredSocket.cancel() } + val drainJobField = HermesWsClient::class.java.getDeclaredField("outboundDrainJob") + drainJobField.isAccessible = true + assertNotNull(drainJobField.get(HermesWsClient)) + } + + @Test + fun testFailureDuringOutboundDrainSchedulesReconnect() { + mockkObject(NetworkMonitor) + every { AuthManager.isAutoReconnect() } returns true + every { NetworkMonitor.isConnected } returns MutableStateFlow(true) + + val pressuredSocket = mockk(relaxed = true) + every { pressuredSocket.send(any()) } returns false + every { pressuredSocket.queueSize() } returns 1024L + + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, pressuredSocket) + + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(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 backoffField = HermesWsClient::class.java.getDeclaredField("currentBackoff") + backoffField.isAccessible = true + backoffField.setLong(HermesWsClient, 1_000L) + + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello")) + + 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(pressuredSocket, IOException("connection lost"), null) + + assertEquals(2_000L, backoffField.getLong(HermesWsClient)) + } + + @Test + fun testReplayPressureSchedulesRecoveryWithoutDroppingHead() { + val pressuredSocket = mockk(relaxed = true) + every { pressuredSocket.send(any()) } returnsMany listOf(true, false) + every { pressuredSocket.queueSize() } returns 1024L + + val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue") + queueField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue + queue.add("{\"jsonrpc\":\"2.0\",\"id\":\"1\"}") + queue.add("{\"jsonrpc\":\"2.0\",\"id\":\"2\"}") + + 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 socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, pressuredSocket) + + 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.onOpen(pressuredSocket, mockk(relaxed = true)) + + assertFalse(HermesWsClient.isConnected) + assertEquals(1, queue.size) + assertTrue(queue.peek()?.contains("\"2\"") == true) + io.mockk.verify(exactly = 0) { pressuredSocket.cancel() } + val drainJobField = HermesWsClient::class.java.getDeclaredField("outboundDrainJob") + drainJobField.isAccessible = true + assertNotNull(drainJobField.get(HermesWsClient)) + } + + @Test + fun testStaleOnOpenDoesNotReplayQueue() { + val staleSocket = mockk(relaxed = true) + val currentSocket = mockk(relaxed = true) + every { currentSocket.send(any()) } returns true + + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, currentSocket) + + 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(2) + + val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue") + queueField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue + queue.add("{\"jsonrpc\":\"2.0\",\"id\":\"1\"}") + + val listenerClass = Class.forName("com.m57.hermescontrol.data.ws.HermesWsClient\$WsListenerImpl") + val constructor = listenerClass.declaredConstructors.single() + constructor.isAccessible = true + val staleListener = constructor.newInstance(1) as WebSocketListener + val currentListener = constructor.newInstance(2) as WebSocketListener + val response = mockk(relaxed = true) + + staleListener.onOpen(staleSocket, response) + currentListener.onOpen(currentSocket, response) + + io.mockk.verify(exactly = 0) { staleSocket.send(any()) } + io.mockk.verify(exactly = 1) { currentSocket.send(any()) } + assertTrue(queue.isEmpty()) + } + + @Test + fun testRejectedSendAfterDisconnectIsNotQueued() { + val staleSocket = mockk(relaxed = true) + every { staleSocket.send(any()) } answers { + HermesWsClient.disconnect(clearPendingMessages = true) + false + } + + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, staleSocket) + + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(true) + + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello")) + + 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()) + assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value) + } + + @Test + fun testDisconnectPreservesQueuedMessagesUnlessExplicitlyCleared() { + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello")) + + val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue") + queueField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue + assertEquals(1, queue.size) + + HermesWsClient.disconnect() + assertEquals(1, queue.size) + + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "during reconnect")) + assertEquals(2, queue.size) + + HermesWsClient.disconnect(clearPendingMessages = true) + assertTrue(queue.isEmpty()) + + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "after logout")) + assertTrue(queue.isEmpty()) + } + + @Test + fun testRejectAllPendingRemovesQueuedAwaitedRpc() { + val intentionalCloseField = HermesWsClient::class.java.getDeclaredField("intentionalClose") + intentionalCloseField.isAccessible = true + (intentionalCloseField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(false) + + val deferred = HermesWsClient.request(WsMethods.PROCESS_LIST, mapOf("session_id" to "s1")) + + val queueField = HermesWsClient::class.java.getDeclaredField("messageQueue") + queueField.isAccessible = true + @Suppress("UNCHECKED_CAST") + val queue = queueField.get(HermesWsClient) as java.util.concurrent.ConcurrentLinkedQueue + assertEquals(1, queue.size) + + HermesWsClient.rejectAllPending() + + assertTrue(deferred.isCompleted) + assertTrue(queue.isEmpty()) + } + + @Test + fun testAuthClosingCodeSurvivesRejectedSendRecovery() { + val socket = mockk(relaxed = true) + every { socket.send(any()) } returns false + every { socket.queueSize() } returns 0L + + val socketField = HermesWsClient::class.java.getDeclaredField("webSocket") + socketField.isAccessible = true + socketField.set(HermesWsClient, socket) + + val connectedField = HermesWsClient::class.java.getDeclaredField("connected") + connectedField.isAccessible = true + (connectedField.get(HermesWsClient) as java.util.concurrent.atomic.AtomicBoolean).set(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 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.onClosing(socket, 4401, "expired") + HermesWsClient.send(WsMethods.PROMPT_SUBMIT, mapOf("session_id" to "s1", "text" to "hello")) + + assertEquals(ConnectionStatus.AUTH_EXPIRED, HermesWsClient.connectionStatus.value) + io.mockk.verify(exactly = 0) { socket.cancel() } + } + @Test fun testReceiveMessage() { var serverWebSocket: WebSocket? = null @@ -211,7 +612,7 @@ class HermesWsClientTest { runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } } assertTrue(serverLatch.await(5, TimeUnit.SECONDS)) - HermesWsClient.disconnect() + HermesWsClient.disconnect(clearPendingMessages = true) assertFalse(HermesWsClient.isConnected) assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value) @@ -378,7 +779,7 @@ class HermesWsClientTest { runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } } // Disconnect — this sets intentionalClose = true and cancels reconnect - HermesWsClient.disconnect() + HermesWsClient.disconnect(clearPendingMessages = true) assertFalse(HermesWsClient.isConnected) assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value) @@ -482,7 +883,7 @@ class HermesWsClientTest { runBlocking { withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } } } // Disconnect (sets intentionalClose) — after this, reconnect should be prevented - HermesWsClient.disconnect() + HermesWsClient.disconnect(clearPendingMessages = true) assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value) assertFalse(HermesWsClient.isConnected) } @@ -557,4 +958,54 @@ class HermesWsClientTest { ticketServer.shutdown() } + + @Test + fun testGatedMode_permanentTicketFailureStopsRetrying() { + every { AuthManager.isAutoReconnect() } returns true + every { AuthManager.serverStore } returns + mockk().also { + every { it.getLatestState() } returns + com.m57.hermescontrol.data.config + .ServerStoreState(wsAuthParam = "ticket") + } + + val ticketServer = MockWebServer() + ticketServer.start() + ticketServer.enqueue(MockResponse().setResponseCode(400)) + every { AuthManager.endpointForBuild() } returns + ServerEndpoint.parse( + ticketServer.url("/").toString(), + CleartextPolicy.ALLOW_WITH_WARNING, + ) + + HermesWsClient.connect() + + assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value) + ticketServer.shutdown() + } + + @Test + fun testGatedMode_transientTicketFailureKeepsReconnectFlow() { + every { AuthManager.isAutoReconnect() } returns true + every { AuthManager.serverStore } returns + mockk().also { + every { it.getLatestState() } returns + com.m57.hermescontrol.data.config + .ServerStoreState(wsAuthParam = "ticket") + } + + val unavailableTicketServer = MockWebServer() + unavailableTicketServer.start() + val unavailableEndpoint = unavailableTicketServer.url("/").toString() + unavailableTicketServer.shutdown() + every { AuthManager.endpointForBuild() } returns + ServerEndpoint.parse( + unavailableEndpoint, + CleartextPolicy.ALLOW_WITH_WARNING, + ) + + HermesWsClient.connect() + + assertEquals(ConnectionStatus.RECONNECTING, HermesWsClient.connectionStatus.value) + } }