Adopt isolate mutex implementations

Author: simolus3

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

@@ -19,6 +19,8 @@ import 'package:powersync_core/src/streaming_sync.dart';
 import 'package:powersync_core/src/sync_status.dart';
 import 'package:sqlite_async/sqlite3_common.dart';
 import 'package:sqlite_async/sqlite_async.dart';
+// ignore: implementation_imports
+import 'package:sqlite_async/src/native/native_isolate_mutex.dart';
 
 /// A PowerSync managed database.
 ///
@@ -44,6 +46,8 @@ class PowerSyncDatabaseImpl
   @protected
   late Future<void> isInitialized;
 
+  final SimpleMutex _syncMutex = SimpleMutex(), _crudMutex = SimpleMutex();
+
   @override
 
   /// The Logger used by this [PowerSyncDatabase].
@@ -224,7 +228,13 @@ class PowerSyncDatabaseImpl
     await Isolate.spawn(
       _syncIsolate,
       _PowerSyncDatabaseIsolateArgs(
-          receiveMessages.sendPort, dbRef, retryDelay, clientParams),
+        receiveMessages.sendPort,
+        dbRef,
+        retryDelay,
+        clientParams,
+        _crudMutex.shared,
+        _syncMutex.shared,
+      ),
       debugName: 'Sync ${database.openFactory.path}',
       onError: receiveUnhandledErrors.sendPort,
       errorsAreFatal: true,
@@ -259,16 +269,32 @@ 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 {
   final SendPort sPort;
   final IsolateConnectionFactory dbRef;
   final Duration retryDelay;
   final Map<String, dynamic>? parameters;
+  final SerializedMutex crudMutex;
+  final SerializedMutex syncMutex;
 
   _PowerSyncDatabaseIsolateArgs(
-      this.sPort, this.dbRef, this.retryDelay, this.parameters);
+    this.sPort,
+    this.dbRef,
+    this.retryDelay,
+    this.parameters,
+    this.crudMutex,
+    this.syncMutex,
+  );
 }
 
 Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
@@ -277,6 +303,9 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
   StreamController<String> crudUpdateController = StreamController.broadcast();
   final upstreamDbClient = args.dbRef.upstreamPort.open();
 
+  final crudMutex = args.crudMutex.open();
+  final syncMutex = args.syncMutex.open();
+
   CommonDatabase? db;
   final Mutex mutex = args.dbRef.mutex.open();
   StreamingSyncImplementation? openedStreamingSync;
@@ -294,6 +323,8 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
     // 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...
@@ -348,14 +379,17 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
 
     final storage = BucketStorage(connection);
     final sync = StreamingSyncImplementation(
-        adapter: storage,
-        credentialsCallback: loadCredentials,
-        invalidCredentialsCallback: invalidateCredentials,
-        uploadCrud: uploadCrud,
-        crudUpdateTriggerStream: crudUpdateController.stream,
-        retryDelay: args.retryDelay,
-        client: http.Client(),
-        syncParameters: args.parameters);
+      adapter: storage,
+      credentialsCallback: loadCredentials,
+      invalidCredentialsCallback: invalidateCredentials,
+      uploadCrud: uploadCrud,
+      crudUpdateTriggerStream: crudUpdateController.stream,
+      retryDelay: args.retryDelay,
+      client: http.Client(),
+      syncParameters: args.parameters,
+      crudMutex: crudMutex,
+      syncMutex: syncMutex,
+    );
     openedStreamingSync = sync;
     sync.streamingSync();
     sync.statusStream.listen((event) {

packages/powersync_core/lib/src/streaming_sync.dart

@@ -74,21 +74,23 @@ class StreamingSyncImplementation implements StreamingSync {
 
   String? clientId;
 
-  StreamingSyncImplementation(
-      {required this.adapter,
-      required this.credentialsCallback,
-      this.invalidCredentialsCallback,
-      required this.uploadCrud,
-      required this.crudUpdateTriggerStream,
-      required this.retryDelay,
-      this.syncParameters,
-      required http.Client client,
-
-      /// A unique identifier for this streaming sync implementation
-      /// A good value is typically the DB file path which it will mutate when syncing.
-      String? identifier = "unknown"})
-      : syncMutex = Mutex(identifier: "sync-$identifier"),
-        crudMutex = Mutex(identifier: "crud-$identifier"),
+  StreamingSyncImplementation({
+    required this.adapter,
+    required this.credentialsCallback,
+    this.invalidCredentialsCallback,
+    required this.uploadCrud,
+    required this.crudUpdateTriggerStream,
+    required this.retryDelay,
+    this.syncParameters,
+    required http.Client client,
+    Mutex? syncMutex,
+    Mutex? crudMutex,
+
+    /// A unique identifier for this streaming sync implementation
+    /// A good value is typically the DB file path which it will mutate when syncing.
+    String? identifier = "unknown",
+  })  : syncMutex = syncMutex ?? Mutex(identifier: "sync-$identifier"),
+        crudMutex = crudMutex ?? Mutex(identifier: "crud-$identifier"),
         _userAgentHeaders = userAgentHeaders() {
     _client = client;
     statusStream = _statusStreamController.stream;