Simplifying refreshSchema implementation by introducing exlusiveLock and refreshSchema in sqlite_database.

Author: Chriztiaan

packages/drift_sqlite_async/pubspec.yaml

@@ -15,7 +15,7 @@ environment:
   sdk: ">=3.0.0 <4.0.0"
 dependencies:
   drift: ">=2.15.0 <2.19.0"
-  sqlite_async: ^0.8.1
+  sqlite_async: ^0.8.2
 dev_dependencies:
   build_runner: ^2.4.8
   drift_dev: ">=2.15.0 <2.19.0"

packages/sqlite_async/lib/src/common/connection/sync_sqlite_connection.dart

@@ -50,11 +50,6 @@ class SyncSqliteConnection extends SqliteConnection with SqliteQueries {
   Future<bool> getAutoCommit() async {
     return db.autocommit;
   }
-
-  @override
-  Future<void> refreshSchema() async {
-    db.execute("PRAGMA table_info('sqlite_master')");
-  }
 }
 
 class SyncReadContext implements SqliteReadContext {

packages/sqlite_async/lib/src/common/sqlite_database.dart

@@ -38,6 +38,21 @@ mixin SqliteDatabaseMixin implements SqliteConnection, SqliteQueries {
   ///
   /// Use this to access the database in background isolates.
   IsolateConnectionFactory isolateConnectionFactory();
+
+  /// TODO Improve on this definition by supporting a writeable context.
+  Future<void> exclusiveLock<T>(
+    Future<T> Function(SqliteReadContext ctx) callback,
+  ) {
+    return writeLock(callback);
+  }
+
+  /// Ensures that all connections are aware of the latest schema changes applied (if any).
+  /// Queries and watch calls can potentially use outdated schema information after a schema update.
+  Future<void> refreshSchema() {
+    return exclusiveLock((ctx) async {
+      return ctx.get("PRAGMA table_info('sqlite_master')");
+    });
+  }
 }
 
 /// A SQLite database instance.

packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart

@@ -64,9 +64,4 @@ class SqliteDatabaseImpl
   Future<bool> getAutoCommit() {
     throw UnimplementedError();
   }
-
-  @override
-  Future<void> refreshSchema() {
-    throw UnimplementedError();
-  }
 }

packages/sqlite_async/lib/src/native/database/connection_pool.dart

@@ -54,6 +54,69 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
     _writeConnection?.updates?.forEach(updatesController.add);
   }
 
+  /// Executes a provided callback function exclusively across all read and
+  /// write connections in the pool.
+  ///
+  /// This function first locks all read and write connections, collecting their
+  /// contexts. It then executes the provided [callback] function on each of these
+  /// contexts. After the [callback] completex for each context, the locks are released
+  ///
+  /// Example usage:
+  /// ```dart
+  /// await runExclusive((ctx) async {
+  ///   // Perform some database operation with the ctx
+  ///   await ctx.execute('PRAGMA schema_version');
+  /// });
+  /// ```
+  ///
+  ///
+  exclusiveLock<T>(
+    Future<T> Function(SqliteReadContext tx) callback,
+  ) async {
+    final List<Completer<SqliteReadContext>> completers = [];
+    final List<Completer<void>> releasers = [];
+
+    for (final read in _allReadConnections) {
+      final completer = Completer<SqliteReadContext>();
+
+      completers.add(completer);
+      read.readLock((ctx) async {
+        completer.complete(ctx);
+
+        final releaser = Completer();
+        releasers.add(releaser);
+
+        // Keep this active, close the context when finished
+        await releaser.future;
+      });
+    }
+
+    final writeCompleter = Completer<SqliteReadContext>();
+    completers.add(writeCompleter);
+    _writeConnection?.writeLock((ctx) async {
+      writeCompleter.complete(ctx);
+
+      final releaser = Completer();
+      releasers.add(releaser);
+      await releaser.future;
+    });
+
+    // Get all the connection contexts and execute the callback on each of them
+    final List<SqliteReadContext> contexts = [];
+    for (final completer in completers) {
+      contexts.add(await completer.future);
+    }
+
+    for (final c in contexts) {
+      await callback(c);
+    }
+
+    // Release all the releasers
+    for (final r in releasers) {
+      r.complete();
+    }
+  }
+
   /// Returns true if the _write_ connection is currently in autocommit mode.
   @override
   Future<bool> getAutoCommit() async {
@@ -221,15 +284,6 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
     // read-only connections first.
     await _writeConnection?.close();
   }
-
-  @override
-  Future<void> refreshSchema() async {
-    final toRefresh = _allReadConnections.toList();
-
-    for (var connection in toRefresh) {
-      await connection.refreshSchema();
-    }
-  }
 }
 
 typedef ReadCallback<T> = Future<T> Function(SqliteReadContext tx);

packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart

@@ -158,11 +158,6 @@ class SqliteConnectionImpl
       });
     }, timeout: lockTimeout);
   }
