Merge pull request #3721 from vitorpamplona/fix/tor-guard-sample-self-heal

fix(tor): self-heal a guard sample that is rotten at runtime, not just on disk

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/ArtiGuardState.kt

@@ -48,12 +48,34 @@ object ArtiGuardState {
     fun parse(json: String): JsonNode = mapper.readTree(json)
 
     /**
-     * True when at least one non-empty guard selection has *zero* usable guards —
-     * every guard permanently `disabled` or dropped from the consensus
-     * (`unlisted_since`). This is the AllGuardsDown wedge: Arti can neither build
-     * circuits nor replenish the sample (it's full of unusable entries), so it
-     * stays broken across restarts until the on-disk state is wiped. A single
-     * usable guard is enough to recover, so this only trips on a total wipeout.
+     * Smallest sample worth judging by ratio. Below this a low usable count is just a young
+     * sample that Arti is still filling, and wiping it would loop the bootstrap.
+     */
+    const val MIN_SAMPLE_TO_JUDGE_RATIO = 10
+
+    /** A sample is wedged when fewer than 1 in [USABLE_RATIO_DIVISOR] of its guards are usable. */
+    const val USABLE_RATIO_DIVISOR = 10
+
+    /**
+     * True when a non-empty guard selection has no usable guards left, or so few that Arti cannot
+     * realistically recover from them — every guard permanently `disabled` or dropped from the
+     * consensus (`unlisted_since`).
+     *
+     * This is the AllGuardsDown wedge: Arti can neither build circuits nor replenish the sample, so
+     * it stays broken across restarts until the on-disk state is wiped.
+     *
+     * ### Why this is not `usable == 0`
+     *
+     * It used to be, on the assumption that "a single usable guard is enough to recover". Observed in
+     * the field: a sample of 60 with **59 disabled and 1 usable**, while Arti rejected all sixty at
+     * runtime (`AllGuardsDown { n_accepted: 0, n_rejected: 60 }`) across repeated app restarts. The
+     * lone survivor was just as unreachable as the rest — useless for recovery, but enough to veto
+     * it, because `usable == 0` never became true. ~87% of relay connections failed indefinitely.
+     *
+     * So the test is proportional: a sample of at least [MIN_SAMPLE_TO_JUDGE_RATIO] whose usable
+     * guards are under 1/[USABLE_RATIO_DIVISOR] of the total is wedged. A small, young sample is
+     * still only judged by the strict `usable == 0` rule, so a legitimate first bootstrap that has
+     * sampled one or two guards is never wiped out from under itself.
      */
     fun hasNoUsableGuards(root: JsonNode): Boolean {
         var wedged = false
@@ -61,7 +83,11 @@ object ArtiGuardState {
             val guards = selection.get("guards") ?: return@forEach
             if (guards.isArray && guards.size() > 0) {
                 val usable = guards.count { !it.isDisabled() && !it.isUnlisted() }
-                if (usable == 0) wedged = true
+                if (usable == 0) {
+                    wedged = true
+                } else if (guards.size() >= MIN_SAMPLE_TO_JUDGE_RATIO && usable * USABLE_RATIO_DIVISOR < guards.size()) {
+                    wedged = true
+                }
             }
         }
         return wedged

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorBackend.kt

@@ -20,6 +20,7 @@
  */
 package com.vitorpamplona.amethyst.ui.tor
 
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.StateFlow
 
 /**
@@ -46,4 +47,16 @@ interface TorBackend {
      * this is a `suspend` call.
      */
     suspend fun hasBootstrappedBefore(): Boolean
+
+    /**
+     * Emits when Arti has reported `AllGuardsDown` persistently enough that the guard sample is
+     * considered rotten at runtime, regardless of what `guards.json` claims.
+     *
+     * The on-disk heuristic ([ArtiGuardState.hasNoUsableGuards]) only sees guards Arti has
+     * permanently retired. A guard that is merely unreachable stays "usable" on disk forever, so a
+     * sample can be entirely dead in practice while still looking healthy to that check — which is
+     * exactly the state that survived repeated restarts in the field. This is the runtime half:
+     * Arti itself says every guard was rejected, so believe it.
+     */
+    val guardsDownSignal: Flow<Unit>
 }

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorManager.kt

@@ -250,6 +250,24 @@ class TorManager(
                 if (it is TorServiceStatus.Active) hasEverBootstrapped = true
             }.launchIn(scope)
 
+        // Rotten guard sample while Tor is otherwise UP. The watchdog above only fires on a status
+        // stuck at Connecting, and the on-disk check only sees guards Arti has permanently retired —
+        // so a sample whose guards are all merely *unreachable* falls between them: status reaches
+        // Active, `guards.json` still lists usable entries, and every circuit fails anyway. Observed
+        // in the field surviving repeated restarts with ~87% of relay connections failing. Arti's own
+        // AllGuardsDown log is the only reliable signal, so route it through the same rate-limited
+        // wipe. Always a clean-state reset: the whole point is that the persisted sample is the
+        // problem.
+        service.guardsDownSignal
+            .onEach {
+                val now = nowMs()
+                if (now - lastSelfHealAtMs < SELF_HEAL_COOLDOWN_MS) return@onEach
+                lastSelfHealAtMs = now
+                Log.w("TorManager") { "Arti guard sample rotten at runtime — self-healing (drop client + wipe state)" }
+                service.resetWithCleanState()
+                resetEpoch.update { it + 1 }
+            }.launchIn(scope)
+
         selfHealSignal
             .onEach {
                 val now = nowMs()

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorService.kt

@@ -25,8 +25,12 @@ import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
@@ -34,6 +38,15 @@ import kotlinx.coroutines.withContext
 import java.io.File
 import java.util.concurrent.atomic.AtomicBoolean
 
+/** Arti's log marker for "no usable guard could be found". */
+private const val ALL_GUARDS_DOWN_MARKER = "AllGuardsDown"
+
+/** How many [ALL_GUARDS_DOWN_MARKER] lines inside [GUARDS_DOWN_WINDOW_MS] mean the sample is rotten. */
+private const val GUARDS_DOWN_THRESHOLD = 40
+
+/** Window for [GUARDS_DOWN_THRESHOLD]. Wide enough that ordinary transient churn never trips it. */
+private const val GUARDS_DOWN_WINDOW_MS = 60_000L
+
 private const val DEFAULT_SOCKS_PORT = 17392
 private const val MAX_PORT_RETRIES = 10
 
@@ -85,6 +98,47 @@ class TorService(
     private val _status = MutableStateFlow<TorServiceStatus>(TorServiceStatus.Off)
     override val status: StateFlow<TorServiceStatus> = _status.asStateFlow()
 
+    /**
+     * Runtime detector for a rotten guard sample. Arti logs [ALL_GUARDS_DOWN_MARKER] every time it
+     * fails to find a usable guard; [GUARDS_DOWN_THRESHOLD] of those inside [GUARDS_DOWN_WINDOW_MS]
+     * means every guard in the sample is unreachable *right now* — which the on-disk check cannot
+     * see, because an unreachable guard is not a `disabled` one (see
+     * [ArtiGuardState.hasNoUsableGuards]).
+     *
+     * Counted here because the log callback is the only place Arti surfaces it; acted on by
+     * [TorManager], which owns the reset cadence and its rate limiting. Both counters are only
+     * touched from Arti's log thread.
+     */
+    private var guardsDownCount = 0
+
+    private var guardsDownWindowStartMs = 0L
+
+    private val _guardsDownSignal =
+        MutableSharedFlow<Unit>(extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+
+    override val guardsDownSignal: Flow<Unit> = _guardsDownSignal.asSharedFlow()
+
+    /**
+     * Feeds one Arti log line to the [guardsDownSignal] detector. Cheap by design: this runs for
+     * every log line, and a wedged Arti emits tens of thousands of them.
+     */
+    private fun trackGuardFailures(text: String) {
+        if (!text.contains(ALL_GUARDS_DOWN_MARKER)) return
+
+        val now = System.currentTimeMillis()
+        if (now - guardsDownWindowStartMs > GUARDS_DOWN_WINDOW_MS) {
+            guardsDownWindowStartMs = now
+            guardsDownCount = 0
+        }
+        guardsDownCount++
+        if (guardsDownCount >= GUARDS_DOWN_THRESHOLD) {
+            guardsDownCount = 0
+            guardsDownWindowStartMs = now
+            Log.w("TorService") { "Arti reported $ALL_GUARDS_DOWN_MARKER $GUARDS_DOWN_THRESHOLD times in ${GUARDS_DOWN_WINDOW_MS}ms — guard sample is rotten at runtime" }
+            _guardsDownSignal.tryEmit(Unit)
+        }
+    }
+
     private fun artiDataDir() = File(context.filesDir, "arti")
 
     /** Diagnostic: total bytes of the consensus/descriptor cache, to correlate with bootstrap time. */
@@ -181,6 +235,7 @@ class TorService(
                     // (gated on proxyRunning) intermittently dropped the transition. start()
                     // now sets Active deterministically once the proxy is bound.
                     ArtiNative.setLogCallback { text ->
+                        trackGuardFailures(text)
                         Log.d("TorService") {
                             val newLine = text.indexOf('\n')
                             if (newLine > 1) {

amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/ArtiGuardStateTest.kt

@@ -100,6 +100,75 @@ class ArtiGuardStateTest {
         assertTrue(ArtiGuardState.hasConfirmedGuard(root))
     }
 
+    /**
+     * The field case this ratio rule exists for: a 60-guard sample with 59 permanently disabled and
+     * exactly ONE still nominally usable, while Arti rejected all sixty at runtime
+     * (`AllGuardsDown { n_accepted: 0, n_rejected: 60 }`) across repeated app restarts. Under the old
+     * `usable == 0` rule the lone survivor vetoed the wipe forever, so Tor stayed ~87% broken and
+     * Concord never folded.
+     */
+    @Test
+    fun `one usable guard out of sixty is wedged, not healthy`() {
+        val guards =
+            buildString {
+                append("""{ "default": { "guards": [""")
+                repeat(59) {
+                    append("""{ "confirmed_at": "2026-07-01T00:00:00Z", "disabled": { "type": "TooManyIndeterminateFailures" }, "unlisted_since": null },""")
+                }
+                append("""{ "confirmed_at": null, "disabled": null, "unlisted_since": null }""")
+                append("] } }")
+            }
+        val root = ArtiGuardState.parse(guards)
+
+        assertTrue(
+            "1 usable of 60 is unrecoverable in practice → must wipe",
+            ArtiGuardState.hasNoUsableGuards(root),
+        )
+    }
+
+    /**
+     * The opposite guard-rail: a young sample must never be wiped just for being small, or a genuine
+     * first bootstrap would loop. Only the strict `usable == 0` rule applies below
+     * [ArtiGuardState.MIN_SAMPLE_TO_JUDGE_RATIO].
+     */
+    @Test
+    fun `small fresh sample is never wedged by the ratio rule`() {
+        val root =
+            ArtiGuardState.parse(
+                """
+                { "default": { "guards": [
+                    { "confirmed_at": null, "disabled": null, "unlisted_since": null },
+                    { "confirmed_at": null, "disabled": { "type": "TooManyIndeterminateFailures" } },
+                    { "confirmed_at": null, "disabled": { "type": "TooManyIndeterminateFailures" } }
+                ] } }
+                """.trimIndent(),
+            )
+
+        assertFalse(
+            "1 usable of 3 is a young sample Arti is still filling → leave it alone",
+            ArtiGuardState.hasNoUsableGuards(root),
+        )
+    }
+
+    /** A healthy majority must not trip the ratio rule. */
+    @Test
+    fun `mostly usable large sample is not wedged`() {
+        val guards =
+            buildString {
+                append("""{ "default": { "guards": [""")
+                repeat(20) {
+                    append("""{ "confirmed_at": null, "disabled": null, "unlisted_since": null },""")
+                }
+                append("""{ "confirmed_at": null, "disabled": { "type": "TooManyIndeterminateFailures" } }""")
+                append("] } }")
+            }
+
+        assertFalse(
+            "20 usable of 21 is healthy",
+            ArtiGuardState.hasNoUsableGuards(ArtiGuardState.parse(guards)),
+        )
+    }
+
     @Test
     fun `empty selection is neither wedged nor confirmed`() {
         val root = ArtiGuardState.parse("""{ "default": { "guards": [] }, "restricted": { "guards": [] } }""")

amethyst/src/test/java/com/vitorpamplona/amethyst/ui/tor/TorManagerTest.kt

@@ -22,6 +22,8 @@ package com.vitorpamplona.amethyst.ui.tor
 
 import com.vitorpamplona.amethyst.commons.tor.TorType
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -506,6 +508,11 @@ private class FakeTorBackend : TorBackend {
     /** Simulates a persisted confirmed guard on disk (prior successful bootstrap). */
     var bootstrappedBefore = false
 
+    /** Lets a test fire Arti's "every guard was rejected" signal (see [TorBackend.guardsDownSignal]). */
+    val guardsDown = MutableSharedFlow<Unit>(extraBufferCapacity = 1)
+
+    override val guardsDownSignal: Flow<Unit> = guardsDown
+
     override suspend fun hasBootstrappedBefore(): Boolean = bootstrappedBefore
 
     override suspend fun start() {