feat: use navigator locks

Author: stevensJourney

packages/sqlite_async/lib/src/common/mutex.dart

@@ -1,8 +1,12 @@
 import 'package:sqlite_async/src/impl/mutex_impl.dart';
 
 abstract class Mutex {
-  factory Mutex() {
-    return MutexImpl();
+  factory Mutex(
+      {
+      /// An optional identifier for this Mutex instance.
+      /// This could be used for platform specific logic or debugging purposes.
+      String? identifier}) {
+    return MutexImpl(identifier: identifier);
   }
 
   /// timeout is a timeout for acquiring the lock, not for the callback

packages/sqlite_async/lib/src/impl/stub_mutex.dart

@@ -1,6 +1,10 @@
 import 'package:sqlite_async/src/common/mutex.dart';
 
 class MutexImpl implements Mutex {
+  String? identifier;
+
+  MutexImpl({this.identifier});
+
   @override
   Future<void> close() {
     throw UnimplementedError();

packages/sqlite_async/lib/src/web/web_mutex.dart

@@ -1,13 +1,36 @@
+import 'dart:async';
+import 'dart:js_interop_unsafe';
+import 'dart:math';
+
 import 'package:mutex/mutex.dart' as mutex;
 import 'package:sqlite_async/src/common/mutex.dart';
+import 'dart:js_interop';
+import 'package:web/web.dart';
+
+@JS('navigator')
+external Navigator get _navigator;
+
+@JS('AbortController')
+external AbortController get _abortController;
 
 /// Web implementation of [Mutex]
-/// This should use `navigator.locks` in future
 class MutexImpl implements Mutex {
-  late final mutex.Mutex m;
+  late final mutex.Mutex fallback;
+  String? identifier;
+  String _resolvedIdentifier;
+
+  MutexImpl({this.identifier})
 
-  MutexImpl() {
-    m = mutex.Mutex();
+      /// On web a lock name is required for Navigator locks.
+      /// Having exclusive Mutex instances requires a somewhat unique lock name.
+      /// This provides a best effort unique identifier, if no identifier is provided.
+      /// This should be fine for most use cases:
+      ///    - The uuid package could be added for better uniqueness if required.
+      ///    - This would add another package dependency to `sqlite_async` which is potentially unnecessary at this point.
+      /// An identifier should be supplied for better exclusion.
+      : _resolvedIdentifier = identifier ??
+            "${DateTime.now().microsecondsSinceEpoch}-${Random().nextDouble()}" {
+    fallback = mutex.Mutex();
   }
 
   @override
@@ -17,8 +40,77 @@ class MutexImpl implements Mutex {
 
   @override
   Future<T> lock<T>(Future<T> Function() callback, {Duration? timeout}) {
-    // Note this lock is only valid in a single web tab
-    return m.protect(callback);
+    if ((_navigator as JSObject).hasProperty('locks'.toJS).toDart) {
+      return _webLock(callback, timeout: timeout);
+    } else {
+      return _fallbackLock(callback, timeout: timeout);
+    }
+  }
+
+  Future<T> _fallbackLock<T>(Future<T> Function() callback,
+      {Duration? timeout}) {
+    final completer = Completer<T>();
+    // Need to implement timeout manually for this
+    bool isTimedOut = false;
+    bool lockObtained = false;
+    if (timeout != null) {
+      Future.delayed(timeout, () {
+        isTimedOut = true;
+        if (lockObtained == false) {
+          completer.completeError(LockError('Timeout reached'));
+        }
+      });
+    }
+
+    fallback.protect(() async {
+      try {
+        if (isTimedOut) {
+          // Don't actually run logic
+          return;
+        }
+        lockObtained = true;
+        final result = await callback();
+        completer.complete(result);
+      } catch (ex) {
+        completer.completeError(ex);
+      }
+    });
+
+    return completer.future;
+  }
+
+  Future<T> _webLock<T>(Future<T> Function() callback, {Duration? timeout}) {
+    final completer = Completer<T>();
+    // Navigator locks can be timed out by using an AbortSignal
+    final controller = AbortController();
+
+    bool lockAcquired = false;
+    if (timeout != null) {
+      // Can't really abort the `delayed` call easily :(
+      Future.delayed(timeout, () {
+        if (lockAcquired == true) {
+          return;
+        }
+        completer.completeError(LockError('Timeout reached'));
+        controller.abort('Timeout'.toJS);
+      });
+    }
+
+    JSPromise jsCallback(JSAny lock) {
+      lockAcquired = true;
+      callback().then((value) {
+        completer.complete(value);
+      }).catchError((error) {
+        completer.completeError(error);
+      });
+      return completer.future.toJS;
+    }
+
+    final lockOptions = JSObject();
+    lockOptions['signal'] = controller.signal;
+    _navigator.locks.request(_resolvedIdentifier, lockOptions, jsCallback.toJS);
+
+    return completer.future;
   }
 
   @override

packages/sqlite_async/pubspec.yaml

@@ -18,6 +18,7 @@ dependencies:
   collection: ^1.17.0
   mutex: ^3.1.0
   meta: ^1.10.0
+  web: ^0.5.1
 
 dev_dependencies:
   dcli: ^4.0.0

packages/sqlite_async/test/mutex_test.dart

@@ -1,67 +1,80 @@
-@TestOn('!browser')
-import 'dart:isolate';
+import 'dart:math';
 
-import 'package:sqlite_async/src/native/native_isolate_mutex.dart';
+import 'package:sqlite_async/sqlite_async.dart';
 import 'package:test/test.dart';
 
+import 'utils/test_utils_impl.dart';
+
+final testUtils = TestUtils();
+
 void main() {
-  group('Mutex Tests', () {
-    test('Closing', () async {
-      // Test that locks are properly released when calling SharedMutex.close()
-      // in in Isolate.
-      // A timeout in this test indicates a likely error.
-      for (var i = 0; i < 50; i++) {
-        final mutex = SimpleMutex();
-        final serialized = mutex.shared;
-
-        final result = await Isolate.run(() async {
-          return _lockInIsolate(serialized);
+  group('Shared Mutex Tests', () {
+    test('Queue exclusive operations', () async {
+      final m = Mutex();
+      final collection = List.generate(10, (index) => index);
+      final results = <int>[];
+
+      final futures = collection.map((element) async {
+        return m.lock(() async {
+          // Simulate some asynchronous work
+          await Future.delayed(Duration(milliseconds: Random().nextInt(100)));
+          results.add(element);
+          return element;
         });
+      }).toList();
 
-        await mutex.lock(() async {});
+      // Await all the promises
+      await Future.wait(futures);
 
-        expect(result, equals(5));
-      }
+      // Check if the results are in ascending order
+      expect(results, equals(collection));
     });
+  });
 
-    test('Re-use after closing', () async {
-      // Test that shared locks can be opened and closed multiple times.
-      final mutex = SimpleMutex();
-      final serialized = mutex.shared;
+  test('Timeout should throw a LockError', () async {
+    final m = Mutex();
+    m.lock(() async {
+      await Future.delayed(Duration(milliseconds: 300));
+    });
 
-      final result = await Isolate.run(() async {
-        return _lockInIsolate(serialized);
-      });
+    await expectLater(
+        m.lock(() async {
+          print('This should not get executed');
+        }, timeout: Duration(milliseconds: 200)),
+        throwsA((e) => e is LockError && e.message.contains('Timeout')));
+  });
 
-      final result2 = await Isolate.run(() async {
-        return _lockInIsolate(serialized);
-      });
+  test('In-time timeout should function normally', () async {
+    final m = Mutex();
+    final results = [];
+    m.lock(() async {
+      await Future.delayed(Duration(milliseconds: 100));
+      results.add(1);
+    });
 
-      await mutex.lock(() async {});
+    await m.lock(() async {
+      results.add(2);
+    }, timeout: Duration(milliseconds: 200));
 
-      expect(result, equals(5));
-      expect(result2, equals(5));
-    });
-  }, timeout: const Timeout(Duration(milliseconds: 5000)));
-}
+    expect(results, equals([1, 2]));
+  });
 
-Future<Object> _lockInIsolate(
-  SerializedMutex smutex,
-) async {
-  final mutex = smutex.open();
-  // Start a "thread" that repeatedly takes a lock
-  _infiniteLock(mutex).ignore();
-  await Future.delayed(const Duration(milliseconds: 10));
-  // Then close the mutex while the above loop is running.
-  await mutex.close();
-
-  return 5;
-}
+  test('Different Mutex instances should cause separate locking', () async {
+    final m1 = Mutex();
+    final m2 = Mutex();
 
-Future<void> _infiniteLock(SharedMutex mutex) async {
-  while (true) {
-    await mutex.lock(() async {
-      await Future.delayed(const Duration(milliseconds: 1));
+    final results = [];
+    final p1 = m1.lock(() async {
+      await Future.delayed(Duration(milliseconds: 300));
+      results.add(1);
     });
-  }
+
+    final p2 = m2.lock(() async {
+      results.add(2);
+    });
+
+    await p1;
+    await p2;
+    expect(results, equals([2, 1]));
+  });
 }

packages/sqlite_async/test/native/native_mutex_test.dart

@@ -0,0 +1,67 @@
+@TestOn('!browser')
+import 'dart:isolate';
+
+import 'package:sqlite_async/src/native/native_isolate_mutex.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('Mutex Tests', () {
+    test('Closing', () async {
+      // Test that locks are properly released when calling SharedMutex.close()
+      // in in Isolate.
+      // A timeout in this test indicates a likely error.
+      for (var i = 0; i < 50; i++) {
+        final mutex = SimpleMutex();
+        final serialized = mutex.shared;
+
+        final result = await Isolate.run(() async {
+          return _lockInIsolate(serialized);
+        });
+
+        await mutex.lock(() async {});
+
+        expect(result, equals(5));
+      }
+    });
+
+    test('Re-use after closing', () async {
+      // Test that shared locks can be opened and closed multiple times.
+      final mutex = SimpleMutex();
+      final serialized = mutex.shared;
+
+      final result = await Isolate.run(() async {
+        return _lockInIsolate(serialized);
+      });
+
+      final result2 = await Isolate.run(() async {
+        return _lockInIsolate(serialized);
+      });
+
+      await mutex.lock(() async {});
+
+      expect(result, equals(5));
+      expect(result2, equals(5));
+    });
+  }, timeout: const Timeout(Duration(milliseconds: 5000)));
+}
+
+Future<Object> _lockInIsolate(
+  SerializedMutex smutex,
+) async {
+  final mutex = smutex.open();
+  // Start a "thread" that repeatedly takes a lock
+  _infiniteLock(mutex).ignore();
+  await Future.delayed(const Duration(milliseconds: 10));
+  // Then close the mutex while the above loop is running.
+  await mutex.close();
+
+  return 5;
+}
+
+Future<void> _infiniteLock(SharedMutex mutex) async {
+  while (true) {
+    await mutex.lock(() async {
+      await Future.delayed(const Duration(milliseconds: 1));
+    });
+  }
+}

packages/sqlite_async/test/web/web_mutex_test.dart

@@ -0,0 +1,29 @@
+import 'package:sqlite_async/sqlite_async.dart';
+import 'package:test/test.dart';
+
+import '../utils/test_utils_impl.dart';
+
+final testUtils = TestUtils();
+
+void main() {
+  group('Web Mutex Tests', () {
+    test('Web should share locking with identical identifiers', () async {
+      final m1 = Mutex(identifier: 'sync');
+      final m2 = Mutex(identifier: 'sync');
+
+      final results = [];
+      final p1 = m1.lock(() async {
+        results.add(1);
+      });
+
+      final p2 = m2.lock(() async {
+        results.add(2);
+      });
+
+      await p1;
+      await p2;
+      // It should be correctly ordered as if it was the same mutex
+      expect(results, equals([1, 2]));
+    });
+  });
+}