signalapp/Signal-iOS
signalapp/Signal-iOS
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 4 Wiki Activity

Refine NSE completion blocking, message processing.

Browse Source
This commit is contained in:
Matthew Chen 2021-09-25 10:09:51 -03:00
parent 73998244e2
commit e70302ac8f
19 changed files with 170 additions and 466 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
NotificationServiceExtension/NotificationService.swift
View File

@ -149,29 +149,19 @@ class NotificationService: UNNotificationServiceExtension {
let processingCompletePromise = firstly {
self.messageProcessor.processingCompletePromise()
}.then(on: .global()) { () -> Promise<Void> in
// Wait until all notifications are enqueued.
if DebugFlags.internalLogging {
Logger.info("Waiting on notificationPresenter.")
}
return NotificationPresenter.pendingNotificationsPromise()
}.then(on: .global()) { () -> Promise<Void> in
if DebugFlags.internalLogging {
Logger.info("Waiting on acks.")
}
// Wait until all ACKs are enqueued.
return Self.messageFetcherJob.pendingAcksPromise()
}.then(on: .global()) { () -> Promise<Void> 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<Void> 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<Void>] = [
// 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

2
Pods

@ -1 +1 @@
Subproject commit df04141b244c85b1de4ab571b3d7b7bc71850288
Subproject commit 43b2a1a14e5bfb0b6773a04c97df97ec2be88d17

2
Signal/src/ViewControllers/DebugUI/DebugUIMessages.swift
View File

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

88
SignalMessaging/Notifications/AppNotifications.swift
View File

@ -158,6 +158,8 @@ let kNotificationDelayForRemoteRead: TimeInterval = 20
let kAudioNotificationsThrottleCount = 2
let kAudioNotificationsThrottleInterval: TimeInterval = 5
typealias NotificationCompletion = () -> Void
protocol NotificationPresenterAdaptee: AnyObject {
func registerNotificationSettings() -> Promise<Void>
@ -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<Void> {
// 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<Void>.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)
}
}

12
SignalMessaging/Notifications/UserNotificationsAdaptee.swift
View File

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

14
SignalServiceKit/Resources/schema.sql
View File

@ -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"
)
;

26
SignalServiceKit/src/Messages/MessageFetcherJob.swift
View File

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

189
SignalServiceKit/src/Messages/MessageProcessor.swift
View File

@ -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<DecryptedEnvelope, Error> {
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
}

2
SignalServiceKit/src/Messages/MessageSender.m
View File

@ -334,7 +334,7 @@ NSString *const MessageSenderSpamChallengeResolvedException = @"SpamChallengeRes
}
_sendingQueueMap = [NSMutableDictionary new];
_pendingTasks = [PendingTasks new];
_pendingTasks = [[PendingTasks alloc] initWithLabel:@"Message Sends"];
OWSSingletonAssert();

2
SignalServiceKit/src/Messages/OWSMessageDecrypter.swift
View File

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

4
SignalServiceKit/src/Messages/OWSOutgoingReceiptManager.m
View File

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

12
SignalServiceKit/src/Network/API/HTTPUtils.swift
View File

@ -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)")
}
}
}

30
SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift
View File

@ -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)
}
}
}

3
SignalServiceKit/src/Storage/Database/GRDBDatabaseStorageAdapter.swift
View File

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

17
SignalServiceKit/src/Storage/Database/GRDBSchemaMigrator.swift
View File

@ -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)")
}

11
SignalServiceKit/src/Util/OWSError.swift
View File

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

20
SignalServiceKit/src/Util/PendingTasks.swift
View File

@ -7,14 +7,32 @@ import SignalCoreKit
@objc
public class PendingTasks: NSObject {
fileprivate let label: String
private let pendingTasks = AtomicDictionary<UInt, PendingTask>()
@objc
public required init(label: String) {
self.label = label
}
public func pendingTasksPromise() -> Promise<Void> {
// 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

4
SignalServiceKit/tests/Messages/MessageProcessingIntegrationTest.swift
View File

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

162
SignalServiceKit/tests/Messages/MessageProcessorTest.swift
View File

@ -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..<mockRecordCount {
let timestamp: UInt64 = now + 25 + index
let serverGuid = UUID().uuidString
XCTAssertEqual(.nonDuplicate,
MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp,
serviceTimestamp: timestamp,
serverGuid: serverGuid,
transaction: transaction))
}
XCTAssertEqual(.nonDuplicate,
MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp2,
serviceTimestamp: timestamp2,
serverGuid: serverGuid2,
transaction: transaction))
// Due to culling, this is no longer a duplicate.
XCTAssertEqual(.nonDuplicate,
MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp1,
serviceTimestamp: timestamp1,
serverGuid: serverGuid1,
transaction: transaction))
// Still a duplicate; not yet culled.
XCTAssertEqual(.duplicate,
MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp2,
serviceTimestamp: timestamp2,
serverGuid: serverGuid2,
transaction: transaction))
}
}
func testCullingByRecordCount() {
MessageDecryptDeduplicationRecord.cullCount.set(0)
write { transaction in
let now = Date.ows_millisecondTimestamp()
let peakRecordCount = (MessageDecryptDeduplicationRecord.cullFrequency +
UInt64(MessageDecryptDeduplicationRecord.maxRecordCount))
// Create enough records to ensure that culling is triggered multiple times.
let mockRecordCount = (MessageDecryptDeduplicationRecord.cullFrequency * 10 +
UInt64(MessageDecryptDeduplicationRecord.maxRecordCount))
for index in 0..<mockRecordCount {
XCTAssertTrue(MessageDecryptDeduplicationRecord.recordCount(transaction: transaction) < peakRecordCount)
let timestamp: UInt64 = now + index
let serverGuid = UUID().uuidString
XCTAssertEqual(.nonDuplicate,
MessageDecryptDeduplicationRecord.deduplicate(envelopeTimestamp: timestamp,
serviceTimestamp: timestamp,
serverGuid: serverGuid,
transaction: transaction))
XCTAssertTrue(MessageDecryptDeduplicationRecord.recordCount(transaction: transaction) < peakRecordCount)
}
}
// The exact value doesn't matter; we just want to verify that the
// behavior is deterministic and that we're not culling too often.
XCTAssertEqual(20, MessageDecryptDeduplicationRecord.cullCount.get())
}
}