fix(db): a failed reconnect gives its lease back in the pools (#1306)

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<void> _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<void> _reconnect(_PoolEntry entry, MysqlSessionMode mode) {

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<void> _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<void> _reconnect(_PoolEntry entry, PgSessionMode mode) {

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<void> _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;

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<void> 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<void>.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();
+    });
+  });
 }

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<void> 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<void>.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();
+    });
+  });
 }

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<void> 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<void>.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();
+    });
+  });
 }