Merge pull request #3221 from nrobi144/fix/relay-health-threading-and-sleep-resume

fix: RelayHealthStore threading + sleep-resume socket recovery (follow-up to #3186)
2026-08-11 00:37:41 +00:00 · 2026-06-16 08:40:37 -04:00
parent 72c0558bb9 552540e77d
commit 085e2466e6
4 changed files with 234 additions and 5 deletions
@@ -37,6 +37,7 @@ import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
+import kotlin.concurrent.Volatile

 /**
 * Per-account, durable record of relay liveness used to drive the "unhealthy relay" review UI.
@@ -55,6 +56,10 @@ class RelayHealthStore(
    private val persistence: RelayHealthPersistence,
    private val torEnabledProvider: () -> Boolean = { false },
    parentScope: CoroutineScope? = null,
+    // Caller-supplied dispatcher used for both classification and persistence I/O.
+    // Default is `Dispatchers.Default` so commonMain stays iOS-compatible — JVM hosts
+    // (Android/Desktop) should pass `Dispatchers.IO` so `prefs.flush()` doesn't sit on
+    // a CPU-bound worker.
    private val ioDispatcher: CoroutineDispatcher = Dispatchers.Default,
 ) {
    companion object {
@@ -84,8 +89,11 @@ class RelayHealthStore(
    private val _unhealthy = MutableStateFlow<PersistentList<UnhealthyRelay>>(persistentListOf())
    val unhealthy: StateFlow<PersistentList<UnhealthyRelay>> = _unhealthy.asStateFlow()

-    private var persistJob: Job? = null
-    private var tickJob: Job? = null
+    @Volatile private var persistJob: Job? = null
+
+    @Volatile private var tickJob: Job? = null
+
+    @Volatile private var closed = false

    init {
        // Persist the firstScanAt seed if we just stamped it.
@@ -214,20 +222,38 @@ class RelayHealthStore(
    }

    private fun schedulePersist() {
+        if (closed) return
        persistJob?.cancel()
        persistJob =
            scope.launch {
                delay(PERSIST_DEBOUNCE_MS)
                val snapshot = state.value
-                runCatching { persistence.save(snapshot) }
+                withContext(ioDispatcher) {
+                    runCatching { persistence.save(snapshot) }
+                }
            }
    }

+    /**
+     * Tear down internal jobs and fire the final persist off-thread. Safe to call from
+     * the composition / Main thread: the blocking I/O is dispatched to [ioDispatcher]
+     * on a detached, self-cancelling scope so the last debounce window isn't lost when
+     * the parent composition scope is about to cancel.
+     */
    fun close() {
-        // Flush pending writes synchronously before tearing down.
+        if (closed) return
+        closed = true
        persistJob?.cancel()
-        runCatching { persistence.save(state.value) }
        tickJob?.cancel()
+        val finalSnapshot = state.value
+        val flushScope = CoroutineScope(SupervisorJob() + ioDispatcher)
+        flushScope.launch {
+            try {
+                runCatching { persistence.save(finalSnapshot) }
+            } finally {
+                flushScope.cancel()
+            }
+        }
        if (ownsScope) scope.cancel()
    }
 }
@@ -0,0 +1,123 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.relays.health
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.test.StandardTestDispatcher
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runTest
+import kotlin.concurrent.Volatile
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class RelayHealthStoreCloseTest {
+    private class CountingPersistence : RelayHealthPersistence {
+        @Volatile var saves: Int = 0
+
+        @Volatile var lastSnapshot: RelayHealthSnapshot? = null
+
+        override fun load(): RelayHealthSnapshot = RelayHealthSnapshot()
+
+        override fun save(snapshot: RelayHealthSnapshot) {
+            saves++
+            lastSnapshot = snapshot
+        }
+    }
+
+    private val url = RelayUrlNormalizer.normalizeOrNull("wss://example.com")!!
+
+    @Test
+    fun close_is_idempotent() =
+        runTest {
+            val persistence = CountingPersistence()
+            val dispatcher = StandardTestDispatcher(testScheduler)
+            val scope = CoroutineScope(SupervisorJob() + dispatcher)
+            val store =
+                RelayHealthStore(
+                    persistence = persistence,
+                    parentScope = scope,
+                    ioDispatcher = dispatcher,
+                )
+
+            store.close()
+            store.close()
+            store.close()
+
+            advanceUntilIdle()
+            // First close runs a fire-and-forget final save; later closes are no-ops.
+            assertEquals(1, persistence.saves)
+            scope.cancel()
+        }
+
+    @Test
+    fun recordIncoming_after_close_does_not_schedule_persist() =
+        runTest {
+            val persistence = CountingPersistence()
+            val dispatcher = StandardTestDispatcher(testScheduler)
+            val scope = CoroutineScope(SupervisorJob() + dispatcher)
+            val store =
+                RelayHealthStore(
+                    persistence = persistence,
+                    parentScope = scope,
+                    ioDispatcher = dispatcher,
+                )
+
+            advanceUntilIdle()
+            val baseline = persistence.saves
+
+            store.close()
+            advanceUntilIdle()
+            val afterClose = persistence.saves
+            assertTrue(afterClose >= baseline + 1, "close should run the final save")
+
+            store.recordIncoming(url, atSeconds = 1_700_000_000L)
+            advanceUntilIdle()
+            // closed → schedulePersist short-circuits, no extra save.
+            assertEquals(afterClose, persistence.saves)
+            scope.cancel()
+        }
+
+    @Test
+    fun close_with_internal_scope_owns_its_lifecycle() =
+        runTest {
+            val persistence = CountingPersistence()
+            val store =
+                RelayHealthStore(
+                    persistence = persistence,
+                    parentScope = null, // store owns its scope
+                    ioDispatcher = Dispatchers.Default,
+                )
+
+            // Sanity: schedulePersist on internal scope worked.
+            store.recordConnect(url, atSeconds = 1_700_000_000L)
+
+            store.close()
+            // Idempotent + does not throw.
+            store.close()
+        }
+}
@@ -839,6 +839,18 @@ fun App(
        relayManager.startMetricsSnapshot(this)
    }

