fix(core): fix connection pool eviction and exception wrapping in SQLite and MySQL (#332)

lib/core/database/mysql_connection.dart

@@ -461,3 +461,18 @@ class MysqlConnection {
     return out;
   }
 }
+
+class MysqlConnectionException implements Exception {
+  MysqlConnectionException(
+    this.message, {
+    this.cause,
+    this.stackTrace,
+  });
+
+  final String message;
+  final Object? cause;
+  final StackTrace? stackTrace;
+
+  @override
+  String toString() => message;
+}

lib/core/database/mysql_connection_pool.dart

@@ -75,12 +75,28 @@ class MysqlConnectionPool {
       return MysqlLease._(this, k, entry.connection);
     }
 
-    await _creationLock.createIfAbsent(k, () async {
-      _evictIfNeededBeforeNewSlot();
-      final conn = await createAndConnect(row, database: database, mode: mode);
-      _pool[k] = _PoolEntry(conn);
-      return conn;
-    });
+    try {
+      await _creationLock.createIfAbsent(k, () async {
+        _evictIfNeededBeforeNewSlot();
+        final conn =
+            await createAndConnect(row, database: database, mode: mode);
+        _pool[k] = _PoolEntry(conn);
+        return conn;
+      });
+    } on StateError {
+      rethrow;
+    } on MysqlConnectionException {
+      rethrow;
+    } catch (e, st) {
+      Error.throwWithStackTrace(
+        MysqlConnectionException(
+          'Failed to acquire MySQL connection for database "$database": $e',
+          cause: e,
+          stackTrace: st,
+        ),
+        st,
+      );
+    }
 
     entry = _pool[k]!;
     entry.touch();

lib/core/database/sqlite_connection.dart

@@ -196,3 +196,18 @@ class SqliteConnection {
     return '"${id.replaceAll('"', '""')}"';
   }
 }
+
+class SqliteConnectionException implements Exception {
+  SqliteConnectionException(
+    this.message, {
+    this.cause,
+    this.stackTrace,
+  });
+
+  final String message;
+  final Object? cause;
+  final StackTrace? stackTrace;
+
+  @override
+  String toString() => message;
+}

lib/core/database/sqlite_connection_pool.dart

@@ -83,12 +83,27 @@ class SqliteConnectionPool {
       return SqliteLease._(this, k, entry.connection);
     }
 
-    await _creationLock.createIfAbsent(k, () async {
-      _evictIfNeededBeforeNewSlot();
-      final conn = await createAndConnect(row, mode: mode);
-      _pool[k] = _PoolEntry(conn);
-      return conn;
-    });
+    try {
+      await _creationLock.createIfAbsent(k, () async {
+        _evictIfNeededBeforeNewSlot();
+        final conn = await createAndConnect(row, mode: mode);
+        _pool[k] = _PoolEntry(conn);
+        return conn;
+      });
+    } on StateError {
+      rethrow;
+    } on SqliteConnectionException {
+      rethrow;
+    } catch (e, st) {
+      Error.throwWithStackTrace(
+        SqliteConnectionException(
+          'Failed to acquire SQLite connection: $e',
+          cause: e,
+          stackTrace: st,
+        ),
+        st,
+      );
+    }
 
     entry = _pool[k]!;
     entry.touch();
@@ -105,15 +120,22 @@ class SqliteConnectionPool {
     while (_pool.length >= maxEntries) {
       final idle = _pool.entries.where((e) => e.value.refs == 0).toList();
       if (idle.isEmpty) {
-        break;
+        throw StateError(
+          'SQLite connection pool exhausted: $maxEntries slots in use.',
+        );
       }
       idle.sort((a, b) => a.value.lastUsed.compareTo(b.value.lastUsed));
-      final oldestKey = idle.first.key;
-      final oldestEntry = _pool.remove(oldestKey);
-      oldestEntry?.connection.disconnect();
+      _removeEntryClosing(idle.first.key);
     }
   }
 
