Fix transport reliability gaps: Tor stalls, weak-signal sends, GCS input validation

NostrRelayManager no longer strands work when Tor is slow to bootstrap: failed readiness waits retry (bounded by nostrTorReadyMaxWaitAttempts) instead of dropping queued relay connections, parked EOSE callbacks fire after exhaustion so callers never hang, and sends made before Tor is ready are queued locally (capped) instead of being dropped on a failed wait - still strictly fail-closed. MessageRouter now prefers a connected transport over a merely window-reachable one, and sends made on a weak reachability signal are retained in the outbox until a delivery/read ack confirms receipt (receivers dedup by message ID), with resends bounded by attempt count. GCS sync filters from the wire are bounds-checked (p in 1...32, m > 1) at both the packet decode and filter decode layers; oversized Golomb parameters previously decoded to garbage via silent shift overflow. BLELinkStateStore is now explicitly pinned to bleQueue: debug builds trap any access from another queue, enforcing the ownership discipline the surrounding code already relied on by convention. CI gains an iOS simulator build job (arm64 only; the vendored Arti xcframework has no x86_64 simulator slice) so iOS-conditional code paths are compile-checked - SPM tests only cover the macOS slice. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-24 22:25:18 +00:00 · 2026-06-10 16:22:35 +01:00
parent 4093ee6733
commit 6bda919dd4
10 changed files with 577 additions and 39 deletions
@@ -40,3 +40,25 @@ jobs:

      - name: Run Tests
        run: swift test --parallel --quiet --package-path ${{ matrix.path }}
+
+  # SPM tests above only compile the macOS slice; this job covers the
+  # iOS-conditional code paths (UIKit, CoreBluetooth restoration, etc.).
+  ios-build:
+    name: Build iOS app (simulator)
+    runs-on: macos-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v5
+
+      - name: Build iOS (simulator, no signing)
+        # arm64 only: the vendored arti.xcframework has no x86_64 simulator slice.
+        run: |
+          set -o pipefail
+          xcodebuild -project bitchat.xcodeproj \
+            -scheme "bitchat (iOS)" \
+            -sdk iphonesimulator \
+            -destination 'generic/platform=iOS Simulator' \
+            ARCHS=arm64 \
+            CODE_SIGNING_ALLOWED=NO \
+            build
@@ -96,7 +96,7 @@ struct RequestSyncPacket {
            }
        }

-        guard let pp = p, let mm = m, let dd = payload, pp >= 1, mm > 0 else { return nil }
+        guard let pp = p, let mm = m, let dd = payload, pp >= 1, pp <= GCSFilter.maxP, mm > 0 else { return nil }
        return RequestSyncPacket(p: pp, m: mm, data: dd, types: types, sinceTimestamp: sinceTimestamp, fragmentIdFilter: fragmentIdFilter)
    }
 }
