add tests for connection pool close

Author: stevensJourney

core/build.gradle.kts

@@ -271,9 +271,6 @@ kotlin {
         // tests.
         jvmTest.get().dependsOn(commonIntegrationTest)
 
-        androidMain.get().dependsOn(commonJDBC)
-        jvmMain.get().dependsOn(commonJDBC)
-
         // We're linking the xcframework for the simulator tests, so they can use integration tests too
         iosSimulatorArm64Test.orNull?.dependsOn(commonIntegrationTest)
     }

core/src/androidMain/kotlin/com/powersync/DatabaseDriverFactory.android.kt

@@ -1,7 +1,10 @@
 package com.powersync
 
 import android.content.Context
+import com.powersync.db.JdbcSqliteDriver
+import com.powersync.db.buildDefaultWalProperties
 import com.powersync.db.internal.InternalSchema
+import com.powersync.db.migrateDriver
 import kotlinx.coroutines.CoroutineScope
 import org.sqlite.SQLiteCommitListener
 

core/src/commonIntegrationTest/kotlin/com/powersync/DatabaseTest.kt

@@ -11,9 +11,12 @@ import com.powersync.db.schema.Schema
 import com.powersync.testutils.UserRow
 import com.powersync.testutils.waitFor
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
 import kotlin.test.AfterTest
 import kotlin.test.BeforeTest
 import kotlin.test.Test
@@ -58,7 +61,11 @@ class DatabaseTest {
 
     @AfterTest
     fun tearDown() {
-        runBlocking { database.disconnectAndClear(true) }
+        runBlocking {
+            if (!database.closed) {
+                database.disconnectAndClear(true)
+            }
+        }
         com.powersync.testutils.cleanup("testdb")
     }
 
@@ -220,6 +227,47 @@ class DatabaseTest {
             }
         }
 
