nik/bitchat
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat.git synced 2026-08-22 07:16:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix rotating NDR relay subscriptions

Browse Source
This commit is contained in:
Dev 2026-07-28 17:06:42 +03:00
parent 78c23dfce4
commit 3d0a97d2e5
4 changed files with 414 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
bitchat/Nostr/NostrRelayManager.swift
View File

@ -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<String>] = [:] // 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<String>
) {
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)
}
}
}
}

42
bitchat/Services/NdrNostrService.swift
View File

@ -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":

106
bitchatTests/DoubleRatchet/NdrOutOfBandTransportTests.swift
View File

@ -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 {

198
bitchatTests/Services/NostrRelayManagerTests.swift
View File

@ -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<URLSessionWebSocketTask.Message, Error>) -> Void) {
if !pendingResults.isEmpty {
completionHandler(pendingResults.removeFirst())