Web: Use navigator locks

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - Update Kotlin to 2.4.10.
 - Compile against Android SDK 37.
+- Web: Use navigator locks to guard the sync client.
 
 ## 1.14.1
 

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

@@ -4,6 +4,7 @@ import co.touchlab.kermit.Logger
 import co.touchlab.stately.concurrency.AtomicBoolean
 import co.touchlab.stately.concurrency.Synchronizable
 import co.touchlab.stately.concurrency.synchronize
+import com.powersync.utils.maybeSharedMutex
 import kotlinx.coroutines.sync.Mutex
 
 /**
@@ -12,7 +13,7 @@ import kotlinx.coroutines.sync.Mutex
 internal expect fun disposeWhenDeallocated(resource: ActiveDatabaseResource): Any
 
 /**
- * An collection of PowerSync databases with the same path / identifier.
+ * A 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.
@@ -25,8 +26,8 @@ internal class ActiveDatabaseGroup(
     private val collection: GroupsCollection,
 ) {
     internal var refCount = 0 // Guarded by companion object
-    internal val syncMutex = Mutex()
-    internal val writeLockMutex = Mutex()
+    internal val syncMutex = maybeSharedMutex("sync-$identifier")
+    internal val writeLockMutex = Mutex() // Not used on the web, can be local.
 
     fun removeUsage() {
         collection.synchronize {

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

@@ -15,10 +15,8 @@ import com.powersync.db.crud.CrudRow
 import com.powersync.db.crud.CrudTransaction
 import com.powersync.db.driver.SQLiteConnectionLease
 import com.powersync.db.driver.SQLiteConnectionPool
-import com.powersync.db.internal.ConnectionContext
 import com.powersync.db.internal.InternalDatabaseImpl
 import com.powersync.db.internal.InternalTable
-import com.powersync.db.internal.PowerSyncTransaction
 import com.powersync.db.internal.PowerSyncVersion
 import com.powersync.db.schema.Schema
 import com.powersync.sync.CoreSyncStatus
@@ -27,6 +25,7 @@ import com.powersync.sync.SyncOptions
 import com.powersync.sync.SyncStatus
 import com.powersync.sync.SyncStatusData
 import com.powersync.sync.SyncStream
+import com.powersync.utils.HeldMutex
 import com.powersync.utils.JsonParam
 import com.powersync.utils.JsonUtil
 import com.powersync.utils.throttle
@@ -36,7 +35,6 @@ import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.async
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.completeWith
 import kotlinx.coroutines.ensureActive
@@ -196,12 +194,12 @@ internal class PowerSyncDatabaseImpl(
                 val streamMutex = resource.group.syncMutex
 
                 // Poke the streaming mutex to see if another client is using it
-                var obtainedLock = false
+                var obtainedLock: HeldMutex? = null
                 try {
                     // This call will throw if the lock is already held by this db client.
                     // We should never reach that point since we disconnect before connecting.
-                    obtainedLock = streamMutex.tryLock(db)
-                    if (!obtainedLock) {
+                    obtainedLock = streamMutex.tryAcquire(db)
+                    if (obtainedLock == null) {
                         // The mutex is held already by another PowerSync instance (owner).
                         // (The tryLock should throw if this client already holds the lock).
                         logger.w(streamConflictMessage)
@@ -211,17 +209,15 @@ internal class PowerSyncDatabaseImpl(
                 }
 
                 // This effectively queues operations
-                if (!obtainedLock) {
+                if (obtainedLock == null) {
                     // This will throw a CancellationException if the job was cancelled while waiting.
-                    streamMutex.lock(db)
+                    obtainedLock = streamMutex.acquire(db)
                 }
 
                 // We have a lock if we reached here
-                try {
+                obtainedLock.use {
                     ensureActive()
                     stream.streamingSync()
-                } finally {
-                    streamMutex.unlock(db)
                 }
             }
 

common/src/commonMain/kotlin/com/powersync/utils/PowerSyncMutex.kt

@@ -0,0 +1,53 @@
+package com.powersync.utils
+
+import kotlinx.coroutines.sync.Mutex
+
+/**
+ * The subset of [Mutex] used by the PowerSync SDK.
+ *
+ * This mutex also has a web-specific implementation based on the web locks API.
+ */
+internal interface PowerSyncMutex {
+    suspend fun acquire(owner: Any? = null): HeldMutex
+
+    suspend fun tryAcquire(owner: Any? = null): HeldMutex?
+}
+
+private class LocalMutex : PowerSyncMutex {
+    private val mutex = Mutex()
+
+    override suspend fun acquire(owner: Any?): HeldMutex {
+        mutex.lock(owner)
+
+        return object : HeldMutex {
+            override fun close() {
+                mutex.unlock(owner)
+            }
+        }
+    }
+
+    override suspend fun tryAcquire(owner: Any?): HeldMutex? {
+        if (!mutex.tryLock(owner)) return null
+
+        return object : HeldMutex {
+            override fun close() {
+                mutex.unlock(owner)
+            }
+        }
+    }
+}
+
+/**
+ * A unique [PowerSyncMutex] that is not shared across multiple processes or tabs.
+ */
+internal fun localMutex(): PowerSyncMutex = LocalMutex()
+
+/**
+ * A mutex that is shared across tabs on the web (that is, [PowerSyncMutex.acquire] is serialized
+ * even across tabs).
+ *
+ * On native and JVM targets, this returns a [localMutex].
+ */
+internal expect fun maybeSharedMutex(name: String): PowerSyncMutex
+
+internal interface HeldMutex : AutoCloseable

