Merge pull request #266 from powersync-ja/warn-on-duplicate-database

Warn on duplicate database instances

Author: simolus3

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

@@ -0,0 +1,40 @@
+import 'package:meta/meta.dart';
+import 'package:sqlite_async/sqlite_async.dart';
+
+/// A collection of PowerSync database instances that are using the same
+/// underlying SQLite database.
+///
+/// We expect that each group will only ever have one database because we
+/// encourage users to manage their databases as singletons. So, we print a
+/// warning when two databases are part of the same group.
+///
+/// This can only detect two database instances being opened on the same
+/// isolate, we can't provide these checks acros isolates. Since most users
+/// aren't opening databases on background isolates though, this still guards
+/// against most misuses.
+@internal
+final class ActiveDatabaseGroup {
+  int refCount = 0;
+
+  /// Use to prevent multiple connections from being opened concurrently
+  final Mutex syncConnectMutex = Mutex();
+  final String identifier;
+
+  ActiveDatabaseGroup._(this.identifier);
+
+  void close() {
+    if (--refCount == 0) {
+      final removedGroup = _activeGroups.remove(identifier);
+      assert(removedGroup == this);
+    }
+  }
+
+  static final Map<String, ActiveDatabaseGroup> _activeGroups = {};
+
+  static ActiveDatabaseGroup referenceDatabase(String identifier) {
+    final group = _activeGroups.putIfAbsent(
+        identifier, () => ActiveDatabaseGroup._(identifier));
+    group.refCount++;
+    return group;
+  }
+}

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

@@ -73,6 +73,6 @@ abstract class PowerSyncDatabase
       required SqliteDatabase database,
       Logger? loggers}) {
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, database: database);
+        schema: schema, database: database, logger: loggers);
   }
 }

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

@@ -80,7 +80,7 @@ class PowerSyncDatabaseImpl
   factory PowerSyncDatabaseImpl.withDatabase(
       {required Schema schema,
       required SqliteDatabase database,
-      Logger? loggers}) {
+      Logger? logger}) {
     throw UnimplementedError();
   }
 

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

@@ -7,6 +7,7 @@ import 'package:powersync_core/sqlite_async.dart';
 import 'package:powersync_core/src/abort_controller.dart';
 import 'package:powersync_core/src/connector.dart';
 import 'package:powersync_core/src/crud.dart';
+import 'package:powersync_core/src/database/active_instances.dart';
 import 'package:powersync_core/src/database/core_version.dart';
 import 'package:powersync_core/src/powersync_update_notification.dart';
 import 'package:powersync_core/src/schema.dart';
@@ -45,8 +46,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   StreamController<SyncStatus> statusStreamController =
       StreamController<SyncStatus>.broadcast();
 
-  /// Use to prevent multiple connections from being opened concurrently
-  final Mutex _connectMutex = Mutex();
+  late final ActiveDatabaseGroup _activeGroup;
 
   @override
 
@@ -71,6 +71,22 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
 
   @protected
   Future<void> baseInit() async {
+    String identifier = 'memory';
+    try {
+      identifier = database.openFactory.path;
+    } catch (ignore) {
+      // The in-memory database used in some tests doesn't have an open factory.
+    }
+
+    _activeGroup = ActiveDatabaseGroup.referenceDatabase(identifier);
+    if (_activeGroup.refCount > 1) {
+      logger.warning(
+        'Multiple instances for the same database have been detected. '
+        'This can cause unexpected results, please check your PowerSync client '
+        'instantiation logic if this is not intentional',
+      );
+    }
+
     statusStream = statusStreamController.stream;
     updates = powerSyncUpdateNotifications(database.updates);
 
@@ -209,12 +225,16 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     await isInitialized;
     // Disconnect any active sync connection.
     await disconnect();
-    // Now we can close the database
-    await database.close();
 
-    // If there are paused subscriptionso n the status stream, don't delay
-    // closing the database because of that.
-    unawaited(statusStreamController.close());
+    if (!database.closed) {
+      // Now we can close the database
+      await database.close();
+
+      // If there are paused subscriptionso n the status stream, don't delay
+      // closing the database because of that.
+      unawaited(statusStreamController.close());
+      _activeGroup.close();
+    }
   }
 
   /// Connect to the PowerSync service, and keep the databases in sync.
@@ -233,7 +253,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     Zone current = Zone.current;
 
     Future<void> reconnect() {
-      return _connectMutex.lock(() => baseConnect(
+      return _activeGroup.syncConnectMutex.lock(() => baseConnect(
           connector: connector,
           crudThrottleTime: crudThrottleTime,
           // The reconnect function needs to run in the original zone,

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

@@ -37,8 +37,6 @@ class PowerSyncDatabaseImpl
   @override
   SqliteDatabase database;
 
-  late final DefaultSqliteOpenFactory openFactory;
-
   @override
   @protected
   late Future<void> isInitialized;
@@ -95,8 +93,7 @@ class PowerSyncDatabaseImpl
       Logger? logger}) {
     final db = SqliteDatabase.withFactory(openFactory, maxReaders: 1);
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, logger: logger, database: db)
-      ..openFactory = openFactory;
+        schema: schema, logger: logger, database: db);
   }
 
   /// Open a PowerSyncDatabase on an existing [SqliteDatabase].

packages/powersync_core/lib/src/web/sync_controller.dart

@@ -115,6 +115,6 @@ class SyncWorkerHandle implements StreamingSync {
   @override
   Future<void> streamingSync() async {
     await _channel.startSynchronization(
-        database.openFactory.path, crudThrottleTimeMs, syncParams);
+        database.database.openFactory.path, crudThrottleTimeMs, syncParams);
   }
 }

