Refactor update notifications; avoid global locks where we can.

Author: rkistner

lib/src/connection_pool.dart

@@ -160,11 +160,8 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       return readLock((ctx) => callback(ctx as SqliteWriteContext),
           lockTimeout: lockTimeout, debugContext: debugContext);
     } else {
-      // FIXME:
-      // This should avoid using global locks, but then we need to fix
-      // update notifications to only fire after commit.
       return _writeLock(callback,
-          lockTimeout: lockTimeout, debugContext: debugContext, global: true);
+          lockTimeout: lockTimeout, debugContext: debugContext, global: false);
     }
   }
 

lib/src/sqlite_connection_impl.dart

@@ -105,6 +105,21 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
     return _connectionMutex.locked;
   }
 
+  @override
+  Future<T> lock<T>(Future<T> Function(SqliteWriteContext tx) callback,
+      {bool? readOnly, 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 {
+      final ctx = _TransactionContext(_isolateClient);
+      try {
+        return await callback(ctx);
+      } finally {
+        await ctx.close();
+      }
+    }, timeout: lockTimeout);
+  }
+
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
       {Duration? lockTimeout, String? debugContext}) async {
@@ -262,7 +277,6 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
   final server = params.portServer;
   final commandPort = ReceivePort();
 
-  Timer? updateDebouncer;
   Set<String> updatedTables = {};
   int? txId;
   Object? txError;
@@ -271,25 +285,18 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
     if (updatedTables.isNotEmpty) {
       client.fire(UpdateNotification(updatedTables));
       updatedTables.clear();
-      updateDebouncer?.cancel();
-      updateDebouncer = null;
     }
   }
 
   db.updates.listen((event) {
     updatedTables.add(event.tableName);
-
-    // This handles two cases:
-    // 1. Update arrived after _SqliteIsolateClose (not sure if this could happen).
-    // 2. Long-running _SqliteIsolateClosure that should fire updates while running.
-    updateDebouncer ??=
-        Timer(const Duration(milliseconds: 10), maybeFireUpdates);
   });
 
   server.open((data) async {
     if (data is _SqliteIsolateClose) {
       if (txId != null) {
         if (!db.autocommit) {
+          updatedTables.clear();
           db.execute('ROLLBACK');
         }
         txId = null;

lib/src/sqlite_database.dart

@@ -117,26 +117,11 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
   }
 
   void _listenForEvents() {
-    UpdateNotification? updates;
-
     Map<SendPort, StreamSubscription> subscriptions = {};
 
     _eventsPort = PortServer((message) async {
       if (message is UpdateNotification) {
-        if (updates == null) {
-          updates = message;
-          // Use the mutex to only send updates after the current transaction.
-          // Do take care to avoid getting a lock for each individual update -
-          // that could add massive performance overhead.
-          mutex.lock(() async {
-            if (updates != null) {
-              _updatesController.add(updates!);
-              updates = null;
-            }
-          });
-        } else {
-          updates!.tables.addAll(message.tables);
-        }
+        _updatesController.add(message);
         return null;
       } else if (message is InitDb) {
         await _initialized;

lib/src/sqlite_queries.dart

@@ -164,11 +164,11 @@ mixin SqliteQueries implements SqliteWriteContext, SqliteConnection {
       return readTransaction((ctx) => callback(ctx as SqliteWriteContext),
           lockTimeout: lockTimeout);
     } else {
-      // FIXME:
-      // This should avoid using global locks, but then we need to fix
-      // update notifications to only fire after commit.
-      // ignore: deprecated_member_use_from_same_package
-      return writeTransaction(callback, lockTimeout: lockTimeout);
+      // Uses connection-level lock, unlike writeTransaction which uses a
+      // database-level lock.
+      return lock((ctx) async {
+        return await internalWriteTransaction(ctx, callback);
+      }, lockTimeout: lockTimeout, debugContext: 'writeTransaction()');
     }
   }
 }