common/src/commonNonWeb/kotlin/com/powersync/utils/PowerSyncMutex.commonNonWeb.kt

@@ -0,0 +1,3 @@
+package com.powersync.utils
+
+internal actual fun maybeSharedMutex(name: String): PowerSyncMutex = localMutex()

common/src/webMain/kotlin/com/powersync/utils/AbortSignal.kt

@@ -0,0 +1,47 @@
+@file:OptIn(ExperimentalWasmJsInterop::class)
+
+package com.powersync.utils
+
+import com.powersync.internal.InternalPowerSyncAPI
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
+import kotlin.js.ExperimentalWasmJsInterop
+import kotlin.js.JsAny
+
+/**
+ * Runs a suspending function with a JavaScript abort signal hooked to its job.
+ */
+@InternalPowerSyncAPI
+public suspend fun <T> withAbortSignal(block: suspend CoroutineScope.(JsAny) -> T): T {
+    val controller = AbortController()
+    var isCompleted = false
+
+    return coroutineScope {
+        val abortOnCancellation =
+            launch {
+                try {
+                    awaitCancellation()
+                } finally {
+                    if (!isCompleted) {
+                        controller.abort()
+                    }
+                }
+            }
+
+        try {
+            block(controller.signal)
+        } finally {
+            isCompleted = true
+            abortOnCancellation.cancel()
+        }
+    }
+}
+
+@InternalPowerSyncAPI
+private external class AbortController : JsAny {
+    val signal: JsAny
+
+    fun abort()
+}

common/src/webMain/kotlin/com/powersync/utils/PowerSyncMutex.web.kt

@@ -0,0 +1,79 @@
+@file:OptIn(ExperimentalWasmJsInterop::class)
+
+package com.powersync.utils
+
+import com.powersync.internal.InternalPowerSyncAPI
+import com.powersync.web.LockManager
+import com.powersync.web.navigator
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.withContext
+import kotlin.js.ExperimentalWasmJsInterop
+import kotlin.js.JsAny
+import kotlin.js.JsBoolean
+import kotlin.js.Promise
+import kotlin.js.asJsException
+import kotlin.js.js
+import kotlin.js.toBoolean
+
+internal actual fun maybeSharedMutex(name: String): PowerSyncMutex = NavigatorLocksMutex(name)
+
+@OptIn(InternalPowerSyncAPI::class)
+private class NavigatorLocksMutex(
+    private val name: String,
+    private val locks: LockManager = navigator().locks,
+) : PowerSyncMutex {
+    override suspend fun acquire(owner: Any?): HeldMutex =
+        withAbortSignal { signal ->
+            // If we don't pass ifAvailable, this will never return null.
+            acquireInternal(lockOptions(signal))!!
+        }
+
+    override suspend fun tryAcquire(owner: Any?): HeldMutex? = acquireInternal(ifAvailableLockOptions())
+
+    private suspend fun acquireInternal(options: JsAny): HeldMutex? {
+        val acquiredLock = CompletableDeferred<HeldMutex?>()
+
+        locks
+            .request(name, options) { lock ->
+                Promise { resolve, _ ->
+                    acquiredLock.complete(
+                        if (lock == null) {
+                            resolve(null)
+                            null
+                        } else {
+                            PromiseBasedHeldMutex(resolve)
+                        },
+                    )
+                }
+            }.catch { rejection ->
+                if (isAbortError(rejection as JsAny).toBoolean()) {
+                    acquiredLock.completeExceptionally(CancellationException("Acquiring navigator lock $name cancelled"))
+                } else {
+                    acquiredLock.completeExceptionally(rejection.asJsException())
+                }
+
+                null
+            }
+
+        // Make this non-cancellable, we abort the lock request if needed.
+        return withContext(NonCancellable) {
+            acquiredLock.await()
+        }
+    }
+}
+
+private class PromiseBasedHeldMutex(
+    private val resolve: (JsAny?) -> Unit,
+) : HeldMutex {
+    override fun close() {
+        resolve(null)
+    }
+}
+
+private fun lockOptions(abortSignal: JsAny): JsAny = js("({ signal: abortSignal })")
+
+private fun ifAvailableLockOptions(): JsAny = js("({ ifAvailable: true })")
+
+private fun isAbortError(e: JsAny): JsBoolean = js("e.name === 'AbortError'")

