From 3d0a97d2e59d2c3b3709c9c466c49e6594a30cab Mon Sep 17 00:00:00 2001 From: Dev Date: Tue, 28 Jul 2026 17:06:42 +0300 Subject: [PATCH] Fix rotating NDR relay subscriptions --- bitchat/Nostr/NostrRelayManager.swift | 113 +++++++--- bitchat/Services/NdrNostrService.swift | 42 ++-- .../NdrOutOfBandTransportTests.swift | 106 +++++++++- .../Services/NostrRelayManagerTests.swift | 198 ++++++++++++++++++ 4 files changed, 414 insertions(+), 45 deletions(-) diff --git a/bitchat/Nostr/NostrRelayManager.swift b/bitchat/Nostr/NostrRelayManager.swift index 98b81c3c..2679acff 100644 --- a/bitchat/Nostr/NostrRelayManager.swift +++ b/bitchat/Nostr/NostrRelayManager.swift @@ -200,7 +200,10 @@ final class NostrRelayManager: ObservableObject { private var hasMutualFavorites: Bool = false private var hasLocationPermission: Bool = false private var connections: [String: NostrRelayConnectionProtocol] = [:] - private var subscriptions: [String: Set] = [:] // relay URL -> active subscription IDs + // Relay URL -> subscription ID -> logical request generation currently + // installed on that socket. A same-ID REQ with a newer generation + // atomically replaces its filter under NIP-01. + private var subscriptions: [String: [String: UInt64]] = [:] // 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. @@ -788,6 +791,14 @@ final class NostrRelayManager: ObservableObject { forKey: id ) } + if let previousRequestState { + let removedRelayURLs = previousRequestState.relayURLs + .subtracting(requestState.relayURLs) + closeSubscription( + id: id, + on: removedRelayURLs + ) + } messageHandlers[id] = handler if previousRequestState == requestState, subscriptionStateExists(id: id, requestState: requestState) @@ -899,6 +910,15 @@ final class NostrRelayManager: ObservableObject { /// Unsubscribe from a subscription func unsubscribe(id: String) { + var relayURLs = subscriptionRequestState[id]?.relayURLs ?? [] + for (relayURL, active) in subscriptions where active[id] != nil { + relayURLs.insert(relayURL) + } + for (relayURL, pending) in pendingSubscriptions + where pending[id] != nil { + relayURLs.insert(relayURL) + } + messageHandlers.removeValue(forKey: id) subscriptionGenerations.removeValue(forKey: id) removeRecentInboundEvents(forSubscriptionID: id) @@ -906,22 +926,29 @@ final class NostrRelayManager: ObservableObject { subscriptionRequestState.removeValue(forKey: id) pendingEOSECallbacks.removeValue(forKey: id) eoseTrackers.removeValue(forKey: id) - for url in Array(pendingSubscriptions.keys) { - pendingSubscriptions[url]?.removeValue(forKey: id) + closeSubscription(id: id, on: relayURLs) + } + + private func closeSubscription( + id: String, + on relayURLs: Set + ) { + guard !relayURLs.isEmpty else { return } + for relayURL in relayURLs { + subscriptions[relayURL]?.removeValue(forKey: id) + pendingSubscriptions[relayURL]?.removeValue(forKey: id) } - + let req = NostrRequest.close(id: id) let message = try? encoder.encode(req) - guard let messageData = message, let messageString = String(data: messageData, encoding: .utf8) else { return } - - // Send unsubscribe to all relays - for (relayUrl, connection) in connections { - if subscriptions[relayUrl]?.contains(id) == true { - subscriptions[relayUrl]?.remove(id) + + for relayURL in relayURLs { + if let connection = connections[relayURL] { connection.send(.string(messageString)) { _ in - // Local state is cleared before sending so callers can re-subscribe immediately. + // Local state is cleared first so a later same-ID REQ can + // register immediately and stale callbacks remain inert. } } } @@ -1036,10 +1063,14 @@ final class NostrRelayManager: ObservableObject { } private func subscriptionStateExists(id: String, requestState: SubscriptionRequestState) -> Bool { - guard !requestState.relayURLs.isEmpty else { return true } + guard !requestState.relayURLs.isEmpty, + let generation = subscriptionGenerations[id] + else { + return requestState.relayURLs.isEmpty + } return requestState.relayURLs.allSatisfy { url in pendingSubscriptions[url]?[id]?.messageString == requestState.messageString || - subscriptions[url]?.contains(id) == true + subscriptions[url]?[id] == generation } } @@ -1239,12 +1270,32 @@ 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] ?? [:]).mapValues(\.messageString) - for (id, state) in subscriptionRequestState where state.relayURLs.contains(relayUrl) && toSend[id] == nil { - toSend[id] = state.messageString + var toSend: [ + String: (messageString: String, generation: UInt64) + ] = [:] + for (id, pending) in pendingSubscriptions[relayUrl] ?? [:] { + guard let state = subscriptionRequestState[id], + state.relayURLs.contains(relayUrl), + state.messageString == pending.messageString, + let generation = subscriptionGenerations[id] + else { + pendingSubscriptions[relayUrl]?.removeValue(forKey: id) + continue + } + toSend[id] = (pending.messageString, generation) } - for (id, messageString) in toSend { - if self.subscriptions[relayUrl]?.contains(id) == true { + for (id, state) in subscriptionRequestState + where state.relayURLs.contains(relayUrl) + { + guard let generation = subscriptionGenerations[id] else { + continue + } + toSend[id] = (state.messageString, generation) + } + for (id, request) in toSend { + let messageString = request.messageString + let generation = request.generation + if subscriptions[relayUrl]?[id] == generation { // Already subscribed on this relay (e.g. a tracker promoted // after an earlier flush): its EOSE is coming, count it. markEOSESubscribed(id: id, relayUrl: relayUrl) @@ -1264,12 +1315,26 @@ final class NostrRelayManager: ObservableObject { // 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) + // Old sockets and old same-ID filters can complete + // after a replacement has already become canonical. + guard let connection, + self.connections[relayUrl] === connection, + self.subscriptionGenerations[id] == generation, + let desired = + self.subscriptionRequestState[id], + desired.relayURLs.contains(relayUrl), + desired.messageString == messageString + else { + return + } + self.subscriptions[relayUrl, default: [:]][id] = + generation + if self.pendingSubscriptions[relayUrl]?[id]? + .messageString == messageString + { + self.pendingSubscriptions[relayUrl]? + .removeValue(forKey: id) + } } } } diff --git a/bitchat/Services/NdrNostrService.swift b/bitchat/Services/NdrNostrService.swift index 777644ec..d912d74e 100644 --- a/bitchat/Services/NdrNostrService.swift +++ b/bitchat/Services/NdrNostrService.swift @@ -880,30 +880,32 @@ final class NdrNostrService { synchronousAcks.append(action.actionId) continue } - if !activeSubIDs.contains(subscriptionID) { - let registered = relayManager.subscribe( - filter: filter, - id: subscriptionID, - relayUrls: nil, - handler: { [weak self, manager] event in - self?.processInboundNostrEvent( - event, - manager: manager, - epoch: epoch - ) - }, - onEOSE: nil - ) - guard registered else { - deferForTransientRetry( - action.actionId, + // The native runtime deliberately reuses its account-scoped + // subscription ID while rotating ephemeral sender authors. + // Register every durable replacement before acknowledging it; + // NIP-01 replaces the relay's live REQ atomically by ID. + let registered = relayManager.subscribe( + filter: filter, + id: subscriptionID, + relayUrls: nil, + handler: { [weak self, manager] event in + self?.processInboundNostrEvent( + event, manager: manager, epoch: epoch ) - continue - } - activeSubIDs.insert(subscriptionID) + }, + onEOSE: nil + ) + guard registered else { + deferForTransientRetry( + action.actionId, + manager: manager, + epoch: epoch + ) + continue } + activeSubIDs.insert(subscriptionID) synchronousAcks.append(action.actionId) case "unsubscribe": diff --git a/bitchatTests/DoubleRatchet/NdrOutOfBandTransportTests.swift b/bitchatTests/DoubleRatchet/NdrOutOfBandTransportTests.swift index 6f768c04..0e32bade 100644 --- a/bitchatTests/DoubleRatchet/NdrOutOfBandTransportTests.swift +++ b/bitchatTests/DoubleRatchet/NdrOutOfBandTransportTests.swift @@ -32,7 +32,9 @@ final class FakeRelayManager: NostrRelayManaging { var subscriptionRegistrationSucceeds = true var activeSubscriptions: [Subscription] { - subscriptions.filter { activeSubscriptionIDs.contains($0.id) } + activeSubscriptionIDs.compactMap { id in + subscriptions.last(where: { $0.id == id }) + } } func resetSentEvents() { @@ -89,6 +91,18 @@ final class FakeRelayManager: NostrRelayManaging { } subscription.handler(event) } + + @discardableResult + func deliverMatching(_ event: NostrEvent) -> Bool { + guard let subscription = activeSubscriptions.first(where: { + ($0.filter.kinds?.contains(event.kind) ?? true) + && ($0.filter.authors?.contains(event.pubkey) ?? true) + }) else { + return false + } + subscription.handler(event) + return true + } } @MainActor @@ -2571,6 +2585,96 @@ struct NdrOutOfBandTransportTests { #expect(restored.hasActiveSession(with: alice.publicKeyHex)) } + @Test("Rotated NDR filters stay live across bidirectional relay turns") + @MainActor + func rotatedSubscriptionFiltersReplaceTheLiveRelayRequest() throws { + let alice = try NostrIdentity.generate() + let bob = try NostrIdentity.generate() + let aliceRelay = FakeRelayManager() + let bobRelay = FakeRelayManager() + let aliceService = try makeService( + label: "subscription-rotation-alice", + relay: aliceRelay + ) + let bobService = try makeService( + label: "subscription-rotation-bob", + relay: bobRelay + ) + aliceService.configureIfNeeded(identity: alice) + bobService.configureIfNeeded(identity: bob) + try establishPairwiseSessions( + aliceService, + bobService, + firstIdentity: alice, + secondIdentity: bob, + firstRelay: aliceRelay, + secondRelay: bobRelay + ) + + var receivedByAlice: [String] = [] + var receivedByBob: [String] = [] + aliceService.onDecryptedMessage = { message, completion in + receivedByAlice.append(message.event.content) + completion(.consumed) + } + bobService.onDecryptedMessage = { message, completion in + receivedByBob.append(message.event.content) + completion(.consumed) + } + aliceRelay.resetSentEvents() + bobRelay.resetSentEvents() + + guard case let .sent(_, firstOuterID) = aliceService.send( + "bitchat1:first-turn", + to: bob.publicKeyHex + ), + let firstOuter = aliceRelay.sentEvents.first(where: { + $0.id == firstOuterID + }) + else { + Issue.record("Expected Alice's first pairwise send") + return + } + #expect(bobRelay.deliverMatching(firstOuter)) + #expect(receivedByBob == ["bitchat1:first-turn"]) + + guard case let .sent(_, replyOuterID) = bobService.send( + "bitchat1:reply-turn", + to: alice.publicKeyHex + ), + let replyOuter = bobRelay.sentEvents.first(where: { + $0.id == replyOuterID + }) + else { + Issue.record("Expected Bob's pairwise reply") + return + } + #expect(aliceRelay.deliverMatching(replyOuter)) + #expect(receivedByAlice == ["bitchat1:reply-turn"]) + + guard case let .sent(_, secondOuterID) = aliceService.send( + "bitchat1:second-turn", + to: bob.publicKeyHex + ), + let secondOuter = aliceRelay.sentEvents.first(where: { + $0.id == secondOuterID + }) + else { + Issue.record("Expected Alice's second pairwise send") + return + } + #expect( + bobRelay.deliverMatching(secondOuter), + "the live relay filter must follow the rotated sender key" + ) + #expect( + receivedByBob + == ["bitchat1:first-turn", "bitchat1:second-turn"] + ) + #expect(aliceRelay.unsubscribedIDs.isEmpty) + #expect(bobRelay.unsubscribedIDs.isEmpty) + } + @Test("Absolute message expiry survives the pairwise round trip") @MainActor func outboundExpirationRoundTripsToDelivery() throws { diff --git a/bitchatTests/Services/NostrRelayManagerTests.swift b/bitchatTests/Services/NostrRelayManagerTests.swift index 5d889113..bff005f0 100644 --- a/bitchatTests/Services/NostrRelayManagerTests.swift +++ b/bitchatTests/Services/NostrRelayManagerTests.swift @@ -798,6 +798,195 @@ final class NostrRelayManagerTests: XCTestCase { XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 0) } + func test_subscribe_changedActiveRequestReplacesSameID() async { + let relayURL = "wss://rotating-subscribe.example" + let context = makeContext(permission: .denied) + let initialFilter = makeFilter() + + context.manager.subscribe( + filter: initialFilter, + id: "rotating-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + let firstSent = await waitUntil { + context.sessionFactory.latestConnection(for: relayURL)? + .sentStrings.count == 1 + } + XCTAssertTrue(firstSent) + + var rotatedFilter = initialFilter + rotatedFilter.authors = [String(repeating: "a", count: 64)] + context.manager.subscribe( + filter: rotatedFilter, + id: "rotating-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + + let replacementSent = await waitUntil { + context.sessionFactory.latestConnection(for: relayURL)? + .sentStrings.count == 2 + } + XCTAssertTrue( + replacementSent, + "NIP-01 replaces a live subscription atomically when a new REQ reuses its ID" + ) + let messages = context.sessionFactory + .latestConnection(for: relayURL)?.sentStrings ?? [] + XCTAssertTrue(messages.allSatisfy { $0.contains("\"REQ\"") }) + XCTAssertNotEqual(messages.first, messages.last) + let replacementCommitted = await waitUntil { + context.manager.debugPendingSubscriptionCount( + for: relayURL + ) == 0 + } + XCTAssertTrue(replacementCommitted) + } + + func test_subscribe_staleCompletionCannotRegressReplacement() async { + let relayURL = "wss://subscription-completion-order.example" + let context = makeContext(permission: .denied) + context.sessionFactory.deferSendCompletions = true + let initialFilter = makeFilter() + + context.manager.subscribe( + filter: initialFilter, + id: "ordered-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + let firstSent = await waitUntil { + context.sessionFactory.latestConnection(for: relayURL)? + .sentStrings.count == 1 + } + XCTAssertTrue(firstSent) + + var replacementFilter = initialFilter + replacementFilter.authors = [String(repeating: "b", count: 64)] + context.manager.subscribe( + filter: replacementFilter, + id: "ordered-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + let replacementSent = await waitUntil { + context.sessionFactory.latestConnection(for: relayURL)? + .sentStrings.count == 2 + } + XCTAssertTrue(replacementSent) + + let connection = context.sessionFactory.latestConnection( + for: relayURL + ) + connection?.flushDeferredSendCompletion(at: 1) + let replacementCommitted = await waitUntil { + context.manager.debugPendingSubscriptionCount( + for: relayURL + ) == 0 + } + XCTAssertTrue(replacementCommitted) + + // The first REQ completes last. It must not restore the old logical + // generation or make the same replacement look absent. + connection?.flushDeferredSendCompletion(at: 0) + try? await Task.sleep(nanoseconds: 20_000_000) + context.manager.subscribe( + filter: replacementFilter, + id: "ordered-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + try? await Task.sleep(nanoseconds: 20_000_000) + + XCTAssertEqual(connection?.sentStrings.count, 2) + XCTAssertEqual( + context.manager.debugPendingSubscriptionCount(for: relayURL), + 0 + ) + } + + func test_subscribe_reconnectReplaysOnlyLatestReplacement() async { + let relayURL = "wss://subscription-reconnect.example" + let context = makeContext(permission: .denied) + let initialFilter = makeFilter() + + context.manager.subscribe( + filter: initialFilter, + id: "reconnecting-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + let oldConnection = context.sessionFactory.latestConnection( + for: relayURL + ) + oldConnection?.deferSendCompletions = true + let initialSent = await waitUntil { + oldConnection?.sentStrings.count == 1 + } + XCTAssertTrue(initialSent) + + var replacementFilter = initialFilter + replacementFilter.authors = [String(repeating: "c", count: 64)] + context.manager.subscribe( + filter: replacementFilter, + id: "reconnecting-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + let replacementSent = await waitUntil { + oldConnection?.sentStrings.count == 2 + } + XCTAssertTrue(replacementSent) + + oldConnection?.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.count == 1 + } + XCTAssertTrue(replayed) + + let connections = + context.sessionFactory.connectionsByURL[relayURL] ?? [] + XCTAssertEqual( + connections.last?.sentStrings.first, + oldConnection?.sentStrings.last + ) + XCTAssertNotEqual( + oldConnection?.sentStrings.first, + oldConnection?.sentStrings.last + ) + + oldConnection?.flushDeferredSendCompletions() + try? await Task.sleep(nanoseconds: 20_000_000) + context.manager.subscribe( + filter: replacementFilter, + id: "reconnecting-sub", + relayUrls: [relayURL], + handler: { _ in } + ) + try? await Task.sleep(nanoseconds: 20_000_000) + + XCTAssertEqual(connections.last?.sentStrings.count, 1) + XCTAssertEqual( + context.manager.debugPendingSubscriptionCount(for: relayURL), + 0 + ) + } + func test_subscribe_waitsForTorReadinessAndPreservesEOSECallback() async throws { let relayURL = "wss://tor-subscribe.example" let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false) @@ -2374,6 +2563,7 @@ private final class MockRelaySessionFactory: NostrRelaySessionProtocol { private(set) var connectionsByURL: [String: [MockRelayConnection]] = [:] var pingErrorByURL: [String: Error?] = [:] var sendErrorByURL: [String: Error?] = [:] + var deferSendCompletions = false var allConnections: [MockRelayConnection] { connectionsByURL.values.flatMap { $0 } @@ -2386,6 +2576,7 @@ private final class MockRelaySessionFactory: NostrRelaySessionProtocol { pingError: pingErrorByURL[url.absoluteString] ?? nil, sendError: sendErrorByURL[url.absoluteString] ?? nil ) + connection.deferSendCompletions = deferSendCompletions connectionsByURL[url.absoluteString, default: []].append(connection) return connection } @@ -2446,6 +2637,13 @@ private final class MockRelayConnection: NostrRelayConnectionProtocol { } } + func flushDeferredSendCompletion(at index: Int) { + guard deferredSendCompletions.indices.contains(index) else { + return + } + deferredSendCompletions.remove(at: index)(sendError) + } + func receive(completionHandler: @escaping (Result) -> Void) { if !pendingResults.isEmpty { completionHandler(pendingResults.removeFirst())