Refactor locks and retry logic around connect

Author: simolus3

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

@@ -282,6 +282,8 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
   StreamSubscription<void>? localUpdatesSubscription;
 
   Future<void> shutdown() async {
+    await openedStreamingSync?.abort();
+
     localUpdatesSubscription?.cancel();
     db?.dispose();
     crudUpdateController.close();
@@ -291,9 +293,10 @@ 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 openedStreamingSync?.abort();
-
     rPort.close();
+
+    // TODO: If we closed our resources properly, this wouldn't be necessary...
+    Isolate.current.kill();
   }
 
   rPort.listen((message) async {
@@ -377,12 +380,12 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
       updateDebouncer ??=
           Timer(const Duration(milliseconds: 1), maybeFireUpdates);
     });
-  }, (error, stack) async {
+  }, (error, stack) {
     // 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.
-    await shutdown();
+    shutdown();
     throw error;
   });
 }

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

@@ -274,28 +274,27 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
         abort: thisConnectAborter,
       );
 
-      retryHandler();
+      thisConnectAborter.onCompletion.whenComplete(retryHandler);
     }
 
     // If the sync encounters a failure without being aborted, retry
-    retryHandler = () {
-      thisConnectAborter.onCompletion.then((_) async {
-        _activeGroup.syncConnectMutex.lock(() async {
-          // Still supposed to be active? (abort is only called within mutex)
-          if (!thisConnectAborter.aborted) {
-            // We only change _abortActiveSync after disconnecting, which resets
-            // the abort controller.
-            assert(identical(_abortActiveSync, thisConnectAborter));
-
-            // We need a new abort controller for this attempt
-            _abortActiveSync = thisConnectAborter = AbortController();
-
-            logger.warning('Sync client failed, retrying...');
-            await connectWithSyncLock();
-          }
-        });
+    retryHandler = Zone.current.bindCallback(() async {
+      _activeGroup.syncConnectMutex.lock(() async {
+        // Is this still supposed to be active? (abort is only called within
+        // mutex)
+        if (!thisConnectAborter.aborted) {
+          // We only change _abortActiveSync after disconnecting, which resets
+          // the abort controller.
+          assert(identical(_abortActiveSync, thisConnectAborter));
+
+          // We need a new abort controller for this attempt
+          _abortActiveSync = thisConnectAborter = AbortController();
+
+          logger.warning('Sync client failed, retrying...');
+          await connectWithSyncLock();
+        }
       });
-    };
+    });
 
     await _activeGroup.syncConnectMutex.lock(() async {
       // Disconnect a previous sync client, if one is active.

packages/powersync_core/test/schema_test.dart

@@ -176,14 +176,11 @@ void main() {
         ]),
       ]);
 
-      try {
-        powersync.updateSchema(schema2);
-      } catch (e) {
-        expect(
-            e,
-            isA<AssertionError>().having((e) => e.message, 'message',
-                'Invalid characters in table name: #notworking'));
-      }
+      await expectLater(
+        () => powersync.updateSchema(schema2),
+        throwsA(isA<AssertionError>().having((e) => e.message, 'message',
+            'Invalid characters in table name: #notworking')),
+      );
     });
   });
 

packages/powersync_core/test/streaming_sync_test.dart

@@ -5,6 +5,7 @@ library;
 import 'dart:async';
 import 'dart:math';
 
+import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
 import 'package:test/test.dart';
 
@@ -27,6 +28,39 @@ void main() {
       await testUtils.cleanDb(path: path);
     });
 
+    test('repeated connect and disconnect calls', () async {
+      final random = Random();
+      final server = await createServer();
+      final ignoreLogger = Logger.detached('powersync.test');
+
+      final pdb =
+          await testUtils.setupPowerSync(path: path, logger: ignoreLogger);
+      pdb.retryDelay = Duration(milliseconds: 5000);
+      final connector = TestConnector(() async {
+        return PowerSyncCredentials(endpoint: server.endpoint, token: 'token');
+      });
+
+      Duration nextDelay() {
+        return Duration(milliseconds: random.nextInt(100));
+      }
+
+      Future<void> connectAndDisconnect() async {
+        for (var i = 0; i < 10; i++) {
+          await Future<void>.delayed(nextDelay());
+          await pdb.connect(connector: connector);
+
+          await Future<void>.delayed(nextDelay());
+          await pdb.disconnect();
+        }
+      }
+
+      // Create a bunch of tasks calling connect and disconnect() concurrently.
+      await Future.wait([for (var i = 0; i < 10; i++) connectAndDisconnect()]);
+
+      expect(server.maxConnectionCount, lessThanOrEqualTo(1));
+      server.close();
+    });
+
     test('full powersync reconnect', () async {
       // Test repeatedly creating new PowerSync connections, then disconnect
       // and close the connection.