Use custom mutex implementation instead of async-mutex (#900)

Author: simolus3

.changeset/shaggy-games-approve.md

@@ -0,0 +1,12 @@
+---
+'@powersync/react-native': minor
+'@powersync/common': minor
+'@powersync/web': minor
+'@powersync/op-sqlite': patch
+'@powersync/adapter-sql-js': patch
+'@powersync/capacitor': patch
+'@powersync/node': patch
+'@powersync/nuxt': patch
+---
+
+Remove `async-mutex` dependency in favor of internal implementation.

demos/react-multi-client/package.json

@@ -16,7 +16,6 @@
     "@supabase/supabase-js": "^2.43.1",
     "@vitejs/plugin-react": "^4.2.1",
     "@webflow/webflow-cli": "^1.6.9",
-    "async-mutex": "^0.5.0",
     "autoprefixer": "10.4.14",
     "lodash": "^4.17.21",
     "postcss": "8.4.27",

demos/react-multi-client/src/library/SupabaseConnector.ts

@@ -3,6 +3,7 @@ import {
   BaseListener,
   BaseObserver,
   CrudEntry,
+  Mutex,
   PowerSyncBackendConnector,
   UpdateType
 } from '@powersync/web';
@@ -14,8 +15,6 @@ export interface SupabaseConnectorListener extends BaseListener {
   onCRUDEvent: (event: { crudType: UpdateType; elapsedTimeMs: number }) => void;
 }
 
-import { Mutex } from 'async-mutex';
-
 export class SupabaseConnector extends BaseObserver<SupabaseConnectorListener> implements PowerSyncBackendConnector {
   static SHARED_MUTEX = new Mutex();
   readonly client: SupabaseClient;

packages/adapter-sql-js/package.json

@@ -32,8 +32,7 @@
     "test": "vitest"
   },
   "dependencies": {
-    "@powersync/common": "workspace:^",
-    "async-mutex": "catalog:"
+    "@powersync/common": "workspace:^"
   },
   "devDependencies": {
     "@powersync/sql-js": "0.0.8",

packages/adapter-sql-js/rollup.config.mjs

@@ -31,6 +31,6 @@ export default () => {
         ]
       })
     ],
-    external: ['@powersync/common', 'async-mutex']
+    external: ['@powersync/common']
   };
 };

packages/adapter-sql-js/src/SQLJSAdapter.ts

@@ -12,13 +12,14 @@ import {
   DBLockOptions,
   ILogger,
   LockContext,
+  Mutex,
   QueryResult,
   SqlExecutor,
   SQLOpenFactory,
   SQLOpenOptions,
+  timeoutSignal,
   Transaction
 } from '@powersync/common';
-import { Mutex } from 'async-mutex';
 // This uses a pure JS version which avoids the need for WebAssembly, which is not supported in React Native.
 import SQLJs from '@powersync/sql-js/dist/sql-asm.js';
 
@@ -165,7 +166,7 @@ class SqlJsConnectionPool extends BaseObserver<DBAdapterListener> implements Con
       this.tableUpdateCache.clear();
       this.iterateListeners((l) => l.tablesUpdated?.(notification));
       return result;
-    });
+    }, timeoutSignal(options?.timeoutMs));
   }
 
   async refreshSchema(): Promise<void> {

packages/capacitor/package.json

@@ -96,8 +96,5 @@
     "android": {
       "src": "android"
     }
-  },
-  "dependencies": {
-    "async-mutex": "catalog:"
   }
 }

packages/capacitor/src/adapter/CapacitorSQLiteAdapter.ts

@@ -10,11 +10,11 @@ import {
   DBAdapterListener,
   DBLockOptions,
   LockContext,
-  mutexRunExclusive,
+  Mutex,
   QueryResult,
+  timeoutSignal,
   Transaction
 } from '@powersync/web';
-import { Mutex } from 'async-mutex';
 import { PowerSyncCore } from '../plugin/PowerSyncCore.js';
 import { messageForErrorCode } from '../plugin/PowerSyncPlugin.js';
 import { CapacitorSQLiteOpenFactoryOptions, DEFAULT_SQLITE_OPTIONS } from './CapacitorSQLiteOpenFactory.js';
@@ -228,39 +228,31 @@ class CapacitorConnectionPool extends BaseObserver<DBAdapterListener> implements
   }
 
   readLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions): Promise<T> {
-    return mutexRunExclusive(
-      this.readMutex,
-      async () => {
-        await this.initializedPromise;
-        return await fn(this.generateLockContext(this.readConnection));
-      },
-      options
-    );
+    return this.readMutex.runExclusive(async () => {
+      await this.initializedPromise;
+      return fn(this.generateLockContext(this.readConnection));
+    }, timeoutSignal(options?.timeoutMs));
   }
 
   writeLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions): Promise<T> {
-    return mutexRunExclusive(
-      this.writeMutex,
-      async () => {
-        await this.initializedPromise;
-        const result = await fn(this.generateLockContext(this.writeConnection));
-
-        // Fetch table updates
-        const updates = await this.writeConnection.query("SELECT powersync_update_hooks('get') AS table_name");
-        const jsonUpdates = updates.values?.[0];
-        if (!jsonUpdates || !jsonUpdates.table_name) {
-          throw new Error('Could not fetch table updates');
-        }
-        const notification: BatchedUpdateNotification = {
-          rawUpdates: [],
-          tables: JSON.parse(jsonUpdates.table_name),
-          groupedUpdates: {}
-        };
-        this.iterateListeners((l) => l.tablesUpdated?.(notification));
-        return result;
-      },
-      options
-    );
+    return this.writeMutex.runExclusive(async () => {
+      await this.initializedPromise;
+      const result = await fn(this.generateLockContext(this.writeConnection));
+
+      // Fetch table updates
+      const updates = await this.writeConnection.query("SELECT powersync_update_hooks('get') AS table_name");
+      const jsonUpdates = updates.values?.[0];
+      if (!jsonUpdates || !jsonUpdates.table_name) {
+        throw new Error('Could not fetch table updates');
+      }
+      const notification: BatchedUpdateNotification = {
+        rawUpdates: [],
+        tables: JSON.parse(jsonUpdates.table_name),
+        groupedUpdates: {}
+      };
+      this.iterateListeners((l) => l.tablesUpdated?.(notification));
+      return result;
+    }, timeoutSignal(options?.timeoutMs));
   }
 
   refreshSchema(): Promise<void> {

packages/capacitor/src/sync/CapacitorSyncImplementation.ts

@@ -1,5 +1,4 @@
-import { AbstractStreamingSyncImplementation, LockOptions, LockType, mutexRunExclusive } from '@powersync/web';
-import { Mutex } from 'async-mutex';
+import { AbstractStreamingSyncImplementation, LockOptions, LockType, Mutex } from '@powersync/web';
 
 type MutexMap = {
   /**
@@ -56,11 +55,8 @@ export class CapacitorStreamingSyncImplementation extends AbstractStreamingSyncI
     mutexRecord.tracking.add(this.instanceId);
     const mutex = mutexRecord.locks[lockOptions.type];
 
-    return mutexRunExclusive(mutex, async () => {
-      if (lockOptions.signal?.aborted) {
-        throw new Error('Aborted');
-      }
+    return mutex.runExclusive(async () => {
       return await lockOptions.callback();
-    });
+    }, lockOptions.signal);
   }
 }

packages/common/package.json

@@ -55,7 +55,6 @@
     "test:exports": "attw --pack ."
   },
   "dependencies": {
-    "async-mutex": "catalog:",
     "event-iterator": "^2.0.0"
   },
   "devDependencies": {