diff --git a/Bitkit/Services/BackupService.swift b/Bitkit/Services/BackupService.swift index 970474c94..dd24da660 100644 --- a/Bitkit/Services/BackupService.swift +++ b/Bitkit/Services/BackupService.swift @@ -113,6 +113,7 @@ class BackupService { @MainActor weak var usdtWallet: UsdtWalletManager? private var backupJobs: [BackupCategory: Task] = [:] private var runningBackupTasks: [BackupCategory: Task] = [:] + // Register and cancel subscriptions on the synchronous backup queue. private var cancellables = Set() private var periodicCheckTask: Task? private var isObserving = false @@ -282,20 +283,24 @@ class BackupService { @discardableResult func startObservingBackups() -> Task? { guard !shouldSkipBackup() else { return nil } + let generation = stateQueue.sync { foregroundObservationGeneration } return Task { - let shouldStart = try? await ServiceQueue.background(.backup) { - guard !self.isObserving else { return false } - self.isObserving = true - return true + let publishers = await MainActor.run { + let settings = SettingsViewModel.shared + return (settings.settingsPublisher, settings.widgetsPublisher, settings.appStatePublisher) } + try? await ServiceQueue.background(.backup) { + guard !self.isObserving, !self.shouldSkipBackup(), + self.stateQueue.sync(execute: { self.foregroundObservationGeneration == generation }) + else { return } + self.isObserving = true - guard shouldStart == true else { return } - - Logger.debug("Start observing backup statuses and data store changes", context: "BackupService") - startBackupStatusObservers() - startDataStoreListeners() - startPeriodicBackupFailureCheck() + Logger.debug("Start observing backup statuses and data store changes", context: "BackupService") + self.startBackupStatusObservers(generation: generation) + self.startDataStoreListeners(settings: publishers.0, widgets: publishers.1, appState: publishers.2) + self.startPeriodicBackupFailureCheck() + } } } @@ -311,13 +316,13 @@ class BackupService { self.runningBackupTasks.removeAll() self.periodicCheckTask?.cancel() self.periodicCheckTask = nil + self.cancellables.removeAll() return wasObserving } await walletBackupWriter.waitUntilIdle() await usdtBackupWriter.waitUntilIdle() guard shouldStop == true else { return } - cancellables.removeAll() Logger.debug("Stopped observing backup statuses and data store changes", context: "BackupService") } @@ -710,7 +715,7 @@ class BackupService { // MARK: - Private Methods - private func startBackupStatusObservers() { + private func startBackupStatusObservers(generation: UInt64) { for category in BackupCategory.allCases { let categoryPublisher = backupStatusesPublisher .prepend(getAllBackupStatuses()) @@ -728,7 +733,7 @@ class BackupService { guard let self, !self.shouldSkipBackup(category: category) else { return } if status.isRequired && !status.running { Task { - await self.scheduleBackup(category: category) + await self.scheduleBackup(category: category, foregroundGeneration: generation) } } } @@ -738,27 +743,28 @@ class BackupService { Logger.debug("Started \(BackupCategory.allCases.count) backup status observers", context: "BackupService") } - private func startDataStoreListeners() { - Task { @MainActor [weak self] in - guard let self else { return } - // SETTINGS - SettingsViewModel.shared.settingsPublisher - .debounce(for: .milliseconds(500), scheduler: DispatchQueue.main) - .sink { [weak self] _ in - guard let self, !self.shouldSkipBackup() else { return } - markBackupRequired(category: .settings) - } - .store(in: &cancellables) + private func startDataStoreListeners( + settings: AnyPublisher<[String: Any], Never>, + widgets: AnyPublisher, + appState: AnyPublisher + ) { + // SETTINGS + settings + .debounce(for: .milliseconds(500), scheduler: DispatchQueue.main) + .sink { [weak self] _ in + guard let self, !self.shouldSkipBackup() else { return } + markBackupRequired(category: .settings) + } + .store(in: &cancellables) - // WIDGETS - SettingsViewModel.shared.widgetsPublisher - .debounce(for: .milliseconds(500), scheduler: DispatchQueue.main) - .sink { [weak self] _ in - guard let self, !self.shouldSkipBackup() else { return } - markBackupRequired(category: .widgets) - } - .store(in: &cancellables) - } + // WIDGETS + widgets + .debounce(for: .milliseconds(500), scheduler: DispatchQueue.main) + .sink { [weak self] _ in + guard let self, !self.shouldSkipBackup() else { return } + markBackupRequired(category: .widgets) + } + .store(in: &cancellables) // Record requirements synchronously; the scheduled upload already has a debounce. WalletBackupChangeTracker.localChanges @@ -806,16 +812,13 @@ class BackupService { .store(in: &cancellables) // APP STATE (UserDefaults changes, etc.) - Task { @MainActor [weak self] in - guard let self else { return } - SettingsViewModel.shared.appStatePublisher - .debounce(for: .milliseconds(500), scheduler: DispatchQueue.main) - .sink { [weak self] _ in - guard let self, !self.shouldSkipBackup() else { return } - markBackupRequired(category: .metadata) - } - .store(in: &cancellables) - } + appState + .debounce(for: .milliseconds(500), scheduler: DispatchQueue.main) + .sink { [weak self] _ in + guard let self, !self.shouldSkipBackup() else { return } + markBackupRequired(category: .metadata) + } + .store(in: &cancellables) // BLOCKTANK CoreService.shared.blocktank.stateChangedPublisher @@ -845,15 +848,11 @@ class BackupService { } private func startPeriodicBackupFailureCheck() { - Task { - try? await ServiceQueue.background(.backup) { - self.periodicCheckTask = Task { - while !Task.isCancelled { - try? await Task.sleep(nanoseconds: UInt64(Self.backupFailureCheckInterval * 1_000_000_000)) - guard !Task.isCancelled else { break } - self.checkForFailedBackups() - } - } + periodicCheckTask = Task { + while !Task.isCancelled { + try? await Task.sleep(nanoseconds: UInt64(Self.backupFailureCheckInterval * 1_000_000_000)) + guard !Task.isCancelled else { break } + self.checkForFailedBackups() } } } diff --git a/Bitkit/Services/PaykitPaymentRequestService.swift b/Bitkit/Services/PaykitPaymentRequestService.swift index 9cd0f1e01..bfff5a91d 100644 --- a/Bitkit/Services/PaykitPaymentRequestService.swift +++ b/Bitkit/Services/PaykitPaymentRequestService.swift @@ -1608,6 +1608,9 @@ final class PaykitPaymentRequestManager { guard generation == eligibilityGeneration, currentStateGeneration == stateGeneration else { return } + if case PaykitPaymentRequestError.operationInProgress = error { + return + } eligibleTargets = eligibleTargets.filter { target in savedPublicKeys.contains { PubkyPublicKeyFormat.matches($0, target.publicKey) } } diff --git a/Bitkit/Services/PrivatePaykitService+Contacts.swift b/Bitkit/Services/PrivatePaykitService+Contacts.swift index 8f0e0515a..b0f701970 100644 --- a/Bitkit/Services/PrivatePaykitService+Contacts.swift +++ b/Bitkit/Services/PrivatePaykitService+Contacts.swift @@ -1,5 +1,6 @@ import Foundation import Paykit +import UIKit // MARK: - Saved Contacts @@ -14,12 +15,13 @@ extension PrivatePaykitService { } struct EndpointPublicationOperations { - let currentPublicKey: () async -> String? + let currentPublicKey: () async throws -> String? let ensureLink: (_ publicKey: String) async throws -> LinkedPeerState let buildEndpoints: (_ publicKey: String) async throws -> [PublicPaykitService.Endpoint] let syncPaymentLists: (_ updates: [PrivatePaymentListReservationUpdateInput]) async throws -> PrivatePaymentListDeliveryReport var linkedPeers: () async throws -> [LinkedPeerRecord] = { [] } - var canPublish: () async -> Bool = { true } + var canPublish: () async throws -> Bool = { true } + var isApplicationActive: () async -> Bool = { true } } struct PrivateMessageDrainOperations { @@ -45,6 +47,7 @@ extension PrivatePaykitService { struct PrivateMessageRetryOperations { var now: () -> Date = Date.init var sleep: (UInt64) async throws -> Void = { try await Task.sleep(nanoseconds: $0) } + var shouldDeferRefresh: () async -> Bool = { await PaykitPaymentActivity.shared.shouldDeferRefresh } var currentPublicKey: (PaykitSdkOperationLock.Priority) async throws -> String? = { try await PaykitSdkService.shared.identityStatus(priority: $0)?.publicKey } @@ -127,7 +130,7 @@ extension PrivatePaykitService { return await prepareSavedContacts( publicKeys, publicationUnavailableReason: privateEndpointPublicationUnavailabilityReason(wallet: wallet), - prepareLinks: { await self.prepareRelevantPrivateLinksIfAvailable($0, reason: "prepare") }, + prepareLinks: { try? await self.prepareRelevantPrivateLinksIfAvailable($0, reason: "prepare") }, publishEndpoints: { publicKeys in await PrivatePaykitAddressReservationStore.shared.reconcileReservedIndexesWithLdk() return await self.syncLocalEndpointPublication( @@ -293,6 +296,22 @@ extension PrivatePaykitService { return generation == preparationGeneration && !Task.isCancelled } + private func waitForForeground(generation: Int, isApplicationActive: () async -> Bool) async -> Bool { + guard generation == preparationGeneration, !Task.isCancelled else { return false } + let id = UUID() + let (stream, continuation) = AsyncStream.makeStream() + backgroundWorkWaiters[id] = continuation + defer { + backgroundWorkWaiters.removeValue(forKey: id) + continuation.finish() + } + if await isApplicationActive() { + continuation.finish() + } + for await _ in stream {} + return generation == preparationGeneration && !Task.isCancelled + } + private func resumeBackgroundWorkWaiters() { let waiters = backgroundWorkWaiters.values backgroundWorkWaiters.removeAll() @@ -317,33 +336,66 @@ extension PrivatePaykitService { reason: String = "refresh", isSessionCurrent: (@MainActor () -> Bool)? = nil ) async -> Error? { - if let isSessionCurrent, await !isSessionCurrent() { - return nil - } let generation = preparationGeneration - if !requireImmediatePublication, await !waitForBackgroundWork(generation: generation) { - return nil - } let operations = endpointPublicationOperations( wallet: wallet, forceRefreshLightning: forceRefreshLightning, readPriority: requireImmediatePublication ? .ordered : .background ) - guard await operations.canPublish() else { - await prepareRelevantPrivateLinksIfAvailable(publicKeys, reason: reason, isBackgroundWork: !requireImmediatePublication) - return requireImmediatePublication && !publicKeys.isEmpty ? PrivatePaykitError.privateUnavailable : nil - } - guard generation == preparationGeneration else { - return requireImmediatePublication ? PrivatePaykitError.privateUnavailable : nil + var expectedIdentity: String? + while !Task.isCancelled, generation == preparationGeneration { + if !requireImmediatePublication { + do { + try await PaykitPaymentActivity.shared.waitUntilIdle(includingReviews: true) + } catch { + return nil + } + guard await waitForBackgroundWork(generation: generation) else { return nil } + } + if let isSessionCurrent, await !isSessionCurrent() { + return nil + } + guard generation == preparationGeneration else { return nil } + do { + guard try await operations.canPublish() else { + if !requireImmediatePublication, await !operations.isApplicationActive() { + guard await waitForForeground(generation: generation, isApplicationActive: operations.isApplicationActive) else { return nil } + continue + } + try await prepareRelevantPrivateLinksIfAvailable( + publicKeys, reason: reason, isBackgroundWork: !requireImmediatePublication, + currentPublicKey: operations.currentPublicKey + ) + return requireImmediatePublication && !publicKeys.isEmpty ? PrivatePaykitError.privateUnavailable : nil + } + guard generation == preparationGeneration else { + return requireImmediatePublication ? PrivatePaykitError.privateUnavailable : nil + } + guard !normalizedSavedContactKeys(publicKeys).isEmpty else { return nil } + guard let identity = try await operations.currentPublicKey() else { + return requireImmediatePublication ? PubkyServiceError.sessionNotActive : nil + } + guard generation == preparationGeneration else { + return requireImmediatePublication ? PrivatePaykitError.privateUnavailable : nil + } + if let expectedIdentity, expectedIdentity != identity { + return requireImmediatePublication ? PubkyServiceError.identityChanged : nil + } + expectedIdentity = identity + let error = await syncLocalEndpointPublication( + for: publicKeys, reason: reason, requireImmediatePublication: requireImmediatePublication, + isSessionCurrent: isSessionCurrent, operations: operations, expectedIdentity: identity + ) + guard !requireImmediatePublication, case PaykitPaymentRequestError.operationInProgress? = error else { return error } + } catch PaykitPaymentRequestError.operationInProgress { + if requireImmediatePublication { + return PaykitPaymentRequestError.operationInProgress + } + } catch { + return error + } } - - return await syncLocalEndpointPublication( - for: publicKeys, - reason: reason, - requireImmediatePublication: requireImmediatePublication, - isSessionCurrent: isSessionCurrent, - operations: operations - ) + return nil } func removePublishedEndpoints( @@ -356,7 +408,7 @@ extension PrivatePaykitService { operations: EndpointCleanupOperations( linkedPeers: { try await PaykitSdkService.shared.linkedPeers() }, clearPaymentLists: { try await PaykitSdkService.shared.clearPrivatePaymentLists(to: $0, isSessionCurrent: isSessionCurrent) }, - drainMessages: { await self.drainPendingPrivateMessages(reason: "cleanup", advancing: $0) }, + drainMessages: { _ = await self.drainPendingPrivateMessages(reason: "cleanup", advancing: $0) }, pendingDrainKeys: { await self.pendingPrivateMessageDrainKeys($0) }, syncApp: { try await PublicPaykitService.syncPaykitApp() } ) @@ -688,7 +740,7 @@ extension PrivatePaykitService { } return EndpointPublicationOperations( currentPublicKey: { - await PubkyService.currentPublicKey() + try await PubkyService.currentPublicKey(priority: readPriority) }, ensureLink: { publicKey in try await PaykitSdkService.shared.ensureLinkWithPeer(publicKey, priority: readPriority).state @@ -703,11 +755,13 @@ extension PrivatePaykitService { syncPaymentLists: { updates in try await PaykitSdkService.shared.syncPrivatePaymentListsWithReservations( updates, - clearUnlistedLinkedPeers: false + clearUnlistedLinkedPeers: false, + isBackgroundWork: readPriority == .background ) }, linkedPeers: { try await PaykitSdkService.shared.linkedPeers(priority: readPriority) }, - canPublish: { await self.canPublishPrivateEndpoints(wallet: wallet) } + canPublish: { try await self.privateEndpointPublicationUnavailabilityReason(wallet: wallet, priority: readPriority) == nil }, + isApplicationActive: { await MainActor.run { UIApplication.shared.applicationState == .active } } ) } @@ -716,7 +770,8 @@ extension PrivatePaykitService { reason: String, requireImmediatePublication: Bool, isSessionCurrent: (@MainActor () -> Bool)? = nil, - operations: EndpointPublicationOperations + operations: EndpointPublicationOperations, + expectedIdentity: String? = nil ) async -> Error? { let publicKeys = normalizedSavedContactKeys(publicKeys) guard !publicKeys.isEmpty else { return nil } @@ -725,8 +780,18 @@ extension PrivatePaykitService { if !requireImmediatePublication, await !waitForBackgroundWork(generation: generation) { return nil } - guard let identity = await operations.currentPublicKey() else { - return requireImmediatePublication ? PubkyServiceError.sessionNotActive : nil + let identity: String + if let expectedIdentity { + identity = expectedIdentity + } else { + do { + guard let currentIdentity = try await operations.currentPublicKey() else { + return requireImmediatePublication ? PubkyServiceError.sessionNotActive : nil + } + identity = currentIdentity + } catch { + return error + } } var firstError: Error? @@ -738,6 +803,8 @@ extension PrivatePaykitService { return nil } peerStates = try await Dictionary(uniqueKeysWithValues: operations.linkedPeers().map { ($0.counterparty, $0.state) }) + } catch PaykitPaymentRequestError.operationInProgress { + return PaykitPaymentRequestError.operationInProgress } catch { Logger.warn("Failed to read private Paykit links during \(reason): \(error)", context: "PrivatePaykit") return requireImmediatePublication ? error : nil @@ -762,6 +829,8 @@ extension PrivatePaykitService { linkRetryKeys.append(publicKey) } unavailableLinkRetryAt[publicKey] = nil + } catch PaykitPaymentRequestError.operationInProgress { + return PaykitPaymentRequestError.operationInProgress } catch PaykitError.NotFound { unavailableLinkRetryAt[publicKey] = messageRetryOperations.now().addingTimeInterval(5 * 60) continue @@ -798,7 +867,7 @@ extension PrivatePaykitService { throw PubkyServiceError.sessionNotActive } guard generation == preparationGeneration, - await operations.currentPublicKey() == identity, + try await operations.currentPublicKey() == identity, !UserDefaults.standard.bool(forKey: Self.cleanupPendingKey) else { throw PrivatePaykitError.privateUnavailable } var updates = [PrivatePaymentListReservationUpdateInput]() @@ -831,12 +900,21 @@ extension PrivatePaykitService { let retryKeys = linkRetryKeys + privatePaymentListDeliveryRetryKeys(from: report) schedulePendingPrivateMessageDrainRetries(reason: reason, retryKeys: retryKeys) } + } catch PaykitPaymentRequestError.operationInProgress { + return PaykitPaymentRequestError.operationInProgress } catch { Logger.warn("Failed to sync private Paykit endpoint publications during \(reason): \(error)", context: "PrivatePaykit") firstError = firstError ?? error if !linkRetryKeys.isEmpty { let sessionIsCurrent = await isSessionCurrent?() ?? true - let currentIdentity = await operations.currentPublicKey() + let currentIdentity: String? + do { + currentIdentity = try await operations.currentPublicKey() + } catch PaykitPaymentRequestError.operationInProgress { + return PaykitPaymentRequestError.operationInProgress + } catch { + currentIdentity = nil + } if !Task.isCancelled, sessionIsCurrent, generation == preparationGeneration, currentIdentity == identity, !UserDefaults.standard.bool(forKey: Self.cleanupPendingKey) { @@ -850,42 +928,44 @@ extension PrivatePaykitService { return requireImmediatePublication ? firstError : nil } - private func prepareRelevantPrivateLinksIfAvailable(_ publicKeys: [String], reason: String, isBackgroundWork: Bool = false) async { + private func prepareRelevantPrivateLinksIfAvailable( + _ publicKeys: [String], + reason: String, + isBackgroundWork: Bool = false, + currentPublicKey: () async throws -> String? = { await PubkyService.currentPublicKey() } + ) async throws { let generation = preparationGeneration if isBackgroundWork, await !waitForBackgroundWork(generation: generation) { return } - guard !UserDefaults.standard.bool(forKey: Self.cleanupPendingKey), await canUsePrivateLinks(), generation == preparationGeneration, - !Task.isCancelled else { return } + guard !UserDefaults.standard.bool(forKey: Self.cleanupPendingKey), PaykitFeatureFlags.isUIEnabled, + let ownPublicKey = try await currentPublicKey(), PubkyProfileManager.hasLocalSecretKey(for: ownPublicKey), + generation == preparationGeneration, !Task.isCancelled else { return } - await drainAndSchedulePrivateLinkRetries( + try await drainAndSchedulePrivateLinkRetries( reason: reason, - retryKeys: normalizedSavedContactKeys(publicKeys), + retryKeys: normalizedSavedContactKeys(publicKeys).filter { knownSavedContactKeys.contains($0) }, isBackgroundWork: isBackgroundWork ) } - private func canUsePrivateLinks() async -> Bool { - guard PaykitFeatureFlags.isUIEnabled, - let ownPublicKey = await PubkyService.currentPublicKey() - else { return false } - - return PubkyProfileManager.hasLocalSecretKey(for: ownPublicKey) - } - - private func drainAndSchedulePrivateLinkRetries(reason: String, retryKeys: [String], isBackgroundWork: Bool) async { + private func drainAndSchedulePrivateLinkRetries(reason: String, retryKeys: [String], isBackgroundWork: Bool) async throws { let generation = preparationGeneration let priority: PaykitSdkOperationLock.Priority = isBackgroundWork ? .background : .ordered let retryKeys = Array(Set(retryKeys).subtracting(pendingMessageDrainRetryKeys)) guard !retryKeys.isEmpty else { return } - let drainKeys = await pendingPrivateMessageDrainKeys(retryKeys, retryMissingPeers: true, priority: priority) + let operations = messageRetryOperations.drain(priority) + let drainKeys = await pendingPrivateMessageDrainKeys(retryKeys, retryMissingPeers: true, priority: priority, operations: operations) guard generation == preparationGeneration, !Task.isCancelled, !drainKeys.isEmpty else { return } - await drainPendingPrivateMessages( - reason: reason, advancing: Array(drainKeys), isBackgroundWork: isBackgroundWork, operations: .live(readPriority: priority) - ) - let pendingRetryKeys = await pendingPrivateMessageDrainKeys(Array(drainKeys), priority: priority) + if let deferral = await drainPendingPrivateMessages( + reason: reason, advancing: Array(drainKeys), isBackgroundWork: isBackgroundWork, operations: operations + ) { + throw deferral + } + guard generation == preparationGeneration, !Task.isCancelled else { return } + let pendingRetryKeys = await pendingPrivateMessageDrainKeys(Array(drainKeys), priority: priority, operations: operations) guard generation == preparationGeneration, !Task.isCancelled else { return } if !pendingRetryKeys.isEmpty { schedulePendingPrivateMessageDrainRetries(reason: reason, retryKeys: Array(pendingRetryKeys)) @@ -896,6 +976,7 @@ extension PrivatePaykitService { normalizedSavedContactKeys((report.queued + report.cleared).map(\.counterparty) + report.failedToDeliver.map(\.counterparty)) } + @discardableResult func drainPendingPrivateMessages( reason: String, advancing retryKeys: [String], @@ -904,16 +985,16 @@ extension PrivatePaykitService { schedulingSnapshot: PrivateMessageSchedulingSnapshot? = nil, isCurrent: () async -> Bool = { true }, onReceived: (Set) -> Void = { _ in } - ) async { + ) async -> PaykitPaymentRequestError? { var retryKeys = Set(retryKeys.map { PubkyPublicKeyFormat.normalized($0) ?? $0 }) - guard !retryKeys.isEmpty, !Task.isCancelled else { return } + guard !retryKeys.isEmpty, !Task.isCancelled else { return nil } let generation = preparationGeneration var receivedKeys = Set() do { if isBackgroundWork, await !waitForBackgroundWork(generation: generation) { - return + return nil } - guard await isCurrent() else { return } + guard await isCurrent() else { return nil } let peers: [LinkedPeerRecord] = if let schedulingSnapshot, isSchedulingSnapshotCurrent(schedulingSnapshot) { schedulingSnapshot.peers } else { @@ -927,9 +1008,9 @@ extension PrivatePaykitService { }) for retryKey in retryKeys.sorted() { if isBackgroundWork, await !waitForBackgroundWork(generation: generation) { - return + return nil } - guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return } + guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return nil } if alreadyLinkedKeys.contains(retryKey) { continue } @@ -942,6 +1023,8 @@ extension PrivatePaykitService { if generation == preparationGeneration, !Task.isCancelled { unavailableLinkRetryAt[retryKey] = messageRetryOperations.now().addingTimeInterval(5 * 60) } + } catch PaykitPaymentRequestError.operationInProgress { + throw PaykitPaymentRequestError.operationInProgress } catch { Logger.warn( "Failed to advance private Paykit link for \(PubkyPublicKeyFormat.redacted(retryKey)) during \(reason): \(error)", @@ -950,17 +1033,19 @@ extension PrivatePaykitService { } } if isBackgroundWork, await !waitForBackgroundWork(generation: generation) { - return + return nil } - guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return } + guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return nil } let pendingKeys = try await retryKeys.intersection(operations.pendingOutbound().map { PubkyPublicKeyFormat.normalized($0) ?? $0 }) for publicKey in pendingKeys.sorted() { if isBackgroundWork, await !waitForBackgroundWork(generation: generation) { - return + return nil } - guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return } + guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return nil } do { try await operations.processPending(publicKey) + } catch PaykitPaymentRequestError.operationInProgress { + throw PaykitPaymentRequestError.operationInProgress } catch { Logger.warn( "Failed to send private Paykit messages during \(reason): \(PaykitResolutionFailureDiagnostics.reason(for: error))", @@ -969,20 +1054,22 @@ extension PrivatePaykitService { } } if isBackgroundWork, await !waitForBackgroundWork(generation: generation) { - return + return nil } - guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return } + guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return nil } let linkedKeys = try await Set(operations.linkedPeers().filter { $0.state == .linked }.map { PubkyPublicKeyFormat.normalized($0.counterparty) ?? $0.counterparty }) for publicKey in retryKeys.intersection(linkedKeys).sorted() { if isBackgroundWork, await !waitForBackgroundWork(generation: generation) { - return + return nil } - guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return } + guard generation == preparationGeneration, !Task.isCancelled, await isCurrent() else { return nil } do { try await operations.receive(publicKey) receivedKeys.insert(publicKey) + } catch PaykitPaymentRequestError.operationInProgress { + throw PaykitPaymentRequestError.operationInProgress } catch { Logger.warn( "Failed to receive private Paykit messages during \(reason): \(PaykitResolutionFailureDiagnostics.reason(for: error))", @@ -990,10 +1077,14 @@ extension PrivatePaykitService { ) } } + } catch PaykitPaymentRequestError.operationInProgress { + onReceived(receivedKeys) + return .operationInProgress } catch { Logger.warn("Failed to process pending private Paykit messages during \(reason): \(error)", context: "PrivatePaykit") } onReceived(receivedKeys) + return nil } private func withLinkPreparation(_ publicKey: String, operation: () async throws -> T) async throws -> T? { @@ -1077,11 +1168,13 @@ extension PrivatePaykitService { while !Task.isCancelled, retryGeneration == pendingMessageDrainRetryGeneration { guard await waitForBackgroundWork(generation: preparationGeneration) else { break } guard retryGeneration == pendingMessageDrainRetryGeneration else { return } + let shouldDeferBackground = await messageRetryOperations.shouldDeferRefresh() + guard !Task.isCancelled, retryGeneration == pendingMessageDrainRetryGeneration else { break } let now = messageRetryOperations.now() let due = pendingMessageDrainRetries.filter { $0.value.nextAttemptAt <= now } let foreground = due.filter { $0.value.priority(at: now) == .interactive } let background = due.filter { $0.value.priority(at: now) == .background } - let candidates = !foreground.isEmpty && (foregroundAttempts < 3 || background.isEmpty) + let candidates = !foreground.isEmpty && (shouldDeferBackground || foregroundAttempts < 3 || background.isEmpty) ? foreground : (background.isEmpty ? pendingMessageDrainRetries : background) guard let next = candidates.min(by: { if $0.value.nextAttemptAt == $1.value.nextAttemptAt { @@ -1089,6 +1182,29 @@ extension PrivatePaykitService { } return $0.value.nextAttemptAt < $1.value.nextAttemptAt }) else { break } + if next.value.priority(at: now) == .background, shouldDeferBackground { + schedulingSnapshot = nil + let foregroundDeadline = pendingMessageDrainRetries.values + .filter { $0.priority(at: now) == .interactive } + .map(\.nextAttemptAt).min() + let sleep = messageRetryOperations.sleep + let idle = Task { + await withTaskGroup(of: Void.self) { group in + group.addTask { _ = try? await PaykitPaymentActivity.shared.waitUntilIdle(includingReviews: true) } + if let foregroundDeadline { + let delay = max(0, foregroundDeadline.timeIntervalSince(now)) + group.addTask { _ = try? await sleep(UInt64(delay * 1_000_000_000)) } + } + await group.next() + group.cancelAll() + } + } + pendingMessageDrainRetrySleep = idle + await idle.value + guard retryGeneration == pendingMessageDrainRetryGeneration else { return } + pendingMessageDrainRetrySleep = nil + continue + } let delay = next.value.nextAttemptAt.timeIntervalSince(now) if delay > 0 { schedulingSnapshot = nil @@ -1122,6 +1238,7 @@ extension PrivatePaykitService { guard let retry = pendingMessageDrainRetries[publicKey] else { return } let generation = preparationGeneration var identityInspectionFailed = false + var wasDeferred = false var linkUnavailable = retry.isMissing var inspectedSchedulingGeneration: Int? defer { @@ -1129,7 +1246,7 @@ extension PrivatePaykitService { pendingMessageDrainRetries[publicKey]?.foregroundUntil == retry.foregroundUntil { pendingMessageDrainRetries[publicKey]?.isMissing = linkUnavailable - if identityInspectionFailed { + if identityInspectionFailed, !wasDeferred { pendingMessageDrainRetries[publicKey]?.completeAttempt(at: messageRetryOperations.now()) } } @@ -1144,6 +1261,10 @@ extension PrivatePaykitService { guard try await self.messageRetryOperations.currentPublicKey(priority) == identity else { return false } } catch { identityInspectionFailed = true + if case PaykitPaymentRequestError.operationInProgress = error { + wasDeferred = true + return false + } Logger.warn( "Failed to inspect private Paykit identity during \(reason): \(PaykitResolutionFailureDiagnostics.reason(for: error))", context: "PrivatePaykit" @@ -1196,7 +1317,12 @@ extension PrivatePaykitService { }, schedulingSnapshot: { defer { inspectedSchedulingGeneration = nil } - return try await self.retryDrainOperations(for: publicKey).schedulingSnapshot() + do { + return try await self.retryDrainOperations(for: publicKey).schedulingSnapshot() + } catch PaykitPaymentRequestError.operationInProgress { + wasDeferred = true + throw PaykitPaymentRequestError.operationInProgress + } }, processPending: { defer { inspectedSchedulingGeneration = nil } @@ -1221,18 +1347,23 @@ extension PrivatePaykitService { } drainKeys = schedulingSnapshot?.drainKeys([publicKey]) ?? [publicKey] } - guard await isCurrent() else { return } + guard !wasDeferred, await isCurrent() else { return } var received = Set() - await drainPendingPrivateMessages( + let deferral = await drainPendingPrivateMessages( reason: reason, advancing: Array(drainKeys), isBackgroundWork: true, operations: operations, schedulingSnapshot: schedulingSnapshot, isCurrent: isCurrent, onReceived: { received = $0 } ) guard await waitForBackgroundWork(generation: generation), await isCurrent() else { return } + if deferral != nil { + schedulingSnapshot = nil + return + } // Post-mutation reads stay fresh; only the next due peer's scheduling can reuse them. schedulingSnapshot = await readPrivateMessageSchedulingSnapshot( priority: .background, operations: operations, isCurrent: isCurrent ) + guard !wasDeferred else { return } schedulingSnapshot?.expectedIdentity = retry.expectedIdentity if schedulingSnapshot?.peers.contains(where: { $0.state == .linked && PubkyPublicKeyFormat.normalized($0.counterparty) == publicKey @@ -1312,6 +1443,8 @@ extension PrivatePaykitService { pendingOutbound: Set(snapshot.pendingOutboundCounterparties.compactMap(PubkyPublicKeyFormat.normalized)), preparationGeneration: generation, schedulingGeneration: schedulingGeneration ) + } catch PaykitPaymentRequestError.operationInProgress { + return nil } catch { Logger.warn("Failed to inspect private Paykit scheduling state: \(error)", context: "PrivatePaykit") return nil diff --git a/Bitkit/Services/PrivatePaykitService+Invoices.swift b/Bitkit/Services/PrivatePaykitService+Invoices.swift index 733d1d671..0beb64c7e 100644 --- a/Bitkit/Services/PrivatePaykitService+Invoices.swift +++ b/Bitkit/Services/PrivatePaykitService+Invoices.swift @@ -88,12 +88,23 @@ extension PrivatePaykitService { @MainActor func privateEndpointPublicationUnavailabilityReason(wallet: WalletViewModel) async -> String? { + do { + return try await privateEndpointPublicationUnavailabilityReason(wallet: wallet, priority: .ordered) + } catch { + return "the Pubky session is not active" + } + } + + @MainActor + func privateEndpointPublicationUnavailabilityReason( + wallet: WalletViewModel, priority: PaykitSdkOperationLock.Priority + ) async throws -> String? { guard PaykitFeatureFlags.isUIEnabled else { return "Paykit UI is disabled" } guard UserDefaults.standard.bool(forKey: Self.publishingEnabledKey) else { return "private publication is disabled" } guard UIApplication.shared.applicationState == .active else { return "the app is not active" } guard wallet.walletExists == true else { return "the wallet is unavailable" } guard wallet.nodeLifecycleState == .running else { return "the Lightning node is not running" } - guard let ownPublicKey = await PubkyService.currentPublicKey() else { return "the Pubky session is not active" } + guard let ownPublicKey = try await PubkyService.currentPublicKey(priority: priority) else { return "the Pubky session is not active" } guard PubkyProfileManager.hasLocalSecretKey(for: ownPublicKey) else { return "the local Pubky secret key is unavailable" } return nil diff --git a/Bitkit/Services/PrivatePaykitService+Payments.swift b/Bitkit/Services/PrivatePaykitService+Payments.swift index a60f1ff29..9e7814989 100644 --- a/Bitkit/Services/PrivatePaykitService+Payments.swift +++ b/Bitkit/Services/PrivatePaykitService+Payments.swift @@ -48,7 +48,6 @@ extension PrivatePaykitService { if prePaymentPublicationKeys.insert(normalizedKey).inserted { Task { defer { prePaymentPublicationKeys.remove(normalizedKey) } - guard await canPublishPrivateEndpoints(wallet: wallet) else { return } _ = await refreshSavedContactEndpointsReturningError( for: [normalizedKey], wallet: wallet, diff --git a/Bitkit/Services/PubkyService.swift b/Bitkit/Services/PubkyService.swift index 3813f6b70..e8c51278b 100644 --- a/Bitkit/Services/PubkyService.swift +++ b/Bitkit/Services/PubkyService.swift @@ -70,6 +70,16 @@ enum PubkyService { try? await PaykitSdkService.shared.currentPublicKey() } + static func currentPublicKey(priority: PaykitSdkOperationLock.Priority) async throws -> String? { + do { + return try await PaykitSdkService.shared.currentPublicKey(priority: priority) + } catch PaykitPaymentRequestError.operationInProgress { + throw PaykitPaymentRequestError.operationInProgress + } catch { + return nil + } + } + // MARK: - Auth Approval (Bitkit as authenticator) /// Parse a pubkyauth:// URL to extract details for UI display. @@ -614,12 +624,13 @@ actor PaykitSdkService { } } - func currentPublicKey() async throws -> String? { - try await withSdk { sdk in + func currentPublicKey(priority: PaykitSdkOperationLock.Priority = .ordered) async throws -> String? { + try await withSdk(priority: priority, beforeOperation: { try await self.checkRefreshAdmission(priority) }) { sdk in if let status = try await sdk.identityStatus(), let publicKey = status.publicKey { return publicKey } + try await checkRefreshAdmission(priority) guard let publicKey = try await sdk.initialize().publicKey else { return nil } @@ -633,8 +644,11 @@ actor PaykitSdkService { } func identityStatus(priority: PaykitSdkOperationLock.Priority) async throws -> IdentityStatus? { - try await withSdk(priority: priority) { sdk in - try await readRetryingSharedStateBusy { try await sdk.identityStatus() } + try await withSdk(priority: priority, beforeOperation: { try await self.checkRefreshAdmission(priority) }) { sdk in + try await readRetryingSharedStateBusy { + try await self.checkRefreshAdmission(priority) + return try await sdk.identityStatus() + } } } @@ -983,10 +997,18 @@ actor PaykitSdkService { func syncPrivatePaymentListsWithReservations( _ updates: [PrivatePaymentListReservationUpdateInput], - clearUnlistedLinkedPeers: Bool + clearUnlistedLinkedPeers: Bool, + isBackgroundWork: Bool = false ) async throws -> PrivatePaymentListDeliveryReport { let withdraws = clearUnlistedLinkedPeers || updates.contains { $0.reservations.isEmpty } - return try await withStateRevisionTracking(priority: withdraws ? .ordered : .background) { sdk in + return try await withStateRevisionTracking( + priority: withdraws ? .ordered : .background, + beforeOperation: { + if isBackgroundWork, !withdraws { + try await self.checkRefreshAdmission(.background) + } + } + ) { sdk in try await sdk.syncPrivatePaymentListsWithReservationsAndProcessOutbound( updates: updates, clearUnlistedLinkedPeers: clearUnlistedLinkedPeers @@ -999,7 +1021,7 @@ actor PaykitSdkService { maxAdvanceSteps: UInt32 = 1, priority: PaykitSdkOperationLock.Priority = .ordered ) async throws -> LinkedPeerHandshakeReport { - try await withStateRevisionTracking(priority: priority) { sdk in + try await withStateRevisionTracking(priority: priority, beforeOperation: { try await self.checkRefreshAdmission(priority) }) { sdk in try await sdk.ensureLinkWithPeer(counterparty: counterparty, maxAdvanceSteps: maxAdvanceSteps) } } @@ -1371,21 +1393,27 @@ actor PaykitSdkService { } func linkedPeers(priority: PaykitSdkOperationLock.Priority) async throws -> [LinkedPeerRecord] { - try await withSdk(priority: priority) { sdk in - try await readRetryingSharedStateBusy { try await sdk.linkedPeers() } + try await withSdk(priority: priority, beforeOperation: { try await self.checkRefreshAdmission(priority) }) { sdk in + try await readRetryingSharedStateBusy { + try await self.checkRefreshAdmission(priority) + return try await sdk.linkedPeers() + } } } func privateMessageSchedulingSnapshot( priority: PaykitSdkOperationLock.Priority = .ordered ) async throws -> Paykit.PrivateMessageSchedulingSnapshot { - try await withSdk(priority: priority) { sdk in - try await readRetryingSharedStateBusy { try await sdk.privateMessageSchedulingSnapshot() } + try await withSdk(priority: priority, beforeOperation: { try await self.checkRefreshAdmission(priority) }) { sdk in + try await readRetryingSharedStateBusy { + try await self.checkRefreshAdmission(priority) + return try await sdk.privateMessageSchedulingSnapshot() + } } } func pendingOutboundPrivateCounterparties(priority: PaykitSdkOperationLock.Priority = .ordered) async throws -> [String] { - try await withSdk(priority: priority) { sdk in + try await withSdk(priority: priority, beforeOperation: { try await self.checkRefreshAdmission(priority) }) { sdk in try await sdk.pendingOutboundPrivateCounterparties() } } diff --git a/Bitkit/ViewModels/NavigationViewModel.swift b/Bitkit/ViewModels/NavigationViewModel.swift index 54a6f2f3b..0c5d5e457 100644 --- a/Bitkit/ViewModels/NavigationViewModel.swift +++ b/Bitkit/ViewModels/NavigationViewModel.swift @@ -211,6 +211,19 @@ class NavigationViewModel: ObservableObject { path.removeLast() } + func returnToConnectionsAfterClosing(channelId: String, from expectedPath: [Route]) { + guard path == expectedPath, path.count >= 2, + case let .connectionDetail(detailChannelId) = path[path.count - 2], detailChannelId == channelId, + case let .closeConnection(channel) = path[path.count - 1], channel.channelIdString == channelId + else { return } + + var remainingPath = Array(path.dropLast(2)) + if remainingPath.last != .connections { + remainingPath.append(.connections) + } + path = remainingPath + } + func returnToContactsAfterRemoving(publicKey: String) { let remainingPath = path.filter { route in switch route { diff --git a/Bitkit/Views/Settings/Advanced/CloseConnectionConfirmation.swift b/Bitkit/Views/Settings/Advanced/CloseConnectionConfirmation.swift index f6071d3ab..14f5d5320 100644 --- a/Bitkit/Views/Settings/Advanced/CloseConnectionConfirmation.swift +++ b/Bitkit/Views/Settings/Advanced/CloseConnectionConfirmation.swift @@ -7,6 +7,7 @@ struct CloseConnectionConfirmation: View { @EnvironmentObject var app: AppViewModel @EnvironmentObject var transfer: TransferViewModel @EnvironmentObject var sheets: SheetViewModel + @EnvironmentObject var navigation: NavigationViewModel @Environment(\.dismiss) private var dismiss @State private var isClosing = false @@ -46,7 +47,9 @@ struct CloseConnectionConfirmation: View { .bottomSafeAreaPadding() } + @MainActor private func closeChannel() async { + let expectedPath = navigation.path isClosing = true do { @@ -54,11 +57,9 @@ struct CloseConnectionConfirmation: View { let failedChannels = try await transfer.closeChannels(channels: [channel]) if failedChannels.isEmpty { - // Success - dismiss this view and show success toast DispatchQueue.main.async { - dismiss() + navigation.returnToConnectionsAfterClosing(channelId: channel.channelIdString, from: expectedPath) - // Show success toast app.toast( type: .success, title: t("lightning__close_success_title"), @@ -114,6 +115,7 @@ struct CloseConnectionConfirmation: View { .environmentObject(TransferViewModel()) .environmentObject(AppViewModel()) .environmentObject(SheetViewModel()) + .environmentObject(NavigationViewModel()) } .preferredColorScheme(.dark) } diff --git a/BitkitTests/BackupForegroundRecoveryTests.swift b/BitkitTests/BackupForegroundRecoveryTests.swift index 4bed741bc..7b4a37aba 100644 --- a/BitkitTests/BackupForegroundRecoveryTests.swift +++ b/BitkitTests/BackupForegroundRecoveryTests.swift @@ -140,6 +140,33 @@ final class BackupForegroundRecoveryTests: XCTestCase { await service.stopObservingBackups() } + func testObservationRestartRetainsAndRecoversPendingBackup() async throws { + let fixture = try Fixture(categories: [.settings]) + defer { fixture.removeDefaults() } + let uploads = UploadRecorder() + let service = fixture.service(uploadBackup: { key, _ in await uploads.record(key) }) + + for _ in 0 ..< 8 { + await startAfterPendingJobsSettle(service, categories: fixture.categories) + await service.stopObservingBackups() + + let status = service.getBackupStatus(category: .settings) + XCTAssertTrue(status.isRequired) + XCTAssertEqual(status.required, fixture.required) + XCTAssertFalse(status.running) + } + + await startAfterPendingJobsSettle(service, categories: fixture.categories) + await service.setAppActive(true)?.value + + let recorded = await uploads.counts + XCTAssertEqual(recorded[BackupCategory.settings.rawValue], 1) + XCTAssertFalse(service.getBackupStatus(category: .settings).isRequired) + XCTAssertFalse(service.getBackupStatus(category: .settings).running) + service.setAppActive(false) + await service.stopObservingBackups() + } + func testStoppingRecoveryRetainsPendingBackupAndDoesNotUpload() async throws { let fixture = try Fixture(categories: [.settings]) defer { fixture.removeDefaults() } diff --git a/BitkitTests/ConnectionNavigationTests.swift b/BitkitTests/ConnectionNavigationTests.swift new file mode 100644 index 000000000..a22d13a80 --- /dev/null +++ b/BitkitTests/ConnectionNavigationTests.swift @@ -0,0 +1,43 @@ +@testable import Bitkit +import Combine +import LDKNode +import XCTest + +@MainActor +final class ConnectionNavigationTests: XCTestCase { + func testSuccessfulCloseReturnsToConnectionsInOneUpdate() { + let channel = ChannelDetails.mock() + let navigation = NavigationViewModel() + + for prefix: [Route] in [[.connections], [.activityList], []] { + let expectedPath = prefix + [.connectionDetail(channelId: channel.channelIdString), .closeConnection(channel: channel)] + navigation.path = expectedPath + var updates: [[Route]] = [] + let subscription = navigation.$path.dropFirst().sink { updates.append($0) } + + navigation.returnToConnectionsAfterClosing(channelId: channel.channelIdString, from: expectedPath) + + let destination = prefix.last == .connections ? prefix : prefix + [.connections] + XCTAssertEqual(navigation.path, destination) + XCTAssertEqual(updates, [destination]) + subscription.cancel() + } + } + + func testCloseCompletionPreservesChangedOrUnrelatedNavigation() { + let channel = ChannelDetails.mock() + let original: [Route] = [.connections, .connectionDetail(channelId: channel.channelIdString), .closeConnection(channel: channel)] + let mismatchedDetail: [Route] = [.connections, .connectionDetail(channelId: "another-channel"), .closeConnection(channel: channel)] + let navigation = NavigationViewModel() + + for (path, expectedPath, channelId) in [ + (original + [.activityList], original, channel.channelIdString), + (original, original, "another-channel"), + (mismatchedDetail, mismatchedDetail, channel.channelIdString), + ] { + navigation.path = path + navigation.returnToConnectionsAfterClosing(channelId: channelId, from: expectedPath) + XCTAssertEqual(navigation.path, path) + } + } +} diff --git a/BitkitTests/PaykitSdkIdentityCheckTests.swift b/BitkitTests/PaykitSdkIdentityCheckTests.swift index fcd762027..1c220867f 100644 --- a/BitkitTests/PaykitSdkIdentityCheckTests.swift +++ b/BitkitTests/PaykitSdkIdentityCheckTests.swift @@ -86,6 +86,35 @@ final class PaykitSdkIdentityCheckTests: XCTestCase { } } + @MainActor + func testBackgroundReadsRecheckPaymentAdmissionBeforeContentionRetry() async throws { + let operations: [(PaykitSdkService) async throws -> Void] = [ + { _ = try await $0.identityStatus(priority: .background) }, + { _ = try await $0.linkedPeers(priority: .background) }, + { _ = try await $0.privateMessageSchedulingSnapshot(priority: .background) }, + ] + for operation in operations { + let sdk = IdentitySwitchingSdk(noPointer: .init()) + sdk.readErrors = [sharedStateBusy] + sdk.schedulingRead = { + await sdk.reads.record() + throw sharedStateBusy + } + let service = PaykitSdkService(sdkFactory: { sdk }) + let read = Task { try await operation(service) } + try await sdk.reads.waitForEntries(count: 1) + let payment = PaykitPaymentActivity.shared.begin() + defer { PaykitPaymentActivity.shared.end(payment) } + do { + try await read.value + XCTFail("A new payment must prevent the delayed acquisition retry") + } catch PaykitPaymentRequestError.operationInProgress {} + let attempts = await sdk.reads.count + XCTAssertEqual(attempts, 1) + XCTAssertTrue(sdk.writes.isEmpty) + } + } + func testPaymentRequestReadRetainsItsRuntimeThroughContentionAndFiltersRecords() async throws { let sdk = IdentitySwitchingSdk(noPointer: .init()) let freshSdk = IdentitySwitchingSdk(noPointer: .init()) diff --git a/BitkitTests/PaykitSdkOperationLockTests.swift b/BitkitTests/PaykitSdkOperationLockTests.swift index bd1970984..422aa319b 100644 --- a/BitkitTests/PaykitSdkOperationLockTests.swift +++ b/BitkitTests/PaykitSdkOperationLockTests.swift @@ -283,15 +283,50 @@ final class PaykitSdkOperationLockTests: XCTestCase { ) } + private func publicationReservation() -> PrivatePaymentEndpointReservationInput { + .init(reservationId: "reservation", identifier: "bitcoin:onchain", payload: "{}", expiresAt: nil, attribution: [:]) + } + + @MainActor + func testExplicitPublicationAndWithdrawalRemainAvailableDuringPayment() async throws { + let sdk = PublicReadSdk(noPointer: .init()) + let recorder = Recorder() + sdk.intake = { await recorder.record("published") } + let service = PaykitSdkService(sdkFactory: { sdk }) + let payment = PaykitPaymentActivity.shared.begin() + defer { PaykitPaymentActivity.shared.end(payment) } + _ = try await service.syncPrivatePaymentListsWithReservations( + [.init(counterparty: "peer", reservations: [publicationReservation()])], clearUnlistedLinkedPeers: false + ) + _ = try await service.syncPrivatePaymentListsWithReservations( + [.init(counterparty: "peer", reservations: [])], clearUnlistedLinkedPeers: false, isBackgroundWork: true + ) + _ = try await service.syncPrivatePaymentListsWithReservations([], clearUnlistedLinkedPeers: true) + let events = await recorder.events + XCTAssertEqual(events, ["published", "published", "published"]) + } + @MainActor func testPassiveRefreshRejectsPaymentContentionAndLeavesRecoveryAvailable() async throws { - let operations: [(PaykitSdkService, PaykitSdkOperationLock.Priority) async throws -> Void] = [ - { _ = try await $0.receivePrivateMessagesFromLinkedPeers(priority: $1) }, - { _ = try await $0.processPendingPrivateMessages(priority: $1) }, - { _ = try await $0.observeState(priority: $1) }, + let operations: [(readsRevision: Bool, run: (PaykitSdkService, PaykitSdkOperationLock.Priority) async throws -> Void)] = [ + (true, { _ = try await $0.receivePrivateMessagesFromLinkedPeers(priority: $1) }), + (true, { _ = try await $0.processPendingPrivateMessages(priority: $1) }), + (false, { _ = try await $0.observeState(priority: $1) }), + (false, { _ = try await $0.identityStatus(priority: $1) }), + (false, { _ = try await $0.currentPublicKey(priority: $1) }), + (false, { _ = try await $0.linkedPeers(priority: $1) }), + (false, { _ = try await $0.privateMessageSchedulingSnapshot(priority: $1) }), + (false, { _ = try await $0.pendingOutboundPrivateCounterparties(priority: $1) }), + (true, { _ = try await $0.ensureLinkWithPeer("peer", priority: $1) }), + (true, { service, priority in + _ = try await service.syncPrivatePaymentListsWithReservations( + [.init(counterparty: "peer", reservations: [self.publicationReservation()])], + clearUnlistedLinkedPeers: false, isBackgroundWork: priority == .background + ) + }), ] - for (index, operation) in operations.enumerated() { - for queued in index == 2 ? [true] : [true, false] { + for operation in operations { + for queued in operation.readsRevision ? [true, false] : [true] { let (gate, release) = AsyncStream.makeStream() defer { release.finish() } let started = expectation(description: "Refresh waiting before admission") @@ -309,12 +344,18 @@ final class PaykitSdkOperationLockTests: XCTestCase { sdk.backupRevisionRead = pause } sdk.intake = { await recorder.record("refresh") } + sdk.pendingRead = { await recorder.record("refresh") } + sdk.identityRead = { + await recorder.record("refresh") + return IdentityStatus(publicKey: "identity", capability: .privateLinkCapable) + } let service = PaykitSdkService(sdkFactory: { sdk }) + let initialBackupVersion = await service.backupStateVersion let holder = queued ? Task { try await service.contactRecords() } : nil if queued { await fulfillment(of: [started], timeout: 2) } - let refresh = Task { try await operation(service, .background) } + let refresh = Task { try await operation.run(service, .background) } defer { refresh.cancel() } if queued { try await Task.sleep(for: .milliseconds(50)) @@ -331,10 +372,12 @@ final class PaykitSdkOperationLockTests: XCTestCase { } catch PaykitPaymentRequestError.operationInProgress {} let skippedEvents = await recorder.events XCTAssertEqual(skippedEvents, ["waiting"]) + let skippedBackupVersion = await service.backupStateVersion + XCTAssertEqual(skippedBackupVersion, initialBackupVersion) // Recovery must not wait for the payment whose failure it is refreshing. - try await operation(service, .ordered) + try await operation.run(service, .ordered) PaykitPaymentActivity.shared.end(payment) - try await operation(service, .background) + try await operation.run(service, .background) let completedEvents = await recorder.events XCTAssertEqual(completedEvents, ["waiting", "refresh", "refresh"]) } @@ -346,6 +389,10 @@ final class PaykitSdkOperationLockTests: XCTestCase { let recorder = Recorder() let sdk = PublicReadSdk(noPointer: .init()) sdk.intake = { await recorder.record("refresh") } + sdk.identityRead = { + await recorder.record("refresh") + return IdentityStatus(publicKey: "identity", capability: .privateLinkCapable) + } sdk.backupExport = { await recorder.record("backup") } sdk.outboundDelivery = { await recorder.record("delivery") } sdk.paymentMutation = { await recorder.record($0) } @@ -360,6 +407,10 @@ final class PaykitSdkOperationLockTests: XCTestCase { { _ = try await $0.receivePrivateMessagesFromLinkedPeers(priority: .background) }, { _ = try await $0.processPendingPrivateMessages(priority: .background) }, { _ = try await $0.observeState(priority: .background) }, + { _ = try await $0.identityStatus(priority: .background) }, + { _ = try await $0.linkedPeers(priority: .background) }, + { _ = try await $0.privateMessageSchedulingSnapshot(priority: .background) }, + { _ = try await $0.ensureLinkWithPeer("peer", priority: .background) }, ] for operation in operations { do { @@ -953,6 +1004,7 @@ private final class PublicReadSdk: PaykitSdk, @unchecked Sendable { } var pendingOutboundCounterparties: [String] = [] + var pendingRead: () async -> Void = {} override func processOutboundPrivateMessages(counterparty: String) async throws -> OutboundPrivateSendReport { await outboundDelivery() @@ -961,7 +1013,25 @@ private final class PublicReadSdk: PaykitSdk, @unchecked Sendable { } override func pendingOutboundPrivateCounterparties() async throws -> [String] { - pendingOutboundCounterparties + await pendingRead() + return pendingOutboundCounterparties + } + + override func linkedPeers() async throws -> [LinkedPeerRecord] { + await intake() + return [] + } + + override func privateMessageSchedulingSnapshot() async throws -> Paykit.PrivateMessageSchedulingSnapshot { + await intake() + return .init(linkedPeers: [], pendingOutboundCounterparties: []) + } + + override func syncPrivatePaymentListsWithReservationsAndProcessOutbound( + updates _: [PrivatePaymentListReservationUpdateInput], clearUnlistedLinkedPeers _: Bool + ) async throws -> PrivatePaymentListDeliveryReport { + await intake() + return .init(queued: [], cleared: [], failedToQueue: [], failedToDeliver: []) } override func submitPaymentProof( diff --git a/BitkitTests/PrivatePaykitServiceTests.swift b/BitkitTests/PrivatePaykitServiceTests.swift index 4321863a3..14ca3038a 100644 --- a/BitkitTests/PrivatePaykitServiceTests.swift +++ b/BitkitTests/PrivatePaykitServiceTests.swift @@ -345,6 +345,7 @@ final class PrivatePaykitServiceTests: XCTestCase { } func testExplicitRetryBacksOffIdentityInspectionFailuresAndRecoversAfterAdmittedWrite() async { + let inspectionError = PaykitError.Transport(code: "offline", context: "Registry unavailable") PrivatePaykitService.setContactSharingCleanupPending(false) let key = "pubky" + String(repeating: "y", count: 52) let startedAt = Date(timeIntervalSince1970: 100) @@ -374,7 +375,7 @@ final class PrivatePaykitServiceTests: XCTestCase { }, currentPublicKey: { _ in if inspectionFails { - throw PaykitError.Transport(code: "offline", context: "Registry unavailable") + throw inspectionError } return "identity" }, @@ -630,7 +631,7 @@ final class PrivatePaykitServiceTests: XCTestCase { caller.cancel() await caller.value continuation.finish() - await preparation.value + _ = await preparation.value await fulfillment(of: [completed], timeout: 2) XCTAssertEqual(advances, 1) XCTAssertGreaterThan(receives, 0) @@ -798,14 +799,18 @@ final class PrivatePaykitServiceTests: XCTestCase { func testFailedSchedulingReadRetainsRetryIntent() async { PrivatePaykitService.setContactSharingCleanupPending(false) let key = "pubky" + String(repeating: "y", count: 52) - let now = Date(timeIntervalSince1970: 100) + var now = Date(timeIntervalSince1970: 100) + var recovered = false + let (resume, continuation) = AsyncStream.makeStream() + defer { continuation.finish() } let waiting = expectation(description: "Failed inspection remains scheduled") var received = 0 let service = PrivatePaykitService(messageRetryOperations: .init( now: { now }, - sleep: { _ in + sleep: { delay in waiting.fulfill() - try await Task.sleep(for: .seconds(3600)) + for await _ in resume {} + now = now.addingTimeInterval(TimeInterval(delay) / 1_000_000_000) }, currentPublicKey: { _ in "identity" }, drain: { _ in @@ -813,7 +818,10 @@ final class PrivatePaykitServiceTests: XCTestCase { ensureLink: { _ in XCTFail("Already linked") }, pendingOutbound: { [] }, linkedPeers: { [self.drainPeer(key)] }, - schedulingSnapshot: { throw PrivatePaykitError.privateUnavailable }, + schedulingSnapshot: { + guard recovered else { throw PrivatePaykitError.privateUnavailable } + return .init(linkedPeers: [], pendingOutboundCounterparties: []) + }, processPending: { _ in XCTFail("No outbound messages") }, receive: { _ in received += 1 } ) @@ -826,6 +834,12 @@ final class PrivatePaykitServiceTests: XCTestCase { let retries = await service.pendingMessageDrainRetryKeys XCTAssertEqual(retries, [key]) XCTAssertEqual(received, 1) + recovered = true + let task = await service.pendingMessageDrainRetryTask + continuation.finish() + await task?.value + let remaining = await service.pendingMessageDrainRetryKeys + XCTAssertTrue(remaining.isEmpty) await service.invalidateContactPreparation() } @@ -1326,7 +1340,7 @@ final class PrivatePaykitServiceTests: XCTestCase { first.cancel() } continuation.finish() - await first.value + _ = await first.value await service.drainPendingPrivateMessages(reason: "test", advancing: ["peer"], operations: operations) XCTAssertEqual(advances, 2) } @@ -1366,11 +1380,11 @@ final class PrivatePaykitServiceTests: XCTestCase { } await fulfillment(of: [newStarted], timeout: 2) oldContinuation.finish() - await old.value + _ = await old.value await service.drainPendingPrivateMessages(reason: "test", advancing: ["peer"], operations: operations) XCTAssertEqual(advances, 2) newContinuation.finish() - await current.value + _ = await current.value } func testPrivateMessageDrainAdvancesRecoveryDetectedDuringSendOnNextPass() async { @@ -1508,7 +1522,7 @@ final class PrivatePaykitServiceTests: XCTestCase { receive: { _ in XCTFail("Cancelled retry must not receive") } )) } - await task.value + _ = await task.value } private func drainPeer(_ publicKey: String, state: LinkedPeerState = .linked) -> LinkedPeerRecord { @@ -2624,6 +2638,729 @@ final class PrivatePaykitServiceTests: XCTestCase { XCTAssertNil(error) } + @MainActor + func testBackgroundPreparationRetainsDeferredPublicationWithoutHoldingItsLock() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + for stage in ["availability", "identity", "peers", "publication", "failedPublicationIdentity"] { + let key = "pubky" + String(repeating: "y", count: 52) + let deferred = expectation(description: "Background \(stage) deferred") + var review: UUID? + var didDefer = false + var publicationFailed = false + var publications = 0 + let deferOnce: (String) async throws -> Void = { currentStage in + if currentStage == stage, !didDefer { + didDefer = true + review = await MainActor.run { PaykitPaymentActivity.shared.beginReview() } + deferred.fulfill() + throw PaykitPaymentRequestError.operationInProgress + } + } + let endpoint = PublicPaykitService.Endpoint( + methodId: .regtestOnchainP2wpkh, value: "bcrt1qendpoint", min: nil, max: nil, rawPayload: "{}" + ) + let operations = PrivatePaykitService.EndpointPublicationOperations( + currentPublicKey: { + try await deferOnce(publicationFailed ? "failedPublicationIdentity" : "identity") + return "identity" + }, + ensureLink: { _ in stage == "failedPublicationIdentity" && !publicationFailed ? .linking : .linked }, + buildEndpoints: { _ in [endpoint] }, + syncPaymentLists: { updates in + try await deferOnce("publication") + if stage == "failedPublicationIdentity", !publicationFailed { + publicationFailed = true + throw PaykitError.Transport(code: "offline", context: "Unavailable homeserver") + } + XCTAssertEqual(updates.map(\.counterparty), [key]) + publications += 1 + return .init(queued: [], cleared: [], failedToQueue: [], failedToDeliver: []) + }, + linkedPeers: { try await deferOnce("peers"); return [] }, + canPublish: { try await deferOnce("availability"); return true } + ) + let service = PrivatePaykitService(publicationOperations: operations) + _ = await service.prepareSavedContacts([key], wallet: WalletViewModel()) + await fulfillment(of: [deferred], timeout: 2) + let task = await service.preparationTask + defer { task?.cancel() } + let token = try XCTUnwrap(review) + defer { PaykitPaymentActivity.shared.endReview(token) } + // Explicit publication takes the same lock while routine preparation remains deferred. + let explicit = await service.syncLocalEndpointPublication( + for: [key], reason: "explicit", requireImmediatePublication: true, operations: operations + ) + XCTAssertNil(explicit) + XCTAssertEqual(publications, 1) + PaykitPaymentActivity.shared.endReview(token) + await task?.value + XCTAssertEqual(publications, 2) + await service.invalidateContactPreparation() + } + } + + @MainActor + func testUnavailablePublicationRetainsBackgroundPreparationAfterIdentityDeferral() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + let key = "pubky" + String(repeating: "y", count: 52) + let deferred = expectation(description: "Unavailable publication defers its identity read") + var review: UUID? + var identityReads = 0 + var canPublish = false + var publications = 0 + let endpoint = PublicPaykitService.Endpoint( + methodId: .regtestOnchainP2wpkh, value: "bcrt1qendpoint", min: nil, max: nil, rawPayload: "{}" + ) + let service = PrivatePaykitService(publicationOperations: .init( + currentPublicKey: { + identityReads += 1 + if identityReads == 1 { + review = await MainActor.run { PaykitPaymentActivity.shared.beginReview() } + deferred.fulfill() + throw PaykitPaymentRequestError.operationInProgress + } + return canPublish ? "identity" : nil + }, + ensureLink: { _ in .linked }, + buildEndpoints: { _ in [endpoint] }, + syncPaymentLists: { updates in + XCTAssertEqual(updates.map(\.counterparty), [key]) + publications += 1 + return .init(queued: [], cleared: [], failedToQueue: [], failedToDeliver: []) + }, + canPublish: { canPublish } + )) + _ = await service.prepareSavedContacts([key], wallet: WalletViewModel()) + await fulfillment(of: [deferred], timeout: 2) + let task = await service.preparationTask + defer { task?.cancel() } + let token = try XCTUnwrap(review) + defer { PaykitPaymentActivity.shared.endReview(token) } + // Explicit setup can inspect availability while automatic preparation waits for the review. + let explicit = await service.refreshSavedContactEndpointsReturningError( + for: [key], wallet: WalletViewModel(), forceRefreshLightning: false, requireImmediatePublication: true + ) + guard case PrivatePaykitError.privateUnavailable? = explicit else { + XCTFail("Explicit unavailable publication must complete during review") + return + } + XCTAssertEqual(identityReads, 2) + XCTAssertEqual(publications, 0) + canPublish = true + PaykitPaymentActivity.shared.endReview(token) + await task?.value + XCTAssertEqual(publications, 1) + await service.invalidateContactPreparation() + } + + @MainActor + func testUnavailablePublicationRetainsDeferredLinkBeforePruningMissingPeer() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + UserDefaults.standard.set(true, forKey: PaykitFeatureFlags.uiEnabledKey) + let savedSecret = try Keychain.load(key: .pubkySecretKey) + addTeardownBlock { + if let savedSecret { + try Keychain.upsert(key: .pubkySecretKey, data: savedSecret) + } else { + try Keychain.delete(key: .pubkySecretKey) + } + } + let secret = String(repeating: "01", count: 32) + try Keychain.upsert(key: .pubkySecretKey, data: Data(secret.utf8)) + let identity = try XCTUnwrap(PubkyPublicKeyFormat.normalized(PubkyService.pubkyPublicKeyFromSecret(secretKeyHex: secret))) + let key = "pubky" + String(repeating: "y", count: 52) + for outcome in ["retry", "session", "removed", "invalidated", "cancelled", "notFound"] { + let rejected = expectation(description: "Link admission rejected during review: \(outcome)") + let (resume, continuation) = AsyncStream.makeStream() + defer { continuation.finish() } + var review: UUID? + var sessionCurrent = true + var availabilityChecks = 0 + var snapshots = 0 + var linkAttempts = 0 + var linked = false + let operations = PrivatePaykitService.PrivateMessageDrainOperations( + ensureLink: { _ in + linkAttempts += 1 + if linkAttempts == 1 { + review = await MainActor.run { PaykitPaymentActivity.shared.beginReview() } + rejected.fulfill() + for await _ in resume {} + if outcome == "notFound" { + throw PaykitError.NotFound(code: "not_found", context: "No App Registry") + } + throw PaykitPaymentRequestError.operationInProgress + } + linked = true + }, + pendingOutbound: { [] }, + linkedPeers: { linked ? [self.drainPeer(key)] : [] }, + schedulingSnapshot: { + snapshots += 1 + if linkAttempts == 1, outcome != "notFound" { + XCTAssertGreaterThan(availabilityChecks, 1, "Deferral must return to preparation before a missing-peer read") + } + return .init(linkedPeers: linked ? [self.drainPeer(key)] : [], pendingOutboundCounterparties: []) + }, + processPending: { _ in XCTFail("No outbound work exists") }, + receive: { XCTAssertEqual($0, key) } + ) + let service = PrivatePaykitService( + publicationOperations: .init( + currentPublicKey: { identity }, ensureLink: { _ in XCTFail("Publication is unavailable"); return .unknown }, + buildEndpoints: { _ in XCTFail("Publication is unavailable"); return [] }, + syncPaymentLists: { _ in XCTFail("Publication is unavailable"); throw PrivatePaykitError.privateUnavailable }, + canPublish: { availabilityChecks += 1; return false } + ), + messageRetryOperations: .init(drain: { priority in + XCTAssertEqual(priority, .background) + return operations + }) + ) + _ = await service.prepareSavedContacts([key], wallet: WalletViewModel(), isSessionCurrent: { sessionCurrent }) + await fulfillment(of: [rejected], timeout: 2) + let task = await service.preparationTask + defer { task?.cancel() } + let token = try XCTUnwrap(review) + defer { PaykitPaymentActivity.shared.endReview(token) } + XCTAssertEqual(snapshots, 1) + switch outcome { + case "session": sessionCurrent = false + case "removed": _ = await service.rememberSavedContacts([], replacing: true) + case "invalidated": await service.invalidateContactPreparation() + case "cancelled": task?.cancel() + default: break + } + // End review before the rejected SDK call unwinds, so a later read could otherwise prune the absent peer. + PaykitPaymentActivity.shared.endReview(token) + continuation.finish() + await task?.value + XCTAssertEqual(linkAttempts, outcome == "retry" ? 2 : 1, outcome) + XCTAssertEqual(linked, outcome == "retry", outcome) + XCTAssertEqual(snapshots, outcome == "retry" ? 3 : (outcome == "notFound" ? 2 : 1), outcome) + let retries = await service.pendingMessageDrainRetryKeys + XCTAssertTrue(retries.isEmpty, outcome) + await service.invalidateContactPreparation() + } + } + + @MainActor + func testDeferredForcedPreparationWaitsForForegroundAfterReviewEnds() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + let key = "pubky" + String(repeating: "y", count: 52) + let deferred = expectation(description: "Publication deferred for payment review") + let inspectedWhilePaused = expectation(description: "Availability must not be inspected while backgrounded") + inspectedWhilePaused.isInverted = true + var review: UUID? + var isPaused = false + var publicationAttempts = 0 + let endpoint = PublicPaykitService.Endpoint( + methodId: .regtestOnchainP2wpkh, value: "bcrt1qendpoint", min: nil, max: nil, rawPayload: "{}" + ) + let service = PrivatePaykitService(publicationOperations: .init( + currentPublicKey: { "identity" }, + ensureLink: { _ in .linked }, + buildEndpoints: { _ in [endpoint] }, + syncPaymentLists: { _ in + publicationAttempts += 1 + if publicationAttempts == 1 { + review = await MainActor.run { PaykitPaymentActivity.shared.beginReview() } + deferred.fulfill() + throw PaykitPaymentRequestError.operationInProgress + } + return .init(queued: [], cleared: [], failedToQueue: [], failedToDeliver: []) + }, + canPublish: { + if isPaused { + inspectedWhilePaused.fulfill() + } + return !isPaused + } + )) + _ = await service.rememberSavedContacts([key], replacing: true) + let wallet = WalletViewModel() + await service.scheduleContactPreparation([key], forceRefreshLightning: true) { keys, forceRefresh in + XCTAssertEqual(keys, [key]) + XCTAssertTrue(forceRefresh) + _ = await service.refreshSavedContactEndpointsReturningError( + for: keys, wallet: wallet, forceRefreshLightning: forceRefresh, + requireImmediatePublication: false + ) + } + await fulfillment(of: [deferred], timeout: 2) + let task = await service.preparationTask + defer { task?.cancel() } + let token = try XCTUnwrap(review) + defer { PaykitPaymentActivity.shared.endReview(token) } + isPaused = true + await service.setBackgroundWorkPaused(true) + PaykitPaymentActivity.shared.endReview(token) + await fulfillment(of: [inspectedWhilePaused], timeout: 0.1) + XCTAssertEqual(publicationAttempts, 1) + isPaused = false + await service.setBackgroundWorkPaused(false) + await task?.value + XCTAssertEqual(publicationAttempts, 2) + await service.invalidateContactPreparation() + } + + @MainActor + func testDeferredForcedPreparationSurvivesReviewEndingBeforeBackgroundPause() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + for change in [ + "resume", + "reviewBeforePreparation", + "resumeBeforeWait", + "identity", + "session", + "removed", + "disabled", + "invalidated", + "cancelled", + ] { + let key = "pubky" + String(repeating: "y", count: 52) + let deferred = expectation(description: "Publication deferred for \(change)") + let waiting = expectation(description: "Inactive publication observed for \(change)") + let finished = expectation(description: "Preparation completes for \(change)") + var review: UUID? + var isActive = true + var sharingEnabled = true + var sessionCurrent = true + var identity = "identity" + var activeReads = 0 + var publicationAttempts = 0 + var service: PrivatePaykitService! + let endpoint = PublicPaykitService.Endpoint( + methodId: .regtestOnchainP2wpkh, value: "bcrt1qendpoint", min: nil, max: nil, rawPayload: "{}" + ) + service = PrivatePaykitService(publicationOperations: .init( + currentPublicKey: { identity }, + ensureLink: { _ in .linked }, + buildEndpoints: { _ in [endpoint] }, + syncPaymentLists: { _ in + publicationAttempts += 1 + if publicationAttempts == 1, change != "reviewBeforePreparation" { + review = await MainActor.run { PaykitPaymentActivity.shared.beginReview() } + deferred.fulfill() + throw PaykitPaymentRequestError.operationInProgress + } + return .init(queued: [], cleared: [], failedToQueue: [], failedToDeliver: []) + }, + canPublish: { isActive && sharingEnabled }, + isApplicationActive: { + activeReads += 1 + if activeReads == 1, change == "resumeBeforeWait" { + isActive = true + await service.setBackgroundWorkPaused(false) + return false + } + if activeReads == 2 { + waiting.fulfill() + } + return isActive + } + )) + _ = await service.rememberSavedContacts([key], replacing: true) + let wallet = WalletViewModel() + if change == "reviewBeforePreparation" { + review = PaykitPaymentActivity.shared.beginReview() + deferred.fulfill() + } + await service.scheduleContactPreparation([key], forceRefreshLightning: true) { keys, forceRefresh in + XCTAssertEqual(keys, [key]) + XCTAssertTrue(forceRefresh) + _ = await service.refreshSavedContactEndpointsReturningError( + for: keys, wallet: wallet, forceRefreshLightning: forceRefresh, + requireImmediatePublication: false, isSessionCurrent: { sessionCurrent } + ) + finished.fulfill() + } + await fulfillment(of: [deferred], timeout: 2) + let task = await service.preparationTask + defer { task?.cancel() } + let token = try XCTUnwrap(review) + defer { PaykitPaymentActivity.shared.endReview(token) } + isActive = false + PaykitPaymentActivity.shared.endReview(token) + await fulfillment(of: [waiting], timeout: 2) + try await service.withPublicationLock {} + switch change { + case "identity": identity = "other" + case "session": sessionCurrent = false + case "removed": _ = await service.rememberSavedContacts([], replacing: true) + case "disabled": sharingEnabled = false + case "invalidated": await service.invalidateContactPreparation() + case "cancelled": task?.cancel() + default: break + } + if change != "resumeBeforeWait", change != "invalidated", change != "cancelled" { + await service.setBackgroundWorkPaused(true) + isActive = true + await service.setBackgroundWorkPaused(false) + } + await fulfillment(of: [finished], timeout: 2) + await task?.value + XCTAssertEqual(publicationAttempts, change.hasPrefix("resume") ? 2 : 1, change) + let waiterCount = await service.backgroundWorkWaiters.count + XCTAssertEqual(waiterCount, 0, change) + await service.invalidateContactPreparation() + } + } + + @MainActor + func testBackgroundDrainRetryDoesNotBurnBackoffDuringPaymentReview() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + let key = "pubky" + String(repeating: "y", count: 52) + var now = Date(timeIntervalSince1970: 100) + var sleeps: [UInt64] = [] + var rejectedAttempts = 0 + let drained = expectation(description: "Retry drains once the review ends") + drained.assertForOverFulfill = false + let reject: () async throws -> Void = { + if PaykitPaymentActivity.shared.shouldDeferRefresh { + throw PaykitPaymentRequestError.operationInProgress + } + } + let service = PrivatePaykitService(messageRetryOperations: .init( + now: { now }, + sleep: { delay in + sleeps.append(delay) + if sleeps.count >= 4 { + try await Task.sleep(for: .seconds(3600)) + } + now = now.addingTimeInterval(TimeInterval(delay) / 1_000_000_000) + }, + currentPublicKey: { _ in "identity" }, + drain: { _ in + .init( + ensureLink: { _ in try await reject() }, + pendingOutbound: { + try await reject() + return [] + }, + linkedPeers: { + do { try await reject() } catch { rejectedAttempts += 1; throw error } + drained.fulfill() + return [self.drainPeer(key)] + }, + processPending: { _ in try await reject() }, + receive: { _ in try await reject() } + ) + }, didLink: { _, _ in } + )) + let review = PaykitPaymentActivity.shared.beginReview() + defer { PaykitPaymentActivity.shared.endReview(review) } + _ = await service.rememberSavedContacts([key], replacing: true) + await service.setTestRetry(.init(nextAttemptAt: now), for: key) + await service.schedulePrivatePaymentRecovery(for: key) + try await Task.sleep(for: .milliseconds(300)) + let duringReview = await service.pendingMessageDrainRetries[key] + XCTAssertLessThanOrEqual( + try XCTUnwrap(duringReview).nextAttemptAt, + now.addingTimeInterval(1), + "rejected \(rejectedAttempts) times, sleeps \(sleeps)" + ) + PaykitPaymentActivity.shared.endReview(review) + await fulfillment(of: [drained], timeout: 2) + await service.invalidateContactPreparation() + } + + @MainActor + func testDeferredDrainRetryPreservesBackoffAndCanBeInterrupted() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + for boundary in ["identity", "snapshot", "link", "receive", "postWriteIdentity", "postWriteSnapshot", "interactive", "invalidated"] { + let key = "pubky" + String(repeating: "y", count: 52) + let now = Date(timeIntervalSince1970: 100) + let deferred = expectation(description: "Admission deferred at \(boundary)") + let completed = expectation(description: "Retry completes at \(boundary)") + var review: UUID? + var didDefer = false + var didReceive = false + var didProcess = false + var linked = boundary != "link" && boundary != "interactive" && boundary != "invalidated" + var priorities: [PaykitSdkOperationLock.Priority] = [] + let deferAt: (String) async throws -> Void = { stage in + let target: String = switch boundary { + case "interactive", "invalidated": "link" + case "postWriteIdentity": didProcess ? "identity" : "postWriteIdentity" + case "postWriteSnapshot": didReceive ? "snapshot" : "postWriteSnapshot" + default: boundary + } + if stage == target, !didDefer { + didDefer = true + review = PaykitPaymentActivity.shared.beginReview() + deferred.fulfill() + throw PaykitPaymentRequestError.operationInProgress + } + } + let service = PrivatePaykitService(messageRetryOperations: .init( + now: { now }, + sleep: { _ in + XCTFail("Payment deferral must not schedule network backoff: \(boundary)") + try await Task.sleep(for: .seconds(3600)) + }, + currentPublicKey: { _ in try await deferAt("identity"); return "identity" }, + drain: { priority in + priorities.append(priority) + let peers = { [self.drainPeer(key, state: linked ? .linked : .linking)] } + let pending = { didReceive || boundary == "identity" ? [] : [key] } + return .init( + ensureLink: { _ in try await deferAt("link"); linked = true }, + pendingOutbound: { pending() }, + linkedPeers: { peers() }, + schedulingSnapshot: { + try await deferAt("snapshot") + return .init(linkedPeers: peers(), pendingOutboundCounterparties: pending()) + }, + processPending: { _ in didProcess = true }, + receive: { _ in + try await deferAt("receive") + didReceive = true + } + ) + }, didLink: { _, _ in } + )) + _ = await service.rememberSavedContacts([key], replacing: true) + await service.setTestRetry( + .init(nextAttemptAt: now, expectedIdentity: boundary == "identity" || boundary == "postWriteIdentity" ? "identity" : nil), + for: key + ) + await service.schedulePrivatePaymentRecovery(for: key) + let task = await service.pendingMessageDrainRetryTask + await fulfillment(of: [deferred], timeout: 2) + let token = try XCTUnwrap(review) + defer { PaykitPaymentActivity.shared.endReview(token) } + for _ in 0 ..< 1000 { + if await service.pendingMessageDrainRetrySleep != nil { + break + } + await Task.yield() + } + let idleWait = await service.pendingMessageDrainRetrySleep + XCTAssertNotNil(idleWait, boundary) + let retry = await service.pendingMessageDrainRetries[key] + XCTAssertEqual(retry?.retryIndex, 0, boundary) + XCTAssertEqual(retry?.nextAttemptAt, now, boundary) + let observer = Task { await task?.value; completed.fulfill() } + defer { observer.cancel() } + switch boundary { + case "interactive": await service.scheduleExplicitContactLink(publicKey: key, identity: "identity") + case "invalidated": await service.invalidateContactPreparation() + default: PaykitPaymentActivity.shared.endReview(token) + } + await fulfillment(of: [completed], timeout: 2) + XCTAssertEqual(didReceive, boundary != "invalidated", boundary) + if boundary == "interactive" { + XCTAssertTrue(PaykitPaymentActivity.shared.shouldDeferRefresh) + XCTAssertEqual(priorities.last, .interactive) + } + await service.invalidateContactPreparation() + } + } + + @MainActor + func testCancelledDrainWorkerPreservesRetryAfterAdmissionSuspends() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + for invalidate in [false, true] { + let key = "pubky" + String(repeating: "y", count: 52) + let now = Date(timeIntervalSince1970: 100) + let firstAdmission = expectation(description: "First worker awaits admission") + let nextAdmission = expectation(description: "Next worker awaits admission") + var firstContinuation: CheckedContinuation? + var nextContinuation: CheckedContinuation? + defer { + firstContinuation?.resume(returning: false) + nextContinuation?.resume(returning: false) + } + var admissionReads = 0 + var identity = "original" + var received: [String] = [] + var linked: [String] = [] + let service = PrivatePaykitService(messageRetryOperations: .init( + now: { now }, + sleep: { _ in XCTFail("Linked peer must complete without backoff"); throw CancellationError() }, + shouldDeferRefresh: { + admissionReads += 1 + switch admissionReads { + case 1: + return await withCheckedContinuation { + firstContinuation = $0 + firstAdmission.fulfill() + } + case 2: + return await withCheckedContinuation { + nextContinuation = $0 + nextAdmission.fulfill() + } + default: + return false + } + }, + currentPublicKey: { _ in identity }, + drain: { priority in + XCTAssertEqual(priority, .interactive) + return .init( + ensureLink: { _ in XCTFail("Peer is already linked") }, + pendingOutbound: { [] }, + linkedPeers: { [self.drainPeer(key)] }, + processPending: { _ in XCTFail("No outbound messages") }, + receive: { received.append($0) } + ) + }, + didLink: { identity, _ in linked.append(identity) } + )) + _ = await service.rememberSavedContacts([key], replacing: true) + await service.scheduleExplicitContactLink(publicKey: key, identity: identity) + await fulfillment(of: [firstAdmission], timeout: 2) + let firstWorker = await service.pendingMessageDrainRetryTask + let firstTask = try XCTUnwrap(firstWorker) + if invalidate { + await service.invalidateContactPreparation() + identity = "replacement" + await service.scheduleExplicitContactLink(publicKey: key, identity: identity) + await fulfillment(of: [nextAdmission], timeout: 2) + } else { + firstTask.cancel() + } + let retainedRetry = await service.pendingMessageDrainRetries[key] + XCTAssertNotNil(retainedRetry) + firstContinuation?.resume(returning: false) + firstContinuation = nil + await firstTask.value + let afterCancellation = await service.pendingMessageDrainRetries[key] + XCTAssertEqual(afterCancellation?.id, retainedRetry?.id) + guard let retainedRetry, afterCancellation?.id == retainedRetry.id else { + await service.invalidateContactPreparation() + continue + } + XCTAssertTrue(received.isEmpty) + XCTAssertTrue(linked.isEmpty) + if !invalidate { + let cancelledTask = await service.pendingMessageDrainRetryTask + XCTAssertNil(cancelledTask) + await service.schedulePrivatePaymentRecovery(for: key) + await fulfillment(of: [nextAdmission], timeout: 2) + } + let nextTask = await service.pendingMessageDrainRetryTask + XCTAssertNotNil(nextTask) + nextContinuation?.resume(returning: false) + nextContinuation = nil + await nextTask?.value + XCTAssertEqual(received, [key]) + XCTAssertEqual(linked, [identity]) + let remaining = await service.pendingMessageDrainRetryKeys + XCTAssertTrue(remaining.isEmpty) + await service.invalidateContactPreparation() + } + } + + @MainActor + func testPaymentReviewDoesNotDelayAnotherContactsInteractiveRetryDeadline() async { + PrivatePaykitService.setContactSharingCleanupPending(false) + let backgroundKey = "pubky" + String(repeating: "y", count: 52) + let foregroundKey = "pubky3rsduhcxpw74snwyct86m38c63j3pq8x4ycqikxg64roik8yw5xg" + var now = Date(timeIntervalSince1970: 100) + let deadline = now.addingTimeInterval(1) + let received = expectation(description: "Interactive intake runs at its retry deadline during review") + var delays: [UInt64] = [] + let service = PrivatePaykitService(messageRetryOperations: .init( + now: { now }, + sleep: { delay in + delays.append(delay) + XCTAssertEqual(delay, 1_000_000_000) + now = deadline + }, + currentPublicKey: { priority in XCTAssertEqual(priority, .interactive); return "identity" }, + drain: { priority in + XCTAssertEqual(priority, .interactive) + return .init( + ensureLink: { _ in XCTFail("Peer is already linked") }, + pendingOutbound: { [] }, + linkedPeers: { [self.drainPeer(foregroundKey)] }, + processPending: { _ in XCTFail("No outbound messages") }, + receive: { key in XCTAssertEqual(key, foregroundKey) } + ) + }, + didLink: { identity, key in + XCTAssertEqual(identity, "identity") + XCTAssertEqual(key, foregroundKey) + received.fulfill() + } + )) + let review = PaykitPaymentActivity.shared.beginReview() + defer { PaykitPaymentActivity.shared.endReview(review) } + _ = await service.rememberSavedContacts([backgroundKey, foregroundKey], replacing: true) + await service.setTestRetry(.init(nextAttemptAt: now), for: backgroundKey) + await service.setTestRetry( + .init(nextAttemptAt: deadline, foregroundUntil: now.addingTimeInterval(20), expectedIdentity: "identity"), + for: foregroundKey + ) + await service.schedulePrivatePaymentRecovery(for: backgroundKey) + await fulfillment(of: [received], timeout: 2) + XCTAssertTrue(PaykitPaymentActivity.shared.shouldDeferRefresh) + XCTAssertEqual(delays, [1_000_000_000]) + let background = await service.pendingMessageDrainRetries[backgroundKey] + XCTAssertEqual(background?.retryIndex, 0) + XCTAssertEqual(background?.nextAttemptAt, Date(timeIntervalSince1970: 100)) + let task = await service.pendingMessageDrainRetryTask + await service.invalidateContactPreparation() + await task?.value + } + + @MainActor + func testDeferredPublicationRevalidatesOwnershipAndSavedContacts() async throws { + PrivatePaykitService.setContactSharingCleanupPending(false) + for change in ["identity", "session", "generation", "removed", "disabled", "cancelled"] { + let key = "pubky" + String(repeating: "y", count: 52) + let deferred = expectation(description: "Publication deferred before \(change)") + var review: UUID? + var identity = "identity" + var currentSession = true + var canPublish = true + var attempts = 0 + let endpoint = PublicPaykitService.Endpoint( + methodId: .regtestOnchainP2wpkh, value: "bcrt1qendpoint", min: nil, max: nil, rawPayload: "{}" + ) + let service = PrivatePaykitService(publicationOperations: .init( + currentPublicKey: { identity }, + ensureLink: { _ in .linked }, + buildEndpoints: { _ in [endpoint] }, + syncPaymentLists: { _ in + attempts += 1 + XCTAssertEqual(attempts, 1, "Invalidated publication must never be attempted again") + review = await MainActor.run { PaykitPaymentActivity.shared.beginReview() } + deferred.fulfill() + throw PaykitPaymentRequestError.operationInProgress + }, + canPublish: { canPublish } + )) + _ = await service.prepareSavedContacts([key], wallet: WalletViewModel(), isSessionCurrent: { currentSession }) + await fulfillment(of: [deferred], timeout: 2) + let task = await service.preparationTask + defer { task?.cancel() } + let token = try XCTUnwrap(review) + defer { PaykitPaymentActivity.shared.endReview(token) } + // This also proves the deferred batch released its publication lock. + try await service.withPublicationLock {} + switch change { + case "identity": identity = "other" + case "session": currentSession = false + case "generation": await service.beginProfileDeletion() + case "removed": _ = await service.rememberSavedContacts([], replacing: true) + case "disabled": canPublish = false + case "cancelled": task?.cancel() + default: XCTFail("Unhandled case") + } + if change == "cancelled" { + await task?.value + } + PaykitPaymentActivity.shared.endReview(token) + await task?.value + XCTAssertEqual(attempts, 1) + await service.invalidateContactPreparation() + } + } + func testBackgroundPreparationCoalescesRepeatedRequests() async { let service = PrivatePaykitService() let publicKey = "pubky3rsduhcxpw74snwyct86m38c63j3pq8x4ycqikxg64roik8yw5xg" @@ -2819,7 +3556,7 @@ final class PrivatePaykitServiceTests: XCTestCase { case "invalidate": await service.invalidateContactPreparation() default: task.cancel() } - await task.value + _ = await task.value XCTAssertEqual(events, outcome == "resume" ? order : prefix) let remainingWaiters = await service.backgroundWorkWaiters.count XCTAssertEqual(remainingWaiters, 0) diff --git a/changelog.d/next/1054.fixed.md b/changelog.d/next/1054.fixed.md new file mode 100644 index 000000000..5b81fc0b6 --- /dev/null +++ b/changelog.d/next/1054.fixed.md @@ -0,0 +1 @@ +Pause automatic contact updates during Paykit payment review or submission, resume them safely afterward, prevent crashes when backup observation starts or stops, and return to the connections list after closing a Lightning connection. diff --git a/journeys/transfer/close-connection.xml b/journeys/transfer/close-connection.xml new file mode 100644 index 000000000..a28d60b56 --- /dev/null +++ b/journeys/transfer/close-connection.xml @@ -0,0 +1,18 @@ + + + Verifies the iOS route after a successful cooperative channel close. Use the local E2E + Docker/LND fixture and an existing funded, usable Lightning connection. Start on Lightning + Connections with the connection visible. Creating or funding the fixture is a prerequisite. + + Closing returns directly to the connections list, so reloading the removed connection's + detail cannot replace the success toast with "Connection Not Found". This journey checks + navigation and acknowledgement; it does not establish settlement or final transfer balance. + + + Open the existing connection (identifier "Channel") + Open Close Connection (identifier "CloseConnection") + Confirm Close (identifier "CloseConnectionButton") + Verify "Transfer Initiated" appears and the current screen is Lightning Connections + Verify "Connection Not Found" does not appear after the success toast disappears + +