common/src/webMain/kotlin/com/powersync/web/WebLocks.kt

@@ -0,0 +1,24 @@
+@file:OptIn(ExperimentalWasmJsInterop::class)
+
+package com.powersync.web
+
+import kotlin.js.ExperimentalWasmJsInterop
+import kotlin.js.JsAny
+import kotlin.js.Promise
+import kotlin.js.js
+
+internal external interface LockManager : JsAny {
+    fun request(
+        name: String,
+        options: JsAny,
+        callback: (lock: Lock?) -> Promise<JsAny?>,
+    ): Promise<JsAny>
+}
+
+internal external interface NavigatorLocksOwner : JsAny {
+    val locks: LockManager
+}
+
+internal external interface Lock
+
+internal fun navigator(): NavigatorLocksOwner = js("navigator")

common/src/webTest/kotlin/com/powersync/utils/PowerSyncMutexTest.kt

@@ -0,0 +1,62 @@
+@file:OptIn(ExperimentalWasmJsInterop::class)
+
+package com.powersync.utils
+
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.matchers.shouldBe
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.await
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+import kotlin.js.ExperimentalWasmJsInterop
+import kotlin.js.JsAny
+import kotlin.js.JsArray
+import kotlin.js.Promise
+import kotlin.js.js
+import kotlin.js.length
+import kotlin.test.Test
+import kotlin.time.Duration.Companion.milliseconds
+
+class PowerSyncMutexTest {
+    @Test
+    fun `can acquire mutex`() =
+        runTest {
+            val mutex = maybeSharedMutex("can-acquire")
+            mutex.acquire().use { }
+        }
+
+    @Test
+    fun `can abort`() =
+        runTest {
+            val mutex = maybeSharedMutex("can-abort")
+            val held = mutex.acquire()
+
+            withContext(Dispatchers.Default) {
+                shouldThrow<CancellationException> {
+                    withTimeout(100.milliseconds) {
+                        mutex.acquire()
+                    }
+                }
+            }
+
+            // Should not have a pending lock request after aborting.
+            val pending = pendingRequests().await()
+            pending.length shouldBe 0
+
+            held.close()
+        }
+
+    @Test
+    fun tryAcquire() =
+        runTest {
+            val mutex = maybeSharedMutex("try-acquire")
+            val held = mutex.tryAcquire()!!
+
+            mutex.tryAcquire() shouldBe null
+            held.close()
+        }
+}
+
+private fun pendingRequests(): Promise<JsArray<JsAny>> = js("navigator.locks.query().then((e) => e.pending)")

core/src/webMain/kotlin/com/powersync/web/DartWorkerDatabase.kt

@@ -6,6 +6,7 @@ import androidx.sqlite.SQLiteStatement
 import com.powersync.db.driver.SQLiteConnectionLease
 import com.powersync.db.driver.SQLiteConnectionPool
 import com.powersync.internal.InternalPowerSyncAPI
+import com.powersync.utils.withAbortSignal
 import kotlinx.coroutines.await
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow