From 54e48c460a9ebc59206bd85e572a5525b2f577c4 Mon Sep 17 00:00:00 2001 From: ZhuchkaTriplesix Date: Sat, 10 Oct 2026 19:22:07 +0300 Subject: [PATCH 1/2] fix(db): reflect underlying driver socket state in isConnected and mark session closed on socket drop - Check _conn.isOpen in PostgresConnection.isConnected. - Check _conn.connected and listen to onClose in MysqlConnection. - Track socket.done and disconnect on socket errors in RedisConnection. - Reset stale closed sockets upon reconnecting. - Add forceClose triggers on connection lost / host unreachable errors. - Add unit tests for socket state reflection and pool drop reconnect. Closes #1373 --- lib/core/database/mysql_connection.dart | 34 ++++++++++- lib/core/database/postgres_connection.dart | 46 +++++++++++++-- lib/core/database/redis_connection.dart | 56 ++++++++++++++++++- .../database/mysql_connection_pool_test.dart | 26 +++++++++ test/core/database/mysql_connection_test.dart | 33 +++++++++++ .../postgres_connection_pool_test.dart | 26 +++++++++ .../database/postgres_connection_test.dart | 33 +++++++++++ 7 files changed, 243 insertions(+), 11 deletions(-) diff --git a/lib/core/database/mysql_connection.dart b/lib/core/database/mysql_connection.dart index e56d6d69..dd2d790d 100644 --- a/lib/core/database/mysql_connection.dart +++ b/lib/core/database/mysql_connection.dart @@ -166,7 +166,13 @@ class MysqlConnection { bool _isConnected = false; bool _inTransaction = false; - bool get isConnected => _isConnected && _conn != null; + bool get isConnected => _isConnected && _conn != null && _conn!.connected; + + @visibleForTesting + void setConnectionForTest(MySQLConnection? conn, {bool isConnected = true}) { + _conn = conn; + _isConnected = isConnected; + } /// Scrubs sensitive in-memory credentials once the network handshake completes. void scrubCredentials() { @@ -186,7 +192,7 @@ class MysqlConnection { /// for that attempt. Without this, two callers opened two sockets and two SSH /// tunnels, and a failing attempt dropped the session the other one had made. Future connect({int connectTimeoutMs = 10000}) { - if (_isConnected && _conn != null) return Future.value(); + if (isConnected) return Future.value(); final inFlight = _connecting; if (inFlight != null) return inFlight; final attempt = _connectOnce(connectTimeoutMs); @@ -201,7 +207,13 @@ class MysqlConnection { int _generation = 0; Future _connectOnce(int connectTimeoutMs) async { - if (_isConnected && _conn != null) return; + if (isConnected) return; + if (_conn != null && !_conn!.connected) { + try { + await _conn?.close(); + } catch (_) {} + _conn = null; + } final generation = _generation; MySQLConnection? opened; SshTunnelHandle? openedHandle; @@ -296,6 +308,11 @@ class MysqlConnection { } if (generation != _generation) throw _ConnectSuperseded(); _isConnected = true; + opened.onClose(() { + if (identical(_conn, opened)) { + _isConnected = false; + } + }); scrubCredentials(); } catch (e, st) { if (generation != _generation) { @@ -538,6 +555,17 @@ class MysqlConnection { // wait in the queue and are not affected. unawaited(forceClose()); rethrow; + } catch (e) { + if (c != null && !c.connected) { + unawaited(forceClose()); + } else { + final mapped = mapDatabaseError(e, driver: DatabaseDriver.mysql); + if (mapped is ConnectionLostException || + mapped is HostUnreachableException) { + unawaited(forceClose()); + } + } + rethrow; } }); } diff --git a/lib/core/database/postgres_connection.dart b/lib/core/database/postgres_connection.dart index b2e49f45..276175a4 100644 --- a/lib/core/database/postgres_connection.dart +++ b/lib/core/database/postgres_connection.dart @@ -159,7 +159,13 @@ class PostgresConnection { bool _isConnected = false; bool _inTransaction = false; - bool get isConnected => _isConnected && _conn != null; + bool get isConnected => _isConnected && _conn != null && _conn!.isOpen; + + @visibleForTesting + void setConnectionForTest(Connection? conn, {bool isConnected = true}) { + _conn = conn; + _isConnected = isConnected; + } /// Scrubs sensitive in-memory credentials once the network handshake completes. void scrubCredentials() { @@ -218,7 +224,7 @@ class PostgresConnection { /// for that attempt. Without this, two callers opened two sockets and two SSH /// tunnels, and a failing attempt dropped the session the other one had made. Future connect() { - if (_isConnected && _conn != null) return Future.value(); + if (isConnected) return Future.value(); final inFlight = _connecting; if (inFlight != null) return inFlight; final attempt = _connectOnce(); @@ -229,7 +235,13 @@ class PostgresConnection { } Future _connectOnce() async { - if (_isConnected && _conn != null) return; + if (isConnected) return; + if (_conn != null && !_conn!.isOpen) { + try { + await _conn?.close(force: true); + } catch (_) {} + _conn = null; + } var effectivePassword = _password; var effectiveConnectionString = _connectionString; @@ -440,6 +452,17 @@ class PostgresConnection { } on TimeoutException catch (e) { if (e is! PgException) unawaited(forceClose()); rethrow; + } catch (e) { + if (c != null && !c.isOpen) { + unawaited(forceClose()); + } else { + final mapped = mapDatabaseError(e, driver: DatabaseDriver.postgres); + if (mapped is ConnectionLostException || + mapped is HostUnreachableException) { + unawaited(forceClose()); + } + } + rethrow; } }); } @@ -449,12 +472,25 @@ class PostgresConnection { /// call's default timeout while it waits, and could cancel the statement /// that is running (#1216). Future _queued(Future Function(Connection c) call) => - statementQueue.run(() { + statementQueue.run(() async { final c = _conn; if (!isConnected || c == null) { throw StateError('Not connected to PostgreSQL'); } - return call(c); + try { + return await call(c); + } catch (e) { + if (!c.isOpen) { + unawaited(forceClose()); + } else { + final mapped = mapDatabaseError(e, driver: DatabaseDriver.postgres); + if (mapped is ConnectionLostException || + mapped is HostUnreachableException) { + unawaited(forceClose()); + } + } + rethrow; + } }); /// [execute] with [timeout] as the statement's own limit. The limit starts diff --git a/lib/core/database/redis_connection.dart b/lib/core/database/redis_connection.dart index 4df63544..89d5b13c 100644 --- a/lib/core/database/redis_connection.dart +++ b/lib/core/database/redis_connection.dart @@ -119,10 +119,24 @@ class RedisConnection { redis.RedisConnection? _conn; redis.Command? _command; + Socket? _socket; bool _isConnected = false; bool get isConnected => _isConnected && _command != null; + @visibleForTesting + void setConnectionForTest({ + redis.RedisConnection? conn, + redis.Command? command, + Socket? socket, + bool isConnected = true, + }) { + _conn = conn; + _command = command; + _socket = socket; + _isConnected = isConnected; + } + /// Scrubs sensitive in-memory credentials once the network handshake completes. void scrubCredentials() { _password = null; @@ -133,7 +147,7 @@ class RedisConnection { /// for that attempt. Without this, two callers opened two sockets and two SSH /// tunnels, and the second overwrote the first's connection (#1307). Future connect() { - if (_isConnected && _command != null) return Future.value(); + if (isConnected) return Future.value(); final inFlight = _connecting; if (inFlight != null) return inFlight; final attempt = _connectOnce(); @@ -156,7 +170,15 @@ class RedisConnection { } Future _openConnection() async { - if (_isConnected && _command != null) return; + if (isConnected) return; + if (_socket != null) { + try { + _socket?.destroy(); + } catch (_) {} + _socket = null; + _command = null; + _conn = null; + } var effectivePassword = _password; var effectiveConnectionString = _connectionString; @@ -218,6 +240,17 @@ class RedisConnection { effectivePort, timeout: limit, ); + _socket = socket; + final activeSocket = socket; + activeSocket.done.then((_) { + if (identical(_socket, activeSocket)) { + _isConnected = false; + } + }, onError: (_) { + if (identical(_socket, activeSocket)) { + _isConnected = false; + } + }); final command = await client.connectWithSocket(socket); _command = command; command.setParser(redis.RedisParserBulkBinary()); @@ -245,6 +278,7 @@ class RedisConnection { _isConnected = false; _conn = null; _command = null; + _socket = null; // The attempt may have got as far as an open socket (a wrong password, // a silent server): close it, or every failed try leaves one behind. try { @@ -278,6 +312,11 @@ class RedisConnection { final wasConnected = _isConnected; _isConnected = false; _command = null; + final sock = _socket; + _socket = null; + try { + sock?.destroy(); + } catch (_) {} final c = _conn; _conn = null; if (c != null && wasConnected) { @@ -351,7 +390,18 @@ class RedisConnection { if (!isConnected || _command == null) { throw StateError('Not connected to Redis'); } - return _command!.send_object(args); + try { + return await _command!.send_object(args); + } catch (e) { + final mapped = mapDatabaseError(e, driver: DatabaseDriver.redis); + if (mapped is ConnectionLostException || + mapped is HostUnreachableException || + e is SocketException) { + _isConnected = false; + unawaited(disconnect()); + } + rethrow; + } } /// SELECT database index. diff --git a/test/core/database/mysql_connection_pool_test.dart b/test/core/database/mysql_connection_pool_test.dart index 8bccae23..d3353935 100644 --- a/test/core/database/mysql_connection_pool_test.dart +++ b/test/core/database/mysql_connection_pool_test.dart @@ -275,6 +275,32 @@ void main() { l.release(); } }); + + test('a dropped socket (isConnected = false) triggers reconnect upon acquire (#1373)', + () async { + final pool = MysqlConnectionPool( + createAndConnect: (row, {required database, required mode}) async { + final c = FakeMysqlConnection(); + await c.connect(); + return c; + }, + ); + final r = _row(); + final lease1 = await pool.acquire(r, database: 'testdb'); + final conn = lease1.connection as FakeMysqlConnection; + expect(conn.connectCount, 1); + lease1.release(); + + // Simulate remote socket drop without client calling forceClose() + conn._connected = false; + expect(conn.isConnected, isFalse); + + final lease2 = await pool.acquire(r, database: 'testdb'); + expect(conn.connectCount, 2, + reason: 'reconnected because isConnected was false'); + expect(conn.isConnected, isTrue); + lease2.release(); + }); }); group('MCP session (#1217)', () { diff --git a/test/core/database/mysql_connection_test.dart b/test/core/database/mysql_connection_test.dart index 9c895712..ae1cdfb3 100644 --- a/test/core/database/mysql_connection_test.dart +++ b/test/core/database/mysql_connection_test.dart @@ -269,4 +269,37 @@ void main() { ); }); }); + + group('MysqlConnection socket state reflection (#1373)', () { + test('isConnected reflects underlying connection connected state', () { + final conn = MysqlConnection( + id: 1, + name: 'test', + host: 'localhost', + ); + expect(conn.isConnected, isFalse); + + conn.setConnectionForTest(null, isConnected: true); + expect(conn.isConnected, isFalse, reason: '_conn is null'); + }); + + test('execute force-closes and marks disconnected on connection lost error', () async { + final conn = MysqlConnection( + id: 1, + name: 'test', + host: 'localhost', + ); + conn.setConnectionForTest(null, isConnected: true); + conn.runStatementForTest = (sql, params, iterable, timeout) => + Future.error(Exception('Lost connection to MySQL server during query')); + + try { + await conn.execute('SELECT 1'); + fail('Should have thrown'); + } catch (_) {} + + await Future.delayed(const Duration(milliseconds: 10)); + expect(conn.isConnected, isFalse); + }); + }); } diff --git a/test/core/database/postgres_connection_pool_test.dart b/test/core/database/postgres_connection_pool_test.dart index 219e74d2..281d4226 100644 --- a/test/core/database/postgres_connection_pool_test.dart +++ b/test/core/database/postgres_connection_pool_test.dart @@ -635,6 +635,32 @@ void main() { l.release(); } }); + + test('a dropped socket (isConnected = false) triggers reconnect upon acquire (#1373)', + () async { + final pool = PostgresConnectionPool( + createAndConnect: (row, {required database, required mode}) async { + final c = FakePostgresConnection(); + await c.connect(); + return c; + }, + ); + final r = _row(); + final lease1 = await pool.acquire(r, database: 'postgres'); + final conn = lease1.connection as FakePostgresConnection; + expect(conn.connectCount, 1); + lease1.release(); + + // Simulate remote socket drop without client calling forceClose() + conn._connected = false; + expect(conn.isConnected, isFalse); + + final lease2 = await pool.acquire(r, database: 'postgres'); + expect(conn.connectCount, 2, + reason: 'reconnected because isConnected was false'); + expect(conn.isConnected, isTrue); + lease2.release(); + }); }); group('MCP session (#1217)', () { diff --git a/test/core/database/postgres_connection_test.dart b/test/core/database/postgres_connection_test.dart index 29dc064e..92683ad1 100644 --- a/test/core/database/postgres_connection_test.dart +++ b/test/core/database/postgres_connection_test.dart @@ -575,4 +575,37 @@ void main() { ); }); }); + + group('PostgresConnection socket state reflection (#1373)', () { + test('isConnected reflects underlying connection isOpen state', () { + final conn = PostgresConnection( + id: 1, + name: 'test', + host: 'localhost', + ); + expect(conn.isConnected, isFalse); + + conn.setConnectionForTest(null, isConnected: true); + expect(conn.isConnected, isFalse, reason: '_conn is null'); + }); + + test('execute force-closes and marks disconnected on connection lost error', () async { + final conn = PostgresConnection( + id: 1, + name: 'test', + host: 'localhost', + ); + conn.setConnectionForTest(null, isConnected: true); + conn.runStatementForTest = (sql, timeout) => + Future.error(Exception('Connection reset by peer')); + + try { + await conn.execute('SELECT 1'); + fail('Should have thrown'); + } catch (_) {} + + await Future.delayed(const Duration(milliseconds: 10)); + expect(conn.isConnected, isFalse); + }); + }); } From fc349c540d6ce7db982d1362d33381c49f6bf8b1 Mon Sep 17 00:00:00 2001 From: ZhuchkaTriplesix Date: Sat, 10 Oct 2026 19:30:24 +0300 Subject: [PATCH 2/2] fix(db): import the mapped exception types in the drivers, await-lint on the socket watcher (#1373) --- lib/core/database/mysql_connection.dart | 1 + lib/core/database/postgres_connection.dart | 1 + lib/core/database/redis_connection.dart | 6 ++++-- 3 files changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/core/database/mysql_connection.dart b/lib/core/database/mysql_connection.dart index dd2d790d..e269fa8d 100644 --- a/lib/core/database/mysql_connection.dart +++ b/lib/core/database/mysql_connection.dart @@ -4,6 +4,7 @@ import 'package:flutter/foundation.dart'; import 'package:mysql_client/mysql_client.dart'; import 'package:querya_desktop/core/database/statement_queue.dart'; import 'package:querya_desktop/core/database/database_error_mapper.dart'; +import 'package:querya_desktop/core/database/querya_database_exception.dart'; import 'package:querya_desktop/core/database/mysql_result_cells.dart'; import 'package:querya_desktop/core/database/table_schema_meta.dart'; import 'package:querya_desktop/core/security/ssh_tunnel_config.dart'; diff --git a/lib/core/database/postgres_connection.dart b/lib/core/database/postgres_connection.dart index 276175a4..f7aed16b 100644 --- a/lib/core/database/postgres_connection.dart +++ b/lib/core/database/postgres_connection.dart @@ -5,6 +5,7 @@ import 'package:flutter/foundation.dart'; import 'package:postgres/postgres.dart'; import 'package:querya_desktop/core/database/statement_queue.dart'; import 'package:querya_desktop/core/database/database_error_mapper.dart'; +import 'package:querya_desktop/core/database/querya_database_exception.dart'; import 'package:querya_desktop/core/security/ssh_tunnel_config.dart'; import 'package:querya_desktop/core/security/ssh_tunnel_manager.dart'; import 'package:querya_desktop/core/storage/connection_secrets_store.dart'; diff --git a/lib/core/database/redis_connection.dart b/lib/core/database/redis_connection.dart index 89d5b13c..e199041a 100644 --- a/lib/core/database/redis_connection.dart +++ b/lib/core/database/redis_connection.dart @@ -3,6 +3,7 @@ import 'dart:io'; import 'package:flutter/foundation.dart'; import 'package:querya_desktop/core/database/database_error_mapper.dart'; +import 'package:querya_desktop/core/database/querya_database_exception.dart'; import 'package:querya_desktop/core/security/ssh_tunnel_config.dart'; import 'package:querya_desktop/core/security/ssh_tunnel_manager.dart'; import 'package:querya_desktop/core/security/ssl_certificate_support.dart'; @@ -242,7 +243,8 @@ class RedisConnection { ); _socket = socket; final activeSocket = socket; - activeSocket.done.then((_) { + // Watches the socket for a drop; the future itself is not awaited. + unawaited(activeSocket.done.then((_) { if (identical(_socket, activeSocket)) { _isConnected = false; } @@ -250,7 +252,7 @@ class RedisConnection { if (identical(_socket, activeSocket)) { _isConnected = false; } - }); + })); final command = await client.connectWithSocket(socket); _command = command; command.setParser(redis.RedisParserBulkBinary());