feat(ws): adopt WS ping method for heartbeat and latency tracking (#1036)
This commit is contained in:
3 files changed
+99
No files matched your search
@@ -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<Long?>(null)
|
||||
|
||||
/** Last measured round-trip latency in milliseconds via [ping], or null when disconnected / unmeasured. */
|
||||
val lastLatencyMs: StateFlow<Long?> = _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 ─────────────────────────────────────────
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user