chore: migrate locks to async-mutex

Author: stevensJourney

.changeset/beige-pigs-occur.md

@@ -0,0 +1,6 @@
+---
+'@powersync/op-sqlite': minor
+'@powersync/capacitor': minor
+---
+
+Removed `async-lock` dependency in favor of `async-mutex`.

.changeset/cool-crews-hug.md

@@ -0,0 +1,6 @@
+---
+'@powersync/op-sqlite': patch
+'@powersync/capacitor': patch
+---
+
+Fixed potential issue where extreme amounts of concurrent calls to `writeLock` could reject with the error "Too many pending tasks in queue"

.changeset/little-dancers-beam.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': minor
+---
+
+Exported common Mutex helper functions

demos/example-capacitor/ios/App/Podfile.lock

@@ -9,7 +9,7 @@ PODS:
   - CapacitorSplashScreen (7.0.0):
     - Capacitor
   - powersync-sqlite-core (0.4.10)
-  - PowersyncCapacitor (0.2.0):
+  - PowersyncCapacitor (0.3.0):
     - Capacitor
     - powersync-sqlite-core (~> 0.4.10)
     - SQLCipher (~> 4.0)
@@ -51,7 +51,7 @@ SPEC CHECKSUMS:
   CapacitorCordova: bf648a636f3c153f652d312ae145fb508b6ffced
   CapacitorSplashScreen: f4e58cc02aafd91c7cbaf32a3d1b44d02a115125
   powersync-sqlite-core: b30017e077c91915d53faebc5f7245384df78275
-  PowersyncCapacitor: 4fe9ac76191de407fd5cc9b74b9792d41e99bf24
+  PowersyncCapacitor: a966fb4c3544e20bfdcc7ec7aad9d364b753e884
   SQLCipher: eb79c64049cb002b4e9fcb30edb7979bf4706dfc
   ZIPFoundation: dfd3d681c4053ff7e2f7350bc4e53b5dba3f5351
 

packages/capacitor/package.json

@@ -75,7 +75,6 @@
     "@ionic/eslint-config": "^0.4.0",
     "@ionic/prettier-config": "^4.0.0",
     "@ionic/swiftlint-config": "^2.0.0",
-    "@types/async-lock": "catalog:",
     "eslint": "^8.57.0",
     "prettier": "catalog:",
     "prettier-plugin-java": "^2.6.6",
@@ -100,6 +99,6 @@
     }
   },
   "dependencies": {
-    "async-lock": "catalog:"
+    "async-mutex": "catalog:"
   }
 }

packages/capacitor/src/adapter/CapacitorSQLiteAdapter.ts

@@ -8,10 +8,11 @@ import {
   DBAdapterListener,
   DBLockOptions,
   LockContext,
+  mutexRunExclusive,
   QueryResult,
   Transaction
 } from '@powersync/web';
-import Lock from 'async-lock';
+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';
@@ -39,13 +40,15 @@ export class CapacitorSQLiteAdapter extends BaseObserver<DBAdapterListener> impl
   protected _writeConnection: SQLiteDBConnection | null;
   protected _readConnection: SQLiteDBConnection | null;
   protected initializedPromise: Promise<void>;
-  protected lock: Lock;
+  protected writeMutex: Mutex;
+  protected readMutex: Mutex;
 
   constructor(protected options: CapacitorSQLiteOpenFactoryOptions) {
     super();
     this._writeConnection = null;
     this._readConnection = null;
-    this.lock = new Lock();
+    this.writeMutex = new Mutex();
+    this.readMutex = new Mutex();
     this.initializedPromise = this.init();
   }
 
