From 6e188632b171a0398d7651a9e327c01cf7a2a4e1 Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Thu, 23 Sep 2021 22:44:00 -0700 Subject: [PATCH 1/7] Skip acks of duplicate REST messages Also more logging to correlate un-acked messages --- SignalServiceKit/src/Messages/MessageProcessor.swift | 2 +- SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/SignalServiceKit/src/Messages/MessageProcessor.swift b/SignalServiceKit/src/Messages/MessageProcessor.swift index 7e303a45c0..0d2ee68a39 100644 --- a/SignalServiceKit/src/Messages/MessageProcessor.swift +++ b/SignalServiceKit/src/Messages/MessageProcessor.swift @@ -192,7 +192,7 @@ public class MessageProcessor: NSObject { let result = pendingEnvelopes.enqueue(encryptedEnvelope: encryptedEnvelope) switch result { case .duplicate: - Logger.warn("Duplicate envelope, envelopeSource: \(envelopeSource).") + Logger.warn("Duplicate envelope \(encryptedEnvelopeProto.timestamp). Server timestamp: \(serverDeliveryTimestamp). EnvelopeSource: \(envelopeSource).") completion(MessageProcessingError.duplicatePendingEnvelope) case .enqueued: drainPendingEnvelopes() diff --git a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift index 906e87c63c..fd7657fde0 100644 --- a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift +++ b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift @@ -525,7 +525,7 @@ public class OWSWebSocket: NSObject { Logger.info("\(currentWebSocket.logPrefix) 1") } - let ackMessage = { (processingError: Error?) in + let ackMessage = { (processingError: Error?, serverTimestamp: UInt64) in if Self.verboseLogging { Logger.info("\(currentWebSocket.logPrefix) 2 \(processingError?.localizedDescription ?? "success!")") } @@ -573,7 +573,7 @@ public class OWSWebSocket: NSObject { } guard let encryptedEnvelope = message.body else { - ackMessage(OWSGenericError("Missing encrypted envelope on message \(currentWebSocket.logPrefix)")) + ackMessage(OWSGenericError("Missing encrypted envelope on message \(currentWebSocket.logPrefix)"), serverDeliveryTimestamp) return } let envelopeSource: EnvelopeSource = { @@ -596,7 +596,7 @@ public class OWSWebSocket: NSObject { Logger.info("\(currentWebSocket.logPrefix) 3") } Self.serialQueue.async { - ackMessage(outcome) + ackMessage(outcome, serverDeliveryTimestamp) } } } From a452ed3061c44bc558c26949098879a18e079c9a Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Thu, 23 Sep 2021 23:28:07 -0700 Subject: [PATCH 2/7] Wait for RingRTC before we finish handling VoIP push --- .../NSECallMessageHandler.swift | 3 +- .../CallKit/CallKitCallUIAdaptee.swift | 2 + .../environment/PushRegistrationManager.swift | 97 ++++++++++++++----- SignalMessaging/utils/CallMessageRelay.swift | 37 +++++-- 4 files changed, 104 insertions(+), 35 deletions(-) diff --git a/NotificationServiceExtension/NSECallMessageHandler.swift b/NotificationServiceExtension/NSECallMessageHandler.swift index 1b7c07fcb1..d7aecad6fd 100644 --- a/NotificationServiceExtension/NSECallMessageHandler.swift +++ b/NotificationServiceExtension/NSECallMessageHandler.swift @@ -57,7 +57,8 @@ public class NSECallMessageHandler: NSObject, OWSCallMessageHandler { transaction: transaction ) - CXProvider.reportNewIncomingVoIPPushPayload(payload) { error in + Logger.info("Notifying primary app of incoming call with push payload: \(payload)") + CXProvider.reportNewIncomingVoIPPushPayload(payload.payloadDict) { error in if let error = error { owsFailDebug("Failed to notify main app of call message: \(error)") } else { diff --git a/Signal/src/Calls/UserInterface/Individual/CallKit/CallKitCallUIAdaptee.swift b/Signal/src/Calls/UserInterface/Individual/CallKit/CallKitCallUIAdaptee.swift index 724821d29e..023ca1f053 100644 --- a/Signal/src/Calls/UserInterface/Individual/CallKit/CallKitCallUIAdaptee.swift +++ b/Signal/src/Calls/UserInterface/Individual/CallKit/CallKitCallUIAdaptee.swift @@ -169,6 +169,8 @@ final class CallKitCallUIAdaptee: NSObject, CallUIAdaptee, CXProviderDelegate { Only add incoming call to the app's list of calls if the call was allowed (i.e. there was no error) since calls may be "denied" for various legitimate reasons. See CXErrorCodeIncomingCallError. */ + self.pushRegistrationManager.didFinishReportingIncomingCall() + guard error == nil else { completion(error) Logger.error("failed to report new incoming call, error: \(error!)") diff --git a/Signal/src/environment/PushRegistrationManager.swift b/Signal/src/environment/PushRegistrationManager.swift index eba1ab77f1..92885607fc 100644 --- a/Signal/src/environment/PushRegistrationManager.swift +++ b/Signal/src/environment/PushRegistrationManager.swift @@ -24,6 +24,15 @@ public enum PushRegistrationError: Error { SwiftSingletons.register(self) } + private let pendingCallSignal = DispatchSemaphore(value: 0) + private let isWaitingForSignal = AtomicBool(false) + // Private callout queue that we can use to synchronously wait for our call to start + // TODO: Rewrite call message routing to be able to synchronously report calls + private let calloutQueue = DispatchQueue( + label: "org.whispersystems.signal.PKPushRegistry", + autoreleaseFrequency: .workItem + ) + private var vanillaTokenPromise: Promise? private var vanillaTokenFuture: Future? @@ -53,6 +62,12 @@ public enum PushRegistrationError: Error { } } + public func didFinishReportingIncomingCall() { + if isWaitingForSignal.tryToClearFlag() { + pendingCallSignal.signal() + } + } + // MARK: Vanilla push token @objc @@ -92,36 +107,69 @@ public enum PushRegistrationError: Error { // MARK: PKPushRegistryDelegate - voIP Push Token public func pushRegistry(_ registry: PKPushRegistry, didReceiveIncomingPushWith payload: PKPushPayload, for type: PKPushType) { - AssertIsOnMainThread() - Logger.info("isAppReady: \(AppReadiness.isAppReady)") + dispatchPrecondition(condition: .onQueue(calloutQueue)) owsAssertDebug(type == .voIP) - AppReadiness.runNowOrWhenAppDidBecomeReadySync { - AssertIsOnMainThread() - if CallMessageRelay.handleVoipPayload(payload.dictionaryPayload) { - // Do nothing. This was a call message relayed from the NSE - Logger.info("Handled call message from NSE.") - } else if let preauthChallengeFuture = self.preauthChallengeFuture, - let challenge = payload.dictionaryPayload["challenge"] as? String { - Logger.info("received preauth challenge") - preauthChallengeFuture.resolve(challenge) - self.preauthChallengeFuture = nil - } else { - owsAssertDebug(!FeatureFlags.notificationServiceExtension) - Logger.info("Fetching messages.") - var backgroundTask: OWSBackgroundTask? = OWSBackgroundTask(label: "Push fetch.") - firstly { () -> Promise in - self.messageFetcherJob.run().promise - }.done(on: .main) { - owsAssertDebug(backgroundTask != nil) - backgroundTask = nil - }.catch { error in - owsFailDebug("Error: \(error)") + + let callRelayPayload = CallMessagePushPayload(payload.dictionaryPayload) + if let callRelayPayload = callRelayPayload { + Logger.info("Received VoIP push from the NSE: \(callRelayPayload)") + owsAssertDebug(isWaitingForSignal.tryToSetFlag()) + } + + // One of the few places we dispatch_sync, this should be safe since we can only block our + // private calloutQueue while waiting for a chance to run on the main thread. + // This should be deadlock free since the only thing dispatching to our calloutQueue is PushKit. + DispatchQueue.main.sync { + AppReadiness.runNowOrWhenAppDidBecomeReadySync { + AssertIsOnMainThread() + if let callRelayPayload = callRelayPayload { + CallMessageRelay.handleVoipPayload(callRelayPayload) + } else if let preauthChallengeFuture = self.preauthChallengeFuture, + let challenge = payload.dictionaryPayload["challenge"] as? String { + Logger.info("received preauth challenge") + preauthChallengeFuture.resolve(challenge) + self.preauthChallengeFuture = nil + } else { + owsAssertDebug(!FeatureFlags.notificationServiceExtension) + Logger.info("Fetching messages.") + var backgroundTask: OWSBackgroundTask? = OWSBackgroundTask(label: "Push fetch.") + firstly { () -> Promise in + self.messageFetcherJob.run().promise + }.done(on: .main) { + owsAssertDebug(backgroundTask != nil) + backgroundTask = nil + }.catch { error in + owsFailDebug("Error: \(error)") + } } } } + + if let callRelayPayload = callRelayPayload { + // We need to handle the incoming call push in the same runloop as it was delivered + // RingRTC will callback to us async to start the call after we hand them the call message, + // but once we return from here it's too late. + // + // We'll probably need RingRTC changes to handle this properly and synchronously. Until then, + // let's just block the thread for a bit until we hear back that the call was started.. + Logger.info("Waiting for call to start: \(callRelayPayload)") + let waitInterval = DispatchTimeInterval.seconds(5) + + if pendingCallSignal.wait(timeout: .now() + waitInterval) == .timedOut { + owsFailDebug("Call didn't start within \(waitInterval) seconds. Continuing anyway, expecting to be killed.") + // We want to make sure we increment the semaphore exactly once per call to reset state + // for the next call. If we timed-out on the semaphore, we could race with another thread + // signaling the semaphore at this instant. We consult the atomic bool before re-incrementing. + if isWaitingForSignal.tryToClearFlag() { + pendingCallSignal.signal() + } + } + Logger.info("Returning back to PushKit. Good luck! \(callRelayPayload)") + } } public func pushRegistry(_ registry: PKPushRegistry, didUpdate credentials: PKPushCredentials, for type: PKPushType) { + dispatchPrecondition(condition: .onQueue(calloutQueue)) Logger.info("") owsAssertDebug(type == .voIP) owsAssertDebug(credentials.type == .voIP) @@ -133,6 +181,7 @@ public enum PushRegistrationError: Error { public func pushRegistry(_ registry: PKPushRegistry, didInvalidatePushTokenFor type: PKPushType) { // It's not clear when this would happen. We've never previously handled it, but we should at // least start learning if it happens. + dispatchPrecondition(condition: .onQueue(calloutQueue)) owsFailDebug("Invalid state") } @@ -231,7 +280,7 @@ public enum PushRegistrationError: Error { AssertIsOnMainThread() guard voipRegistry == nil else { return } - let voipRegistry = PKPushRegistry(queue: nil) + let voipRegistry = PKPushRegistry(queue: calloutQueue) self.voipRegistry = voipRegistry voipRegistry.desiredPushTypes = [.voIP] voipRegistry.delegate = self diff --git a/SignalMessaging/utils/CallMessageRelay.swift b/SignalMessaging/utils/CallMessageRelay.swift index f5a49b5e96..bb3cf4844e 100644 --- a/SignalMessaging/utils/CallMessageRelay.swift +++ b/SignalMessaging/utils/CallMessageRelay.swift @@ -4,15 +4,35 @@ import Foundation +public class CallMessagePushPayload: CustomStringConvertible { + private static let identifierKey = "CallMessageRelayPayload" + public let identifier: String + + fileprivate init() { + identifier = UUID().uuidString + } + + public init?(_ payloadDict: [AnyHashable: Any]) { + guard let payloadId = payloadDict[Self.identifierKey] as? String else { return nil } + identifier = payloadId + } + + public var payloadDict: [String: String] { + [Self.identifierKey: identifier] + } + + public var description: String { + "\(type(of: self)): \(identifier.suffix(6))" + } +} + @objc public class CallMessageRelay: NSObject { - private static let callMessagePayloadKey = "CallMessageRelayPayload" private static let pendingCallMessageStore = SDSKeyValueStore(collection: "PendingCallMessageStore") - @objc - public static func handleVoipPayload(_ payload: [AnyHashable: Any]) -> Bool { - guard let payload = payload[callMessagePayloadKey] as? Bool, payload == true else { return false } - + public static func handleVoipPayload(_ payload: CallMessagePushPayload) { + Logger.info("Handling incoming VoIP payload: \(payload)") + defer { Logger.info("Finished handling incoming VoIP payload: \(payload)") } // Process all the pending call messages from the NSE in 1 batch. // This should almost always be a batch of one. databaseStorage.write { transaction in @@ -42,8 +62,6 @@ public class CallMessageRelay: NSObject { ) } } - - return true } public static func enqueueCallMessageForMainApp( @@ -52,7 +70,7 @@ public class CallMessageRelay: NSObject { wasReceivedByUD: Bool, serverDeliveryTimestamp: UInt64, transaction: SDSAnyWriteTransaction - ) throws -> [String: Any] { + ) throws -> CallMessagePushPayload { let payload = Payload( envelope: envelope, plaintextData: plaintextData, @@ -61,8 +79,7 @@ public class CallMessageRelay: NSObject { ) try pendingCallMessageStore.setCodable(payload, key: "\(envelope.timestamp)", transaction: transaction) - - return [callMessagePayloadKey: true] + return CallMessagePushPayload() } private struct Payload: Codable { From 60d334ef1b9766c09e3831f11f4cf3f9d2b144e3 Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Fri, 24 Sep 2021 00:37:02 -0700 Subject: [PATCH 3/7] Pause message processing in NSE when waking main app --- .../NSECallMessageHandler.swift | 7 ++++++ .../Messages/MessagePipelineSupervisor.swift | 4 ++-- .../src/Messages/MessageProcessor.swift | 23 +++++++++++++++---- 3 files changed, 28 insertions(+), 6 deletions(-) diff --git a/NotificationServiceExtension/NSECallMessageHandler.swift b/NotificationServiceExtension/NSECallMessageHandler.swift index d7aecad6fd..ed22f6c6a1 100644 --- a/NotificationServiceExtension/NSECallMessageHandler.swift +++ b/NotificationServiceExtension/NSECallMessageHandler.swift @@ -57,6 +57,13 @@ public class NSECallMessageHandler: NSObject, OWSCallMessageHandler { transaction: transaction ) + // We don't want to risk consuming any call messages that the main app needs to perform the call + // We suspend message processing in our process to give the main app a chance to wake and take over + let suspension = messagePipelineSupervisor.suspendMessageProcessing(for: "Waking main app for \(payload)") + DispatchQueue.sharedUtility.asyncAfter(deadline: .now() + .seconds(10)) { + suspension.invalidate() + } + Logger.info("Notifying primary app of incoming call with push payload: \(payload)") CXProvider.reportNewIncomingVoIPPushPayload(payload.payloadDict) { error in if let error = error { diff --git a/SignalServiceKit/src/Messages/MessagePipelineSupervisor.swift b/SignalServiceKit/src/Messages/MessagePipelineSupervisor.swift index 258897c226..e30be32540 100644 --- a/SignalServiceKit/src/Messages/MessagePipelineSupervisor.swift +++ b/SignalServiceKit/src/Messages/MessagePipelineSupervisor.swift @@ -76,11 +76,11 @@ public class MessagePipelineSupervisor: NSObject { // MARK: - Private private func incrementSuspensionCount(for reason: String) { - Logger.verbose("Incrementing suspension refcount for reason: \(reason)") let updatedCount: Int = lock.withLock { suspensionCount += 1 return suspensionCount } + Logger.info("Incremented suspension refcount to \(updatedCount) for reason: \(reason)") if updatedCount == 1 { notifyOfSuspensionStateChange() } @@ -91,7 +91,7 @@ public class MessagePipelineSupervisor: NSObject { suspensionCount -= 1 return suspensionCount } - Logger.verbose("Decremented suspension refcount for reason: \(reason)") + Logger.info("Decremented suspension refcount to \(updatedCount) for reason: \(reason)") assert(updatedCount >= 0, "Suspension refcount dipped below zero") if updatedCount == 0 { diff --git a/SignalServiceKit/src/Messages/MessageProcessor.swift b/SignalServiceKit/src/Messages/MessageProcessor.swift index 0d2ee68a39..d1a59af6d7 100644 --- a/SignalServiceKit/src/Messages/MessageProcessor.swift +++ b/SignalServiceKit/src/Messages/MessageProcessor.swift @@ -289,12 +289,23 @@ public class MessageProcessor: NSObject { Logger.info("Processing batch of \(batchEnvelopes.count)/\(pendingEnvelopesCount) received envelope(s).") + var skippedEnvelopes: [PendingEnvelope] = [] SDSDatabaseStorage.shared.write { transaction in - batchEnvelopes.forEach { self.processEnvelope($0, transaction: transaction) } + batchEnvelopes.forEach { + if messagePipelineSupervisor.isMessageProcessingPermitted { + self.processEnvelope($0, transaction: transaction) + } else { + skippedEnvelopes.append($0) + } + } } // Remove the processed envelopes from the pending list. - pendingEnvelopes.removeProcessedBatch(batch) + if skippedEnvelopes.isEmpty { + pendingEnvelopes.removeProcessedBatch(batch) + } else { + pendingEnvelopes.removePartiallySuccessfulBatch(batch, failedEnvelopes: skippedEnvelopes) + } return true } @@ -748,7 +759,7 @@ public class PendingEnvelopes { } } - fileprivate func removeProcessedBatch(_ batch: Batch) { + fileprivate func removePartiallySuccessfulBatch(_ batch: Batch, failedEnvelopes: [PendingEnvelope]) { unfairLock.withLock { let batchEnvelopes = batch.batchEnvelopes guard pendingEnvelopes.count > batchEnvelopes.count else { @@ -756,7 +767,7 @@ public class PendingEnvelopes { return } let oldCount = pendingEnvelopes.count - pendingEnvelopes = Array(pendingEnvelopes.suffix(from: batchEnvelopes.count)) + pendingEnvelopes = failedEnvelopes + Array(pendingEnvelopes.suffix(from: batchEnvelopes.count)) let newCount = pendingEnvelopes.count if DebugFlags.internalLogging { Logger.info("\(oldCount) -> \(newCount)") @@ -764,6 +775,10 @@ public class PendingEnvelopes { } } + fileprivate func removeProcessedBatch(_ batch: Batch) { + removePartiallySuccessfulBatch(batch, failedEnvelopes: []) + } + fileprivate func enqueue(decryptedEnvelope: DecryptedEnvelope) { unfairLock.withLock { let oldCount = pendingEnvelopes.count From 4984270635a360d619773bd2586fcb15c3a4e192 Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Fri, 24 Sep 2021 01:01:30 -0700 Subject: [PATCH 4/7] Rebalance semaphore count --- .../src/environment/PushRegistrationManager.swift | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/Signal/src/environment/PushRegistrationManager.swift b/Signal/src/environment/PushRegistrationManager.swift index 92885607fc..5fc3666c4e 100644 --- a/Signal/src/environment/PushRegistrationManager.swift +++ b/Signal/src/environment/PushRegistrationManager.swift @@ -157,11 +157,16 @@ public enum PushRegistrationError: Error { if pendingCallSignal.wait(timeout: .now() + waitInterval) == .timedOut { owsFailDebug("Call didn't start within \(waitInterval) seconds. Continuing anyway, expecting to be killed.") - // We want to make sure we increment the semaphore exactly once per call to reset state - // for the next call. If we timed-out on the semaphore, we could race with another thread - // signaling the semaphore at this instant. We consult the atomic bool before re-incrementing. - if isWaitingForSignal.tryToClearFlag() { - pendingCallSignal.signal() + // To prepare for the next VoIP push, we need to make sure that when we leave this function: + // - isWaitingForSignal is set false + // - pendingCallSignal has a count of 0 + // + // There's a tiny race where we timeout then immediately after the call tries to wake + // the signal. To guard against this, we check the isWaitingForSignal. If it's been cleared, + // we know another thread must have signaled, so we should wait on our semaphore to decrement + // it back to zero. + if isWaitingForSignal.tryToClearFlag() == false { + pendingCallSignal.wait() } } Logger.info("Returning back to PushKit. Good luck! \(callRelayPayload)") From 4f3c72dbe7b5219fa78d5b73c8f21f331841ea89 Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Fri, 24 Sep 2021 12:50:01 -0700 Subject: [PATCH 5/7] Add a bunch more comments explaining the concurrency logic I had written most of this in PR comments but was afraid it'd be too verbose for comments. PR feedback said this was useful enough to include inline. --- .../environment/PushRegistrationManager.swift | 86 ++++++++++++++----- 1 file changed, 65 insertions(+), 21 deletions(-) diff --git a/Signal/src/environment/PushRegistrationManager.swift b/Signal/src/environment/PushRegistrationManager.swift index 5fc3666c4e..9891c141b6 100644 --- a/Signal/src/environment/PushRegistrationManager.swift +++ b/Signal/src/environment/PushRegistrationManager.swift @@ -24,8 +24,10 @@ public enum PushRegistrationError: Error { SwiftSingletons.register(self) } + // Coordinates blocking of the calloutQueue while we wait for an incoming call private let pendingCallSignal = DispatchSemaphore(value: 0) private let isWaitingForSignal = AtomicBool(false) + // Private callout queue that we can use to synchronously wait for our call to start // TODO: Rewrite call message routing to be able to synchronously report calls private let calloutQueue = DispatchQueue( @@ -63,6 +65,13 @@ public enum PushRegistrationError: Error { } public func didFinishReportingIncomingCall() { + owsAssertDebug(CurrentAppContext().isMainApp) + + // If we successfully clear the flag, we know we have someone waiting on the calloutQueue + // They may be blocked, in which case the signal will wake them up + // They could also have timed out, in which case they'll detect the cleared flag and decrement + // Either way, we should only signal if we can clear the flag, otherwise the extra increment will + // prevent the calloutQueue from blocking in the future. if isWaitingForSignal.tryToClearFlag() { pendingCallSignal.signal() } @@ -107,9 +116,17 @@ public enum PushRegistrationError: Error { // MARK: PKPushRegistryDelegate - voIP Push Token public func pushRegistry(_ registry: PKPushRegistry, didReceiveIncomingPushWith payload: PKPushPayload, for type: PKPushType) { - dispatchPrecondition(condition: .onQueue(calloutQueue)) + assertOnQueue(calloutQueue) + owsAssertDebug(CurrentAppContext().isMainApp) owsAssertDebug(type == .voIP) + // Concurrency invariants: + // At the start of this function: isWaitingForSignal: false. pendingCallSignal: +0. + // During this function (if a call message): isWaitingForSignal: true. pendingCallSignal: +{0,1}. + // Before returning: isWaitingForSignal: false. pendingCallSignal: +0. + owsAssertDebug(isWaitingForSignal.get() == false) + // owsAssertDebug(pendingCallSignal.count == 0) // Not exposed so we can't actually assert this. + let callRelayPayload = CallMessagePushPayload(payload.dictionaryPayload) if let callRelayPayload = callRelayPayload { Logger.info("Received VoIP push from the NSE: \(callRelayPayload)") @@ -146,35 +163,62 @@ public enum PushRegistrationError: Error { } if let callRelayPayload = callRelayPayload { - // We need to handle the incoming call push in the same runloop as it was delivered - // RingRTC will callback to us async to start the call after we hand them the call message, - // but once we return from here it's too late. + // iOS will kill our app and refuse to launch it again for an incoming call if we return from + // this function without reporting an incoming call. // - // We'll probably need RingRTC changes to handle this properly and synchronously. Until then, - // let's just block the thread for a bit until we hear back that the call was started.. + // You may see a crash here: -[PKPushRegistry _terminateAppIfThereAreUnhandledVoIPPushes] + // Or a log message like: + // > "Apps receving VoIP pushes must post an incoming call via CallKit in the same run loop as + // pushRegistry:didReceiveIncomingPushWithPayload:forType:[withCompletionHandler:] without delay" + // > "Killing app because it never posted an incoming call to the system after receiving a PushKit VoIP push." + // + // We should be better about handling these pushes faster and synchronously, but for now we + // can get away with just block this thread and wait for a call to be reported to signal us. Logger.info("Waiting for call to start: \(callRelayPayload)") let waitInterval = DispatchTimeInterval.seconds(5) - - if pendingCallSignal.wait(timeout: .now() + waitInterval) == .timedOut { + let didTimeout = (pendingCallSignal.wait(timeout: .now() + waitInterval) == .timedOut) + if didTimeout { owsFailDebug("Call didn't start within \(waitInterval) seconds. Continuing anyway, expecting to be killed.") - // To prepare for the next VoIP push, we need to make sure that when we leave this function: - // - isWaitingForSignal is set false - // - pendingCallSignal has a count of 0 - // - // There's a tiny race where we timeout then immediately after the call tries to wake - // the signal. To guard against this, we check the isWaitingForSignal. If it's been cleared, - // we know another thread must have signaled, so we should wait on our semaphore to decrement - // it back to zero. - if isWaitingForSignal.tryToClearFlag() == false { - pendingCallSignal.wait() - } } + + // Three cases that we need to account for. + // In all of these cases, we need to make sure that we return from this function with + // Semaphore: +0. isWaitingForSignal: false. + switch (didTimeout: didTimeout, didClearFlag: isWaitingForSignal.tryToClearFlag()) { + + // 1. We're successfully signaled by a reported call: + case (didTimeout: false, let didClearFlag): + // If we've been signaled, another thread must have called `didFinishReportingIncomingCall` + // It should have already cleared the flag for us, so let's assert that we haven't: + owsAssertDebug(didClearFlag == false) + // It should have also signaled the semaphore to +1. Our successful wait would have decremented back to +0. + // Invariant restored ✅: Semaphore: +0. isWaitingForSignal: false + + // 2. A call isn't reported in time, so we timeout before another thread calls `didFinishReportingIncomingCall` + case (didTimeout: true, didClearFlag: true): + // We successfully cleared the flag, so we know the semaphore cannot be incremented at this point. + // Invariant restored ✅: Semaphore: +0. isWaitingForSignal: false + break + + // 3. A race. We timeout at the same time that another thread tries to signal us + case (didTimeout: true, didClearFlag: false): + // We failed to clear the flag, so another thread beat us to clearing it by calling `didFinishReportingIncomingCall` + // This means that the semaphore is either at a +1 count, or is about to be signaled to +1 + // We can safely wait to re-decrement the semaphore: + + // Semaphore: +1. isWaitingForSema: false + pendingCallSignal.wait() + // Invariant restored ✅: Semaphore: +0. isWaitingForSignal: false + } + + owsAssertDebug(isWaitingForSignal.get() == false) + // owsAssertDebug(pendingCallSignal.count == 0) // Not exposed so we can't actually assert this. Logger.info("Returning back to PushKit. Good luck! \(callRelayPayload)") } } public func pushRegistry(_ registry: PKPushRegistry, didUpdate credentials: PKPushCredentials, for type: PKPushType) { - dispatchPrecondition(condition: .onQueue(calloutQueue)) + assertOnQueue(calloutQueue) Logger.info("") owsAssertDebug(type == .voIP) owsAssertDebug(credentials.type == .voIP) @@ -186,7 +230,7 @@ public enum PushRegistrationError: Error { public func pushRegistry(_ registry: PKPushRegistry, didInvalidatePushTokenFor type: PKPushType) { // It's not clear when this would happen. We've never previously handled it, but we should at // least start learning if it happens. - dispatchPrecondition(condition: .onQueue(calloutQueue)) + assertOnQueue(calloutQueue) owsFailDebug("Invalid state") } From d32358007f744ad8507c391a42fb39937ad31215 Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Fri, 24 Sep 2021 13:00:05 -0700 Subject: [PATCH 6/7] Restore some log messages post-rebase --- SignalServiceKit/src/Messages/MessageFetcherJob.swift | 2 ++ SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift | 2 ++ 2 files changed, 4 insertions(+) diff --git a/SignalServiceKit/src/Messages/MessageFetcherJob.swift b/SignalServiceKit/src/Messages/MessageFetcherJob.swift index 97a8c33492..53197077b0 100644 --- a/SignalServiceKit/src/Messages/MessageFetcherJob.swift +++ b/SignalServiceKit/src/Messages/MessageFetcherJob.swift @@ -322,6 +322,8 @@ public class MessageFetcherJob: NSObject { if shouldAck { Self.messageFetcherJob.acknowledgeDelivery(envelopeInfo: envelopeInfo) + } else { + Logger.info("Skipping ack of message with timestamp \(envelope.timestamp) because of error: \(String(describing: processingError))") } } } catch { diff --git a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift index fd7657fde0..c9ea4af63f 100644 --- a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift +++ b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift @@ -550,6 +550,8 @@ public class OWSWebSocket: NSObject { if shouldAck { self.sendWebSocketMessageAcknowledgement(message, currentWebSocket: currentWebSocket) + } else { + Logger.info("Skipping ack of message with serverTimestamp \(serverTimestamp) because of error: \(String(describing: processingError))") } owsAssertDebug(backgroundTask != nil) From f4247dc91bcabf0ac2df8d7049dce70844bde342 Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Fri, 24 Sep 2021 13:33:15 -0700 Subject: [PATCH 7/7] Only drain head-of-batch envelopes when processing a batch --- .../src/Messages/MessageProcessor.swift | 40 ++++++++----------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/SignalServiceKit/src/Messages/MessageProcessor.swift b/SignalServiceKit/src/Messages/MessageProcessor.swift index d1a59af6d7..be2f2ca5d1 100644 --- a/SignalServiceKit/src/Messages/MessageProcessor.swift +++ b/SignalServiceKit/src/Messages/MessageProcessor.swift @@ -260,11 +260,16 @@ public class MessageProcessor: NSObject { guard !self.isDrainingPendingEnvelopes else { return } self.isDrainingPendingEnvelopes = true self.drainNextBatch() + self.isDrainingPendingEnvelopes = false + if self.pendingEnvelopes.isEmpty { + NotificationCenter.default.postNotificationNameAsync(Self.messageProcessorDidFlushQueue, object: nil) + } } } private func drainNextBatch() { assertOnQueue(serialQueue) + owsAssertDebug(isDrainingPendingEnvelopes) let shouldContinue: Bool = autoreleasepool { // We want a value that is just high enough to yield perf benefits. @@ -278,35 +283,29 @@ public class MessageProcessor: NSObject { let batchEnvelopes = batch.batchEnvelopes let pendingEnvelopesCount = batch.pendingEnvelopesCount - guard !batchEnvelopes.isEmpty else { - isDrainingPendingEnvelopes = false + guard !batchEnvelopes.isEmpty, messagePipelineSupervisor.isMessageProcessingPermitted else { if DebugFlags.internalLogging { Logger.info("Processing complete: \(self.queuedContentCount).") } - NotificationCenter.default.postNotificationNameAsync(Self.messageProcessorDidFlushQueue, object: nil) return false } Logger.info("Processing batch of \(batchEnvelopes.count)/\(pendingEnvelopesCount) received envelope(s).") - var skippedEnvelopes: [PendingEnvelope] = [] + var processedEnvelopes: [PendingEnvelope] = [] SDSDatabaseStorage.shared.write { transaction in - batchEnvelopes.forEach { + for envelope in batchEnvelopes { if messagePipelineSupervisor.isMessageProcessingPermitted { - self.processEnvelope($0, transaction: transaction) + self.processEnvelope(envelope, transaction: transaction) + processedEnvelopes.append(envelope) } else { - skippedEnvelopes.append($0) + // If we're skipping one message, we have to skip them all to preserve ordering + // Next time around we can process the skipped messages in order + break } } } - - // Remove the processed envelopes from the pending list. - if skippedEnvelopes.isEmpty { - pendingEnvelopes.removeProcessedBatch(batch) - } else { - pendingEnvelopes.removePartiallySuccessfulBatch(batch, failedEnvelopes: skippedEnvelopes) - } - + pendingEnvelopes.removeProcessedEnvelopes(processedEnvelopes) return true } @@ -759,15 +758,14 @@ public class PendingEnvelopes { } } - fileprivate func removePartiallySuccessfulBatch(_ batch: Batch, failedEnvelopes: [PendingEnvelope]) { + fileprivate func removeProcessedEnvelopes(_ processedEnvelopes: [PendingEnvelope]) { unfairLock.withLock { - let batchEnvelopes = batch.batchEnvelopes - guard pendingEnvelopes.count > batchEnvelopes.count else { + guard pendingEnvelopes.count > processedEnvelopes.count else { pendingEnvelopes = [] return } let oldCount = pendingEnvelopes.count - pendingEnvelopes = failedEnvelopes + Array(pendingEnvelopes.suffix(from: batchEnvelopes.count)) + pendingEnvelopes = Array(pendingEnvelopes.suffix(from: processedEnvelopes.count)) let newCount = pendingEnvelopes.count if DebugFlags.internalLogging { Logger.info("\(oldCount) -> \(newCount)") @@ -775,10 +773,6 @@ public class PendingEnvelopes { } } - fileprivate func removeProcessedBatch(_ batch: Batch) { - removePartiallySuccessfulBatch(batch, failedEnvelopes: []) - } - fileprivate func enqueue(decryptedEnvelope: DecryptedEnvelope) { unfairLock.withLock { let oldCount = pendingEnvelopes.count