Handle uncaught errors inside connection isolates.

Author: rkistner

lib/src/connection_pool.dart

@@ -58,6 +58,9 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       var completer = Completer<T>();
 
       var futures = _readConnections.sublist(0).map((connection) async {
+        if (connection.closed) {
+          _readConnections.remove(connection);
+        }
         try {
           return await connection.readLock((ctx) async {
             if (haveLock) {
@@ -107,6 +110,9 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
     if (closed) {
       throw AssertionError('Closed');
     }
+    if (_writeConnection?.closed == true) {
+      _writeConnection = null;
+    }
     _writeConnection ??= SqliteConnectionImpl(
         upstreamPort: _upstreamPort,
         primary: false,

lib/src/sqlite_connection.dart

@@ -103,4 +103,7 @@ abstract class SqliteConnection extends SqliteWriteContext {
       {Duration? lockTimeout, String? debugContext});
 
   Future<void> close();
+
+  /// Returns true if the connection is closed
+  bool get closed;
 }

lib/src/sqlite_connection_impl.dart

@@ -44,6 +44,10 @@ class SqliteConnectionImpl with SqliteQueries implements SqliteConnection {
     await _isolateClient.ready;
   }
 
+  bool get closed {
+    return _isolateClient.closed;
+  }
+
   Future<void> _open(SqliteOpenFactory openFactory,
       {required bool primary,
       required SerializedPortClient upstreamPort}) async {
@@ -212,9 +216,20 @@ void _sqliteConnectionIsolate(_SqliteConnectionParams params) async {
     // running migrations, and other setup.
     await client.post(const InitDb());
   }
+
   final db = await params.openFactory.open(SqliteOpenOptions(
       primaryConnection: params.primary, readOnly: params.readOnly));
 
+  runZonedGuarded(() async {
+    await _sqliteConnectionIsolateInner(params, client, db);
+  }, (error, stack) {
+    db.dispose();
+    throw error;
+  });
+}
+
+Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
+    ChildPortClient client, sqlite.Database db) async {
   final server = params.portServer;
   final commandPort = ReceivePort();
 

lib/src/sqlite_database.dart

@@ -104,6 +104,10 @@ class SqliteDatabase with SqliteQueries implements SqliteConnection {
     await _initialized;
   }
 
+  bool get closed {
+    return _pool.closed;
+  }
+
   void _listenForEvents() {
     UpdateNotification? updates;
 

test/basic_test.dart

@@ -249,6 +249,78 @@ void main() {
         await db.execute('BEGIN');
       }, throwsA((e) => e is sqlite.SqliteException));
     });
+
+    test('should handle normal errors', () async {
+      final db = await setupDatabase(path: path);
+      await createTables(db);
+      Error? caughtError;
+      final syntheticError = ArgumentError('foobar');
+      await db.computeWithDatabase<void>((db) async {
+        throw syntheticError;
+      }).catchError((error) {
+        caughtError = error;
+      });
+      expect(caughtError.toString(), equals(syntheticError.toString()));
+
+      // Check that we can still continue afterwards
+      final computed = await db.computeWithDatabase((db) async {
+        return 5;
+      });
+      expect(computed, equals(5));
+    });
+
+    test('should handle uncaught errors', () async {
+      final db = await setupDatabase(path: path);
+      await createTables(db);
+      Object? caughtError;
+      await db.computeWithDatabase<void>((db) async {
+        Future<void> asyncCompute() async {
+          throw ArgumentError('uncaught async error');
+        }
+
+        asyncCompute();
+      }).catchError((error) {
+        caughtError = error;
+      });
+      // This may change into a better error in the future
+      expect(caughtError.toString(), equals("Instance of 'ClosedException'"));
+
+      // Check that we can still continue afterwards
+      final computed = await db.computeWithDatabase((db) async {
+        return 5;
+      });
+      expect(computed, equals(5));
+    });
+
+    test('should handle uncaught errors in read connections', () async {
+      final db = await setupDatabase(path: path);
+      await createTables(db);
+      for (var i = 0; i < 10; i++) {
+        Object? caughtError;
+
+        await db.readTransaction((ctx) async {
+          await ctx.computeWithDatabase((db) async {
+            Future<void> asyncCompute() async {
+              throw ArgumentError('uncaught async error');
+            }
+
+            asyncCompute();
+          });
+        }).catchError((error) {
+          caughtError = error;
+        });
+        // This may change into a better error in the future
+        expect(caughtError.toString(), equals("Instance of 'ClosedException'"));
+      }
+
+      // Check that we can still continue afterwards
+      final computed = await db.readTransaction((ctx) async {
+        return await ctx.computeWithDatabase((db) async {
+          return 5;
+        });
+      });
+      expect(computed, equals(5));
+    });
   });
 }