Merge pull request #34 from powersync-ja/multi-isolates

Use busy_timeout for better lock handling across Isolates

Author: rkistner

.github/workflows/test.yaml

@@ -32,19 +32,19 @@ jobs:
         include:
           - sqlite_version: "3440200"
             sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3440200.tar.gz"
-            dart_sdk: 3.2.4
+            dart_sdk: 3.3.3
           - sqlite_version: "3430200"
             sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3430200.tar.gz"
-            dart_sdk: 3.2.4
+            dart_sdk: 3.3.3
           - sqlite_version: "3420000"
             sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3420000.tar.gz"
-            dart_sdk: 3.2.4
+            dart_sdk: 3.3.3
           - sqlite_version: "3410100"
             sqlite_url: "https://www.sqlite.org/2023/sqlite-autoconf-3410100.tar.gz"
-            dart_sdk: 3.2.4
+            dart_sdk: 3.3.3
           - sqlite_version: "3380000"
             sqlite_url: "https://www.sqlite.org/2022/sqlite-autoconf-3380000.tar.gz"
-            dart_sdk: 3.2.0
+            dart_sdk: 3.3.3
     steps:
       - uses: actions/checkout@v3
       - uses: dart-lang/setup-dart@v1

lib/src/sqlite_connection.dart

@@ -84,7 +84,8 @@ abstract class SqliteConnection extends SqliteWriteContext {
   /// Open a read-write transaction.
   ///
   /// This takes a global lock - only one write transaction can execute against
-  /// the database at a time.
+  /// the database at a time. This applies even when constructing separate
+  /// [SqliteDatabase] instances for the same database file.
   ///
   /// Statements within the transaction must be done on the provided
   /// [SqliteWriteContext] - attempting statements on the [SqliteConnection]
@@ -104,13 +105,20 @@ abstract class SqliteConnection extends SqliteWriteContext {
 
   /// Takes a read lock, without starting a transaction.
   ///
+  /// The lock only applies to a single [SqliteConnection], and multiple
+  /// connections may hold read locks at the same time.
+  ///
   /// In most cases, [readTransaction] should be used instead.
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
       {Duration? lockTimeout, String? debugContext});
 
   /// Takes a global lock, without starting a transaction.
   ///
   /// In most cases, [writeTransaction] should be used instead.
+  ///
+  /// The lock applies to all [SqliteConnection] instances for a [SqliteDatabase].
+  /// Locks for separate [SqliteDatabase] instances on the same database file
+  /// may be held concurrently.
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout, String? debugContext});
 

lib/src/sqlite_database.dart

@@ -15,8 +15,10 @@ import 'update_notification.dart';
 
 /// A SQLite database instance.
 ///
-/// Use one instance per database file. If multiple instances are used, update
-/// notifications may not trigger, and calls may fail with "SQLITE_BUSY" errors.
+/// Use one instance per database file where feasible.
+///
+/// If multiple instances are used, update notifications will not be propagated between them.
+/// For update notifications across isolates, use [isolateConnectionFactory].
 class SqliteDatabase with SqliteQueries implements SqliteConnection {
   /// The maximum number of concurrent read transactions if not explicitly specified.
   static const int defaultMaxReaders = 5;

lib/src/sqlite_open_factory.dart

@@ -1,6 +1,6 @@
 import 'dart:async';
 
-import 'package:sqlite3/sqlite3.dart' as sqlite;
+import 'package:sqlite_async/sqlite3.dart' as sqlite;
 
 import 'sqlite_options.dart';
 
@@ -29,6 +29,12 @@ class DefaultSqliteOpenFactory implements SqliteOpenFactory {
   List<String> pragmaStatements(SqliteOpenOptions options) {
     List<String> statements = [];
 
+    if (sqliteOptions.lockTimeout != null) {
+      // May be replaced by a Dart-level retry mechanism in the future
+      statements.add(
+          'PRAGMA busy_timeout = ${sqliteOptions.lockTimeout!.inMilliseconds}');
+    }
+
     if (options.primaryConnection && sqliteOptions.journalMode != null) {
       // Persisted - only needed on the primary connection
       statements
@@ -51,8 +57,21 @@ class DefaultSqliteOpenFactory implements SqliteOpenFactory {
     final mode = options.openMode;
     var db = sqlite.sqlite3.open(path, mode: mode, mutex: false);
 
+    // Pragma statements don't have the same BUSY_TIMEOUT behavior as normal statements.
+    // We add a manual retry loop for those.
     for (var statement in pragmaStatements(options)) {
-      db.execute(statement);
+      for (var tries = 0; tries < 30; tries++) {
+        try {
+          db.execute(statement);
+          break;
+        } on sqlite.SqliteException catch (e) {
+          if (e.resultCode == sqlite.SqlError.SQLITE_BUSY && tries < 29) {
+            continue;
+          } else {
+            rethrow;
+          }
+        }
+      }
     }
     return db;
   }

lib/src/sqlite_options.dart

@@ -11,15 +11,22 @@ class SqliteOptions {
   /// attempt to truncate the file afterwards.
   final int? journalSizeLimit;
 
+  /// Timeout waiting for locks to be released by other connections.
+  /// Defaults to 30 seconds.
+  /// Set to null or [Duration.zero] to fail immediately when the database is locked.
+  final Duration? lockTimeout;
+
   const SqliteOptions.defaults()
       : journalMode = SqliteJournalMode.wal,
         journalSizeLimit = 6 * 1024 * 1024, // 1.5x the default checkpoint size
-        synchronous = SqliteSynchronous.normal;
+        synchronous = SqliteSynchronous.normal,
+        lockTimeout = const Duration(seconds: 30);
 
   const SqliteOptions(
       {this.journalMode = SqliteJournalMode.wal,
       this.journalSizeLimit = 6 * 1024 * 1024,
-      this.synchronous = SqliteSynchronous.normal});
+      this.synchronous = SqliteSynchronous.normal,
+      this.lockTimeout = const Duration(seconds: 30)});
 }
 
 /// SQLite journal mode. Set on the primary connection.

test/basic_test.dart

@@ -64,6 +64,39 @@ void main() {
       }
     });
 
+    test('Concurrency 2', () async {
+      final db1 =
+          SqliteDatabase.withFactory(testFactory(path: path), maxReaders: 3);
+
+      final db2 =
+          SqliteDatabase.withFactory(testFactory(path: path), maxReaders: 3);
+      await db1.initialize();
+      await createTables(db1);
+      await db2.initialize();
+      print("${DateTime.now()} start");
+
+      var futures1 = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11].map((i) {
+        return db1.execute(
+            "INSERT OR REPLACE INTO test_data(id, description) SELECT ? as i, test_sleep(?) || ' ' || test_connection_name() || ' 1 ' || datetime() as connection RETURNING *",
+            [
+              i,
+              5 + Random().nextInt(20)
+            ]).then((value) => print("${DateTime.now()} $value"));
+      }).toList();
+
+      var futures2 = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11].map((i) {
+        return db2.execute(
+            "INSERT OR REPLACE INTO test_data(id, description) SELECT ? as i, test_sleep(?) || ' ' || test_connection_name() || ' 2 ' || datetime() as connection RETURNING *",
+            [
+              i,
+              5 + Random().nextInt(20)
+            ]).then((value) => print("${DateTime.now()} $value"));
+      }).toList();
+      await Future.wait(futures1);
+      await Future.wait(futures2);
+      print("${DateTime.now()} done");
+    });
+
     test('read-only transactions', () async {
       final db = await setupDatabase(path: path);
       await createTables(db);