Fetch stream if outstanding requests are cancelled

Closes #2744

drift/CHANGELOG.md

@@ -3,6 +3,11 @@
 - Add the `QueryInterceptor` to easily monitor all database calls made by drift.
 - Add the `count()` extension on tables to easily count rows in tables or views.
 
+## 2.13.2
+
+- Fix a race condition causing query streams to not emit new data around some
+  transaction setups.
+
 ## 2.13.1
 
 - Fix a bug when running batches over serialized communication channels.

drift/lib/src/runtime/executor/stream_queries.dart

@@ -303,6 +303,7 @@ class QueryStream {
     for (final op in _runningOperations) {
       op.cancel();
     }
+    _runningOperations.clear();
   }
 
   Future<void> fetchAndEmitData() async {

drift/test/integration_tests/regress_2744_test.dart

@@ -0,0 +1,30 @@
+import 'package:drift/drift.dart';
+import 'package:test/test.dart';
+
+import '../generated/todos.dart';
+import '../test_utils/test_utils.dart';
+
+void main() {
+  test('updates after transaction', () async {
+    // Regression test for https://github.com/simolus3/drift/issues/2744
+    final db = TodoDb(testInMemoryDatabase());
+    final categories = db.categories.all().watch();
+
+    expect(categories, emits(isEmpty));
+    await db.categories
+        .insertOne(CategoriesCompanion.insert(description: 'desc1'));
+    expect(categories, emits(hasLength(1)));
+
+    await db.categories.deleteAll();
+    await db.batch((batch) {
+      batch.insert(
+          db.categories, CategoriesCompanion.insert(description: 'desc2'));
+    });
+
+    expect(
+        categories,
+        emits([
+          isA<Category>().having((e) => e.description, 'description', 'desc2')
+        ]));
+  });
+}

examples/migrations_example/lib/database.dart

@@ -21,7 +21,7 @@ class Database extends _$Database {
   @override
   int get schemaVersion => latestSchemaVersion;
 
-  Database(DatabaseConnection super.connection);
+  Database(super.connection);
 
   @override
   MigrationStrategy get migration {