Fix some errors when closing client while connecting.

Author: rkistner

packages/powersync/lib/src/powersync_database.dart

@@ -191,13 +191,24 @@ class PowerSyncDatabase with SqliteQueries implements SqliteConnection {
       /// Throttle time between CRUD operations
       /// Defaults to 10 milliseconds.
       Duration crudThrottleTime = const Duration(milliseconds: 10)}) async {
-    _connectMutex.lock(() =>
-        _connect(connector: connector, crudThrottleTime: crudThrottleTime));
+    Zone current = Zone.current;
+
+    Future<void> reconnect() {
+      return _connectMutex.lock(() => _connect(
+          connector: connector,
+          crudThrottleTime: crudThrottleTime,
+          // The reconnect function needs to run in the original zone,
+          // to avoid recursive lock errors.
+          reconnect: current.bindCallback(reconnect)));
+    }
+
+    await reconnect();
   }
 
   Future<void> _connect(
       {required PowerSyncBackendConnector connector,
-      required Duration crudThrottleTime}) async {
+      required Duration crudThrottleTime,
+      required Future<void> Function() reconnect}) async {
     await initialize();
 
     // Disconnect if connected
@@ -266,7 +277,9 @@ class PowerSyncDatabase with SqliteQueries implements SqliteConnection {
       logger.severe('Sync Isolate error', message);
 
       // Reconnect
-      connect(connector: connector);
+      // Use the param like this instead of directly calling connect(), to avoid recursive
+      // locks in some edge cases.
+      reconnect();
     });
 
     disconnected() {
@@ -532,7 +545,7 @@ Future<void> _powerSyncDatabaseIsolate(
 
   CommonDatabase? db;
   final mutex = args.dbRef.mutex.open();
-  StreamingSyncImplementation? _sync;
+  StreamingSyncImplementation? openedStreamingSync;
 
   rPort.listen((message) async {
     if (message is List) {
@@ -548,7 +561,7 @@ Future<void> _powerSyncDatabaseIsolate(
         updateController.close();
         upstreamDbClient.close();
         // Abort any open http requests, and wait for it to be closed properly
-        await _sync?.abort();
+        await openedStreamingSync?.abort();
         // No kill the Isolate
         Isolate.current.kill();
       }
@@ -596,7 +609,7 @@ Future<void> _powerSyncDatabaseIsolate(
         uploadCrud: uploadCrud,
         updateStream: updateController.stream,
         retryDelay: args.retryDelay);
-    _sync = sync;
+    openedStreamingSync = sync;
     sync.streamingSync();
     sync.statusStream.listen((event) {
       sPort.send(['status', event]);

packages/powersync/lib/src/streaming_sync.dart

@@ -42,6 +42,8 @@ class StreamingSyncImplementation {
 
   AbortController? _abort;
 
+  bool _safeToClose = true;
+
   StreamingSyncImplementation(
       {required this.adapter,
       required this.credentialsCallback,
@@ -65,10 +67,17 @@ class StreamingSyncImplementation {
     // According to the documentation, the behavior is undefined when calling
     // close() while requests are pending. However, this is no other
     // known way to cancel open streams, and this appears to end the stream with
-    // a consistent ClientException.
-    _client.close();
+    // a consistent ClientException if a request is open.
+    // We avoid closing the client while opening a request, as that does cause
+    // unpredicable uncaught errors.
+    if (_safeToClose) {
+      _client.close();
+    }
     // wait for completeAbort() to be called
     await future;
+
+    // Now close the client in all cases not covered above
+    _client.close();
   }
 
   bool get aborted {
@@ -375,7 +384,18 @@ class StreamingSyncImplementation {
     request.headers['Authorization'] = "Token ${credentials.token}";
     request.body = convert.jsonEncode(data);
 
-    final res = await _client.send(request);
+    http.StreamedResponse res;
+    try {
+      // Do not close the client during the request phase - this causes uncaught errors.
+      _safeToClose = false;
+      res = await _client.send(request);
+    } finally {
+      _safeToClose = true;
+    }
+    if (aborted) {
+      return;
+    }
+
     if (res.statusCode == 401) {
       if (invalidCredentialsCallback != null) {
         await invalidCredentialsCallback!();