Fix more bugs (#46)

- Fix race condition when subscribing to a database change publisher
- Unregister file presenter on deinit
- Return true from SQLiteError.isDatabaseClosed if the database is suspended

Author: atdrendel

Sources/SQLite/CrossProcessChangeNotifier.swift

@@ -8,7 +8,6 @@ final class CrossProcessChangeNotifier: NSObject, @unchecked Sendable {
     private let changeTrackerURL: URL?
     private let isStarted = Locked<Bool>(false)
 
-    private let localChange = PassthroughSubject<Void, Never>()
     private var localChangeSubscription: AnyCancellable?
 
     private let remoteChange = PassthroughSubject<Void, Never>()
@@ -25,6 +24,7 @@ final class CrossProcessChangeNotifier: NSObject, @unchecked Sendable {
 
     init(
         databasePath: String,
+        databaseChangePublisher: AnyPublisher<Void, Error>,
         onRemoteChange: @Sendable @escaping () -> Void
     ) {
         changeTrackerURL = Self.changeTrackerURL(
@@ -33,30 +33,23 @@ final class CrossProcessChangeNotifier: NSObject, @unchecked Sendable {
 
         super.init()
 
-        localChangeSubscription = localChange.throttle(
-            for: .seconds(1),
-            scheduler: RunLoop.main,
-            latest: true
-        )
-        .receive(on: queue)
-        .sink { [weak self] _ in
-            guard let self, let url = changeTrackerURL else { return }
-            let coordinator = NSFileCoordinator(filePresenter: self)
-            coordinator.coordinate(
-                writingItemAt: url,
-                options: .forReplacing,
-                error: nil,
-                byAccessor: touch
+        localChangeSubscription = databaseChangePublisher
+            .replaceError(with: ())
+            .receive(on: queue)
+            .sink { [weak self] in self?.notifyOtherProcesses() }
+
+        remoteChangeSubscription = remoteChange
+            .throttle(
+                for: .seconds(1),
+                scheduler: RunLoop.main,
+                latest: true
             )
-        }
+            .receive(on: queue)
+            .sink { onRemoteChange() }
+    }
 
-        remoteChangeSubscription = remoteChange.throttle(
-            for: .seconds(1),
-            scheduler: RunLoop.main,
-            latest: true
-        )
-        .receive(on: queue)
-        .sink { onRemoteChange() }
+    deinit {
+        removeFilePresenter()
     }
 
     func start() {
@@ -99,7 +92,14 @@ extension CrossProcessChangeNotifier: NSFilePresenter {
     }
 
     private func notifyOtherProcesses() {
-        localChange.send()
+        guard let url = changeTrackerURL else { return }
+        let coordinator = NSFileCoordinator(filePresenter: self)
+        coordinator.coordinate(
+            writingItemAt: url,
+            options: .forReplacing,
+            error: nil,
+            byAccessor: touch
+        )
     }
 
     private var touch: (URL?) -> Void {

Sources/SQLite/SQLiteDatabase.swift

@@ -73,6 +73,7 @@ public final class SQLiteDatabase: DatabaseProtocol, @unchecked Sendable {
         self.sqliteVersion = sqliteVersion.description
         changeNotifier = CrossProcessChangeNotifier(
             databasePath: path,
+            databaseChangePublisher: database.databaseChangePublisher(),
             onRemoteChange: { [weak self] in
                 self?.triggerObservers.send()
             }
@@ -85,6 +86,10 @@ public final class SQLiteDatabase: DatabaseProtocol, @unchecked Sendable {
         changeNotifier.start()
     }
 
+    deinit {
+        changeNotifier.stop()
+    }
+
     func resume() {
         NotificationCenter.default.post(
             name: Self.resumeNotification,
@@ -808,6 +813,14 @@ private enum Database {
         .mapToSQLiteError(sql: sql)
         .eraseToAnyPublisher()
     }
+
+    func databaseChangePublisher() -> AnyPublisher<Void, Error> {
+        let observation = DatabaseRegionObservation(tracking: .fullDatabase)
+        return observation
+            .publisher(in: writer)
+            .map { _ in }
+            .eraseToAnyPublisher()
+    }
 }
 
 private final class SQLitePublisher: Publisher, @unchecked Sendable {
@@ -817,8 +830,6 @@ private final class SQLitePublisher: Publisher, @unchecked Sendable {
     private let sql: SQL
     private let arguments: SQLiteArguments
     private let request: SQLRequest<Row>
-    private let trigger: AnyPublisher<Void, Never>
-    private let queue: DispatchQueue
 
     private let subject = CurrentValueSubject<Output, Failure>([])
     private var subscriptions = Locked<Set<AnyCancellable>>([])
@@ -836,34 +847,31 @@ private final class SQLitePublisher: Publisher, @unchecked Sendable {
             sql: sql,
             arguments: arguments.statementArguments
         )
-        self.trigger = trigger
-        self.queue = queue
 
-        var didFetchInitialValue = false
+        let demands = Locked(Demands { [database, subject] in
+            do {
+                subject.send(try database.read(
+                    sql, arguments: arguments
+                ))
+            } catch {
+                subject.send(completion: .failure(error))
+            }
+        })
+
         let observationSub = DatabaseRegionObservation(tracking: [request])
             .publisher(in: database.writer)
+            .handleEvents(receiveRequest: { demand in
+                demands.access { demands in
+                    demands.receiveObservationDemand(demand)
+                }
+            })
             .receive(on: queue)
             .tryMap { _ in try database.read(sql, arguments: arguments) }
-            .handleEvents(
-                receiveRequest: { [subject] demand in
-                    // NOTE: To match the previous version's behavior, we
-                    //       fetch the initial value as soon as a downstream
-                    //       publisher requests some values.
-                    guard !didFetchInitialValue,
-                          demand > .none
-                    else { return }
-
-                    didFetchInitialValue = true
-
-                    do {
-                        subject.send(try database.read(
-                            sql, arguments: arguments
-                        ))
-                    } catch {
-                        subject.send(completion: .failure(error))
-                    }
+            .handleEvents(receiveRequest: { demand in
+                demands.access { demands in
+                    demands.receiveObservationDownstreamDemand(demand)
                 }
-            )
+            })
             .sink(
                 receiveCompletion: { [subject] completion in
                     subject.send(completion: completion)
@@ -874,6 +882,11 @@ private final class SQLitePublisher: Publisher, @unchecked Sendable {
             )
 
         let triggerSub = trigger
+            .handleEvents(receiveRequest: { demand in
+                demands.access { demands in
+                    demands.receiveTriggerDemand(demand)
+                }
+            })
             .receive(on: queue)
             .tryMap { _ in try database.read(sql, arguments: arguments) }
             .sink(
@@ -963,3 +976,62 @@ private extension GRDB.Database {
         }
     }
 }
+
+private enum Demands {
+    struct Source: OptionSet {
+        let rawValue: Int
+
+        static let observation = Source(rawValue: 1 << 0)
+        static let observationDownstream = Source(rawValue: 1 << 1)
+        static let trigger = Source(rawValue: 1 << 2)
+
+        var isComplete: Bool {
+            self == [.observation, .observationDownstream, .trigger]
+        }
+    }
+
+    case finished
+    case waiting(Source, () -> Void)
+
+    init(_ block: @escaping () -> Void) {
+        self = .waiting(Source(), block)
+    }
+
+    mutating func receiveObservationDemand(
+        _ demand: Subscribers.Demand
+    ) {
+        receiveDemand(demand, source: .observation)
+    }
+
+    mutating func receiveObservationDownstreamDemand(
+        _ demand: Subscribers.Demand
+    ) {
+        receiveDemand(demand, source: .observationDownstream)
+    }
+
+    mutating func receiveTriggerDemand(
+        _ demand: Subscribers.Demand
+    ) {
+        receiveDemand(demand, source: .trigger)
+    }
+
+    private mutating func receiveDemand(
+        _ demand: Subscribers.Demand,
+        source: Source
+    ) {
+        guard case .waiting(var sources, let block) = self,
+              demand > .none
+        else {
+            return
+        }
+
+        sources.insert(source)
+
+        if sources.isComplete {
+            self = .finished
+            block()
+        } else {
+            self = .waiting(sources, block)
+        }
+    }
+}

Sources/SQLite/SQLiteError.swift

@@ -137,7 +137,11 @@ extension SQLiteError: CustomStringConvertible {
 
 public extension SQLiteError {
     var isBusy: Bool { self == .SQLITE_BUSY }
-    var isDatabaseClosed: Bool { self == .SQLITE_MISUSE }
+    var isDatabaseClosed: Bool {
+        self == .SQLITE_MISUSE // Database is closed error
+            || self == .SQLITE_INTERRUPT // Database is suspended
+            || self == .SQLITE_ABORT // Database is suspended
+    }
 
     var isInterrupt: Bool {
         switch self {

Tests/SQLiteTests/CrossProcessChangeNotifierTests.swift

@@ -51,7 +51,8 @@ final class CrossProcessChangeNotifierTests: XCTestCase {
                 arguments: ["id": .integer(1), "text": .text("one")]
             )
 
-            wait(for: [ex1, ex2], timeout: 4) // Coordinated writes can be very slow
+            // Coordinated writes can be very slow
+            wait(for: [ex1, ex2], timeout: 4)
         }
     }
 

Tests/SQLiteTests/SQLitePublisherTests.swift

@@ -52,20 +52,45 @@ final class SQLitePublisherTests: XCTestCase {
 
             try db.write(Person.createTable)
 
+            let expectedOutput = Locked([
+                (0, []),
+                (1, [_person1]),
+                (2, [_person1, _person2]),
+            ])
             let ex = db
                 .publisher(Person.self, Person.getAll)
                 .removeDuplicates()
-                .expectOutput(
-                    [[], [_person1], [_person1, _person2]],
-                    failsOnCompletion: true
-                )
-
-            try db.write(Person.insert, arguments: _person1.asArguments)
-
-            db.suspend()
-            db.resume()
-
-            try db.write(Person.insert, arguments: _person2.asArguments)
+                .expectOutput({ [_person1, _person2] people in
+                    let (i, expected) = expectedOutput.access { $0.removeFirst() }
+                    XCTAssertEqual(expected, people)
+
+                    switch i {
+                    case 0:
+                        try db.write(
+                            Person.insert,
+                            arguments: _person1.asArguments
+                        )
+
+                        db.suspend()
+                        db.resume()
+
+                        return .moreExpected
+
+                    case 1:
+                        try db.write(
+                            Person.insert,
+                            arguments: _person2.asArguments
+                        )
+                        return .moreExpected
+
+                    case 2:
+                        return .finished
+
+                    default:
+                        XCTFail()
+                        return .finished
+                    }
+                }, failsOnCompletion: true)
 
             wait(for: [ex], timeout: 2)
         }