From 57a6193efc4a049400a58110ca616bba81b19eb4 Mon Sep 17 00:00:00 2001 From: nickshorebird Date: Fri, 22 May 2026 13:23:01 -0400 Subject: [PATCH] refactor(redis_client): extract RedisConnection from RedisClient (#3796) --- .github/workflows/_shorebird_ci_dart.yaml | 9 + .github/workflows/shorebird_ci.yaml | 1 + packages/redis_client/CHANGELOG.md | 4 + .../redis_client/lib/src/redis_client.dart | 174 ++-------------- .../lib/src/redis_connection.dart | 193 ++++++++++++++++++ packages/redis_client/pubspec.yaml | 2 +- .../test/src/redis_client_test.dart | 14 ++ 7 files changed, 242 insertions(+), 155 deletions(-) create mode 100644 packages/redis_client/lib/src/redis_connection.dart diff --git a/.github/workflows/_shorebird_ci_dart.yaml b/.github/workflows/_shorebird_ci_dart.yaml index 86939649..e71c309d 100644 --- a/.github/workflows/_shorebird_ci_dart.yaml +++ b/.github/workflows/_shorebird_ci_dart.yaml @@ -21,6 +21,10 @@ on: required: false default: "" type: string + needs_redis: + required: false + default: false + type: boolean jobs: ci: @@ -29,6 +33,11 @@ jobs: - uses: actions/checkout@v6 with: submodules: recursive + - name: 🐳 Run Redis + if: inputs.needs_redis + run: | + docker pull redis/redis-stack-server:latest + docker run --name test_redis -d -p 6379:6379 --rm -e REDIS_ARGS="--requirepass password" redis/redis-stack-server:latest - uses: dart-lang/setup-dart@v1 - name: Setup Bloc Tools if: inputs.has_bloc_lint diff --git a/.github/workflows/shorebird_ci.yaml b/.github/workflows/shorebird_ci.yaml index d8aa76ca..0a7be55a 100644 --- a/.github/workflows/shorebird_ci.yaml +++ b/.github/workflows/shorebird_ci.yaml @@ -204,6 +204,7 @@ jobs: has_bloc_lint: false has_unit_tests: true subpackages: "" + needs_redis: true stripe_api: needs: changes diff --git a/packages/redis_client/CHANGELOG.md b/packages/redis_client/CHANGELOG.md index 45f6f722..2d1d8efc 100644 --- a/packages/redis_client/CHANGELOG.md +++ b/packages/redis_client/CHANGELOG.md @@ -1,3 +1,7 @@ +# 0.0.13 + +- refactor: extract `RedisConnection` from `RedisClient` to prepare for opt-in connection pooling + # 0.0.12 - fix: more redis query result type handling fixes diff --git a/packages/redis_client/lib/src/redis_client.dart b/packages/redis_client/lib/src/redis_client.dart index 5c84a018..e5a33b4a 100644 --- a/packages/redis_client/lib/src/redis_client.dart +++ b/packages/redis_client/lib/src/redis_client.dart @@ -6,6 +6,8 @@ import 'package:resp_client/resp_client.dart'; import 'package:resp_client/resp_commands.dart'; import 'package:resp_client/resp_server.dart'; +part 'redis_connection.dart'; + /// {@template redis_exception} /// An exception thrown by the Redis client. /// {@endtemplate} @@ -116,43 +118,18 @@ class RedisClient { RedisSocketOptions socket = const RedisSocketOptions(), RedisCommandOptions command = const RedisCommandOptions(), RedisLogger logger = const _NoopRedisLogger(), - }) : _socketOptions = socket, - _commandOptions = command, - _logger = logger; - - /// The socket options for the Redis server. - final RedisSocketOptions _socketOptions; + }) : _commandOptions = command, + _logger = logger, + _connection = RedisConnection(options: socket, logger: logger); /// The command options for the Redis client. final RedisCommandOptions _commandOptions; - /// The underlying connection to the Redis server. - RespServerConnection? _connection; - /// The logger for the Redis client. final RedisLogger _logger; - /// The underlying client for interacting with the Redis server. - RespClient? _client; - - /// Whether the client has been closed. - var _closed = false; - - /// A completer which completes when the client establishes a connection. - var _connected = Completer(); - - /// Whether the client is connected. - var _isConnected = false; - - /// A completer which completes when the client disconnects. - /// Begins in a completed state since the client is initially disconnected. - var _disconnected = Completer()..complete(); - - /// A future which completes when the client establishes a connection. - Future get _untilConnected => _connected.future; - - /// A future which completes when the client disconnects. - Future get _untilDisconnected => _disconnected.future; + /// The underlying connection to the Redis server. + final RedisConnection _connection; /// The Redis JSON commands. RedisJson get json => RedisJson._(client: this); @@ -265,137 +242,34 @@ class RedisClient { } /// Send a command to the Redis server. - Future execute(List command) async { - return _runWithRetry(() async { - final result = await RespCommandsTier0(_client!).execute(command); - if (result.isError) throw RedisException(result.toString()); - return result.payload; - }, command: command.join(' ')); + Future execute(List command) { + return _runWithRetry( + () => _connection.execute(command), + command: command.join(' '), + ); } /// Establish a connection to the Redis server. /// The delay between connection attempts. - Future connect() async { - if (_closed) throw StateError('RedisClient has been closed.'); - - unawaited(_reconnect(retryAttempts: _socketOptions.retryAttempts)); - - return _untilConnected; - } + Future connect() => _connection.connect(); /// Terminate the connection to the Redis server. - Future disconnect() async { - _logger.info('Disconnecting.'); - await _connection?.close(); - _reset(); - await _untilDisconnected; - _logger.info('Disconnected.'); - } + Future disconnect() => _connection.disconnect(); /// Terminate the connection to the Redis server and close the client. /// After this method is called, the client instance is no longer usable. /// Call this method when you are done using the client and/or wish to /// prevent reconnection attempts. - Future close() { - _logger.info('Closing connection.'); - _closed = true; - return disconnect(); - } - - Future _reconnect({required int retryAttempts}) async { - if (retryAttempts <= 0) { - _connected.completeError( - const SocketException('Connection retry limit exceeded'), - StackTrace.current, - ); - return; - } - - Future onConnectionOpened(RespServerConnection connection) async { - _logger.info('Connection opened.'); - _disconnected = Completer(); - _connection = connection; - _client = RespClient(connection); - if (_socketOptions.password != null) { - _logger.info('Authenticating.'); - final username = _socketOptions.username; - final password = _socketOptions.password!; - await RespCommandsTier0(_client!).execute(['AUTH', username, password]); - } - _isConnected = true; - if (!_connected.isCompleted) _connected.complete(); - _logger.info('Connected.'); - } - - void onConnectionClosed([Object? error, StackTrace? stackTrace]) { - if (error == null) { - _logger.info('Connection closed.'); - } else { - _logger.error( - 'Connection closed with error.', - error: error, - stackTrace: stackTrace, - ); - } - - if (_closed) return; - - final wasConnected = _isConnected; - _isConnected = false; - - final retryInterval = _socketOptions.retryInterval; - final totalAttempts = _socketOptions.retryAttempts; - final remainingAttempts = wasConnected - ? totalAttempts - : retryAttempts - 1; - final attemptsMade = totalAttempts - remainingAttempts; - final attemptInfo = attemptsMade > 0 - ? ' ($attemptsMade/$totalAttempts attempts)' - : ''; - - if (wasConnected) _reset(); - - _logger.info( - 'Reconnecting in ${retryInterval.inMilliseconds}ms$attemptInfo.', - ); - Future.delayed( - retryInterval, - () => _reconnect(retryAttempts: remainingAttempts), - ); - } - - try { - _logger.info('Connecting to ${_socketOptions.connectionUri}.'); - final uri = _socketOptions.connectionUri; - final connection = await connectSocket( - uri.host, - port: uri.port, - timeout: _socketOptions.timeout, - ); - unawaited(onConnectionOpened(connection)); - unawaited( - connection.outputSink.done - .then((_) => onConnectionClosed()) - .catchError(onConnectionClosed), - ); - } on Exception catch (error, stackTrace) { - onConnectionClosed(error, stackTrace); - } - } - - void _reset() { - _connected = Completer(); - _connection = null; - _client = null; - if (!_disconnected.isCompleted) _disconnected.complete(); - } + Future close() => _connection.close(); Future _runWithRetry( Future Function() fn, { required String command, int? remainingAttempts, }) async { - if (_closed) throw StateError('RedisClient has been closed.'); + if (_connection.isClosed) { + throw StateError('RedisClient has been closed.'); + } final totalAttempts = _commandOptions.retryAttempts; remainingAttempts ??= _commandOptions.retryAttempts; @@ -407,10 +281,7 @@ class RedisClient { _logger.debug('Executing "$command"$attemptInfo.'); try { - return await Future.sync(() async { - await _untilConnected; - return fn(); - }).timeout(_commandOptions.timeout); + return await Future.sync(fn).timeout(_commandOptions.timeout); } catch (error, stackTrace) { if (error is RedisException) rethrow; if (remainingAttempts > 0) { @@ -431,7 +302,7 @@ class RedisClient { error: error, stackTrace: stackTrace, ); - await _connection?.close(); + await _connection.recycle(); rethrow; } } @@ -988,11 +859,6 @@ class RedisTDigest { } } -extension on RedisSocketOptions { - /// The connection URI for the Redis server derived from the socket options. - Uri get connectionUri => Uri.parse('redis://$host:$port'); -} - final class _NoopRedisLogger implements RedisLogger { const _NoopRedisLogger(); diff --git a/packages/redis_client/lib/src/redis_connection.dart b/packages/redis_client/lib/src/redis_connection.dart new file mode 100644 index 00000000..f4eb67d3 --- /dev/null +++ b/packages/redis_client/lib/src/redis_connection.dart @@ -0,0 +1,193 @@ +part of 'redis_client.dart'; + +/// {@template redis_connection} +/// A single transport-level connection to a Redis server. +/// +/// Owns the socket lifecycle: opening, authenticating, reconnect-on-drop, +/// and graceful close. Commands sent through [execute] wait for the socket +/// to be ready before being written to the wire. +/// +/// This type is internal to the package today and is not exported. A future +/// release will expose it so callers can pin a connection for transactions +/// and pipelining. +/// {@endtemplate} +class RedisConnection { + /// {@macro redis_connection} + RedisConnection({required this.options, required this.logger}); + + /// The socket-level options used to dial and re-dial the server. + final RedisSocketOptions options; + + /// The logger used for connection-lifecycle events. + final RedisLogger logger; + + /// The underlying connection to the Redis server. + RespServerConnection? _connection; + + /// The underlying RESP client for the active connection. + RespClient? _client; + + /// Whether the connection has been permanently closed. + var _closed = false; + + /// A completer which completes when the socket is established. + var _connected = Completer(); + + /// Whether the socket is currently established. + var _isConnected = false; + + /// A completer which completes when the socket is torn down. + /// Begins in a completed state since the connection is initially down. + var _disconnected = Completer()..complete(); + + /// A future which completes when the socket is established. + Future get _untilConnected => _connected.future; + + /// A future which completes when the socket is torn down. + Future get _untilDisconnected => _disconnected.future; + + /// Whether [close] has been called. After this returns true the connection + /// is no longer usable. + bool get isClosed => _closed; + + /// Open the socket. Returns once the connection is established. + Future connect() async { + if (_closed) throw StateError('RedisClient has been closed.'); + + unawaited(_reconnect(retryAttempts: options.retryAttempts)); + + return _untilConnected; + } + + /// Tear down the current socket without closing the connection logically. + /// The reconnect loop will recreate the socket on subsequent use unless + /// [close] has been called. + Future disconnect() async { + logger.info('Disconnecting.'); + await _connection?.close(); + _reset(); + await _untilDisconnected; + logger.info('Disconnected.'); + } + + /// Permanently close the connection. After this method is called, the + /// instance is not usable. + Future close() { + logger.info('Closing connection.'); + _closed = true; + return disconnect(); + } + + /// Send a single command to the server. Waits for the socket to be ready + /// before writing. Throws [RedisException] on a server-side error reply. + /// Throws [StateError] if the connection has been closed. + Future execute(List command) async { + if (_closed) throw StateError('RedisClient has been closed.'); + await _untilConnected; + final result = await RespCommandsTier0(_client!).execute(command); + if (result.isError) throw RedisException(result.toString()); + return result.payload; + } + + /// Close the underlying socket without closing the connection logically. + /// The reconnect loop will recreate the socket on next [execute]. + /// + /// Used by [RedisClient]'s retry policy to recover from a wedged socket + /// after the retry budget is exhausted. + Future recycle() async { + await _connection?.close(); + } + + Future _reconnect({required int retryAttempts}) async { + if (retryAttempts <= 0) { + _connected.completeError( + const SocketException('Connection retry limit exceeded'), + StackTrace.current, + ); + return; + } + + Future onConnectionOpened(RespServerConnection connection) async { + logger.info('Connection opened.'); + _disconnected = Completer(); + _connection = connection; + _client = RespClient(connection); + if (options.password != null) { + logger.info('Authenticating.'); + final username = options.username; + final password = options.password!; + await RespCommandsTier0(_client!).execute(['AUTH', username, password]); + } + _isConnected = true; + if (!_connected.isCompleted) _connected.complete(); + logger.info('Connected.'); + } + + void onConnectionClosed([Object? error, StackTrace? stackTrace]) { + if (error == null) { + logger.info('Connection closed.'); + } else { + logger.error( + 'Connection closed with error.', + error: error, + stackTrace: stackTrace, + ); + } + + if (_closed) return; + + final wasConnected = _isConnected; + _isConnected = false; + + final retryInterval = options.retryInterval; + final totalAttempts = options.retryAttempts; + final remainingAttempts = wasConnected + ? totalAttempts + : retryAttempts - 1; + final attemptsMade = totalAttempts - remainingAttempts; + final attemptInfo = attemptsMade > 0 + ? ' ($attemptsMade/$totalAttempts attempts)' + : ''; + + if (wasConnected) _reset(); + + logger.info( + 'Reconnecting in ${retryInterval.inMilliseconds}ms$attemptInfo.', + ); + Future.delayed( + retryInterval, + () => _reconnect(retryAttempts: remainingAttempts), + ); + } + + try { + logger.info('Connecting to ${options._connectionUri}.'); + final uri = options._connectionUri; + final connection = await connectSocket( + uri.host, + port: uri.port, + timeout: options.timeout, + ); + unawaited(onConnectionOpened(connection)); + unawaited( + connection.outputSink.done + .then((_) => onConnectionClosed()) + .catchError(onConnectionClosed), + ); + } on Exception catch (error, stackTrace) { + onConnectionClosed(error, stackTrace); + } + } + + void _reset() { + _connected = Completer(); + _connection = null; + _client = null; + if (!_disconnected.isCompleted) _disconnected.complete(); + } +} + +extension on RedisSocketOptions { + /// The connection URI for the Redis server derived from the socket options. + Uri get _connectionUri => Uri.parse('redis://$host:$port'); +} diff --git a/packages/redis_client/pubspec.yaml b/packages/redis_client/pubspec.yaml index 3635447a..d348f41a 100644 --- a/packages/redis_client/pubspec.yaml +++ b/packages/redis_client/pubspec.yaml @@ -1,6 +1,6 @@ name: shorebird_redis_client description: A lightweight Dart client library for communicating with a Redis server. Built by Shorebird. -version: 0.0.12 +version: 0.0.13 homepage: https://shorebird.dev repository: https://github.com/shorebirdtech/shorebird/tree/main/packages/redis_client topics: [redis, cache, shorebird] diff --git a/packages/redis_client/test/src/redis_client_test.dart b/packages/redis_client/test/src/redis_client_test.dart index 01c5b7c7..cbef27e7 100644 --- a/packages/redis_client/test/src/redis_client_test.dart +++ b/packages/redis_client/test/src/redis_client_test.dart @@ -108,6 +108,20 @@ void main() { ), ); }); + + test('execute throws StateError when closed', () async { + await client.close(); + await expectLater( + client.execute(['PING']), + throwsA( + isA().having( + (e) => e.message, + 'message', + 'RedisClient has been closed.', + ), + ), + ); + }); }); group('disconnect', () {