Fix update notifications firing before completed transactions

packages/sqlite_async/CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.14.1
+
+- Fix update notifications sometimes firing before a write has completed.
+
 ## 0.14.0
 
 - Support versions 3.x of the `sqlite3` package and 0.7.x of `sqlite3_web`.

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

@@ -314,13 +314,24 @@ final class _LeasedContext extends UnscopedContext {
     pool._returnIsolateWorker(worker);
   }
 
-  Future<T> _runOnWorker<T>(FutureOr<T> Function(PoolConnection db) compute) {
-    return inner.unsafeAccess((connection) {
+  Future<T> _runOnWorker<T>(
+      FutureOr<T> Function(PoolConnection db) compute) async {
+    final (result, autocommit) = await inner.unsafeAccess((connection) async {
       final ptr = connection.unsafePointer.address;
       final checkInTransaction = verifyInTransaction;
 
-      return worker.run(_wrapDbClosure(ptr, checkInTransaction, compute));
+      final result =
+          await worker.run(_wrapDbClosure(ptr, checkInTransaction, compute));
+      return (result, connection.database.autocommit);
     });
+
+    if (autocommit) {
+      if (inner case final ConnectionLease lease) {
+        await lease.notifyUpdates();
+      }
+    }
+
+    return result;
   }
 
   @override

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.14.0
+version: 0.14.1
 resolution: workspace
 repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
@@ -15,7 +15,7 @@ topics:
 dependencies:
   sqlite3: ^3.2.0
   sqlite3_web: ^0.7.0
-  sqlite3_connection_pool: ^0.2.3
+  sqlite3_connection_pool: ^0.2.4
   async: ^2.10.0
   collection: ^1.17.0
   meta: ^1.10.0

packages/sqlite_async/test/native/watch_test.dart

@@ -140,6 +140,23 @@ void main() {
         lastTime = r;
       }
     });
+
+    test('receives notifications before write lock returns', () async {
+      final db = await testUtils.setupDatabase(path: path);
+      await createTables(db);
+
+      final reads = StreamQueue(db.updates);
+      final first = reads.next;
+
+      db.writeLock((ctx) async {
+        await ctx.execute('INSERT INTO customer(name) VALUES (?)', ['test']);
+        // Because we're not in a transaction, this should emit an update. We
+        // shouldn't just collect updates at the end of writeLock to avoid
+        // long-running writers never emitting updates.
+        await first;
+      });
+      reads.cancel();
+    });
   });
 }
 

packages/sqlite_async/test/watch_test.dart

@@ -41,6 +41,7 @@ void main() {
       db.updates.listen((_) => events++);
       db.updates.listen((_) => events++);
       await db.execute('INSERT INTO assets DEFAULT VALUES');
+      await pumpEventQueue();
       expect(events, 2);
     });