Merge pull request #4 from journeyapps/recursive-lock-errors

Improve recursive lock errors

Author: rkistner

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.3.0
+
+- Better error messages for recursive transactions.
+- Breaking change: Error by default when starting a read transaction within a write transaction.
+
 ## 0.2.1
 
 - Fix update notifications missing the first update.

README.md

@@ -23,6 +23,9 @@ query access.
  * Automatically convert query args to JSON where applicable, making JSON1 operations simple.
  * Direct SQL queries - no wrapper classes or code generation required.
 
+See this [blog post](https://www.powersync.co/blog/sqlite-optimizations-for-ultra-high-performance),
+explaining why these features are important for using SQLite.
+
 ## Installation
 
 ```sh
@@ -65,8 +68,8 @@ void main() async {
   // 1. Atomic persistence (all updates are either applied or rolled back).
   // 2. Improved throughput.
   await db.writeTransaction((tx) async {
-    await db.execute('INSERT INTO test_data(data) values(?)', ['Test3']);
-    await db.execute('INSERT INTO test_data(data) values(?)', ['Test4']);
+    await tx.execute('INSERT INTO test_data(data) values(?)', ['Test3']);
+    await tx.execute('INSERT INTO test_data(data) values(?)', ['Test4']);
   });
 
   await db.close();

example/basic_example.dart

@@ -29,8 +29,8 @@ void main() async {
   // 1. Atomic persistence (all updates are either applied or rolled back).
   // 2. Improved throughput.
   await db.writeTransaction((tx) async {
-    await db.execute('INSERT INTO test_data(data) values(?)', ['Test3']);
-    await db.execute('INSERT INTO test_data(data) values(?)', ['Test4']);
+    await tx.execute('INSERT INTO test_data(data) values(?)', ['Test3']);
+    await tx.execute('INSERT INTO test_data(data) values(?)', ['Test4']);
   });
 
   // Close database to release resources

lib/src/connection_pool.dart

@@ -50,58 +50,60 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
-      {Duration? lockTimeout}) async {
+      {Duration? lockTimeout, String? debugContext}) async {
     await _expandPool();
 
-    bool haveLock = false;
-    var completer = Completer<T>();
+    return _runZoned(() async {
+      bool haveLock = false;
+      var completer = Completer<T>();
+
+      var futures = _readConnections.sublist(0).map((connection) async {
+        try {
+          return await connection.readLock((ctx) async {
+            if (haveLock) {
+              // Already have a different lock - release this one.
+              return false;
+            }
+            haveLock = true;
+
+            var future = callback(ctx);
+            completer.complete(future);
+
+            // We have to wait for the future to complete before we can release the
+            // lock.
+            try {
+              await future;
+            } catch (_) {
+              // Ignore
+            }
+
+            return true;
+          }, lockTimeout: lockTimeout, debugContext: debugContext);
+        } on TimeoutException {
+          return false;
+        }
+      });
+
+      final stream = Stream<bool>.fromFutures(futures);
+      var gotAny = await stream.any((element) => element);
+
+      if (!gotAny) {
+        // All TimeoutExceptions
+        throw TimeoutException('Failed to get a read connection', lockTimeout);
+      }
 
-    var futures = _readConnections.sublist(0).map((connection) async {
       try {
-        return await connection.readLock((ctx) async {
-          if (haveLock) {
-            // Already have a different lock - release this one.
-            return false;
-          }
-          haveLock = true;
-
-          var future = callback(ctx);
-          completer.complete(future);
-
-          // We have to wait for the future to complete before we can release the
-          // lock.
-          try {
-            await future;
-          } catch (_) {
-            // Ignore
-          }
-
-          return true;
-        }, lockTimeout: lockTimeout);
-      } on TimeoutException {
-        return false;
+        return await completer.future;
+      } catch (e) {
+        // throw e;
+        rethrow;
       }
-    });
-
-    final stream = Stream<bool>.fromFutures(futures);
-    var gotAny = await stream.any((element) => element);
-
-    if (!gotAny) {
-      // All TimeoutExceptions
-      throw TimeoutException('Failed to get a read connection', lockTimeout);
-    }
-
-    try {
-      return await completer.future;
-    } catch (e) {
-      // throw e;
-      rethrow;
-    }
+    }, debugContext: debugContext ?? 'get*()');
   }
 
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
-      {Duration? lockTimeout}) {
+      {Duration? lockTimeout, String? debugContext}) {
     if (closed) {
       throw AssertionError('Closed');
     }
@@ -113,7 +115,25 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
         mutex: mutex,
         readOnly: false,
         openFactory: _factory);
-    return _writeConnection!.writeLock(callback, lockTimeout: lockTimeout);
+    return _runZoned(() {
+      return _writeConnection!.writeLock(callback,
+          lockTimeout: lockTimeout, debugContext: debugContext);
+    }, debugContext: debugContext ?? 'execute()');
+  }
+
+  /// The [Mutex] on individual connections do already error in recursive locks.
+  ///
+  /// We duplicate the same check here, to:
+  /// 1. Also error when the recursive transaction is handled by a different
+  ///    connection (with a different lock).
+  /// 2. Give a more specific error message when it happens.
+  T _runZoned<T>(T Function() callback, {required String debugContext}) {
+    if (Zone.current[this] != null) {
+      throw LockError(
+          'Recursive lock is not allowed. Use `tx.$debugContext` instead of `db.$debugContext`.');
+    }
+    var zone = Zone.current.fork(zoneValues: {this: true});
+    return zone.run(callback);
   }
 
   Future<void> _expandPool() async {

lib/src/mutex.dart

@@ -40,7 +40,7 @@ class SimpleMutex implements Mutex {
   @override
   Future<T> lock<T>(Future<T> Function() callback, {Duration? timeout}) async {
     if (Zone.current[this] != null) {
-      throw AssertionError('Recursive lock is not allowed');
+      throw LockError('Recursive lock is not allowed');
     }
     var zone = Zone.current.fork(zoneValues: {this: true});
 
@@ -132,7 +132,7 @@ class SharedMutex implements Mutex {
   @override
   Future<T> lock<T>(Future<T> Function() callback, {Duration? timeout}) async {
     if (Zone.current[this] != null) {
-      throw AssertionError('Recursive lock is not allowed');
+      throw LockError('Recursive lock is not allowed');
     }
     return runZoned(() async {
       await _acquire(timeout: timeout);
@@ -223,3 +223,14 @@ class _AcquireMessage {
 class _UnlockMessage {
   const _UnlockMessage();
 }
+
+class LockError extends Error {
+  final String message;
+
+  LockError(this.message);
+
+  @override
+  String toString() {
+    return 'LockError: $message';
+  }
+}

lib/src/sqlite_connection.dart

@@ -94,13 +94,13 @@ abstract class SqliteConnection extends SqliteWriteContext {
   ///
   /// In most cases, [readTransaction] should be used instead.
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
-      {Duration? lockTimeout});
+      {Duration? lockTimeout, String? debugContext});
 
   /// Takes a global lock, without starting a transaction.
   ///
   /// In most cases, [writeTransaction] should be used instead.
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
-      {Duration? lockTimeout});
+      {Duration? lockTimeout, String? debugContext});
 
   Future<void> close();
 }

lib/src/sqlite_connection_impl.dart

@@ -79,7 +79,7 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
 
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
-      {Duration? lockTimeout}) async {
+      {Duration? lockTimeout, String? debugContext}) async {
     // Private lock to synchronize this with other statements on the same connection,
     // to ensure that transactions aren't interleaved.
     return _connectionMutex.lock(() async {
@@ -94,7 +94,7 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
 
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
-      {Duration? lockTimeout}) async {
+      {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.

lib/src/sqlite_database.dart

@@ -209,13 +209,15 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
 
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
-      {Duration? lockTimeout}) {
-    return _pool.readLock(callback, lockTimeout: lockTimeout);
+      {Duration? lockTimeout, String? debugContext}) {
+    return _pool.readLock(callback,
+        lockTimeout: lockTimeout, debugContext: debugContext);
   }
 
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
-      {Duration? lockTimeout}) {
-    return _pool.writeLock(callback, lockTimeout: lockTimeout);
+      {Duration? lockTimeout, String? debugContext}) {
+    return _pool.writeLock(callback,
+        lockTimeout: lockTimeout, debugContext: debugContext);
   }
 }

lib/src/sqlite_queries.dart

@@ -17,30 +17,30 @@ mixin SqliteQueries implements SqliteWriteContext, SqliteConnection {
       [List<Object?> parameters = const []]) async {
     return writeLock((ctx) async {
       return ctx.execute(sql, parameters);
-    });
+    }, debugContext: 'execute()');
   }
 
   @override
   Future<sqlite.ResultSet> getAll(String sql,
       [List<Object?> parameters = const []]) {
     return readLock((ctx) async {
       return ctx.getAll(sql, parameters);
-    });
+    }, debugContext: 'getAll()');
   }
 
   @override
   Future<sqlite.Row> get(String sql, [List<Object?> parameters = const []]) {
     return readLock((ctx) async {
       return ctx.get(sql, parameters);
-    });
+    }, debugContext: 'get()');
   }
 
   @override
   Future<sqlite.Row?> getOptional(String sql,
       [List<Object?> parameters = const []]) {
     return readLock((ctx) async {
       return ctx.getOptional(sql, parameters);
-    });
+    }, debugContext: 'getOptional()');
   }
 
   @override
@@ -103,7 +103,7 @@ mixin SqliteQueries implements SqliteWriteContext, SqliteConnection {
       {Duration? lockTimeout}) async {
     return readLock((ctx) async {
       return await internalReadTransaction(ctx, callback);
-    }, lockTimeout: lockTimeout);
+    }, lockTimeout: lockTimeout, debugContext: 'readTransaction()');
   }
 
   @override
@@ -112,7 +112,7 @@ mixin SqliteQueries implements SqliteWriteContext, SqliteConnection {
       {Duration? lockTimeout}) async {
     return writeLock((ctx) async {
       return await internalWriteTransaction(ctx, callback);
-    }, lockTimeout: lockTimeout);
+    }, lockTimeout: lockTimeout, debugContext: 'writeTransaction()');
   }
 
   /// See [SqliteReadContext.computeWithDatabase].

pubspec.yaml

@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.2.1
+version: 0.3.0
 repository: https://github.com/journeyapps/sqlite_async.dart
 environment:
   sdk: '>=2.19.1 <3.0.0'