+    // Detect host-machine sleep/wake: after a long delay overshoot the OkHttp
+    // sockets we held are dead even though needsToReconnect() still reads false,
+    // so force a hard disconnect+connect. See SleepResumeMonitor.kt.
+    LaunchedEffect(relayManager) {
+        com.vitorpamplona.amethyst.desktop.network.runSleepResumeMonitor {
+            relayManager.client.reconnect(
+                onlyIfChanged = false,
+                ignoreRetryDelays = true,
+            )
+        }
+    }
+
    // Subscriptions coordinator — uses default relay URLs for metadata indexing.
    // Feed subscriptions (inside MainContent) drive actual relay pool connections.
    val subscriptionsCoordinator =
@@ -1441,6 +1453,8 @@ fun MainContent(
                    torStateForHealth.settings.torType != com.vitorpamplona.amethyst.commons.tor.TorType.OFF
                },
                parentScope = scope,
+                // `prefs.flush()` is blocking — keep it off the composition scope's Main dispatcher.
+                ioDispatcher = kotlinx.coroutines.Dispatchers.IO,
            )
        }
    DisposableEffect(relayHealthStore, relayManager) {
@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.desktop.network
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlin.coroutines.coroutineContext
+
+/**
+ * Detects host-machine sleep/wake transitions by watching wall-clock overshoot
+ * of a tight delay loop. When the OS suspends the JVM, [delay] returns far past
+ * its scheduled deadline; the OkHttp websocket connections we held are dead by
+ * then even though [com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.BasicRelayClient]
+ * still reports `isConnected() == true` until the next ping fails — so the
+ * standard keep-alive reconnect path is a no-op and the offline banner stays
+ * stuck until a manual reload.
+ *
+ * Lives here in the desktop app (per Vitor's review of #3221) instead of the
+ * cross-platform NostrClient because sleep/wake semantics differ across
+ * platforms — Android has Doze + network change broadcasts, iOS has app
+ * lifecycle events, and macOS/Linux/Windows desktops can grow real OS-level
+ * sleep hooks here later (NSWorkspace notifications, D-Bus PrepareForSleep,
+ * WM_POWERBROADCAST) without touching Quartz.
+ *
+ * Real OS sleep events would be more precise, but the wall-clock heuristic
+ * needs zero native deps and catches the symptom for v1.
+ */
+suspend fun runSleepResumeMonitor(
+    intervalMs: Long = DEFAULT_INTERVAL_MS,
+    wakeThresholdMs: Long = DEFAULT_WAKE_THRESHOLD_MS,
+    nowMs: () -> Long = { System.currentTimeMillis() },
+    onWake: () -> Unit,
+) {
+    var lastTickMs = nowMs()
+    while (coroutineContext.isActive) {
+        delay(intervalMs)
+        val now = nowMs()
+        val elapsed = now - lastTickMs
+        lastTickMs = now
+        if (elapsed > wakeThresholdMs) onWake()
+    }
+}
+
+private const val DEFAULT_INTERVAL_MS: Long = 60_000L
+
+// 5x the tick — wide enough to ignore GC stalls / brief scheduler hiccups, tight
+// enough to recover quickly after a real sleep.
+private const val DEFAULT_WAKE_THRESHOLD_MS: Long = 5 * DEFAULT_INTERVAL_MS