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..edc357cf2a --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTracker.kt @@ -0,0 +1,153 @@ +/* + * 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 a *sustained* failure streak: fire only when, while Tor + * is Active and connectivity is up, there is an unbroken run of Tor-routed failures that (a) holds + * at least [FAIL_THRESHOLD] failures and (b) has lasted at least [SUSTAINED_MS], with **zero** + * Tor-routed successes interrupting it. Any Tor success ends the streak — one good open means + * circuits work, so suppress. + * + * The [SUSTAINED_MS] floor is essential, not cosmetic: the instant Tor flips Active the pool dials + * the whole Tor-routed relay set at once, and on freshly built circuits a burst of ≥8 can fail + * *before* the first Tor relay completes its handshake. Counting failures in a short window would + * fire ~2s into a perfectly healthy warmup and wipe the good client (observed on device). Requiring + * the streak to span [SUSTAINED_MS] gives a natural post-Active warmup grace — a real success + * during warmup clears the streak; only genuinely dead circuits keep failing for the full span. + * A gap longer than [SUSTAINED_MS] between failures also restarts the streak, so sparse unrelated + * failures never accumulate. + * + * 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 streak state 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 the first failure in the current unbroken streak, or -1 when no streak is open. */ + private var streakStartMs: Long = -1L + + /** Epoch-millis of the most recent failure in the current streak. */ + private var lastFailureMs: Long = -1L + + /** Failures in the current streak. */ + private var streakCount: Int = 0 + + 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 — end the streak. + synchronized(this@TorCircuitHealthTracker) { resetStreak() } + } + + override fun onCannotConnect( + relay: IRelayClient, + errorMessage: String, + ) { + if (!isTorRouted(relay.url)) return + if (!isTorActive() || !isConnectivityActive()) return + + val now = nowMs() + val fire = + synchronized(this@TorCircuitHealthTracker) { + // Start a fresh streak on the first failure, or whenever the gap since the + // last failure is too long for them to count as the same sustained outage. + if (streakStartMs < 0 || now - lastFailureMs > SUSTAINED_MS) { + streakStartMs = now + streakCount = 0 + } + streakCount++ + lastFailureMs = now + + if (streakCount >= FAIL_THRESHOLD && now - streakStartMs >= SUSTAINED_MS) { + resetStreak() // re-arm; a fresh streak must build before firing again + true + } else { + false + } + } + + if (fire) { + Log.w(TAG) { "Tor Active but $FAIL_THRESHOLD+ Tor relays failed for ${SUSTAINED_MS}ms+ with no success — requesting self-heal" } + onCircuitsDead() + } + } + } + + private fun resetStreak() { + streakStartMs = -1L + lastFailureMs = -1L + streakCount = 0 + } + + fun register() { + client.addConnectionListener(listener) + } + + fun unregister() { + client.removeConnectionListener(listener) + } + + companion object { + const val TAG = "TorCircuitHealthTracker" + + /** Unbroken Tor-routed failures needed (alongside the [SUSTAINED_MS] floor) to declare circuits dead. */ + const val FAIL_THRESHOLD = 8 + + /** + * The failure streak must span at least this long before it counts as a dead transport. + * Doubles as the post-Active warmup grace and the max gap between failures of one streak. + */ + const val SUSTAINED_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..e672ec5a51 --- /dev/null +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt @@ -0,0 +1,214 @@ +/* + * 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/") + + /** + * Drives a sustained, threshold-meeting failure streak: spreads count-1 failures in small + * continuous steps, then jumps the clock so the final failure pushes the span just past the + * floor — guaranteeing both the count and the SUSTAINED_MS span are met. + */ + private fun Harness.sustainedFailures(count: Int = TorCircuitHealthTracker.FAIL_THRESHOLD) { + val start = now + repeat(count - 1) { + failTor() + now += 1_000 // small continuous gaps, well under SUSTAINED_MS + } + now = start + TorCircuitHealthTracker.SUSTAINED_MS + 1 + failTor() + } + + @Test + fun `fires on a sustained threshold-meeting failure streak`() { + val h = Harness() + h.sustainedFailures() + assertEquals(1, h.deadCount) + } + + @Test + fun `does not fire on a short burst even past threshold`() { + val h = Harness() + // 8+ failures but all within ~2s — the post-Active warmup burst that wrongly wiped the + // good client on device. Must NOT fire: the streak hasn't spanned SUSTAINED_MS. + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD * 2) { + h.failTor() + h.now += 200 + } + assertEquals("a short burst is warmup, not a dead transport", 0, h.deadCount) + } + + @Test + fun `does not fire below threshold even when sustained`() { + val h = Harness() + // Spread a few failures across well over the span, but never reach the count. + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD - 1) { + h.failTor() + h.now += TorCircuitHealthTracker.SUSTAINED_MS + } + assertEquals(0, h.deadCount) + } + + @Test + fun `a single Tor success ends the streak`() { + val h = Harness() + h.sustainedFailures(count = TorCircuitHealthTracker.FAIL_THRESHOLD - 1) + h.succeedTor() // warmup completed — streak ends + 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 * 4) { + h.failClear() + h.now += TorCircuitHealthTracker.SUSTAINED_MS + } + assertEquals(0, h.deadCount) + } + + @Test + fun `does not fire when connectivity is down`() { + val h = Harness(connectivityActive = false) + h.sustainedFailures() + 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) + h.sustainedFailures() + assertEquals(0, h.deadCount) + } + + @Test + fun `a long gap restarts the streak`() { + val h = Harness() + // Almost reach the count, sustained... + repeat(TorCircuitHealthTracker.FAIL_THRESHOLD - 1) { + h.failTor() + h.now += 3_000 + } + // ...then a gap longer than the span: the next failure starts a brand-new streak, so a + // single fresh failure can't reach threshold. + h.now += TorCircuitHealthTracker.SUSTAINED_MS + 1 + h.failTor() + assertEquals(0, h.deadCount) + } + + @Test + fun `re-arms after firing — needs a fresh sustained streak to fire again`() { + val h = Harness() + h.sustainedFailures() + assertEquals(1, h.deadCount) + + // Immediately after firing the streak is reset, so the next failure can't re-fire until a + // new streak builds count AND spans the floor again. + h.failTor() + assertEquals(1, h.deadCount) + + h.now += TorCircuitHealthTracker.SUSTAINED_MS + 1 // gap resets, then a fresh sustained streak + h.sustainedFailures() + 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 // ------------------------------------------------------------------