+    @Test
+    fun testClosingReadPool() =
+        runTest {
+            val pausedLock = CompletableDeferred<Unit>()
+            val inLock = CompletableDeferred<Unit>()
+            // Request a lock
+            val lockJob =
+                async {
+                    database.readLock {
+                        inLock.complete(Unit)
+                        runBlocking {
+                            pausedLock.await()
+                        }
+                    }
+                }
+
+            // Wait for the lock to be active
+            inLock.await()
+
+            // Close the database. This should close the read pool
+            // The pool should wait for jobs to complete before closing
+            val closeJob =
+                async {
+                    database.close()
+                }
+
+            // Wait a little for testing
+            // Spawns in a different context for the delay to actually take affect
+            async { withContext(Dispatchers.Default) { delay(500) } }.await()
+
+            // The database should not close yet
+            assertEquals(actual = database.closed, expected = false)
+
+            // Release the lock
+            pausedLock.complete(Unit)
+            lockJob.await()
+            closeJob.await()
+
+            assertEquals(actual = database.closed, expected = true)
+        }
+
     @Test
     fun warnsMultipleInstances() =
         runTest {

core/src/commonJDBC/kotlin/com/powersync/JdbcPreparedStatement.kt renamed to core/src/commonJava/kotlin/com/powersync/db/JdbcPreparedStatement.kt

@@ -1,4 +1,4 @@
-package com.powersync
+package com.powersync.db
 
 import app.cash.sqldelight.db.QueryResult
 import app.cash.sqldelight.db.SqlCursor

core/src/commonJDBC/kotlin/com/powersync/JdbcSqliteDriver.kt renamed to core/src/commonJava/kotlin/com/powersync/db/JdbcSqliteDriver.kt

@@ -1,4 +1,4 @@
-package com.powersync
+package com.powersync.db
 
 import app.cash.sqldelight.Query
 import app.cash.sqldelight.Transacter

core/src/commonJDBC/kotlin/com/powersync/WalProperties.kt renamed to core/src/commonJava/kotlin/com/powersync/db/WalProperties.kt

@@ -1,4 +1,4 @@
-package com.powersync
+package com.powersync.db
 
 import java.util.Properties
 

core/src/commonMain/kotlin/com/powersync/PowerSyncDatabase.kt

@@ -19,6 +19,12 @@ import kotlin.coroutines.cancellation.CancellationException
  * All changes to local tables are automatically recorded, whether connected or not. Once connected, the changes are uploaded.
  */
 public interface PowerSyncDatabase : Queries {
+    /**
+     * Indicates if the PowerSync client has been closed.
+     * A new client is required after a client has been closed.
+     */
+    public val closed: Boolean
+
     /**
      * Identifies the database client.
      * This is typically the database name.

core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -70,19 +70,22 @@ internal class PowerSyncDatabaseImpl(
 
     override val identifier = dbDirectory + dbFilename
 
+    private val activeDatabaseGroup = ActiveDatabaseGroup.referenceDatabase(logger, identifier)
+    private val resource = activeDatabaseGroup.first
+    private val clearResourceWhenDisposed = activeDatabaseGroup.second
+
     private val internalDb =
         InternalDatabaseImpl(
             factory = factory,
             scope = scope,
             dbFilename = dbFilename,
             dbDirectory = dbDirectory,
+            writeLockMutex = resource.group.writeLockMutex,
         )
 
     internal val bucketStorage: BucketStorage = BucketStorageImpl(internalDb, logger)
-    private val resource: ActiveDatabaseResource
-    private val clearResourceWhenDisposed: Any
 
-    var closed = false
+    override var closed = false
 
     /**
      * The current sync status.
@@ -98,10 +101,6 @@ internal class PowerSyncDatabaseImpl(
     private var powerSyncVersion: String? = null
 
     init {
-        val res = ActiveDatabaseGroup.referenceDatabase(logger, identifier)
-        resource = res.first
-        clearResourceWhenDisposed = res.second
-
         runBlocking {
             val sqliteVersion = internalDb.get("SELECT sqlite_version()") { it.getString(0)!! }
             logger.d { "SQLiteVersion: $sqliteVersion" }
@@ -323,23 +322,14 @@ internal class PowerSyncDatabaseImpl(
 
     override suspend fun <R> readTransaction(callback: ThrowableTransactionCallback<R>): R = internalDb.writeTransaction(callback)
 
-    override suspend fun <R> writeLock(callback: ThrowableLockCallback<R>): R =
-        resource.group.writeLockMutex.withLock {
-            internalDb.writeLock(callback)
-        }
+    override suspend fun <R> writeLock(callback: ThrowableLockCallback<R>): R = internalDb.writeLock(callback)
 
-    override suspend fun <R> writeTransaction(callback: ThrowableTransactionCallback<R>): R =
-        resource.group.writeLockMutex.withLock {
-            internalDb.writeTransaction(callback)
-        }
+    override suspend fun <R> writeTransaction(callback: ThrowableTransactionCallback<R>): R = internalDb.writeTransaction(callback)
 
     override suspend fun execute(
         sql: String,
         parameters: List<Any?>?,
-    ): Long =
-        resource.group.writeLockMutex.withLock {
-            internalDb.execute(sql, parameters)
-        }
+    ): Long = internalDb.execute(sql, parameters)
 
     private suspend fun handleWriteCheckpoint(
         lastTransactionId: Int,

core/src/commonMain/kotlin/com/powersync/db/internal/ConnectionPool.kt

@@ -18,24 +18,22 @@ internal class ConnectionPool(
 ) {
     private val available = Channel<Pair<TransactorDriver, CompletableDeferred<Unit>>>()
     private val connections: List<Job> =
-        buildList {
-            repeat(size) {
-                scope.launch {
-                    val driver = TransactorDriver(factory())
-                    try {
-                        while (true) {
-                            val done = CompletableDeferred<Unit>()
-                            try {
-                                available.send(driver to done)
-                            } catch (_: ClosedSendChannelException) {
-                                break // Pool closed
-                            }
-
-                            done.await()
+        List(size) {
+            scope.launch {
+                val driver = TransactorDriver(factory())
+                try {
+                    while (true) {
+                        val done = CompletableDeferred<Unit>()
+                        try {
+                            available.send(driver to done)
+                        } catch (_: ClosedSendChannelException) {
+                            break // Pool closed
                         }
-                    } finally {
-                        driver.driver.close()
+
+                        done.await()
                     }
+                } finally {
+                    driver.driver.close()
                 }
             }
         }

core/src/commonMain/kotlin/com/powersync/db/internal/InternalDatabase.kt

@@ -1,11 +1,10 @@
 package com.powersync.db.internal
 
-import app.cash.sqldelight.db.Closeable
 import com.powersync.db.Queries
 import kotlinx.coroutines.flow.SharedFlow
 
-internal interface InternalDatabase :
-    Queries,
-    Closeable {
+internal interface InternalDatabase : Queries {
     fun updatesOnTables(): SharedFlow<Set<String>>
+
+    suspend fun close(): Unit
 }