fix(redis): connect timeout, no socket left after a failed AUTH, single-flight connect (#1307)

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<void> 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<void> connect() {
+    if (_isConnected && _command != null) return Future<void>.value();
+    final inFlight = _connecting;
+    if (inFlight != null) return inFlight;
+    final attempt = _connectOnce();
+    _connecting = attempt;
+    return attempt.whenComplete(() {
+      if (identical(_connecting, attempt)) _connecting = null;
+    });
+  }
+
+  Future<void>? _connecting;
+
+  Future<void> _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<void> _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<void> disconnect() async {
     final wasConnected = _isConnected;
     _isConnected = false;

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 <password>` 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 = <Socket>[];
+
+  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 = <Socket>{};
+  bool _needsAuth(Socket s) => !_authed.contains(s);
+
+  Future<void> close() async {
+    for (final s in _sockets) {
+      s.destroy();
+    }
+    await _server.close();
+  }
+}
+
+Future<void> _until(bool Function() condition) async {
+  for (var i = 0; i < 100 && !condition(); i++) {
+    await Future<void>.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<AuthFailedException>()));
+
+    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<ConnectionTimeoutException>()));
+
+    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<ConnectionTimeoutException>()));
+    await expectLater(c.connect(), throwsA(isA<ConnectionTimeoutException>()));
+    expect(server.accepted, 2);
+  });
+}

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<void> testExecutable(FutureOr<void> 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();