Avoid zone workarounds

Author: simolus3

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

@@ -120,6 +120,7 @@ class PowerSyncDatabaseImpl
     required PowerSyncBackendConnector connector,
     required Duration crudThrottleTime,
     required AbortController abort,
+    required Zone asyncWorkZone,
     Map<String, dynamic>? params,
   }) async {
     final dbRef = database.isolateConnectionFactory();
@@ -157,7 +158,7 @@ class PowerSyncDatabaseImpl
       await waitForShutdown();
     }
 
-    receiveMessages.listen((data) async {
+    Future<void> handleMessage(Object? data) async {
       if (data is List) {
         String action = data[0] as String;
         if (action == "getCredentials") {
@@ -192,7 +193,14 @@ class PowerSyncDatabaseImpl
               record.level, record.message, record.error, record.stackTrace);
         }
       }
-    });
+    }
+
+    // This function is called in a Zone marking the connection lock as locked.
+    // This is used to prevent reentrant calls to the lock (which would be a
+    // deadlock). However, the lock is returned as soon as this function
+    // returns - and handleMessage may run later. So, make sure we run those
+    // callbacks in the parent zone.
+    receiveMessages.listen(asyncWorkZone.bindUnaryCallback(handleMessage));
 
     receiveUnhandledErrors.listen((message) async {
       // Sample error:

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

@@ -114,6 +114,7 @@ class PowerSyncDatabaseImpl
       {required PowerSyncBackendConnector connector,
       required Duration crudThrottleTime,
       required AbortController abort,
+      required Zone asyncWorkZone,
       Map<String, dynamic>? params}) {
     throw UnimplementedError();
   }

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

@@ -282,6 +282,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
 
     clientParams = params;
     var thisConnectAborter = AbortController();
+    final zone = Zone.current;
 
     late void Function() retryHandler;
 
@@ -296,6 +297,9 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
         crudThrottleTime: crudThrottleTime,
         params: params,
         abort: thisConnectAborter,
+        // Run follow-up async tasks in the parent zone, a new one is introduced
+        // while we hold the lock (and async tasks won't hold the sync lock).
+        asyncWorkZone: zone,
       );
 
       thisConnectAborter.onCompletion.whenComplete(retryHandler);
@@ -347,6 +351,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     required PowerSyncBackendConnector connector,
     required Duration crudThrottleTime,
     required AbortController abort,
+    required Zone asyncWorkZone,
     Map<String, dynamic>? params,
   });
 

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

@@ -116,6 +116,7 @@ class PowerSyncDatabaseImpl
     required PowerSyncBackendConnector connector,
     required Duration crudThrottleTime,
     required AbortController abort,
+    required Zone asyncWorkZone,
     Map<String, dynamic>? params,
   }) async {
     final crudStream =

packages/powersync_core/test/streaming_sync_test.dart

@@ -84,16 +84,21 @@ void main() {
       final didDisconnect = Completer<void>();
 
       connector.fetchCredentialsCallback = expectAsync0(() async {
-        Zone.current.parent!.scheduleMicrotask(() {
-          didDisconnect.complete(pdb.disconnect());
-        });
+        didDisconnect.complete(pdb.disconnect());
 
         throw 'deliberate disconnect';
       });
 
       service.addKeepAlive(0);
       await didDisconnect.future;
       expect(pdb.currentStatus.connected, isFalse);
+      // The error should be cleared after calling disconnect
+      expect(pdb.currentStatus.downloadError, isNull);
+
+      // Wait for a short while to make sure the database doesn't reconnect.
+      for (var i = 0; i < 10; i++) {
+        expect(pdb.currentStatus.connecting, isFalse);
+      }
     });
 
     test('can connect as initial operation', () async {