diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt index 80466dcc26..0a57cee574 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/AppModules.kt @@ -68,6 +68,7 @@ import com.vitorpamplona.amethyst.service.playback.pip.BackgroundMedia import com.vitorpamplona.amethyst.service.playback.service.PlaybackServiceClient import com.vitorpamplona.amethyst.service.relayClient.CacheClientConnector import com.vitorpamplona.amethyst.service.relayClient.RelayProxyClientConnector +import com.vitorpamplona.amethyst.service.relayClient.TorCircuitHealthTracker import com.vitorpamplona.amethyst.service.relayClient.authCommand.model.AuthCoordinator import com.vitorpamplona.amethyst.service.relayClient.notifyCommand.model.NotifyCoordinator import com.vitorpamplona.amethyst.service.relayClient.reqCommand.RelaySubscriptionsCoordinator @@ -481,6 +482,18 @@ class AppModules( // Provides a relay pool val client: INostrClient = NostrClient(websocketBuilder, applicationIOScope) + // Self-heals the "Tor Active but every circuit dead" state the lifecycle watchdogs can't + // see (they only arm while Connecting). Watches Tor-routed relay outcomes and, when enough + // fail with zero successes in the window, pokes TorManager to drop + re-init Arti. + val torCircuitHealthTracker = + TorCircuitHealthTracker( + client = client, + isTorRouted = { torEvaluatorFlow.shouldUseTorForRelay(it) }, + isTorActive = { torManager.isSocksReady() }, + isConnectivityActive = { connManager.status.value is ConnectivityStatus.Active }, + onCircuitsDead = { torManager.onTorCircuitsDead() }, + ).also { it.register() } + // Watches for changes on Tor and Relay List Settings val relayProxyClientConnector = RelayProxyClientConnector( diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTracker.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTracker.kt new file mode 100644 index 0000000000..8acdb71101 --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTracker.kt @@ -0,0 +1,137 @@ +/* + * 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.service.relayClient + +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import com.vitorpamplona.quartz.utils.Log + +/** + * Detects the "Tor is Active but every circuit is dead" state and asks [onCircuitsDead] to + * self-heal it. + * + * Tor can reach [com.vitorpamplona.amethyst.ui.tor.TorServiceStatus.Active] — SOCKS proxy up, + * bootstrap "succeeded" off cached consensus — while no exit circuit actually works (the + * `ExitTimeout` / `RESOLVEFAILED` barrage). The lifecycle watchdogs in + * [com.vitorpamplona.amethyst.ui.tor.TorManager] can't see this: they only arm while status is + * `Connecting`. The relay layer is the only place that has both halves of the signal — per-relay + * success ([RelayConnectionListener.onConnected]) and failure + * ([RelayConnectionListener.onCannotConnect]) — plus knowledge of which relays are Tor-routed. + * + * The discriminator (the whole point) is: fire only when, while Tor is Active and connectivity is + * up, there have been at least [FAIL_THRESHOLD] Tor-routed failures within [WINDOW_MS] **and zero + * Tor-routed successes in that window**. One successful Tor open means circuits work — suppress. + * Gating on connectivity avoids resetting Tor during a general network outage (where clearnet + * relays fail too). Arti emits no per-stream success log, so this success signal can only come + * from the relay layer — a failure-only signal can't tell "all dead" from "some dead", and would + * reset Tor every time a few dead relays are dialed. + * + * Register with [register] and tear down with [unregister]. All callbacks arrive on relay/OkHttp + * threads, so the sliding window is guarded by an intrinsic lock. + */ +class TorCircuitHealthTracker( + private val client: INostrClient, + private val isTorRouted: (NormalizedRelayUrl) -> Boolean, + private val isTorActive: () -> Boolean, + private val isConnectivityActive: () -> Boolean, + private val onCircuitsDead: () -> Unit, + private val nowMs: () -> Long = System::currentTimeMillis, +) { + /** Epoch-millis of Tor-routed failures still inside the rolling [WINDOW_MS]. */ + private val failures = ArrayDeque() + + /** + * Epoch-millis of the last Tor-routed success, or null if we've never observed one. Null + * (not "now") at construction on purpose: before any success we must NOT assume circuits are + * healthy, otherwise an all-dead-from-cold-start state would be suppressed for a full window. + */ + @Volatile private var lastTorSuccessAtMs: Long? = null + + private val listener = + object : RelayConnectionListener { + override fun onConnected( + relay: IRelayClient, + pingMillis: Int, + compressed: Boolean, + ) { + if (!isTorRouted(relay.url)) return + // One Tor success means circuits work — disarm the window. + synchronized(this@TorCircuitHealthTracker) { + lastTorSuccessAtMs = nowMs() + failures.clear() + } + } + + override fun onCannotConnect( + relay: IRelayClient, + errorMessage: String, + ) { + if (!isTorRouted(relay.url)) return + if (!isTorActive() || !isConnectivityActive()) return + + val now = nowMs() + val fire = + synchronized(this@TorCircuitHealthTracker) { + failures.addLast(now) + while (failures.isNotEmpty() && now - failures.first() > WINDOW_MS) { + failures.removeFirst() + } + val lastSuccess = lastTorSuccessAtMs + val noSuccessInWindow = lastSuccess == null || now - lastSuccess >= WINDOW_MS + if (failures.size >= FAIL_THRESHOLD && noSuccessInWindow) { + // Re-arm: drop the window and push the success marker forward so a + // burst of late failures from the same dead state doesn't re-fire + // before the reset has had a chance to take effect. + failures.clear() + lastTorSuccessAtMs = now + true + } else { + false + } + } + + if (fire) { + Log.w(TAG) { "Tor Active but $FAIL_THRESHOLD+ Tor relays failed with no success in ${WINDOW_MS}ms — requesting self-heal" } + onCircuitsDead() + } + } + } + + fun register() { + client.addConnectionListener(listener) + } + + fun unregister() { + client.removeConnectionListener(listener) + } + + companion object { + const val TAG = "TorCircuitHealthTracker" + + /** Tor-routed failures within [WINDOW_MS] (and zero successes) needed to declare circuits dead. */ + const val FAIL_THRESHOLD = 8 + + /** Rolling window for counting failures and the "no success since" check. */ + const val WINDOW_MS = 30_000L + } +} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorManager.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorManager.kt index b0d750f83a..46a491b3f3 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorManager.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorManager.kt @@ -287,6 +287,37 @@ class TorManager( } } + /** + * Tor is [TorServiceStatus.Active] (SOCKS proxy up, bootstrap "succeeded" off cached + * consensus) yet every Tor-routed relay is failing — no successful Tor open while exit + * failures pile up. The circuits behind the proxy are dead, but the lifecycle can't see + * it: the stuck-Connecting watchdog and the [connectionFailure] dialog only arm while + * status is [TorServiceStatus.Connecting], not Active. The relay layer (which knows both + * the per-relay success/failure outcome and the Tor-routing of each url) detects the + * all-failing condition and pokes us here — analogous to [onNetworkChange]. + * + * Recovery mirrors the post-Active stuck-Connecting path: drop the client and wipe + * `arti/state/` (we *did* bootstrap, so a fully-dead exit set behind a healthy-looking + * guards.json points at a bad persisted guard/circuit sample worth rebuilding), then bump + * [resetEpoch] so the status combine re-enters the INTERNAL branch and runs a full + * re-init. Shares [lastSelfHealAtMs]/[SELF_HEAL_COOLDOWN_MS] with the Connecting watchdog + * so the two can't thrash — at most one self-heal per cooldown window. If circuits are + * still dead after the reset, the cooldown suppresses further resets and the 60s + * [connectionFailure] dialog still offers the user the bypass. + */ + fun onTorCircuitsDead() { + if (sessionBypass.value) return + if (status.value !is TorServiceStatus.Active) return + val now = nowMs() + if (now - lastSelfHealAtMs < SELF_HEAL_COOLDOWN_MS) return + lastSelfHealAtMs = now + Log.w("TorManager") { "Tor Active but all circuits failing — self-healing (drop client + wipe state)" } + scope.launch(ioDispatcher) { + service.resetWithCleanState() + resetEpoch.update { it + 1 } + } + } + fun isSocksReady() = status.value is TorServiceStatus.Active fun socksPort(): Int = (status.value as? TorServiceStatus.Active)?.port ?: 17392 diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt new file mode 100644 index 0000000000..4a9053f150 --- /dev/null +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt @@ -0,0 +1,177 @@ +/* + * 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.service.relayClient + +import com.vitorpamplona.quartz.nip01Core.relay.client.EmptyNostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient +import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener +import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient +import com.vitorpamplona.quartz.nip01Core.relay.commands.toRelay.Command +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * Unit tests for [TorCircuitHealthTracker]'s discriminator. Drives the captured + * [RelayConnectionListener] directly with a virtual clock — no Arti, no real relay client. + */ +class TorCircuitHealthTrackerTest { + private val torRelay = NormalizedRelayUrl("wss://torrelay.example/") + private val clearRelay = NormalizedRelayUrl("wss://clearrelay.example/") + + @Test + fun `fires after threshold Tor failures with no success in window`() { + val h = Harness() + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + assertEquals(1, h.deadCount) + } + + @Test + fun `does not fire below threshold`() { + val h = Harness() + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD - 1) { h.failTor() } + assertEquals(0, h.deadCount) + } + + @Test + fun `a single Tor success disarms the window`() { + val h = Harness() + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD - 1) { h.failTor() } + h.succeedTor() // resets the failure window AND the no-success clock + h.failTor() + assertEquals("one success means circuits work — suppress", 0, h.deadCount) + } + + @Test + fun `clearnet relay outcomes are ignored`() { + val h = Harness() + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD * 2) { h.failClear() } + assertEquals(0, h.deadCount) + } + + @Test + fun `does not fire when connectivity is down`() { + val h = Harness(connectivityActive = false) + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + assertEquals("general outage must not reset Tor", 0, h.deadCount) + } + + @Test + fun `does not fire when Tor is not Active`() { + val h = Harness(torActive = false) + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + assertEquals(0, h.deadCount) + } + + @Test + fun `failures older than the window do not accumulate`() { + val h = Harness() + // Fill almost to threshold, then let the window slide fully past those failures. + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD - 1) { h.failTor() } + h.now += TorCircuitHealthTracker.WINDOW_MS + 1 + // The no-success clause holds (we never succeeded), but the stale failures must have aged + // out of the window, so a single fresh failure can't reach threshold. + h.failTor() + assertEquals(0, h.deadCount) + } + + @Test + fun `re-arms after firing — needs a fresh window to fire again`() { + val h = Harness() + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + assertEquals(1, h.deadCount) + + // Immediately after firing the window is cleared and the no-success clock reset, so the + // next failure can't re-fire until both the window refills and WINDOW_MS elapses. + h.failTor() + assertEquals(1, h.deadCount) + + h.now += TorCircuitHealthTracker.WINDOW_MS + 1 + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + assertEquals(2, h.deadCount) + } + + // ------------------------------------------------------------------ + // harness + // ------------------------------------------------------------------ + + private inner class Harness( + torActive: Boolean = true, + connectivityActive: Boolean = true, + ) { + var now: Long = 1_000_000_000_000L + var deadCount: Int = 0 + private set + + private val client = CapturingClient() + + init { + TorCircuitHealthTracker( + client = client, + isTorRouted = { it == torRelay }, + isTorActive = { torActive }, + isConnectivityActive = { connectivityActive }, + onCircuitsDead = { deadCount++ }, + nowMs = { now }, + ).register() + } + + fun failTor() = client.listener!!.onCannotConnect(relayClient(torRelay), "SOCKS: Connection refused") + + fun failClear() = client.listener!!.onCannotConnect(relayClient(clearRelay), "boom") + + fun succeedTor() = client.listener!!.onConnected(relayClient(torRelay), 400, false) + } + + private fun relayClient(relayUrl: NormalizedRelayUrl): IRelayClient = + object : IRelayClient { + override val url = relayUrl + + override fun connect() = error("unused") + + override fun needsToReconnect() = error("unused") + + override fun connectAndSyncFiltersIfDisconnected(ignoreRetryDelays: Boolean) = error("unused") + + override fun isConnected() = error("unused") + + override fun sendOrConnectAndSync(cmd: Command) = error("unused") + + override fun sendIfConnected(cmd: Command) = error("unused") + + override fun disconnect() = error("unused") + } + + /** Minimal [INostrClient] (delegating to [EmptyNostrClient]) that just captures the listener. */ + private class CapturingClient( + private val delegate: INostrClient = EmptyNostrClient(), + ) : INostrClient by delegate { + var listener: RelayConnectionListener? = null + + override fun addConnectionListener(listener: RelayConnectionListener) { + this.listener = listener + } + + override fun removeConnectionListener(listener: RelayConnectionListener) { + if (this.listener === listener) this.listener = null + } + } +} diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/TorManagerTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/TorManagerTest.kt index 8afa86072e..7da36be10b 100644 --- a/amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/TorManagerTest.kt +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/TorManagerTest.kt @@ -166,6 +166,90 @@ class TorManagerTest { assertEquals(0, backend.resetWithCleanStateCount) } + // ------------------------------------------------------------------ + // onTorCircuitsDead — Active-but-failing self-heal + // ------------------------------------------------------------------ + + @Test + fun `onTorCircuitsDead wipes state and re-inits when Active`() = + runTest(UnconfinedTestDispatcher()) { + val backend = FakeTorBackend() + val manager = buildManager(backend = backend, clock = { 1_000_000_000_000L }) + advanceUntilIdle() + backend.setActive(17392) + advanceUntilIdle() + + manager.onTorCircuitsDead() + advanceUntilIdle() + + assertEquals("Active-but-failing recovery wipes state", 1, backend.resetWithCleanStateCount) + // resetEpoch bump re-enters INTERNAL → start() runs again. + assertTrue("re-init should call start() again", backend.startCount >= 2) + } + + @Test + fun `onTorCircuitsDead is a no-op while not Active`() = + runTest(UnconfinedTestDispatcher()) { + val backend = FakeTorBackend() + val manager = buildManager(backend = backend, clock = { 1_000_000_000_000L }) + advanceUntilIdle() + // Still Connecting (never reached Active). + assertEquals(TorServiceStatus.Connecting, manager.status.value) + + manager.onTorCircuitsDead() + advanceUntilIdle() + + assertEquals(0, backend.resetWithCleanStateCount) + } + + @Test + fun `onTorCircuitsDead is a no-op while bypassing`() = + runTest(UnconfinedTestDispatcher()) { + val backend = FakeTorBackend() + val manager = buildManager(backend = backend, clock = { 1_000_000_000_000L }) + advanceUntilIdle() + backend.setActive(17392) + advanceUntilIdle() + manager.sessionBypass.value = true + advanceUntilIdle() + + manager.onTorCircuitsDead() + advanceUntilIdle() + + assertEquals(0, backend.resetWithCleanStateCount) + } + + @Test + fun `onTorCircuitsDead shares the self-heal cooldown`() = + runTest(UnconfinedTestDispatcher()) { + val backend = FakeTorBackend() + var clockNow = 1_000_000_000_000L + val manager = buildManager(backend = backend, clock = { clockNow }) + advanceUntilIdle() + backend.setActive(17392) + advanceUntilIdle() + + manager.onTorCircuitsDead() + advanceUntilIdle() + assertEquals(1, backend.resetWithCleanStateCount) + + // Backend is Active again (re-init bootstrapped). A second call inside the cooldown + // window must be suppressed. + backend.setActive(17392) + advanceUntilIdle() + manager.onTorCircuitsDead() + advanceUntilIdle() + assertEquals("cooldown should suppress the second self-heal", 1, backend.resetWithCleanStateCount) + + // Past the cooldown it can fire again. + clockNow += TorManager.SELF_HEAL_COOLDOWN_MS + 1_000L + backend.setActive(17392) + advanceUntilIdle() + manager.onTorCircuitsDead() + advanceUntilIdle() + assertEquals("after cooldown elapses, self-heal fires again", 2, backend.resetWithCleanStateCount) + } + // ------------------------------------------------------------------ // stuck-Connecting watchdog // ------------------------------------------------------------------