Fix crash when multiple updates create publishers simultaneously

Sources/Persist/Persister.swift

@@ -131,6 +131,10 @@ public final class Persister<Value> {
     /// The updates subject that publishes updates.
     @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
     private var updatesSubject: PassthroughSubject<UpdatePayload, Never> {
+        updatesSubjectLock.lock()
+        defer {
+            updatesSubjectLock.unlock()
+        }
         if let updatesSubject = _updatesSubject as? PassthroughSubject<UpdatePayload, Never> {
             return updatesSubject
         }
@@ -140,13 +144,28 @@ public final class Persister<Value> {
         return updatesSubject
     }
 
+    @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+    private var updatesSubjectIfCreated: PassthroughSubject<UpdatePayload, Never>? {
+        updatesSubjectLock.lock()
+        defer {
+            updatesSubjectLock.unlock()
+        }
+        return _updatesSubject.flatMap { $0 as? PassthroughSubject<UpdatePayload, Never> }
+    }
+
+    private let updatesSubjectLock = NSLock()
+
     /// An `Any` value that will always be a `PassthroughSubject<UpdatePayload, Never>`.
     /// This is required because Swift does not support marking stored properties as `available`.
     private var _updatesSubject: Any?
 
     /// The updates subject that publishes updates.
     @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
     private var subject: CurrentValueSubject<Value, Never> {
+        subjectLock.lock()
+        defer {
+            subjectLock.unlock()
+        }
         if let subject = _subject as? CurrentValueSubject<Value, Never> {
             return subject
         }
@@ -156,6 +175,17 @@ public final class Persister<Value> {
         return subject
     }
 
+    @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+    private var subjectIfCreated: CurrentValueSubject<Value, Never>? {
+        subjectLock.lock()
+        defer {
+            subjectLock.unlock()
+        }
+        return _subject.flatMap { $0 as? CurrentValueSubject<Value, Never> }
+    }
+
+    private let subjectLock = NSLock()
+
     /// An `Any` value that will always be a `CurrentValueSubject<Value, Never>`.
     /// This is required because Swift does not support marking stored properties as `available`.
     private var _subject: Any?
@@ -857,13 +887,23 @@ public final class Persister<Value> {
 
         #if canImport(Combine)
         if #available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *) {
-            updatesSubject.send(result)
-
-            switch result {
-            case .success(let update):
-                subject.send(update.newValue)
-            case .failure:
-                break
+            // Send the update to the Combine subjects. If they're not yet
+            // created this will do nothing. Subjects are created when accessed
+            // via the public API.
+            //
+            // We do not need to worry about thread safety when calling `send`
+            // here because the computed properties used are protected with a
+            // lock and the `send(_:)` function is thread-safe:
+            // https://forums.swift.org/t/thread-safety-for-combine-publishers/29491/13
+            updatesSubjectIfCreated?.send(result)
+
+            if let subject = subjectIfCreated {
+                switch result {
+                case .success(let update):
+                    subject.send(update.newValue)
+                case .failure:
+                    break
+                }
             }
         }
         #endif

Tests/PersistTests/PersisterTests.swift

@@ -649,6 +649,65 @@ final class PersisterTests: XCTestCase {
 
         waitForExpectations(timeout: 1, handler: nil)
     }
+
+    /// Prior to 1.3.0 subjects (which are exposed as publishers) were created
+    /// lazily when the first update occurred _or_ one of the publisher
+    /// properties was accessed.
+    ///
+    /// This could crash if multiple updates were sent across multiple queues
+    /// and the properties had not yet been created.
+    ///
+    /// This would crash quite consistently when it's the only test being run.
+    @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+    func testPersistingValueAcrossQueuesWithoutPublishers() throws {
+        let storage = InMemoryStorage<Int>()
+        let persister = Persister(key: "test", storedBy: storage, defaultValue: nil)
+
+        let updatesCount = 10_000
+
+        let persistsValueExpectation = expectation(description: "Persists value")
+        persistsValueExpectation.expectedFulfillmentCount = updatesCount
+        DispatchQueue.concurrentPerform(iterations: updatesCount) { index in
+            try? persister.persist(index)
+            persistsValueExpectation.fulfill()
+        }
+
+        waitForExpectations(timeout: 1, handler: nil)
+    }
+
+    /// This is the same as ``testPersistingValueAcrossQueuesWithoutPublishers``
+    /// but with both publishers created prior to sending an update. This is
+    /// more of a test that the publishers support access across threads, which
+    /// is explicitly supported according to
+    /// https://forums.swift.org/t/thread-safety-for-combine-publishers/29491/13
+    @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+    func testPersistingValueAcrossQueuesWithPublishers() throws {
+        let storage = InMemoryStorage<Int>()
+        let persister = Persister(key: "test", storedBy: storage, defaultValue: nil)
+
+        let updatesCount = 10_000
+
+        let notifiesPublisherExpectation = expectation(description: "Calls update listener")
+        notifiesPublisherExpectation.expectedFulfillmentCount = updatesCount
+        let publisherSubscription = persister.publisher.dropFirst().sink { _ in
+            notifiesPublisherExpectation.fulfill()
+        }
+
+        let notifiesUpdatesPublisherExpectation = expectation(description: "Calls update listener")
+        notifiesUpdatesPublisherExpectation.expectedFulfillmentCount = updatesCount
+        let updatesPublisherSubscription = persister.updatesPublisher.sink { _ in
+            notifiesUpdatesPublisherExpectation.fulfill()
+        }
+
+        DispatchQueue.concurrentPerform(iterations: updatesCount) { index in
+            try? persister.persist(index)
+        }
+
+        waitForExpectations(timeout: 3, handler: nil)
+
+        _ = publisherSubscription
+        _ = updatesPublisherSubscription
+    }
     #endif
 
 }