Bound egress canary probes with a watchdog and cancel them on invalidate

The Tor egress self-check probe had no outer bound: the proxied session
sets waitsForConnectivity = true, which can defer the per-request timer
indefinitely, leaving a canary request bounded only by URLSession's
default 7-day resource timeout. Because verify() joins concurrent
callers on the in-flight task and invalidate() neither cancelled nor
cleared it, one hung probe wedged every subsequent verify() caller --
even across a Tor restart -- parking awaitingTorForConnections in
NostrRelayManager until process restart.

Fixes (liveness only; the fail-closed policy is unchanged):
- Race every probe against an independent async watchdog
  (probeTimeout, default 20s = the live probe's request timeout, via
  TorEgressVerifier.defaultProbeTimeout). On timeout the probe task is
  cancelled (cooperatively cancelling the underlying URLSessionTask),
  the verdict is the fail-closed .unreachable, and the in-flight slot
  is cleared so the next verify() starts fresh.
- invalidate() now cancels the in-flight probe and clears it (plus the
  throttle timestamp it already cleared), so Tor restart/dormant/
  shutdown genuinely resets the verifier.
- A probe generation counter keeps actor reentrancy safe: a cancelled/
  superseded probe cannot re-seed the throttle/cache that invalidate()
  just cleared or clobber a fresh probe's in-flight slot; its awaiting
  callers resolve promptly as false.

Concurrent-caller join semantics are preserved (one shared probe), and
the race resolves exactly once behind an NSLock never held across an
await. Tests cover the hung-probe timeout bound, invalidate-cancels-
in-flight, post-restart recovery, and the shared-probe invariant, all
with the injected probe/clock harness (no real network).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bitchatTests/Nostr/TorEgressVerifierTests.swift

@@ -19,20 +19,72 @@ struct TorEgressVerifierTests {
         private let lock = NSLock()
         private var _now = Date(timeIntervalSince1970: 1_000_000)
         private var _result: TorEgressVerifier.ProbeResult = .verifiedTor
-        private(set) var probeCount = 0
+        private var _hanging = false
+        private var _gated = false
+        private var _released = false
+        private var _probeCount = 0
+        private var _cancelledCount = 0
 
         var now: Date {
             lock.lock(); defer { lock.unlock() }; return _now
         }
+        var probeCount: Int {
+            lock.lock(); defer { lock.unlock() }; return _probeCount
+        }
+        /// Number of hung probes that observed cooperative cancellation.
+        var cancelledCount: Int {
+            lock.lock(); defer { lock.unlock() }; return _cancelledCount
+        }
         func advance(_ seconds: TimeInterval) {
             lock.lock(); _now = _now.addingTimeInterval(seconds); lock.unlock()
         }
         func setResult(_ r: TorEgressVerifier.ProbeResult) {
             lock.lock(); _result = r; lock.unlock()
         }
+        /// When `true`, probes park forever and only exit via cooperative
+        /// cancellation — models a canary request wedged by
+        /// `waitsForConnectivity` deferring the request timer.
+        func setHanging(_ hanging: Bool) {
+            lock.lock(); _hanging = hanging; lock.unlock()
+        }
+        /// When `true`, probes wait for `release()` before returning — models
+        /// a slow-but-completing canary for join-semantics tests.
+        func setGated(_ gated: Bool) {
+            lock.lock(); _gated = gated; lock.unlock()
+        }
+        func release() {
+            lock.lock(); _released = true; lock.unlock()
+        }
+        /// Suspends until at least `n` probes have started.
+        func waitUntilProbeCount(atLeast n: Int) async {
+            while probeCount < n {
+                try? await Task.sleep(nanoseconds: 1_000_000)
+            }
+        }
         func makeProbe() -> @Sendable () async -> TorEgressVerifier.ProbeResult {
             return { [self] in
-                lock.lock(); probeCount += 1; let r = _result; lock.unlock()
+                lock.lock()
+                _probeCount += 1
+                let r = _result
+                let hang = _hanging
+                let gated = _gated
+                lock.unlock()
+                if hang {
+                    // Park until cancelled (verifier watchdog or invalidate());
+                    // cancellation-responsive so no task outlives the test.
+                    while !Task.isCancelled {
+                        try? await Task.sleep(nanoseconds: 2_000_000)
+                    }
+                    lock.lock(); _cancelledCount += 1; lock.unlock()
+                    return .unreachable("hung probe cancelled")
+                }
+                if gated {
+                    while !Task.isCancelled {
+                        lock.lock(); let released = _released; lock.unlock()
+                        if released { break }
+                        try? await Task.sleep(nanoseconds: 1_000_000)
+                    }
+                }
                 return r
             }
         }
@@ -41,8 +93,19 @@ struct TorEgressVerifierTests {
         }
     }
 