packages/powersync_core/test/in_memory_sync_test.dart

@@ -15,6 +15,8 @@ import 'utils/in_memory_http.dart';
 import 'utils/test_utils_impl.dart';
 
 void main() {
+  final ignoredLogger = Logger.detached('powersync.test')..level = Level.OFF;
+
   group('in-memory sync tests', () {
     late final testUtils = TestUtils();
 
@@ -100,7 +102,8 @@ void main() {
       await expectLater(
           status, emits(isSyncStatus(downloading: false, hasSynced: true)));
 
-      final independentDb = factory.wrapRaw(raw);
+      final independentDb = factory.wrapRaw(raw, logger: ignoredLogger);
+      addTearDown(independentDb.close);
       // Even though this database doesn't have a sync client attached to it,
       // is should reconstruct hasSynced from the database.
       await independentDb.initialize();
@@ -272,7 +275,8 @@ void main() {
         await database.waitForFirstSync(priority: BucketPriority(1));
         expect(database.currentStatus.hasSynced, isFalse);
 
-        final independentDb = factory.wrapRaw(raw);
+        final independentDb = factory.wrapRaw(raw, logger: ignoredLogger);
+        addTearDown(independentDb.close);
         await independentDb.initialize();
         expect(independentDb.currentStatus.hasSynced, isFalse);
         // Completing a sync for prio 1 implies a completed sync for prio 0

packages/powersync_core/test/powersync_shared_test.dart

@@ -1,3 +1,4 @@
+import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
 import 'package:sqlite_async/mutex.dart';
 import 'package:test/test.dart';
@@ -20,6 +21,33 @@ void main() {
       await testUtils.cleanDb(path: path);
     });
 
+    test('warns on duplicate database', () async {
+      final logger = Logger.detached('powersync.test')..level = Level.WARNING;
+      final events = <LogRecord>[];
+      final subscription = logger.onRecord.listen(events.add);
+      addTearDown(subscription.cancel);
+
+      final firstInstance =
+          await testUtils.setupPowerSync(path: path, logger: logger);
+      await firstInstance.initialize();
+      expect(events, isEmpty);
+
+      final secondInstance =
+          await testUtils.setupPowerSync(path: path, logger: logger);
+      await secondInstance.initialize();
+      expect(
+        events,
+        contains(
+          isA<LogRecord>().having(
+            (e) => e.message,
+            'message',
+            contains(
+                'Multiple instances for the same database have been detected.'),
+          ),
+        ),
+      );
+    });
+
     test('should not allow direct db calls within a transaction callback',
         () async {
       final db = await testUtils.setupPowerSync(path: path);

packages/powersync_core/test/utils/abstract_test_utils.dart

@@ -5,6 +5,7 @@ import 'package:powersync_core/src/bucket_storage.dart';
 import 'package:powersync_core/src/streaming_sync.dart';
 import 'package:sqlite_async/sqlite3_common.dart';
 import 'package:sqlite_async/sqlite_async.dart';
+import 'package:test/test.dart';
 import 'package:test_api/src/backend/invoker.dart';
 
 const schema = Schema([
@@ -63,11 +64,15 @@ abstract mixin class TestPowerSyncFactory implements PowerSyncOpenFactory {
     return (raw, wrapRaw(raw));
   }
 
-  PowerSyncDatabase wrapRaw(CommonDatabase raw) {
+  PowerSyncDatabase wrapRaw(
+    CommonDatabase raw, {
+    Logger? logger,
+  }) {
     return PowerSyncDatabase.withDatabase(
       schema: schema,
       database: SqliteDatabase.singleConnection(
           SqliteConnection.synchronousWrapper(raw)),
+      loggers: logger,
     );
   }
 }
@@ -95,6 +100,7 @@ abstract class AbstractTestUtils {
         schema: schema ?? defaultSchema,
         logger: logger ?? _makeTestLogger(name: _testName));
     await db.initialize();
+    addTearDown(db.close);
     return db;
   }
 

packages/powersync_core/test/utils/native_test_utils.dart

@@ -19,7 +19,8 @@ class TestOpenFactory extends PowerSyncOpenFactory with TestPowerSyncFactory {
       return DynamicLibrary.open('libsqlite3.so.0');
     });
     sqlite_open.open.overrideFor(sqlite_open.OperatingSystem.macOS, () {
-      return DynamicLibrary.open('libsqlite3.dylib');
+      return DynamicLibrary.open(
+          '/opt/homebrew/opt/sqlite/lib/libsqlite3.dylib');
     });
   }
 

packages/powersync_core/test/utils/web_test_utils.dart

@@ -78,7 +78,7 @@ class TestUtils extends AbstractTestUtils {
   Future<PowerSyncDatabase> setupPowerSync(
       {String? path, Schema? schema, Logger? logger}) async {
     await _isInitialized;
-    return super.setupPowerSync(path: path, schema: schema);
+    return super.setupPowerSync(path: path, schema: schema, logger: logger);
   }
 
   @override