From bb400881e6ddee443ae3a35a8d314bfe46b68bf4 Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Sat, 22 Aug 2026 18:49:50 +0300 Subject: [PATCH] ws: harden HermesWsClient (stale-socket watchdog, JSON ticket parse, dead-branch removal) (#954) * ws: watchdog cancels stale sockets to trigger reconnect * ws: parse ws-ticket response with kotlinx.serialization * ws: remove unreachable buffer-overflow warning * style: ktlint format ws client tests --------- Co-authored-by: hy4ri --- .../hermescontrol/data/ws/HermesWsClient.kt | 63 ++++++++++--- .../hermescontrol/data/ws/JsonRpcModels.kt | 11 +++ .../data/ws/HermesWsClientTest.kt | 89 +++++++++++++++++++ 3 files changed, 153 insertions(+), 10 deletions(-) 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 8a8d529f..2baad4e1 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 @@ -79,6 +79,15 @@ object HermesWsClient { private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024 private const val OUTBOUND_DRAIN_TIMEOUT_MS = 60_000L + /** + * No inbound frame for this long means the link is dead even though the + * TCP session may still look open (silent NAT/VPN drop). OkHttp pings + * every 30s (OkHttpProvider.websocket) and any inbound frame refreshes + * [lastPongTimestamp], so a healthy link never exceeds ~30s of silence. + * 90s tolerates exactly one missed ping cycle before acting. + */ + private const val STALE_THRESHOLD_MS = 90_000L + // ── Internal state (all access through synchronized / atomic) ──────── private val requestId = AtomicInteger(0) @@ -135,13 +144,41 @@ object HermesWsClient { wsScope.launch { while (connected.get()) { delay(30_000L) - if (connected.get() && System.currentTimeMillis() - lastPongTimestamp > 60_000L) { - Log.w(TAG, "WebSocket connection appears unhealthy (no frames received for > 60s)") - } + if (!runHealthCheckPass()) break } } } + /** + * One liveness pass. Returns false when the tracking loop should stop: + * either the connection is already gone or the watchdog fired a cancel. + * + * Cancelling (not close()) forces an immediate onFailure, which rides the + * existing teardown path: generation bump, RECONNECTING, scheduleReconnect. + */ + private fun runHealthCheckPass(): Boolean { + if (!connected.get()) return false + val staleMs = System.currentTimeMillis() - lastPongTimestamp + if (staleMs <= STALE_THRESHOLD_MS) return true + Log.w(TAG, "WebSocket stale (${staleMs / 1000}s without frames) — cancelling to trigger reconnect") + synchronized(outboundLock) { + if (!connected.get()) return false + webSocket?.cancel() + } + return false + } + + /** + * Test hook: backdate liveness past the staleness threshold and force one + * synchronous watchdog pass, so reconnect tests are deterministic instead + * of racing the real 30s/90s cadence. + */ + @VisibleForTesting + internal fun forceHealthCheckForTest(staleMillis: Long) { + lastPongTimestamp = System.currentTimeMillis() - staleMillis + runHealthCheckPass() + } + private fun stopHealthTracking() { healthJob?.cancel() healthJob = null @@ -462,12 +499,18 @@ object HermesWsClient { } if (code in 200..299) { - val ticketMatch = Regex("""\"ticket\":\"([^\"]+)\"""").find(body) - val ticket = ticketMatch?.groupValues?.getOrNull(1) + // Real JSON decode — the ticket is base64url and may contain + // characters a naive [^"]+ regex cannot extract (see + // testGatedMode_parsesEscapedTicketFromRealJsonShape). + val parsed = + runCatching { + OkHttpProvider.json.decodeFromString(body) + }.getOrNull() + val ticket = parsed?.ticket if (!ticket.isNullOrBlank()) { return TicketRequestResult(ticket, null) } - Log.w(TAG, "WS ticket mint failed: response body did not contain ticket") + Log.w(TAG, "WS ticket mint failed: unparseable ticket response") } else { Log.w(TAG, "WS ticket mint failed: HTTP $code") return TicketRequestResult(null, code) @@ -926,10 +969,10 @@ object HermesWsClient { else -> Unit } - val emitted = parsedEvents.tryEmit(event) - if (!emitted && BuildConfig.DEBUG) { - Log.w(TAG, "WebSocket message dropped due to buffer overflow") - } + // tryEmit on a DROP_OLDEST flow only returns false when the + // buffer is full AND no subscriber is draining; with extraBuffer=512 + // and always-on init collectors this is unreachable in practice. + parsedEvents.tryEmit(event) } override fun onClosing( diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt index f5ebe880..8c82476a 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/JsonRpcModels.kt @@ -75,3 +75,14 @@ fun JsonElement.toAny(): Any? = map { it.toAny() } } } + +/** + * Response body of POST /api/auth/ws-ticket (gated mode). Verified against + * hermes-agent dashboard_auth/routes.py — the ticket is base64url and can + * therefore contain characters that only a real JSON decoder survives. + */ +@Serializable +data class WsTicketResponse( + val ticket: String, + val ttl_seconds: Int? = null, +) 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 c219109b..c8e8c102 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 @@ -1419,4 +1419,93 @@ class HermesWsClientTest { assertEquals(ConnectionStatus.RECONNECTING, HermesWsClient.connectionStatus.value) } + + @Test + fun testGatedMode_parsesEscapedTicketFromRealJsonShape() { + every { AuthManager.isAutoReconnect() } returns true + every { AuthManager.serverStore } returns + mockk().also { + every { it.getLatestState() } returns + com.m57.hermescontrol.data.config + .ServerStoreState(wsAuthParam = "ticket") + } + every { AuthManager.setToken(any()) } returns Unit + + val ticketServer = MockWebServer() + ticketServer.start() + // Real backend shape with an ESCAPED QUOTE inside the ticket value. + // Built from char vals so nobody has to count backslashes again: + // wire body == {"ticket":"ay","ttl_seconds":30} + // The old regex ([^"]+) stops at the inner dquote and extracts "a", + // which never equals the real ticket — only the JSON parser survives. + val bs = 92.toChar() // backslash + val dq = 34.toChar() // dquote + val wireTicket = "a" + bs + dq + "y" + + fun jsonEscape(s: String): String = s.replace("$bs", "$bs$bs").replace("$dq", "$bs$dq") + val wireBody = """{"ticket":"${jsonEscape(wireTicket)}","ttl_seconds":30}""" + ticketServer.enqueue(MockResponse().setResponseCode(200).setBody(wireBody)) + every { AuthManager.endpointForBuild() } returns + ServerEndpoint.parse( + ticketServer.url("/").toString(), + CleartextPolicy.ALLOW_WITH_WARNING, + ) + + HermesWsClient.connect() + + io.mockk.verify(timeout = 5000) { AuthManager.setToken(wireTicket) } + ticketServer.shutdown() + } + + @Test + fun testStaleSocketIsCancelledByWatchdog() { + var serverWebSocket: WebSocket? = null + val serverLatch = CountDownLatch(1) + mockWebServer.enqueue( + MockResponse().withWebSocketUpgrade( + object : WebSocketListener() { + override fun onOpen( + webSocket: WebSocket, + response: okhttp3.Response, + ) { + serverWebSocket = webSocket + serverLatch.countDown() + } + + override fun onMessage( + webSocket: WebSocket, + text: String, + ) { + // Any inbound frame refreshes liveness — send nothing, + // simulating a dead NAT'd link where only client pings flow. + } + }, + ), + ) + + HermesWsClient.connect() + runBlocking { + withTimeout(5000) { + HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } + } + } + assertTrue(serverLatch.await(5, TimeUnit.SECONDS)) + assertTrue(HermesWsClient.isConnected) + + // Backdate liveness past the staleness threshold, then force one + // synchronous watchdog pass. Deterministic — no real-time waiting. + HermesWsClient.forceHealthCheckForTest(staleMillis = 200_000L) + + // Cancel fires onFailure on an OkHttp thread; with auto-reconnect off + // (setUp default) the terminal state settles at DISCONNECTED. + val deadline = System.currentTimeMillis() + 5000 + while (HermesWsClient.isConnected && System.currentTimeMillis() < deadline) { + Thread.sleep(25) + } + assertFalse("Stale socket was not cancelled", HermesWsClient.isConnected) + assertTrue( + "Status stuck at CONNECTED after stale cancel", + HermesWsClient.connectionStatus.value != ConnectionStatus.CONNECTED, + ) + } }