diff --git a/lib/core/database/redis_connection.dart b/lib/core/database/redis_connection.dart index c1366e1a..f76023a0 100644 --- a/lib/core/database/redis_connection.dart +++ b/lib/core/database/redis_connection.dart @@ -1,3 +1,4 @@ +import 'dart:async'; import 'dart:io'; import 'package:flutter/foundation.dart'; @@ -29,9 +30,25 @@ class RedisConnection { String? connectionString, this.sshConfig, this.sshSecrets, - }) : _password = password, + Duration? connectTimeout, + }) : _connectTimeout = connectTimeout, + _password = password, _connectionString = connectionString; + /// How long the TCP connect, the TLS handshake and `AUTH` / `PING` may take, + /// each. The same 10 s as PostgreSQL and MySQL (#1307). [Duration.zero] means + /// no limit. + /// + /// A test seam: widget tests that leave a connect pending when they end + /// (a view opening a real connection to a closed port) would fail on the + /// timer the limit creates, so `flutter_test_config.dart` sets it to zero. + @visibleForTesting + static Duration defaultConnectTimeout = const Duration(seconds: 10); + + final Duration? _connectTimeout; + + Duration get connectTimeout => _connectTimeout ?? defaultConnectTimeout; + /// Parses [ConnectionRow.connectionString] (`redis://` / `rediss://`) for /// host, port, userinfo, and TLS. Pass [id] `-1` for a sidebar probe that /// must not share the workspace pool id (and skips secrets lookup). @@ -112,7 +129,23 @@ class RedisConnection { _connectionString = null; } - Future connect() async { + /// Single-flight: a caller that arrives while an attempt is in progress waits + /// 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(); + final inFlight = _connecting; + if (inFlight != null) return inFlight; + final attempt = _connectOnce(); + _connecting = attempt; + return attempt.whenComplete(() { + if (identical(_connecting, attempt)) _connecting = null; + }); + } + + Future? _connecting; + + Future _connectOnce() async { try { await _openConnection(); } on RedisConnectionException { @@ -139,10 +172,14 @@ class RedisConnection { } catch (_) {} } - _conn = redis.RedisConnection(); + final client = redis.RedisConnection(); + _conn = client; + // Owned here so that a failed attempt can close it, wherever it failed. + Socket? socket; final sslPaths = extractSslCertificatePathsFromString(effectiveConnectionString); final secure = useSSL || sslPaths.hasAny; + final limit = connectTimeout > Duration.zero ? connectTimeout : null; try { if (sshConfig != null && sshConfig!.enabled) { @@ -168,35 +205,33 @@ class RedisConnection { final effectiveHost = _sshTunnelHandle?.localHost ?? host; final effectivePort = _sshTunnelHandle?.localPort ?? port; - if (secure) { - final context = buildSecurityContext(sslPaths); - final socket = await SecureSocket.connect( - effectiveHost, - effectivePort, - context: context, - ); - _command = await _conn!.connectWithSocket(socket); - } else { - _command = await _conn!.connect(effectiveHost, effectivePort); - } - _command!.setParser(redis.RedisParserBulkBinary()); - if (effectivePassword != null && effectivePassword.isNotEmpty) { - if (username != null && username!.trim().isNotEmpty) { - await _command! - .send_object(['AUTH', username!.trim(), effectivePassword]); - } else { - await _command!.send_object(['AUTH', effectivePassword]); - } - } - final result = await _command!.send_object(['PING']); - if (result == null || result.toString().toUpperCase() != 'PONG') { - try { - await _conn?.close(); - } catch (_) {} - _conn = null; - _command = null; - throw RedisConnectionException('PING failed'); - } + // The socket is opened here, with a timeout, for plain and TLS alike. + socket = secure + ? await SecureSocket.connect( + effectiveHost, + effectivePort, + context: buildSecurityContext(sslPaths), + timeout: limit, + ) + : await Socket.connect( + effectiveHost, + effectivePort, + timeout: limit, + ); + final command = await client.connectWithSocket(socket); + _command = command; + command.setParser(redis.RedisParserBulkBinary()); + // A server that accepts the TCP connection and then says nothing must + // not hold the caller either. + final handshake = _authenticate(command, effectivePassword); + await (limit == null + ? handshake + : handshake.timeout( + limit, + onTimeout: () => throw TimeoutException( + 'Redis did not answer within ${limit.inSeconds} s', + ), + )); _isConnected = true; if (_clientReadOnly) { try { @@ -210,6 +245,11 @@ class RedisConnection { _isConnected = false; _conn = null; _command = 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 { + socket?.destroy(); + } catch (_) {} try { await _sshTunnelHandle?.release(); } catch (_) {} @@ -218,6 +258,22 @@ class RedisConnection { } } + /// `AUTH` when there is a password, then `PING`. + Future _authenticate(redis.Command command, String? password) async { + if (password != null && password.isNotEmpty) { + final user = username?.trim(); + if (user != null && user.isNotEmpty) { + await command.send_object(['AUTH', user, password]); + } else { + await command.send_object(['AUTH', password]); + } + } + final result = await command.send_object(['PING']); + if (result == null || result.toString().toUpperCase() != 'PONG') { + throw RedisConnectionException('PING failed'); + } + } + Future disconnect() async { final wasConnected = _isConnected; _isConnected = false; diff --git a/test/core/database/redis_connect_lifecycle_test.dart b/test/core/database/redis_connect_lifecycle_test.dart new file mode 100644 index 00000000..6c0f110c --- /dev/null +++ b/test/core/database/redis_connect_lifecycle_test.dart @@ -0,0 +1,142 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:querya_desktop/core/database/querya_database_exception.dart'; +import 'package:querya_desktop/core/database/redis_connection.dart'; + +/// #1307: connect timeout, no socket left after a failed AUTH, single-flight. +/// +/// A tiny RESP server: `AUTH ` is checked against [password], `PING` +/// is answered, and a [silent] server accepts the connection and says nothing. +class _FakeRedis { + _FakeRedis._(this._server, {this.password, this.silent = false}) { + _server.listen(_onClient); + } + + static Future<_FakeRedis> start({String? password, bool silent = false}) async { + final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0); + return _FakeRedis._(server, password: password, silent: silent); + } + + final ServerSocket _server; + final String? password; + final bool silent; + + int accepted = 0; + int closedByClient = 0; + final _sockets = []; + + int get port => _server.port; + + void _onClient(Socket socket) { + accepted++; + _sockets.add(socket); + socket.listen( + (data) { + if (silent) return; + final text = utf8.decode(data, allowMalformed: true); + final upper = text.toUpperCase(); + if (upper.contains('AUTH')) { + final ok = password != null && text.contains(password!); + if (ok) _authed.add(socket); + socket.write(ok + ? '+OK\r\n' + : '-WRONGPASS invalid username-password pair or user is ' + 'disabled.\r\n'); + } else if (upper.contains('PING')) { + if (password != null && _needsAuth(socket)) { + socket.write('-NOAUTH Authentication required.\r\n'); + } else { + socket.write('+PONG\r\n'); + } + } + }, + onDone: () { + closedByClient++; + socket.destroy(); + }, + onError: (_) {}, + ); + } + + final _authed = {}; + bool _needsAuth(Socket s) => !_authed.contains(s); + + Future close() async { + for (final s in _sockets) { + s.destroy(); + } + await _server.close(); + } +} + +Future _until(bool Function() condition) async { + for (var i = 0; i < 100 && !condition(); i++) { + await Future.delayed(const Duration(milliseconds: 20)); + } +} + +void main() { + late _FakeRedis server; + + tearDown(() async => server.close()); + + RedisConnection client({ + String? password, + Duration connectTimeout = const Duration(seconds: 2), + }) => + RedisConnection( + id: 0, + name: 'fake', + host: '127.0.0.1', + port: server.port, + password: password, + connectTimeout: connectTimeout, + ); + + test('a wrong password fails and the socket is closed', () async { + server = await _FakeRedis.start(password: 'right'); + final c = client(password: 'wrong'); + + await expectLater(c.connect(), throwsA(isA())); + + expect(c.isConnected, isFalse); + await _until(() => server.closedByClient > 0); + expect(server.accepted, 1); + expect(server.closedByClient, 1); + }); + + test('a server that never answers fails after the timeout and is closed', + () async { + server = await _FakeRedis.start(silent: true); + final c = client(connectTimeout: const Duration(milliseconds: 300)); + + final started = DateTime.now(); + await expectLater( + c.connect(), throwsA(isA())); + + expect(DateTime.now().difference(started).inSeconds, lessThan(5)); + await _until(() => server.closedByClient > 0); + expect(server.closedByClient, 1); + }); + + test('two concurrent connects open one connection', () async { + server = await _FakeRedis.start(); + final c = client(); + + await Future.wait([c.connect(), c.connect()]); + + expect(c.isConnected, isTrue); + expect(server.accepted, 1); + await c.disconnect(); + }); + + test('a failed attempt can be retried', () async { + server = await _FakeRedis.start(silent: true); + final c = client(connectTimeout: const Duration(milliseconds: 200)); + await expectLater(c.connect(), throwsA(isA())); + await expectLater(c.connect(), throwsA(isA())); + expect(server.accepted, 2); + }); +} diff --git a/test/flutter_test_config.dart b/test/flutter_test_config.dart index 17e62959..c8bc6ec8 100644 --- a/test/flutter_test_config.dart +++ b/test/flutter_test_config.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'package:querya_desktop/app/app_wiring.dart'; +import 'package:querya_desktop/core/database/redis_connection.dart'; import 'package:querya_desktop/core/storage/app_data_root.dart'; import 'package:querya_desktop/core/storage/connection_secrets_store.dart'; @@ -13,6 +14,9 @@ Future testExecutable(FutureOr Function() testMain) async { installAppWiring(); ConnectionSecretsStore.backend = testMemorySecrets; testMemorySecrets.clear(); + // A connect limit is a timer; a widget test that ends with a connect pending + // would fail on it (tests that exercise the limit pass their own). + RedisConnection.defaultConnectTimeout = Duration.zero; // Avoid copying the developer's real legacy profile into test temp dirs. AppDataRoot.mockLegacySupportCandidates = const []; await testMain();