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 <hy4ri@users.noreply.github.com>
This commit is contained in:
1 parent
a11761f120
commit
bb400881e6
3 files changed
+152
-9
No files matched your search
@@ -79,6 +79,15 @@ object HermesWsClient {
|
|||||||
private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024
|
private const val MAX_OUTBOUND_MESSAGE_BYTES = 16 * 1024 * 1024
|
||||||
private const val OUTBOUND_DRAIN_TIMEOUT_MS = 60_000L
|
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) ────────
|
// ── Internal state (all access through synchronized / atomic) ────────
|
||||||
|
|
||||||
private val requestId = AtomicInteger(0)
|
private val requestId = AtomicInteger(0)
|
||||||
@@ -135,11 +144,39 @@ object HermesWsClient {
|
|||||||
wsScope.launch {
|
wsScope.launch {
|
||||||
while (connected.get()) {
|
while (connected.get()) {
|
||||||
delay(30_000L)
|
delay(30_000L)
|
||||||
if (connected.get() && System.currentTimeMillis() - lastPongTimestamp > 60_000L) {
|
if (!runHealthCheckPass()) break
|
||||||
Log.w(TAG, "WebSocket connection appears unhealthy (no frames received for > 60s)")
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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() {
|
private fun stopHealthTracking() {
|
||||||
@@ -462,12 +499,18 @@ object HermesWsClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (code in 200..299) {
|
if (code in 200..299) {
|
||||||
val ticketMatch = Regex("""\"ticket\":\"([^\"]+)\"""").find(body)
|
// Real JSON decode — the ticket is base64url and may contain
|
||||||
val ticket = ticketMatch?.groupValues?.getOrNull(1)
|
// characters a naive [^"]+ regex cannot extract (see
|
||||||
|
// testGatedMode_parsesEscapedTicketFromRealJsonShape).
|
||||||
|
val parsed =
|
||||||
|
runCatching {
|
||||||
|
OkHttpProvider.json.decodeFromString<WsTicketResponse>(body)
|
||||||
|
}.getOrNull()
|
||||||
|
val ticket = parsed?.ticket
|
||||||
if (!ticket.isNullOrBlank()) {
|
if (!ticket.isNullOrBlank()) {
|
||||||
return TicketRequestResult(ticket, null)
|
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 {
|
} else {
|
||||||
Log.w(TAG, "WS ticket mint failed: HTTP $code")
|
Log.w(TAG, "WS ticket mint failed: HTTP $code")
|
||||||
return TicketRequestResult(null, code)
|
return TicketRequestResult(null, code)
|
||||||
@@ -926,10 +969,10 @@ object HermesWsClient {
|
|||||||
|
|
||||||
else -> Unit
|
else -> Unit
|
||||||
}
|
}
|
||||||
val emitted = parsedEvents.tryEmit(event)
|
// tryEmit on a DROP_OLDEST flow only returns false when the
|
||||||
if (!emitted && BuildConfig.DEBUG) {
|
// buffer is full AND no subscriber is draining; with extraBuffer=512
|
||||||
Log.w(TAG, "WebSocket message dropped due to buffer overflow")
|
// and always-on init collectors this is unreachable in practice.
|
||||||
}
|
parsedEvents.tryEmit(event)
|
||||||
}
|
}
|
||||||
|
|
||||||
override fun onClosing(
|
override fun onClosing(
|
||||||
|
|||||||
@@ -75,3 +75,14 @@ fun JsonElement.toAny(): Any? =
|
|||||||
map { it.toAny() }
|
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,
|
||||||
|
)
|
||||||
@@ -1419,4 +1419,93 @@ class HermesWsClientTest {
|
|||||||
|
|
||||||
assertEquals(ConnectionStatus.RECONNECTING, HermesWsClient.connectionStatus.value)
|
assertEquals(ConnectionStatus.RECONNECTING, HermesWsClient.connectionStatus.value)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun testGatedMode_parsesEscapedTicketFromRealJsonShape() {
|
||||||
|
every { AuthManager.isAutoReconnect() } returns true
|
||||||
|
every { AuthManager.serverStore } returns
|
||||||
|
mockk<com.m57.hermescontrol.data.config.ServerStore>().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":"a<backslash><dquote>y","ttl_seconds":30}
|
||||||
|
// The old regex ([^"]+) stops at the inner dquote and extracts "a<bs>",
|
||||||
|
// 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,
|
||||||
|
)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Reference in new issue
Block a user