Share crud and sync mutexes

Author: simolus3

packages/powersync_core/lib/src/database/active_instances.dart

@@ -18,14 +18,23 @@ final class ActiveDatabaseGroup {
 
   /// Use to prevent multiple connections from being opened concurrently
   final Mutex syncConnectMutex = Mutex();
+  final Mutex syncMutex;
+  final Mutex crudMutex;
+
   final String identifier;
 
-  ActiveDatabaseGroup._(this.identifier);
+  ActiveDatabaseGroup._(this.identifier)
+      : syncMutex = Mutex(identifier: '$identifier-sync'),
+        crudMutex = Mutex(identifier: '$identifier-crud');
 
-  void close() {
+  Future<void> close() async {
     if (--refCount == 0) {
       final removedGroup = _activeGroups.remove(identifier);
       assert(removedGroup == this);
+
+      await syncConnectMutex.close();
+      await syncMutex.close();
+      await crudMutex.close();
     }
   }
 

packages/powersync_core/lib/src/database/native/native_powersync_database.dart

@@ -46,8 +46,6 @@ class PowerSyncDatabaseImpl
   @protected
   late Future<void> isInitialized;
 
-  final SimpleMutex _syncMutex = SimpleMutex(), _crudMutex = SimpleMutex();
-
   @override
 
   /// The Logger used by this [PowerSyncDatabase].
@@ -223,6 +221,9 @@ class PowerSyncDatabaseImpl
       receivedIsolateExit.complete();
     });
 
+    final crudMutex = group.crudMutex as SimpleMutex;
+    final syncMutex = group.syncMutex as SimpleMutex;
+
     // Spawning the isolate can't be interrupted
     triedSpawningIsolate = true;
     await Isolate.spawn(
@@ -232,8 +233,8 @@ class PowerSyncDatabaseImpl
         dbRef,
         retryDelay,
         clientParams,
-        _crudMutex.shared,
-        _syncMutex.shared,
+        crudMutex.shared,
+        syncMutex.shared,
       ),
       debugName: 'Sync ${database.openFactory.path}',
       onError: receiveUnhandledErrors.sendPort,
@@ -269,14 +270,6 @@ class PowerSyncDatabaseImpl
     return database.writeLock(callback,
         debugContext: debugContext, lockTimeout: lockTimeout);
   }
-
-  @override
-  Future<void> close() async {
-    await super.close();
-
-    await _crudMutex.close();
-    await _crudMutex.close();
-  }
 }
 
 class _PowerSyncDatabaseIsolateArgs {
@@ -311,24 +304,32 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
   StreamingSyncImplementation? openedStreamingSync;
   StreamSubscription<void>? localUpdatesSubscription;
 
-  Future<void> shutdown() async {
-    await openedStreamingSync?.abort();
-
-    localUpdatesSubscription?.cancel();
-    db?.dispose();
-    crudUpdateController.close();
-    upstreamDbClient.close();
-
-    // The SyncSqliteConnection uses this mutex
-    // It needs to be closed before killing the isolate
-    // in order to free the mutex for other operations.
-    await mutex.close();
-    await crudMutex.close();
-    await syncMutex.close();
-    rPort.close();
+  Completer<void> shutdownCompleter = Completer();
+
+  Future<void> shutdown() {
+    if (!shutdownCompleter.isCompleted) {
+      shutdownCompleter.complete(Future(() async {
+        await openedStreamingSync?.abort();
+
+        localUpdatesSubscription?.cancel();
+        db?.dispose();
+        crudUpdateController.close();
+        upstreamDbClient.close();
+
+        // The SyncSqliteConnection uses this mutex
+        // It needs to be closed before killing the isolate
+        // in order to free the mutex for other operations.
+        await mutex.close();
+        await crudMutex.close();
+        await syncMutex.close();
+        rPort.close();
+
+        // TODO: If we closed our resources properly, this wouldn't be necessary...
+        Isolate.current.kill();
+      }));
+    }
 
-    // TODO: If we closed our resources properly, this wouldn't be necessary...
-    Isolate.current.kill();
+    return shutdownCompleter.future;
   }
 
   rPort.listen((message) async {
@@ -415,12 +416,12 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
       updateDebouncer ??=
           Timer(const Duration(milliseconds: 1), maybeFireUpdates);
     });
-  }, (error, stack) {
+  }, (error, stack) async {
     // Properly dispose the database if an uncaught error occurs.
     // Unfortunately, this does not handle disposing while the database is opening.
     // This should be rare - any uncaught error is a bug. And in most cases,
     // it should occur after the database is already open.
-    shutdown();
+    await shutdown();
     throw error;
   });
 }

packages/powersync_core/lib/src/database/powersync_db_mixin.dart

@@ -52,6 +52,14 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
 
   late final ActiveDatabaseGroup _activeGroup;
 
+  /// An [ActiveDatabaseGroup] sharing mutexes for the sync client.
+  ///
+  /// This is used to ensure that, even if two databases to the same file are
+  /// open concurrently, they won't both open a sync stream. Doing so would
+  /// waste resources.
+  @internal
+  ActiveDatabaseGroup get group => _activeGroup;
+
   @override
 
   /// Broadcast stream that is notified of any table updates.
@@ -242,7 +250,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
       // If there are paused subscriptionso n the status stream, don't delay
       // closing the database because of that.
       unawaited(statusStreamController.close());
-      _activeGroup.close();
+      await _activeGroup.close();
     }
   }
 

packages/powersync_core/lib/src/streaming_sync.dart

@@ -150,8 +150,7 @@ class StreamingSyncImplementation implements StreamingSync {
             invalidCredentials = false;
           }
           // Protect sync iterations with exclusivity (if a valid Mutex is provided)
-          await syncMutex.lock(
-              () => streamingSyncIteration(abortController: _abort),
+          await syncMutex.lock(() => streamingSyncIteration(),
               timeout: retryDelay);
         } catch (e, stacktrace) {
           if (aborted && e is http.ClientException) {
@@ -345,11 +344,13 @@ class StreamingSyncImplementation implements StreamingSync {
     return (initialRequests, localDescriptions);
   }
 
-  Future<void> streamingSyncIteration(
-      {AbortController? abortController}) async {
+  Future<void> streamingSyncIteration() async {
     adapter.startSession();
 
     var (bucketRequests, bucketMap) = await _collectLocalBucketState();
+    if (aborted) {
+      return;
+    }
 
     Checkpoint? targetCheckpoint;
     Checkpoint? validatedCheckpoint;
@@ -388,8 +389,7 @@ class StreamingSyncImplementation implements StreamingSync {
           await adapter.removeBuckets([...bucketsToDelete]);
           _updateStatus(downloading: true);
         case StreamingSyncCheckpointComplete():
-          final result =
-              await _applyCheckpoint(targetCheckpoint!, abortController);
+          final result = await _applyCheckpoint(targetCheckpoint!, _abort);
           if (result.abort) {
             return;
           }
@@ -479,8 +479,7 @@ class StreamingSyncImplementation implements StreamingSync {
                 downloadError: _noError,
                 lastSyncedAt: DateTime.now());
           } else if (validatedCheckpoint == targetCheckpoint) {
-            final result =
-                await _applyCheckpoint(targetCheckpoint!, abortController);
+            final result = await _applyCheckpoint(targetCheckpoint!, _abort);
             if (result.abort) {
               return;
             }