-
-  @override
-  Future<void> refreshSchema() async {
-    await get("PRAGMA table_info('sqlite_master')");
-  }
 }
 
 int _nextCtxId = 1;

packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart

@@ -168,7 +168,8 @@ class SqliteDatabaseImpl
   }
 
   @override
-  Future<void> refreshSchema() async {
-    await _pool.refreshSchema();
+  Future<void> exclusiveLock<T>(
+      Future<T> Function(SqliteReadContext ctx) callback) {
+    return _pool.exclusiveLock(callback);
   }
 }

packages/sqlite_async/lib/src/sqlite_connection.dart

@@ -128,10 +128,6 @@ abstract class SqliteConnection extends SqliteWriteContext {
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout, String? debugContext});
 
-  /// Ensures that all connections are aware of the latest schema changes applied (if any).
-  /// Queries and watch calls can potentially use outdated schema information after a schema update.
-  Future<void> refreshSchema();
-
   Future<void> close();
 
   /// Returns true if the connection is closed

packages/sqlite_async/lib/src/web/database.dart

@@ -129,11 +129,6 @@ class WebDatabase
       }
     }
   }
-
-  @override
-  Future<void> refreshSchema() async {
-    _database.execute("PRAGMA table_info('sqlite_master')");
-  }
 }
 
 class _SharedContext implements SqliteReadContext {

packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart

@@ -139,9 +139,4 @@ class SqliteDatabaseImpl
     await isInitialized;
     return _connection.getAutoCommit();
   }
-
-  @override
-  Future<void> refreshSchema() async {
-    await _connection.refreshSchema();
-  }
 }

packages/sqlite_async/pubspec.yaml

@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.8.1
+version: 0.8.2
 repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
   sdk: ">=3.4.0 <4.0.0"

packages/sqlite_async/test/native/schema_test.dart

@@ -0,0 +1,103 @@
+@TestOn('!browser')
+import 'dart:async';
+
+import 'package:sqlite_async/sqlite_async.dart';
+import 'package:sqlite_async/src/utils/shared_utils.dart';
+import 'package:test/test.dart';
+
+import '../utils/test_utils_impl.dart';
+
+final testUtils = TestUtils();
+
+void main() {
+  group('Schema Tests', () {
+    late String path;
+
+    setUp(() async {
+      path = testUtils.dbPath();
+      await testUtils.cleanDb(path: path);
+    });
+
+    tearDown(() async {
+      await testUtils.cleanDb(path: path);
+    });
+
+    createTables(SqliteDatabase db) async {
+      await db.writeTransaction((tx) async {
+        await tx.execute(
+            'CREATE TABLE _customers(id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)');
+        await tx.execute(
+            'CREATE TABLE _local_customers(id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)');
+        await tx
+            .execute('CREATE VIEW customers AS SELECT * FROM _local_customers');
+      });
+    }
+
+    updateTables(SqliteDatabase db) async {
+      await db.writeTransaction((tx) async {
+        await tx.execute('DROP VIEW IF EXISTS customers');
+        await tx.execute('CREATE VIEW customers AS SELECT * FROM _customers');
+      });
+    }
+
+    test('should refresh schema views', () async {
+      final db = await testUtils.setupDatabase(path: path);
+      await createTables(db);
+
+      final customerTables =
+          await getSourceTables(db, "select * from customers");
+      expect(customerTables.contains('_local_customers'), true);
+      await updateTables(db);
+
+      // without this, source tables are outdated
+      await db.refreshSchema();
+
+      final updatedCustomerTables =
+          await getSourceTables(db, "select * from customers");
+      expect(updatedCustomerTables.contains('_customers'), true);
+    });
+
+    test('should complete refresh schema after transaction', () async {
+      var completer1 = Completer<void>();
+      var transactionCompleted = false;
+
+      final db = await testUtils.setupDatabase(path: path);
+      await createTables(db);
+
+      // Start a read transaction
+      db.readTransaction((tx) async {
+        completer1.complete();
+        await tx.get('select test_sleep(2000)');
+
+        transactionCompleted = true;
+      });
+
+      // Wait for the transaction to start
+      await completer1.future;
+
+      var refreshSchemaFuture = db.refreshSchema();
+
+      // Setup check that refreshSchema completes after the transaction has completed
+      var refreshAfterTransaction = false;
+      refreshSchemaFuture.then((_) {
+        if (transactionCompleted) {
+          refreshAfterTransaction = true;
+        }
+      });
+
+      await refreshSchemaFuture;
+
+      expect(refreshAfterTransaction, isTrue,
+          reason: 'refreshSchema completed before transaction finished');
+
+      // Sanity check
+      expect(transactionCompleted, isTrue,
+          reason: 'Transaction did not complete as expected');
+    });
+  });
+}
+
+// For some reason, future.ignore() doesn't actually ignore errors in these tests.
+void ignore(Future future) {
+  future.then((_) {}, onError: (_) {});
+}