-    private func makeVerifier(_ h: Harness, ttl: TimeInterval = 300, minRetry: TimeInterval = 5) -> TorEgressVerifier {
-        TorEgressVerifier(ttl: ttl, minRetryInterval: minRetry, now: h.nowProvider(), probe: h.makeProbe())
+    private func makeVerifier(
+        _ h: Harness,
+        ttl: TimeInterval = 300,
+        minRetry: TimeInterval = 5,
+        probeTimeout: TimeInterval = TorEgressVerifier.defaultProbeTimeout
+    ) -> TorEgressVerifier {
+        TorEgressVerifier(
+            ttl: ttl,
+            minRetryInterval: minRetry,
+            probeTimeout: probeTimeout,
+            now: h.nowProvider(),
+            probe: h.makeProbe()
+        )
     }
 
     @Test("verifiedTor allows and is cached within TTL (single probe)")
@@ -217,4 +280,124 @@ struct TorEgressVerifierTests {
         _ = await v.verify()
         #expect(await v.lastProbeResult() == .notTor)
     }
+
+    // MARK: - Liveness (probe timeout watchdog + invalidate cancellation)
+
+    @Test("a probe that never completes is bounded by probeTimeout and fails closed")
+    func hungProbeIsBoundedByTimeout() async {
+        let h = Harness()
+        h.setHanging(true)
+        let v = makeVerifier(h, ttl: 300, minRetry: 0, probeTimeout: 0.05)
+
+        // Without the watchdog this would wedge: waitsForConnectivity can
+        // defer the request timer, leaving the canary bounded only by the
+        // 7-day resource timeout.
+        #expect(await v.verify() == false)
+        let last = await v.lastProbeResult()
+        switch last {
+        case .unreachable:
+            break // fail-closed timeout verdict recorded
+        default:
+            Issue.record("expected .unreachable after probe timeout, got \(String(describing: last))")
+        }
+        #expect(v.hasFreshVerification == false)
+
+        // The hung probe task itself was cancelled (URLSession task would be
+        // torn down), not abandoned.
+        while h.cancelledCount < 1 {
+            try? await Task.sleep(nanoseconds: 1_000_000)
+        }
+        #expect(h.cancelledCount == 1)
+
+        // The in-flight slot was cleared: the next verify() starts a fresh
+        // probe (does not join the hung one) and recovers.
+        h.setHanging(false)
+        h.setResult(.verifiedTor)
+        #expect(await v.verify() == true)
+        #expect(h.probeCount == 2)
+        #expect(v.hasFreshVerification == true)
+    }
+
+    @Test("invalidate() cancels the in-flight probe and the awaiting caller fails closed")
+    func invalidateCancelsInFlightProbe() async {
+        let h = Harness()
+        h.setHanging(true)
+        // Long (real-time) timeout and throttle: only invalidate() can
+        // unblock the caller, and only invalidate() clearing the throttle
+        // lets the follow-up probe run without advancing the clock.
+        let v = makeVerifier(h, ttl: 300, minRetry: 600, probeTimeout: 600)
+
+        let first = Task { await v.verify() }
+        await h.waitUntilProbeCount(atLeast: 1)
+        await v.invalidate()
+
+        // The awaiting caller resolves promptly (no 600s wait) and refuses.
+        #expect(await first.value == false)
+        #expect(v.hasFreshVerification == false)
+
+        // The hung probe observed cancellation (a Tor restart genuinely
+        // shakes the wedged canary request).
+        while h.cancelledCount < 1 {
+            try? await Task.sleep(nanoseconds: 1_000_000)
+        }
+
+        // Recovery: a fresh verify() runs a NEW probe — it neither joins the
+        // cancelled one nor inherits its throttle/last-result state.
+        h.setHanging(false)
+        h.setResult(.verifiedTor)
+        #expect(await v.verify() == true)
+        #expect(h.probeCount == 2)
+        #expect(v.hasFreshVerification == true)
+    }
+
+    @Test("recovery after a hung probe survives invalidate + Tor restart cycle")
+    func hungThenInvalidatedThenRecovers() async {
+        let h = Harness()
+        h.setHanging(true)
+        let v = makeVerifier(h, ttl: 300, minRetry: 5, probeTimeout: 600)
+
+        // Wedge one probe, then simulate a Tor restart mid-flight.
+        let wedged = Task { await v.verify() }
+        await h.waitUntilProbeCount(atLeast: 1)
+        await v.invalidate()
+        #expect(await wedged.value == false)
+
+        // Canary still down right after restart: fresh probe, fail closed —
+        // and the throttle applies to the FRESH result (bounded retry intact).
+        h.setHanging(false)
+        h.setResult(.unreachable("circuit not built"))
+        #expect(await v.verify() == false)
+        #expect(h.probeCount == 2)
+        h.advance(1)
+        #expect(await v.verify() == false)
+        #expect(h.probeCount == 2) // throttled, no hammering
+
+        // Canary returns after the retry window: verification recovers.
+        h.setResult(.verifiedTor)
+        h.advance(5)
+        #expect(await v.verify() == true)
+        #expect(v.hasFreshVerification == true)
+    }
+
+    @Test("concurrent verify() callers still share a single in-flight probe")
+    func concurrentCallersShareOneProbe() async {
+        let h = Harness()
+        h.setResult(.verifiedTor)
+        h.setGated(true)
+        let v = makeVerifier(h, ttl: 300, minRetry: 0)
+
+        let t1 = Task { await v.verify() }
+        await h.waitUntilProbeCount(atLeast: 1)
+        let t2 = Task { await v.verify() }
+        // Give t2 a chance to join the in-flight probe before releasing it.
+        // Either way the invariant holds: t2 joins the shared probe, or (if
+        // scheduled after completion) hits the fresh TTL cache — exactly one
+        // probe runs.
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        h.release()
+
+        #expect(await t1.value == true)
+        #expect(await t2.value == true)
+        #expect(h.probeCount == 1)
+    }
 }

