From b91519157e7907f6cbeb992504219fe6260db2d8 Mon Sep 17 00:00:00 2001 From: nrobi144 Date: Mon, 15 Jun 2026 13:21:32 +0300 Subject: [PATCH 1/2] fix(commons,quartz): RelayHealthStore threading + sleep-resume socket recovery Follow-up to #3186, addressing the unresolved review feedback: - RelayHealthStore.schedulePersist() wrapped the blocking save() in withContext(ioDispatcher) so prefs.flush() no longer sits on the Compose composition thread on Desktop. - close() now fires the final save on a detached IO-bound scope instead of blocking the composition thread for ~50ms during account switch / app exit. - @Volatile on persistJob/tickJob and a closed-flag guard so the relay-network thread and composition thread no longer race on plain vars (and post-close work is dropped). - desktopApp/Main.kt passes Dispatchers.IO to RelayHealthStore so persistence flushes land on the IO dispatcher instead of Dispatchers.Default. Plus a separate-but-related fix to the offline-banner-stuck-after-Mac-sleep issue: NostrClient.keepAliveJob now tracks wall-clock overshoot of its scheduled tick. If the OS suspended us (laptop lid closed, system sleep), delay() returns far past its deadline and the OkHttp websockets we held are dead even though BasicRelayClient.isConnected() still reads true until the next ping fails. On a >5x interval overshoot, force relayPool.disconnect() + connect() instead of trusting needsToReconnect(), so feeds resume without an app restart. --- .../commons/relays/health/RelayHealthStore.kt | 36 ++++- .../health/RelayHealthStoreCloseTest.kt | 123 ++++++++++++++++++ .../vitorpamplona/amethyst/desktop/Main.kt | 2 + .../nip01Core/relay/client/NostrClient.kt | 23 +++- 4 files changed, 178 insertions(+), 6 deletions(-) create mode 100644 commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/relays/health/RelayHealthStoreCloseTest.kt 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..68780b4d62 100644 --- a/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt +++ b/desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt @@ -1441,6 +1441,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/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index 693aa74661..092ae94eea 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt @@ -36,6 +36,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder +import com.vitorpamplona.quartz.utils.TimeUtils import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.FlowPreview @@ -164,12 +165,27 @@ class NostrClient( * error code) would stay disconnected forever in the absence of any * subscription change. The per-relay [BasicRelayClient] backoff still * gates the actual reconnect attempt, so dead relays are not hammered. + * + * Also detects system sleep/resume by tracking wall-clock overshoot of the + * scheduled tick. If the [delay] returned far later than expected the host + * was almost certainly suspended (laptop lid closed, OS sleep), and the + * OkHttp websockets we held are dead even though [isConnected] still reads + * true until the next ping fails. In that case force a hard reconnect. */ private val keepAliveJob = scope.launch { + var lastTickMs = TimeUtils.nowMillis() while (true) { delay(KEEP_ALIVE_INTERVAL_MS) - if (this@NostrClient.isActive) { + if (!this@NostrClient.isActive) continue + val now = TimeUtils.nowMillis() + val elapsed = now - lastTickMs + lastTickMs = now + if (elapsed > KEEP_ALIVE_WAKE_THRESHOLD_MS) { + // System likely resumed from sleep — force a hard reconnect. + relayPool.disconnect() + relayPool.connect() + } else { relayPool.reconnectIfNeedsTo(ignoreRetryDelays = false) } } @@ -177,6 +193,11 @@ class NostrClient( companion object { private const val KEEP_ALIVE_INTERVAL_MS = 60_000L + + // Treat any tick that overshoots the scheduled delay by more than this many + // milliseconds as a probable system-sleep resume. 5x interval (5 min) avoids + // firing on routine GC stalls or brief OS scheduler pauses. + private const val KEEP_ALIVE_WAKE_THRESHOLD_MS = 5 * KEEP_ALIVE_INTERVAL_MS } override fun reconnect( From 552540e77d50cff363412e3f1855dffae300c68e Mon Sep 17 00:00:00 2001 From: nrobi144 Date: Tue, 16 Jun 2026 09:57:22 +0300 Subject: [PATCH 2/2] fix(desktopApp): move sleep/resume detection out of Quartz MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per Vitor's review of #3221: wake-detection is platform-specific UX, not NostrClient's job. Quartz already exposes `reconnect(onlyIfChanged = false, ignoreRetryDelays = true)` which does the full disconnect + connect — the app layer just needs to call it when it detects a wake. - Revert the keep-alive heuristic in NostrClient.kt; the loop is back to the conservative `reconnectIfNeedsTo` path it had before. - Add `runSleepResumeMonitor` (desktopApp/network/SleepResumeMonitor.kt): a 60s tick that watches for wall-clock overshoot and calls the supplied `onWake` lambda. No native deps. - Wire it in `Main.kt` next to the metrics LaunchedEffect: on >5x overshoot call `relayManager.client.reconnect(onlyIfChanged = false, ignoreRetryDelays = true)`. Real OS sleep events (NSWorkspace on macOS, D-Bus PrepareForSleep on Linux, WM_POWERBROADCAST on Windows) can be layered in later as platform improvements without touching Quartz again. --- .../vitorpamplona/amethyst/desktop/Main.kt | 12 ++++ .../desktop/network/SleepResumeMonitor.kt | 66 +++++++++++++++++++ .../nip01Core/relay/client/NostrClient.kt | 23 +------ 3 files changed, 79 insertions(+), 22 deletions(-) create mode 100644 desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/network/SleepResumeMonitor.kt 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 68780b4d62..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 = 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 diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt index 092ae94eea..693aa74661 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt @@ -36,7 +36,6 @@ import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder -import com.vitorpamplona.quartz.utils.TimeUtils import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.FlowPreview @@ -165,27 +164,12 @@ class NostrClient( * error code) would stay disconnected forever in the absence of any * subscription change. The per-relay [BasicRelayClient] backoff still * gates the actual reconnect attempt, so dead relays are not hammered. - * - * Also detects system sleep/resume by tracking wall-clock overshoot of the - * scheduled tick. If the [delay] returned far later than expected the host - * was almost certainly suspended (laptop lid closed, OS sleep), and the - * OkHttp websockets we held are dead even though [isConnected] still reads - * true until the next ping fails. In that case force a hard reconnect. */ private val keepAliveJob = scope.launch { - var lastTickMs = TimeUtils.nowMillis() while (true) { delay(KEEP_ALIVE_INTERVAL_MS) - if (!this@NostrClient.isActive) continue - val now = TimeUtils.nowMillis() - val elapsed = now - lastTickMs - lastTickMs = now - if (elapsed > KEEP_ALIVE_WAKE_THRESHOLD_MS) { - // System likely resumed from sleep — force a hard reconnect. - relayPool.disconnect() - relayPool.connect() - } else { + if (this@NostrClient.isActive) { relayPool.reconnectIfNeedsTo(ignoreRetryDelays = false) } } @@ -193,11 +177,6 @@ class NostrClient( companion object { private const val KEEP_ALIVE_INTERVAL_MS = 60_000L - - // Treat any tick that overshoots the scheduled delay by more than this many - // milliseconds as a probable system-sleep resume. 5x interval (5 min) avoids - // firing on routine GC stalls or brief OS scheduler pauses. - private const val KEEP_ALIVE_WAKE_THRESHOLD_MS = 5 * KEEP_ALIVE_INTERVAL_MS } override fun reconnect(