From 250f8a4b8be64c60cc5a63bd1a9c4bf63694768f Mon Sep 17 00:00:00 2001 From: ZhuchkaTriplesix Date: Sat, 10 Oct 2026 16:33:34 +0300 Subject: [PATCH] fix(db): a failed reconnect gives its lease back in the PostgreSQL, MySQL and SQLite pools (#1306) acquire counted the lease before reconnecting a dropped entry. When the reconnect threw, nobody held the lease to release it, so the entry stayed in use for good: never idle-disposed, never evicted, and a few of them filled the pool with pool exhausted errors. The count is now given back when the reconnect fails. --- lib/core/database/mysql_connection_pool.dart | 17 +++++- .../database/postgres_connection_pool.dart | 17 +++++- lib/core/database/sqlite_connection_pool.dart | 21 +++++--- .../database/mysql_connection_pool_test.dart | 54 +++++++++++++++++++ .../postgres_connection_pool_test.dart | 54 +++++++++++++++++++ .../database/sqlite_connection_pool_test.dart | 51 ++++++++++++++++++ 6 files changed, 204 insertions(+), 10 deletions(-) diff --git a/lib/core/database/mysql_connection_pool.dart b/lib/core/database/mysql_connection_pool.dart index fd933c37..274426ef 100644 --- a/lib/core/database/mysql_connection_pool.dart +++ b/lib/core/database/mysql_connection_pool.dart @@ -87,7 +87,7 @@ class MysqlConnectionPool { entry.idleTimer?.cancel(); entry.idleTimer = null; entry.refs++; - if (!entry.connection.isConnected) await _reconnect(entry, mode); + await _ensureConnected(entry, mode); return MysqlLease._(this, entry, entry.connection); } @@ -120,7 +120,7 @@ class MysqlConnectionPool { entry.idleTimer?.cancel(); entry.idleTimer = null; entry.refs++; - if (!entry.connection.isConnected) await _reconnect(entry, mode); + await _ensureConnected(entry, mode); return MysqlLease._(this, entry, entry.connection); } @@ -147,6 +147,19 @@ class MysqlConnectionPool { /// Releases one lease on [entry]. A lease taken before an [interrupt] points /// at an entry that is no longer in the pool: it must not touch the entry /// that replaced it under the same key, so it does nothing. + /// Makes the entry's connection usable for a lease that was just counted. + /// When the reconnect fails the count is given back, so the entry can be + /// idle-disposed or evicted instead of staying "in use" for good (#1306). + Future _ensureConnected(_PoolEntry entry, MysqlSessionMode mode) async { + if (entry.connection.isConnected) return; + try { + await _reconnect(entry, mode); + } catch (_) { + _release(entry); + rethrow; + } + } + /// Brings a dropped entry back with one reconnect and one session setting, /// even when several callers ask at once; they all wait for the same attempt. Future _reconnect(_PoolEntry entry, MysqlSessionMode mode) { diff --git a/lib/core/database/postgres_connection_pool.dart b/lib/core/database/postgres_connection_pool.dart index cc5065db..5eb51c5c 100644 --- a/lib/core/database/postgres_connection_pool.dart +++ b/lib/core/database/postgres_connection_pool.dart @@ -99,7 +99,7 @@ class PostgresConnectionPool { entry.idleTimer?.cancel(); entry.idleTimer = null; entry.refs++; - if (!entry.connection.isConnected) await _reconnect(entry, mode); + await _ensureConnected(entry, mode); return PgLease._(this, entry, entry.connection); } @@ -132,10 +132,23 @@ class PostgresConnectionPool { entry.idleTimer?.cancel(); entry.idleTimer = null; entry.refs++; - if (!entry.connection.isConnected) await _reconnect(entry, mode); + await _ensureConnected(entry, mode); return PgLease._(this, entry, entry.connection); } + /// Makes the entry's connection usable for a lease that was just counted. + /// When the reconnect fails the count is given back, so the entry can be + /// idle-disposed or evicted instead of staying "in use" for good (#1306). + Future _ensureConnected(_PoolEntry entry, PgSessionMode mode) async { + if (entry.connection.isConnected) return; + try { + await _reconnect(entry, mode); + } catch (_) { + _release(entry); + rethrow; + } + } + /// Brings a dropped entry back with one reconnect and one session setting, /// even when several callers ask at once; they all wait for the same attempt. Future _reconnect(_PoolEntry entry, PgSessionMode mode) { diff --git a/lib/core/database/sqlite_connection_pool.dart b/lib/core/database/sqlite_connection_pool.dart index eab1bacc..42e0e800 100644 --- a/lib/core/database/sqlite_connection_pool.dart +++ b/lib/core/database/sqlite_connection_pool.dart @@ -91,9 +91,7 @@ class SqliteConnectionPool { entry.idleTimer?.cancel(); entry.idleTimer = null; entry.refs++; - if (!entry.connection.isConnected) { - await entry.connection.connect(); - } + await _ensureConnected(entry); return SqliteLease._(this, entry, entry.connection); } @@ -125,9 +123,7 @@ class SqliteConnectionPool { entry.idleTimer?.cancel(); entry.idleTimer = null; entry.refs++; - if (!entry.connection.isConnected) { - await entry.connection.connect(); - } + await _ensureConnected(entry); return SqliteLease._(this, entry, entry.connection); } @@ -144,6 +140,19 @@ class SqliteConnectionPool { } } + /// Reconnects a dropped entry for a lease that was just counted. When the + /// reconnect fails the count is given back, so the entry can be + /// idle-disposed or evicted instead of staying "in use" for good (#1306). + Future _ensureConnected(_PoolEntry entry) async { + if (entry.connection.isConnected) return; + try { + await entry.connection.connect(); + } catch (_) { + _release(entry); + rethrow; + } + } + void _removeEntryClosing(String k) { final entry = _pool.remove(k); if (entry == null) return; diff --git a/test/core/database/mysql_connection_pool_test.dart b/test/core/database/mysql_connection_pool_test.dart index 9d6f04ab..a5f0e19a 100644 --- a/test/core/database/mysql_connection_pool_test.dart +++ b/test/core/database/mysql_connection_pool_test.dart @@ -27,12 +27,20 @@ class FakeMysqlConnection extends MysqlConnection { bool? lastReadOnly; bool? openTransaction; + /// Thrown (once) by the next [connect]. + Object? connectError; + @override bool get isConnected => _connected; @override Future connect({int connectTimeoutMs = 10000}) async { connectCount++; + final error = connectError; + if (error != null) { + connectError = null; + throw error; + } _connected = true; } @@ -297,4 +305,50 @@ void main() { mcp.release(); }); }); + + group('a failed reconnect gives the lease back (#1306)', () { + test('the entry is still idle-disposed afterwards', () async { + final fake = FakeMysqlConnection(); + final pool = MysqlConnectionPool( + idleDisposeDelay: const Duration(milliseconds: 20), + createAndConnect: (row, {required database, required mode}) async { + await fake.connect(); + return fake; + }, + ); + final first = await pool.acquire(_row(), database: 'db'); + await fake.forceClose(); // the session dropped + fake.connectError = StateError('server restarted'); + await expectLater( + pool.acquire(_row(), database: 'db'), throwsStateError); + + first.release(); + await Future.delayed(const Duration(milliseconds: 80)); + expect(fake.disconnectCount, 1); + }); + + test('failed reconnects do not exhaust the pool', () async { + final pool = MysqlConnectionPool( + maxEntries: 2, + createAndConnect: (row, {required database, required mode}) async { + final c = FakeMysqlConnection(); + await c.connect(); + return c; + }, + ); + for (final db in ['a', 'b']) { + final lease = await pool.acquire(_row(), database: db); + final fake = lease.connection as FakeMysqlConnection; + await fake.forceClose(); + fake.connectError = StateError('down'); + await expectLater( + pool.acquire(_row(), database: db), throwsStateError); + lease.release(); + } + // Both slots are idle, so a third key evicts one instead of failing. + final c = await pool.acquire(_row(), database: 'c'); + expect(c.connection.isConnected, isTrue); + c.release(); + }); + }); } diff --git a/test/core/database/postgres_connection_pool_test.dart b/test/core/database/postgres_connection_pool_test.dart index aeef623c..a37416fa 100644 --- a/test/core/database/postgres_connection_pool_test.dart +++ b/test/core/database/postgres_connection_pool_test.dart @@ -28,12 +28,20 @@ class FakePostgresConnection extends PostgresConnection { bool? lastReadOnly; bool? openTransaction; + /// Thrown (once) by the next [connect]. + Object? connectError; + @override bool get isConnected => _connected; @override Future connect() async { connectCount++; + final error = connectError; + if (error != null) { + connectError = null; + throw error; + } _connected = true; } @@ -664,4 +672,50 @@ void main() { mcp.release(); }); }); + + group('a failed reconnect gives the lease back (#1306)', () { + test('the entry is still idle-disposed afterwards', () async { + final fake = FakePostgresConnection(); + final pool = PostgresConnectionPool( + idleDisposeDelay: const Duration(milliseconds: 20), + createAndConnect: (row, {required database, required mode}) async { + await fake.connect(); + return fake; + }, + ); + final first = await pool.acquire(_row(), database: 'db'); + await fake.forceClose(); // the session dropped + fake.connectError = StateError('server restarted'); + await expectLater( + pool.acquire(_row(), database: 'db'), throwsStateError); + + first.release(); + await Future.delayed(const Duration(milliseconds: 80)); + expect(fake.disconnectCount, 1); + }); + + test('failed reconnects do not exhaust the pool', () async { + final pool = PostgresConnectionPool( + maxEntries: 2, + createAndConnect: (row, {required database, required mode}) async { + final c = FakePostgresConnection(); + await c.connect(); + return c; + }, + ); + for (final db in ['a', 'b']) { + final lease = await pool.acquire(_row(), database: db); + final fake = lease.connection as FakePostgresConnection; + await fake.forceClose(); + fake.connectError = StateError('down'); + await expectLater( + pool.acquire(_row(), database: db), throwsStateError); + lease.release(); + } + // Both slots are idle, so a third key evicts one instead of failing. + final c = await pool.acquire(_row(), database: 'c'); + expect(c.connection.isConnected, isTrue); + c.release(); + }); + }); } diff --git a/test/core/database/sqlite_connection_pool_test.dart b/test/core/database/sqlite_connection_pool_test.dart index 954e3370..2dc493de 100644 --- a/test/core/database/sqlite_connection_pool_test.dart +++ b/test/core/database/sqlite_connection_pool_test.dart @@ -24,12 +24,20 @@ class FakeSqliteConnection extends SqliteConnection { int forceCloseCount = 0; bool? openTransaction; + /// Thrown (once) by the next [connect]. + Object? connectError; + @override bool get isConnected => _connected; @override Future connect() async { connectCount++; + final error = connectError; + if (error != null) { + connectError = null; + throw error; + } _connected = true; } @@ -242,4 +250,47 @@ void main() { expect(replacement.disconnectCount, 1); }); }); + + group('a failed reconnect gives the lease back (#1306)', () { + test('the entry is still idle-disposed afterwards', () async { + final fake = FakeSqliteConnection(); + final pool = SqliteConnectionPool( + idleDisposeDelay: const Duration(milliseconds: 20), + createAndConnect: (row, {required mode}) async { + await fake.connect(); + return fake; + }, + ); + final first = await pool.acquire(_row()); + await fake.forceClose(); // the file handle was lost + fake.connectError = StateError('file is gone'); + await expectLater(pool.acquire(_row()), throwsStateError); + + first.release(); + await Future.delayed(const Duration(milliseconds: 80)); + expect(fake.disconnectCount, 1); + }); + + test('failed reconnects do not exhaust the pool', () async { + final pool = SqliteConnectionPool( + maxEntries: 2, + createAndConnect: (row, {required mode}) async { + final c = FakeSqliteConnection(); + await c.connect(); + return c; + }, + ); + for (final id in [1, 2]) { + final lease = await pool.acquire(_row(id: id)); + final fake = lease.connection as FakeSqliteConnection; + await fake.forceClose(); + fake.connectError = StateError('down'); + await expectLater(pool.acquire(_row(id: id)), throwsStateError); + lease.release(); + } + final c = await pool.acquire(_row(id: 3)); + expect(c.connection.isConnected, isTrue); + c.release(); + }); + }); }