Skip to content

fix(db): reflect underlying driver socket state in isConnected and mark session closed on socket drop (#1373) #1379

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ZhuchkaTriplesix merged 2 commits into from
Oct 10, 2026
Merged

fix(db): reflect underlying driver socket state in isConnected and mark session closed on socket drop (#1373) #1379

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
54e48c4
fix(db): reflect underlying driver socket state in isConnected and ma…
ZhuchkaTriplesix Oct 10, 2026
fc349c5
fix(db): import the mapped exception types in the drivers, await-lint…
ZhuchkaTriplesix Oct 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 32 additions & 3 deletions lib/core/database/mysql_connection.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -166,7 +167,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() {
Expand All @@ -186,7 +193,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<void> connect({int connectTimeoutMs = 10000}) {
if (_isConnected && _conn != null) return Future<void>.value();
if (isConnected) return Future<void>.value();
final inFlight = _connecting;
if (inFlight != null) return inFlight;
final attempt = _connectOnce(connectTimeoutMs);
Expand All @@ -201,7 +208,13 @@ class MysqlConnection {
int _generation = 0;

Future<void> _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;
Expand Down Expand Up @@ -296,6 +309,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) {
Expand Down Expand Up @@ -538,6 +556,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;
}
});
}
Expand Down
47 changes: 42 additions & 5 deletions lib/core/database/postgres_connection.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -159,7 +160,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() {
Expand Down Expand Up @@ -218,7 +225,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<void> connect() {
if (_isConnected && _conn != null) return Future<void>.value();
if (isConnected) return Future<void>.value();
final inFlight = _connecting;
if (inFlight != null) return inFlight;
final attempt = _connectOnce();
Expand All @@ -229,7 +236,13 @@ class PostgresConnection {
}

Future<void> _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;
Expand Down Expand Up @@ -440,6 +453,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;
}
});
}
Expand All @@ -449,12 +473,25 @@ class PostgresConnection {
/// call's default timeout while it waits, and could cancel the statement
/// that is running (#1216).
Future<Result> _queued(Future<Result> 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
Expand Down
58 changes: 55 additions & 3 deletions lib/core/database/redis_connection.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -119,10 +120,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;
Expand All @@ -133,7 +148,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<void> connect() {
if (_isConnected && _command != null) return Future<void>.value();
if (isConnected) return Future<void>.value();
final inFlight = _connecting;
if (inFlight != null) return inFlight;
final attempt = _connectOnce();
Expand All @@ -156,7 +171,15 @@ class RedisConnection {
}

Future<void> _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;
Expand Down Expand Up @@ -218,6 +241,18 @@ class RedisConnection {
effectivePort,
timeout: limit,
);
_socket = socket;
final activeSocket = socket;
// Watches the socket for a drop; the future itself is not awaited.
unawaited(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());
Expand Down Expand Up @@ -245,6 +280,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 {
Expand Down Expand Up @@ -278,6 +314,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) {
Expand Down Expand Up @@ -351,7 +392,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.
Expand Down
26 changes: 26 additions & 0 deletions test/core/database/mysql_connection_pool_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)', () {
Expand Down
33 changes: 33 additions & 0 deletions test/core/database/mysql_connection_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
});
});
}
26 changes: 26 additions & 0 deletions test/core/database/postgres_connection_pool_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)', () {
Expand Down
Loading
Loading