fix: Guard EventParser.lastEventId against a data race

.github/workflows/ci.yml

@@ -56,6 +56,20 @@ jobs:
 
       - uses: ./.github/actions/test-swiftpm
 
+  test-thread-sanitizer:
+    runs-on: ubuntu-latest
+    container: swift:6.1
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Test with ThreadSanitizer
+        # halt_on_error makes a detected data race abort with a non-zero exit so the job fails;
+        # without it TSan only prints a warning and the step would stay green.
+        env:
+          TSAN_OPTIONS: halt_on_error=1
+        run: swift test --sanitize=thread
+
   contract-tests:
     runs-on: macos-15
 

Source/EventParser.swift

@@ -1,6 +1,10 @@
 import Foundation
 
-class EventParser {
+// `parse`, `reset`, and event dispatch must be driven from a single serialized context (the
+// EventSource delegate queue). `getLastEventId()` is the one entry point called concurrently — from
+// arbitrary caller threads via `EventSource.getLastEventId()` — so the `lastEventId` it reads is
+// lock-guarded. `@unchecked Sendable` reflects that split, which the compiler cannot verify.
+final class EventParser: @unchecked Sendable {
     private struct Constants {
         static let dataLabel: Substring = "data"
         static let idLabel: Substring = "id"
@@ -13,12 +17,28 @@ class EventParser {
     private var data: String = ""
     private var eventType: String = ""
     private var lastEventIdBuffer: String?
-    private var lastEventId: String
     private var currentRetry: TimeInterval
 
+    // Written on the serialized parse path and read from any thread via `getLastEventId()`, so its
+    // access is guarded by a lock.
+    private let lastEventIdLock = NSLock()
+    private var _lastEventId: String
+    private var lastEventId: String {
+        get {
+            lastEventIdLock.lock()
+            defer { lastEventIdLock.unlock() }
+            return _lastEventId
+        }
+        set {
+            lastEventIdLock.lock()
+            defer { lastEventIdLock.unlock() }
+            _lastEventId = newValue
+        }
+    }
+
     init(handler: EventHandler, initialEventId: String, initialRetry: TimeInterval) {
         self.handler = handler
-        self.lastEventId = initialEventId
+        self._lastEventId = initialEventId
         self.currentRetry = initialRetry
     }
 

Tests/EventParserTests.swift

@@ -306,4 +306,30 @@ final class EventParserTests {
         #expect(handler.events.maybeEvent() == .comment("bar"))
         #expect(handler.events.maybeEvent() == .message("msg", MessageEvent(data: "foo", lastEventId: "")))
     }
+
+    // MARK: Concurrency
+    // getLastEventId() is read from arbitrary threads while events are parsed on another; the read
+    // must be race-free. Uses a local parser/handler (not the suite fixtures) so the concurrent
+    // writes don't trip the deinit "no leftover events" check. Run under `--sanitize=thread` to
+    // actually catch a regression here.
+    @Test func getLastEventIdIsSafeDuringConcurrentParsing() async {
+        let parser = EventParser(handler: MockHandler(), initialEventId: "", initialRetry: 1.0)
+        await withTaskGroup(of: Void.self) { group in
+            // Writer: drive events on a single task, mirroring the serialized delegate queue.
+            group.addTask {
+                for id in 0..<2000 {
+                    parser.parse(line: "id: \(id)")
+                    parser.parse(line: "data: x")
+                    parser.parse(line: "")
+                }
+            }
+            // Reader: hammer getLastEventId() from a separate task, concurrent with the writer.
+            group.addTask {
+                for _ in 0..<2000 {
+                    _ = parser.getLastEventId()
+                }
+            }
+        }
+        #expect(parser.getLastEventId() == "1999")
+    }
 }