diff --git a/NotificationServiceExtension/NotificationService.swift b/NotificationServiceExtension/NotificationService.swift index b93d48689e..7273bed422 100644 --- a/NotificationServiceExtension/NotificationService.swift +++ b/NotificationServiceExtension/NotificationService.swift @@ -149,29 +149,19 @@ class NotificationService: UNNotificationServiceExtension { let processingCompletePromise = firstly { self.messageProcessor.processingCompletePromise() }.then(on: .global()) { () -> Promise in - // Wait until all notifications are enqueued. - if DebugFlags.internalLogging { - Logger.info("Waiting on notificationPresenter.") - } - return NotificationPresenter.pendingNotificationsPromise() - }.then(on: .global()) { () -> Promise in - if DebugFlags.internalLogging { - Logger.info("Waiting on acks.") - } - // Wait until all ACKs are enqueued. - return Self.messageFetcherJob.pendingAcksPromise() - }.then(on: .global()) { () -> Promise in - if DebugFlags.internalLogging { - Logger.info("Waiting on outgoing receipt send enqueues.") - } - // Wait until all outgoing receipt sends are enqueued. - return Self.outgoingReceiptManager.pendingSendsPromise() - }.then(on: .global()) { () -> Promise in - if DebugFlags.internalLogging { - Logger.info("Waiting on sends.") - } - // Wait until all outgoing messages are sent. - return Self.messageSender.pendingSendsPromise() + // Wait until all async side effects of + // message processing are complete. + let completionPromises: [Promise] = [ + // Wait until all notifications are posted. + NotificationPresenter.pendingNotificationsPromise(), + // Wait until all ACKs are complete. + Self.messageFetcherJob.pendingAcksPromise(), + // Wait until all outgoing receipt sends are complete. + Self.outgoingReceiptManager.pendingSendsPromise(), + // Wait until all outgoing messages are sent. + Self.messageSender.pendingSendsPromise() + ] + return Promise.when(resolved: completionPromises).asVoid() } processingCompletePromise.timeout(seconds: 20, description: "Message Processing Timeout.") { NotificationServiceError.timeout diff --git a/Pods b/Pods index df04141b24..43b2a1a14e 160000 --- a/Pods +++ b/Pods @@ -1 +1 @@ -Subproject commit df04141b244c85b1de4ab571b3d7b7bc71850288 +Subproject commit 43b2a1a14e5bfb0b6773a04c97df97ec2be88d17 diff --git a/Signal/src/ViewControllers/DebugUI/DebugUIMessages.swift b/Signal/src/ViewControllers/DebugUI/DebugUIMessages.swift index 26dc11d200..5681475333 100644 --- a/Signal/src/ViewControllers/DebugUI/DebugUIMessages.swift +++ b/Signal/src/ViewControllers/DebugUI/DebugUIMessages.swift @@ -36,8 +36,6 @@ public extension DebugUIMessages { switch error { case MessageProcessingError.duplicatePendingEnvelope?: Logger.warn("duplicatePendingEnvelope.") - case MessageProcessingError.duplicateDecryption?: - Logger.warn("duplicateDecryption.") case let otherError?: owsFailDebug("Error: \(otherError)") case nil: diff --git a/SignalMessaging/Notifications/AppNotifications.swift b/SignalMessaging/Notifications/AppNotifications.swift index 69804eb31e..947b16fc26 100644 --- a/SignalMessaging/Notifications/AppNotifications.swift +++ b/SignalMessaging/Notifications/AppNotifications.swift @@ -158,6 +158,8 @@ let kNotificationDelayForRemoteRead: TimeInterval = 20 let kAudioNotificationsThrottleCount = 2 let kAudioNotificationsThrottleInterval: TimeInterval = 5 +typealias NotificationCompletion = () -> Void + protocol NotificationPresenterAdaptee: AnyObject { func registerNotificationSettings() -> Promise @@ -168,7 +170,8 @@ protocol NotificationPresenterAdaptee: AnyObject { threadIdentifier: String?, userInfo: [AnyHashable: Any], interaction: INInteraction?, - sound: OWSSound?) + sound: OWSSound?, + completion: NotificationCompletion?) func notify(category: AppNotificationCategory, title: String?, @@ -177,7 +180,8 @@ protocol NotificationPresenterAdaptee: AnyObject { userInfo: [AnyHashable: Any], interaction: INInteraction?, sound: OWSSound?, - replacingIdentifier: String?) + replacingIdentifier: String?, + completion: NotificationCompletion?) func cancelNotifications(threadId: String) func cancelNotifications(messageId: String) @@ -187,6 +191,8 @@ protocol NotificationPresenterAdaptee: AnyObject { var hasReceivedSyncMessageRecently: Bool { get } } +// MARK: - + extension NotificationPresenterAdaptee { var hasReceivedSyncMessageRecently: Bool { return OWSDeviceManager.shared().hasReceivedSyncMessage(inLastSeconds: 60) @@ -259,7 +265,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { interaction = wrapper } - notificationQueue.async { + notifyAsyncOnMainThread { completion in self.adaptee.notify(category: .incomingCall, title: notificationTitle, body: notificationBody, @@ -267,7 +273,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { userInfo: userInfo, interaction: interaction, sound: nil, - replacingIdentifier: call.localId.uuidString) + replacingIdentifier: call.localId.uuidString, + completion: completion) } } @@ -309,7 +316,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { interaction = wrapper } - notificationQueue.async { + notifyAsyncOnMainThread { completion in let sound = self.requestSound(thread: thread) self.adaptee.notify(category: category, title: notificationTitle, @@ -318,7 +325,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { userInfo: userInfo, interaction: interaction, sound: sound, - replacingIdentifier: call.localId.uuidString) + replacingIdentifier: call.localId.uuidString, + completion: completion) } } @@ -343,7 +351,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { AppNotificationUserInfoKey.threadId: thread.uniqueId ] - notificationQueue.async { + notifyAsyncOnMainThread { completion in let sound = self.requestSound(thread: thread) self.adaptee.notify(category: .missedCallFromNoLongerVerifiedIdentity, title: notificationTitle, @@ -352,7 +360,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { userInfo: userInfo, interaction: nil, sound: sound, - replacingIdentifier: call.localId.uuidString) + replacingIdentifier: call.localId.uuidString, + completion: completion) } } @@ -378,7 +387,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { let category: AppNotificationCategory = (shouldShowActions ? .missedCallWithActions : .missedCallWithoutActions) - notificationQueue.async { + notifyAsyncOnMainThread { completion in let sound = self.requestSound(thread: thread) self.adaptee.notify(category: category, title: notificationTitle, @@ -387,7 +396,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { userInfo: userInfo, interaction: nil, sound: sound, - replacingIdentifier: call.localId.uuidString) + replacingIdentifier: call.localId.uuidString, + completion: completion) } } @@ -526,7 +536,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { interaction = wrapper } - notificationQueue.async { + notifyAsyncOnMainThread { completion in let sound = self.requestSound(thread: thread) self.adaptee.notify(category: category, title: notificationTitle, @@ -534,7 +544,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { threadIdentifier: threadIdentifier, userInfo: userInfo, interaction: interaction, - sound: sound) + sound: sound, + completion: completion) } } @@ -544,16 +555,33 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { private static var notificationQueue: DispatchQueue { notifyOnMainThread ? .main : serialQueue } private var notificationQueue: DispatchQueue { Self.notificationQueue } + private static let pendingTasks = PendingTasks(label: "Notifications") + public static func pendingNotificationsPromise() -> Promise { // This promise blocks on all pending notifications already in flight, // but will not block on new notifications enqueued after this promise // is created. That's intentional to ensure that NotificationService // instances complete in a timely way. - let (promise, future) = Promise.pending() + pendingTasks.pendingTasksPromise() + } + + private func notifyAsyncOnMainThread(_ block: @escaping (@escaping NotificationCompletion) -> Void) { + let pendingTask = Self.pendingTasks.buildPendingTask(label: "Notification") notificationQueue.async { - future.resolve(()) + block { + pendingTask.complete() + } + } + } + + private func notifyInAsyncCompletionOnMainThread(transaction: SDSAnyWriteTransaction, + _ block: @escaping (@escaping NotificationCompletion) -> Void) { + let pendingTask = Self.pendingTasks.buildPendingTask(label: "Notification") + transaction.addAsyncCompletion(queue: notificationQueue) { + block { + pendingTask.complete() + } } - return promise } public func notifyUser(forReaction reaction: OWSReaction, @@ -659,7 +687,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { interaction = wrapper } - notificationQueue.async { + notifyAsyncOnMainThread { completion in let sound = self.requestSound(thread: thread) self.adaptee.notify( category: category, @@ -668,7 +696,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { threadIdentifier: thread.uniqueId, userInfo: userInfo, interaction: interaction, - sound: sound + sound: sound, + completion: completion ) } } @@ -688,7 +717,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { AppNotificationUserInfoKey.threadId: threadId ] - notificationQueue.async { + notifyAsyncOnMainThread { completion in let sound = self.requestSound(thread: thread) self.adaptee.notify(category: .infoOrErrorMessage, title: notificationTitle, @@ -696,7 +725,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { threadIdentifier: nil, // show ungrouped userInfo: userInfo, interaction: nil, - sound: sound) + sound: sound, + completion: completion) } } @@ -711,7 +741,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { comment: "Format string for an error alert notification message. Embes {{ error string }}") let message = String(format: messageFormat, errorString) - notificationQueue.async { + notifyAsyncOnMainThread { completion in self.adaptee.notify( category: .internalError, title: title, @@ -721,7 +751,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { AppNotificationUserInfoKey.defaultAction: AppNotificationAction.submitDebugLogs.rawValue ], interaction: nil, - sound: self.requestGlobalSound()) + sound: self.requestGlobalSound(), + completion: completion) } } @@ -741,7 +772,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { AppNotificationUserInfoKey.defaultAction: AppNotificationAction.showCallLobby.rawValue ] - notificationQueue.async { + notifyAsyncOnMainThread { completion in let sound = self.requestSound(thread: thread) self.adaptee.notify(category: .infoOrErrorMessage, title: notificationTitle, @@ -749,7 +780,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { threadIdentifier: nil, // show ungrouped userInfo: userInfo, interaction: nil, - sound: sound) + sound: sound, + completion: completion) } } @@ -822,7 +854,7 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { interaction = wrapper } - transaction.addAsyncCompletion(queue: notificationQueue) { + notifyInAsyncCompletionOnMainThread(transaction: transaction) { completion in let sound = wantsSound ? self.requestSound(thread: thread) : nil self.adaptee.notify(category: .infoOrErrorMessage, title: notificationTitle, @@ -830,7 +862,8 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { threadIdentifier: threadIdentifier, userInfo: userInfo, interaction: interaction, - sound: sound) + sound: sound, + completion: completion) } } @@ -838,14 +871,15 @@ public class NotificationPresenter: NSObject, NotificationsProtocol { transaction: SDSAnyWriteTransaction) { let notificationBody = errorMessage.previewText(transaction: transaction) - transaction.addAsyncCompletion(queue: notificationQueue) { + notifyInAsyncCompletionOnMainThread(transaction: transaction) { completion in self.adaptee.notify(category: .threadlessErrorMessage, title: nil, body: notificationBody, threadIdentifier: nil, userInfo: [:], interaction: nil, - sound: self.requestGlobalSound()) + sound: self.requestGlobalSound(), + completion: completion) } } diff --git a/SignalMessaging/Notifications/UserNotificationsAdaptee.swift b/SignalMessaging/Notifications/UserNotificationsAdaptee.swift index 09c9d15f98..3ab1502bea 100644 --- a/SignalMessaging/Notifications/UserNotificationsAdaptee.swift +++ b/SignalMessaging/Notifications/UserNotificationsAdaptee.swift @@ -136,6 +136,8 @@ public class UserNotificationConfig { } +// MARK: - + class UserNotificationPresenterAdaptee: NSObject, NotificationPresenterAdaptee { private let notificationCenter: UNUserNotificationCenter @@ -167,16 +169,19 @@ class UserNotificationPresenterAdaptee: NSObject, NotificationPresenterAdaptee { } } - func notify(category: AppNotificationCategory, title: String?, body: String, threadIdentifier: String?, userInfo: [AnyHashable: Any], interaction: INInteraction?, sound: OWSSound?) { + func notify(category: AppNotificationCategory, title: String?, body: String, threadIdentifier: String?, userInfo: [AnyHashable: Any], interaction: INInteraction?, sound: OWSSound?, + completion: NotificationCompletion?) { AssertIsOnMainThread() - notify(category: category, title: title, body: body, threadIdentifier: threadIdentifier, userInfo: userInfo, interaction: interaction, sound: sound, replacingIdentifier: nil) + notify(category: category, title: title, body: body, threadIdentifier: threadIdentifier, userInfo: userInfo, interaction: interaction, sound: sound, replacingIdentifier: nil, completion: completion) } - func notify(category: AppNotificationCategory, title: String?, body: String, threadIdentifier: String?, userInfo: [AnyHashable: Any], interaction: INInteraction?, sound: OWSSound?, replacingIdentifier: String?) { + func notify(category: AppNotificationCategory, title: String?, body: String, threadIdentifier: String?, userInfo: [AnyHashable: Any], interaction: INInteraction?, sound: OWSSound?, replacingIdentifier: String?, + completion: NotificationCompletion?) { AssertIsOnMainThread() guard tsAccountManager.isOnboarded() else { Logger.info("suppressing notification since user hasn't yet completed onboarding.") + completion?() return } @@ -235,6 +240,7 @@ class UserNotificationPresenterAdaptee: NSObject, NotificationPresenterAdaptee { if let error = error { owsFailDebug("Error: \(error)") } + completion?() } } #if swift(>=5.5) // TODO Temporary for Xcode 12 support. diff --git a/SignalServiceKit/Resources/schema.sql b/SignalServiceKit/Resources/schema.sql index b72f158df3..c71ba66790 100644 --- a/SignalServiceKit/Resources/schema.sql +++ b/SignalServiceKit/Resources/schema.sql @@ -1232,17 +1232,3 @@ CREATE WHERE recordType IS NOT 70 ; - -CREATE - TABLE - IF NOT EXISTS "MessageDecryptDeduplication" ( - "id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL - ,"serverGuid" TEXT NOT NULL - ) -; - -CREATE - INDEX "MessageDecryptDeduplication_serverGuid" - ON "MessageDecryptDeduplication"("serverGuid" - ) -; diff --git a/SignalServiceKit/src/Messages/MessageFetcherJob.swift b/SignalServiceKit/src/Messages/MessageFetcherJob.swift index 4236f8a493..6b9d8aba81 100644 --- a/SignalServiceKit/src/Messages/MessageFetcherJob.swift +++ b/SignalServiceKit/src/Messages/MessageFetcherJob.swift @@ -268,10 +268,10 @@ public class MessageFetcherJob: NSObject { return operationQueue }() - private let pendingAcks = PendingTasks() + private let pendingAcks = PendingTasks(label: "Acks") private func acknowledgeDelivery(envelopeInfo: EnvelopeInfo) { - let pendingAck = pendingAcks.buildPendingTask(label: "ack, timestamp: \(envelopeInfo.timestamp), serviceTimestamp: \(envelopeInfo.serviceTimestamp)") + let pendingAck = pendingAcks.buildPendingTask(label: "Ack, timestamp: \(envelopeInfo.timestamp), serviceTimestamp: \(envelopeInfo.serviceTimestamp)") let ackConcurrentOperation = MessageAckOperation(envelopeInfo: envelopeInfo, pendingAck: pendingAck) ackOperationQueue.addOperation(ackConcurrentOperation) @@ -307,23 +307,13 @@ public class MessageFetcherJob: NSObject { let envelopeInfo = Self.buildEnvelopeInfo(envelope: envelope) do { let envelopeData = try envelope.serializedData() - return EnvelopeJob(encryptedEnvelopeData: envelopeData, encryptedEnvelope: envelope) { processingError in - var shouldAck: Bool - switch processingError { - case nil: - shouldAck = true - case MessageProcessingError.duplicatePendingEnvelope?: - shouldAck = false - case MessageProcessingError.duplicateDecryption?: - shouldAck = true - default: - shouldAck = true - } - - if shouldAck { + return EnvelopeJob(encryptedEnvelopeData: envelopeData, encryptedEnvelope: envelope) { error in + let ackBehavior = MessageProcessor.handleMessageProcessingOutcome(error: error) + switch ackBehavior { + case .shouldAck: Self.messageFetcherJob.acknowledgeDelivery(envelopeInfo: envelopeInfo) - } else { - Logger.info("Skipping ack of message with timestamp \(envelope.timestamp) because of error: \(String(describing: processingError))") + case .shouldNotAck(let error): + Logger.info("Skipping ack of message with timestamp \(envelope.timestamp) because of error: \(error)") } } } catch { diff --git a/SignalServiceKit/src/Messages/MessageProcessor.swift b/SignalServiceKit/src/Messages/MessageProcessor.swift index be2f2ca5d1..e221e61372 100644 --- a/SignalServiceKit/src/Messages/MessageProcessor.swift +++ b/SignalServiceKit/src/Messages/MessageProcessor.swift @@ -443,6 +443,34 @@ public class MessageProcessor: NSObject { self.drainPendingEnvelopes() } } + + public enum MessageAckBehavior { + case shouldAck + case shouldNotAck(error: Error) + } + + public static func handleMessageProcessingOutcome(error: Error?) -> MessageAckBehavior { + guard let error = error else { + // Success. + return .shouldAck + } + if case MessageProcessingError.duplicatePendingEnvelope = error { + // _DO NOT_ ACK if de-duplicated before decryption. + return .shouldNotAck(error: error) + } else if let owsError = error as? OWSError, + owsError.errorCode == OWSErrorCode.failedToDecryptDuplicateMessage.rawValue { + // _DO_ ACK if de-duplicated during decryption. + return .shouldAck + } else { + Logger.warn("Failed to process message: \(error)") + Self.databaseStorage.write { transaction in + let errorMessage = ThreadlessErrorMessage.corruptedMessageInUnknownThread() + Self.notificationsManager?.notifyUser(forThreadlessErrorMessage: errorMessage, + transaction: transaction) + } + return .shouldAck + } + } } // MARK: - @@ -464,156 +492,6 @@ private protocol PendingEnvelope { // MARK: - -class MessageDecryptDeduplicationRecord: Codable, FetchableRecord, PersistableRecord { - static let databaseTableName = "MessageDecryptDeduplication" - - var id: Int64? - let serverGuid: String - - init(serverGuid: String) { - self.serverGuid = serverGuid - } - - func didInsert(with rowID: Int64, for column: String?) { - guard column == "id" else { return owsFailDebug("Expected id") } - id = rowID - } - - public enum Outcome { - case nonDuplicate - case duplicate - } - - public static func deduplicate( - encryptedEnvelope: SSKProtoEnvelope, - transaction: SDSAnyWriteTransaction, - skipCull: Bool = false - ) -> Outcome { - deduplicate(envelopeTimestamp: encryptedEnvelope.timestamp, - serviceTimestamp: encryptedEnvelope.serverTimestamp, - serverGuid: encryptedEnvelope.serverGuid, - transaction: transaction, - skipCull: skipCull) - } - - public static func deduplicate( - envelopeTimestamp: UInt64, - serviceTimestamp: UInt64, - serverGuid: String?, - transaction: SDSAnyWriteTransaction, - skipCull: Bool = false - ) -> Outcome { - guard envelopeTimestamp > 0 else { - owsFailDebug("Invalid envelopeTimestamp.") - return .nonDuplicate - } - guard serviceTimestamp > 0 else { - owsFailDebug("Invalid serviceTimestamp.") - return .nonDuplicate - } - guard let serverGuid = serverGuid?.nilIfEmpty else { - owsFailDebug("Missing serverGuid.") - return .nonDuplicate - } - do { - let isDuplicate: Bool = try { - let sql = """ - SELECT EXISTS ( SELECT 1 - FROM \(MessageDecryptDeduplicationRecord.databaseTableName) - WHERE serverGuid = ? ) - """ - let arguments: StatementArguments = [serverGuid] - return try Bool.fetchOne(transaction.unwrapGrdbWrite.database, sql: sql, arguments: arguments) ?? false - }() - guard !isDuplicate else { - Logger.warn("Discarding duplicate envelope with envelopeTimestamp: \(envelopeTimestamp), serviceTimestamp: \(serviceTimestamp), serverGuid: \(serverGuid)") - return .duplicate - } - - // No existing record found. Create a new one and insert it. - let record = MessageDecryptDeduplicationRecord(serverGuid: serverGuid) - try record.insert(transaction.unwrapGrdbWrite.database) - - if !skipCull, shouldCull() { - cull(transaction: transaction) - } - - if DebugFlags.internalLogging { - Logger.info("Proceeding with envelopeTimestamp: \(envelopeTimestamp), serviceTimestamp: \(serviceTimestamp), serverGuid: \(serverGuid)") - } - return .nonDuplicate - } catch { - owsFailDebug("Error: \(error)") - // If anything goes wrong with our bookkeeping, we must - // proceed with message processing. - return .nonDuplicate - } - } - - static let maxRecordCount: UInt = 1000 - - static let cullCount = AtomicUInt(0) - - private static func cull(transaction: SDSAnyWriteTransaction) { - - cullCount.increment() - - let oldCount = recordCount(transaction: transaction) - - guard oldCount > maxRecordCount else { - return - } - - do { - // It is sufficient to cull by record count in batches. - // The batch size must be larger than our cull frequency to bound total record count. - let cullCount: Int = min(Int(cullFrequency) * 2, Int(oldCount) - Int(maxRecordCount)) - - Logger.info("Culling \(cullCount) records.") - - // Find and delete the oldest N records. - let records = try MessageDecryptDeduplicationRecord.order(GRDB.Column("serviceTimestamp")) - .limit(cullCount) - .fetchAll(transaction.unwrapGrdbRead.database) - for record in records { - try record.delete(transaction.unwrapGrdbWrite.database) - } - - let newCount = recordCount(transaction: transaction) - if oldCount != newCount { - Logger.info("Culled by count: \(oldCount) -> \(newCount)") - } - owsAssertDebug(newCount <= maxRecordCount) - } catch { - owsFailDebug("Error: \(error)") - } - } - - public static func recordCount(transaction: SDSAnyReadTransaction) -> UInt { - MessageDecryptDeduplicationRecord.ows_fetchCount(transaction.unwrapGrdbRead.database) - } - - private static let unfairLock = UnfairLock() - private static var counter: UInt64 = 0 - static let cullFrequency: UInt64 = 100 - - private static func shouldCull() -> Bool { - unfairLock.withLock { - // Cull records once per N decryptions. - // - // NOTE: this always return true the first time that this - // method is called for a given launch of the process. - // We need to err on the side of culling too often to bound - // total record count. - let shouldCull = counter % cullFrequency == 0 - counter += 1 - return shouldCull - } - } -} - -// MARK: - - private struct EncryptedEnvelope: PendingEnvelope, Dependencies { let encryptedEnvelopeData: Data let encryptedEnvelope: SSKProtoEnvelope @@ -631,16 +509,6 @@ private struct EncryptedEnvelope: PendingEnvelope, Dependencies { } func decrypt(transaction: SDSAnyWriteTransaction) -> Swift.Result { - let deduplicationOutcome = MessageDecryptDeduplicationRecord.deduplicate(encryptedEnvelope: encryptedEnvelope, - transaction: transaction) - switch deduplicationOutcome { - case .nonDuplicate: - // Proceed with decryption. - break - case .duplicate: - return .failure(MessageProcessingError.duplicateDecryption) - } - let result = Self.messageDecrypter.decryptEnvelope( encryptedEnvelope, envelopeData: encryptedEnvelopeData, @@ -813,5 +681,4 @@ public class PendingEnvelopes { public enum MessageProcessingError: Error { case duplicatePendingEnvelope - case duplicateDecryption } diff --git a/SignalServiceKit/src/Messages/MessageSender.m b/SignalServiceKit/src/Messages/MessageSender.m index 30462c9fcd..54afcbc1ce 100644 --- a/SignalServiceKit/src/Messages/MessageSender.m +++ b/SignalServiceKit/src/Messages/MessageSender.m @@ -334,7 +334,7 @@ NSString *const MessageSenderSpamChallengeResolvedException = @"SpamChallengeRes } _sendingQueueMap = [NSMutableDictionary new]; - _pendingTasks = [PendingTasks new]; + _pendingTasks = [[PendingTasks alloc] initWithLabel:@"Message Sends"]; OWSSingletonAssert(); diff --git a/SignalServiceKit/src/Messages/OWSMessageDecrypter.swift b/SignalServiceKit/src/Messages/OWSMessageDecrypter.swift index 6b5076533a..e49249a024 100644 --- a/SignalServiceKit/src/Messages/OWSMessageDecrypter.swift +++ b/SignalServiceKit/src/Messages/OWSMessageDecrypter.swift @@ -293,7 +293,7 @@ public class OWSMessageDecrypter: OWSMessageHandler { let logString = "Error while decrypting \(Self.description(for: envelope)), error: \(error)" if case SignalError.duplicatedMessage(_) = error { - Logger.info(logString) + Logger.warn(logString) // Duplicate messages are not recorded in the database. return OWSError(error: .failedToDecryptDuplicateMessage, description: "Duplicate message", diff --git a/SignalServiceKit/src/Messages/OWSOutgoingReceiptManager.m b/SignalServiceKit/src/Messages/OWSOutgoingReceiptManager.m index 327d38ec02..b3ab1cb03b 100644 --- a/SignalServiceKit/src/Messages/OWSOutgoingReceiptManager.m +++ b/SignalServiceKit/src/Messages/OWSOutgoingReceiptManager.m @@ -72,7 +72,7 @@ NSString *NSStringForOWSReceiptType(OWSReceiptType receiptType) OWSSingletonAssert(); - _pendingTasks = [PendingTasks new]; + _pendingTasks = [[PendingTasks alloc] initWithLabel:@"Receipt Sends"]; // We skip any sends to untrusted identities since we know they'll fail anyway. If an identity state changes // we should recheck our pendingReceipts to re-attempt a send to formerly untrusted recipients. @@ -312,7 +312,7 @@ NSString *NSStringForOWSReceiptType(OWSReceiptType receiptType) return; } - NSString *label = [NSString stringWithFormat:@"receipt %@", NSStringForOWSReceiptType(receiptType)]; + NSString *label = [NSString stringWithFormat:@"Receipt Send: %@", NSStringForOWSReceiptType(receiptType)]; PendingTask *pendingTask = [self.pendingTasks buildPendingTaskWithLabel:label]; MessageReceiptSet *persistedSet = [self fetchReceiptSetWithType:receiptType diff --git a/SignalServiceKit/src/Network/API/HTTPUtils.swift b/SignalServiceKit/src/Network/API/HTTPUtils.swift index 058d2d1127..baef9ff54b 100644 --- a/SignalServiceKit/src/Network/API/HTTPUtils.swift +++ b/SignalServiceKit/src/Network/API/HTTPUtils.swift @@ -44,7 +44,7 @@ extension HTTPUtils { responseError: Error?, responseData: Data?) -> OWSHTTPError { - var errorDescription = "URL: \(requestUrl.absoluteString), status: \(responseStatus)" + var errorDescription = "URL: \(request.httpMethod) \(requestUrl.absoluteString), status: \(responseStatus)" if let responseError = responseError { errorDescription += ", error: \(responseError)" } @@ -68,7 +68,7 @@ extension HTTPUtils { switch responseStatus { case 0: - Logger.warn("The network request failed because of a connectivity error: \(requestUrl.absoluteString)") + Logger.warn("The network request failed because of a connectivity error: \(request.httpMethod) \(requestUrl.absoluteString)") let error = OWSHTTPError.networkFailure(requestUrl: requestUrl) return error case 400: @@ -101,7 +101,7 @@ extension HTTPUtils { recoverySuggestion: recoverySuggestion) return error case 413: - Logger.warn("Rate limit exceeded: \(requestUrl.absoluteString)") + Logger.warn("Rate limit exceeded: \(request.httpMethod) \(requestUrl.absoluteString)") let description = NSLocalizedString("REGISTER_RATE_LIMITING_ERROR", comment: "") let recoverySuggestion = NSLocalizedString("REGISTER_RATE_LIMITING_BODY", comment: "") let error = buildServiceResponseError(description: description, @@ -109,7 +109,7 @@ extension HTTPUtils { return error case 417: // TODO: Is this response code obsolete? - Logger.warn("The number is already registered on a relay. Please unregister there first: \(requestUrl.absoluteString)") + Logger.warn("The number is already registered on a relay. Please unregister there first: \(request.httpMethod) \(requestUrl.absoluteString)") let description = NSLocalizedString("REGISTRATION_ERROR", comment: "") let recoverySuggestion = NSLocalizedString("RELAY_REGISTERED_ERROR_RECOVERY", comment: "") let error = buildServiceResponseError(description: description, @@ -149,11 +149,11 @@ extension HTTPUtils { if Self.tsAccountManager.isRegisteredAndReady { Self.tsAccountManager.setIsDeregistered(true) } else { - Logger.warn("Ignoring auth failure not registered and ready: \(requestUrl.absoluteString).") + Logger.warn("Ignoring auth failure not registered and ready: \(request.httpMethod) \(requestUrl.absoluteString).") } } } else { - Logger.warn("Ignoring \(statusCode) for URL: \(requestUrl.absoluteString)") + Logger.warn("Ignoring \(statusCode) for URL: \(request.httpMethod) \(requestUrl.absoluteString)") } } } diff --git a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift index c9ea4af63f..b5e48850eb 100644 --- a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift +++ b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift @@ -530,28 +530,12 @@ public class OWSWebSocket: NSObject { Logger.info("\(currentWebSocket.logPrefix) 2 \(processingError?.localizedDescription ?? "success!")") } - var shouldAck: Bool - switch processingError { - case nil: - shouldAck = true - case MessageProcessingError.duplicatePendingEnvelope?: - shouldAck = false - case MessageProcessingError.duplicateDecryption?: - shouldAck = true - case let alertableError?: - shouldAck = true - Logger.warn("Failed to process message: \(alertableError)") - Self.databaseStorage.write { transaction in - let errorMessage = ThreadlessErrorMessage.corruptedMessageInUnknownThread() - Self.notificationsManager?.notifyUser(forThreadlessErrorMessage: errorMessage, - transaction: transaction) - } - } - - if shouldAck { + let ackBehavior = MessageProcessor.handleMessageProcessingOutcome(error: processingError) + switch ackBehavior { + case .shouldAck: self.sendWebSocketMessageAcknowledgement(message, currentWebSocket: currentWebSocket) - } else { - Logger.info("Skipping ack of message with serverTimestamp \(serverTimestamp) because of error: \(String(describing: processingError))") + case .shouldNotAck(let error): + Logger.info("Skipping ack of message with serverTimestamp \(serverTimestamp) because of error: \(error)") } owsAssertDebug(backgroundTask != nil) @@ -593,12 +577,12 @@ public class OWSWebSocket: NSObject { } Self.messageProcessor.processEncryptedEnvelopeData(encryptedEnvelope, serverDeliveryTimestamp: serverDeliveryTimestamp, - envelopeSource: envelopeSource) { outcome in + envelopeSource: envelopeSource) { error in if Self.verboseLogging { Logger.info("\(currentWebSocket.logPrefix) 3") } Self.serialQueue.async { - ackMessage(outcome, serverDeliveryTimestamp) + ackMessage(error, serverDeliveryTimestamp) } } } diff --git a/SignalServiceKit/src/Storage/Database/GRDBDatabaseStorageAdapter.swift b/SignalServiceKit/src/Storage/Database/GRDBDatabaseStorageAdapter.swift index 9828e80c02..beafbcaeed 100644 --- a/SignalServiceKit/src/Storage/Database/GRDBDatabaseStorageAdapter.swift +++ b/SignalServiceKit/src/Storage/Database/GRDBDatabaseStorageAdapter.swift @@ -130,8 +130,7 @@ public class GRDBDatabaseStorageAdapter: NSObject { MediaGalleryRecord.self, MessageSendLog.Payload.self, MessageSendLog.Recipient.self, - MessageSendLog.Message.self, - MessageDecryptDeduplicationRecord.self + MessageSendLog.Message.self ] // MARK: - DatabaseChangeObserver diff --git a/SignalServiceKit/src/Storage/Database/GRDBSchemaMigrator.swift b/SignalServiceKit/src/Storage/Database/GRDBSchemaMigrator.swift index a24055c279..55ed6b1496 100644 --- a/SignalServiceKit/src/Storage/Database/GRDBSchemaMigrator.swift +++ b/SignalServiceKit/src/Storage/Database/GRDBSchemaMigrator.swift @@ -119,7 +119,7 @@ public class GRDBSchemaMigrator: NSObject { case updateMessageSendLogColumnTypes case addRecordTypeIndex case tunedConversationLoadIndices - case messageDecryptDeduplicationV5 + case messageDecryptDeduplicationV6 // NOTE: Every time we add a migration id, consider // incrementing grdbSchemaVersionLatest. @@ -1454,24 +1454,11 @@ public class GRDBSchemaMigrator: NSObject { } } - migrator.registerMigration(MigrationId.messageDecryptDeduplicationV5.rawValue) { db in + migrator.registerMigration(MigrationId.messageDecryptDeduplicationV6.rawValue) { db in do { if try db.tableExists("MessageDecryptDeduplication") { try db.drop(table: "MessageDecryptDeduplication") } - - try db.create(table: "MessageDecryptDeduplication") { table in - table.autoIncrementedPrimaryKey("id") - .notNull() - table.column("serverGuid", .text) - .notNull() - } - - try db.create( - index: "MessageDecryptDeduplication_serverGuid", - on: "MessageDecryptDeduplication", - columns: ["serverGuid"] - ) } catch { owsFail("Error: \(error)") } diff --git a/SignalServiceKit/src/Util/OWSError.swift b/SignalServiceKit/src/Util/OWSError.swift index eabe7fcb72..0137c30ebb 100644 --- a/SignalServiceKit/src/Util/OWSError.swift +++ b/SignalServiceKit/src/Util/OWSError.swift @@ -55,6 +55,17 @@ public class OWSError: NSObject, CustomNSError, IsRetryableProvider, UserErrorDe isRetryable: customIsRetryable) as Error as NSError } + @objc + public override var description: String { + var result = "[OWSError code: \(errorCode), description: \(customLocalizedDescription)" + if let customUserInfo = self.customUserInfo, + !customUserInfo.isEmpty { + result += ", userInfo: \(customUserInfo)" + } + result += "]" + return result + } + // MARK: - CustomNSError // NSError bridging: the domain of the error. diff --git a/SignalServiceKit/src/Util/PendingTasks.swift b/SignalServiceKit/src/Util/PendingTasks.swift index 3740f77610..57d834c3d0 100644 --- a/SignalServiceKit/src/Util/PendingTasks.swift +++ b/SignalServiceKit/src/Util/PendingTasks.swift @@ -7,14 +7,32 @@ import SignalCoreKit @objc public class PendingTasks: NSObject { + + fileprivate let label: String + private let pendingTasks = AtomicDictionary() + @objc + public required init(label: String) { + self.label = label + } + public func pendingTasksPromise() -> Promise { // This promise blocks on all pending tasks already in flight, // but will not block on new tasks added after this promise // is created. + let label = self.label + if DebugFlags.internalLogging { + Logger.info("Waiting \(label).") + } let promises = pendingTasks.allValues.map { $0.promise } - return Promise.when(resolved: promises).asVoid() + return firstly(on: .global()) { + Promise.when(resolved: promises).asVoid() + }.map(on: .global()) { + if DebugFlags.internalLogging { + Logger.info("Complete \(label).") + } + } } @objc diff --git a/SignalServiceKit/tests/Messages/MessageProcessingIntegrationTest.swift b/SignalServiceKit/tests/Messages/MessageProcessingIntegrationTest.swift index 561a924030..6b9aa77087 100644 --- a/SignalServiceKit/tests/Messages/MessageProcessingIntegrationTest.swift +++ b/SignalServiceKit/tests/Messages/MessageProcessingIntegrationTest.swift @@ -107,8 +107,6 @@ class MessageProcessingIntegrationTest: SSKBaseTestSwift { switch error { case MessageProcessingError.duplicatePendingEnvelope?: XCTFail("duplicatePendingEnvelope") - case MessageProcessingError.duplicateDecryption?: - XCTFail("duplicateDecryption") case .some(_): XCTFail("failure") case nil: @@ -181,8 +179,6 @@ class MessageProcessingIntegrationTest: SSKBaseTestSwift { switch error { case MessageProcessingError.duplicatePendingEnvelope?: XCTFail("duplicatePendingEnvelope") - case MessageProcessingError.duplicateDecryption?: - XCTFail("duplicateDecryption") case .some(_): XCTFail("failure") case nil: diff --git a/SignalServiceKit/tests/Messages/MessageProcessorTest.swift b/SignalServiceKit/tests/Messages/MessageProcessorTest.swift deleted file mode 100644 index f531d44343..0000000000 --- a/SignalServiceKit/tests/Messages/MessageProcessorTest.swift +++ /dev/null @@ -1,162 +0,0 @@ -// -// Copyright (c) 2021 Open Whisper Systems. All rights reserved. -// - -import XCTest -@testable import SignalServiceKit -import GRDB - -class MessageProcessorTest: SSKBaseTestSwift { - - // MARK: - Hooks - - override func setUp() { - super.setUp() - } - - override func tearDown() { - super.tearDown() - } - - // MARK: - Tests - - func testDecryptDeduplication_withoutCulling() { - MessageDecryptDeduplicationRecord.cullCount.set(0) - - write { transaction in - let now = Date.ows_millisecondTimestamp() - let timestamp1: UInt64 = now - let timestamp2: UInt64 = now + 10 - let serverGuid1 = UUID().uuidString - let serverGuid2 = UUID().uuidString - - XCTAssertEqual(.nonDuplicate, - MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp1, - serviceTimestamp: timestamp1, - serverGuid: serverGuid1, - transaction: transaction, - skipCull: true)) - // A duplicate even if serviceTimestamp doesn't match. - XCTAssertEqual(.duplicate, - MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp2, - serviceTimestamp: timestamp2, - serverGuid: serverGuid1, - transaction: transaction, - skipCull: true)) - // Not a duplicate if serverGuid doesn't match. - XCTAssertEqual(.nonDuplicate, - MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp1, - serviceTimestamp: timestamp1, - serverGuid: serverGuid2, - transaction: transaction, - skipCull: true)) - // A duplicate if both match. - XCTAssertEqual(.duplicate, - MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp1, - serviceTimestamp: timestamp1, - serverGuid: serverGuid1, - transaction: transaction, - skipCull: true)) - // A duplicate if both match even if you ask twice. - XCTAssertEqual(.duplicate, - MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp1, - serviceTimestamp: timestamp1, - serverGuid: serverGuid1, - transaction: transaction, - skipCull: true)) - } - - XCTAssertEqual(0, MessageDecryptDeduplicationRecord.cullCount.get()) - } - - func testDecryptDeduplication_withCullingByRecordCount() { - MessageDecryptDeduplicationRecord.cullCount.set(0) - - write { transaction in - let now = Date.ows_millisecondTimestamp() - let timestamp1: UInt64 = now - let timestamp2: UInt64 = now + 10 - let serverGuid1 = UUID().uuidString - let serverGuid2 = UUID().uuidString - - XCTAssertEqual(.nonDuplicate, - MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp1, - serviceTimestamp: timestamp1, - serverGuid: serverGuid1, - transaction: transaction)) - // A duplicate if both match. - XCTAssertEqual(.duplicate, - MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp1, - serviceTimestamp: timestamp1, - serverGuid: serverGuid1, - transaction: transaction)) - - // Create enough records to: - // - // * Overflow culling by record count. - // * Ensure that culling is triggered afterward. - let mockRecordCount = (MessageDecryptDeduplicationRecord.cullFrequency + - UInt64(MessageDecryptDeduplicationRecord.maxRecordCount)) - for index in 0..