feat(events): persist events to disk so they survive the process (tier 3)

LaunchDarkly/GeneratedCode/mocks.generated.swift

@@ -38,53 +38,6 @@ final class DarklyStreamingProviderMock: DarklyStreamingProvider {
     }
 }
 
-// MARK: - DiagnosticCachingMock
-final class DiagnosticCachingMock: DiagnosticCaching {
-
-    var getDiagnosticIdCallCount = 0
-    var getDiagnosticIdCallback: (() throws -> Void)?
-    var getDiagnosticIdReturnValue: DiagnosticId!
-    func getDiagnosticId() -> DiagnosticId {
-        getDiagnosticIdCallCount += 1
-        try! getDiagnosticIdCallback?()
-        return getDiagnosticIdReturnValue
-    }
-
-    var getCurrentStatsAndResetCallCount = 0
-    var getCurrentStatsAndResetCallback: (() throws -> Void)?
-    var getCurrentStatsAndResetReturnValue: DiagnosticStats!
-    func getCurrentStatsAndReset() -> DiagnosticStats {
-        getCurrentStatsAndResetCallCount += 1
-        try! getCurrentStatsAndResetCallback?()
-        return getCurrentStatsAndResetReturnValue
-    }
-
-    var incrementDroppedEventCountCallCount = 0
-    var incrementDroppedEventCountCallback: (() throws -> Void)?
-    func incrementDroppedEventCount() {
-        incrementDroppedEventCountCallCount += 1
-        try! incrementDroppedEventCountCallback?()
-    }
-
-    var recordEventsInLastBatchCallCount = 0
-    var recordEventsInLastBatchCallback: (() throws -> Void)?
-    var recordEventsInLastBatchReceivedEventsInLastBatch: Int?
-    func recordEventsInLastBatch(eventsInLastBatch: Int) {
-        recordEventsInLastBatchCallCount += 1
-        recordEventsInLastBatchReceivedEventsInLastBatch = eventsInLastBatch
-        try! recordEventsInLastBatchCallback?()
-    }
-
-    var addStreamInitCallCount = 0
-    var addStreamInitCallback: (() throws -> Void)?
-    var addStreamInitReceivedStreamInit: DiagnosticStreamInit?
-    func addStreamInit(streamInit: DiagnosticStreamInit) {
-        addStreamInitCallCount += 1
-        addStreamInitReceivedStreamInit = streamInit
-        try! addStreamInitCallback?()
-    }
-}
-
 // MARK: - DiagnosticReportingMock
 final class DiagnosticReportingMock: DiagnosticReporting {
 
@@ -174,8 +127,8 @@ final class EnvironmentReportingMock: EnvironmentReporting {
     }
 }
 
-// MARK: - EventReportingMock
-final class EventReportingMock: EventReporting {
+// MARK: - EventProcessorMock
+final class EventProcessorMock: EventProcessor {
 
     var isOnlineSetCount = 0
     var setIsOnlineCallback: (() throws -> Void)?
@@ -230,6 +183,36 @@ final class EventReportingMock: EventReporting {
         flushReportingOutcomeReceivedCompletion = completion
         try! flushReportingOutcomeCallback?()
     }
+
+    var flushBypassingBackoffCallCount = 0
+    var flushBypassingBackoffCallback: (() throws -> Void)?
+    var flushBypassingBackoffReceivedCompletion: (FlushOutcomeClosure)?
+    func flushBypassingBackoff(completion: @escaping FlushOutcomeClosure) {
+        flushBypassingBackoffCallCount += 1
+        flushBypassingBackoffReceivedCompletion = completion
+        try! flushBypassingBackoffCallback?()
+    }
+
+    var flushAtCloseCallCount = 0
+    var flushAtCloseCallback: (() throws -> Void)?
+    func flushAtClose() {
+        flushAtCloseCallCount += 1
+        try! flushAtCloseCallback?()
+    }
+
+    var commitRecordedEventsCallCount = 0
+    var commitRecordedEventsCallback: (() throws -> Void)?
+    func commitRecordedEvents() {
+        commitRecordedEventsCallCount += 1
+        try! commitRecordedEventsCallback?()
+    }
+
+    var commitWherePersistingCallCount = 0
+    var commitWherePersistingCallback: (() throws -> Void)?
+    func commitWherePersisting() {
+        commitWherePersistingCallCount += 1
+        try! commitWherePersistingCallback?()
+    }
 }
 
 // MARK: - FeatureFlagCachingMock

LaunchDarkly/LaunchDarkly/Extensions/Data+CRC.swift

@@ -0,0 +1,15 @@
+import Foundation
+
+extension Crc32 {
+    /// Advance the current checksum with bytes already in memory, without wrapping them in `Data` first.
+    /// - parameter bytes: bytes to advance the checksum
+    mutating func advance(withBytes bytes: UnsafeRawBufferPointer) {
+        guard let base = bytes.baseAddress?.assumingMemoryBound(to: UInt8.self)
+        else { return }
+        if let fastCrc32 = Crc32.zLibCrc32 {
+            checksum = fastCrc32(checksum, base, UInt32(bytes.count))
+        } else {
+            checksum = slowCrc32(start: checksum, data: Data(bytes))
+        }
+    }
+}

LaunchDarkly/LaunchDarkly/Extensions/Data+Compression.swift

@@ -258,12 +258,12 @@ internal extension Data {
 
 /// Struct based type representing a Crc32 checksum.
 struct Crc32: CustomStringConvertible {
-    private static let zLibCrc32: ZLibCrc32FuncPtr? = loadCrc32fromZLib()
+    static let zLibCrc32: ZLibCrc32FuncPtr? = loadCrc32fromZLib()
 
     init() {}
 
     // C convention function pointer type matching the signature of `libz::crc32`
-    private typealias ZLibCrc32FuncPtr = @convention(c) (
+    typealias ZLibCrc32FuncPtr = @convention(c) (
         _ cks: UInt32,
         _ buf: UnsafePointer<UInt8>,
         _ len: UInt32
@@ -300,7 +300,7 @@ struct Crc32: CustomStringConvertible {
     /// Rudimentary fallback implementation of the crc32 checksum. This is only a backup used
     /// when zlib can't be found under '/usr/lib/libz.dylib'.
     /// - returns: crc32 checksum (4 byte)
-    private func slowCrc32(start: UInt32, data: Data) -> UInt32 {
+    func slowCrc32(start: UInt32, data: Data) -> UInt32 {
         return ~data.reduce(~start) { (crc: UInt32, next: UInt8) -> UInt32 in
             let tableOffset = (crc ^ UInt32(next)) & 0xff
             return lookUpTable[Int(tableOffset)] ^ crc >> 8

LaunchDarkly/LaunchDarkly/LDClient.swift

@@ -66,7 +66,7 @@ public class LDClient {
                 let oldValue = _isOnline
                 _isOnline = newValue
                 flagSynchronizer.isOnline = _isOnline
-                eventReporter.isOnline = _isOnline
+                eventProcessor.isOnline = _isOnline
                 diagnosticReporter.setMode(runMode, online: _isOnline)
                 if _isOnline != oldValue {
                     connectionInformation = ConnectionInformation.onlineSetCheck(connectionInformation: connectionInformation, ldClient: self, config: config, online: _isOnline)
@@ -247,39 +247,25 @@ public class LDClient {
      There is almost no reason to stop the LDClient. Normally, set the LDClient offline to stop communication with the LaunchDarkly servers. Stop the LDClient to stop recording events. There is no need to stop the LDClient prior to suspending, moving to the background, or terminating the app. The SDK will respond to these events as the system requires and as configured in LDConfig.
     */
     public func close() {
-        LDClient.instancesQueue.sync(flags: .barrier) {
-            LDClient.instances?.forEach { $1.internalClose() }
-            LDClient.instances = nil
+        // Closed outside `instancesQueue`: each close can wait on its final event commit, and every other call that
+        // reads `instances` would wait behind it.
+        let clients = LDClient.instancesQueue.sync(flags: .barrier) { () -> [LDClient] in
+            defer { LDClient.instances = nil }
+            return LDClient.instances.map { Array($0.values) } ?? []
         }
+        clients.forEach { $0.internalClose() }
     }
 
     private func internalClose() {
         os_log("%s stopping", log: config.logger, type: .debug, typeName(and: #function))
-        internalFlush()
+        eventProcessor.flushAtClose()
         internalSetOnline(false)
         hasStarted = false
         os_log("%s stopped", log: config.logger, type: .debug, typeName(and: #function))
     }
 
     @objc private func didEnterBackground() {
         os_log("%s", log: config.logger, type: .debug, typeName(and: #function))
-        // A backgrounded process is suspended once idle, so deliver inside an activity assertion. Only where suspension
-        // is real: macOS posts this notification whenever the application loses focus, and never suspends it for that.
-        #if os(iOS) || os(tvOS)
-        backgroundDelivery.run { [weak self] finished in
-            guard let self = self
-            else {
-                finished()
-                return
-            }
-            self.eventReporter.flushReportingOutcome { delivered in
-                if !delivered {
-                    os_log("%s events did not reach LaunchDarkly before suspension", log: self.config.logger, type: .debug, self.typeName(and: #function))
-                }
-                finished()
-            }
-        }
-        #endif
         Thread.performOnMain {
             runMode = .background
         }
@@ -292,9 +278,6 @@ public class LDClient {
         }
     }
 
-    /// One assertion per client, however many times the application is backgrounded while a delivery is running.
-    private let backgroundDelivery: BackgroundActivity
-
     let config: LDConfig
     /// Identifies this client's environment to a hook, without handing it the mobile key that identifies the
     /// environment to LaunchDarkly. Hashed once here rather than per evaluation, because a deduping hook asks for it on
@@ -475,7 +458,7 @@ public class LDClient {
 
         internalIdentifyQueue.sync {
             if self.context == updatedContext {
-                self.eventReporter.record(IdentifyEvent(context: self.context))
+                self.eventProcessor.record(IdentifyEvent(context: self.context))
                 completion?()
                 return
             }
@@ -525,7 +508,7 @@ public class LDClient {
                                                                    onSyncComplete: self.onFlagSyncComplete)
 
             if self.hasStarted {
-                self.eventReporter.record(IdentifyEvent(context: self.context))
+                self.eventProcessor.record(IdentifyEvent(context: self.context))
             }
 
             if self.isOnline && self.isInSupportedRunMode {
@@ -769,7 +752,7 @@ public class LDClient {
             key,
             String(describing: data),
             String(describing: metricValue))
-        eventReporter.record(event)
+        eventProcessor.record(event)
 
         executeAfterTrackHooks(key: key, data: data, metricValue: metricValue)
     }
@@ -979,7 +962,7 @@ public class LDClient {
     private(set) var flagCache: FeatureFlagCaching
     private(set) var flagSynchronizer: LDFlagSynchronizing
     var flagChangeNotifier: FlagChangeNotifying
-    private(set) var eventReporter: EventReporting
+    private(set) var eventProcessor: EventProcessor
     private(set) var environmentReporter: EnvironmentReporting
     private(set) var throttler: Throttling
     private(set) var diagnosticReporter: DiagnosticReporting
@@ -1035,14 +1018,13 @@ public class LDClient {
         throttler = self.serviceFactory.makeThrottler(environmentReporter: environmentReporter)
 
         config = configuration
-        backgroundDelivery = BackgroundActivity(reason: "LaunchDarkly event delivery", logger: configuration.logger)
         context = startContext ?? LDContext()
 
         if config.autoEnvAttributes {
             context = AutoEnvContextModifier(environmentReporter: environmentReporter, logger: config.logger).modifyContext(context)
         }
 
-        var hookState: IdentifyHookState? = nil
+        var hookState: IdentifyHookState?
         if !hooks.isEmpty {
             let seriesContext = IdentifySeriesContext(context: context, methodName: "init")
             let seriesData = hooks.map { hook in hook.beforeIdentify(seriesContext: seriesContext, seriesData: EvaluationSeriesData()) }
@@ -1054,7 +1036,7 @@ public class LDClient {
         // Reports over the logger rather than over this client, which is not usable until every stored property
         // below has a value, so that the reporter can be built once rather than built and replaced.
         let logger = configuration.logger
-        eventReporter = self.serviceFactory.makeEventReporter(config: config, service: service) { result in
+        eventProcessor = self.serviceFactory.makeEventProcessor(config: config, service: service) { result in
             LDClient.logEventSyncComplete(result, logger)
         }
         connectionInformation = self.serviceFactory.makeConnectionInformation()
@@ -1069,6 +1051,7 @@ public class LDClient {
         if let backgroundNotification = SystemCapabilities.backgroundNotification {
             NotificationCenter.default.addObserver(self, selector: #selector(didEnterBackground), name: backgroundNotification, object: nil)
         }
+        LDClient.observeBackgroundingForEvents()
         if let foregroundNotification = SystemCapabilities.foregroundNotification {
             NotificationCenter.default.addObserver(self, selector: #selector(willEnterForeground), name: foregroundNotification, object: nil)
         }
@@ -1078,12 +1061,13 @@ public class LDClient {
         service.resetFlagResponseCache(etag: cachedData.etag)
         // Reports to this client, so it could not be given at construction: `self` is only usable from here.
         flagSynchronizer.onSyncComplete = onFlagSyncComplete
+        giveFlagValuesTheNetworkFirst()
 
         if let cachedFlags = cachedData.items, !cachedFlags.isEmpty {
             flagStore.replaceStore(newStoredItems: cachedFlags)
         }
 
-        eventReporter.record(IdentifyEvent(context: context))
+        eventProcessor.record(IdentifyEvent(context: context))
         self.connectionInformation = ConnectionInformation.uncacheConnectionInformation(config: config, ldClient: self, clientServiceFactory: self.serviceFactory)
 
         internalSetOnline(configuration.startOnline) {
@@ -1098,12 +1082,35 @@ public class LDClient {
 
 extension LDClient: TypeIdentifying { }
 
-private extension LDClient {
-    func onEventSyncComplete(result: SynchronizingError?) {
-        if let synchronizingError = result {
-            os_log("%s result: %s", log: config.logger, type: .debug, typeName(and: #function), String(describing: synchronizingError))
-        } else {
-            os_log("%s result: success", log: config.logger, type: .debug, typeName(and: #function))
+extension LDClient {
+    /// Has the diagnostic init event wait for the event processor's launch gate, as a previous run's events do, whether
+    /// or not events persist.
+    fileprivate func giveFlagValuesTheNetworkFirst() {
+        if let eventProcessor = eventProcessor as? DirectEventProcessor {
+            (diagnosticReporter as? DiagnosticReporter)?.sendsInitAfter = eventProcessor.whenLaunchGateOpens
+        }
+    }
+
+    /// Observes the notification once for the events of every environment, however many clients are made.
+    private static let backgroundEvents: BackgroundEventDelivery? = {
+        guard let backgroundNotification = SystemCapabilities.backgroundNotification
+        else { return nil }
+        let delivery = BackgroundEventDelivery(environments: LDClient.currentEnvironments)
+        // Run on the posting thread, so that every environment has committed by the time the notification returns.
+        NotificationCenter.default.addObserver(forName: backgroundNotification, object: nil, queue: nil) { _ in
+            delivery.applicationDidEnterBackground()
+        }
+        return delivery
+    }()
+
+    fileprivate static func observeBackgroundingForEvents() {
+        _ = backgroundEvents
+    }
+
+    private static func currentEnvironments() -> [BackgroundEventDelivery.Environment] {
+        let clients = instancesQueue.sync { instances } ?? [:]
+        return clients.map { name, client in
+            BackgroundEventDelivery.Environment(name: name, eventProcessor: client.eventProcessor, logger: client.config.logger)
         }
     }
 }