Make location notes robust: durable relay subscriptions, failure decay, auto-recovery (#1333)

* Make location notes robust: durable relay subscriptions, failure decay, auto-recovery Location notes (and geohash chat / DMs) intermittently stopped showing events because the Nostr relay layer lost subscriptions and blacklisted relays: - Replay active subscriptions on every relay (re)connect. Relays drop REQs with the socket; previously a drop silently killed the subscription on that relay for the rest of the session. Durable subscription intent now also survives disconnect()/resetAllConnections (background -> foreground). - Keep failed REQ sends queued instead of dropping them. - Raise the EOSE fallback from a fixed 2s Timer to a 10s injected schedule (Tor needs more than 2s), and settle EOSE trackers when a relay disconnects before answering so initial load doesn't stall. - Decay "permanently failed" relay markings after a 10-minute cooldown; previously ~9 minutes of outage (or one DNS hiccup) excluded a relay until app restart, with nothing resetting it on macOS. - Make geo relay selection deterministic (distance, then host) so publishers and subscribers with the same directory agree on relays. - Post .geoRelayDirectoryDidRefresh after a directory fetch and let LocationNotesManager auto-resubscribe out of the "no relays" state instead of requiring a manual retry. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Guard subscription activation on connection identity A REQ send completion from a dead socket could land after handleDisconnection cleared the relay's subscriptions and re-mark the subscription active, making the next connection skip the durable replay and leave that relay silent. Only mark a subscription active if the completing socket is still the relay's live connection. Regression test defers send completions in the mock so the stale completion deterministically interleaves between disconnect and reconnect. Addresses Codex review on #1333. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: jack <jackjackbits@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-24 21:45:20 +00:00 · 2026-06-11 09:28:23 +01:00
parent 93d01b8fa6
commit 75dd83d9cc
7 changed files with 388 additions and 66 deletions
@@ -7,6 +7,11 @@ import UIKit
 import AppKit
 #endif

+extension Notification.Name {
+    /// Posted after the geo relay directory successfully refreshes its entries.
+    static let geoRelayDirectoryDidRefresh = Notification.Name("bitchat.geoRelayDirectoryDidRefresh")
+}
+
 /// Directory of online Nostr relays with approximate GPS locations, used for geohash routing.
 struct GeoRelayDirectoryDependencies {
    var userDefaults: UserDefaults
@@ -165,33 +170,16 @@ final class GeoRelayDirectory {
    }

    /// Returns up to `count` relay URLs (wss://) closest to the given coordinate.
+    /// Ties break by host so every device with the same directory picks the
+    /// same relay set — publishers and subscribers must agree on relays.
    func closestRelays(toLat lat: Double, lon: Double, count: Int = 5) -> [String] {
        guard !entries.isEmpty, count > 0 else { return [] }

-        if entries.count <= count {
-            return entries
-                .sorted { a, b in
-                    haversineKm(lat, lon, a.lat, a.lon) < haversineKm(lat, lon, b.lat, b.lon)
-                }
-                .map { "wss://\($0.host)" }
-        }
-
-        var best: [(entry: Entry, distance: Double)] = []
-        best.reserveCapacity(count)
-
-        for entry in entries {
-            let distance = haversineKm(lat, lon, entry.lat, entry.lon)
-            if best.count < count {
-                let idx = best.firstIndex { $0.distance > distance } ?? best.count
-                best.insert((entry, distance), at: idx)
-            } else if let worstDistance = best.last?.distance, distance < worstDistance {
-                let idx = best.firstIndex { $0.distance > distance } ?? best.count
-                best.insert((entry, distance), at: idx)
-                best.removeLast()
-            }
-        }
-
-        return best.map { "wss://\($0.entry.host)" }
+        return entries
+            .map { (entry: $0, distance: haversineKm(lat, lon, $0.lat, $0.lon)) }
+            .sorted { ($0.distance, $0.entry.host) < ($1.distance, $1.entry.host) }
+            .prefix(count)
+            .map { "wss://\($0.entry.host)" }
    }

    // MARK: - Remote Fetch
@@ -289,6 +277,8 @@ final class GeoRelayDirectory {
        isFetching = false
        retryAttempt = 0
        cancelRetry()
+        // Let waiters (e.g. location notes stuck in a "no relays" state) retry.
+        dependencies.notificationCenter.post(name: .geoRelayDirectoryDidRefresh, object: nil)
    }

    @MainActor
@@ -171,9 +171,10 @@ final class NostrRelayManager: ObservableObject {
    private struct EOSETracker {
        var pendingRelays: Set<String>
        var callback: () -> Void
-        var timer: Timer?
+        let epoch: Int
    }
    private var eoseTrackers: [String: EOSETracker] = [:]
+    private var eoseTrackerEpoch = 0
    private var pendingEOSECallbacks: [String: () -> Void] = [:]
    
    // Message queue for reliability
@@ -264,15 +265,18 @@ final class NostrRelayManager: ObservableObject {
            task.cancel(with: .goingAway, reason: nil)
        }
        connections.removeAll()
-        // Clear known subscriptions and any queued subs since connections are gone
+        // Sockets are gone, so per-relay subscription state is cleared — but
+        // durable intent (subscriptionRequestState, messageHandlers, parked
+        // EOSE callbacks) is kept so REQs replay when relays reconnect
+        // (e.g. background → foreground).
        subscriptions.removeAll()
        pendingSubscriptions.removeAll()
-        subscriptionRequestState.removeAll()
-        pendingEOSECallbacks.removeAll()
-        for (_, tracker) in eoseTrackers {
-            tracker.timer?.invalidate()
-        }
+        // Settle in-flight initial loads instead of leaving callers hanging.
+        let trackers = eoseTrackers
        eoseTrackers.removeAll()
+        for (_, tracker) in trackers {
+            tracker.callback()
+        }
        pendingTorConnectionURLs.removeAll()
        awaitingTorForConnections = false
        torReadyWaitAttempts = 0
@@ -516,7 +520,6 @@ final class NostrRelayManager: ObservableObject {
        subscribeCoalesce.removeValue(forKey: id)
        subscriptionRequestState.removeValue(forKey: id)
        pendingEOSECallbacks.removeValue(forKey: id)
-        eoseTrackers[id]?.timer?.invalidate()
        eoseTrackers.removeValue(forKey: id)
        for url in Array(pendingSubscriptions.keys) {
            pendingSubscriptions[url]?.removeValue(forKey: id)
@@ -630,20 +633,18 @@ final class NostrRelayManager: ObservableObject {
    }

    private func startEOSETracking(id: String, relayURLs: Set<String>, callback: @escaping () -> Void) {
-        eoseTrackers[id]?.timer?.invalidate()
-        var tracker = EOSETracker(pendingRelays: relayURLs, callback: callback, timer: nil)
+        eoseTrackerEpoch += 1
+        let epoch = eoseTrackerEpoch
+        eoseTrackers[id] = EOSETracker(pendingRelays: relayURLs, callback: callback, epoch: epoch)
        // Fallback timeout to avoid hanging if a relay never sends EOSE.
-        tracker.timer = Timer.scheduledTimer(withTimeInterval: 2.0, repeats: false) { [weak self] _ in
-            Task { @MainActor in
+        dependencies.scheduleAfter(TransportConfig.nostrSubscriptionEOSEFallbackSeconds) { [weak self] in
+            Task { @MainActor [weak self] in
                guard let self else { return }
-                if let tracker = self.eoseTrackers[id] {
-                    tracker.timer?.invalidate()
-                    self.eoseTrackers.removeValue(forKey: id)
-                    callback()
-                }
+                guard let tracker = self.eoseTrackers[id], tracker.epoch == epoch else { return }
+                self.eoseTrackers.removeValue(forKey: id)
+                tracker.callback()
            }
        }
-        eoseTrackers[id] = tracker
    }

    private func startPendingEOSETrackingIfNeeded(id: String) {
@@ -764,26 +765,35 @@ final class NostrRelayManager: ObservableObject {
        }
    }

-    /// Send any queued subscriptions for a relay that just connected.
+    /// Send queued subscriptions and replay durable ones for a relay that just
+    /// (re)connected. Relays drop subscriptions with the socket, so every
+    /// active subscription targeting this relay must be re-sent.
    private func flushPendingSubscriptions(for relayUrl: String) {
-        guard let map = pendingSubscriptions[relayUrl], !map.isEmpty else { return }
        guard let connection = connections[relayUrl] else { return }
-        for (id, messageString) in map {
+        var toSend = pendingSubscriptions[relayUrl] ?? [:]
+        for (id, state) in subscriptionRequestState where state.relayURLs.contains(relayUrl) && toSend[id] == nil {
+            toSend[id] = state.messageString
+        }
+        for (id, messageString) in toSend {
            if self.subscriptions[relayUrl]?.contains(id) == true { continue }
            startPendingEOSETrackingIfNeeded(id: id)
-            connection.send(.string(messageString)) { error in
-                if let error = error {
-                    SecureLogger.error("❌ Failed to send pending subscription to \(relayUrl): \(error)", category: .session)
-                } else {
-                    Task { @MainActor in
-                        var subs = self.subscriptions[relayUrl] ?? Set<String>()
-                        subs.insert(id)
-                        self.subscriptions[relayUrl] = subs
+            connection.send(.string(messageString)) { [weak self, weak connection] error in
+                Task { @MainActor [weak self] in
+                    guard let self else { return }
+                    if let error = error {
+                        // Keep the pending entry; the next (re)connect retries it.
+                        SecureLogger.error("❌ Failed to send pending subscription to \(relayUrl): \(error)", category: .session)
+                    } else {
+                        // A stale completion from a socket that has since been
+                        // replaced must not mark the subscription active, or
+                        // the next connection would skip replaying it.
+                        guard let connection, self.connections[relayUrl] === connection else { return }
+                        self.subscriptions[relayUrl, default: []].insert(id)
+                        self.pendingSubscriptions[relayUrl]?.removeValue(forKey: id)
                    }
                }
            }
        }
-        pendingSubscriptions[relayUrl] = nil
    }
    
    private func receiveMessage(from task: NostrRelayConnectionProtocol, relayUrl: String) {
@@ -850,7 +860,6 @@ final class NostrRelayManager: ObservableObject {
            if var tracker = eoseTrackers[subId] {
                tracker.pendingRelays.remove(relayUrl)
                if tracker.pendingRelays.isEmpty {
-                    tracker.timer?.invalidate()
                    eoseTrackers.removeValue(forKey: subId)
                    tracker.callback()
                } else {
@@ -924,17 +933,30 @@ final class NostrRelayManager: ObservableObject {
        isConnected = relays.contains { $0.isConnected }
    }
    
-    private func handleDisconnection(relayUrl: String, error: Error) {
-        // If networking is disallowed, do not schedule reconnection
-        if !dependencies.activationAllowed() {
-            connections.removeValue(forKey: relayUrl)
-            subscriptions.removeValue(forKey: relayUrl)
-            updateRelayStatus(relayUrl, isConnected: false, error: error)
-            return
+    /// A relay that drops before sending EOSE must not stall initial-load
+    /// callbacks; treat it as done and let the remaining relays (or the
+    /// fallback timeout) drive completion.
+    private func settleEOSETrackers(droppingRelay relayUrl: String) {
+        for (id, var tracker) in eoseTrackers where tracker.pendingRelays.contains(relayUrl) {
+            tracker.pendingRelays.remove(relayUrl)
+            if tracker.pendingRelays.isEmpty {
+                eoseTrackers.removeValue(forKey: id)
+                tracker.callback()
+            } else {
+                eoseTrackers[id] = tracker
+            }
        }
+    }
+
+    private func handleDisconnection(relayUrl: String, error: Error) {
        connections.removeValue(forKey: relayUrl)
        subscriptions.removeValue(forKey: relayUrl)
        updateRelayStatus(relayUrl, isConnected: false, error: error)
+        settleEOSETrackers(droppingRelay: relayUrl)
+        // If networking is disallowed, do not schedule reconnection
+        if !dependencies.activationAllowed() {
+            return
+        }
        
        // Check if this is a DNS or handshake error; treat as permanent
        let errorDescription = error.localizedDescription.lowercased()
@@ -1064,6 +1086,13 @@ final class NostrRelayManager: ObservableObject {
    // MARK: - Failure classification
    private func isPermanentlyFailed(_ url: String) -> Bool {
        guard let r = relays.first(where: { $0.url == url }) else { return false }
+        // Failures decay: after a cooldown the relay gets another chance, so a
+        // long network outage or transient relay trouble can't blacklist it
+        // for the rest of the process lifetime.
+        if let lastDisconnect = r.lastDisconnectedAt,
+           dependencies.now().timeIntervalSince(lastDisconnect) >= TransportConfig.nostrRelayFailureCooldownSeconds {
+            return false
+        }
        if r.reconnectAttempts >= maxReconnectAttempts { return true }
        if let ns = r.lastError as NSError?, ns.domain == NSURLErrorDomain {
            if ns.code == NSURLErrorBadServerResponse || ns.code == NSURLErrorCannotFindHost {
@@ -1,4 +1,5 @@
 import BitLogger
+import Combine
 import Foundation

 /// Dependencies for location notes, allowing tests to stub relay/identity behavior.
@@ -14,7 +15,9 @@ struct LocationNotesDependencies {
    var sendEvent: SendEvent
    var deriveIdentity: (_ geohash: String) throws -> NostrIdentity
    var now: () -> Date
-    
+    // Fires when the geo relay directory refreshes; used to retry after "no relays".
+    var relayDirectoryUpdates: AnyPublisher<Void, Never> = Empty(completeImmediately: false).eraseToAnyPublisher()
+
    private static let idBridge = NostrIdentityBridge()

    static let live = LocationNotesDependencies(
@@ -39,7 +42,11 @@ struct LocationNotesDependencies {
        deriveIdentity: { geohash in
            try idBridge.deriveIdentity(forGeohash: geohash)
        },
-        now: { Date() }
+        now: { Date() },
+        relayDirectoryUpdates: NotificationCenter.default
+            .publisher(for: .geoRelayDirectoryDidRefresh)
+            .map { _ in () }
+            .eraseToAnyPublisher()
    )
 }

@@ -77,6 +84,7 @@ final class LocationNotesManager: ObservableObject {
    @Published private(set) var errorMessage: String?
    private var subscriptionID: String?
    private var noteIDs = Set<String>() // O(1) duplicate detection
+    private var directoryUpdateCancellable: AnyCancellable?
    private let dependencies: LocationNotesDependencies
    private let maxNotesInMemory = 500 // Defensive cap (relay limit is 200)

@@ -101,6 +109,15 @@ final class LocationNotesManager: ObservableObject {
            SecureLogger.warning("LocationNotesManager: invalid geohash '\(norm)' (expected 8 valid base32 chars)", category: .session)
        }
        subscribe()
+        // The relay directory may load after init (remote fetch over Tor);
+        // retry automatically instead of staying stuck on "no relays".
+        directoryUpdateCancellable = dependencies.relayDirectoryUpdates
+            .sink { [weak self] in
+                Task { @MainActor [weak self] in
+                    guard let self, self.state == .noRelays else { return }
+                    self.subscribe()
+                }
+            }
    }

    func setGeohash(_ newGeohash: String) {
@@ -156,6 +156,11 @@ enum TransportConfig {
    // bootstrap deadline) to attempt before unblocking pending EOSE callers.
    static let nostrTorReadyMaxWaitAttempts: Int = 3
    static let nostrPendingSendQueueCap: Int = 200
+    // Fallback deadline for treating a subscription's initial fetch as complete
+    // when a relay never sends EOSE (generous to cover Tor circuit setup).
+    static let nostrSubscriptionEOSEFallbackSeconds: TimeInterval = 10.0
+    // After this long, a relay marked permanently failed gets another chance.
+    static let nostrRelayFailureCooldownSeconds: TimeInterval = 600.0

    // Geo relay directory
    static let geoRelayFetchIntervalSeconds: TimeInterval = 60 * 60 * 24
@@ -1,4 +1,5 @@
 import Testing
+import Combine
 import Foundation
@testable import bitchat

@@ -26,6 +27,41 @@ struct LocationNotesManagerTests {
        #expect(manager.errorMessage == String(localized: "location_notes.error.no_relays"))
    }

+    @Test
+    func noRelays_resubscribesWhenDirectoryRefreshes() async throws {
+        var relays: [String] = []
+        var subscribeCount = 0
+        let directorySubject = PassthroughSubject<Void, Never>()
+        let deps = LocationNotesDependencies(
+            relayLookup: { _, _ in relays },
+            subscribe: { _, _, _, _, _ in
+                subscribeCount += 1
+            },
+            unsubscribe: { _ in },
+            sendEvent: { _, _ in },
+            deriveIdentity: { _ in try NostrIdentity.generate() },
+            now: { Date() },
+            relayDirectoryUpdates: directorySubject.eraseToAnyPublisher()
+        )
+
+        let manager = LocationNotesManager(geohash: "u4pruydq", dependencies: deps)
+        #expect(manager.state == .noRelays)
+        #expect(subscribeCount == 0)
+
+        // Directory loads later (e.g. remote fetch finished after Tor came up).
+        relays = ["wss://relay.one"]
+        directorySubject.send(())
+
+        let deadline = Date().addingTimeInterval(1.0)
+        while manager.state == .noRelays && Date() < deadline {
+            try await Task.sleep(nanoseconds: 10_000_000)
+        }
+
+        #expect(subscribeCount == 1)
+        #expect(manager.state == .loading)
+        #expect(manager.errorMessage == nil)
+    }
+
    @Test
    func sendWithoutRelays_surfacesNoRelaysError() {
        var sendCalled = false
@@ -55,6 +55,47 @@ final class GeoRelayDirectoryTests: XCTestCase {
        )
    }

+    func test_closestRelays_breaksDistanceTiesDeterministicallyByHost() {
+        // Same coordinates for all entries: selection must still be stable so
+        // publishers and subscribers using the same directory agree on relays.
+        let harness = makeHarness(
+            cacheCSV: """
+            relay url,lat,lon
+            zeta.example,10,10
+            alpha.example,10,10
+            mike.example,10,10
+            """
+        )
+        let directory = GeoRelayDirectory(dependencies: harness.dependencies)
+
+        XCTAssertEqual(
+            directory.closestRelays(toLat: 10, lon: 10, count: 2),
+            ["wss://alpha.example", "wss://mike.example"]
+        )
+    }
+
+    func test_fetchSuccess_postsDirectoryRefreshNotification() async {
+        let harness = makeHarness(fetchCSV: """
+        relay url,lat,lon
+        notify.example,1,2
+        """)
+        let directory = GeoRelayDirectory(dependencies: harness.dependencies)
+
+        var notified = 0
+        let observer = harness.notificationCenter.addObserver(
+            forName: .geoRelayDirectoryDidRefresh,
+            object: nil,
+            queue: .main
+        ) { _ in
+            notified += 1
+        }
+        defer { harness.notificationCenter.removeObserver(observer) }
+
+        directory.prefetchIfNeeded()
+        let refreshed = await waitUntil { notified == 1 }
+        XCTAssertTrue(refreshed)
+    }
+
    func test_loadLocalEntries_prefersCacheThenBundleThenWorkingDirectory() {
        let cacheHarness = makeHarness(
            cacheCSV: """
@@ -761,7 +761,12 @@ final class NostrRelayManagerTests: XCTestCase {
        }
        XCTAssertTrue(subscribed)

-        let timedOut = await waitUntil(timeout: 3.0) { eoseCount == 1 }
+        // The fallback is scheduled but has not fired yet.
+        XCTAssertEqual(context.scheduler.scheduled.first?.delay, TransportConfig.nostrSubscriptionEOSEFallbackSeconds)
+        XCTAssertEqual(eoseCount, 0)
+
+        context.scheduler.runNext()
+        let timedOut = await waitUntil { eoseCount == 1 }
        XCTAssertTrue(timedOut)

        try context.sessionFactory.latestConnection(for: relayURL)?.emitEOSE(subscriptionID: "timeout")
@@ -769,6 +774,192 @@ final class NostrRelayManagerTests: XCTestCase {
        XCTAssertEqual(eoseCount, 1)
    }

+    func test_eose_completesWhenRelayDisconnectsBeforeEOSE() async throws {
+        let relayOne = "wss://eose-drop-one.example"
+        let relayTwo = "wss://eose-drop-two.example"
+        let context = makeContext(permission: .denied)
+        var eoseCount = 0
+
+        context.manager.subscribe(
+            filter: makeFilter(),
+            id: "eose-drop",
+            relayUrls: [relayOne, relayTwo],
+            handler: { _ in },
+            onEOSE: { eoseCount += 1 }
+        )
+
+        let subscribed = await waitUntil {
+            context.sessionFactory.latestConnection(for: relayOne)?.sentStrings.count == 1 &&
+            context.sessionFactory.latestConnection(for: relayTwo)?.sentStrings.count == 1
+        }
+        XCTAssertTrue(subscribed)
+
+        try context.sessionFactory.latestConnection(for: relayOne)?.emitEOSE(subscriptionID: "eose-drop")
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(eoseCount, 0)
+
+        context.sessionFactory.latestConnection(for: relayTwo)?.fail(
+            error: NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
+        )
+        let completed = await waitUntil { eoseCount == 1 }
+        XCTAssertTrue(completed)
+    }
+
+    func test_reconnect_replaysActiveSubscriptionsAndDeliversEvents() async throws {
+        let relayURL = "wss://replay.example"
+        let context = makeContext(permission: .denied)
+        var received: [NostrEvent] = []
+
+        context.manager.subscribe(
+            filter: makeFilter(),
+            id: "replay-sub",
+            relayUrls: [relayURL],
+            handler: { received.append($0) }
+        )
+        let subscribed = await waitUntil {
+            context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.contains { $0.contains("replay-sub") } == true
+        }
+        XCTAssertTrue(subscribed)
+
+        // Drop the socket; the relay forgets the subscription with it.
+        context.sessionFactory.latestConnection(for: relayURL)?.fail(
+            error: NSError(domain: NSURLErrorDomain, code: NSURLErrorNetworkConnectionLost)
+        )
+        let retryScheduled = await waitUntil { !context.scheduler.scheduled.isEmpty }
+        XCTAssertTrue(retryScheduled)
+        context.scheduler.runNext()
+
+        let replayed = await waitUntil {
+            let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
+            return connections.count == 2 &&
+                connections.last?.sentStrings.contains { $0.contains("replay-sub") } == true
+        }
+        XCTAssertTrue(replayed)
+
+        let event = try makeSignedEvent(content: "after reconnect")
+        try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "replay-sub", event: event)
+        let delivered = await waitUntil { received.count == 1 }
+        XCTAssertTrue(delivered)
+    }
+
+    func test_disconnectThenConnect_restoresSubscriptions() async {
+        let relayURL = "wss://restore.example"
+        let context = makeContext(permission: .denied)
+
+        context.manager.subscribe(filter: makeFilter(), id: "restore-sub", relayUrls: [relayURL], handler: { _ in })
+        let subscribed = await waitUntil {
+            context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.contains { $0.contains("restore-sub") } == true
+        }
+        XCTAssertTrue(subscribed)
+
+        // Background → foreground: connections reset, subscriptions must survive.
+        context.manager.disconnect()
+        context.manager.connect()
+
+        let resubscribed = await waitUntil {
+            let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
+            return connections.count == 2 &&
+                connections.last?.sentStrings.contains { $0.contains("restore-sub") } == true
+        }
+        XCTAssertTrue(resubscribed)
+    }
+
+    func test_subscriptionSendFailure_retriesOnReconnect() async {
+        let relayURL = "wss://flaky-send.example"
+        let context = makeContext(permission: .denied)
+        context.sessionFactory.sendErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
+
+        context.manager.subscribe(filter: makeFilter(), id: "flaky-sub", relayUrls: [relayURL], handler: { _ in })
+        let attempted = await waitUntil {
+            context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.isEmpty == false
+        }
+        XCTAssertTrue(attempted)
+
+        // The REQ send failed; the subscription must survive for the next connection.
+        context.sessionFactory.sendErrorByURL[relayURL] = nil
+        context.sessionFactory.latestConnection(for: relayURL)?.fail(
+            error: NSError(domain: NSURLErrorDomain, code: NSURLErrorNetworkConnectionLost)
+        )
+        let retryScheduled = await waitUntil { !context.scheduler.scheduled.isEmpty }
+        XCTAssertTrue(retryScheduled)
+        context.scheduler.runNext()
+
+        let resubscribed = await waitUntil {
+            let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
+            return connections.count == 2 &&
+                connections.last?.sentStrings.contains { $0.contains("flaky-sub") } == true
+        }
+        XCTAssertTrue(resubscribed)
+    }
+
+    func test_staleSendCompletionFromDeadSocket_doesNotBlockReplayOnNextConnection() async {
+        let relayURL = "wss://stale-completion.example"
+        let context = makeContext(permission: .denied)
+
+        context.manager.subscribe(filter: makeFilter(), id: "stale-sub", relayUrls: [relayURL], handler: { _ in })
+        // The connection exists synchronously; its REQ flush lands on a later
+        // main-queue tick, so deferring completions here is race-free.
+        let connectionA = context.sessionFactory.latestConnection(for: relayURL)
+        XCTAssertNotNil(connectionA)
+        connectionA?.deferSendCompletions = true
+
+        let reqSent = await waitUntil {
+            connectionA?.sentStrings.contains { $0.contains("stale-sub") } == true
+        }
+        XCTAssertTrue(reqSent)
+
+        // Socket dies while the REQ's send completion is still in flight.
+        connectionA?.fail(error: NSError(domain: NSURLErrorDomain, code: NSURLErrorNetworkConnectionLost))
+        let disconnected = await waitUntil {
+            context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == false
+        }
+        XCTAssertTrue(disconnected)
+
+        // The stale success completion must not mark the subscription active.
+        connectionA?.flushDeferredSendCompletions()
+        try? await Task.sleep(nanoseconds: 20_000_000)
+
+        context.scheduler.runNext()
+        let replayed = await waitUntil {
+            let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
+            return connections.count == 2 &&
+                connections.last?.sentStrings.contains { $0.contains("stale-sub") } == true
+        }
+        XCTAssertTrue(replayed)
+    }
+
+    func test_permanentFailure_decaysAfterCooldownAndRetries() async {
+        let relayURL = "wss://cooldown.example"
+        let context = makeContext(permission: .denied)
+        context.sessionFactory.pingErrorByURL[relayURL] = NSError(
+            domain: NSURLErrorDomain,
+            code: NSURLErrorCannotFindHost,
+            userInfo: [NSLocalizedDescriptionKey: "DNS failure"]
+        )
+
+        context.manager.ensureConnections(to: [relayURL])
+        let failed = await waitUntil {
+            context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts == TransportConfig.nostrRelayMaxReconnectAttempts
+        }
+        XCTAssertTrue(failed)
+
+        // Within the cooldown the relay is skipped.
+        let countBefore = context.sessionFactory.requestedURLs.count
+        context.manager.ensureConnections(to: [relayURL])
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(context.sessionFactory.requestedURLs.count, countBefore)
+
+        // After the cooldown it gets another chance and recovers.
+        context.sessionFactory.pingErrorByURL[relayURL] = nil
+        context.clock.now = context.clock.now.addingTimeInterval(TransportConfig.nostrRelayFailureCooldownSeconds + 1)
+        context.manager.ensureConnections(to: [relayURL])
+        let retried = await waitUntil {
+            context.sessionFactory.requestedURLs.count == countBefore + 1 &&
+            context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
+        }
+        XCTAssertTrue(retried)
+    }
+
    func test_receiveFailure_schedulesReconnectWithBackoff() async {
        let relayURL = "wss://retry.example"
        let context = makeContext(permission: .denied)
@@ -1222,9 +1413,22 @@ private final class MockRelayConnection: NostrRelayConnectionProtocol {
        cancelCallCount += 1
    }

+    var deferSendCompletions = false
+    private var deferredSendCompletions: [(Error?) -> Void] = []
+
    func send(_ message: URLSessionWebSocketTask.Message, completionHandler: @escaping (Error?) -> Void) {
        sentMessages.append(message)
-        completionHandler(sendError)
+        if deferSendCompletions {
+            deferredSendCompletions.append(completionHandler)
+        } else {
+            completionHandler(sendError)
+        }
+    }
+
+    func flushDeferredSendCompletions() {
+        let pending = deferredSendCompletions
+        deferredSendCompletions = []
+        pending.forEach { $0(sendError) }
    }

    func receive(completionHandler: @escaping (Result<URLSessionWebSocketTask.Message, Error>) -> Void) {