Fixing recovered logs not having resources from the session (#66)

Sources/EmbraceCore/Embrace.swift

@@ -209,13 +209,11 @@ To start the SDK you first need to configure it using an `Embrace.Options` insta
                         storage: self?.storage,
                         upload: self?.upload,
                         otel: self,
+                        logController: self?.logController,
                         currentSessionId: self?.sessionController.currentSession?.id,
                         crashReporter: self?.captureServices.crashReporter
                     )
 
-                    // upload persisted logs
-                    self?.logController.uploadAllPersistedLogs()
-
                     // retry any remaining cached upload data
                     self?.upload?.retryCachedData()
                 }

Sources/EmbraceCore/Internal/Logs/Exporter/DefaultLogBatcher.swift

@@ -13,6 +13,7 @@ protocol LogBatcherDelegate: AnyObject {
 
 protocol LogBatcher {
     func addLogRecord(logRecord: LogRecord)
+    func renewBatch(withLogs logRecords: [LogRecord])
 }
 
 class DefaultLogBatcher: LogBatcher {
@@ -50,7 +51,7 @@ class DefaultLogBatcher: LogBatcher {
     }
 }
 
-private extension DefaultLogBatcher {
+internal extension DefaultLogBatcher {
     func renewBatch(withLogs logRecords: [LogRecord] = []) {
         processorQueue.async {
             guard let batch = self.batch else {

Sources/EmbraceCore/Internal/Logs/Exporter/StorageEmbraceLogExporter.swift

@@ -26,6 +26,22 @@ class StorageEmbraceLogExporter: LogRecordExporter {
         self.state = state
         self.logBatcher = logBatcher
         self.validation = LogDataValidation(validators: validators)
+
+        NotificationCenter.default.addObserver(
+            self,
+            selector: #selector(onSessionEnd),
+            name: .embraceSessionWillEnd,
+            object: nil
+        )
+    }
+
+    deinit {
+        NotificationCenter.default.removeObserver(self)
+    }
+
+    @objc func onSessionEnd(noticication: Notification) {
+        // forcefully start a new batch of logs when a session ends
+        logBatcher.renewBatch(withLogs: [])
     }
 
     func export(logRecords: [ReadableLogRecord], explicitTimeout: TimeInterval?) -> ExportResult {

Sources/EmbraceCore/Internal/Logs/LogController.swift

@@ -6,6 +6,7 @@ import Foundation
 import EmbraceStorageInternal
 import EmbraceUploadInternal
 import EmbraceCommonInternal
+import EmbraceSemantics
 
 protocol LogControllable: LogBatcherDelegate {
     func uploadAllPersistedLogs()
@@ -63,23 +64,40 @@ private extension LogController {
         guard batches.count > 0 else {
             return
         }
-        guard let sessionId = sessionController?.currentSession?.id else {
-            return
-        }
 
-        do {
-            let resourcePayload = try createResourcePayload(sessionId: sessionId)
-            let metadataPayload = try createMetadataPayload(sessionId: sessionId)
+        for batch in batches {
+            do {
+                guard batch.logs.count > 0 else {
+                    continue
+                }
+
+                // Since we always end batches when a session ends
+                // all the logs still in storage when the app starts should come
+                // from the last session before the app closes.
+                //
+                // We grab the first valid sessionId from the stored logs
+                // and assume all of them come from the same session.
+                //
+                // If we can't find a sessionId, we use the processId instead
+
+                var sessionId: SessionIdentifier?
+                if let log = batch.logs.first(where: { $0.attributes[LogSemantics.keySessionId] != nil }) {
+                    sessionId = SessionIdentifier(string: log.attributes[LogSemantics.keySessionId]?.description)
+                }
+
+                let processId = batch.logs[0].processIdentifier
+
+                let resourcePayload = try createResourcePayload(sessionId: sessionId, processId: processId)
+                let metadataPayload = try createMetadataPayload(sessionId: sessionId, processId: processId)
 
-            batches.forEach { batch in
                 send(
                     logs: batch.logs,
                     resourcePayload: resourcePayload,
                     metadataPayload: metadataPayload
                 )
+            } catch let exception {
+                Error.couldntCreatePayload(reason: exception.localizedDescription).log()
             }
-        } catch let exception {
-            Error.couldntCreatePayload(reason: exception.localizedDescription).log()
         }
     }
 
@@ -137,24 +155,41 @@ private extension LogController {
         return batches
     }
 
-    func createResourcePayload(sessionId: SessionIdentifier) throws -> ResourcePayload {
+    func createResourcePayload(sessionId: SessionIdentifier?,
+                               processId: ProcessIdentifier = ProcessIdentifier.current
+    ) throws -> ResourcePayload {
         guard let storage = storage else {
             throw Error.couldntAccessStorageModule
         }
-        let resources = try storage.fetchResourcesForSessionId(sessionId)
+
+        var resources: [MetadataRecord] = []
+
+        if let sessionId = sessionId {
+            resources = try storage.fetchResourcesForSessionId(sessionId)
+        } else {
+            resources = try storage.fetchResourcesForProcessId(processId)
+        }
+
         return ResourcePayload(from: resources)
     }
 
-    func createMetadataPayload(sessionId: SessionIdentifier) throws -> MetadataPayload {
+    func createMetadataPayload(sessionId: SessionIdentifier?,
+                               processId: ProcessIdentifier = ProcessIdentifier.current
+    ) throws -> MetadataPayload {
         guard let storage = storage else {
             throw Error.couldntAccessStorageModule
         }
 
         var metadata: [MetadataRecord] = []
-        let properties = try storage.fetchCustomPropertiesForSessionId(sessionId)
-        let tags = try storage.fetchPersonaTagsForSessionId(sessionId)
-        metadata.append(contentsOf: properties)
-        metadata.append(contentsOf: tags)
+
+        if let sessionId = sessionId {
+            let properties = try storage.fetchCustomPropertiesForSessionId(sessionId)
+            let tags = try storage.fetchPersonaTagsForSessionId(sessionId)
+            metadata.append(contentsOf: properties)
+            metadata.append(contentsOf: tags)
+        } else {
+            metadata = try storage.fetchPersonaTagsForProcessId(processId)
+        }
 
         return MetadataPayload(from: metadata)
     }

Sources/EmbraceCore/Public/Embrace+OTel.swift

@@ -153,7 +153,7 @@ extension Embrace: EmbraceOpenTelemetry {
          However that would cause to always add a frame to the stacktrace.
          */
         if stackTraceBehavior == .default && (severity == .warn || severity == .error) {
-            var stackTrace: [String] = Thread.callStackSymbols
+            let stackTrace: [String] = Thread.callStackSymbols
             attributesBuilder.addStackTrace(stackTrace)
         }
 

Sources/EmbraceCore/Session/DataRecovery/UnsentDataHandler.swift

@@ -13,6 +13,7 @@ class UnsentDataHandler {
         storage: EmbraceStorage?,
         upload: EmbraceUpload?,
         otel: EmbraceOpenTelemetry?,
+        logController: LogControllable? = nil,
         currentSessionId: SessionIdentifier? = nil,
         crashReporter: CrashReporter? = nil
     ) {
@@ -22,6 +23,9 @@ class UnsentDataHandler {
             return
         }
 
+        // send any logs in storage first before we clean up the resources
+        logController?.uploadAllPersistedLogs()
+
         // if we have a crash reporter, we fetch the unsent crash reports first
         // and save their identifiers to the corresponding sessions
         if let crashReporter = crashReporter {

Tests/EmbraceCoreTests/Internal/Logs/Exporter/StorageEmbraceLogExporterTests.swift

@@ -108,6 +108,12 @@ class StorageEmbraceLogExporterTests: XCTestCase {
         thenBatchAdded(count: 0)
         thenResult(is: .success)
     }
+
+    func test_endBatch_onSessionEnd() {
+        givenStorageEmbraceLogExporter(initialState: .active)
+        whenSessionEnds()
+        thenBatchRenewed()
+    }
 }
 
 private extension StorageEmbraceLogExporterTests {
@@ -128,6 +134,10 @@ private extension StorageEmbraceLogExporterTests {
         result = sut.forceFlush()
     }
 
+    func whenSessionEnds() {
+        NotificationCenter.default.post(name: .embraceSessionWillEnd, object: nil)
+    }
+
     func thenState(is newState: StorageEmbraceLogExporter.State) {
         XCTAssertEqual(sut.state, newState)
     }
@@ -159,16 +169,27 @@ private extension StorageEmbraceLogExporterTests {
             randomLogData(body: body)
         }
     }
+
+    func thenBatchRenewed() {
+        XCTAssert(batcher.didCallRenewBatch)
+    }
 }
 
 class SpyLogBatcher: LogBatcher {
-    private (set) var didCallAddLogRecord: Bool = false
-    private (set) var addLogRecordInvocationCount: Int = 0
-    private (set) var logRecords = [LogRecord]()
+    private(set) var didCallAddLogRecord: Bool = false
+    private(set) var addLogRecordInvocationCount: Int = 0
+    private(set) var logRecords = [LogRecord]()
 
     func addLogRecord(logRecord: LogRecord) {
         didCallAddLogRecord = true
         addLogRecordInvocationCount += 1
         logRecords.append(logRecord)
     }
+
+    private(set) var didCallRenewBatch: Bool = false
+    private(set) var renewBatchInvocationCount: Int = 0
+    func renewBatch(withLogs logRecords: [LogRecord]) {
+        didCallRenewBatch = true
+        renewBatchInvocationCount += 1
+    }
 }

Tests/EmbraceCoreTests/Internal/Logs/LogControllerTests.swift

@@ -49,25 +49,39 @@ class LogControllerTests: XCTestCase {
     }
 
     func testHavingLogs_onSetup_fetchesResourcesFromStorage() throws {
-        givenStorage(withLogs: [randomLogRecord()])
+        let sessionId = SessionIdentifier.random
+        let log = randomLogRecord(sessionId: sessionId)
+
+        givenStorage(withLogs: [log])
         givenLogController()
         whenInvokingSetup()
-        try thenFetchesResourcesForCurrentSessionIdFromStorage()
+        try thenFetchesResourcesFromStorage(sessionId: sessionId)
     }
 
     func testHavingLogs_onSetup_fetchesMetadataFromStorage() throws {
-        givenStorage(withLogs: [randomLogRecord()])
+        let sessionId = SessionIdentifier.random
+        let log = randomLogRecord(sessionId: sessionId)
+
+        givenStorage(withLogs: [log])
         givenLogController()
         whenInvokingSetup()
-        try thenFetchesMetadataForCurrentSessionIdFromStorage()
+        try thenFetchesMetadataFromStorage(sessionId: sessionId)
     }
 
-    func testHavingLogsButNoSession_onSetup_wontTryToUploadAnything() {
-        givenSessionControllerWithoutSession()
-        givenStorage(withLogs: [randomLogRecord()])
+    func testHavingLogsWithNoSessionId_onSetup_fetchesResourcesFromStorage() throws {
+        let log = randomLogRecord()
+        givenStorage(withLogs: [log])
         givenLogController()
         whenInvokingSetup()
-        thenDoesntTryToUploadAnything()
+        try thenFetchesResourcesFromStorage(processId: log.processIdentifier)
+    }
+
+    func testHavingLogsWithNoSessionId_onSetup_fetchesMetadataFromStorage() throws {
+        let log = randomLogRecord()
+        givenStorage(withLogs: [log])
+        givenLogController()
+        whenInvokingSetup()
+        try thenFetchesMetadataFromStorage(processId: log.processIdentifier)
     }
 
     func testHavingLogsForLessThanABatch_onSetup_logUploaderShouldSendASingleBatch() {
@@ -115,19 +129,19 @@ class LogControllerTests: XCTestCase {
     func testHavingLogs_onBatchFinished_fetchesResourcesFromStorage() throws {
         givenLogController()
         whenInvokingBatchFinished(withLogs: [randomLogRecord()])
-        try thenFetchesResourcesForCurrentSessionIdFromStorage()
+        try thenFetchesResourcesFromStorage(sessionId: sessionController.currentSession?.id)
     }
 
     func testHavingLogs_onBatchFinished_fetchesMetadataFromStorage() throws {
         givenLogController()
         whenInvokingBatchFinished(withLogs: [randomLogRecord()])
-        try thenFetchesMetadataForCurrentSessionIdFromStorage()
+        try thenFetchesMetadataFromStorage(sessionId: sessionController.currentSession?.id)
     }
 
     func testHavingLogs_onBatchFinished_logUploaderShouldSendASingleBatch() throws {
         givenLogController()
         whenInvokingBatchFinished(withLogs: [randomLogRecord()])
-        try thenFetchesMetadataForCurrentSessionIdFromStorage()
+        try thenFetchesMetadataFromStorage(sessionId: sessionController.currentSession?.id)
     }
 
     func testHavingThrowingStorage_onBatchFinished_wontTryToUploadAnything() {
@@ -250,29 +264,45 @@ private extension LogControllerTests {
         XCTAssertFalse(unwrappedStorage.didCallRemoveLogs)
     }
 
-    func thenFetchesResourcesForCurrentSessionIdFromStorage() throws {
+    func thenFetchesResourcesFromStorage(sessionId: SessionIdentifier?) throws {
         let unwrappedStorage = try XCTUnwrap(storage)
-        let currentSessionId = sessionController.currentSession?.id
         XCTAssertTrue(unwrappedStorage.didCallFetchResourcesForSessionId)
-        XCTAssertEqual(unwrappedStorage.fetchResourcesForSessionIdReceivedParameter, currentSessionId)
+        XCTAssertEqual(unwrappedStorage.fetchResourcesForSessionIdReceivedParameter, sessionId)
     }
 
-    func thenFetchesMetadataForCurrentSessionIdFromStorage() throws {
+    func thenFetchesMetadataFromStorage(sessionId: SessionIdentifier?) throws {
         let unwrappedStorage = try XCTUnwrap(storage)
-        let currentSessionId = sessionController.currentSession?.id
         XCTAssertTrue(unwrappedStorage.didCallFetchCustomPropertiesForSessionId)
-        XCTAssertEqual(unwrappedStorage.fetchCustomPropertiesForSessionIdReceivedParameter, currentSessionId)
+        XCTAssertEqual(unwrappedStorage.fetchCustomPropertiesForSessionIdReceivedParameter, sessionId)
         XCTAssertTrue(unwrappedStorage.didCallFetchCustomPropertiesForSessionId)
-        XCTAssertEqual(unwrappedStorage.fetchPersonaTagsForSessionIdReceivedParameter, currentSessionId)
+        XCTAssertEqual(unwrappedStorage.fetchPersonaTagsForSessionIdReceivedParameter, sessionId)
     }
 
-    func randomLogRecord() -> LogRecord {
-        .init(
+    func thenFetchesResourcesFromStorage(processId: ProcessIdentifier) throws {
+        let unwrappedStorage = try XCTUnwrap(storage)
+        XCTAssertTrue(unwrappedStorage.didCallFetchResourcesForProcessId)
+        XCTAssertEqual(unwrappedStorage.fetchResourcesForProcessIdReceivedParameter, processId)
+    }
+
+    func thenFetchesMetadataFromStorage(processId: ProcessIdentifier) throws {
+        let unwrappedStorage = try XCTUnwrap(storage)
+        XCTAssertTrue(unwrappedStorage.didCallFetchPersonaTagsForProcessId)
+        XCTAssertEqual(unwrappedStorage.fetchPersonaTagsForProcessIdReceivedParameter, processId)
+    }
+
+    func randomLogRecord(sessionId: SessionIdentifier? = nil) -> LogRecord {
+
+        var attributes: [String: PersistableValue] = [:]
+        if let sessionId = sessionId {
+            attributes["emb.session_id"] = PersistableValue(sessionId.toString)
+        }
+
+        return LogRecord(
             identifier: .random,
             processIdentifier: .random,
             severity: .info,
             body: UUID().uuidString,
-            attributes: .empty()
+            attributes: attributes
         )
     }