diff --git a/CHANGELOG.md b/CHANGELOG.md index a3a2af8d..4ddb5643 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/common/src/commonMain/kotlin/com/powersync/db/ActiveInstanceStore.kt b/common/src/commonMain/kotlin/com/powersync/db/ActiveInstanceStore.kt index 1ba3faed..46c5b240 100644 --- a/common/src/commonMain/kotlin/com/powersync/db/ActiveInstanceStore.kt +++ b/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 { diff --git a/common/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt b/common/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt index c0b5a475..02615ec7 100644 --- a/common/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt +++ b/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) } } diff --git a/common/src/commonMain/kotlin/com/powersync/utils/PowerSyncMutex.kt b/common/src/commonMain/kotlin/com/powersync/utils/PowerSyncMutex.kt new file mode 100644 index 00000000..9e46b4a6 --- /dev/null +++ b/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 diff --git a/common/src/commonNonWeb/kotlin/com/powersync/utils/PowerSyncMutex.commonNonWeb.kt b/common/src/commonNonWeb/kotlin/com/powersync/utils/PowerSyncMutex.commonNonWeb.kt new file mode 100644 index 00000000..f721a836 --- /dev/null +++ b/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() diff --git a/common/src/webMain/kotlin/com/powersync/utils/AbortSignal.kt b/common/src/webMain/kotlin/com/powersync/utils/AbortSignal.kt new file mode 100644 index 00000000..0fbc407d --- /dev/null +++ b/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 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() +} diff --git a/common/src/webMain/kotlin/com/powersync/utils/PowerSyncMutex.web.kt b/common/src/webMain/kotlin/com/powersync/utils/PowerSyncMutex.web.kt new file mode 100644 index 00000000..1e3c8ecf --- /dev/null +++ b/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() + + 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'") diff --git a/common/src/webMain/kotlin/com/powersync/web/WebLocks.kt b/common/src/webMain/kotlin/com/powersync/web/WebLocks.kt new file mode 100644 index 00000000..085720ac --- /dev/null +++ b/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, + ): Promise +} + +internal external interface NavigatorLocksOwner : JsAny { + val locks: LockManager +} + +internal external interface Lock + +internal fun navigator(): NavigatorLocksOwner = js("navigator") diff --git a/common/src/webTest/kotlin/com/powersync/utils/PowerSyncMutexTest.kt b/common/src/webTest/kotlin/com/powersync/utils/PowerSyncMutexTest.kt new file mode 100644 index 00000000..30f8d1c3 --- /dev/null +++ b/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 { + 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> = js("navigator.locks.query().then((e) => e.pending)") diff --git a/core/src/webMain/kotlin/com/powersync/web/DartWorkerDatabase.kt b/core/src/webMain/kotlin/com/powersync/web/DartWorkerDatabase.kt index d223adc2..cd4dd4bb 100644 --- a/core/src/webMain/kotlin/com/powersync/web/DartWorkerDatabase.kt +++ b/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 diff --git a/core/src/webMain/kotlin/com/powersync/web/Utils.kt b/core/src/webMain/kotlin/com/powersync/web/Utils.kt index 6e5e13ca..87828a44 100644 --- a/core/src/webMain/kotlin/com/powersync/web/Utils.kt +++ b/core/src/webMain/kotlin/com/powersync/web/Utils.kt @@ -5,10 +5,6 @@ package com.powersync.web import androidx.sqlite.SQLiteException import com.powersync.internal.InternalPowerSyncAPI import kotlinx.coroutines.CancellationException -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.awaitCancellation -import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.launch import kotlin.coroutines.resume import kotlin.coroutines.resumeWithException import kotlin.coroutines.suspendCoroutine @@ -112,42 +108,6 @@ internal suspend fun Promise.awaitSafe(): T = ) } -/** - * Runs a suspending function with a JavaScript abort signal hooked to its job. - */ -@OptIn(InternalPowerSyncAPI::class) -internal suspend fun 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 -public external class AbortController : JsAny { - public val signal: JsAny - - public fun abort() -} - private fun isSqliteException(exception: JsAny): Boolean = js("'extendedResultCode' in exception") private fun domErrorName(domError: JsAny): String = js("domError.name")