Fix concurrency issues around close()

CHANGELOG.md

@@ -20,6 +20,7 @@
   Backends that process uploads asynchronously can handle checkpoint requests themselves by
   implementing `CustomCheckpointRequestConnector` on their connector.
   These APIs are in alpha and may change in future releases.
+* Fix races around `PowerSyncDatabase.close()` crashing the process.
 
 ## 1.15.1
 

Sources/PowerSync/Implementation/AsyncConnectionPool.swift

@@ -101,7 +101,7 @@ final class AsyncConnectionPool: SQLiteConnectionPoolProtocol {
     }
 
     private func configureConnection(connection: borrowing RawSqliteConnection, isWriter: Bool) throws {
-        let context = connection.asLease()
+        let context = try connection.asLease()
         for stmt in initialStatements {
             let _ = try context.execute(sql: stmt, parameters: [])
         }

Sources/PowerSync/Implementation/PowerSyncDatabaseImpl.swift

@@ -245,7 +245,7 @@ private actor DatabaseInitializationAction {
 
     func ensureInitialized(db: PowerSyncDatabaseImpl) async throws {
         if closed {
-            throw PowerSyncError.operationFailed(message: "Attempted to use closed PowerSync database")
+            throw PowerSyncError.databaseClosedError()
         }
         if isInitialized {
             return
@@ -269,9 +269,16 @@ private actor DatabaseInitializationAction {
     }
 
     func close(action: () async throws -> ()) async rethrows {
-        if !closed {
-            closed = true
+        if closed {
+            return
+        }
+
+        closed = true
+        do {
             try await action()
+        } catch {
+            closed = false
+            throw error
         }
     }
 }

Sources/PowerSync/Implementation/sqlite3/NativeConnectionPool.swift

@@ -56,21 +56,21 @@ final class NativeConnectionPool: Sendable {
         // No dedicated readers? Acquire write connection for this then
         let semaphore = readers ?? writer
         let connection = try await semaphore.acquire(count: 1)
-        let lease = connection.acquiredItems[0].asLease()
+        let lease = try connection.acquiredItems[0].asLease()
         return try await onConnection(lease)
     }
 
     func write<T>(onConnection: (NativeConnectionLease) async throws -> T) async throws -> T {
         let connection = try await writer.acquire(count: 1)
-        let lease = connection.acquiredItems[0].asLease()
+        let lease = try connection.acquiredItems[0].asLease()
         defer { dispatchWrites(lease: lease) }
         let result = try await onConnection(lease)
         return result
     }
 
     func withAllConnections<T>(onConnection: (NativeConnectionLease, [NativeConnectionLease]) async throws -> T) async throws -> T {
         let write = try await writer.acquire(count: 1)
-        let writeLease = write.acquiredItems[0].asLease()
+        let writeLease = try write.acquiredItems[0].asLease()
         defer { dispatchWrites(lease: writeLease) }
 
         let result: T
@@ -80,7 +80,7 @@ final class NativeConnectionPool: Sendable {
 
             let span = acquiredReaders.acquiredItems.span
             for idx in span.indices {
-                readerLeases.append(span[idx].asLease())
+                readerLeases.append(try span[idx].asLease())
             }
             result = try await onConnection(writeLease, readerLeases)
         } else {
@@ -128,8 +128,11 @@ struct RawSqliteConnection: ~Copyable {
         sqlite3_close_v2(connection)
     }
 
-    func asLease() -> NativeConnectionLease {
-        precondition(!closed)
+    func asLease() throws(PowerSyncError) -> NativeConnectionLease {
+        if closed {
+            throw .databaseClosedError()
+        }
+
         return NativeConnectionLease(pointer: self.connection)
     }
 }

Sources/PowerSync/Protocol/PowerSyncError.swift

@@ -49,4 +49,8 @@ public enum PowerSyncError: Error, LocalizedError {
             return msg
         }
     }
+
+    internal static func databaseClosedError() -> Self {
+        .operationFailed(message: "Attempted to use closed PowerSync database")
+    }
 }

Tests/PowerSyncTests/Implementation/ConcurrentOpenTests.swift

@@ -12,7 +12,10 @@ import Testing
 /// retry with backoff. Reproduced here in-process with a system-SQLite connection holding
 /// a reserved lock, which is byte-for-byte the same file-level contention two processes
 /// produce.
-@Suite("Concurrent open")
+/// 
+/// We run these tests without parallelization as they can block threads (while waiting for
+/// a busy lock), which interferes with other concurrent tests.
+@Suite("Concurrent open", .serialized)
 struct ConcurrentOpenTests {
     /// Sendable wrapper for the lock-holding SQLite connection used across tasks.
     private final class LockHolder: @unchecked Sendable {

Tests/PowerSyncTests/Implementation/DatabaseImplementationTests.swift

@@ -42,4 +42,24 @@ struct DatabaseImplementationTests {
         try #require(results == [2, 4, 6])
         try await db.close()
     }
+
+    @Test func canCancelClose() async throws {
+        let db = PowerSyncDatabase(
+            schema: Schema(),
+            dbFilename: "cancel-close-test",
+            logger: DefaultLogger()
+        )
+
+        let result = try await db.readLock { reader in
+            // Try to close the database, this can't work because of the busy read connection.
+            let task = Task { try await db.close() };
+            task.cancel();
+            return task
+        }.result
+        #expect(throws: CancellationError.self) { try result.get() }
+
+        // Verify that the database is not in a half-closed state by updating the schema, which runs statements
+        // on all connections.
+        try await db.updateSchema(schema: Schema(Table(name: "users", columns: [.text("name")])))
+    }
 }

Tests/PowerSyncTests/Implementation/StatementTests.swift

@@ -5,7 +5,7 @@ import Testing
 struct StatementTests {
     @Test func bindValues() throws {
         let connection = try DatabaseLocation.inMemory.openConnection(writer: true)
-        let lease = connection.asLease()
+        let lease = try connection.asLease()
         try lease.withIterator(
             sql: "SELECT ?, ?, ?, ?, ?, ?, typeof(?), typeof(?)",
             parameters: [