From eaf07fd4add5195625dbf364a16faface7520fbb Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Tue, 16 Jun 2026 12:10:56 -0400 Subject: [PATCH 1/2] feat(tor): self-heal when Tor is Active but every circuit is dead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TorManager's watchdogs (selfHealSignal, connectionFailure) only arm while status is Connecting. Once Arti logs "Sufficiently bootstrapped" off cached consensus, status flips to Active and every watchdog disarms — even if no exit circuit actually works (the ExitTimeout/RESOLVEFAILED barrage seen on device). The SOCKS proxy is up, so nothing in the lifecycle looks wrong, and Tor can sit Active-but-useless indefinitely with no recovery. The relay layer is the only place with both halves of the signal: per-relay success (onConnected) and failure (onCannotConnect), plus the Tor-routing of each url. Arti emits no per-stream success log, so a Tor-internal trigger could only count failures — and couldn't tell "all dead" from "some dead", resetting Tor every time a few dead relays are dialed. Add TorCircuitHealthTracker, a RelayConnectionListener that fires only when, while Tor is Active and connectivity is up, there are >= FAIL_THRESHOLD (8) Tor-routed failures within WINDOW_MS (30s) AND zero Tor-routed successes in that window. The zero-success clause is the whole discriminator: one successful Tor open means circuits work, so it suppresses. It pokes TorManager.onTorCircuitsDead(), which mirrors the post-Active stuck-Connecting recovery — resetWithCleanState() (wipe the suspect guard/circuit state behind a healthy-looking guards.json) plus a resetEpoch bump to force a full re-init — and shares lastSelfHealAtMs / SELF_HEAL_COOLDOWN_MS with the Connecting watchdog so the two can't thrash. If circuits are still dead after the reset, the cooldown suppresses further resets and the 60s connectionFailure dialog still offers the user the bypass. Tests: TorManagerTest covers onTorCircuitsDead (fires + wipes + re-inits when Active; no-op while not-Active / bypassing; shares the cooldown). TorCircuitHealthTrackerTest covers the discriminator (threshold, single-success disarm, clearnet ignored, connectivity/Active gating, window aging, re-arm). Verified on device: no crash, tracker stays silent during normal operation (168 relays opened, 0 self-heal fires). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../com/vitorpamplona/amethyst/AppModules.kt | 13 ++ .../relayClient/TorCircuitHealthTracker.kt | 137 ++++++++++++++ .../amethyst/ui/tor/TorManager.kt | 31 +++ .../TorCircuitHealthTrackerTest.kt | 177 ++++++++++++++++++ .../amethyst/ui/tor/TorManagerTest.kt | 84 +++++++++ 5 files changed, 442 insertions(+) create mode 100644 amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTracker.kt create mode 100644 amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt 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 // ------------------------------------------------------------------ From 829afba141ba808c59465a81e27a83464adabef3 Mon Sep 17 00:00:00 2001 From: Vitor Pamplona Date: Tue, 16 Jun 2026 13:32:42 -0400 Subject: [PATCH 2/2] fix(tor): require a sustained failure streak before circuit self-heal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On-device testing caught a false positive: the original 8-failures-in-a-30s- window detector fired ~2s after Tor flipped Active. The instant Tor goes 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 relay completes its handshake — so the detector wiped the good, just-bootstrapped client. Replace the count-window with a sustained-streak model: fire only when an unbroken run of Tor-routed failures both reaches FAIL_THRESHOLD and spans at least SUSTAINED_MS (30s), with zero successes interrupting it (any success ends the streak). A gap longer than SUSTAINED_MS also restarts the streak. The span floor doubles as a post-Active warmup grace, so the warmup burst — and the identical reconnection burst on app resume — no longer trips it. Verified on device across cold start, WiFi<->Cellular handover, airplane on/off, and app pause/resume: 0 false self-heals. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../relayClient/TorCircuitHealthTracker.kt | 86 +++++++++++-------- .../TorCircuitHealthTrackerTest.kt | 81 ++++++++++++----- 2 files changed, 110 insertions(+), 57 deletions(-) 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 index 8acdb71101..edc357cf2a 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTracker.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTracker.kt @@ -38,16 +38,28 @@ import com.vitorpamplona.quartz.utils.Log * 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. + * 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. + * 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. + * threads, so the streak state is guarded by an intrinsic lock. */ class TorCircuitHealthTracker( private val client: INostrClient, @@ -57,15 +69,14 @@ class TorCircuitHealthTracker( 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 first failure in the current unbroken streak, or -1 when no streak is open. */ + private var streakStartMs: Long = -1L - /** - * 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 + /** 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 { @@ -75,11 +86,8 @@ class TorCircuitHealthTracker( compressed: Boolean, ) { if (!isTorRouted(relay.url)) return - // One Tor success means circuits work — disarm the window. - synchronized(this@TorCircuitHealthTracker) { - lastTorSuccessAtMs = nowMs() - failures.clear() - } + // One Tor success means circuits work — end the streak. + synchronized(this@TorCircuitHealthTracker) { resetStreak() } } override fun onCannotConnect( @@ -92,18 +100,17 @@ class TorCircuitHealthTracker( val now = nowMs() val fire = synchronized(this@TorCircuitHealthTracker) { - failures.addLast(now) - while (failures.isNotEmpty() && now - failures.first() > WINDOW_MS) { - failures.removeFirst() + // 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 } - 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 + 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 @@ -111,12 +118,18 @@ class TorCircuitHealthTracker( } if (fire) { - Log.w(TAG) { "Tor Active but $FAIL_THRESHOLD+ Tor relays failed with no success in ${WINDOW_MS}ms — requesting self-heal" } + 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) } @@ -128,10 +141,13 @@ class TorCircuitHealthTracker( companion object { const val TAG = "TorCircuitHealthTracker" - /** Tor-routed failures within [WINDOW_MS] (and zero successes) needed to declare circuits dead. */ + /** Unbroken Tor-routed failures needed (alongside the [SUSTAINED_MS] floor) 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 + /** + * 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/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt index 4a9053f150..e672ec5a51 100644 --- a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/TorCircuitHealthTrackerTest.kt @@ -37,25 +37,56 @@ 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 after threshold Tor failures with no success in window`() { + fun `fires on a sustained threshold-meeting failure streak`() { val h = Harness() - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + h.sustainedFailures() assertEquals(1, h.deadCount) } @Test - fun `does not fire below threshold`() { + fun `does not fire on a short burst even past threshold`() { val h = Harness() - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD - 1) { h.failTor() } + // 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 disarms the window`() { + fun `a single Tor success ends the streak`() { val h = Harness() - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD - 1) { h.failTor() } - h.succeedTor() // resets the failure window AND the no-success clock + 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) } @@ -63,49 +94,55 @@ class TorCircuitHealthTrackerTest { @Test fun `clearnet relay outcomes are ignored`() { val h = Harness() - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD * 2) { h.failClear() } + 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) - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + 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) - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + h.sustainedFailures() assertEquals(0, h.deadCount) } @Test - fun `failures older than the window do not accumulate`() { + fun `a long gap restarts the streak`() { 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. + // 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 window to fire again`() { + fun `re-arms after firing — needs a fresh sustained streak to fire again`() { val h = Harness() - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + h.sustainedFailures() 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. + // 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.WINDOW_MS + 1 - repeat(TorCircuitHealthTracker.FAIL_THRESHOLD) { h.failTor() } + h.now += TorCircuitHealthTracker.SUSTAINED_MS + 1 // gap resets, then a fresh sustained streak + h.sustainedFailures() assertEquals(2, h.deadCount) }