+  void _removeEntryClosing(String k) {
+    final entry = _pool.remove(k);
+    if (entry == null) return;
+    entry.idleTimer?.cancel();
+    unawaited(entry.connection.forceClose());
+  }
+
   void _release(String key) {
     final entry = _pool[key];
     if (entry == null) return;
@@ -122,10 +144,11 @@ class SqliteConnectionPool {
       entry.refs = 0;
       entry.idleTimer?.cancel();
       entry.idleTimer = Timer(idleDisposeDelay, () {
-        if (_pool[key] == entry && entry.refs == 0) {
-          _pool.remove(key);
-          entry.connection.disconnect();
-        }
+        final e = _pool[key];
+        if (e == null || e.refs > 0) return;
+        e.idleTimer = null;
+        unawaited(e.connection.disconnect());
+        _pool.remove(key);
       });
     }
   }
@@ -135,19 +158,15 @@ class SqliteConnectionPool {
     SqliteSessionMode mode = SqliteSessionMode.readOnly,
   }) {
     final k = keyFor(row.id, mode);
-    final entry = _pool.remove(k);
-    if (entry != null) {
-      entry.idleTimer?.cancel();
-      entry.connection.forceClose();
-    }
+    _removeEntryClosing(k);
   }
 
   Future<void> disconnectAll() async {
     final entries = _pool.values.toList();
     _pool.clear();
     for (final entry in entries) {
       entry.idleTimer?.cancel();
-      await entry.connection.disconnect();
+      await entry.connection.forceClose();
     }
   }
 }

test/core/database/mysql_connection_pool_test.dart

@@ -0,0 +1,99 @@
+import 'package:flutter_test/flutter_test.dart';
+import 'package:querya_desktop/core/database/mysql_connection.dart';
+import 'package:querya_desktop/core/database/mysql_connection_pool.dart';
+import 'package:querya_desktop/core/storage/local_db.dart';
+
+ConnectionRow _row({int? id = 1}) => ConnectionRow(
+      id: id,
+      type: 'mysql',
+      name: 'test',
+      createdAt: '2020-01-01T00:00:00Z',
+    );
+
+class FakeMysqlConnection extends MysqlConnection {
+  FakeMysqlConnection({super.id = 1})
+      : super(
+          name: 'fake',
+          host: 'localhost',
+          port: 3306,
+          database: 'testdb',
+        );
+
+  bool _connected = false;
+  int connectCount = 0;
+  int disconnectCount = 0;
+  int forceCloseCount = 0;
+  int setReadOnlyCount = 0;
+
+  @override
+  bool get isConnected => _connected;
+
+  @override
+  Future<void> connect({int connectTimeoutMs = 10000}) async {
+    connectCount++;
+    _connected = true;
+  }
+
+  @override
+  Future<void> disconnect() async {
+    disconnectCount++;
+    _connected = false;
+  }
+
+  @override
+  Future<void> forceClose() async {
+    forceCloseCount++;
+    _connected = false;
+  }
+
+  @override
+  Future<void> setSessionReadOnly(bool readOnly) async {
+    setReadOnlyCount++;
+  }
+}
+
+void main() {
+  group('MysqlConnectionPool', () {
+    test('acquire increments refs and connects if needed', () async {
+      final fake = FakeMysqlConnection();
+      final pool = MysqlConnectionPool(
+        createAndConnect: (row, {required database, required mode}) async => fake,
+      );
+
+      final lease = await pool.acquire(_row(id: 1), database: 'testdb');
+      expect(fake.connectCount, 1);
+      expect(fake.setReadOnlyCount, 1);
+      expect(fake.isConnected, isTrue);
+
+      lease.release();
+    });
+
+    test('wraps unknown exceptions in MysqlConnectionException', () async {
+      final pool = MysqlConnectionPool(
+        createAndConnect: (row, {required database, required mode}) async {
+          throw Exception('Connection refused');
+        },
+      );
+
+      expect(
+        () => pool.acquire(_row(id: 1), database: 'testdb'),
+        throwsA(isA<MysqlConnectionException>()),
+      );
+    });
+
+    test('rethrows StateError directly when pool exhausted', () async {
+      final pool = MysqlConnectionPool(
+        maxEntries: 1,
+        createAndConnect: (row, {required database, required mode}) async =>
+            FakeMysqlConnection(id: row.id ?? 1),
+      );
+
+      await pool.acquire(_row(id: 1), database: 'db1'); // busy
+
+      expect(
+        () => pool.acquire(_row(id: 2), database: 'db2'),
+        throwsA(isA<StateError>()),
+      );
+    });
+  });
+}