[Alpha] Fix mutex reuse (#32)

* Fix Mutexed reuse in alpha

Author: stevensJourney

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.7.0-alpha.2
+
+- Fix re-using a shared Mutex from https://github.com/powersync-ja/sqlite_async.dart/pull/31
+
 ## 0.7.0-alpha.1
 
 - Added initial support for web platform.

lib/src/isolate_connection_factory.dart

@@ -1,5 +1,4 @@
 // This follows the pattern from here: https://stackoverflow.com/questions/58710226/how-to-import-platform-specific-dependency-in-flutter-dart-combine-web-with-an
 // To conditionally export an implementation for either web or "native" platforms
 // The sqlite library uses dart:ffi which is not supported on web
-
 export 'impl/isolate_connection_factory_impl.dart';

lib/src/native/native_isolate_connection_factory.dart

@@ -44,8 +44,8 @@ class IsolateConnectionFactoryImpl
         readOnly: readOnly,
         debugName: debugName,
         updates: updates.stream,
-        closeFunction: () {
-          openMutex.close();
+        closeFunction: () async {
+          await openMutex.close();
           updates.close();
         });
   }
@@ -83,7 +83,7 @@ class _IsolateUpdateListener {
 }
 
 class _IsolateSqliteConnection extends SqliteConnectionImpl {
-  final void Function() closeFunction;
+  final Future<void> Function() closeFunction;
 
   _IsolateSqliteConnection(
       {required super.openFactory,
@@ -97,6 +97,6 @@ class _IsolateSqliteConnection extends SqliteConnectionImpl {
   @override
   Future<void> close() async {
     await super.close();
-    closeFunction();
+    await closeFunction();
   }
 }

lib/src/native/native_isolate_mutex.dart

@@ -112,7 +112,7 @@ class SimpleMutex implements MutexImpl {
 class SerializedMutex implements Mutex {
   final SerializedPortClient client;
 
-  SerializedMutex(this.client);
+  const SerializedMutex(this.client);
 
   @override
   SharedMutex open() {
@@ -202,7 +202,8 @@ class SharedMutex implements Mutex {
     closed = true;
     // Wait for any existing locks to complete, then prevent any further locks from being taken out.
     await _acquire();
-    client.fire(const _CloseMessage());
+    // Release the lock
+    _unlock();
     // Close client immediately after _unlock(),
     // so that we're sure no further locks are acquired.
     // This also cancels any lock request in process.
@@ -215,7 +216,6 @@ class _SharedMutexServer {
   Completer? unlock;
   late final SerializedMutex serialized;
   final Mutex mutex;
-  bool closed = false;
 
   late final PortServer server;
 
@@ -230,11 +230,6 @@ class _SharedMutexServer {
     if (arg is _AcquireMessage) {
       var lock = Completer.sync();
       mutex.lock(() async {
-        if (closed) {
-          // The client will error already - we just need to ensure
-          // we don't take out another lock.
-          return;
-        }
         assert(unlock == null);
         unlock = Completer.sync();
         lock.complete();
@@ -245,10 +240,6 @@ class _SharedMutexServer {
     } else if (arg is _UnlockMessage) {
       assert(unlock != null);
       unlock!.complete();
-    } else if (arg is _CloseMessage) {
-      // Unlock and close (from client side)
-      closed = true;
-      unlock?.complete();
     }
   }
 
@@ -264,8 +255,3 @@ class _AcquireMessage {
 class _UnlockMessage {
   const _UnlockMessage();
 }
-
-/// Unlock and close
-class _CloseMessage {
-  const _CloseMessage();
-}

pubspec.yaml

@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.7.0-alpha.1
+version: 0.7.0-alpha.2
 repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
   sdk: '>=3.2.0 <4.0.0'

test/mutex_test.dart

@@ -23,6 +23,25 @@ void main() {
         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)));
 }