diff --git a/NotificationServiceExtension/NSECallMessageHandler.swift b/NotificationServiceExtension/NSECallMessageHandler.swift index 1b7c07fcb1..ed22f6c6a1 100644 --- a/NotificationServiceExtension/NSECallMessageHandler.swift +++ b/NotificationServiceExtension/NSECallMessageHandler.swift @@ -57,7 +57,15 @@ public class NSECallMessageHandler: NSObject, OWSCallMessageHandler { transaction: transaction ) - CXProvider.reportNewIncomingVoIPPushPayload(payload) { error in + // 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 { 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..9891c141b6 100644 --- a/Signal/src/environment/PushRegistrationManager.swift +++ b/Signal/src/environment/PushRegistrationManager.swift @@ -24,6 +24,17 @@ 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( + label: "org.whispersystems.signal.PKPushRegistry", + autoreleaseFrequency: .workItem + ) + private var vanillaTokenPromise: Promise? private var vanillaTokenFuture: Future? @@ -53,6 +64,19 @@ 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() + } + } + // MARK: Vanilla push token @objc @@ -92,36 +116,109 @@ 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)") + assertOnQueue(calloutQueue) + owsAssertDebug(CurrentAppContext().isMainApp) 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)") + + // 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)") + 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 { + // 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. + // + // 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) + let didTimeout = (pendingCallSignal.wait(timeout: .now() + waitInterval) == .timedOut) + if didTimeout { + owsFailDebug("Call didn't start within \(waitInterval) seconds. Continuing anyway, expecting to be killed.") + } + + // 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) { + assertOnQueue(calloutQueue) Logger.info("") owsAssertDebug(type == .voIP) owsAssertDebug(credentials.type == .voIP) @@ -133,6 +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. + assertOnQueue(calloutQueue) owsFailDebug("Invalid state") } @@ -231,7 +329,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 { 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/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 7e303a45c0..be2f2ca5d1 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() @@ -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,24 +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 processedEnvelopes: [PendingEnvelope] = [] SDSDatabaseStorage.shared.write { transaction in - batchEnvelopes.forEach { self.processEnvelope($0, transaction: transaction) } + for envelope in batchEnvelopes { + if messagePipelineSupervisor.isMessageProcessingPermitted { + self.processEnvelope(envelope, transaction: transaction) + processedEnvelopes.append(envelope) + } else { + // 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. - pendingEnvelopes.removeProcessedBatch(batch) - + pendingEnvelopes.removeProcessedEnvelopes(processedEnvelopes) return true } @@ -748,15 +758,14 @@ public class PendingEnvelopes { } } - fileprivate func removeProcessedBatch(_ batch: Batch) { + 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 = Array(pendingEnvelopes.suffix(from: batchEnvelopes.count)) + pendingEnvelopes = Array(pendingEnvelopes.suffix(from: processedEnvelopes.count)) let newCount = pendingEnvelopes.count if DebugFlags.internalLogging { Logger.info("\(oldCount) -> \(newCount)") diff --git a/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift b/SignalServiceKit/src/Network/WebSockets/OWSWebSocket.swift index 906e87c63c..c9ea4af63f 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!")") } @@ -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) @@ -573,7 +575,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 +598,7 @@ public class OWSWebSocket: NSObject { Logger.info("\(currentWebSocket.logPrefix) 3") } Self.serialQueue.async { - ackMessage(outcome) + ackMessage(outcome, serverDeliveryTimestamp) } } }