Merge pull request #31 from powersync-ja/fix-mutex-reuse

Fix issue with re-using a shared Mutex or isolateConnectionFactory

Author: rkistner

lib/src/isolate_connection_factory.dart

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

lib/src/mutex.dart

@@ -188,7 +188,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.
@@ -201,7 +202,6 @@ class _SharedMutexServer {
   Completer? unlock;
   late final SerializedMutex serialized;
   final Mutex mutex;
-  bool closed = false;
 
   late final PortServer server;
 
@@ -216,11 +216,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();
@@ -231,10 +226,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();
     }
   }
 
@@ -251,11 +242,6 @@ class _UnlockMessage {
   const _UnlockMessage();
 }
 
-/// Unlock and close
-class _CloseMessage {
-  const _CloseMessage();
-}
-
 class LockError extends Error {
   final String message;
 

test/mutex_test.dart

@@ -22,6 +22,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)));
 }