From ed86ed1065e2d0eea4e8ca15a8eb26e184f472a3 Mon Sep 17 00:00:00 2001 From: jack Date: Thu, 11 Jun 2026 15:05:44 +0200 Subject: [PATCH] Surface router drops as failed status; bound relay pending subscriptions; jitter backoff MessageRouter outbox drops (attempt cap, TTL expiry in flush and cleanup, per-peer overflow eviction) now invoke onMessageDropped, wired to mark the message .failed in the ConversationStore - guarded so a late failure never downgrades an already delivered/read status. NostrRelayManager pending subscriptions gain a per-relay cap (64, oldest-by-sequence eviction; durable intent still replays from subscriptionRequestState) and a 10-minute age sweep on the existing connect path. Reconnect backoff gets injectable +/-20% jitter so recovering relays don't thundering-herd. Co-Authored-By: Claude Fable 5 --- bitchat/Nostr/NostrRelayManager.swift | 83 +++++++++++-- bitchat/Services/MessageRouter.swift | 30 ++++- bitchat/Services/TransportConfig.swift | 8 ++ .../ChatViewModelBootstrapper.swift | 17 +++ .../ChatViewModelDeliveryStatusTests.swift | 74 ++++++++++++ .../Services/MessageRouterTests.swift | 98 +++++++++++++++ .../Services/NostrRelayManagerTests.swift | 112 +++++++++++++++++- 7 files changed, 405 insertions(+), 17 deletions(-) diff --git a/bitchat/Nostr/NostrRelayManager.swift b/bitchat/Nostr/NostrRelayManager.swift index 5b8842a3..b2eaa53f 100644 --- a/bitchat/Nostr/NostrRelayManager.swift +++ b/bitchat/Nostr/NostrRelayManager.swift @@ -66,6 +66,9 @@ struct NostrRelayManagerDependencies { var makeSession: () -> NostrRelaySessionProtocol var scheduleAfter: @Sendable (TimeInterval, @escaping @Sendable () -> Void) -> Void var now: () -> Date + /// Uniform random value in [0, 1) used to jitter reconnect backoff. + /// Injectable so tests can pin or sweep the jitter deterministically. + var jitterUnit: () -> Double } private extension NostrRelayManagerDependencies { @@ -93,7 +96,8 @@ private extension NostrRelayManagerDependencies { scheduleAfter: { delay, action in DispatchQueue.main.asyncAfter(deadline: .now() + delay, execute: action) }, - now: Date.init + now: Date.init, + jitterUnit: { Double.random(in: 0..<1) } ) } } @@ -140,7 +144,16 @@ final class NostrRelayManager: ObservableObject { private var hasLocationPermission: Bool = false private var connections: [String: NostrRelayConnectionProtocol] = [:] private var subscriptions: [String: Set] = [:] // relay URL -> active subscription IDs - private var pendingSubscriptions: [String: [String: String]] = [:] // relay URL -> (subscription id -> encoded REQ JSON) + // Not-yet-flushed REQs per relay, bounded by a per-relay cap (oldest by + // insertion order evicted) and an age sweep on connect attempts. Dicts are + // unordered, so each entry carries an insertion sequence and queue time. + private struct PendingSubscription { + let messageString: String // encoded REQ JSON + let queuedAt: Date + let sequence: UInt64 + } + private var pendingSubscriptions: [String: [String: PendingSubscription]] = [:] // relay URL -> (subscription id -> pending REQ) + private var pendingSubscriptionSequence: UInt64 = 0 private var messageHandlers: [String: (NostrEvent) -> Void] = [:] private struct InboundEventKey: Hashable { let subscriptionID: String @@ -432,9 +445,7 @@ final class NostrRelayManager: ObservableObject { existingSet.insert(url) } for url in urls { - var map = self.pendingSubscriptions[url] ?? [:] - map[id] = messageString - self.pendingSubscriptions[url] = map + queuePendingSubscription(id: id, messageString: messageString, for: url) } // Initialize EOSE tracking if requested if let onEOSE = onEOSE { @@ -550,6 +561,7 @@ final class NostrRelayManager: ObservableObject { private func connectToRelays(_ relayUrls: [String], shouldLog: Bool = false) { guard dependencies.activationAllowed() else { return } + sweepStalePendingSubscriptions() let targets = allowedRelayList(from: relayUrls).filter { connections[$0] == nil && !isPermanentlyFailed($0) } @@ -627,11 +639,53 @@ final class NostrRelayManager: ObservableObject { private func subscriptionStateExists(id: String, requestState: SubscriptionRequestState) -> Bool { guard !requestState.relayURLs.isEmpty else { return true } return requestState.relayURLs.allSatisfy { url in - pendingSubscriptions[url]?[id] == requestState.messageString || + pendingSubscriptions[url]?[id]?.messageString == requestState.messageString || subscriptions[url]?.contains(id) == true } } + private func queuePendingSubscription(id: String, messageString: String, for url: String) { + var map = pendingSubscriptions[url] ?? [:] + pendingSubscriptionSequence &+= 1 + map[id] = PendingSubscription( + messageString: messageString, + queuedAt: dependencies.now(), + sequence: pendingSubscriptionSequence + ) + // Bound per-relay pending REQs; evict oldest by insertion order. The + // durable intent stays in subscriptionRequestState, so an evicted REQ + // is still replayed if its subscription is active when the relay + // (re)connects. + while map.count > TransportConfig.nostrPendingSubscriptionsPerRelayCap, + let oldest = map.min(by: { $0.value.sequence < $1.value.sequence }) { + map.removeValue(forKey: oldest.key) + SecureLogger.warning( + "📋 Pending subscription overflow for \(url) - evicted oldest id=\(oldest.key)", + category: .session + ) + } + pendingSubscriptions[url] = map + } + + /// Drop pending REQs older than the TTL. Runs on connect attempts (the + /// natural maintenance path: connect/ensureConnections/reconnects all + /// funnel through connectToRelays) so stale entries for relays that never + /// come up cannot accumulate without bound. + private func sweepStalePendingSubscriptions() { + let now = dependencies.now() + for (url, map) in pendingSubscriptions { + let fresh = map.filter { + now.timeIntervalSince($0.value.queuedAt) <= TransportConfig.nostrPendingSubscriptionTTLSeconds + } + guard fresh.count != map.count else { continue } + SecureLogger.debug( + "📋 Swept \(map.count - fresh.count) stale pending subscription(s) for \(url)", + category: .session + ) + pendingSubscriptions[url] = fresh.isEmpty ? nil : fresh + } + } + private func startEOSETracking(id: String, relayURLs: Set, callback: @escaping () -> Void) { eoseTrackerEpoch += 1 let epoch = eoseTrackerEpoch @@ -770,7 +824,7 @@ final class NostrRelayManager: ObservableObject { /// active subscription targeting this relay must be re-sent. private func flushPendingSubscriptions(for relayUrl: String) { guard let connection = connections[relayUrl] else { return } - var toSend = pendingSubscriptions[relayUrl] ?? [:] + var toSend = (pendingSubscriptions[relayUrl] ?? [:]).mapValues(\.messageString) for (id, state) in subscriptionRequestState where state.relayURLs.contains(relayUrl) && toSend[id] == nil { toSend[id] = state.messageString } @@ -987,12 +1041,16 @@ final class NostrRelayManager: ObservableObject { return } - // Calculate backoff interval - let backoffInterval = min( + // Calculate backoff interval with ±jitterRatio random jitter so relays + // that dropped together don't all reconnect at the same instant. + let baseBackoffInterval = min( initialBackoffInterval * pow(backoffMultiplier, Double(relays[index].reconnectAttempts - 1)), maxBackoffInterval ) - + let jitterRatio = TransportConfig.nostrRelayBackoffJitterRatio + let jitterFactor = 1.0 + (dependencies.jitterUnit() * 2.0 - 1.0) * jitterRatio + let backoffInterval = baseBackoffInterval * jitterFactor + let nextReconnectTime = dependencies.now().addingTimeInterval(backoffInterval) relays[index].nextReconnectTime = nextReconnectTime @@ -1054,6 +1112,11 @@ final class NostrRelayManager: ObservableObject { pendingSubscriptions[relayUrl]?.count ?? 0 } + func debugPendingSubscriptionIDs(for relayUrl: String) -> Set { + guard let map = pendingSubscriptions[relayUrl] else { return [] } + return Set(map.keys) + } + var debugDuplicateInboundEventDropCount: Int { duplicateInboundEventDropCount } diff --git a/bitchat/Services/MessageRouter.swift b/bitchat/Services/MessageRouter.swift index 2f00955c..af0b3291 100644 --- a/bitchat/Services/MessageRouter.swift +++ b/bitchat/Services/MessageRouter.swift @@ -6,6 +6,13 @@ import Foundation @MainActor final class MessageRouter { private let transports: [Transport] + private let now: () -> Date + + /// Invoked whenever a retained private message is dropped without a + /// delivery ack (attempt cap, TTL expiry, or per-peer overflow eviction) + /// so the UI can surface the failure instead of leaving the message in a + /// stale "sending/sent" state forever. + var onMessageDropped: ((_ messageID: String, _ peerID: PeerID) -> Void)? // Outbox entry with timestamp for TTL-based eviction private struct QueuedMessage { @@ -25,8 +32,9 @@ final class MessageRouter { // get a delivery ack (e.g. peer on an old client that doesn't ack). private static let maxSendAttempts = 8 - init(transports: [Transport]) { + init(transports: [Transport], now: @escaping () -> Date = Date.init) { self.transports = transports + self.now = now // Observe favorites changes to learn Nostr mapping and flush queued messages NotificationCenter.default.addObserver( @@ -72,7 +80,7 @@ final class MessageRouter { return } - let message = QueuedMessage(content: content, nickname: recipientNickname, messageID: messageID, timestamp: Date(), sendAttempts: 1) + let message = QueuedMessage(content: content, nickname: recipientNickname, messageID: messageID, timestamp: now(), 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. @@ -108,6 +116,7 @@ final class MessageRouter { 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) + onMessageDropped?(evicted.messageID, peerID) } outbox[peerID] = queue } @@ -142,13 +151,14 @@ final class MessageRouter { guard let queued = outbox[peerID], !queued.isEmpty else { return } SecureLogger.debug("Flushing outbox for \(peerID.id.prefix(8))… count=\(queued.count)", category: .session) - let now = Date() + let now = now() var remaining: [QueuedMessage] = [] for message in queued { // Skip expired messages (TTL exceeded) if now.timeIntervalSince(message.timestamp) > Self.messageTTLSeconds { SecureLogger.debug("⏰ Expired queued message for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… (age: \(Int(now.timeIntervalSince(message.timestamp)))s)", category: .session) + onMessageDropped?(message.messageID, peerID) continue } @@ -161,6 +171,7 @@ final class MessageRouter { // 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) + onMessageDropped?(message.messageID, peerID) continue } SecureLogger.debug("Outbox -> \(type(of: transport)) (reachable) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session) @@ -186,12 +197,21 @@ final class MessageRouter { /// Periodically clean up expired messages from all outboxes func cleanupExpiredMessages() { - let now = Date() + let now = now() for peerID in Array(outbox.keys) { - outbox[peerID]?.removeAll { now.timeIntervalSince($0.timestamp) > Self.messageTTLSeconds } + var expiredMessageIDs: [String] = [] + outbox[peerID]?.removeAll { message in + guard now.timeIntervalSince(message.timestamp) > Self.messageTTLSeconds else { return false } + expiredMessageIDs.append(message.messageID) + return true + } if outbox[peerID]?.isEmpty == true { outbox.removeValue(forKey: peerID) } + for messageID in expiredMessageIDs { + SecureLogger.debug("⏰ Expired queued message for \(peerID.id.prefix(8))… id=\(messageID.prefix(8))…", category: .session) + onMessageDropped?(messageID, peerID) + } } } } diff --git a/bitchat/Services/TransportConfig.swift b/bitchat/Services/TransportConfig.swift index 7a6e069d..87255245 100644 --- a/bitchat/Services/TransportConfig.swift +++ b/bitchat/Services/TransportConfig.swift @@ -148,11 +148,19 @@ enum TransportConfig { static let nostrRelayMaxBackoffSeconds: TimeInterval = 300.0 static let nostrRelayBackoffMultiplier: Double = 2.0 static let nostrRelayMaxReconnectAttempts: Int = 10 + // Reconnect delays get ±20% random jitter so relays that dropped together + // (e.g. a network blip) don't thundering-herd the same reconnect instant. + static let nostrRelayBackoffJitterRatio: Double = 0.2 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 + // Pending (not-yet-flushed) REQs are bounded per relay: oldest-by-insertion + // eviction at the cap, plus an age sweep on connect attempts. Durable + // subscription intent survives in subscriptionRequestState either way. + static let nostrPendingSubscriptionsPerRelayCap: Int = 64 + static let nostrPendingSubscriptionTTLSeconds: TimeInterval = 600.0 // 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 diff --git a/bitchat/ViewModels/ChatViewModelBootstrapper.swift b/bitchat/ViewModels/ChatViewModelBootstrapper.swift index 3ec15fa8..6e5d6027 100644 --- a/bitchat/ViewModels/ChatViewModelBootstrapper.swift +++ b/bitchat/ViewModels/ChatViewModelBootstrapper.swift @@ -78,6 +78,23 @@ private extension ChatViewModelBootstrapper { viewModel.privateChatManager.messageRouter = viewModel.messageRouter viewModel.privateChatManager.unifiedPeerService = viewModel.unifiedPeerService viewModel.unifiedPeerService.messageRouter = viewModel.messageRouter + // Surface silent outbox drops (attempt cap, TTL expiry, overflow + // eviction) as a visible failure. The store's no-downgrade rule does + // not cover `.failed` over confirmed receipts, so guard here: a drop + // of an already-delivered/read message (e.g. a stale retained copy) + // must not downgrade its status. + viewModel.messageRouter.onMessageDropped = { [weak viewModel] messageID, _ in + guard let viewModel else { return } + switch viewModel.conversations.deliveryStatus(forMessageID: messageID) { + case .delivered, .read: + return + default: + viewModel.conversations.setDeliveryStatus( + .failed(reason: "Not delivered"), + forMessageID: messageID + ) + } + } viewModel.commandProcessor.contextProvider = viewModel viewModel.commandProcessor.meshService = viewModel.meshService viewModel.participantTracker.configure(context: viewModel) diff --git a/bitchatTests/ChatViewModelDeliveryStatusTests.swift b/bitchatTests/ChatViewModelDeliveryStatusTests.swift index bf859abb..78365b66 100644 --- a/bitchatTests/ChatViewModelDeliveryStatusTests.swift +++ b/bitchatTests/ChatViewModelDeliveryStatusTests.swift @@ -349,6 +349,80 @@ struct ChatViewModelDeliveryStatusTests { #expect(isRead(viewModel.privateChats[peerID]?.last?.deliveryStatus)) } + // MARK: - MessageRouter Drop Wiring Tests + + /// Drives a real outbox drop (per-peer overflow eviction with no + /// reachable transport) and proves the bootstrapper wiring marks the + /// dropped message `.failed` in the conversation store. + @Test @MainActor + func messageRouterDrop_marksMessageFailedInStore() async { + let (viewModel, transport) = makeTestableViewModel() + let peerID = PeerID(str: "0102030405060708") + let droppedID = "router-drop-0" + + let message = BitchatMessage( + id: droppedID, + sender: viewModel.nickname, + content: "Will be dropped", + timestamp: Date(), + isRelay: false, + isPrivate: true, + recipientNickname: "Peer", + senderPeerID: transport.myPeerID, + deliveryStatus: .sent + ) + viewModel.seedPrivateChat([message], for: peerID) + + // No transport is reachable, so every send is queued; the 101st + // enqueue for this peer evicts the oldest queued message. + viewModel.messageRouter.sendPrivate("Will be dropped", to: peerID, recipientNickname: "Peer", messageID: droppedID) + for i in 1...100 { + viewModel.messageRouter.sendPrivate("Filler \(i)", to: peerID, recipientNickname: "Peer", messageID: "router-drop-\(i)") + } + + let status = viewModel.conversations.deliveryStatus(forMessageID: droppedID) + #expect({ + if case .failed = status { return true } + return false + }()) + } + + /// The store's no-downgrade rule does not cover `.failed` over confirmed + /// receipts, so the wiring guards it: a drop of an already-delivered + /// message must not downgrade its status. + @Test @MainActor + func messageRouterDrop_doesNotDowngradeDeliveredStatus() async { + let (viewModel, transport) = makeTestableViewModel() + let peerID = PeerID(str: "0102030405060708") + let droppedID = "router-drop-delivered" + + let message = BitchatMessage( + id: droppedID, + sender: viewModel.nickname, + content: "Already delivered", + timestamp: Date(), + isRelay: false, + isPrivate: true, + recipientNickname: "Peer", + senderPeerID: transport.myPeerID, + deliveryStatus: .delivered(to: "Peer", at: Date()) + ) + viewModel.seedPrivateChat([message], for: peerID) + + // Same eviction-driven drop as above, but the store already recorded + // a delivery confirmation for the message. + viewModel.messageRouter.sendPrivate("Already delivered", to: peerID, recipientNickname: "Peer", messageID: droppedID) + for i in 1...100 { + viewModel.messageRouter.sendPrivate("Filler \(i)", to: peerID, recipientNickname: "Peer", messageID: "router-keep-\(i)") + } + + let status = viewModel.conversations.deliveryStatus(forMessageID: droppedID) + #expect({ + if case .delivered = status { return true } + return false + }()) + } + // MARK: - Status Rank Tests (for deduplication) @Test @MainActor diff --git a/bitchatTests/Services/MessageRouterTests.swift b/bitchatTests/Services/MessageRouterTests.swift index 01aea3ff..13f020a9 100644 --- a/bitchatTests/Services/MessageRouterTests.swift +++ b/bitchatTests/Services/MessageRouterTests.swift @@ -110,6 +110,98 @@ struct MessageRouterTests { #expect(transport.sentPrivateMessages.count == 8) } + // MARK: - Drop visibility (onMessageDropped) + + @Test @MainActor + func flushOutbox_attemptCapDropInvokesOnMessageDropped() async { + let peerID = PeerID(str: "0000000000000009") + let transport = MockTransport() + transport.reachablePeers.insert(peerID) + + let router = MessageRouter(transports: [transport]) + var dropped: [(messageID: String, peerID: PeerID)] = [] + router.onMessageDropped = { dropped.append(($0, $1)) } + + router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m9") + for _ in 0..<10 { + router.flushOutbox(for: peerID) + } + + #expect(dropped.count == 1) + #expect(dropped.first?.messageID == "m9") + #expect(dropped.first?.peerID == peerID) + } + + @Test @MainActor + func flushOutbox_ttlExpiryInvokesOnMessageDroppedAndDoesNotResend() async { + let peerID = PeerID(str: "000000000000000a") + let transport = MockTransport() + let clock = MutableTestClock() + + let router = MessageRouter(transports: [transport], now: { clock.now }) + var dropped: [String] = [] + router.onMessageDropped = { messageID, _ in dropped.append(messageID) } + + // No reachable transport: the message is queued, never sent. + router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m10") + #expect(transport.sentPrivateMessages.isEmpty) + + // Past the 24h TTL the flush must drop it (visibly), not send it. + clock.now = clock.now.addingTimeInterval(24 * 60 * 60 + 1) + transport.reachablePeers.insert(peerID) + router.flushOutbox(for: peerID) + + #expect(dropped == ["m10"]) + #expect(transport.sentPrivateMessages.isEmpty) + + // The drop is final: nothing is retained for later flushes. + router.flushOutbox(for: peerID) + #expect(dropped == ["m10"]) + } + + @Test @MainActor + func cleanupExpiredMessages_invokesOnMessageDroppedForExpiredOnly() async { + let peerID = PeerID(str: "000000000000000b") + let transport = MockTransport() + let clock = MutableTestClock() + + let router = MessageRouter(transports: [transport], now: { clock.now }) + var dropped: [String] = [] + router.onMessageDropped = { messageID, _ in dropped.append(messageID) } + + router.sendPrivate("Old", to: peerID, recipientNickname: "Peer", messageID: "m11-old") + clock.now = clock.now.addingTimeInterval(24 * 60 * 60 - 60) + router.sendPrivate("Fresh", to: peerID, recipientNickname: "Peer", messageID: "m11-fresh") + clock.now = clock.now.addingTimeInterval(120) + + router.cleanupExpiredMessages() + + #expect(dropped == ["m11-old"]) + + // The fresh message survived and still flushes once reachable. + transport.reachablePeers.insert(peerID) + router.flushOutbox(for: peerID) + #expect(transport.sentPrivateMessages.map(\.messageID) == ["m11-fresh"]) + } + + @Test @MainActor + func enqueue_perPeerOverflowEvictionInvokesOnMessageDropped() async { + let peerID = PeerID(str: "000000000000000c") + let transport = MockTransport() + + let router = MessageRouter(transports: [transport]) + var dropped: [String] = [] + router.onMessageDropped = { messageID, _ in dropped.append(messageID) } + + // No reachable transport: everything queues. The cap is 100 per peer, + // so the 101st enqueue evicts the oldest. + for i in 0...100 { + router.sendPrivate("Hello \(i)", to: peerID, recipientNickname: "Peer", messageID: "q\(i)") + } + + #expect(dropped == ["q0"]) + } + @Test @MainActor func sendReadReceipt_usesReachableTransport() async { let peerID = PeerID(str: "0000000000000003") @@ -135,3 +227,9 @@ struct MessageRouterTests { #expect(transport.sentFavoriteNotifications.count == 1) } } + +/// Mutable wall clock injected into `MessageRouter` so TTL expiry is testable +/// without real waiting. +private final class MutableTestClock { + var now = Date(timeIntervalSince1970: 1_700_000_000) +} diff --git a/bitchatTests/Services/NostrRelayManagerTests.swift b/bitchatTests/Services/NostrRelayManagerTests.swift index 7139b430..c6271973 100644 --- a/bitchatTests/Services/NostrRelayManagerTests.swift +++ b/bitchatTests/Services/NostrRelayManagerTests.swift @@ -1195,6 +1195,98 @@ final class NostrRelayManagerTests: XCTestCase { XCTAssertTrue(reconnected) } + func test_pendingSubscriptions_perRelayCapEvictsOldestByInsertionOrder() async { + let relayURL = "wss://pending-cap.example" + // Tor stalled: nothing flushes, so every REQ stays pending. + let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false) + let cap = TransportConfig.nostrPendingSubscriptionsPerRelayCap + + for i in 0..<(cap + 3) { + context.manager.subscribe(filter: makeFilter(), id: "cap-sub-\(i)", relayUrls: [relayURL], handler: { _ in }) + } + + XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), cap) + let pendingIDs = context.manager.debugPendingSubscriptionIDs(for: relayURL) + // The three oldest entries were evicted; the newest survive. + for i in 0..<3 { + XCTAssertFalse(pendingIDs.contains("cap-sub-\(i)"), "expected cap-sub-\(i) to be evicted") + } + for i in 3..<(cap + 3) { + XCTAssertTrue(pendingIDs.contains("cap-sub-\(i)"), "expected cap-sub-\(i) to be retained") + } + } + + func test_pendingSubscriptions_staleEntriesSweptOnConnectAttempt() async { + let relayURL = "wss://pending-sweep.example" + // Tor stalled: the REQ stays pending and no socket ever opens. + let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false) + + context.manager.subscribe(filter: makeFilter(), id: "stale-pending-sub", relayUrls: [relayURL], handler: { _ in }) + XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1) + + // Just under the TTL the entry survives a connect attempt. + context.clock.now = context.clock.now.addingTimeInterval(TransportConfig.nostrPendingSubscriptionTTLSeconds - 1) + context.manager.ensureConnections(to: [relayURL]) + XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1) + + // Past the TTL the next connect attempt sweeps it. + context.clock.now = context.clock.now.addingTimeInterval(2) + context.manager.ensureConnections(to: [relayURL]) + XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 0) + } + + func test_reconnectBackoff_appliesJitterWithinConfiguredBounds() async { + let relayURL = "wss://jitter-bounds.example" + // Pin the jitter source to the extremes and the midpoint of [0, 1). + let jitter = JitterSequence([0.0, 1.0.nextDown, 0.25]) + let context = makeContext(permission: .denied, jitterUnit: { jitter.next() }) + // Persistent ping failure: every connect attempt fails and schedules + // the next reconnect with an increasing attempt count. + context.sessionFactory.pingErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut) + + context.manager.ensureConnections(to: [relayURL]) + + var delays: [TimeInterval] = [] + for attempt in 1...3 { + let scheduled = await waitUntil { context.scheduler.scheduled.count == 1 } + XCTAssertTrue(scheduled, "reconnect for attempt \(attempt) was not scheduled") + delays.append(context.scheduler.scheduled[0].delay) + context.scheduler.runNext() + } + + // Bases: 1s, 2s, 4s. Jitter factors: 0.8, ~1.2, 0.9. + XCTAssertEqual(delays[0], 0.8 * TransportConfig.nostrRelayInitialBackoffSeconds, accuracy: 1e-9) + XCTAssertEqual(delays[1], 1.2 * TransportConfig.nostrRelayInitialBackoffSeconds * TransportConfig.nostrRelayBackoffMultiplier, accuracy: 1e-6) + XCTAssertEqual(delays[2], 0.9 * TransportConfig.nostrRelayInitialBackoffSeconds * pow(TransportConfig.nostrRelayBackoffMultiplier, 2), accuracy: 1e-9) + } + + func test_reconnectBackoff_realRandomJitterStaysInBoundsAndVaries() async { + let relayURL = "wss://jitter-random.example" + let context = makeContext(permission: .denied, jitterUnit: { Double.random(in: 0..<1) }) + context.sessionFactory.pingErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut) + + context.manager.ensureConnections(to: [relayURL]) + + var factors: [Double] = [] + for attempt in 1...5 { + let scheduled = await waitUntil { context.scheduler.scheduled.count == 1 } + XCTAssertTrue(scheduled, "reconnect for attempt \(attempt) was not scheduled") + let base = min( + TransportConfig.nostrRelayInitialBackoffSeconds * pow(TransportConfig.nostrRelayBackoffMultiplier, Double(attempt - 1)), + TransportConfig.nostrRelayMaxBackoffSeconds + ) + let factor = context.scheduler.scheduled[0].delay / base + XCTAssertGreaterThanOrEqual(factor, 1.0 - TransportConfig.nostrRelayBackoffJitterRatio) + XCTAssertLessThan(factor, 1.0 + TransportConfig.nostrRelayBackoffJitterRatio) + factors.append(factor) + context.scheduler.runNext() + } + + // A real RNG must not produce a constant delay across attempts + // (5 identical uniform doubles is probability ~0). + XCTAssertGreaterThan(Set(factors).count, 1) + } + private func makeContext( permission: LocationChannelManager.PermissionState, favorites: Set = [], @@ -1202,7 +1294,8 @@ final class NostrRelayManagerTests: XCTestCase { userTorEnabled: Bool = false, torEnforced: Bool = false, torIsReady: Bool = true, - torIsForeground: Bool = true + torIsForeground: Bool = true, + jitterUnit: @escaping () -> Double = { 0.5 } // 0.5 -> jitter factor 1.0 (no jitter) ) -> RelayManagerTestContext { let permissionSubject = CurrentValueSubject(permission) let favoritesSubject = CurrentValueSubject, Never>(favorites) @@ -1228,7 +1321,8 @@ final class NostrRelayManagerTests: XCTestCase { scheduleAfter: { delay, action in scheduler.schedule(delay: delay, action: action) }, - now: { clock.now } + now: { clock.now }, + jitterUnit: jitterUnit ) ) return RelayManagerTestContext( @@ -1305,6 +1399,20 @@ private final class MutableClock { } } +/// Deterministic jitter source: returns the queued values in order, then a +/// neutral 0.5 (jitter factor 1.0) once exhausted. +private final class JitterSequence { + private var values: [Double] + + init(_ values: [Double]) { + self.values = values + } + + func next() -> Double { + values.isEmpty ? 0.5 : values.removeFirst() + } +} + private final class MutableBool { var value: Bool