Use busy-timeout for database-level locks.

Author: rkistner

lib/src/sqlite_connection_impl.dart

@@ -123,30 +123,15 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout, String? debugContext}) async {
-    final stopWatch = lockTimeout == null ? null : (Stopwatch()..start());
     // Private lock to synchronize this with other statements on the same connection,
     // to ensure that transactions aren't interleaved.
     return await _connectionMutex.lock(() async {
-      Duration? innerTimeout;
-      if (lockTimeout != null && stopWatch != null) {
-        innerTimeout = lockTimeout - stopWatch.elapsed;
-        stopWatch.stop();
+      final ctx = _TransactionContext(_isolateClient);
+      try {
+        return await callback(ctx);
+      } finally {
+        await ctx.close();
       }
-      // DB lock so that only one write happens at a time
-      return await _writeMutex.lock(() async {
-        final ctx = _TransactionContext(_isolateClient);
-        try {
-          return await callback(ctx);
-        } finally {
-          await ctx.close();
-        }
-      }, timeout: innerTimeout).catchError((error, stackTrace) {
-        if (error is TimeoutException) {
-          return Future<T>.error(TimeoutException(
-              'Failed to acquire global write lock', lockTimeout));
-        }
-        return Future<T>.error(error, stackTrace);
-      });
     }, timeout: lockTimeout);
   }
 }

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,11 @@ class DefaultSqliteOpenFactory implements SqliteOpenFactory {
   List<String> pragmaStatements(SqliteOpenOptions options) {
     List<String> statements = [];
 
+    if (sqliteOptions.busyTimeout != null) {
+      statements.add(
+          'PRAGMA busy_timeout = ${sqliteOptions.busyTimeout!.inMilliseconds}');
+    }
+
     if (options.primaryConnection && sqliteOptions.journalMode != null) {
       // Persisted - only needed on the primary connection
       statements
@@ -51,8 +56,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 write connections.
+  /// Defaults to 30 seconds.
+  /// Set to 0 to fail immediately when the database is locked.
+  final Duration? busyTimeout;
+
   const SqliteOptions.defaults()
       : journalMode = SqliteJournalMode.wal,
         journalSizeLimit = 6 * 1024 * 1024, // 1.5x the default checkpoint size
-        synchronous = SqliteSynchronous.normal;
+        synchronous = SqliteSynchronous.normal,
+        busyTimeout = const Duration(seconds: 30);
 
   const SqliteOptions(
       {this.journalMode = SqliteJournalMode.wal,
       this.journalSizeLimit = 6 * 1024 * 1024,
-      this.synchronous = SqliteSynchronous.normal});
+      this.synchronous = SqliteSynchronous.normal,
+      this.busyTimeout = 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);