Fix deadlock on immediate connect call

Author: simolus3

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

@@ -122,7 +122,6 @@ class PowerSyncDatabaseImpl
     required AbortController abort,
     Map<String, dynamic>? params,
   }) async {
-    await initialize();
     final dbRef = database.isolateConnectionFactory();
 
     bool triedSpawningIsolate = false;

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

@@ -264,6 +264,11 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     Duration crudThrottleTime = const Duration(milliseconds: 10),
     Map<String, dynamic>? params,
   }) async {
+    // The initialization process acquires a sync connect lock (through
+    // updateSchema), so ensure the database is ready before we try to acquire
+    // the lock for the connection.
+    await initialize();
+
     clientParams = params;
     var thisConnectAborter = AbortController();
 

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

@@ -118,8 +118,6 @@ class PowerSyncDatabaseImpl
     required AbortController abort,
     Map<String, dynamic>? params,
   }) async {
-    await initialize();
-
     final crudStream =
         database.onChange(['ps_crud'], throttle: crudThrottleTime);
 

packages/powersync_core/test/streaming_sync_test.dart

@@ -61,6 +61,25 @@ void main() {
       server.close();
     });
 
+    test('can connect as initial operation', () async {
+      final server = await createServer();
+      final ignoreLogger = Logger.detached('powersync.test');
+
+      final pdb = await testUtils.setupPowerSync(
+          path: path, logger: ignoreLogger, initialize: false);
+      pdb.retryDelay = Duration(milliseconds: 5000);
+
+      await pdb.connect(connector: TestConnector(() async {
+        return PowerSyncCredentials(endpoint: server.endpoint, token: 'token');
+      }));
+
+      await expectLater(
+        pdb.statusStream,
+        emitsThrough(
+            isA<SyncStatus>().having((e) => e.connected, 'connected', isTrue)),
+      );
+    });
+
     test('full powersync reconnect', () async {
       // Test repeatedly creating new PowerSync connections, then disconnect
       // and close the connection.

packages/powersync_core/test/utils/abstract_test_utils.dart

@@ -94,12 +94,18 @@ abstract class AbstractTestUtils {
       SqliteOptions options = const SqliteOptions.defaults()});
 
   /// Creates a SqliteDatabaseConnection
-  Future<PowerSyncDatabase> setupPowerSync(
-      {String? path, Schema? schema, Logger? logger}) async {
+  Future<PowerSyncDatabase> setupPowerSync({
+    String? path,
+    Schema? schema,
+    Logger? logger,
+    bool initialize = true,
+  }) async {
     final db = PowerSyncDatabase.withFactory(await testFactory(path: path),
         schema: schema ?? defaultSchema,
         logger: logger ?? _makeTestLogger(name: _testName));
-    await db.initialize();
+    if (initialize) {
+      await db.initialize();
+    }
     addTearDown(db.close);
     return db;
   }

packages/powersync_core/test/utils/web_test_utils.dart

@@ -76,9 +76,13 @@ class TestUtils extends AbstractTestUtils {
 
   @override
   Future<PowerSyncDatabase> setupPowerSync(
-      {String? path, Schema? schema, Logger? logger}) async {
+      {String? path,
+      Schema? schema,
+      Logger? logger,
+      bool initialize = true}) async {
     await _isInitialized;
-    return super.setupPowerSync(path: path, schema: schema, logger: logger);
+    return super.setupPowerSync(
+        path: path, schema: schema, logger: logger, initialize: initialize);
   }
 
   @override