prevent readers from spawning while running withAllConnections

Author: davidmartos96

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

@@ -31,6 +31,8 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   final MutexImpl mutex;
 
+  bool _runningWithAllConnections = false;
+
   @override
   bool closed = false;
 
@@ -88,6 +90,11 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       return;
     }
 
+    if (_availableReadConnections.isEmpty && _runningWithAllConnections) {
+      // Wait until withAllConnections is done
+      return;
+    }
+
     var nextItem = _queue.removeFirst();
     while (nextItem.completer.isCompleted) {
       // This item already timed out - try the next one if available
@@ -237,16 +244,26 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       Future<T> Function(
               SqliteWriteContext writer, List<SqliteReadContext> readers)
           block) async {
-    final blockCompleter = Completer<T>();
-    final (write, reads) = await _lockAllConns<T>(blockCompleter);
+    try {
+      _runningWithAllConnections = true;
+
+      final blockCompleter = Completer<T>();
+      final (write, reads) = await _lockAllConns<T>(blockCompleter);
 
     try {
       final res = await block(write, reads);
       blockCompleter.complete(res);
       return res;
     } catch (e, st) {
-      blockCompleter.completeError(e, st);
-      rethrow;
+        blockCompleter.completeError(e, st);
+        rethrow;
+      }
+    } finally {
+      _runningWithAllConnections = false;
+
+      // Continue processing any pending read requests that may have been queued while
+      // the block was running.
+      Timer.run(_nextRead);
     }
   }
 

packages/sqlite_async/test/native/basic_test.dart

@@ -2,14 +2,17 @@
 library;
 
 import 'dart:async';
+import 'dart:io';
 import 'dart:math';
 
 import 'package:collection/collection.dart';
+import 'package:path/path.dart' show join;
 import 'package:sqlite3/common.dart' as sqlite;
 import 'package:sqlite3/sqlite3.dart' show Row;
 import 'package:sqlite_async/sqlite_async.dart';
 import 'package:test/test.dart';
 
+import '../utils/abstract_test_utils.dart';
 import '../utils/test_utils_impl.dart';
 
 final testUtils = TestUtils();
@@ -126,7 +129,7 @@ void main() {
 
       print("${DateTime.now()} start");
       await db.withAllConnections((writer, readers) async {
-        assert(readers.length == 3);
+        expect(readers.length, 3);
 
         // Run some reads during the block that they should run after the block finishes and releases
         // all locks
@@ -160,6 +163,64 @@ void main() {
       await readsCalledWhileWithAllConnsRunning;
     });
 
+    test('prevent opening new readers while in withAllConnections', () async {
+      final sharedStateDir = Directory.systemTemp.createTempSync();
+      addTearDown(() => sharedStateDir.deleteSync(recursive: true));
+
+      final File sharedStateFile =
+          File(join(sharedStateDir.path, 'shared-state.txt'));
+
+      sharedStateFile.writeAsStringSync('initial');
+
+      final db = SqliteDatabase.withFactory(
+          _TestSqliteOpenFactoryWithSharedStateFile(
+              path: path, sharedStateFilePath: sharedStateFile.path),
+          maxReaders: 3);
+      await db.initialize();
+      await createTables(db);
+
+      // The writer saw 'initial' in the file when opening the connection
+      expect(
+        await db
+            .writeLock((c) => c.get('SELECT file_contents_on_open() AS state')),
+        {'state': 'initial'},
+      );
+
+      final withAllConnectionsCompleter = Completer<void>();
+
+      final withAllConnsFut = db.withAllConnections((writer, readers) async {
+        expect(readers.length, 0); // No readers yet
+
+        // Simulate some work until the file is updated
+        await Future.delayed(const Duration(milliseconds: 200));
+        sharedStateFile.writeAsStringSync('updated');
+
+        await withAllConnectionsCompleter.future;
+      });
+
+      // Start a reader that gets the contents of the shared file
+      bool readFinished = false;
+      final someReadFut =
+          db.get('SELECT file_contents_on_open() AS state', []).then((r) {
+        readFinished = true;
+        return r;
+      });
+
+      // The withAllConnections should prevent the reader from opening
+      await Future.delayed(const Duration(milliseconds: 100));
+      expect(readFinished, isFalse);
+
+      // Free all the locks
+      withAllConnectionsCompleter.complete();
+      await withAllConnsFut;
+
+      final readerInfo = await someReadFut;
+      expect(readFinished, isTrue);
+      // The read should see the updated value in the file. This checks
+      // that a reader doesn't spawn while running withAllConnections
+      expect(readerInfo, {'state': 'updated'});
+    });
+
     test('read-only transactions', () async {
       final db = await testUtils.setupDatabase(path: path);
       await createTables(db);
@@ -439,3 +500,31 @@ class _InvalidPragmaOnOpenFactory extends DefaultSqliteOpenFactory {
     ];
   }
 }
+
+class _TestSqliteOpenFactoryWithSharedStateFile
+    extends TestDefaultSqliteOpenFactory {
+  final String sharedStateFilePath;
+
+  _TestSqliteOpenFactoryWithSharedStateFile(
+      {required super.path, required this.sharedStateFilePath});
+
+  @override
+  sqlite.CommonDatabase open(SqliteOpenOptions options) {
+    final File sharedStateFile = File(sharedStateFilePath);
+    final String sharedState = sharedStateFile.readAsStringSync();
+
+    final db = super.open(options);
+
+    // Function to return the contents of the shared state file at the time of opening
+    // so that we know at which point the factory was called.
+    db.createFunction(
+      functionName: 'file_contents_on_open',
+      argumentCount: const sqlite.AllowedArgumentCount(0),
+      function: (args) {
+        return sharedState;
+      },
+    );
+
+    return db;
+  }
+}