diff --git a/app/src/main/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresher.kt b/app/src/main/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresher.kt
index 705b2195..d27b9e03 100644
--- a/app/src/main/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresher.kt
+++ b/app/src/main/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresher.kt
@@ -2,26 +2,64 @@ package com.m57.hermescontrol.data.remote
import com.m57.hermescontrol.data.local.AuthManager
import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
import okhttp3.OkHttpClient
import okhttp3.Request
+import okhttp3.Response
import okio.Buffer
internal object DashboardSessionTokenRefresher {
private val tokenPattern = Regex("""__HERMES_SESSION_TOKEN__\s*=\s*"([^"]+)"""")
+ private val refreshMutex = Mutex()
// Synchronized so concurrent 401s / reconnect storms don't each fire a
// redundant GET at the dashboard SPA (thundering-herd guard).
@Synchronized
fun refresh(): String? =
try {
- val token = fetch(AuthManager.baseUrl(), OkHttpProvider.probe) ?: return null
+ val token = fetchSync(AuthManager.baseUrl(), OkHttpProvider.probe) ?: return null
AuthManager.setToken(token)
token
} catch (_: Exception) {
null
}
- internal fun fetch(
+ suspend fun refreshAsync(): String? =
+ refreshMutex.withLock {
+ try {
+ val token = fetch(AuthManager.baseUrl(), OkHttpProvider.probe) ?: return@withLock null
+ AuthManager.setToken(token)
+ token
+ } catch (_: Exception) {
+ null
+ }
+ }
+
+ internal suspend fun fetch(
+ baseUrl: String,
+ client: OkHttpClient,
+ ): String? =
+ withContext(Dispatchers.IO) {
+ try {
+ val request =
+ Request
+ .Builder()
+ .url(baseUrl)
+ .get()
+ .build()
+
+ client.newCall(request).await().use { response ->
+ if (!response.isSuccessful) return@withContext null
+ extractToken(response)
+ }
+ } catch (_: Exception) {
+ null
+ }
+ }
+
+ internal fun fetchSync(
baseUrl: String,
client: OkHttpClient,
): String? =
@@ -32,32 +70,31 @@ internal object DashboardSessionTokenRefresher {
.url(baseUrl)
.get()
.build()
- // execute() + body read on Dispatchers.IO: the body read touches
- // the socket on the CALLING thread otherwise — when that is main
- // (loopback ticket mint from the kanban events connect), it throws
- // NetworkOnMainThreadException (same bug class as requestWsTicket).
- kotlinx.coroutines.runBlocking(Dispatchers.IO) {
- client.newCall(request).execute().use { response ->
- if (!response.isSuccessful) return@runBlocking null
- // Cap the read — the SPA HTML can be large, and we only need the
- // injected token near the top of the document. Reading the whole
- // body into a String risks OOM on low-end devices.
- val html =
- response.body.source().use { source ->
- val buffer = Buffer()
- source.read(buffer, MAX_BODY_BYTES)
- buffer.readUtf8()
- }
- tokenPattern
- .find(html)
- ?.groupValues
- ?.getOrNull(1)
- ?.takeIf { it.isNotBlank() }
- }
+
+ client.newCall(request).execute().use { response ->
+ if (!response.isSuccessful) return null
+ extractToken(response)
}
} catch (_: Exception) {
null
}
+ private fun extractToken(response: Response): String? {
+ // Cap the read — the SPA HTML can be large, and we only need the
+ // injected token near the top of the document. Reading the whole
+ // body into a String risks OOM on low-end devices.
+ val html =
+ response.body.source().use { source ->
+ val buffer = Buffer()
+ source.read(buffer, MAX_BODY_BYTES)
+ buffer.readUtf8()
+ }
+ return tokenPattern
+ .find(html)
+ ?.groupValues
+ ?.getOrNull(1)
+ ?.takeIf { it.isNotBlank() }
+ }
+
private const val MAX_BODY_BYTES: Long = 64 * 1024
}
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 ffc4ff17..042e5bf1 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
@@ -8,6 +8,7 @@ import com.m57.hermescontrol.data.remote.CookieManager
import com.m57.hermescontrol.data.remote.DashboardSessionTokenRefresher
import com.m57.hermescontrol.data.remote.NetworkMonitor
import com.m57.hermescontrol.data.remote.OkHttpProvider
+import com.m57.hermescontrol.data.remote.await
import com.m57.hermescontrol.data.session.ActiveSessionHolder
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CompletableDeferred
@@ -547,7 +548,7 @@ object HermesWsClient {
* or because ticket refresh succeeded). Returns false if we are in gated mode
* and ticket refresh failed or cannot be performed.
*/
- internal fun refreshWsTicketIfNeeded(generation: Int): Boolean {
+ internal suspend fun refreshWsTicketIfNeeded(generation: Int): Boolean {
val isGated =
try {
AuthManager.serverStore.getLatestState().wsAuthParam == "ticket"
@@ -563,11 +564,9 @@ object HermesWsClient {
// restart. Refresh it before each WebSocket handshake so automatic
// reconnect does not get stuck in AUTH_EXPIRED with a stale token.
val token =
- synchronized(DashboardSessionTokenRefresher) {
- runCatching {
- DashboardSessionTokenRefresher.fetch(AuthManager.baseUrl(), OkHttpProvider.probe)
- }.getOrNull()
- }
+ runCatching {
+ DashboardSessionTokenRefresher.refreshAsync()
+ }.getOrNull()
synchronized(outboundLock) {
if (isCurrentGeneration(generation) && token != null) {
runCatching { AuthManager.setToken(token) }
@@ -610,7 +609,7 @@ object HermesWsClient {
* ticket. Loopback mode: refresh the dashboard token and return it.
* Returns null when a ticket could not be obtained.
*/
- internal fun mintWsTicket(): String? {
+ internal suspend fun mintWsTicket(): String? {
val isGated =
try {
AuthManager.serverStore.getLatestState().wsAuthParam == "ticket"
@@ -618,7 +617,7 @@ object HermesWsClient {
false
}
if (!isGated) {
- DashboardSessionTokenRefresher.refresh()
+ DashboardSessionTokenRefresher.refreshAsync()
return AuthManager.getToken()
}
return requestWsTicket().ticket
@@ -631,7 +630,7 @@ object HermesWsClient {
)
/** POST /api/auth/ws-ticket (cookie-auth'd via the shared CookieJar) and parse the ticket. */
- private fun requestWsTicket(): TicketRequestResult {
+ private suspend fun requestWsTicket(): TicketRequestResult {
try {
val client = OkHttpProvider.probe
val request =
@@ -641,16 +640,12 @@ object HermesWsClient {
.post("{}".toRequestBody())
.build()
- // Run the ENTIRE call on Dispatchers.IO — execute() already hops,
- // but ResponseBody.string() reads the socket on the CALLING
- // thread. When the caller is main (kanban events connect), that
- // read throws NetworkOnMainThreadException and the mint fails.
val (code, body) =
- kotlinx.coroutines.runBlocking(Dispatchers.IO) {
+ withContext(Dispatchers.IO) {
if (CookieManager.isInitialized()) {
CookieManager.useStore(CookieManager.cookieJar.currentServer())
}
- client.newCall(request).execute().use { resp ->
+ client.newCall(request).await().use { resp ->
resp.code to resp.body.string()
}
}
@@ -1184,22 +1179,24 @@ object HermesWsClient {
if (intentionalClose.get()) return
connectionGeneration.incrementAndGet()
}
- if (!refreshWsTicketIfNeeded(generation)) {
- Log.w(TAG, "Aborting openSocket: WS ticket refresh failed")
- return
- }
- if (!isCurrentGeneration(generation)) return
- val url = AuthManager.wsUrl()
- val safeUrl = url.replace(Regex("token=[^&]+"), "token=REDACTED")
- if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl")
+ wsScope.launch(Dispatchers.IO) {
+ if (!refreshWsTicketIfNeeded(generation)) {
+ Log.w(TAG, "Aborting openSocket: WS ticket refresh failed")
+ return@launch
+ }
+ if (!isCurrentGeneration(generation)) return@launch
+ val url = AuthManager.wsUrl()
+ val safeUrl = url.replace(Regex("token=[^&]+"), "token=REDACTED")
+ if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl")
- val request = Request.Builder().url(url).build()
- val newSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl(generation))
- synchronized(outboundLock) {
- if (connectionGeneration.get() == generation && !intentionalClose.get()) {
- webSocket = newSocket
- } else {
- newSocket.cancel()
+ val request = Request.Builder().url(url).build()
+ val newSocket = OkHttpProvider.websocket.newWebSocket(request, WsListenerImpl(generation))
+ synchronized(outboundLock) {
+ if (connectionGeneration.get() == generation && !intentionalClose.get()) {
+ webSocket = newSocket
+ } else {
+ newSocket.cancel()
+ }
}
}
}
diff --git a/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt b/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt
index 2551a208..6b5f4cd6 100644
--- a/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt
+++ b/app/src/main/java/com/m57/hermescontrol/data/ws/KanbanEventsClient.kt
@@ -5,6 +5,7 @@ import com.m57.hermescontrol.BuildConfig
import com.m57.hermescontrol.data.local.AuthManager
import com.m57.hermescontrol.data.remote.OkHttpProvider
import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
@@ -163,86 +164,89 @@ class KanbanEventsClient {
) {
if (closed || gen != generation) return
- val credential = HermesWsClient.mintWsTicket()
- if (credential.isNullOrBlank()) {
- onStatus(KanbanLiveStatus.AUTH_FAILED)
- return
- }
+ scope.launch(Dispatchers.IO) {
+ val credential = HermesWsClient.mintWsTicket()
+ if (closed || gen != generation) return@launch
+ if (credential.isNullOrBlank()) {
+ scope.launch { onStatus(KanbanLiveStatus.AUTH_FAILED) }
+ return@launch
+ }
- val rawAuthParam = AuthManager.serverStore.getLatestState().wsAuthParam
- val authParam = if (rawAuthParam.isBlank()) "token" else rawAuthParam
- val url =
- AuthManager.endpointForBuild().webSocketUrl(
- path = KANBAN_EVENTS_PATH,
- authParameter = authParam,
- credential = credential,
- extraParams = mapOf("since" to cursor.toString(), "board" to board),
- )
- val safeUrl = url.replace(Regex("(token|ticket)=[^&]+"), "$1=REDACTED")
- if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl")
+ val rawAuthParam = AuthManager.serverStore.getLatestState().wsAuthParam
+ val authParam = if (rawAuthParam.isBlank()) "token" else rawAuthParam
+ val url =
+ AuthManager.endpointForBuild().webSocketUrl(
+ path = KANBAN_EVENTS_PATH,
+ authParameter = authParam,
+ credential = credential,
+ extraParams = mapOf("since" to cursor.toString(), "board" to board),
+ )
+ val safeUrl = url.replace(Regex("(token|ticket)=[^&]+"), "$1=REDACTED")
+ if (BuildConfig.DEBUG) Log.d(TAG, "Connecting to $safeUrl")
- val request = Request.Builder().url(url).build()
- webSocket =
- httpClient.newWebSocket(
- request,
- object : WebSocketListener() {
- override fun onOpen(
- webSocket: WebSocket,
- response: Response,
- ) {
- if (closed || gen != generation) return
- backoffMs = INITIAL_BACKOFF_MS
- scope.launch { onStatus(KanbanLiveStatus.CONNECTED) }
- }
-
- override fun onMessage(
- webSocket: WebSocket,
- text: String,
- ) {
- if (closed || gen != generation) return
- val envelope = parseKanbanEventsFrame(text) ?: return
- if (envelope.events.isEmpty()) return
- envelope.cursor?.let { cursor = it }
- scope.launch { onEvents(envelope) }
- }
-
- override fun onMessage(
- webSocket: WebSocket,
- bytes: ByteString,
- ) = Unit
-
- override fun onClosing(
- webSocket: WebSocket,
- code: Int,
- reason: String,
- ) {
- webSocket.close(code, reason)
- }
-
- override fun onClosed(
- webSocket: WebSocket,
- code: Int,
- reason: String,
- ) {
- if (closed || gen != generation) return
- if (code == WS_1008_POLICY_VIOLATION) {
- scope.launch { onStatus(KanbanLiveStatus.AUTH_FAILED) }
- return
+ val request = Request.Builder().url(url).build()
+ webSocket =
+ httpClient.newWebSocket(
+ request,
+ object : WebSocketListener() {
+ override fun onOpen(
+ webSocket: WebSocket,
+ response: Response,
+ ) {
+ if (closed || gen != generation) return
+ backoffMs = INITIAL_BACKOFF_MS
+ scope.launch { onStatus(KanbanLiveStatus.CONNECTED) }
}
- scheduleReconnect(scope, gen, board, onEvents, onStatus)
- }
- override fun onFailure(
- webSocket: WebSocket,
- t: Throwable,
- response: Response?,
- ) {
- if (closed || gen != generation) return
- if (BuildConfig.DEBUG) Log.d(TAG, "Stream failure: ${t.javaClass.simpleName}")
- scheduleReconnect(scope, gen, board, onEvents, onStatus)
- }
- },
- )
+ override fun onMessage(
+ webSocket: WebSocket,
+ text: String,
+ ) {
+ if (closed || gen != generation) return
+ val envelope = parseKanbanEventsFrame(text) ?: return
+ if (envelope.events.isEmpty()) return
+ envelope.cursor?.let { cursor = it }
+ scope.launch { onEvents(envelope) }
+ }
+
+ override fun onMessage(
+ webSocket: WebSocket,
+ bytes: ByteString,
+ ) = Unit
+
+ override fun onClosing(
+ webSocket: WebSocket,
+ code: Int,
+ reason: String,
+ ) {
+ webSocket.close(code, reason)
+ }
+
+ override fun onClosed(
+ webSocket: WebSocket,
+ code: Int,
+ reason: String,
+ ) {
+ if (closed || gen != generation) return
+ if (code == WS_1008_POLICY_VIOLATION) {
+ scope.launch { onStatus(KanbanLiveStatus.AUTH_FAILED) }
+ return
+ }
+ scheduleReconnect(scope, gen, board, onEvents, onStatus)
+ }
+
+ override fun onFailure(
+ webSocket: WebSocket,
+ t: Throwable,
+ response: Response?,
+ ) {
+ if (closed || gen != generation) return
+ if (BuildConfig.DEBUG) Log.d(TAG, "Stream failure: ${t.javaClass.simpleName}")
+ scheduleReconnect(scope, gen, board, onEvents, onStatus)
+ }
+ },
+ )
+ }
}
private fun scheduleReconnect(
diff --git a/app/src/test/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresherTest.kt b/app/src/test/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresherTest.kt
index 84f4844f..76b42b72 100644
--- a/app/src/test/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresherTest.kt
+++ b/app/src/test/java/com/m57/hermescontrol/data/remote/DashboardSessionTokenRefresherTest.kt
@@ -5,6 +5,7 @@ import io.mockk.every
import io.mockk.mockkObject
import io.mockk.unmockkAll
import io.mockk.verify
+import kotlinx.coroutines.runBlocking
import okhttp3.OkHttpClient
import okhttp3.mockwebserver.MockResponse
import okhttp3.mockwebserver.MockWebServer
@@ -32,50 +33,68 @@ class DashboardSessionTokenRefresherTest {
}
@Test
- fun fetchExtractsInjectedDashboardToken() {
+ fun fetchExtractsInjectedDashboardToken() =
+ runBlocking {
+ server.enqueue(
+ MockResponse()
+ .setResponseCode(200)
+ .setBody(""""""),
+ )
+
+ val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
+
+ assertEquals("new-token", token)
+ assertEquals("/", server.takeRequest().path)
+ }
+
+ @Test
+ fun fetchReturnsNullWhenDashboardDoesNotInjectToken() =
+ runBlocking {
+ server.enqueue(MockResponse().setResponseCode(200).setBody(""))
+
+ val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
+
+ assertNull(token)
+ }
+
+ @Test
+ fun fetchReturnsNullForFailedResponse() =
+ runBlocking {
+ server.enqueue(MockResponse().setResponseCode(500))
+
+ val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
+
+ assertNull(token)
+ }
+
+ @Test
+ fun fetchReturnsNullWhenTokenIsBlank() =
+ runBlocking {
+ server.enqueue(
+ MockResponse()
+ .setResponseCode(200)
+ .setBody(""""""),
+ )
+
+ val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
+
+ assertNull(token)
+ }
+
+ @Test
+ fun fetchSyncExtractsInjectedDashboardToken() {
server.enqueue(
MockResponse()
.setResponseCode(200)
- .setBody(""""""),
+ .setBody(""""""),
)
- val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
+ val token = DashboardSessionTokenRefresher.fetchSync(server.url("/").toString(), OkHttpClient())
- assertEquals("new-token", token)
+ assertEquals("sync-token", token)
assertEquals("/", server.takeRequest().path)
}
- @Test
- fun fetchReturnsNullWhenDashboardDoesNotInjectToken() {
- server.enqueue(MockResponse().setResponseCode(200).setBody(""))
-
- val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
-
- assertNull(token)
- }
-
- @Test
- fun fetchReturnsNullForFailedResponse() {
- server.enqueue(MockResponse().setResponseCode(500))
-
- val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
-
- assertNull(token)
- }
-
- @Test
- fun fetchReturnsNullWhenTokenIsBlank() {
- server.enqueue(
- MockResponse()
- .setResponseCode(200)
- .setBody(""""""),
- )
-
- val token = DashboardSessionTokenRefresher.fetch(server.url("/").toString(), OkHttpClient())
-
- assertNull(token)
- }
-
@Test
fun refreshUpdatesAuthManagerOnSuccess() {
mockkObject(AuthManager)
@@ -94,6 +113,25 @@ class DashboardSessionTokenRefresherTest {
verify { AuthManager.setToken("mocked-token") }
}
+ @Test
+ fun refreshAsyncUpdatesAuthManagerOnSuccess() =
+ runBlocking {
+ mockkObject(AuthManager)
+ every { AuthManager.baseUrl() } returns server.url("/").toString()
+ every { AuthManager.setToken(any()) } returns Unit
+
+ server.enqueue(
+ MockResponse()
+ .setResponseCode(200)
+ .setBody(""""""),
+ )
+
+ val token = DashboardSessionTokenRefresher.refreshAsync()
+
+ assertEquals("async-token", token)
+ verify { AuthManager.setToken("async-token") }
+ }
+
@Test
fun refreshReturnsNullOnException() {
mockkObject(AuthManager)
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 4829bd1b..6e2016f0 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
@@ -10,6 +10,7 @@ import com.m57.hermescontrol.data.remote.OkHttpProvider
import com.m57.hermescontrol.data.remote.ServerEndpoint
import com.m57.hermescontrol.data.remote.buildFakePersistentCookieJar
import com.m57.hermescontrol.data.session.ActiveSessionHolder
+import io.mockk.coEvery
import io.mockk.every
import io.mockk.mockk
import io.mockk.mockkObject
@@ -84,6 +85,10 @@ class HermesWsClientTest {
// the WS stack can build its OkHttp clients without app context.
CookieManager.setJarForTest(buildFakePersistentCookieJar())
+ mockkObject(DashboardSessionTokenRefresher)
+ coEvery { DashboardSessionTokenRefresher.refreshAsync() } returns null
+ every { DashboardSessionTokenRefresher.refresh() } returns null
+
// Reset state
val connectedField = HermesWsClient::class.java.getDeclaredField("connected")
connectedField.isAccessible = true
@@ -850,12 +855,11 @@ class HermesWsClientTest {
@Test
fun testDisconnectThenConnectSupersedesBlockedSocketOpen() =
runBlocking {
- mockkObject(DashboardSessionTokenRefresher)
every { AuthManager.baseUrl() } returns mockWebServer.url("/").toString()
val refreshStarted = CountDownLatch(1)
val releaseRefresh = CountDownLatch(1)
val refreshCalls = AtomicInteger(0)
- every { DashboardSessionTokenRefresher.fetch(any(), any()) } answers {
+ coEvery { DashboardSessionTokenRefresher.refreshAsync() } answers {
if (refreshCalls.getAndIncrement() == 0) {
refreshStarted.countDown()
releaseRefresh.await(5, TimeUnit.SECONDS)
@@ -1181,10 +1185,8 @@ class HermesWsClientTest {
// 5-6s latch windows routinely overshoot under parallel load).
HermesWsClient.setReconnectBackoffForTest(0L)
// The reconnect path refreshes the WS token over the network before
- // opening the socket. Stub it so no real HTTP call sits inside the
+ // opening the socket. Stubbed in setUp() so no real HTTP call sits inside the
// test's timing window (CI network latency was the dominant flake).
- mockkObject(DashboardSessionTokenRefresher)
- every { DashboardSessionTokenRefresher.fetch(any(), any()) } returns null
var serverSocket1: WebSocket? = null
var serverSocket2: WebSocket? = null
@@ -1523,6 +1525,11 @@ class HermesWsClientTest {
HermesWsClient.connect()
+ runBlocking {
+ withTimeout(5000) {
+ HermesWsClient.connectionStatus.first { it == ConnectionStatus.DISCONNECTED }
+ }
+ }
assertEquals(ConnectionStatus.DISCONNECTED, HermesWsClient.connectionStatus.value)
ticketServer.shutdown()
}
@@ -1549,6 +1556,11 @@ class HermesWsClientTest {
HermesWsClient.connect()
+ runBlocking {
+ withTimeout(5000) {
+ HermesWsClient.connectionStatus.first { it == ConnectionStatus.RECONNECTING }
+ }
+ }
assertEquals(ConnectionStatus.RECONNECTING, HermesWsClient.connectionStatus.value)
}