Fix race condition causing batch payload corruption

.gitignore

@@ -102,3 +102,4 @@ XCFrameworkOutput
 .idea
 .vscode
 .editorconfig
+test-output.log

Sources/Segment/Utilities/Storage/TransientDB.swift

@@ -12,30 +12,37 @@ public class TransientDB {
     // keeps items added in the order given.
     internal let syncQueue = DispatchQueue(label: "transientDB.sync")
     private let asyncAppend: Bool
-
+    // create a serial queue we can hit async so events still arrive in an expected order.
+    private let asyncQueue = DispatchQueue(label: "com.segment.transientdb.async", qos: .utility)
+
+
     public var hasData: Bool {
         var result: Bool = false
         syncQueue.sync {
             result = store.hasData
         }
         return result
     }
-    
+
     public var count: Int {
         var result: Int = 0
         syncQueue.sync {
             result = store.count
         }
         return result
     }
-    
+
     public var transactionType: DataTransactionType {
         return store.transactionType
     }
-    
+
     public init(store: any DataStore, asyncAppend: Bool = true) {
         self.store = store
-        self.asyncAppend = asyncAppend
+        if (store is MemoryStore) {
+            self.asyncAppend = false
+        } else {
+            self.asyncAppend = asyncAppend
+        }
     }
 
     public func reset() {
@@ -46,9 +53,13 @@ public class TransientDB {
 
     public func append(data: RawEvent) {
         if asyncAppend {
-            syncQueue.async { [weak self] in
+            // Dispatch to background thread, but execute synchronously on syncQueue
+            // This ensures FIFO ordering while keeping appends off the main thread
+            asyncQueue.async { [weak self] in
                 guard let self else { return }
-                store.append(data: data)
+                self.syncQueue.sync {
+                    self.store.append(data: data)
+                }
             }
         } else {
             syncQueue.sync { [weak self] in
@@ -59,6 +70,9 @@ public class TransientDB {
     }
 
     public func fetch(count: Int? = nil, maxBytes: Int? = nil) -> DataResult? {
+        // syncQueue is serial and all operations use .sync, ensuring FIFO ordering
+        // Appends still in-flight on global queue will execute after this fetch,
+        // and will start a new file (preventing corruption)
         var result: DataResult? = nil
         syncQueue.sync {
             result = store.fetch(count: count, maxBytes: maxBytes)

Tests/Segment-Tests/Analytics_Tests.swift

@@ -145,7 +145,7 @@ final class Analytics_Tests: XCTestCase {
             let expectation = XCTestExpectation(description: "MyDestination Expectation")
             let myDestination = MyDestination(disabled: true) {
                 expectation.fulfill()
-                print("called")
+                //print("called")
                 return true
             }
 
@@ -486,17 +486,32 @@ final class Analytics_Tests: XCTestCase {
 
         analytics.identify(userId: "brandon", traits: MyTraits(email: "blah@blah.com"))
 
-        let currentBatchCount = analytics.storage.read(.events)!.dataFiles!.count
+        // Wait for async append to complete before reading
+        // CI environments (especially simulators) need more time
+        Thread.sleep(forTimeInterval: 0.5)
+
+        let currentBatch = analytics.storage.read(.events)!
+        let currentBatchCount = currentBatch.dataFiles!.count
+
+        let expectation = XCTestExpectation(description: "flush completes")
+        analytics.flush {
+            expectation.fulfill()
+        }
+        wait(for: [expectation], timeout: 5.0)
 
-        analytics.flush()
         analytics.track(name: "test")
 
+        // Wait for async append to complete before reading
+        // CI environments (especially simulators) need more time
+        Thread.sleep(forTimeInterval: 0.5)
+
         let batches = analytics.storage.read(.events)!.dataFiles
         let newBatchCount = batches!.count
-        // 1 new temp file
+        // After flush, the first file is removed (uploaded or 400 error).
+        // So we should have exactly 1 file (from the track call), not currentBatchCount + 1
         XCTAssertTrue(
-            newBatchCount == currentBatchCount + 1,
-            "New Count (\(newBatchCount)) should be \(currentBatchCount) + 1")
+            newBatchCount == 1,
+            "New Count (\(newBatchCount)) should be 1 (file from track after flush removed previous file)")
     }
 
     func testEnabled() {
@@ -741,14 +756,14 @@ final class Analytics_Tests: XCTestCase {
     func testEnrichment() {
         var sourceHit: Bool = false
         let sourceEnrichment: EnrichmentClosure = { event in
-            print("source enrichment applied")
+            //print("source enrichment applied")
             sourceHit = true
             return event
         }
 
         var destHit: Bool = true
         let destEnrichment: EnrichmentClosure = { event in
-            print("destination enrichment applied")
+            //print("destination enrichment applied")
             destHit = true
             return event
         }
@@ -932,7 +947,8 @@ final class Analytics_Tests: XCTestCase {
                 return
             }
 
-            let analytics = Analytics(configuration: Configuration(writeKey: "networkTest"))
+            let analytics = Analytics(configuration: Configuration(writeKey: "networkTest")
+                .operatingMode(.synchronous))
 
             waitUntilStarted(analytics: analytics)
 

Tests/Segment-Tests/CompletionGroup_Tests.swift

@@ -31,38 +31,38 @@ final class CompletionGroup_Tests: XCTestCase {
 
         group.add { group in
             group.enter()
-            print("item1 - sleeping 10")
+            //print("item1 - sleeping 10")
             sleep(10)
-            print("item1 - done sleeping")
+            //print("item1 - done sleeping")
             group.leave()
         }
 
         group.add { group in
             group.enter()
-            print("item2 - launching an async task")
+            //print("item2 - launching an async task")
             DispatchQueue.global(qos: .background).async {
-                print("item2 - background, sleeping 5")
+                //print("item2 - background, sleeping 5")
                 sleep(5)
-                print("item2 - background, done sleeping")
+                //print("item2 - background, done sleeping")
                 group.leave()
             }
         }
 
         group.add { group in
             group.enter()
-            print("item3 - returning real quick")
+            //print("item3 - returning real quick")
             group.leave()
         }
 
         group.add { group in
-            print("item4 - not entering group")
+            //print("item4 - not entering group")
         }
 
         group.run(mode: .asynchronous) {
-            print("all items completed.")
+            //print("all items completed.")
         }
 
-        print("test exited.")
+        //print("test exited.")
     }*/
 
 }

Tests/Segment-Tests/FlushPolicy_Tests.swift

@@ -135,7 +135,10 @@ class FlushPolicyTests: XCTestCase {
 
         waitUntilStarted(analytics: analytics)
         analytics.track(name: "blah", properties: nil)
-
+
+        // Wait for async append to complete
+        Thread.sleep(forTimeInterval: 0.5)
+
         XCTAssertTrue(analytics.hasUnsentEvents)
 
         @Atomic var flushSent = false

Tests/Segment-Tests/JSON_Tests.swift

@@ -48,7 +48,7 @@ class JSONTests: XCTestCase {
             let json = try encoder.encode(userInfo)
             XCTAssertNotNil(json)
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
     }
@@ -72,7 +72,7 @@ class JSONTests: XCTestCase {
             let newTest = try! JSONDecoder.default.decode(TestStruct.self, from: json)
             XCTAssertEqual(newTest.myDate.toString(), "\"\(expectedDateString)\"")
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
 
@@ -132,7 +132,7 @@ class JSONTests: XCTestCase {
             let json = try encoder.encode(object)
             XCTAssertNotNil(json)
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
     }
@@ -262,7 +262,7 @@ class JSONTests: XCTestCase {
                     newValue = 11
                 }
             }
-            print("value = \(value.self)")
+            //print("value = \(value.self)")
             return newValue
         }).dictionaryValue
 
@@ -353,7 +353,7 @@ class JSONTests: XCTestCase {
             let o = try JSON(nan)
             XCTAssertNotNil(o)
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
 
@@ -371,7 +371,7 @@ class JSONTests: XCTestCase {
             XCTAssertNotNil(t)
             XCTAssertTrue(t!.nando == 0)
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
     }
@@ -390,7 +390,7 @@ class JSONTests: XCTestCase {
             let o = try JSON(nan)
             XCTAssertNotNil(o)
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
 
@@ -408,7 +408,7 @@ class JSONTests: XCTestCase {
             XCTAssertNotNil(t)
             XCTAssertNil(t!.nando)
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
     }
@@ -445,7 +445,7 @@ class JSONTests: XCTestCase {
 
         do {
             let json = try JSON(dict)
-            print(json.prettyPrint())
+            //print(json.prettyPrint())
 
             let strEnum: String? = json[keyPath: "strEnum"]
             XCTAssertEqual(strEnum, "test2")
@@ -463,7 +463,7 @@ class JSONTests: XCTestCase {
             XCTAssertEqual(uuid!.count, 36)
 
         } catch {
-            print(error)
+            //print(error)
             XCTFail()
         }
     }

Tests/Segment-Tests/ObjC_Tests.swift

@@ -103,14 +103,14 @@ class ObjC_Tests: XCTestCase {
         analytics.analytics.add(plugin: outputReader)
 
         let sourcePlugin = ObjCBlockPlugin { event in
-            print("source enrichment applied")
+            //print("source enrichment applied")
             sourceHit = true
             return event
         }
         analytics.add(plugin: sourcePlugin)
 
         let destPlugin = ObjCBlockPlugin { event in
-            print("destination enrichment applied")
+            //print("destination enrichment applied")
             destHit = true
             return event
         }