fix issues where Sync stream manager's mutex would not release

Author: stevensJourney

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

@@ -118,6 +118,7 @@ class PowerSyncDatabaseImpl
   }
 
   @override
+  @internal
 
   /// Connect to the PowerSync service, and keep the databases in sync.
   ///
@@ -134,10 +135,11 @@ class PowerSyncDatabaseImpl
 
     // Disconnect if connected
     await disconnect();
-    disconnecter = AbortController();
+    final disconnector = AbortController();
+    disconnecter = disconnector;
 
     await isInitialized;
-    final dbref = database.isolateConnectionFactory();
+    final dbRef = database.isolateConnectionFactory();
     ReceivePort rPort = ReceivePort();
     StreamSubscription? updateSubscription;
     rPort.listen((data) async {
@@ -161,7 +163,7 @@ class PowerSyncDatabaseImpl
           updateSubscription = throttled.listen((event) {
             port.send(['update']);
           });
-          disconnecter?.onAbort.then((_) {
+          disconnector.onAbort.then((_) {
             port.send(['close']);
           }).ignore();
         } else if (action == 'uploadCrud') {
@@ -197,11 +199,11 @@ class PowerSyncDatabaseImpl
       logger.severe('Sync Isolate error', message);
 
       // Reconnect
-      baseConnect(connector: connector);
+      connect(connector: connector, crudThrottleTime: crudThrottleTime);
     });
 
     disconnected() {
-      disconnecter?.completeAbort();
+      disconnector.completeAbort();
       disconnecter = null;
       rPort.close();
       // Clear status apart from lastSyncedAt
@@ -220,7 +222,7 @@ class PowerSyncDatabaseImpl
     }
 
     Isolate.spawn(_powerSyncDatabaseIsolate,
-        _PowerSyncDatabaseIsolateArgs(rPort.sendPort, dbref, retryDelay),
+        _PowerSyncDatabaseIsolateArgs(rPort.sendPort, dbRef, retryDelay),
         debugName: 'PowerSyncDatabase',
         onError: errorPort.sendPort,
         onExit: exitPort.sendPort);
@@ -274,12 +276,17 @@ Future<void> _powerSyncDatabaseIsolate(
   final upstreamDbClient = args.dbRef.upstreamPort.open();
 
   CommonDatabase? db;
-  rPort.listen((message) {
+  final Mutex mutex = args.dbRef.mutex.open();
+  rPort.listen((message) async {
     if (message is List) {
       String action = message[0];
       if (action == 'update') {
         updateController.add('update');
       } else if (action == '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();
         db?.dispose();
         updateController.close();
         upstreamDbClient.close();
@@ -318,7 +325,6 @@ Future<void> _powerSyncDatabaseIsolate(
   }
 
   runZonedGuarded(() async {
-    final mutex = args.dbRef.mutex.open();
     db = await args.dbRef.openFactory
         .open(SqliteOpenOptions(primaryConnection: false, readOnly: false));
     final connection = SyncSqliteConnection(db!, mutex);

packages/powersync/lib/src/database/powersync_database_impl_stub.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 
 import 'package:logging/logging.dart';
+import 'package:meta/meta.dart';
 import 'package:powersync/sqlite_async.dart';
 import 'package:powersync/src/database/powersync_db_mixin.dart';
 import 'package:powersync/src/open_factory/abstract_powersync_open_factory.dart';
@@ -109,6 +110,7 @@ class PowerSyncDatabaseImpl
   Logger get logger => throw UnimplementedError();
 
   @override
+  @internal
   Future<void> baseConnect(
       {required PowerSyncBackendConnector connector,
       Duration crudThrottleTime = const Duration(milliseconds: 10)}) {

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

@@ -137,6 +137,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// classes. This is wrapped inside an exclusive mutex in the [connect]
   /// method.
   @protected
+  @internal
   Future<void> baseConnect(
       {required PowerSyncBackendConnector connector,
 

packages/powersync/lib/src/database/web/web_powersync_database.dart

@@ -113,6 +113,7 @@ class PowerSyncDatabaseImpl
   }
 
   @override
+  @internal
 
   /// Connect to the PowerSync service, and keep the databases in sync.
   ///