@@ -142,11 +142,22 @@ final class NostrRelayManager: ObservableObject {
    private var subscriptions: [String: Set<String>] = [:] // relay URL -> active subscription IDs
    private var pendingSubscriptions: [String: [String: String]] = [:] // relay URL -> (subscription id -> encoded REQ JSON)
    private var messageHandlers: [String: (NostrEvent) -> Void] = [:]
+    private struct InboundEventKey: Hashable {
+        let subscriptionID: String
+        let eventID: String
+    }
+    private let recentInboundEventKeyLimit = TransportConfig.nostrInboundEventDedupCap
+    private let recentInboundEventKeyTrimTarget = TransportConfig.nostrInboundEventDedupTrimTarget
+    private var recentInboundEventKeys = Set<InboundEventKey>()
+    private var recentInboundEventKeyOrder: [InboundEventKey] = []
+    private var duplicateInboundEventDropCount = 0
+    private var duplicateInboundEventDropCountBySubscription: [String: Int] = [:]
    // Coalesce duplicate subscribe requests for the same id within a short window.
    private let subscribeCoalesceInterval: TimeInterval = 1.0
    private var subscribeCoalesce: [String: Date] = [:]
    private var pendingTorConnectionURLs = Set<String>()
    private var awaitingTorForConnections = false
+    private var torReadyWaitAttempts = 0
    private var cancellables = Set<AnyCancellable>()

    private struct SubscriptionRequestState: Equatable {
@@ -263,6 +274,7 @@ final class NostrRelayManager: ObservableObject {
        eoseTrackers.removeAll()
        pendingTorConnectionURLs.removeAll()
        awaitingTorForConnections = false
+        torReadyWaitAttempts = 0
        updateConnectionStatus()
    }
    
@@ -285,11 +297,14 @@ final class NostrRelayManager: ObservableObject {
        // Global network policy gate
        guard dependencies.activationAllowed() else { return }
        if shouldUseTor && dependencies.torEnforced() && !dependencies.torIsReady() {
-            // Defer sends until Tor is ready to avoid premature queueing
-            dependencies.awaitTorReady { [weak self] ready in
-                guard let self = self else { return }
-                if ready { self.sendEvent(event, to: relayUrls) }
-            }
+            // Fail-closed: nothing touches the network until Tor is up. Queue the
+            // event locally so it survives a slow bootstrap (queued sends flush
+            // when relays connect), then kick off connection setup, which itself
+            // waits for Tor readiness.
+            let targetRelays = allowedRelayList(from: relayUrls ?? Self.defaultRelays)
+            guard !targetRelays.isEmpty else { return }
+            enqueuePendingSend(event, pendingRelays: Set(targetRelays))
+            ensureConnections(to: targetRelays)
            return
        }
        let requestedRelays = relayUrls ?? Self.defaultRelays
@@ -307,12 +322,20 @@ final class NostrRelayManager: ObservableObject {
            }
        }
        if !stillPending.isEmpty {
-            messageQueueLock.lock()
-            messageQueue.append(PendingSend(event: event, pendingRelays: stillPending))
-            messageQueueLock.unlock()
+            enqueuePendingSend(event, pendingRelays: stillPending)
        }
    }

+    private func enqueuePendingSend(_ event: NostrEvent, pendingRelays: Set<String>) {
+        messageQueueLock.lock()
+        messageQueue.append(PendingSend(event: event, pendingRelays: pendingRelays))
+        let overflow = messageQueue.count - TransportConfig.nostrPendingSendQueueCap
+        if overflow > 0 {
+            messageQueue.removeFirst(overflow)
+        }
+        messageQueueLock.unlock()
+    }
+
    /// Try to flush any queued messages for relays that are now connected.
    private func flushMessageQueue(for relayUrl: String? = nil) {
        messageQueueLock.lock()
@@ -486,6 +509,8 @@ final class NostrRelayManager: ObservableObject {
    /// Unsubscribe from a subscription
    func unsubscribe(id: String) {
        messageHandlers.removeValue(forKey: id)
+        removeRecentInboundEvents(forSubscriptionID: id)
+        duplicateInboundEventDropCountBySubscription.removeValue(forKey: id)
        // Allow immediate re-subscription by clearing coalescer timestamp
        subscribeCoalesce.removeValue(forKey: id)
        subscriptionRequestState.removeValue(forKey: id)
@@ -561,14 +586,40 @@ final class NostrRelayManager: ObservableObject {
            self.awaitingTorForConnections = false

            guard ready else {
-                SecureLogger.error("❌ Tor not ready; aborting relay connections (fail-closed)", category: .session)
+                self.torReadyWaitAttempts += 1
+                if self.torReadyWaitAttempts < TransportConfig.nostrTorReadyMaxWaitAttempts {
+                    SecureLogger.warning("Tor not ready; re-queueing \(pending.count) relay connection(s) (attempt \(self.torReadyWaitAttempts))", category: .session)
+                    self.queueConnectionsUntilTorReady(pending)
+                } else {
+                    // Still fail-closed (no network), but unblock any callers
+                    // waiting on EOSE so the UI doesn't hang indefinitely.
+                    // Queued subscriptions/sends are kept and flush if a later
+                    // trigger (e.g. app foreground) brings Tor up.
+                    SecureLogger.error("❌ Tor not ready after \(self.torReadyWaitAttempts) wait(s); aborting relay connections (fail-closed)", category: .session)
+                    self.torReadyWaitAttempts = 0
+                    self.unblockPendingEOSECallbacks(reason: "tor-unavailable")
+                }
                return
            }

+            self.torReadyWaitAttempts = 0
            self.connectToRelays(pending, shouldLog: true)
        }
    }

+    /// Fire and clear all EOSE callbacks that are parked waiting for Tor.
+    /// Callers treat EOSE as "initial fetch finished"; firing with no data is
+    /// safe and prevents indefinite hangs when Tor cannot bootstrap.
+    private func unblockPendingEOSECallbacks(reason: String) {
+        guard !pendingEOSECallbacks.isEmpty else { return }
+        let callbacks = pendingEOSECallbacks
+        pendingEOSECallbacks.removeAll()
+        SecureLogger.warning("Unblocking \(callbacks.count) pending EOSE callback(s) without data (\(reason))", category: .session)
+        for (_, callback) in callbacks {
+            callback()
+        }
+    }
+
    private func subscriptionStateExists(id: String, requestState: SubscriptionRequestState) -> Bool {
        guard !requestState.relayURLs.isEmpty else { return true }
        return requestState.relayURLs.allSatisfy { url in
@@ -608,6 +659,53 @@ final class NostrRelayManager: ObservableObject {
            startEOSETracking(id: id, relayURLs: requestState.relayURLs, callback: callback)
        }
    }
+
+    private func shouldDeliverInboundEvent(subscriptionID: String, eventID: String) -> Bool {
+        guard !eventID.isEmpty else { return true }
+        let key = InboundEventKey(subscriptionID: subscriptionID, eventID: eventID)
+        guard recentInboundEventKeys.insert(key).inserted else {
+            recordDuplicateInboundEventDrop(subscriptionID: subscriptionID)
+            return false
+        }
+        recentInboundEventKeyOrder.append(key)
+
+        if recentInboundEventKeyOrder.count > recentInboundEventKeyLimit {
+            let removeCount = recentInboundEventKeyOrder.count - recentInboundEventKeyTrimTarget
+            for staleKey in recentInboundEventKeyOrder.prefix(removeCount) {
+                recentInboundEventKeys.remove(staleKey)
+            }
+            recentInboundEventKeyOrder.removeFirst(removeCount)
+        }
+        return true
+    }
+
+    private func recordDuplicateInboundEventDrop(subscriptionID: String) {
+        duplicateInboundEventDropCount += 1
+        let subscriptionCount = (duplicateInboundEventDropCountBySubscription[subscriptionID] ?? 0) + 1
+        duplicateInboundEventDropCountBySubscription[subscriptionID] = subscriptionCount
+
+        if duplicateInboundEventDropCount == 1 ||
+            duplicateInboundEventDropCount.isMultiple(of: TransportConfig.nostrDuplicateEventLogInterval) {
+            SecureLogger.debug(
+                "Dropped duplicate Nostr event deliveries total=\(duplicateInboundEventDropCount) sub=\(subscriptionID) sub_total=\(subscriptionCount)",
+                category: .session
+            )
+        }
+    }
+
+    private func removeRecentInboundEvents(forSubscriptionID subscriptionID: String) {
+        guard !recentInboundEventKeyOrder.isEmpty else { return }
+        var retainedKeys: [InboundEventKey] = []
+        retainedKeys.reserveCapacity(recentInboundEventKeyOrder.count)
+        for key in recentInboundEventKeyOrder {
+            if key.subscriptionID == subscriptionID {
+                recentInboundEventKeys.remove(key)
+            } else {
+                retainedKeys.append(key)
+            }
+        }
+        recentInboundEventKeyOrder = retainedKeys
+    }
    
    private func connectToRelay(_ urlString: String) {
        // Global network policy gate
@@ -722,12 +820,22 @@ final class NostrRelayManager: ObservableObject {
    private func handleParsedMessage(_ parsed: ParsedInbound, from relayUrl: String) {
        switch parsed {
        case .event(let subId, let event):
-            if event.kind != 1059 {
-                SecureLogger.debug("📥 Event kind=\(event.kind) id=\(event.id.prefix(16))… relay=\(relayUrl)", category: .session)
-            }
            if let index = self.relays.firstIndex(where: { $0.url == relayUrl }) {
                self.relays[index].messagesReceived += 1
            }
+            guard event.isValidSignature() else {
+                SecureLogger.warning(
+                    "⚠️ Dropped invalid Nostr event id=\(event.id.prefix(16))… sub=\(subId) relay=\(relayUrl)",
+                    category: .session
+                )
+                return
+            }
+            guard shouldDeliverInboundEvent(subscriptionID: subId, eventID: event.id) else {
+                return
+            }
+            if event.kind != 1059 {
+                SecureLogger.debug("📥 Event kind=\(event.kind) id=\(event.id.prefix(16))… relay=\(relayUrl)", category: .session)
+            }
            if let handler = self.messageHandlers[subId] {
                handler(event)
            } else {
@@ -919,6 +1027,14 @@ final class NostrRelayManager: ObservableObject {
        pendingSubscriptions[relayUrl]?.count ?? 0
    }

+    var debugDuplicateInboundEventDropCount: Int {
+        duplicateInboundEventDropCount
+    }
+
+    func debugDuplicateInboundEventDropCount(forSubscriptionID subscriptionID: String) -> Int {
+        duplicateInboundEventDropCountBySubscription[subscriptionID] ?? 0
+    }
+
    func debugFlushMessageQueue() {
        flushMessageQueue(for: nil)
    }
@@ -974,8 +1090,7 @@ private enum ParsedInbound {
            if array.count >= 3,
               let subId = array[1] as? String,
               let eventDict = array[2] as? [String: Any],
-               let event = try? NostrEvent(from: eventDict),
-               event.isValidSignature() {
+               let event = try? NostrEvent(from: eventDict) {
                self = .event(subId: subId, event: event)
                return
            }
@@ -26,36 +26,69 @@ struct BLESubscribedCentralSnapshot {
    }
 }

+/// Owns all BLE link state (peripheral connections we hold as central, and
+/// central subscriptions we serve as peripheral). The store has no internal
+/// locking: every access must happen on the single owning queue (the BLE
+/// queue). Other queues must go through BLEService's `readLinkState`, which
+/// hops to that queue. Call `assumeOwnership(of:)` to have debug builds trap
+/// any access from the wrong queue.
 final class BLELinkStateStore {
    private(set) var peripherals: [String: BLEPeripheralLinkState] = [:]
    private(set) var peerToPeripheralUUID: [PeerID: String] = [:]
    private(set) var subscribedCentrals: [CBCentral] = []
    private(set) var centralToPeerID: [String: PeerID] = [:]

+    #if DEBUG
+    private var ownerQueue: DispatchQueue?
+    #endif
+
+    /// Pin the store to its owning queue. Debug-only enforcement; release
+    /// builds are unchanged.
+    func assumeOwnership(of queue: DispatchQueue) {
+        #if DEBUG
+        ownerQueue = queue
+        #endif
+    }
+
+    @inline(__always)
+    private func assertOwned() {
+        #if DEBUG
+        if let queue = ownerQueue {
+            dispatchPrecondition(condition: .onQueue(queue))
+        }
+        #endif
+    }
+
    var peripheralStates: [BLEPeripheralLinkState] {
-        Array(peripherals.values)
+        assertOwned()
+        return Array(peripherals.values)
    }

    var subscribedCentralSnapshot: BLESubscribedCentralSnapshot {
-        BLESubscribedCentralSnapshot(
+        assertOwned()
+        return BLESubscribedCentralSnapshot(
            centrals: subscribedCentrals,
            peerIDsByCentralUUID: centralToPeerID
        )
    }

    var subscribedCentralCount: Int {
-        subscribedCentrals.count
+        assertOwned()
+        return subscribedCentrals.count
    }

    var connectedOrConnectingPeripheralCount: Int {
-        peripherals.values.filter { $0.isConnected || $0.isConnecting }.count
+        assertOwned()
+        return peripherals.values.filter { $0.isConnected || $0.isConnecting }.count
    }

    func state(forPeripheralID peripheralID: String) -> BLEPeripheralLinkState? {
-        peripherals[peripheralID]
+        assertOwned()
+        return peripherals[peripheralID]
    }

    func setPeripheralState(_ state: BLEPeripheralLinkState, for peripheralID: String) {
+        assertOwned()
        peripherals[peripheralID] = state
    }

@@ -64,6 +97,7 @@ final class BLELinkStateStore {
        _ peripheralID: String,
        _ update: (inout BLEPeripheralLinkState) -> Void
    ) -> BLEPeripheralLinkState? {
+        assertOwned()
        guard var state = peripherals[peripheralID] else { return nil }
        update(&state)
        peripherals[peripheralID] = state
@@ -113,10 +147,12 @@ final class BLELinkStateStore {
    }

    func directPeripheralState(for peerID: PeerID) -> BLEPeripheralLinkState? {
-        peerToPeripheralUUID[peerID].flatMap { peripherals[$0] }
+        assertOwned()
+        return peerToPeripheralUUID[peerID].flatMap { peripherals[$0] }
    }

    func directLinkState(for peerID: PeerID) -> BLEDirectLinkState {
+        assertOwned()
        let peripheralUUID = peerToPeripheralUUID[peerID]
        let hasPeripheral = peripheralUUID.flatMap { peripherals[$0]?.isConnected } ?? false
        let hasCentral = centralToPeerID.values.contains(peerID)
@@ -124,6 +160,7 @@ final class BLELinkStateStore {
    }

    func links(to peerID: PeerID?) -> Set<BLEIngressLinkID> {
+        assertOwned()
        guard let peerID else { return [] }

        var links: Set<BLEIngressLinkID> = []
@@ -137,35 +174,42 @@ final class BLELinkStateStore {
    }

    func peerID(forPeripheralID peripheralID: String) -> PeerID? {
-        peripherals[peripheralID]?.peerID
+        assertOwned()
+        return peripherals[peripheralID]?.peerID
    }

    func peerID(forCentralUUID centralUUID: String) -> PeerID? {
-        centralToPeerID[centralUUID]
+        assertOwned()
+        return centralToPeerID[centralUUID]
    }

    func addSubscribedCentral(_ central: CBCentral) {
+        assertOwned()
        guard !subscribedCentrals.contains(central) else { return }
        subscribedCentrals.append(central)
    }

    func removeSubscribedCentral(_ central: CBCentral) -> PeerID? {
+        assertOwned()
        let centralUUID = central.identifier.uuidString
        subscribedCentrals.removeAll { $0.identifier == central.identifier }
        return centralToPeerID.removeValue(forKey: centralUUID)
    }

    func bindCentral(_ centralUUID: String, to peerID: PeerID) {
+        assertOwned()
        centralToPeerID[centralUUID] = peerID
    }

    func bindPeripheral(_ peripheralUUID: String, to peerID: PeerID) {
+        assertOwned()
        if updatePeripheral(peripheralUUID, { $0.peerID = peerID }) != nil {
            peerToPeripheralUUID[peerID] = peripheralUUID
        }
    }

    func removePeripheral(_ peripheralID: String) -> PeerID? {
+        assertOwned()
        let peerID = peripherals.removeValue(forKey: peripheralID)?.peerID
        if let peerID {
            peerToPeripheralUUID.removeValue(forKey: peerID)
@@ -174,6 +218,7 @@ final class BLELinkStateStore {
    }

    func clearPeripherals() -> [PeerID] {
+        assertOwned()
        let peerIDs = peripherals.compactMap { $0.value.peerID }
        peripherals.removeAll()
        peerToPeripheralUUID.removeAll()
@@ -181,6 +226,7 @@ final class BLELinkStateStore {
    }

    func clearCentrals() -> [PeerID] {
+        assertOwned()
        let peerIDs = Array(centralToPeerID.values)
        subscribedCentrals.removeAll()
        centralToPeerID.removeAll()
@@ -188,6 +234,7 @@ final class BLELinkStateStore {
    }

    func clearAll() {
+        assertOwned()
        peripherals.removeAll()
        peerToPeripheralUUID.removeAll()
        subscribedCentrals.removeAll()
@@ -13,6 +13,7 @@ final class MessageRouter {
        let nickname: String
        let messageID: String
        let timestamp: Date
+        var sendAttempts: Int = 0
    }

    private var outbox: [PeerID: [QueuedMessage]] = [:]
@@ -20,6 +21,9 @@ final class MessageRouter {
    // Outbox limits to prevent unbounded memory growth
    private static let maxMessagesPerPeer = 100
    private static let messageTTLSeconds: TimeInterval = 24 * 60 * 60 // 24 hours
+    // Bound resends of messages sent on a weak reachability signal that never
+    // get a delivery ack (e.g. peer on an old client that doesn't ack).
+    private static let maxSendAttempts = 8

    init(transports: [Transport]) {
        self.transports = transports
@@ -61,26 +65,53 @@ final class MessageRouter {
    // MARK: - Message Sending

    func sendPrivate(_ content: String, to peerID: PeerID, recipientNickname: String, messageID: String) {
-        if let transport = reachableTransport(for: peerID) {
-            SecureLogger.debug("Routing PM via \(type(of: transport)) to \(peerID.id.prefix(8))… id=\(messageID.prefix(8))…", category: .session)
+        if let transport = connectedTransport(for: peerID) {
+            // A live link is a strong delivery signal; trust it outright.
+            SecureLogger.debug("Routing PM via \(type(of: transport)) (connected) to \(peerID.id.prefix(8))… id=\(messageID.prefix(8))…", category: .session)
            transport.sendPrivateMessage(content, to: peerID, recipientNickname: recipientNickname, messageID: messageID)
+            return
+        }
+
+        let message = QueuedMessage(content: content, nickname: recipientNickname, messageID: messageID, timestamp: Date(), sendAttempts: 1)
+        if let transport = reachableTransport(for: peerID) {
+            // Reachability without a connection is a freshness heuristic (e.g.
+            // the mesh retention window), so the send can silently go nowhere.
+            // Send now, but retain a copy until a delivery/read ack clears it;
+            // receivers dedup resends by message ID.
+            SecureLogger.debug("Routing PM via \(type(of: transport)) (reachable) to \(peerID.id.prefix(8))… id=\(messageID.prefix(8))…", category: .session)
+            transport.sendPrivateMessage(content, to: peerID, recipientNickname: recipientNickname, messageID: messageID)
+            enqueue(message, for: peerID)
        } else {
-            // Queue for later with timestamp for TTL tracking
-            if outbox[peerID] == nil { outbox[peerID] = [] }
-
-            let message = QueuedMessage(content: content, nickname: recipientNickname, messageID: messageID, timestamp: Date())
-            outbox[peerID]?.append(message)
-
-            // Enforce per-peer size limit with FIFO eviction
-            if let count = outbox[peerID]?.count, count > Self.maxMessagesPerPeer {
-                let evicted = outbox[peerID]?.removeFirst()
-                SecureLogger.warning("📤 Outbox overflow for \(peerID.id.prefix(8))… - evicted oldest message: \(evicted?.messageID.prefix(8) ?? "?")…", category: .session)
-            }
-
+            var unsent = message
+            unsent.sendAttempts = 0
+            enqueue(unsent, for: peerID)
            SecureLogger.debug("Queued PM for \(peerID.id.prefix(8))… (no reachable transport) id=\(messageID.prefix(8))… queue=\(outbox[peerID]?.count ?? 0)", category: .session)
        }
    }

+    /// A delivery or read ack confirms receipt; stop retaining the message.
+    func markDelivered(_ messageID: String) {
+        for (peerID, queue) in outbox {
+            let filtered = queue.filter { $0.messageID != messageID }
+            guard filtered.count != queue.count else { continue }
+            outbox[peerID] = filtered.isEmpty ? nil : filtered
+        }
+    }
+
+    private func enqueue(_ message: QueuedMessage, for peerID: PeerID) {
+        var queue = outbox[peerID] ?? []
+        // Re-sending an already-queued ID replaces the entry (keeps attempt count fresh)
+        queue.removeAll { $0.messageID == message.messageID }
+        queue.append(message)
+
+        // Enforce per-peer size limit with FIFO eviction
+        if queue.count > Self.maxMessagesPerPeer {
+            let evicted = queue.removeFirst()
+            SecureLogger.warning("📤 Outbox overflow for \(peerID.id.prefix(8))… - evicted oldest message: \(evicted.messageID.prefix(8))…", category: .session)
+        }
+        outbox[peerID] = queue
+    }
+
    func sendReadReceipt(_ receipt: ReadReceipt, to peerID: PeerID) {
        if let transport = reachableTransport(for: peerID) {
            SecureLogger.debug("Routing READ ack via \(type(of: transport)) to \(peerID.id.prefix(8))… id=\(receipt.originalMessageID.prefix(8))…", category: .session)
@@ -121,9 +152,22 @@ final class MessageRouter {
                continue
            }

-            if let transport = reachableTransport(for: peerID) {
-                SecureLogger.debug("Outbox -> \(type(of: transport)) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
+            if let transport = connectedTransport(for: peerID) {
+                // Live link: send and stop retaining.
+                SecureLogger.debug("Outbox -> \(type(of: transport)) (connected) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
                transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
+            } else if let transport = reachableTransport(for: peerID) {
+                // Weak signal: send but keep retaining until an ack clears it,
+                // bounded by attempt count for peers that never ack.
+                guard message.sendAttempts < Self.maxSendAttempts else {
+                    SecureLogger.warning("📤 Dropping unacked PM for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… after \(message.sendAttempts) attempts", category: .session)
+                    continue
+                }
+                SecureLogger.debug("Outbox -> \(type(of: transport)) (reachable) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
+                transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
+                var retained = message
+                retained.sendAttempts += 1
+                remaining.append(retained)
            } else {
                remaining.append(message)
            }
@@ -18,6 +18,7 @@ enum TransportConfig {
    static let meshTimelineCap: Int = 1337
    static let geoTimelineCap: Int = 1337
    static let contentLRUCap: Int = 2000
+    static let geoSamplingEventLRUCap: Int = 2000

    // Timers
    static let networkResetGraceSeconds: TimeInterval = 600 // 10 minutes
@@ -43,6 +44,9 @@ enum TransportConfig {

    // Nostr
    static let nostrReadAckInterval: TimeInterval = 0.35 // ~3 per second
+    static let nostrInboundEventDedupCap: Int = 4096
+    static let nostrInboundEventDedupTrimTarget: Int = 3072
+    static let nostrDuplicateEventLogInterval: Int = 50

    // UI thresholds
    static let uiLateInsertThreshold: TimeInterval = 15.0
@@ -146,6 +150,10 @@ enum TransportConfig {
    static let nostrRelayBackoffMultiplier: Double = 2.0
    static let nostrRelayMaxReconnectAttempts: Int = 10
    static let nostrRelayDefaultFetchLimit: Int = 100
+    // How many consecutive Tor-readiness waits (each bounded by TorManager's
+    // bootstrap deadline) to attempt before unblocking pending EOSE callers.
+    static let nostrTorReadyMaxWaitAttempts: Int = 3
+    static let nostrPendingSendQueueCap: Int = 200

    // Geo relay directory
    static let geoRelayFetchIntervalSeconds: TimeInterval = 60 * 60 * 24
@@ -12,6 +12,11 @@ import CryptoKit
 enum GCSFilter {
    struct Params { let p: Int; let m: UInt32; let data: Data }

+    // Highest Golomb-Rice parameter we accept from the wire. P maps to an FPR
+    // of ~1/2^P; beyond 32 the remainder width exceeds any practical filter
+    // and shifts in decode would silently overflow to garbage values.
+    static let maxP = 32
+
    // Derive P from FPR (~ 1 / 2^P)
    static func deriveP(targetFpr: Double) -> Int {
        let f = max(0.000001, min(0.25, targetFpr))
@@ -66,6 +71,9 @@ enum GCSFilter {
    }

    static func decodeToSortedSet(p: Int, m: UInt32, data: Data) -> [UInt64] {
+        // Reject out-of-range parameters rather than decoding garbage: callers
+        // treat the result as "peer has nothing" and fall back to sending data.
+        guard p >= 1, p <= maxP, m > 1 else { return [] }
        var values: [UInt64] = []
        let reader = BitReader(data)
        var acc: UInt64 = 0
@@ -20,4 +20,32 @@ struct GCSFilterTests {
        #expect(bucket != 0)
        #expect(bucket < 2)
    }
+
+    @Test func decodeRejectsOutOfRangeParameters() {
+        let junk = Data(repeating: 0xFF, count: 64)
+        #expect(GCSFilter.decodeToSortedSet(p: 0, m: 1000, data: junk).isEmpty)
+        #expect(GCSFilter.decodeToSortedSet(p: -1, m: 1000, data: junk).isEmpty)
+        #expect(GCSFilter.decodeToSortedSet(p: GCSFilter.maxP + 1, m: 1000, data: junk).isEmpty)
+        #expect(GCSFilter.decodeToSortedSet(p: 255, m: UInt32.max, data: junk).isEmpty)
+        #expect(GCSFilter.decodeToSortedSet(p: 8, m: 0, data: junk).isEmpty)
+        #expect(GCSFilter.decodeToSortedSet(p: 8, m: 1, data: junk).isEmpty)
+    }
+
+    @Test func decodeOfTruncatedDataReturnsOnlyCompleteValues() {
+        let ids = (0..<32).map { i in Data(repeating: UInt8(i), count: 16) }
+        let params = GCSFilter.buildFilter(ids: ids, maxBytes: 128, targetFpr: 0.01)
+        let full = GCSFilter.decodeToSortedSet(p: params.p, m: params.m, data: params.data)
+        let truncated = GCSFilter.decodeToSortedSet(p: params.p, m: params.m, data: params.data.prefix(params.data.count / 2))
+        #expect(truncated.count <= full.count)
+        // Truncation must not invent values that were not in the full set.
+        #expect(truncated.allSatisfy { full.contains($0) })
+    }
+
+    @Test func requestSyncPacketDecodeRejectsOversizedP() {
+        let valid = RequestSyncPacket(p: 8, m: 4096, data: Data([0x01, 0x02]))
+        #expect(RequestSyncPacket.decode(from: valid.encode()) != nil)
+
+        let oversized = RequestSyncPacket(p: 200, m: 4096, data: Data([0x01, 0x02]))
+        #expect(RequestSyncPacket.decode(from: oversized.encode()) == nil)
+    }
 }
@@ -42,6 +42,74 @@ struct MessageRouterTests {
        #expect(transport.sentPrivateMessages.count == 1)
    }

+    @Test @MainActor
+    func sendPrivate_prefersConnectedTransportOverEarlierReachableOne() async {
+        let peerID = PeerID(str: "0000000000000005")
+        let reachableOnly = MockTransport()
+        reachableOnly.reachablePeers.insert(peerID)
+        let connected = MockTransport()
+        connected.connectedPeers.insert(peerID)
+        connected.reachablePeers.insert(peerID)
+
+        let router = MessageRouter(transports: [reachableOnly, connected])
+        router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m5")
+
+        #expect(reachableOnly.sentPrivateMessages.isEmpty)
+        #expect(connected.sentPrivateMessages.count == 1)
+    }
+
+    @Test @MainActor
+    func sendPrivate_reachableOnlySendRetainsUntilDeliveryAck() async {
+        let peerID = PeerID(str: "0000000000000006")
+        let transport = MockTransport()
+        transport.reachablePeers.insert(peerID)
+
+        let router = MessageRouter(transports: [transport])
+        router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m6")
+        #expect(transport.sentPrivateMessages.count == 1)
+
+        // No ack yet: a flush retries over the weak signal.
+        router.flushOutbox(for: peerID)
+        #expect(transport.sentPrivateMessages.count == 2)
+
+        // Ack clears the retained copy; later flushes stop resending.
+        router.markDelivered("m6")
+        router.flushOutbox(for: peerID)
+        #expect(transport.sentPrivateMessages.count == 2)
+    }
+
+    @Test @MainActor
+    func sendPrivate_connectedSendIsNotRetained() async {
+        let peerID = PeerID(str: "0000000000000007")
+        let transport = MockTransport()
+        transport.connectedPeers.insert(peerID)
+        transport.reachablePeers.insert(peerID)
+
+        let router = MessageRouter(transports: [transport])
+        router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m7")
+        #expect(transport.sentPrivateMessages.count == 1)
+
+        router.flushOutbox(for: peerID)
+        #expect(transport.sentPrivateMessages.count == 1)
+    }
+
+    @Test @MainActor
+    func flushOutbox_dropsUnackedMessageAfterAttemptCap() async {
+        let peerID = PeerID(str: "0000000000000008")
+        let transport = MockTransport()
+        transport.reachablePeers.insert(peerID)
+
+        let router = MessageRouter(transports: [transport])
+        router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m8")
+
+        for _ in 0..<10 {
+            router.flushOutbox(for: peerID)
+        }
+
+        // Initial send plus capped resends; never unbounded.
+        #expect(transport.sentPrivateMessages.count == 8)
+    }
+
    @Test @MainActor
    func sendReadReceipt_usesReachableTransport() async {
        let peerID = PeerID(str: "0000000000000003")
@@ -90,6 +90,90 @@ final class NostrRelayManagerTests: XCTestCase {
        XCTAssertFalse(context.manager.isConnected)
    }

+    func test_connect_retriesTorWaitAndConnectsWhenTorBecomesReady() async {
+        let context = makeContext(permission: .authorized, userTorEnabled: true, torEnforced: true, torIsReady: false)
+
+        context.manager.connect()
+        XCTAssertEqual(context.torWaiter.awaitCallCount, 1)
+
+        context.torWaiter.resolve(false)
+
+        // A failed wait re-queues the same targets and waits again instead of dropping them.
+        XCTAssertEqual(context.torWaiter.awaitCallCount, 2)
+        XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
+
+        context.torWaiter.resolve(true)
+
+        let connected = await waitUntil {
+            context.sessionFactory.requestedURLs.count == self.expectedDefaultRelayCount &&
+            context.manager.relays.allSatisfy(\.isConnected)
+        }
+        XCTAssertTrue(connected)
+    }
+
+    func test_subscribe_unblocksDeferredEOSEWhenTorWaitAttemptsExhausted() async {
+        let relayURL = "wss://tor-eose-unblock.example"
+        let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
+        var eoseCount = 0
+
+        context.manager.subscribe(
+            filter: makeFilter(),
+            id: "tor-sub-unblock",
+            relayUrls: [relayURL],
+            handler: { _ in },
+            onEOSE: { eoseCount += 1 }
+        )
+
+        for _ in 0..<TransportConfig.nostrTorReadyMaxWaitAttempts {
+            context.torWaiter.resolve(false)
+        }
+
+        // Fail-closed (no sessions), but the EOSE caller is unblocked.
+        XCTAssertEqual(eoseCount, 1)
+        XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
+    }
+
+    func test_sendEvent_survivesFailedTorWaitAndSendsWhenTorRecovers() async throws {
+        let relayURL = "wss://tor-send-retry.example"
+        let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
+        let event = try makeSignedEvent(content: "queued through tor stall")
+
+        context.manager.sendEvent(event, to: [relayURL])
+
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 1)
+        XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
+
+        context.torWaiter.resolve(false)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 1)
+
+        context.torWaiter.resolve(true)
+
+        let sent = await waitUntil {
+            context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1 &&
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(sent)
+    }
+
+    func test_sendEvent_pendingQueueDropsOldestBeyondCap() async throws {
+        let relayURL = "wss://tor-send-cap.example"
+        let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
+        let identity = try NostrIdentity.generate()
+
+        for i in 0...(TransportConfig.nostrPendingSendQueueCap + 4) {
+            let event = NostrEvent(
+                pubkey: identity.publicKeyHex,
+                createdAt: Date(),
+                kind: .textNote,
+                tags: [],
+                content: "cap-\(i)"
+            )
+            context.manager.sendEvent(try event.sign(with: identity.schnorrSigningKey()), to: [relayURL])
+        }
+
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, TransportConfig.nostrPendingSendQueueCap)
+    }
+
    func test_sendEvent_waitsForTorReadinessBeforeSending() async throws {
        let relayURL = "wss://tor-ready.example"
        let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
@@ -439,6 +523,114 @@ final class NostrRelayManagerTests: XCTestCase {
        XCTAssertEqual(receivedEvent?.id, event.id)
    }

+    func test_receiveEvent_deduplicatesSameSubscriptionEventAcrossRelays() async throws {
+        let firstRelayURL = "wss://events-one.example"
+        let secondRelayURL = "wss://events-two.example"
+        let context = makeContext(permission: .denied)
+        let event = try makeSignedEvent(content: "duplicate")
+        var receivedIDs: [String] = []
+
+        context.manager.subscribe(
+            filter: makeFilter(),
+            id: "events",
+            relayUrls: [firstRelayURL, secondRelayURL]
+        ) { event in
+            receivedIDs.append(event.id)
+        }
+        let subscriptionsSent = await waitUntil {
+            context.sessionFactory.latestConnection(for: firstRelayURL)?.sentStrings.count == 1 &&
+            context.sessionFactory.latestConnection(for: secondRelayURL)?.sentStrings.count == 1
+        }
+        XCTAssertTrue(subscriptionsSent)
+
+        try context.sessionFactory.latestConnection(for: firstRelayURL)?.emitEventMessage(subscriptionID: "events", event: event)
+        try context.sessionFactory.latestConnection(for: secondRelayURL)?.emitEventMessage(subscriptionID: "events", event: event)
+
+        let countedOnBothRelays = await waitUntil {
+            context.manager.relays.first(where: { $0.url == firstRelayURL })?.messagesReceived == 1 &&
+            context.manager.relays.first(where: { $0.url == secondRelayURL })?.messagesReceived == 1
+        }
+        XCTAssertTrue(countedOnBothRelays)
+        XCTAssertEqual(receivedIDs, [event.id])
+        XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, 1)
+        XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "events"), 1)
+    }
+
+    func test_receiveEvent_duplicateFanInDeliversOnceAndCountsDrops() async throws {
+        let relayURLs = (0..<8).map { "wss://fan-in-\($0).example" }
+        let context = makeContext(permission: .denied)
+        let event = try makeSignedEvent(content: "fan-in")
+        var receivedIDs: [String] = []
+
+        context.manager.subscribe(
+            filter: makeFilter(),
+            id: "presence",
+            relayUrls: relayURLs
+        ) { event in
+            receivedIDs.append(event.id)
+        }
+        let subscriptionsSent = await waitUntil {
+            relayURLs.allSatisfy { relayURL in
+                context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
+            }
+        }
+        XCTAssertTrue(subscriptionsSent)
+
+        for relayURL in relayURLs {
+            try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(
+                subscriptionID: "presence",
+                event: event
+            )
+        }
+
+        let countedOnEveryRelay = await waitUntil {
+            relayURLs.allSatisfy { relayURL in
+                context.manager.relays.first(where: { $0.url == relayURL })?.messagesReceived == 1
+            }
+        }
+        XCTAssertTrue(countedOnEveryRelay)
+        XCTAssertEqual(receivedIDs, [event.id])
+        XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, relayURLs.count - 1)
+        XCTAssertEqual(
+            context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "presence"),
+            relayURLs.count - 1
+        )
+    }
+
+    func test_receiveEvent_invalidSignatureDoesNotPoisonDuplicateCache() async throws {
+        let firstRelayURL = "wss://invalid-first-one.example"
+        let secondRelayURL = "wss://invalid-first-two.example"
+        let context = makeContext(permission: .denied)
+        let event = try makeSignedEvent(content: "valid-after-invalid")
+        let invalidEvent = invalidSignatureCopy(of: event)
+        var receivedIDs: [String] = []
+
+        context.manager.subscribe(
+            filter: makeFilter(),
+            id: "events",
+            relayUrls: [firstRelayURL, secondRelayURL]
+        ) { event in
+            receivedIDs.append(event.id)
+        }
+        let subscriptionsSent = await waitUntil {
+            context.sessionFactory.latestConnection(for: firstRelayURL)?.sentStrings.count == 1 &&
+            context.sessionFactory.latestConnection(for: secondRelayURL)?.sentStrings.count == 1
+        }
+        XCTAssertTrue(subscriptionsSent)
+
+        try context.sessionFactory.latestConnection(for: firstRelayURL)?.emitEventMessage(subscriptionID: "events", event: invalidEvent)
+        try context.sessionFactory.latestConnection(for: secondRelayURL)?.emitEventMessage(subscriptionID: "events", event: event)
+
+        let countedOnBothRelays = await waitUntil {
+            context.manager.relays.first(where: { $0.url == firstRelayURL })?.messagesReceived == 1 &&
+            context.manager.relays.first(where: { $0.url == secondRelayURL })?.messagesReceived == 1
+        }
+        XCTAssertTrue(countedOnBothRelays)
+        XCTAssertEqual(receivedIDs, [event.id])
+        XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, 0)
+        XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "events"), 0)
+    }
+
    func test_receiveEvent_withoutHandlerStillTracksReceivedCount() async throws {
        let relayURL = "wss://missing-handler.example"
        let context = makeContext(permission: .denied)
@@ -880,6 +1072,12 @@ final class NostrRelayManagerTests: XCTestCase {
        return try event.sign(with: identity.schnorrSigningKey())
    }

+    private func invalidSignatureCopy(of event: NostrEvent) -> NostrEvent {
+        var invalid = event
+        invalid.sig = String(repeating: "0", count: 128)
+        return invalid
+    }
+
    private func waitUntil(
        timeout: TimeInterval = 1.0,
        condition: @escaping @MainActor () -> Bool