Merge branch 'mlin/PR/RepeatAckStarvation' into release/5.21.2

SignalServiceKit/src/Messages/MessageFetcherJob.swift

@@ -562,6 +562,24 @@ private class MessageAckOperation: OWSOperation {
     private let envelopeInfo: EnvelopeInfo
     private let pendingAck: PendingTask
 
+    // A heuristic to quickly filter out multiple ack attempts for the same message
+    // This doesn't affect correctness, just tries to guard against backing up our operation queue with repeat work
+    static private var inFlightAcks = AtomicSet<String>()
+    private var didRecordAckId = false
+    private var inFlightAckId: String {
+        // All messages *should* have a guid, but we'll handle things correctly if they don't
+        owsAssertDebug(envelopeInfo.serverGuid?.nilIfEmpty != nil)
+
+        if let serverGuid = envelopeInfo.serverGuid?.nilIfEmpty {
+            return serverGuid
+        } else if let sourceUuid = envelopeInfo.sourceAddress?.uuid {
+            return "\(sourceUuid.uuidString)_\(envelopeInfo.timestamp)"
+        } else {
+            // This *could* collide, but we don't have enough info to ack the message anyway. So it should be fine.
+            return "\(envelopeInfo.serviceTimestamp)"
+        }
+    }
+
     fileprivate required init(envelopeInfo: EnvelopeInfo,
                               pendingAck: PendingTask) {
         self.envelopeInfo = envelopeInfo
@@ -574,6 +592,14 @@ private class MessageAckOperation: OWSOperation {
         // MessageAckOperation must have a higher priority than than the
         // operations used to flush the ack operation queue.
         self.queuePriority = .high
+        if Self.inFlightAcks.contains(inFlightAckId) {
+            Logger.info("Cancelling new ack operation for \(envelopeInfo). Duplicate ack already enqueued")
+            pendingAck.complete()
+            self.cancel()
+        } else {
+            Self.inFlightAcks.insert(inFlightAckId)
+            didRecordAckId = true
+        }
     }
 
     public override func run() {
@@ -611,14 +637,11 @@ private class MessageAckOperation: OWSOperation {
     }
 
     @objc
-    public override func didSucceed() {
-        super.didSucceed()
-        pendingAck.complete()
-    }
-
-    @objc
-    public override func didFail(error: Error) {
-        super.didFail(error: error)
+    public override func didComplete() {
+        super.didComplete()
+        if didRecordAckId {
+            Self.inFlightAcks.remove(inFlightAckId)
+        }
         pendingAck.complete()
     }
 }

SignalServiceKit/src/Messages/MessageProcessor.swift

@@ -233,7 +233,7 @@ public class MessageProcessor: NSObject {
     // This is tricky since there are multiple variables (e.g. network
     // perf affects fetch, CPU perf affects processing).
     public var hasSomeQueuedContent: Bool {
-        queuedContentCount >= 25
+        queuedContentCount >= 10
     }
 
     public var queuedContentCount: Int {

SignalServiceKit/src/Util/OWSOperation.h

@@ -53,6 +53,9 @@ typedef NS_ENUM(NSInteger, OWSOperationState) {
 // Called at most one time, once retry is no longer possible.
 - (void)didFailWithError:(NSError *)error NS_SWIFT_NAME(didFail(error:));
 
+// Called exactly once after operation has moved to OWSOperationStateFinished
+- (void)didComplete;
+
 // How long to wait before retry, if possible
 - (NSTimeInterval)retryInterval;
 

SignalServiceKit/src/Util/OWSOperation.m

@@ -111,6 +111,13 @@ NSString *const OWSOperationKeyIsFinished = @"isFinished";
     // Override in subclass if necessary
 }
 
+// Called exactly once after the operation is marked complete, either with success, failure, or cancellation
+- (void)didComplete
+{
+    // no-op
+    // Override in subclass if necessary
+}
+
 #pragma mark - NSOperation overrides
 
 - (NSString *)eventId
@@ -271,6 +278,7 @@ NSString *const OWSOperationKeyIsFinished = @"isFinished";
 
     [self didChangeValueForKey:OWSOperationKeyIsExecuting];
     [self didChangeValueForKey:OWSOperationKeyIsFinished];
+    [self didComplete];
 
     [BenchManager completeEventWithEventId:self.eventId];
 }