diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStore.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStore.kt index 5bf6623393..7fba544322 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStore.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStore.kt @@ -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>(persistentListOf()) val unhealthy: StateFlow> = _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() } } diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStoreCloseTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStoreCloseTest.kt new file mode 100644 index 0000000000..7f1be49b0f --- /dev/null +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStoreCloseTest.kt @@ -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() + } +} diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt index 5f0def4eac..7b59dc24f7 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt @@ -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) { diff --git a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/network/SleepResumeMonitor.kt b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/network/SleepResumeMonitor.kt new file mode 100644 index 0000000000..c2d8d3195d --- /dev/null +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/network/SleepResumeMonitor.kt @@ -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