From d8385a2b0294dab8fdf10906054b71b474e8d15a Mon Sep 17 00:00:00 2001 From: M57 <41645758+Hy4ri@users.noreply.github.com> Date: Tue, 1 Sep 2026 19:44:39 +0300 Subject: [PATCH] feat(ws): adopt WS ping method for heartbeat and latency tracking (#1036) --- .../hermescontrol/data/ws/HermesWsClient.kt | 33 ++++++++++ .../m57/hermescontrol/data/ws/WsMethods.kt | 5 ++ .../data/ws/HermesWsClientTest.kt | 61 +++++++++++++++++++ 3 files changed, 99 insertions(+) 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 335ddaa5..a1d9c710 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 @@ -152,6 +152,12 @@ object HermesWsClient { val isHealthy: Boolean get() = isConnected && (System.currentTimeMillis() - lastPongTimestamp < 60_000L) + // ── Observable latency tracking (issue #1017) ───────────────────────── + private val _lastLatencyMs = MutableStateFlow(null) + + /** Last measured round-trip latency in milliseconds via [ping], or null when disconnected / unmeasured. */ + val lastLatencyMs: StateFlow = _lastLatencyMs.asStateFlow() + private var healthJob: Job? = null private fun startHealthTracking() { @@ -161,11 +167,37 @@ object HermesWsClient { wsScope.launch { while (connected.get()) { delay(30_000L) + if (connected.get()) { + try { + ping(timeoutMs = 10_000L) + } catch (e: Exception) { + Log.w(TAG, "Health check ping failed: ${e.message}") + } + } if (!runHealthCheckPass()) break } } } + /** + * Ultra-lightweight JSON-RPC liveness check and RTT measurement (issue #1017). + * Bypasses agent queues on the backend; measures round-trip time in milliseconds. + * Updates [lastLatencyMs] and [lastPongTimestamp]. + */ + suspend fun ping(timeoutMs: Long = 5_000L): Long { + val start = System.currentTimeMillis() + request(WsMethods.PING, emptyMap(), timeoutMs = timeoutMs).await() + val latency = (System.currentTimeMillis() - start).coerceAtLeast(0L) + lastPongTimestamp = System.currentTimeMillis() + _lastLatencyMs.value = latency + return latency + } + + @VisibleForTesting + internal fun setLastLatencyForTest(latency: Long?) { + _lastLatencyMs.value = latency + } + /** * One liveness pass. Returns false when the tracking loop should stop: * either the connection is already gone or the watchdog fired a cancel. @@ -199,6 +231,7 @@ object HermesWsClient { private fun stopHealthTracking() { healthJob?.cancel() healthJob = null + _lastLatencyMs.value = null } // ── Public observable stream ───────────────────────────────────────── diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt index 92398aaf..dd00f6b5 100644 --- a/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt +++ b/app/src/main/java/com/m57/hermescontrol/data/ws/WsMethods.kt @@ -37,6 +37,11 @@ object WsMethods { const val SUDO_RESPOND = "sudo.respond" const val SECRET_RESPOND = "secret.respond" + // ── Heartbeat & Latency (issue #1017) ────────────────────────────────── + + /** Ultra-lightweight liveness check and round-trip latency measurement. */ + const val PING = "ping" + // ── Commands catalog ────────────────────────────────────────────────── const val COMMANDS_CATALOG = "commands.catalog" const val COMMAND_DISPATCH = "command.dispatch" 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 742da1f2..7994b464 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 @@ -1688,4 +1688,65 @@ class HermesWsClientTest { Thread.sleep(200) assertEquals(6, HermesWsClient.getSeqWatermarks()["s1"]) } + + @Test + fun testPingMethodConstant() { + assertEquals("ping", WsMethods.PING) + } + + @Test + fun testPingMeasuresLatencyAndUpdatesTimestamp() { + var serverWebSocket: WebSocket? = null + val serverLatch = CountDownLatch(1) + val pingLatch = CountDownLatch(1) + var pingReceived = false + + 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, + ) { + if (text.contains(""""method":"ping"""")) { + pingReceived = true + val id = Regex(""""id":"([^"]+)"""").find(text)?.groupValues?.get(1) ?: "1" + webSocket.send( + """{"jsonrpc":"2.0","id":"$id","result":{"pong":true,"timestamp":1700000000.0}}""", + ) + pingLatch.countDown() + } + } + }, + ), + ) + + HermesWsClient.connect() + runBlocking { + withTimeout(5000) { + HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } + } + } + assertTrue(serverLatch.await(5, TimeUnit.SECONDS)) + + val latency = + runBlocking { + HermesWsClient.ping(timeoutMs = 5000) + } + + assertTrue(pingLatch.await(5, TimeUnit.SECONDS)) + assertTrue(pingReceived) + assertTrue(latency >= 0) + assertNotNull(HermesWsClient.lastLatencyMs.value) + assertEquals(latency, HermesWsClient.lastLatencyMs.value) + assertTrue(HermesWsClient.lastPongTimestamp > 0) + } }