Merge remote-tracking branch 'origin/main' into concurrent

Author: stevensJourney

core-tests-android/src/androidTest/java/com/powersync/AndroidDatabaseTest.kt

@@ -34,7 +34,10 @@ class AndroidDatabaseTest {
 
     @After
     fun tearDown() {
-        runBlocking { database.disconnectAndClear(true) }
+        runBlocking {
+            database.disconnectAndClear(true)
+            database.close()
+        }
     }
 
     @Test

core/build.gradle.kts

@@ -217,8 +217,8 @@ kotlin {
             dependsOn(commonTest.get())
         }
 
-        val commonJDBC by creating {
-            kotlin.srcDir("commonJDBC")
+        val commonJava by creating {
+            kotlin.srcDir("commonJava")
             dependsOn(commonMain.get())
             dependencies {
                 implementation(libs.sqlite.jdbc)
@@ -240,14 +240,18 @@ kotlin {
             api(libs.kermit)
         }
 
-        androidMain.dependencies {
-            implementation(libs.ktor.client.okhttp)
-            implementation(libs.sqlite.jdbc)
+        androidMain {
+            dependsOn(commonJava)
+            dependencies.implementation(libs.ktor.client.okhttp)
         }
 
-        jvmMain.dependencies {
-            implementation(libs.ktor.client.okhttp)
-            implementation(libs.sqlite.jdbc)
+        jvmMain {
+            dependsOn(commonJava)
+
+            dependencies {
+                implementation(libs.ktor.client.okhttp)
+                implementation(libs.sqlite.jdbc)
+            }
         }
 
         iosMain.dependencies {

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

@@ -6,7 +6,7 @@ import co.touchlab.kermit.Logger
 import co.touchlab.kermit.Severity
 import co.touchlab.kermit.TestConfig
 import co.touchlab.kermit.TestLogWriter
-import com.powersync.db.PowerSyncDatabaseImpl
+import com.powersync.db.ActiveDatabaseGroup
 import com.powersync.db.schema.Schema
 import com.powersync.testutils.UserRow
 import com.powersync.testutils.waitFor
@@ -228,7 +228,7 @@ class DatabaseTest {
             waitFor {
                 assertNotNull(
                     logWriter.logs.find {
-                        it.message == PowerSyncDatabaseImpl.multipleInstancesMessage
+                        it.message == ActiveDatabaseGroup.multipleInstancesMessage
                     },
                 )
             }

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

@@ -120,7 +120,7 @@ class SyncIntegrationTest {
     fun testPartialSync() =
         runTest {
             val syncStream = syncStream()
-            database.connect(syncStream, 1000L)
+            database.connectInternal(syncStream, 1000L)
 
             val checksums =
                 buildList {
@@ -214,7 +214,7 @@ class SyncIntegrationTest {
     fun testRemembersLastPartialSync() =
         runTest {
             val syncStream = syncStream()
-            database.connect(syncStream, 1000L)
+            database.connectInternal(syncStream, 1000L)
 
             syncLines.send(
                 SyncLine.FullCheckpoint(
@@ -253,7 +253,7 @@ class SyncIntegrationTest {
     fun setsDownloadingState() =
         runTest {
             val syncStream = syncStream()
-            database.connect(syncStream, 1000L)
+            database.connectInternal(syncStream, 1000L)
 
             turbineScope(timeout = 10.0.seconds) {
                 val turbine = database.currentStatus.asFlow().testIn(this)
@@ -291,7 +291,7 @@ class SyncIntegrationTest {
                 val syncStream = syncStream()
                 val turbine = database.currentStatus.asFlow().testIn(this)
 
-                database.connect(syncStream, 1000L)
+                database.connectInternal(syncStream, 1000L)
                 turbine.waitFor { it.connecting }
 
                 database.disconnect()
@@ -308,7 +308,7 @@ class SyncIntegrationTest {
     fun testMultipleSyncsDoNotCreateMultipleStatusEntries() =
         runTest {
             val syncStream = syncStream()
-            database.connect(syncStream, 1000L)
+            database.connectInternal(syncStream, 1000L)
 
             turbineScope(timeout = 10.0.seconds) {
                 val turbine = database.currentStatus.asFlow().testIn(this)

core/src/commonJava/kotlin/com/powersync/db/ActiveInstanceStore.commonJava.kt

@@ -0,0 +1,27 @@
+package com.powersync.db
+
+internal actual fun disposeWhenDeallocated(resource: ActiveDatabaseResource): Any {
+    // We can't do this on Java 8 :(
+    return object {}
+}
+
+// This would require Java 9+
+
+/*
+import java.lang.ref.Cleaner
+
+internal actual fun disposeWhenDeallocated(resource: ActiveDatabaseResource): Any {
+    // Note: It's important that the returned object does not reference the resource directly
+    val wrapper = CleanableWrapper()
+    CleanableWrapper.cleaner.register(wrapper, resource::dispose)
+    return wrapper
+}
+
+private class CleanableWrapper {
+    var cleanable: Cleaner.Cleanable? = null
+
+    companion object {
+        val cleaner: Cleaner = Cleaner.create()
+    }
+}
+*/

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

@@ -1,23 +1,94 @@
 package com.powersync.db
 
-import com.powersync.PowerSyncDatabase
-import com.powersync.utils.ExclusiveMethodProvider
-
-internal class ActiveInstanceStore : ExclusiveMethodProvider() {
-    private val instances = mutableListOf<PowerSyncDatabase>()
-
-    /**
-     * Registers an instance. Returns true if multiple instances with the same identifier are
-     * present.
-     */
-    suspend fun registerAndCheckInstance(db: PowerSyncDatabase) =
-        exclusiveMethod("instances") {
-            instances.add(db)
-            return@exclusiveMethod instances.filter { it.identifier == db.identifier }.size > 1
+import co.touchlab.kermit.Logger
+import co.touchlab.stately.concurrency.AtomicBoolean
+import co.touchlab.stately.concurrency.Synchronizable
+import co.touchlab.stately.concurrency.synchronize
+import kotlinx.coroutines.sync.Mutex
+
+/**
+ * Returns an object that, when deallocated, calls [ActiveDatabaseResource.dispose].
+ */
+internal expect fun disposeWhenDeallocated(resource: ActiveDatabaseResource): Any
+
+/**
+ * An collection of PowerSync databases with the same path / identifier.
+ *
+ * We expect that each group will only ever have one database because we encourage users to write their databases as
+ * singletons. We print a warning when two databases are part of the same group.
+ * Additionally, we want to avoid two databases in the same group having a sync stream open at the same time to avoid
+ * duplicate resources being used. For this reason, each active database group has a coroutine mutex guarding the
+ * sync job.
+ */
+internal class ActiveDatabaseGroup(
+    val identifier: String,
+    private val collection: GroupsCollection,
+) {
+    internal var refCount = 0 // Guarded by companion object
+    internal val syncMutex = Mutex()
+
+    fun removeUsage() {
+        collection.synchronize {
+            if (--refCount == 0) {
+                collection.allGroups.remove(this)
+            }
         }
+    }
+
+    internal open class GroupsCollection : Synchronizable() {
+        internal val allGroups = mutableListOf<ActiveDatabaseGroup>()
+
+        private fun findGroup(
+            warnOnDuplicate: Logger,
+            identifier: String,
+        ): ActiveDatabaseGroup =
+            synchronize {
+                val existing = allGroups.asSequence().firstOrNull { it.identifier == identifier }
+                val resolvedGroup =
+                    if (existing == null) {
+                        val added = ActiveDatabaseGroup(identifier, this)
+                        allGroups.add(added)
+                        added
+                    } else {
+                        existing
+                    }
+
+                if (resolvedGroup.refCount++ != 0) {
+                    warnOnDuplicate.w { multipleInstancesMessage }
+                }
+
+                resolvedGroup
+            }
+
+        internal fun referenceDatabase(
+            warnOnDuplicate: Logger,
+            identifier: String,
+        ): Pair<ActiveDatabaseResource, Any> {
+            val group = findGroup(warnOnDuplicate, identifier)
+            val resource = ActiveDatabaseResource(group)
+
+            return resource to disposeWhenDeallocated(resource)
+        }
+    }
+
+    companion object : GroupsCollection() {
+        internal val multipleInstancesMessage =
+            """
+            Multiple PowerSync instances for the same database have been detected.
+            This can cause unexpected results.
+            Please check your PowerSync client instantiation logic if this is not intentional.
+            """.trimIndent()
+    }
+}
+
+internal class ActiveDatabaseResource(
+    val group: ActiveDatabaseGroup,
+) {
+    val disposed = AtomicBoolean(false)
 
-    suspend fun removeInstance(db: PowerSyncDatabase) =
-        exclusiveMethod("instances") {
-            instances.remove(db)
+    fun dispose() {
+        if (disposed.compareAndSet(false, true)) {
+            group.removeUsage()
         }
+    }
 }