@@ -237,10 +240,14 @@ export class CapacitorSQLiteAdapter extends BaseObserver<DBAdapterListener> impl
   }
 
   readLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions): Promise<T> {
-    return this.lock.acquire('read_lock', async () => {
-      await this.initializedPromise;
-      return await fn(this.generateLockContext(this.readConnection));
-    });
+    return mutexRunExclusive(
+      this.readMutex,
+      async () => {
+        await this.initializedPromise;
+        return await fn(this.generateLockContext(this.readConnection));
+      },
+      options
+    );
   }
 
   readTransaction<T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions): Promise<T> {
@@ -250,24 +257,28 @@ export class CapacitorSQLiteAdapter extends BaseObserver<DBAdapterListener> impl
   }
 
   writeLock<T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions): Promise<T> {
-    return this.lock.acquire('write_lock', 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;
-    });
+    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
+    );
   }
 
   writeTransaction<T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions): Promise<T> {

packages/capacitor/src/sync/CapacitorSyncImplementation.ts

@@ -1,12 +1,60 @@
-import { AbstractStreamingSyncImplementation, LockOptions } from '@powersync/web';
-import Lock from 'async-lock';
+import { AbstractStreamingSyncImplementation, LockOptions, LockType, mutexRunExclusive } from '@powersync/web';
+import { Mutex } from 'async-mutex';
+
+type MutexMap = {
+  /**
+   * Used to track the consumers of this Mutex.
+   * It should be safe to dispose the Mutex if this is empty.
+   */
+  tracking: Set<number>;
+  locks: {
+    [Type in LockType]: Mutex;
+  };
+};
+
+const GLOBAL_MUTEX_STORE: Map<string, MutexMap> = new Map();
+
+/**
+ * Used to identify multiple instances of CapacitorStreamingSyncImplementation
+ */
+let _CAPACITOR_STREAMING_SYNC_SEQUENCE = 0;
 
 export class CapacitorStreamingSyncImplementation extends AbstractStreamingSyncImplementation {
-  static GLOBAL_LOCK = new Lock();
+  // A unique ID for tacking this specific instance of CapacitorStreamingSyncImplementation
+  protected instanceId = _CAPACITOR_STREAMING_SYNC_SEQUENCE++;
+
+  async dispose(): Promise<void> {
+    await super.dispose();
+
+    // Clear up any global mutexes which aren't used anymore
+    for (const mutexEntry of GLOBAL_MUTEX_STORE.entries()) {
+      const [identifier, mutex] = mutexEntry;
+      if (!mutex.tracking.has(this.instanceId)) {
+        continue;
+      }
+      mutex.tracking.delete(this.instanceId);
+      if (mutex.tracking.size == 0) {
+        GLOBAL_MUTEX_STORE.delete(identifier);
+      }
+    }
+  }
 
   async obtainLock<T>(lockOptions: LockOptions<T>): Promise<T> {
-    const identifier = `streaming-sync-${lockOptions.type}-${this.options.identifier}`;
-    return CapacitorStreamingSyncImplementation.GLOBAL_LOCK.acquire(identifier, async () => {
+    // If we don't have an identifier for some reason (should not happen), we use a shared Mutex
+    const { identifier: baseIdentifier = 'DEFAULT' } = this.options;
+    if (!GLOBAL_MUTEX_STORE.has(baseIdentifier)) {
+      GLOBAL_MUTEX_STORE.set(baseIdentifier, {
+        tracking: new Set([this.instanceId]),
+        locks: {
+          [LockType.CRUD]: new Mutex(),
+          [LockType.SYNC]: new Mutex()
+        }
+      });
+    }
+
+    const mutex = GLOBAL_MUTEX_STORE.get(baseIdentifier)!.locks[lockOptions.type];
+
+    return mutexRunExclusive(mutex, async () => {
       if (lockOptions.signal?.aborted) {
         throw new Error('Aborted');
       }

packages/common/src/index.ts

@@ -61,6 +61,7 @@ export * from './utils/BaseObserver.js';
 export * from './utils/ControlledExecutor.js';
 export * from './utils/DataStream.js';
 export * from './utils/Logger.js';
+export * from './utils/mutex.js';
 export * from './utils/parseQuery.js';
 
 export * from './types/types.js';

packages/common/src/utils/mutex.ts

@@ -6,7 +6,7 @@ import { Mutex } from 'async-mutex';
 export async function mutexRunExclusive<T>(
   mutex: Mutex,
   callback: () => Promise<T>,
-  options?: { timeoutMs: number }
+  options?: { timeoutMs?: number }
 ): Promise<T> {
   return new Promise((resolve, reject) => {
     const timeout = options?.timeoutMs;

packages/powersync-op-sqlite/package.json

@@ -72,12 +72,11 @@
   },
   "dependencies": {
     "@powersync/common": "workspace:*",
-    "async-lock": "catalog:"
+    "async-mutex": "catalog:"
   },
   "devDependencies": {
     "@op-engineering/op-sqlite": "catalog:",
     "@react-native/eslint-config": "^0.83.1",
-    "@types/async-lock": "catalog:",
     "@types/react": "^19.1.1",
     "del-cli": "^7.0.0",
     "eslint": "catalog:",