localPackages/Arti/Sources/TorEgressVerifier.swift

@@ -44,6 +44,20 @@ import Foundation
 ///     gate, GeoRelayDirectory backoff) re-probe and succeed once the canary
 ///     is reachable again.
 ///
+/// Liveness:
+///   - Every probe is hard-bounded by `probeTimeout` via an independent async
+///     watchdog. The proxied session sets `waitsForConnectivity = true`, which
+///     can defer the per-request timer indefinitely, leaving the request
+///     bounded only by the default 7-day resource timeout — without the
+///     watchdog a hung canary would wedge every subsequent `verify()` caller.
+///     On timeout the probe task is cancelled (cooperatively cancelling the
+///     underlying `URLSessionTask`), the verdict is the fail-closed
+///     `.unreachable`, and the in-flight slot is cleared so the next
+///     `verify()` (after the retry throttle) starts a fresh probe.
+///   - `invalidate()` cancels any in-flight probe and clears all cached state
+///     (including the retry throttle), so a Tor restart/dormant/shutdown
+///     genuinely resets the verifier: a hung probe cannot survive it.
+///
 /// The probe is injectable so the policy/caching logic is unit-tested without a
 /// live network (see `TorEgressVerifierTests`).
 public actor TorEgressVerifier {
@@ -56,6 +70,11 @@ public actor TorEgressVerifier {
         case unreachable(String)
     }
 
+    /// Hard upper bound for a single canary probe, enforced independently of
+    /// URLSession timers (see the "Liveness" section of the type doc). Matches
+    /// the live probe's per-request timeout.
+    public static let defaultProbeTimeout: TimeInterval = 20
+
     private let probe: @Sendable () async -> ProbeResult
     private let now: @Sendable () -> Date
     private let ttl: TimeInterval
@@ -63,11 +82,18 @@ public actor TorEgressVerifier {
     /// persistent `.unreachable` cannot hammer the canary endpoint on every
     /// reconnect burst.
     private let minRetryInterval: TimeInterval
+    /// Outer wall-clock bound on a single probe (watchdog; fail-closed).
+    private let probeTimeout: TimeInterval
 
     private var lastVerifiedAt: Date?
     private var lastProbeAt: Date?
     private var lastResult: ProbeResult?
     private var inFlight: Task<Bool, Never>?
+    /// Bumped whenever a new probe starts or `invalidate()` runs. A completing
+    /// probe only records its outcome (cache/throttle) and clears `inFlight`
+    /// if its generation is still current, so a cancelled/superseded probe
+    /// cannot clobber state owned by a fresh one (actor-reentrancy safety).
+    private var probeGeneration = 0
 
     /// Lock-protected mirror of "verified within TTL" so synchronous gates
     /// (e.g. `NostrRelayManager`'s connect path) can consult the cache without
@@ -95,18 +121,26 @@ public actor TorEgressVerifier {
     public init(
         ttl: TimeInterval,
         minRetryInterval: TimeInterval = 5.0,
+        probeTimeout: TimeInterval = TorEgressVerifier.defaultProbeTimeout,
         now: @escaping @Sendable () -> Date = Date.init,
         probe: @escaping @Sendable () async -> ProbeResult
     ) {
         self.ttl = ttl
         self.minRetryInterval = minRetryInterval
+        self.probeTimeout = probeTimeout
         self.now = now
         self.probe = probe
     }
 
     /// Drop any cached verification (e.g. after a Tor restart or when the
-    /// network path changes). The next `verify()` re-probes.
+    /// network path changes) AND cancel any in-flight probe. The next
+    /// `verify()` starts a fresh probe — it neither joins the cancelled one
+    /// nor is throttled by its outcome, so a probe hung from before a Tor
+    /// restart cannot wedge callers after it.
     public func invalidate() {
+        probeGeneration += 1
+        inFlight?.cancel()
+        inFlight = nil
         lastVerifiedAt = nil
         lastProbeAt = nil
         lastResult = nil
@@ -137,10 +171,15 @@ public actor TorEgressVerifier {
         }
         if let inFlight { return await inFlight.value }
 
-        let task = Task<Bool, Never> { await self.runProbe() }
+        probeGeneration += 1
+        let generation = probeGeneration
+        let task = Task<Bool, Never> { await self.runProbe(generation: generation) }
         inFlight = task
         let allowed = await task.value
-        inFlight = nil
+        // Only clear the slot if this probe is still the current one: an
+        // `invalidate()` while we were suspended has already cleared it and a
+        // newer probe may occupy it (do not clobber the fresh task).
+        if probeGeneration == generation { inFlight = nil }
         return allowed
     }
 
@@ -158,8 +197,13 @@ public actor TorEgressVerifier {
         }
     }
 
-    private func runProbe() async -> Bool {
-        let result = await probe()
+    private func runProbe(generation: Int) async -> Bool {
+        let result = await boundedProbe()
+        // Superseded by `invalidate()` (Tor restart/dormant/shutdown) while the
+        // probe ran: its verdict predates the reset, so discard it — recording
+        // it would re-seed the throttle/cache that invalidate() just cleared.
+        // Fail closed for the callers that were awaiting this probe.
+        guard generation == probeGeneration else { return false }
         lastProbeAt = now()
         lastResult = result
         switch result {
@@ -185,6 +229,95 @@ public actor TorEgressVerifier {
             return false
         }
     }
+
+    /// Runs the injected probe raced against `probeTimeout`, guaranteeing a
+    /// result in bounded time regardless of URLSession timer behavior (the
+    /// proxied session's `waitsForConnectivity` can defer the per-request
+    /// timeout indefinitely). Whichever side loses the race is cancelled:
+    /// - on timeout, the probe task is cancelled (URLSession's async APIs
+    ///   cancel the underlying `URLSessionTask` cooperatively) and the result
+    ///   is the fail-closed `.unreachable`;
+    /// - on completion, the watchdog's sleep is cancelled so no timer lingers.
+    /// Cancelling the enclosing task (`invalidate()`) resolves immediately as
+    /// `.unreachable` and cancels both sides.
+    ///
+    /// `nonisolated` so the race body never re-enters the actor; it touches
+    /// only immutable `Sendable` state.
+    nonisolated private func boundedProbe() async -> ProbeResult {
+        let probe = self.probe
+        let timeout = self.probeTimeout
+        let race = ProbeRace()
+        return await withTaskCancellationHandler {
+            await withCheckedContinuation { (continuation: CheckedContinuation<ProbeResult, Never>) in
+                race.install(continuation)
+                let probeTask = Task { race.finish(await probe()) }
+                let watchdog = Task {
+                    try? await Task.sleep(nanoseconds: UInt64(max(0, timeout) * 1_000_000_000))
+                    guard !Task.isCancelled else { return }
+                    race.finish(.unreachable("probe timed out after \(Int(timeout))s"))
+                }
+                race.register(probeTask: probeTask, watchdog: watchdog)
+            }
+        } onCancel: {
+            race.finish(.unreachable("probe cancelled"))
+        }
+    }
+
+    /// Resolve-once rendezvous for the probe/watchdog race. Lock-protected
+    /// (never held across an await); the first `finish()` wins, resumes the
+    /// continuation exactly once, and cancels both tasks.
+    private final class ProbeRace: @unchecked Sendable {
+        private let lock = NSLock()
+        private var continuation: CheckedContinuation<ProbeResult, Never>?
+        private var pendingResult: ProbeResult?
+        private var resolved = false
+        private var probeTask: Task<Void, Never>?
+        private var watchdog: Task<Void, Never>?
+
+        func install(_ continuation: CheckedContinuation<ProbeResult, Never>) {
+            lock.lock()
+            if let result = pendingResult {
+                // finish() ran before the continuation existed (e.g. the
+                // enclosing task was already cancelled): resolve immediately.
+                pendingResult = nil
+                lock.unlock()
+                continuation.resume(returning: result)
+                return
+            }
+            self.continuation = continuation
+            lock.unlock()
+        }
+
+        func register(probeTask: Task<Void, Never>, watchdog: Task<Void, Never>) {
+            lock.lock()
+            if resolved {
+                lock.unlock()
+                probeTask.cancel()
+                watchdog.cancel()
+                return
+            }
+            self.probeTask = probeTask
+            self.watchdog = watchdog
+            lock.unlock()
+        }
+
+        func finish(_ result: ProbeResult) {
+            lock.lock()
+            guard !resolved else { lock.unlock(); return }
+            resolved = true
+            let continuation = self.continuation
+            self.continuation = nil
+            if continuation == nil { pendingResult = result }
+            let probeTask = self.probeTask
+            let watchdog = self.watchdog
+            self.probeTask = nil
+            self.watchdog = nil
+            lock.unlock()
+            probeTask?.cancel()
+            watchdog?.cancel()
+            continuation?.resume(returning: result)
+        }
+    }
 }
 
 // MARK: - Live probe
@@ -198,9 +331,13 @@ public extension TorEgressVerifier {
     ///
     /// Follow-up (see PR): make the canary endpoint configurable and add an
     /// onion-service canary so verification does not depend on a single host.
+    /// Note: the per-request `timeoutInterval` below is best-effort only — the
+    /// proxied session's `waitsForConnectivity` can defer it. The authoritative
+    /// bound is the verifier's `probeTimeout` watchdog, whose cancellation
+    /// propagates into `session.data(for:)` and cancels the URLSessionTask.
     static func liveProbe(
         endpoint: URL = URL(string: "https://check.torproject.org/api/ip")!,
-        timeout: TimeInterval = 20
+        timeout: TimeInterval = TorEgressVerifier.defaultProbeTimeout
     ) -> @Sendable () async -> ProbeResult {
         return {
             var request = URLRequest(url: endpoint)