Fix compilation errors

Author: simolus3

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
@@ -122,7 +122,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

@@ -1,5 +1,12 @@
 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 {
@@ -16,3 +23,4 @@ private class CleanableWrapper {
         val cleaner: Cleaner = Cleaner.create()
     }
 }
+*/

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

@@ -71,7 +71,7 @@ internal class ActiveDatabaseGroup(val identifier: String) {
 }
 
 internal class ActiveDatabaseResource(val group: ActiveDatabaseGroup) {
-    var disposed = atomic(false)
+    val disposed = atomic(false)
 
     fun dispose() {
         if (!disposed.getAndSet(true)) {

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

@@ -34,6 +34,8 @@ import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import kotlinx.datetime.Instant
 import kotlinx.datetime.LocalDateTime
 import kotlinx.datetime.TimeZone
@@ -82,10 +84,9 @@ internal class PowerSyncDatabaseImpl(
      */
     override val currentStatus: SyncStatus = SyncStatus()
 
+    private val mutex = Mutex()
     private var syncStream: SyncStream? = null
-
     private var syncJob: Job? = null
-
     private var uploadJob: Job? = null
 
     init {
@@ -117,10 +118,10 @@ internal class PowerSyncDatabaseImpl(
         crudThrottleMs: Long,
         retryDelayMs: Long,
         params: Map<String, JsonParam?>,
-    ) = exclusiveMethod("connect") {
-        disconnect()
+    ) = mutex.withLock {
+        disconnectInternal()
 
-        connect(
+        connectInternal(
             SyncStream(
                 bucketStorage = bucketStorage,
                 connector = connector,
@@ -134,7 +135,7 @@ internal class PowerSyncDatabaseImpl(
     }
 
     @OptIn(FlowPreview::class)
-    internal fun connect(
+    internal fun connectInternal(
         stream: SyncStream,
         crudThrottleMs: Long,
     ) {
@@ -327,7 +328,9 @@ internal class PowerSyncDatabaseImpl(
         }
     }
 
-    override suspend fun disconnect() {
+    override suspend fun disconnect() = mutex.withLock { disconnectInternal() }
+
+    private suspend fun disconnectInternal() {
         if (syncJob != null && syncJob!!.isActive) {
             syncJob?.cancelAndJoin()
         }
@@ -421,11 +424,11 @@ internal class PowerSyncDatabaseImpl(
     }
 
     override suspend fun close() =
-        exclusiveMethod("close") {
+        mutex.withLock {
             if (closed) {
-                return@exclusiveMethod
+                return@withLock
             }
-            disconnect()
+            disconnectInternal()
             internalDb.close()
             resource.dispose()
             closed = true