From 0bfebbc4e43a1d52f162e0d3c63f7c7a4dcec88d Mon Sep 17 00:00:00 2001 From: ZhuchkaTriplesix Date: Mon, 13 Jul 2026 01:38:14 +0300 Subject: [PATCH] fix(core): fix connection pool eviction and exception wrapping in SQLite and MySQL (#332) --- lib/core/database/mysql_connection.dart | 15 ++ lib/core/database/mysql_connection_pool.dart | 28 +++- lib/core/database/sqlite_connection.dart | 15 ++ lib/core/database/sqlite_connection_pool.dart | 59 +++++--- .../database/mysql_connection_pool_test.dart | 99 +++++++++++++ .../database/sqlite_connection_pool_test.dart | 131 ++++++++++++++++++ 6 files changed, 321 insertions(+), 26 deletions(-) create mode 100644 test/core/database/mysql_connection_pool_test.dart create mode 100644 test/core/database/sqlite_connection_pool_test.dart diff --git a/lib/core/database/mysql_connection.dart b/lib/core/database/mysql_connection.dart index d1fb0ad6..fd467dd7 100644 --- a/lib/core/database/mysql_connection.dart +++ b/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; +} diff --git a/lib/core/database/mysql_connection_pool.dart b/lib/core/database/mysql_connection_pool.dart index 63c01f08..1d4289dd 100644 --- a/lib/core/database/mysql_connection_pool.dart +++ b/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(); diff --git a/lib/core/database/sqlite_connection.dart b/lib/core/database/sqlite_connection.dart index 15455a72..3663fce8 100644 --- a/lib/core/database/sqlite_connection.dart +++ b/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; +} diff --git a/lib/core/database/sqlite_connection_pool.dart b/lib/core/database/sqlite_connection_pool.dart index ae2f13c7..c7b2ee3c 100644 --- a/lib/core/database/sqlite_connection_pool.dart +++ b/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,11 +158,7 @@ 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 disconnectAll() async { @@ -147,7 +166,7 @@ class SqliteConnectionPool { _pool.clear(); for (final entry in entries) { entry.idleTimer?.cancel(); - await entry.connection.disconnect(); + await entry.connection.forceClose(); } } } diff --git a/test/core/database/mysql_connection_pool_test.dart b/test/core/database/mysql_connection_pool_test.dart new file mode 100644 index 00000000..0448b593 --- /dev/null +++ b/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 connect({int connectTimeoutMs = 10000}) async { + connectCount++; + _connected = true; + } + + @override + Future disconnect() async { + disconnectCount++; + _connected = false; + } + + @override + Future forceClose() async { + forceCloseCount++; + _connected = false; + } + + @override + Future 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()), + ); + }); + + 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()), + ); + }); + }); +} diff --git a/test/core/database/sqlite_connection_pool_test.dart b/test/core/database/sqlite_connection_pool_test.dart new file mode 100644 index 00000000..1b2b978e --- /dev/null +++ b/test/core/database/sqlite_connection_pool_test.dart @@ -0,0 +1,131 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:querya_desktop/core/database/sqlite_connection.dart'; +import 'package:querya_desktop/core/database/sqlite_connection_pool.dart'; +import 'package:querya_desktop/core/storage/local_db.dart'; + +ConnectionRow _row({int? id = 1}) => ConnectionRow( + id: id, + type: 'sqlite', + name: 'test', + host: '/path/to/test.db', + createdAt: '2020-01-01T00:00:00Z', + ); + +class FakeSqliteConnection extends SqliteConnection { + FakeSqliteConnection({super.id = 1}) + : super( + name: 'fake', + path: '/path/to/test.db', + ); + + bool _connected = false; + int connectCount = 0; + int disconnectCount = 0; + int forceCloseCount = 0; + + @override + bool get isConnected => _connected; + + @override + Future connect() async { + connectCount++; + _connected = true; + } + + @override + Future disconnect() async { + disconnectCount++; + _connected = false; + } + + @override + Future forceClose() async { + forceCloseCount++; + _connected = false; + } +} + +void main() { + group('SqliteConnectionPool', () { + test('acquire increments refs and connects if needed', () async { + final fake = FakeSqliteConnection(); + final pool = SqliteConnectionPool( + createAndConnect: (row, {required mode}) async => fake, + ); + + final lease = await pool.acquire(_row(id: 1)); + expect(fake.connectCount, 1); + expect(fake.isConnected, isTrue); + + lease.release(); + }); + + test('evicts oldest idle entry when maxEntries reached', () async { + final fakes = {}; + final pool = SqliteConnectionPool( + maxEntries: 2, + createAndConnect: (row, {required mode}) async { + final c = FakeSqliteConnection(id: row.id ?? 1); + fakes[row.id ?? 1] = c; + return c; + }, + ); + + final l1 = await pool.acquire(_row(id: 1)); + l1.release(); + + final l2 = await pool.acquire(_row(id: 2)); + l2.release(); + + // At capacity (2 idle). Acquiring #3 should evict #1 (oldest idle). + final l3 = await pool.acquire(_row(id: 3)); + expect(fakes[1]!.forceCloseCount, 1); + l3.release(); + }); + + test('throws StateError when all slots busy at maxEntries', () async { + final pool = SqliteConnectionPool( + maxEntries: 2, + createAndConnect: (row, {required mode}) async => + FakeSqliteConnection(id: row.id ?? 1), + ); + + await pool.acquire(_row(id: 1)); // busy + await pool.acquire(_row(id: 2)); // busy + + expect( + () => pool.acquire(_row(id: 3)), + throwsA(isA()), + ); + }); + + test('rethrows SqliteConnectionException wrapped on acquire error', () async { + final pool = SqliteConnectionPool( + createAndConnect: (row, {required mode}) async { + throw Exception('network boom'); + }, + ); + + expect( + () => pool.acquire(_row(id: 1)), + throwsA(isA()), + ); + }); + + test('disconnectAll force-closes all pooled entries', () async { + final fake1 = FakeSqliteConnection(id: 1); + final fake2 = FakeSqliteConnection(id: 2); + var i = 0; + final pool = SqliteConnectionPool( + createAndConnect: (row, {required mode}) async => i++ == 0 ? fake1 : fake2, + ); + + await pool.acquire(_row(id: 1)); + await pool.acquire(_row(id: 2)); + await pool.disconnectAll(); + + expect(fake1.forceCloseCount, 1); + expect(fake2.forceCloseCount, 1); + }); + }); +}