Some additional cleanup

Author: simolus3

Sources/PowerSync/Implementation/AsyncConnectionPool.swift

@@ -61,26 +61,27 @@ final class AsyncConnectionPool: SQLiteConnectionPoolProtocol {
     private let initialStatements: [String]
     private let logger: any LoggerProtocol
     private let tableUpdatesStream = BroadcastStream<Set<String>>()
-    private let inner: AsyncSemaphore<NativeConnectionPool?> = AsyncSemaphore(singleElement: nil)
+    private let opener = PoolOpener()
 
     init(location: DatabaseLocation, logger: any LoggerProtocol, initialStatements: [String] = []) {
         self.location = location
         self.logger = logger
         self.initialStatements = initialStatements
     }
-    
+
     var tableUpdates: AsyncStream<Set<String>> {
         tableUpdatesStream.subscribe()
     }
-    
+
+    /// Asyncifies a synchronous unit of work on by running it on a suitable background thread.
     private func runBlocking<T>(action: @escaping @Sendable () throws -> T, qos: DispatchQoS.QoSClass = .userInitiated) async throws -> T {
         return try await withCheckedThrowingContinuation { continuation in
             DispatchQueue.global(qos: qos).async {
                 continuation.resume(with: Result(catching: { try action() }))
             }
         }
     }
-    
+
     private func configureConnection(connection: borrowing RawSqliteConnection, isWriter: Bool) throws {
         let context = connection.asLease()
         for stmt in initialStatements {
@@ -111,40 +112,10 @@ final class AsyncConnectionPool: SQLiteConnectionPoolProtocol {
             let _ = try context.execute(sql: "select powersync_update_hooks('install')", parameters: [])
         }
     }
-    
-    private func obtainInner() async throws -> NativeConnectionPool {
-        var lease = try await inner.acquire(count: 1)
-        if let pool = lease.acquiredItems[0] {
-            return pool
-        } else {
-            try registerPowerSyncCoreExtension()
-
-            @Sendable func handleUpdates(_ updates: Set<String>) {
-                self.tableUpdatesStream.dispatch(event: updates)
-            }
-            
-            let pool = try await runBlocking { [self] in
-                let writer = try location.openConnection(writer: true)
-                try configureConnection(connection: writer, isWriter: true)
 
-                if case .inMemory = location {
-                    return NativeConnectionPool(singleConnection: writer, logger: logger, handleUpdates: handleUpdates)
-                } else {
-                    let numReaders = 4
-                    var readers = RigidDeque<RawSqliteConnection>(capacity: numReaders)
-                    while !readers.isFull {
-                        let connection = try location.openConnection(writer: false)
-                        try configureConnection(connection: connection, isWriter: false)
-                        readers.append(connection)
-                    }
-
-                    return NativeConnectionPool(writer: writer, readers: readers, logger: logger, handleUpdates: handleUpdates)
-                }
-            }
-            
-            lease.acquiredItems[0] = pool
-            return pool
-        }
+    /// Opens connections on a background thread to obtain the native connection pool.
+    private func obtainInner() async throws -> NativeConnectionPool {
+        try await opener.obtainPool(pool: self)
     }
 
     func read<T>(onConnection: @escaping @Sendable (any SQLiteConnectionLease) throws -> T) async throws -> T {
@@ -160,7 +131,7 @@ final class AsyncConnectionPool: SQLiteConnectionPoolProtocol {
             try await runBlocking { try onConnection(connection) }
         }
     }
-    
+
     func withAllConnections<T>(onConnection: @escaping @Sendable (any SQLiteConnectionLease, [any SQLiteConnectionLease]) throws -> T) async throws -> T {
         let pool = try await obtainInner()
         return try await pool.withAllConnections { writer, readers in
@@ -169,10 +140,55 @@ final class AsyncConnectionPool: SQLiteConnectionPoolProtocol {
     }
 
     func close() async throws {
-        var lease = try await inner.acquire(count: 1)
-        if let pool = lease.acquiredItems[0] {
-            try await pool.close()
-            lease.acquiredItems[0] = nil
+        try await self.opener.close()
+    }
+
+    private actor PoolOpener {
+        private var pool: NativeConnectionPool? = nil
+        private var isClosed = false
+
+        func obtainPool(pool context: AsyncConnectionPool) async throws -> NativeConnectionPool {
+            if let pool {
+                return pool
+            }
+
+            try registerPowerSyncCoreExtension()
+            let handleUpdates: @Sendable (_: Set<String>) -> () = { [weak context] updates in
+                context?.tableUpdatesStream.dispatch(event: updates)
+            }
+
+            let pool = try await context.runBlocking {
+                let writer = try context.location.openConnection(writer: true)
+                try context.configureConnection(connection: writer, isWriter: true)
+
+                if case .inMemory = context.location {
+                    return NativeConnectionPool(singleConnection: writer, logger: context.logger, handleUpdates: handleUpdates)
+                } else {
+                    let numReaders = 4
+                    var readers = RigidDeque<RawSqliteConnection>(capacity: numReaders)
+                    while !readers.isFull {
+                        let connection = try context.location.openConnection(writer: false)
+                        try context.configureConnection(connection: connection, isWriter: false)
+                        readers.append(connection)
+                    }
+
+                    return NativeConnectionPool(writer: writer, readers: readers, logger: context.logger, handleUpdates: handleUpdates)
+                }
+            }
+
+            self.pool = pool
+            return pool
+        }
+
+        func close() async throws {
+            if isClosed {
+                return
+            }
+
+            isClosed = true
+            if let pool {
+                try await pool.close()
+            }
         }
     }
 }

Sources/PowerSync/Implementation/sqlite3/NativeConnectionPool.swift

@@ -6,6 +6,7 @@ import DequeModule
 ///
 /// This class does not configure or open connections (that is the responsibility of ``AsyncConnectionPool``).
 final class NativeConnectionPool: Sendable {
+    // This could be an async mutex, but AsyncSemaphore has better cancellation support.
     private let writer: AsyncSemaphore<RawSqliteConnection>
     private let readers: AsyncSemaphore<RawSqliteConnection>?
     private let handleUpdates: @Sendable (_: Set<String>) -> ()

Sources/PowerSync/Implementation/sqlite3/NativeStatement.swift

@@ -35,13 +35,12 @@ struct NativeSqliteStatement: ~Copyable {
     var columnCount: Int {
         return Int(sqlite3_column_count(self.stmt))
     }
-    
-    
+
     var columnNames: [String : Int] {
         return resolvedColumnNames!
     }
 
-    borrowing func bindValues(_ parameters: [PowerSyncDataType?]) throws (PowerSyncError) {
+    borrowing func bindValues(_ parameters: [PowerSyncDataType?]) throws(PowerSyncError) {
         for (i, parameter) in parameters.enumerated() {
             let index = Int32(i + 1)
 
@@ -53,7 +52,7 @@ struct NativeSqliteStatement: ~Copyable {
         }
     }
 
-    borrowing func bindValue(_ index: Int32, _ parameter: PowerSyncDataType?) throws (PowerSyncError) {
+    borrowing func bindValue(_ index: Int32, _ parameter: PowerSyncDataType?) throws(PowerSyncError) {
         let rc: Int32
 
         switch parameter {
@@ -98,7 +97,7 @@ struct NativeSqliteStatement: ~Copyable {
         }
     }
 
-    mutating func step() throws (PowerSyncError) -> Bool {
+    mutating func step() throws(PowerSyncError) -> Bool {
         let rc = sqlite3_step(self.stmt)
         if rc == SQLITE_DONE {
             return false

Sources/PowerSync/Utils/MergeItemSequence.swift

@@ -1,4 +1,9 @@
-/// An ``AsyncSequence`` merging all items emitted between ``AsyncIteratorProtocol/next``.
+/// An ``AsyncSequence`` merging all items emitted between calls to ``AsyncIteratorProtocol/next``.
+/// 
+/// This is useful for sequences where we just want to know that an event has occurred, without needing
+/// to know about the exact event. We use this internally to implement `watch()` queries with a throttle:
+/// If any amount of events have occurred between throttled calls to `next()`, we want to dispatch a single
+/// event.
 struct MergeItemSequence<Base: AsyncSequence & Sendable>: AsyncSequence where Base.Element == () {
     typealias AsyncIterator = IteratorImpl
     typealias Element = ()

Sources/PowerSync/resolvePowerSyncLoadableExtensionPath.swift

@@ -5,7 +5,7 @@
 /// function invokes `sqlite3_auto_extension` to load the core extension automatically.
 ///
 /// - Returns: `nil`
-/// - Throws: An error if the extension could not be registered watchOS.
+/// - Throws: An error if the extension could not be registered.
 public func resolvePowerSyncLoadableExtensionPath() throws(PowerSyncError) -> String? {
     try registerPowerSyncCoreExtension()
     return nil