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

Merge branch 'mlin/PR/PushKitWaitForSetupV2'

Browse Source
This commit is contained in:
Michelle Linington 2021-09-24 18:36:56 -07:00
commit 4e3045cf54
8 changed files with 191 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
NotificationServiceExtension/NSECallMessageHandler.swift
View File

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

2
Signal/src/Calls/UserInterface/Individual/CallKit/CallKitCallUIAdaptee.swift
View File

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

146
Signal/src/environment/PushRegistrationManager.swift
View File

@ -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<Data>?
private var vanillaTokenFuture: Future<Data>?
@ -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<Void> 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<Void> 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

37
SignalMessaging/utils/CallMessageRelay.swift
View File

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

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

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

4
SignalServiceKit/src/Messages/MessagePipelineSupervisor.swift
View File

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

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

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

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

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