From 1abd7d4c79348b8c9b8f3a645dd64b10da32f1c3 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 13:32:08 +0200 Subject: [PATCH 01/20] feat(llc)!: bound and authenticate a connection attempt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `StreamWebSocketClient` treated opening the socket as the end of connecting: it called `onConnectionEstablished`, discarded whatever that returned, and waited indefinitely for a health check to arrive. Four consequences, all reachable in the guest flow that motivated this. `options` becomes `optionsBuilder`, called once per attempt. The options carry values that change over a client's lifetime — the auth type a connection needs depends on the token it will present, and a client that switches users presents a different one — so a single instance built at construction time describes only the first attempt. `onConnectionEstablished` becomes `onAuthenticate`, which is what it is called for and when: the socket is open, the state is `Authenticating`, and the connection is not usable until credentials have been sent. It is now a `WebSocketAuthenticator` — handed a `WsSender` and returning a `Result` — so a failure to send them is observed rather than dropped. A `void Function()` could not report one, and silently accepted an `async` callback whose future was then discarded. On failure the connection is closed with the new `AuthenticationFailed` source, carrying the cause, instead of being left waiting for a reply that cannot come. The sender exists because the authenticator runs while the connection is still being established, so it cannot be handed the client itself. `WebSocketOptions.connectTimeout` was declared and never read. It now bounds the whole attempt rather than just opening the socket, since an attempt that opens but never receives its first health check is exactly the one that hangs — and nothing else watches `Authenticating`. Abandoning it reports the new `ConnectTimeout` source. The field is no longer nullable: "the platform default" was never consulted, so `null` meant no timeout at all, and it now defaults to `WebSocketOptions.defaultConnectTimeout`. Neither new source enables automatic reconnection. A handshake that never completes and credentials the server rejected both fail the same way on a retry, unlike an unhealthy connection, which was established once and may be again. Fixes a health check arriving while disconnecting being treated as one arriving on a live connection: it set the state back to `Connected`, which replaced the `Disconnecting` source. A deliberate `UserInitiated` disconnect could therefore close as `ServerInitiated` and be automatically reconnected — the opposite of what the caller asked for. Pongs are now ignored once the connection is on its way down. Adds `ConnectUserDetailsRequest.fromUser`, since an authenticator builds its auth frame from the client's `User` and every product was mapping the same four fields by hand. `role` and `teams` are deliberately left out: the server assigns both and ignores them from a client. `name` comes from `originalName`, so a user with no name does not have their id sent as one. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 9 + .../user/connect_user_details_request.dart | 14 + .../ws/client/engine/web_socket_options.dart | 13 +- .../ws/client/stream_web_socket_client.dart | 91 ++++- .../client/web_socket_connection_state.dart | 40 ++ packages/stream_core/pubspec.yaml | 1 + .../connect_user_details_request_test.dart | 59 +++ .../client/stream_web_socket_client_test.dart | 377 ++++++++++++++++++ .../web_socket_connection_state_test.dart | 53 +++ 9 files changed, 639 insertions(+), 18 deletions(-) create mode 100644 packages/stream_core/test/user/connect_user_details_request_test.dart create mode 100644 packages/stream_core/test/ws/client/stream_web_socket_client_test.dart diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index baeeb305..0db4bf92 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -4,6 +4,9 @@ - Removed the `userId` parameter from `UserToken.anonymous`, anonymous tokens always use `User.anonymousUserId` - Removed the `TokenManager.tokenProvider` setter, use `setTokenProvider` instead +- `StreamWebSocketClient` now takes an `optionsBuilder` instead of `options`, and calls it for every connection attempt +- Renamed `StreamWebSocketClient.onConnectionEstablished` to `onAuthenticate`, which is what it is called for and when +- `StreamWebSocketClient.onAuthenticate` is now a `WebSocketAuthenticator`: it is handed a `WsSender` and returns a `Result`, so a failure to authenticate can be observed - `TokenManager.userId` is now nullable, and is `null` until an identity is configured ### ✨ Features @@ -14,6 +17,11 @@ - Added `User.anonymousUserId`, the id every anonymous user has - Added `TokenManager.unconfigured`, for a client that exists before its user does - Added `TokenManager.reset`, which drops the configured identity and its cached token +- Added `DisconnectionSource.connectTimeout`, reported when a connection attempt is abandoned before the connection is established +- Added `DisconnectionSource.authenticationFailed`, reported with its cause when a connection opens but cannot be authenticated +- `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout` +- Added `WsSender`, the send capability handed to a `WebSocketAuthenticator` +- Added `ConnectUserDetailsRequest.fromUser`, which builds the details a client may send from a `User` - Added `teams` field to `User` class ### 🐛 Bug Fixes @@ -21,6 +29,7 @@ - Fixed `TokenManager.getToken()` contacting the `TokenProvider` on every call instead of returning the cached token - Fixed `DynamicTokenProvider` accepting a token issued for a different user than the one requested - Fixed `TokenManager` caching a token that finished loading after `expireToken` or `setTokenProvider` had invalidated it +- Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect ### 🔄 Changed diff --git a/packages/stream_core/lib/src/user/connect_user_details_request.dart b/packages/stream_core/lib/src/user/connect_user_details_request.dart index 553ba9d2..fea5a7ef 100644 --- a/packages/stream_core/lib/src/user/connect_user_details_request.dart +++ b/packages/stream_core/lib/src/user/connect_user_details_request.dart @@ -1,5 +1,7 @@ import 'package:json_annotation/json_annotation.dart'; +import 'user.dart'; + part 'connect_user_details_request.g.dart'; @JsonSerializable(createFactory: false) @@ -13,6 +15,18 @@ class ConnectUserDetailsRequest { this.custom, }); + factory ConnectUserDetailsRequest.fromUser( + User user, { + bool includeDetails = true, + }) { + return ConnectUserDetailsRequest( + id: user.id, + name: includeDetails ? user.originalName : null, + image: includeDetails ? user.image : null, + custom: includeDetails ? user.custom : null, + ); + } + final String id; final String? image; final bool? invisible; diff --git a/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart b/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart index 0d2f86c0..f4295085 100644 --- a/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart +++ b/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart @@ -22,7 +22,7 @@ class WebSocketOptions { /// Creates a new instance of [WebSocketOptions]. const WebSocketOptions({ required this.url, - this.connectTimeout, + this.connectTimeout = defaultConnectTimeout, this.protocols, this.queryParameters, }); @@ -35,9 +35,14 @@ class WebSocketOptions { /// Maximum time allowed for establishing the WebSocket connection. /// - /// When specified, the connection attempt will timeout if not completed - /// within this duration. If `null`, uses the platform default timeout. - final Duration? connectTimeout; + /// Covers the whole attempt, not just opening the socket: a connection that + /// opens but is never established is abandoned once this elapses. + /// + /// Defaults to [defaultConnectTimeout]. + final Duration connectTimeout; + + /// The [connectTimeout] used when none is given. + static const defaultConnectTimeout = Duration(seconds: 15); /// WebSocket sub-protocols to negotiate during the handshake. /// diff --git a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart index 3e6c4ec4..8a306dff 100644 --- a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart +++ b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart @@ -18,6 +18,32 @@ WsRequest _defaultPingRequestBuilder([HealthCheckInfo? info]) { return HealthCheckPingEvent(connectionId: info?.connectionId); } +/// A function that builds the options for a connection attempt. +/// +/// Called once per attempt, so the options may carry values that change over the +/// client's lifetime. +/// +/// Returns the [WebSocketOptions] to open the connection with. +typedef WebSocketOptionsBuilder = WebSocketOptions Function(); + +/// A function that sends a request over a connection that is not usable yet. +/// +/// Handed to a [WebSocketAuthenticator], which runs while the connection is +/// still being established and so cannot be given the client itself. +/// +/// Returns a [Result] indicating whether the request was sent. +typedef WsSender = Result Function(WsRequest request); + +/// A function that authenticates a newly opened connection. +/// +/// Called once the socket is open, while the state is [Authenticating]. Sending +/// the credentials the server expects is this function's job. +/// +/// Returns a [Future] that completes when the credentials have been sent, and +/// fails if they could not be — in which case the connection is closed with +/// [AuthenticationFailed] rather than left waiting for a reply that never comes. +typedef WebSocketAuthenticator = Future> Function(WsSender send); + /// A WebSocket client with connection management and event handling. /// /// The primary interface for WebSocket connections in the Stream Core SDK that provides @@ -31,11 +57,9 @@ WsRequest _defaultPingRequestBuilder([HealthCheckInfo? info]) { /// ## Example /// ```dart /// final client = StreamWebSocketClient( -/// options: WebSocketOptions(url: 'wss://api.example.com'), +/// optionsBuilder: () => WebSocketOptions(url: 'wss://api.example.com'), /// messageCodec: JsonMessageCodec(), -/// onConnectionEstablished: () { -/// client.send(AuthRequest(token: authToken)); -/// }, +/// onAuthenticate: (send) async => send(AuthRequest(token: authToken)), /// ); /// /// await client.connect(); @@ -43,8 +67,8 @@ WsRequest _defaultPingRequestBuilder([HealthCheckInfo? info]) { class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineListener { /// Creates a new instance of [StreamWebSocketClient]. StreamWebSocketClient({ - required this.options, - this.onConnectionEstablished, + required this.optionsBuilder, + this.onAuthenticate, WebSocketProvider? wsProvider, this.pingRequestBuilder = _defaultPingRequestBuilder, required WebSocketMessageCodec messageCodec, @@ -58,18 +82,34 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL ); } - /// The WebSocket connection options including URL and configuration. - final WebSocketOptions options; + /// The function used to build the connection options for each attempt. + final WebSocketOptionsBuilder optionsBuilder; /// The function used to build ping requests for health checks. final PingRequestBuilder pingRequestBuilder; - /// Called when the WebSocket connection is established and ready for authentication. - final void Function()? onConnectionEstablished; + /// The function used to authenticate a newly opened connection. + final WebSocketAuthenticator? onAuthenticate; late final StreamWebSocketEngine _engine; late final _healthMonitor = WebSocketHealthMonitor(listener: this); + // Bounds a connection attempt that never reaches 'connected'. + Timer? _connectTimeoutTimer; + + void _startConnectTimeout(Duration timeout) { + _connectTimeoutTimer?.cancel(); + _connectTimeoutTimer = Timer(timeout, () { + const source = DisconnectionSource.connectTimeout(); + unawaited(disconnect(source: source)); + }); + } + + void _cancelConnectTimeout() { + _connectTimeoutTimer?.cancel(); + _connectTimeoutTimer = null; + } + /// The event emitter for WebSocket events. /// /// Use this to listen to incoming WebSocket events with type-safe event handling. @@ -116,7 +156,11 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL // Update the connection state to 'connecting'. _connectionState = const WebSocketConnectionState.connecting(); - // Open the connection using the engine. + // Open the connection using the engine, with options built for this attempt. + final options = optionsBuilder.call(); + + // Time the whole handshake: nothing else watches 'authenticating'. + _startConnectTimeout(options.connectTimeout); final result = await _engine.open(options); // If some failure occurs, disconnect and rethrow the error. @@ -136,6 +180,9 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL // If the connection is already disconnected, do nothing. if (connectionState.value is Disconnected) return; + // Stop the timeout from firing later and replacing this source. + _cancelConnectTimeout(); + // Update the connection state to 'disconnecting'. _connectionState = WebSocketConnectionState.disconnecting(source: source); @@ -148,13 +195,24 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL // Update the connection state to 'authenticating'. _connectionState = const WebSocketConnectionState.authenticating(); - // Notify that the connection has been established and we are ready - // to authenticate. - onConnectionEstablished?.call(); + // The socket is open, so authenticate before the connection is usable. + unawaited(_authenticate()); + } + + Future _authenticate() async { + final result = await onAuthenticate?.call(send); + + // Close the connection rather than wait for a reply that cannot come. + if (result?.exceptionOrNull() case final error?) { + final source = DisconnectionSource.authenticationFailed(error: error); + return disconnect(source: source); + } } @override void onClose([int? closeCode, String? closeReason]) { + _cancelConnectTimeout(); + final source = switch (connectionState.value) { // If we were already disconnecting, keep the caller-provided source. Disconnecting(:final source) => source, @@ -217,6 +275,11 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL void _handleHealthCheckEvent(WsEvent event, HealthCheckInfo info) { print('WebSocketClient: Health check pong received: $info'); + // Ignore a pong that arrives once the connection is on its way down. + if (connectionState.value case Disconnecting() || Disconnected()) return; + + _cancelConnectTimeout(); + // Update the connection state with health check info. _connectionState = WebSocketConnectionState.connected(healthCheck: info); diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index 87a6cac2..586c8d1f 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -115,6 +115,8 @@ sealed class WebSocketConnectionState extends Equatable { UnHealthyConnection() => true, SystemInitiated() => true, UserInitiated() => false, + ConnectTimeout() => false, + AuthenticationFailed() => false, }, _ => false, // No automatic reconnection for other states }; @@ -252,6 +254,18 @@ sealed class DisconnectionSource extends Equatable { /// typically when ping requests do not receive pong responses. const factory DisconnectionSource.unHealthyConnection() = UnHealthyConnection; + /// Creates a [ConnectTimeout] disconnection source. + /// + /// Indicates that the connection never became usable within the allotted + /// time, so it was abandoned before it was ever established. + const factory DisconnectionSource.connectTimeout() = ConnectTimeout; + + /// Creates an [AuthenticationFailed] disconnection source. + /// + /// Indicates that the connection opened but could not be authenticated, so it + /// was closed without ever being usable. + const factory DisconnectionSource.authenticationFailed({Object? error}) = AuthenticationFailed; + /// A human-readable description of the disconnection source. /// /// Provides a descriptive string that explains why the connection was closed. @@ -264,6 +278,8 @@ sealed class DisconnectionSource extends Equatable { ServerInitiated() => 'Server initiated disconnection', SystemInitiated() => 'System initiated disconnection', UnHealthyConnection() => 'Unhealthy connection (no pong received)', + ConnectTimeout() => 'Timed out before the connection was established', + AuthenticationFailed() => 'Authentication failed', }; } @@ -320,3 +336,27 @@ final class UnHealthyConnection extends DisconnectionSource { /// Creates an [UnHealthyConnection] disconnection source. const UnHealthyConnection(); } + +/// A disconnection caused by the connection not becoming usable in time. +/// +/// This source indicates that the connection was abandoned while it was still +/// being established, so it was never usable. +final class ConnectTimeout extends DisconnectionSource { + /// Creates a [ConnectTimeout] disconnection source. + const ConnectTimeout(); +} + +/// A disconnection caused by the connection failing to authenticate. +/// +/// This source indicates that the socket opened but authentication did not +/// complete, so the connection was never usable. +final class AuthenticationFailed extends DisconnectionSource { + /// Creates an [AuthenticationFailed] disconnection source. + const AuthenticationFailed({this.error}); + + /// The error that prevented the connection from authenticating. + final Object? error; + + @override + List get props => [error]; +} diff --git a/packages/stream_core/pubspec.yaml b/packages/stream_core/pubspec.yaml index c28d1a61..7fa4b4fa 100644 --- a/packages/stream_core/pubspec.yaml +++ b/packages/stream_core/pubspec.yaml @@ -37,6 +37,7 @@ dependencies: dev_dependencies: build_runner: ^2.10.5 + fake_async: ^1.3.3 json_serializable: ^6.9.5 mocktail: ^1.0.4 test: ^1.26.2 diff --git a/packages/stream_core/test/user/connect_user_details_request_test.dart b/packages/stream_core/test/user/connect_user_details_request_test.dart new file mode 100644 index 00000000..2ff4858a --- /dev/null +++ b/packages/stream_core/test/user/connect_user_details_request_test.dart @@ -0,0 +1,59 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +void main() { + group('ConnectUserDetailsRequest.fromUser', () { + test('carries the fields the server accepts from a client', () { + const user = User( + id: 'user-1', + name: 'Bob', + image: 'https://example.com/bob.png', + custom: {'plan': 'pro'}, + ); + + final details = ConnectUserDetailsRequest.fromUser(user); + + expect(details.id, 'user-1'); + expect(details.name, 'Bob'); + expect(details.image, 'https://example.com/bob.png'); + expect(details.custom, {'plan': 'pro'}); + }); + + test('leaves out the fields the server decides itself', () { + const user = User(id: 'user-1', role: 'admin', teams: ['red']); + + final json = ConnectUserDetailsRequest.fromUser(user).toJson(); + + // Sending either is pointless: the server ignores both from a client. + expect(json, isNot(contains('role'))); + expect(json, isNot(contains('teams'))); + }); + + test('sends the id alone when details are excluded', () { + const user = User( + id: 'user-1', + name: 'Bob', + image: 'https://example.com/bob.png', + custom: {'plan': 'pro'}, + ); + + final details = ConnectUserDetailsRequest.fromUser(user, includeDetails: false); + + expect(details.id, 'user-1'); + expect(details.name, isNull); + expect(details.image, isNull); + expect(details.custom, isNull); + }); + + test('reports the name the user was created with, not the id fallback', () { + // `User.name` falls back to the id; the wire form must not, or a user + // with no name would be given the id as one. + const user = User(id: 'user-1'); + + final details = ConnectUserDetailsRequest.fromUser(user); + + expect(user.name, 'user-1'); + expect(details.name, isNull); + }); + }); +} diff --git a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart new file mode 100644 index 00000000..e458a599 --- /dev/null +++ b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart @@ -0,0 +1,377 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; +import 'package:web_socket_channel/web_socket_channel.dart'; + +class _MockWebSocketChannel extends Mock implements WebSocketChannel {} + +class _MockWebSocketSink extends Mock implements WebSocketSink {} + +/// A codec that is never exercised: these tests drive the client through its +/// engine listener callbacks rather than through encoded frames. +class _NoopCodec implements WebSocketMessageCodec { + const _NoopCodec(); + + @override + Object encode(WsRequest message) => ''; + + @override + WsEvent decode(Object message) => const _HealthCheckEvent(); +} + +final class _HealthCheckEvent extends WsEvent { + const _HealthCheckEvent({this.connectionId = 'connection-id'}); + + final String? connectionId; + + @override + HealthCheckInfo? get healthCheckInfo { + return HealthCheckInfo(connectionId: connectionId); + } +} + +final class _PingRequest extends WsRequest { + const _PingRequest(); + + @override + Map toJson() => const {}; + + @override + List get props => const []; +} + +/// Builds a client whose socket opens successfully but sends nothing, so the +/// handshake only progresses when a test drives it. +({ + StreamWebSocketClient client, + StreamController incoming, + int Function() optionsBuilt, + WebSocketSink sink, +}) +_client({ + Duration connectTimeout = WebSocketOptions.defaultConnectTimeout, + WebSocketAuthenticator? onAuthenticate, +}) { + final incoming = StreamController.broadcast(); + addTearDown(incoming.close); + + final channel = _MockWebSocketChannel(); + when(() => channel.ready).thenAnswer((_) async {}); + when(() => channel.stream).thenAnswer((_) => incoming.stream); + final sink = _MockWebSocketSink(); + when(() => channel.sink).thenReturn(sink); + + var built = 0; + final client = StreamWebSocketClient( + optionsBuilder: () { + built++; + return WebSocketOptions( + url: 'wss://example.com', + connectTimeout: connectTimeout, + ); + }, + onAuthenticate: onAuthenticate, + wsProvider: (_) => channel, + pingRequestBuilder: ([_]) => const _PingRequest(), + messageCodec: const _NoopCodec(), + ); + + return (client: client, incoming: incoming, optionsBuilt: () => built, sink: sink); +} + +void main() { + group('StreamWebSocketClient.optionsBuilder', () { + test('is called for every connection attempt, not once per client', () async { + final (:client, :incoming, :optionsBuilt, sink: _) = _client(); + + await client.connect(); + expect(optionsBuilt(), 1); + + await client.disconnect(); + client.onClose(); + + await client.connect(); + expect(optionsBuilt(), 2); + }); + }); + + group('StreamWebSocketClient.onAuthenticate', () { + test('is called once the socket is open, while authenticating', () async { + WebSocketConnectionState? stateWhenCalled; + late StreamWebSocketClient client; + final built = _client( + onAuthenticate: (_) async { + stateWhenCalled = client.connectionState.value; + return const Result.success(null); + }, + ); + client = built.client; + + await client.connect(); + await pumpEventQueue(); + + expect(stateWhenCalled, isA()); + }); + + test('is called once per connection attempt', () async { + var calls = 0; + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async { + calls++; + return const Result.success(null); + }, + ); + + await client.connect(); + await pumpEventQueue(); + expect(calls, 1); + + await client.disconnect(); + client.onClose(); + + await client.connect(); + await pumpEventQueue(); + expect(calls, 2); + }); + + test('is handed a sender that puts the request on the socket', () async { + Result? sent; + final (:client, incoming: _, optionsBuilt: _, :sink) = _client( + onAuthenticate: (send) async => sent = send(const _PingRequest()), + ); + + await client.connect(); + await pumpEventQueue(); + + // The sender is only useful if it reaches the socket: an authenticator + // that cannot send has nothing to report but failure. + expect(sent, isA>()); + verify(() => sink.add(any())).called(1); + }); + + test('leaves the connection authenticating when it succeeds', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async => const Result.success(null), + ); + + await client.connect(); + await pumpEventQueue(); + + // Sending the credentials does not establish the connection; the server + // answering does. + expect(client.connectionState.value, isA()); + + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + }); + + test('leaves the connection authenticating when there is no authenticator', () async { + // A socket that has nothing to send before it is usable, such as one whose + // protocol authenticates elsewhere. + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + await client.connect(); + await pumpEventQueue(); + + expect(client.connectionState.value, isA()); + + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + }); + }); + + group('StreamWebSocketClient authentication failure', () { + test('closes the connection instead of waiting for a reply', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async => Result.failure(StateError('no token')), + ); + + await client.connect(); + await pumpEventQueue(); + + final state = client.connectionState.value; + expect( + state, + isA().having( + (it) => it.source, + 'source', + isA().having((it) => it.error, 'error', isStateError), + ), + ); + }); + + test('is not retried, since the same credentials would fail again', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async => Result.failure(StateError('no token')), + ); + + await client.connect(); + await pumpEventQueue(); + client.onClose(); + + final state = client.connectionState.value; + expect(state, isA()); + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + }); + + group('StreamWebSocketClient connect timeout', () { + test('abandons an attempt that never becomes connected', () { + fakeAsync((async) { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + + // The socket opened, so the client is authenticating with nothing else + // watching it. + expect(client.connectionState.value, isA()); + + // Still waiting a tick before the timeout is due. + async.elapse(WebSocketOptions.defaultConnectTimeout - const Duration(seconds: 1)); + expect(client.connectionState.value, isA()); + + async.elapse(const Duration(seconds: 1)); + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + + test('abandons an attempt whose authenticator never returns', () { + fakeAsync((async) { + // The realistic hang: an authenticator awaiting something that never + // resolves. Nothing else watches 'authenticating', so only this fires. + final (:client, incoming: _, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) => Completer>().future, + ); + + client.connect().ignore(); + async.flushMicrotasks(); + expect(client.connectionState.value, isA()); + + async.elapse(WebSocketOptions.defaultConnectTimeout); + + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + + test('is armed again for a later attempt', () { + fakeAsync((async) { + final (:client, incoming: _, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + async.elapse(WebSocketOptions.defaultConnectTimeout); + client.onClose(); + expect(client.connectionState.value, isA()); + + client.connect().ignore(); + async.flushMicrotasks(); + async.elapse(WebSocketOptions.defaultConnectTimeout); + + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + + test('honours a timeout given in the options', () { + fakeAsync((async) { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + connectTimeout: const Duration(seconds: 2), + ); + + client.connect().ignore(); + async.flushMicrotasks(); + + async.elapse(const Duration(seconds: 2)); + + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + + test('does not fire once the connection is established', () { + fakeAsync((async) { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + + // Past when the timeout would have fired, but before the health + // monitor's first ping is due. + async.elapse(WebSocketOptions.defaultConnectTimeout + const Duration(seconds: 1)); + + expect(client.connectionState.value, isA()); + }); + }); + + test('does not replace the source of a disconnect that came first', () { + fakeAsync((async) { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.disconnect().ignore(); + async.flushMicrotasks(); + + async.elapse(WebSocketOptions.defaultConnectTimeout * 2); + + // The timeout would otherwise report this deliberate disconnect as a + // timed-out attempt, which reconnects differently. + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + }); + + group('StreamWebSocketClient health check while disconnecting', () { + test('does not report the connection as established again', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + await client.connect(); + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + + await client.disconnect(); + expect(client.connectionState.value, isA()); + + // Arrives before the socket finished closing. + client.onMessage(const _HealthCheckEvent(connectionId: 'late')); + + expect(client.connectionState.value, isA()); + }); + + test('leaves the disconnection source intact once the socket closes', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + await client.connect(); + client.onMessage(const _HealthCheckEvent()); + await client.disconnect(); + client.onMessage(const _HealthCheckEvent(connectionId: 'late')); + client.onClose(); + + // Without the guard the late health check moves the state back to + // connected, and `onClose` then reports a server-initiated disconnect, + // which is eligible for an automatic reconnect. + final state = client.connectionState.value; + expect(state, isA().having((it) => it.source, 'source', isA())); + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + }); +} diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index c8d9a6b9..59d26f43 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -40,5 +40,58 @@ void main() { expect(state.isAutomaticReconnectionEnabled, isTrue); }); + + test( + 'is disabled when a connection attempt timed out, so a handshake that ' + 'never completes is not retried forever', + () { + const state = Disconnected(source: ConnectTimeout()); + + expect(state.isAutomaticReconnectionEnabled, isFalse); + }, + ); + + test( + 'is disabled when a connection could not be authenticated, since the ' + 'same credentials would fail again', + () { + const state = Disconnected(source: AuthenticationFailed(error: 'no token')); + + expect(state.isAutomaticReconnectionEnabled, isFalse); + }, + ); + + test('is enabled when a connected socket stops answering health checks', () { + const state = Disconnected(source: UnHealthyConnection()); + + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); + }); + + group('DisconnectionSource.closeReason', () { + test('reads differently for every source', () { + const sources = [ + UserInitiated(), + ServerInitiated(), + SystemInitiated(), + UnHealthyConnection(), + ConnectTimeout(), + AuthenticationFailed(error: 'no token'), + ]; + + final reasons = sources.map((it) => it.closeReason).toSet(); + + // A shared reason would report two different outcomes identically. + expect(reasons, hasLength(sources.length)); + }); + }); + + group('WebSocketOptions.defaultConnectTimeout', () { + test('is the timeout used when the options do not say', () { + const options = WebSocketOptions(url: 'wss://example.com'); + + expect(options.connectTimeout, WebSocketOptions.defaultConnectTimeout); + expect(WebSocketOptions.defaultConnectTimeout, const Duration(seconds: 15)); + }); }); } From 64c7af78dce040650ea35cc210d6f670d4caae7a Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 17:18:28 +0200 Subject: [PATCH 02/20] fix(llc)!: return the result's own type from Result's failure-side helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `getOrElse`, `getOrDefault`, `recover` and `recoverCatching` each declared a type parameter of their own and then cast the success value into it — `Success(:final data) => data as R`. Nothing constrains `T` to be a subtype of `R`, so the cast is unsound: with a callback that only throws, `R` infers as `Never` and a *successful* result fails with a type error on the path that has nothing wrong with it. getOrElse THREW on a Success: type '(String, int)' is not a subtype of type 'Never' That makes the natural way to turn a failure into an exception — the shorthand `getOrThrow`'s own doc suggests — unusable. Dart cannot express Kotlin's `T : R` bound, so the type parameter goes and the helpers return `T`. Widening is still available through `fold`, which takes its return type honestly. Source-breaking for callers that relied on widening; none exist in this repo or in `stream-feeds-flutter`. Adds the first tests for `Result`, four of which pin the success path of each helper against a throwing callback. Co-Authored-By: Claude Opus 5 (1M context) --- .../stream_core/lib/src/utils/result.dart | 28 ++--- .../stream_core/test/utils/result_test.dart | 110 ++++++++++++++++++ 2 files changed, 124 insertions(+), 14 deletions(-) create mode 100644 packages/stream_core/test/utils/result_test.dart diff --git a/packages/stream_core/lib/src/utils/result.dart b/packages/stream_core/lib/src/utils/result.dart index 07c66ca7..e29afdce 100644 --- a/packages/stream_core/lib/src/utils/result.dart +++ b/packages/stream_core/lib/src/utils/result.dart @@ -58,8 +58,8 @@ extension PatternMatching on Result { /// Returns the encapsulated value if this instance represents [Success] or `null` /// if it is [Failure]. /// - /// This function is a shorthand for `getOrElse(() => null)` or - /// `fold(onSuccess: (it) => it, onFailure: (_) => null)`. + /// This function is a shorthand for + /// `fold(onSuccess: (it) => it, onFailure: (_, _) => null)`. T? getOrNull() { return switch (this) { Success(:final data) => data, @@ -90,7 +90,7 @@ extension PatternMatching on Result { /// Returns the encapsulated value if this instance represents [Success] or throws the encapsulated error /// if it is [Failure]. /// - /// This function is a shorthand for `getOrElse((error) => throw error)`. + /// This function is a shorthand for `getOrElse((error, _) => throw error)`. T getOrThrow() { return switch (this) { Success(:final data) => data, @@ -107,9 +107,9 @@ extension PatternMatching on Result { /// Note, that this function rethrows any error thrown by [onFailure] function. /// /// This function is a shorthand for `fold(onSuccess: (it) => it, onFailure: onFailure)`. - R getOrElse(R Function(Object error, StackTrace? stackTrace) onFailure) { + T getOrElse(T Function(Object error, StackTrace? stackTrace) onFailure) { return switch (this) { - Success(:final data) => data as R, + Success(:final data) => data, Failure(:final error, :final stackTrace) => onFailure(error, stackTrace), }; } @@ -117,10 +117,10 @@ extension PatternMatching on Result { /// Returns the encapsulated value if this instance represents [Success] or the /// [defaultValue] if it is [Failure]. /// - /// This function is a shorthand for `getOrElse((_) => defaultValue)`. - R getOrDefault(R defaultValue) { + /// This function is a shorthand for `getOrElse((_, _) => defaultValue)`. + T getOrDefault(T defaultValue) { return switch (this) { - Success(:final data) => data as R, + Success(:final data) => data, Failure() => defaultValue, }; } @@ -180,11 +180,11 @@ extension PatternMatching on Result { /// /// Note, that this function rethrows any error thrown by [transform] function. /// See [recoverCatching] for an alternative that encapsulates errors. - Result recover( - R Function(Object error, StackTrace? stackTrace) transform, + Result recover( + T Function(Object error, StackTrace? stackTrace) transform, ) { return switch (this) { - Success(:final data) => Result.success(data as R), + Success() => this, Failure(:final error, :final stackTrace) => Result.success( transform(error, stackTrace), ), @@ -196,11 +196,11 @@ extension PatternMatching on Result { /// /// This function catches any error thrown by [transform] function and encapsulates it as a failure. /// See [recover] for an alternative that rethrows errors. - Result recoverCatching( - R Function(Object error, StackTrace? stackTrace) transform, + Result recoverCatching( + T Function(Object error, StackTrace? stackTrace) transform, ) { return switch (this) { - Success(:final data) => Result.success(data as R), + Success() => this, Failure(:final error, :final stackTrace) => runSafelySync( () => transform(error, stackTrace), ), diff --git a/packages/stream_core/test/utils/result_test.dart b/packages/stream_core/test/utils/result_test.dart new file mode 100644 index 00000000..fb789b81 --- /dev/null +++ b/packages/stream_core/test/utils/result_test.dart @@ -0,0 +1,110 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +void main() { + group('Result.getOrElse', () { + test('returns the value when the fallback only throws', () { + const result = Result.success(42); + + // The value used to be cast to the fallback's return type, here `Never`. + final value = result.getOrElse((_, _) => throw StateError('unreachable')); + + expect(value, 42); + }); + + test('throws what the fallback throws, so an error can be reworded', () { + final result = Result.failure(Exception('original')); + + expect( + () => result.getOrElse((error, _) => throw StateError('$error')), + throwsA(isA()), + ); + }); + + test('returns what the fallback returns', () { + final result = Result.failure(Exception('original')); + + expect(result.getOrElse((_, _) => 0), 0); + }); + + test('hands the fallback the error and its stack trace', () { + final stackTrace = StackTrace.current; + final error = Exception('original'); + final result = Result.failure(error, stackTrace); + + Object? seenError; + StackTrace? seenStackTrace; + result.getOrElse((error, stackTrace) { + seenError = error; + seenStackTrace = stackTrace; + return 0; + }); + + expect(seenError, error); + expect(seenStackTrace, stackTrace); + }); + }); + + group('Result.getOrDefault', () { + test('returns the value when there is one', () { + const result = Result.success(42); + + expect(result.getOrDefault(0), 42); + }); + + test('returns the default when there is not', () { + final result = Result.failure(Exception('failed')); + + expect(result.getOrDefault(0), 0); + }); + }); + + group('Result.recover', () { + test('keeps the value when the transform only throws', () { + const result = Result.success(42); + + // Same cast as `getOrElse`. + final recovered = result.recover((_, _) => throw StateError('unreachable')); + + expect(recovered.getOrNull(), 42); + }); + + test('turns a failure into the value the transform returns', () { + final result = Result.failure(Exception('failed')); + + expect(result.recover((_, _) => 0).getOrNull(), 0); + }); + + test('rethrows an error from the transform', () { + final result = Result.failure(Exception('failed')); + + expect( + () => result.recover((_, _) => throw StateError('while recovering')), + throwsA(isA()), + ); + }); + }); + + group('Result.recoverCatching', () { + test('keeps the value when the transform only throws', () { + const result = Result.success(42); + + final recovered = result.recoverCatching( + (_, _) => throw StateError('unreachable'), + ); + + expect(recovered.getOrNull(), 42); + }); + + test('reports an error from the transform as a failure', () { + final result = Result.failure(Exception('failed')); + + final recovered = result.recoverCatching( + (_, _) => throw StateError('while recovering'), + ); + + // Unlike `recover`, the error replaces the original rather than escaping. + expect(recovered.exceptionOrNull(), isA()); + }); + }); +} From 705123f5f407d8563223420b2c71860d8cd5516d Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 17:18:28 +0200 Subject: [PATCH 03/20] fix(llc): make a connection going down report why, once, and stay down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five things about closing a connection, found while wiring `stream-feeds-flutter` onto this and in review of #160. `disconnect` returned while the socket was still closing, so a `connect` straight afterwards raced it: the engine's `open` closes any existing socket first, both closes ran to completion, and `onClose` fired twice — the second landing on a state of `Connecting` and reporting `ServerInitiated`, which is reconnect-eligible. One `disconnect(); connect();` pair could therefore end up with a spurious reconnect alongside the connection it just opened. The close is now awaited, which costs a socket flush: the returned future resolves when the close frame has been written, not when the peer replies. A failed close left the client reporting `Disconnecting` for good. The engine reports such a failure as a `Result` and skips notifying its listener, so nothing moved the state on. The connection is unusable either way, so it is now reported closed. `disconnect` no longer replaces the source of a closure already under way. `onError` sets `Disconnecting(ServerInitiated)` without cancelling the connect timer, so the timer could overwrite a reconnectable server error with a `ConnectTimeout`; the same shape turned a timeout into a late `AuthenticationFailed`. Whoever asked first describes why. An authenticator that throws now fails the connection instead of escaping. The `WebSocketAuthenticator` typedef asks for a `Result`, but the one authenticator everyone writes awaits a token — and loading one throws. The error escaped unhandled, since nothing observes that future, and the connection sat in `Authenticating` until the timeout reported a cause it does not carry. `ConnectTimeout` is now eligible for automatic reconnection. A first health check that never arrives is the same failure as one that stops arriving, which `UnHealthyConnection` already retries; making it permanent meant a backend slow to send that first check went from connecting eventually to staying down. `AuthenticationFailed` stays ineligible: it means the client could not produce credentials, not that the server refused them, and it will fail the same way on a retry. Adds `dispose`, so the client can be released rather than only closed — `StreamFeedsClient.dispose` had nothing to call, leaving both emitters open for the life of the process. It closes the connection, stops the health monitor and closes `events` and `connectionState`, and is idempotent through `Disposable`. Reporting a state guards on the emitter being closed rather than on disposal, so a close event arriving from the engine afterwards is ignored instead of thrown into a closed emitter. Co-Authored-By: Claude Opus 5 (1M context) --- .../ws/client/stream_web_socket_client.dart | 71 ++++++-- .../client/web_socket_connection_state.dart | 7 +- .../client/stream_web_socket_client_test.dart | 167 +++++++++++++++++- .../web_socket_connection_state_test.dart | 6 +- 4 files changed, 227 insertions(+), 24 deletions(-) diff --git a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart index 8a306dff..6ed3f3ba 100644 --- a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart +++ b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart @@ -39,9 +39,9 @@ typedef WsSender = Result Function(WsRequest request); /// Called once the socket is open, while the state is [Authenticating]. Sending /// the credentials the server expects is this function's job. /// -/// Returns a [Future] that completes when the credentials have been sent, and -/// fails if they could not be — in which case the connection is closed with -/// [AuthenticationFailed] rather than left waiting for a reply that never comes. +/// Returns a [Result] that fails when the credentials could not be sent, in +/// which case the connection is closed with [AuthenticationFailed] rather than +/// left waiting for a reply that never comes. typedef WebSocketAuthenticator = Future> Function(WsSender send); /// A WebSocket client with connection management and event handling. @@ -64,7 +64,7 @@ typedef WebSocketAuthenticator = Future> Function(WsSender send); /// /// await client.connect(); /// ``` -class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineListener { +class StreamWebSocketClient with Disposable implements WebSocketHealthListener, WebSocketEngineListener { /// Creates a new instance of [StreamWebSocketClient]. StreamWebSocketClient({ required this.optionsBuilder, @@ -120,11 +120,12 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL /// /// Emits state changes as the WebSocket transitions through different connection states. ConnectionStateEmitter get connectionState => _connectionStateEmitter; - late final _connectionStateEmitter = MutableConnectionStateEmitter( - const WebSocketConnectionState.initialized(), - ); + late final _connectionStateEmitter = MutableConnectionStateEmitter(const .initialized()); set _connectionState(WebSocketConnectionState connectionState) { + // Return early if the emitter is closed. + if (_connectionStateEmitter.isClosed) return; + // Return early if the state hasn't changed. if (_connectionStateEmitter.value == connectionState) return; @@ -144,9 +145,15 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL /// /// The connection state can be monitored through [connectionState] for real-time updates. /// If the connection is already established or in progress, this method returns immediately. + /// It also does nothing once [dispose] has been called. /// - /// Returns a [Future] that completes when the connection attempt finishes. + /// Returns a [Future] that completes once the socket is open — before the + /// connection is authenticated, and well before it is [Connected]. Watch + /// [connectionState] to know when it is usable. Future connect() async { + assert(!isDisposed, 'Cannot connect a disposed StreamWebSocketClient'); + if (isDisposed) return; + // If the connection is already established or in the process of connecting, // do not initiate a new connection. if (connectionState.value is Connecting) return; @@ -177,8 +184,11 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL CloseCode closeCode = CloseCode.normalClosure, DisconnectionSource source = const UserInitiated(), }) async { - // If the connection is already disconnected, do nothing. - if (connectionState.value is Disconnected) return; + // A connection already going down keeps the source it started going down + // with: whoever asked first described why. Without this the connect + // timeout could replace a `ServerInitiated` closure, which is + // reconnectable, with one that is not. + if (connectionState.value case Disconnected() || Disconnecting()) return; // Stop the timeout from firing later and replacing this source. _cancelConnectTimeout(); @@ -186,8 +196,33 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL // Update the connection state to 'disconnecting'. _connectionState = WebSocketConnectionState.disconnecting(source: source); - // Close the connection using the engine. - unawaited(_engine.close(closeCode, source.closeReason)); + // Awaited so the connection is closed, rather than merely closing, once this + // returns: a reconnect straight afterwards would otherwise race the close + // and see the connection go down again. + final result = await _engine.close(closeCode, source.closeReason); + + // The engine reports a failed close rather than throwing, and does not + // notify its listener on that path. The connection is unusable either way, + // so report it closed rather than leave it disconnecting for good. + if (result.isFailure) onClose(closeCode, source.closeReason); + } + + /// Releases every resource held by this client. + /// + /// Closes the connection along with [events] and [connectionState], after which + /// this client cannot be connected again. Use [disconnect] for a connection that + /// may be opened again. + /// + /// Returns a [Future] that completes once everything has been released. + @override + Future dispose() async { + await disconnect(); + _healthMonitor.stop(); + + await _events.close(); + await _connectionStateEmitter.close(); + + return super.dispose(); } @override @@ -200,10 +235,18 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL } Future _authenticate() async { - final result = await onAuthenticate?.call(send); + final authenticate = onAuthenticate; + if (authenticate == null) return; + + // Guarded rather than awaited directly: an authenticator that awaits a + // token throws rather than returning a failure, and nothing observes this + // future, so the error would escape and leave the connection + // authenticating until the timeout reported a cause it does not know. + final outcome = await runSafely(() => authenticate(send)); + final result = outcome.flatten(); // Close the connection rather than wait for a reply that cannot come. - if (result?.exceptionOrNull() case final error?) { + if (result.exceptionOrNull() case final error?) { final source = DisconnectionSource.authenticationFailed(error: error); return disconnect(source: source); } diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index 586c8d1f..d0624095 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -115,7 +115,12 @@ sealed class WebSocketConnectionState extends Equatable { UnHealthyConnection() => true, SystemInitiated() => true, UserInitiated() => false, - ConnectTimeout() => false, + // A handshake that did not complete in time is the same failure as a + // connection that stops answering health checks, at an earlier moment. + ConnectTimeout() => true, + // Not the server refusing the credentials, which arrives as an error + // frame: this is the client failing to load or send them, and it will + // fail the same way on a retry. AuthenticationFailed() => false, }, _ => false, // No automatic reconnection for other states diff --git a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart index e458a599..09724557 100644 --- a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart +++ b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart @@ -63,6 +63,8 @@ _client({ when(() => channel.stream).thenAnswer((_) => incoming.stream); final sink = _MockWebSocketSink(); when(() => channel.sink).thenReturn(sink); + // A socket that closes cleanly; tests that need otherwise re-stub this. + when(() => sink.close(any(), any())).thenAnswer((_) async {}); var built = 0; final client = StreamWebSocketClient( @@ -83,6 +85,87 @@ _client({ } void main() { + group('StreamWebSocketClient.disconnect', () { + test('leaves the connection closed, not closing, once it returns', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + + await client.disconnect(); + + // A caller that reconnects straight away would otherwise race the close + // and see the connection go down again. + expect(client.connectionState.value, isA()); + }); + + test('reports the connection closed even when the socket close fails', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + when(() => sink.close(any(), any())).thenThrow(Exception('close failed')); + await client.connect(); + + await client.disconnect(); + + // The engine swallows the failure and never notifies its listener, which + // used to leave the connection disconnecting for good. + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + + test('can be followed by another connect', () async { + final (:client, :sink, incoming: _, :optionsBuilt) = _client(); + await client.connect(); + await client.disconnect(); + + await client.connect(); + + expect(client.connectionState.value, isA()); + expect(optionsBuilt(), 2); + }); + }); + + group('StreamWebSocketClient.dispose', () { + test('closes the connection and both emitters', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + + await client.dispose(); + + expect(client.isDisposed, isTrue); + expect(client.events.isClosed, isTrue); + expect(client.connectionState.isClosed, isTrue); + }); + + test('does nothing when called again', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + await client.dispose(); + + await expectLater(client.dispose(), completes); + }); + + test('refuses to connect again', () async { + final (:client, :sink, incoming: _, :optionsBuilt) = _client(); + await client.connect(); + await client.dispose(); + + // Asserts rather than throws: a reconnect can come from the recovery + // handler, which does not await it and cannot report an error. + await expectLater(client.connect(), throwsA(isA())); + expect(optionsBuilt(), 1); + }); + + test('ignores a socket event arriving after it', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + await client.dispose(); + + // The state emitter is closed, so a late event must not be reported into + // it rather than throwing. + expect(() => client.onClose(1000, 'late'), returnsNormally); + }); + }); + group('StreamWebSocketClient.optionsBuilder', () { test('is called for every connection attempt, not once per client', () async { final (:client, :incoming, :optionsBuilt, sink: _) = _client(); @@ -195,7 +278,28 @@ void main() { final state = client.connectionState.value; expect( state, - isA().having( + isA().having( + (it) => it.source, + 'source', + isA().having((it) => it.error, 'error', isStateError), + ), + ); + }); + + test('closes the connection when the authenticator throws', () async { + // The natural authenticator awaits a token, and loading one throws rather + // than returning a failure. Left unguarded the error escapes unhandled and + // the connection waits for the timeout, which knows no cause. + final (:client, :sink, incoming: _, optionsBuilt: _) = _client( + onAuthenticate: (_) async => throw StateError('token load failed'), + ); + + await client.connect(); + await pumpEventQueue(); + + expect( + client.connectionState.value, + isA().having( (it) => it.source, 'source', isA().having((it) => it.error, 'error', isStateError), @@ -235,6 +339,10 @@ void main() { expect(client.connectionState.value, isA()); async.elapse(const Duration(seconds: 1)); + // Only the source can be observed here: `disconnect` awaits the + // subscription cancel, which `fakeAsync` never completes, so the state + // settles at 'disconnecting'. The reached state is covered by the + // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, isA().having((it) => it.source, 'source', isA()), @@ -256,6 +364,10 @@ void main() { async.elapse(WebSocketOptions.defaultConnectTimeout); + // Only the source can be observed here: `disconnect` awaits the + // subscription cancel, which `fakeAsync` never completes, so the state + // settles at 'disconnecting'. The reached state is covered by the + // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, isA().having((it) => it.source, 'source', isA()), @@ -277,6 +389,10 @@ void main() { async.flushMicrotasks(); async.elapse(WebSocketOptions.defaultConnectTimeout); + // Only the source can be observed here: `disconnect` awaits the + // subscription cancel, which `fakeAsync` never completes, so the state + // settles at 'disconnecting'. The reached state is covered by the + // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, isA().having((it) => it.source, 'source', isA()), @@ -295,6 +411,10 @@ void main() { async.elapse(const Duration(seconds: 2)); + // Only the source can be observed here: `disconnect` awaits the + // subscription cancel, which `fakeAsync` never completes, so the state + // settles at 'disconnecting'. The reached state is covered by the + // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, isA().having((it) => it.source, 'source', isA()), @@ -319,6 +439,29 @@ void main() { }); }); + test('does not replace the source of a socket error that came first', () { + fakeAsync((async) { + final (:client, incoming: _, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + + // A socket error closes the connection without cancelling the timer. + client.onError(StateError('socket died')); + expect(client.connectionState.value, isA()); + + async.elapse(WebSocketOptions.defaultConnectTimeout * 2); + + // Replacing this with `ConnectTimeout` would have made a reconnectable + // failure permanent, since `ServerInitiated` is eligible and the + // timeout used not to be. + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + test('does not replace the source of a disconnect that came first', () { fakeAsync((async) { final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); @@ -332,6 +475,10 @@ void main() { // The timeout would otherwise report this deliberate disconnect as a // timed-out attempt, which reconnects differently. + // Only the source can be observed here: `disconnect` awaits the + // subscription cancel, which `fakeAsync` never completes, so the state + // settles at 'disconnecting'. The reached state is covered by the + // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, isA().having((it) => it.source, 'source', isA()), @@ -342,29 +489,37 @@ void main() { group('StreamWebSocketClient health check while disconnecting', () { test('does not report the connection as established again', () async { - final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + // Held open so the connection is still closing when the pong arrives. + final closing = Completer(); + when(() => sink.close(any(), any())).thenAnswer((_) => closing.future); await client.connect(); client.onMessage(const _HealthCheckEvent()); expect(client.connectionState.value, isA()); - await client.disconnect(); + client.disconnect().ignore(); expect(client.connectionState.value, isA()); // Arrives before the socket finished closing. client.onMessage(const _HealthCheckEvent(connectionId: 'late')); expect(client.connectionState.value, isA()); + closing.complete(); }); test('leaves the disconnection source intact once the socket closes', () async { - final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + final closing = Completer(); + when(() => sink.close(any(), any())).thenAnswer((_) => closing.future); await client.connect(); client.onMessage(const _HealthCheckEvent()); - await client.disconnect(); + client.disconnect().ignore(); client.onMessage(const _HealthCheckEvent(connectionId: 'late')); - client.onClose(); + + closing.complete(); + await pumpEventQueue(); // Without the guard the late health check moves the state back to // connected, and `onClose` then reports a server-initiated disconnect, diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index 59d26f43..1ff6bdd9 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -42,12 +42,12 @@ void main() { }); test( - 'is disabled when a connection attempt timed out, so a handshake that ' - 'never completes is not retried forever', + 'is enabled when a connection attempt timed out, since a handshake that ' + 'did not complete in time is the same failure as one that stopped', () { const state = Disconnected(source: ConnectTimeout()); - expect(state.isAutomaticReconnectionEnabled, isFalse); + expect(state.isAutomaticReconnectionEnabled, isTrue); }, ); From ba79c6a805691e53101b8a2d347d08458cb9dbcb Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 17:18:28 +0200 Subject: [PATCH 04/20] docs(llc): document fromUser and record this round of changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ConnectUserDetailsRequest.fromUser` shipped in #160 without a dartdoc, against the style guide's own rule for new public code. The two things a caller cannot infer are why `role` and `teams` are absent — the server assigns both and ignores them from a client — and that `includeDetails: false` sends the id alone. Also corrects `connect`'s dartdoc, which claimed its future completes when the connection attempt finishes. It resolves once the socket is open, before authentication and well before the connection is usable — which is precisely what the connect timeout exists to bound. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 10 ++++++++-- .../lib/src/user/connect_user_details_request.dart | 6 ++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 35863dcf..16b7766d 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -8,6 +8,7 @@ - Renamed `StreamWebSocketClient.onConnectionEstablished` to `onAuthenticate`, which is what it is called for and when - `StreamWebSocketClient.onAuthenticate` is now a `WebSocketAuthenticator`: it is handed a `WsSender` and returns a `Result`, so a failure to authenticate can be observed - `TokenManager.userId` is now nullable, and is `null` until an identity is configured +- `Result.getOrElse`, `getOrDefault`, `recover` and `recoverCatching` no longer take a type parameter of their own and return the result's own type. They previously cast the value to the callback's type, which failed on a successful result — most visibly for a callback that only throws. Use `fold` where the return type has to differ ### ✨ Features @@ -17,11 +18,12 @@ - Added `User.anonymousUserId`, the id every anonymous user has - Added `TokenManager.unconfigured`, for a client that exists before its user does - Added `TokenManager.reset`, which drops the configured identity and its cached token -- Added `DisconnectionSource.connectTimeout`, reported when a connection attempt is abandoned before the connection is established +- Added `DisconnectionSource.connectTimeout`, reported when a connection attempt is abandoned before the connection is established; it is eligible for automatic reconnection, since a handshake that did not complete in time is the same failure as a connection that stops answering health checks - Added `DisconnectionSource.authenticationFailed`, reported with its cause when a connection opens but cannot be authenticated -- `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout` +- `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout`. This is a behaviour change as well as an API one: a connection previously waited indefinitely for its first health check, and is now abandoned — and reconnected — after 15 seconds - Added `WsSender`, the send capability handed to a `WebSocketAuthenticator` - Added `ConnectUserDetailsRequest.fromUser`, which builds the details a client may send from a `User` +- Added `StreamWebSocketClient.dispose`, which closes the connection along with `events` and `connectionState`; the client is `Disposable`, so `isDisposed` reports whether it has been called - Added `teams` field to `User` class ### 🐛 Bug Fixes @@ -29,6 +31,10 @@ - Fixed `TokenManager.getToken()` contacting the `TokenProvider` on every call instead of returning the cached token - Fixed `DynamicTokenProvider` accepting a token issued for a different user than the one requested - Fixed `TokenManager` caching a token that finished loading after `expireToken` or `setTokenProvider` had invalidated it +- Fixed `StreamWebSocketClient.disconnect` completing before the socket was closed, so a `connect` straight afterwards raced the closure and saw the connection go down again +- Fixed a failure to close the socket leaving `StreamWebSocketClient` reporting itself as disconnecting for good, since the engine reports such a failure rather than notifying its listener +- Fixed a `WebSocketAuthenticator` that throws, rather than returning a failure, escaping as an unhandled error and leaving the connection authenticating until it timed out — losing the cause, which the timeout does not carry. The natural authenticator throws, since loading a token does +- Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` - Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect ### 🔄 Changed diff --git a/packages/stream_core/lib/src/user/connect_user_details_request.dart b/packages/stream_core/lib/src/user/connect_user_details_request.dart index fea5a7ef..7e27d170 100644 --- a/packages/stream_core/lib/src/user/connect_user_details_request.dart +++ b/packages/stream_core/lib/src/user/connect_user_details_request.dart @@ -15,6 +15,12 @@ class ConnectUserDetailsRequest { this.custom, }); + /// Creates the details a client may send when connecting as [user]. + /// + /// A user's role and teams are left out: the server assigns both and does not + /// accept them from a client. + /// + /// Pass [includeDetails] as `false` to send the id alone. factory ConnectUserDetailsRequest.fromUser( User user, { bool includeDetails = true, From bc46b99f461d61ceae1bc08362f222cc16e4c3bf Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 17:24:59 +0200 Subject: [PATCH 05/20] docs(llc): show how to widen a Result now that the helpers do not MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kotlin's `getOrElse`, `getOrDefault`, `recover` and `recoverCatching` widen through a second type parameter bounded by the receiver's — `` — which is what makes their `value as T` sound. Dart has upper bounds only, so the bound cannot be stated and the previous `` was a cast with nothing behind it. The capability is still reachable, just declared in a different place: `Result` is covariant, so naming the wider type on the result gives the same widening that Kotlin infers from the callback. ```dart final Result widened = intResult; widened.getOrElse((_, _) => 0.5); ``` Documents that on both `get` helpers and pins it with a test, so the migration note is not the only record of it. Worth noting for the reviewer: Kotlin's `recover` also returns the receiver unchanged on success (`null -> this`) rather than rebuilding it, and its non-widening members — `getOrNull`, `getOrThrow` — take no type parameter either, which is the shape these four now have. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 2 +- packages/stream_core/lib/src/utils/result.dart | 7 +++++++ packages/stream_core/test/utils/result_test.dart | 13 +++++++++++++ 3 files changed, 21 insertions(+), 1 deletion(-) diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 16b7766d..cf897343 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -8,7 +8,7 @@ - Renamed `StreamWebSocketClient.onConnectionEstablished` to `onAuthenticate`, which is what it is called for and when - `StreamWebSocketClient.onAuthenticate` is now a `WebSocketAuthenticator`: it is handed a `WsSender` and returns a `Result`, so a failure to authenticate can be observed - `TokenManager.userId` is now nullable, and is `null` until an identity is configured -- `Result.getOrElse`, `getOrDefault`, `recover` and `recoverCatching` no longer take a type parameter of their own and return the result's own type. They previously cast the value to the callback's type, which failed on a successful result — most visibly for a callback that only throws. Use `fold` where the return type has to differ +- `Result.getOrElse`, `getOrDefault`, `recover` and `recoverCatching` no longer take a type parameter of their own and return the result's own type. They previously cast the value to the callback's type, which failed on a successful result — most visibly for a callback that only throws. Kotlin's equivalents widen through a `` bound that Dart cannot express; to widen here, name the wider type on the result (`Result widened = intResult`), which works because `Result` is covariant, or use `fold` ### ✨ Features diff --git a/packages/stream_core/lib/src/utils/result.dart b/packages/stream_core/lib/src/utils/result.dart index e29afdce..7f33c313 100644 --- a/packages/stream_core/lib/src/utils/result.dart +++ b/packages/stream_core/lib/src/utils/result.dart @@ -107,6 +107,10 @@ extension PatternMatching on Result { /// Note, that this function rethrows any error thrown by [onFailure] function. /// /// This function is a shorthand for `fold(onSuccess: (it) => it, onFailure: onFailure)`. + /// + /// [onFailure] returns this result's own type. To fall back to a supertype, + /// widen the result first — `Result widened = intResult` — or use [fold], + /// which takes its return type from both branches. T getOrElse(T Function(Object error, StackTrace? stackTrace) onFailure) { return switch (this) { Success(:final data) => data, @@ -118,6 +122,9 @@ extension PatternMatching on Result { /// [defaultValue] if it is [Failure]. /// /// This function is a shorthand for `getOrElse((_, _) => defaultValue)`. + /// + /// [defaultValue] is of this result's own type; widen the result to fall back + /// to a supertype. T getOrDefault(T defaultValue) { return switch (this) { Success(:final data) => data, diff --git a/packages/stream_core/test/utils/result_test.dart b/packages/stream_core/test/utils/result_test.dart index fb789b81..92234e36 100644 --- a/packages/stream_core/test/utils/result_test.dart +++ b/packages/stream_core/test/utils/result_test.dart @@ -45,6 +45,19 @@ void main() { }); }); + group('Result widening', () { + test('falls back to a supertype when the result is widened', () { + // Kotlin widens through a `` bound Dart has no equivalent for. + // Naming the wider type on the result gets there instead, since `Result` + // is covariant. + final Result widened = Result.failure(Exception('failed')); + + expect(widened.getOrElse((_, _) => 0.5), 0.5); + expect(widened.getOrDefault(0.5), 0.5); + expect(widened.recover((_, _) => 0.5).getOrNull(), 0.5); + }); + }); + group('Result.getOrDefault', () { test('returns the value when there is one', () { const result = Result.success(42); From 39952e56de91426562e95469506e63c52ae27d1b Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 17:27:32 +0200 Subject: [PATCH 06/20] docs(llc): note where widening happens for recover `getOrElse` and `getOrDefault` return the receiver's type, so widening the result before or after the call reads the same. `recover` returns a `Result`, so the order matters: widening afterwards gives a `Result` there is nothing left to widen. Kotlin's returns `Result` and infers it from the transform; ours takes it from the receiver, so the receiver has to be widened first. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/lib/src/utils/result.dart | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/stream_core/lib/src/utils/result.dart b/packages/stream_core/lib/src/utils/result.dart index 7f33c313..0032e597 100644 --- a/packages/stream_core/lib/src/utils/result.dart +++ b/packages/stream_core/lib/src/utils/result.dart @@ -187,6 +187,9 @@ extension PatternMatching on Result { /// /// Note, that this function rethrows any error thrown by [transform] function. /// See [recoverCatching] for an alternative that encapsulates errors. + /// + /// [transform] returns this result's own type, so widening is done on the way + /// in rather than on the way out: widen the result first, then recover. Result recover( T Function(Object error, StackTrace? stackTrace) transform, ) { @@ -203,6 +206,9 @@ extension PatternMatching on Result { /// /// This function catches any error thrown by [transform] function and encapsulates it as a failure. /// See [recover] for an alternative that rethrows errors. + /// + /// [transform] returns this result's own type, so widening is done on the way + /// in rather than on the way out: widen the result first, then recover. Result recoverCatching( T Function(Object error, StackTrace? stackTrace) transform, ) { From b0f94d1ce18aa2010b81534580b55414ae99da80 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 17:39:53 +0200 Subject: [PATCH 07/20] fix(llc): do not open a socket while the previous one is still closing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `connect` guarded `connecting`, `authenticating` and `connected` but not `disconnecting`, so it proceeded while an old socket was on its way out. The old socket's close event then reported the new connection as `Disconnected(ServerInitiated)` — and, since `onClose` cancels the connect timeout, disarmed the timer watching the new attempt, leaving it authenticating with nothing to bound it. Awaiting the engine's close made the sequential case safe; this covers the caller that does not await. Raised in review of #160 as pre-existing. Co-Authored-By: Claude Opus 5 (1M context) --- .../ws/client/stream_web_socket_client.dart | 9 +++++++-- .../client/stream_web_socket_client_test.dart | 18 ++++++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart index 6ed3f3ba..4c1d5ff4 100644 --- a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart +++ b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart @@ -144,8 +144,8 @@ class StreamWebSocketClient with Disposable implements WebSocketHealthListener, /// Establishes a WebSocket connection. /// /// The connection state can be monitored through [connectionState] for real-time updates. - /// If the connection is already established or in progress, this method returns immediately. - /// It also does nothing once [dispose] has been called. + /// If the connection is already established or in progress, this method returns immediately, + /// as it does while a previous connection is still closing, and once [dispose] has been called. /// /// Returns a [Future] that completes once the socket is open — before the /// connection is authenticated, and well before it is [Connected]. Watch @@ -160,6 +160,11 @@ class StreamWebSocketClient with Disposable implements WebSocketHealthListener, if (connectionState.value is Authenticating) return; if (connectionState.value is Connected) return; + // Nor while a previous connection is still closing: the socket it opened + // would be brought down by the old one's close event, which would also + // disarm the new attempt's timeout and leave it authenticating unwatched. + if (connectionState.value is Disconnecting) return; + // Update the connection state to 'connecting'. _connectionState = const WebSocketConnectionState.connecting(); diff --git a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart index 09724557..5d53277f 100644 --- a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart +++ b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart @@ -112,6 +112,24 @@ void main() { ); }); + test('does not open a socket while the previous one is still closing', () async { + final (:client, :sink, incoming: _, :optionsBuilt) = _client(); + // Held open so the connection is still closing when connect is called. + final closing = Completer(); + when(() => sink.close(any(), any())).thenAnswer((_) => closing.future); + await client.connect(); + + client.disconnect().ignore(); + expect(client.connectionState.value, isA()); + await client.connect(); + + // The old socket's close event would otherwise bring the new connection + // down and disarm the timeout meant to be watching it. + expect(optionsBuilt(), 1); + expect(client.connectionState.value, isA()); + closing.complete(); + }); + test('can be followed by another connect', () async { final (:client, :sink, incoming: _, :optionsBuilt) = _client(); await client.connect(); From e3d700bfc4385849eb3ef4c28342185b2f2be5f4 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 17:56:50 +0200 Subject: [PATCH 08/20] fix(llc): allow 30 seconds for a connection to establish, not 15 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default I picked for `connectTimeout` matched neither sibling SDK: Swift feeds waits 30 seconds for the same handshake (`FeedsClient+Connection.swift:62`) and Android core defaults to 10 (`StreamSocketConfig.kt:96`). Fifteen was a number, not a decision. Aligning with the more forgiving of the two is the right way round here, since this timeout went from absent to mandatory in this PR: the customer it can hurt is the one whose backend is slow to send its first health check, and review raised exactly that case. It is per-attempt and configurable, so anyone wanting Android's stricter bound can set it. One test now supplies its own five-second timeout rather than using the default: the default outlives the health monitor's first missed pong (25s + 3s), so elapsing past it on an established connection reports an unhealthy connection instead. That ordering is fine in production — the monitor only runs once a connection is established, and this timeout only bounds getting there — but it leaves no window for a test that wants to elapse past one and not the other. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 2 +- .../lib/src/ws/client/engine/web_socket_options.dart | 5 ++++- .../test/ws/client/stream_web_socket_client_test.dart | 11 +++++++---- .../ws/client/web_socket_connection_state_test.dart | 2 +- 4 files changed, 13 insertions(+), 7 deletions(-) diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 5db81fe9..ec78ef12 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -21,7 +21,7 @@ - Added `TokenManager.reset`, which drops the configured identity and its cached token - Added `DisconnectionSource.connectTimeout`, reported when a connection attempt is abandoned before the connection is established; it is eligible for automatic reconnection, since a handshake that did not complete in time is the same failure as a connection that stops answering health checks - Added `DisconnectionSource.authenticationFailed`, reported with its cause when a connection opens but cannot be authenticated -- `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout`. This is a behaviour change as well as an API one: a connection previously waited indefinitely for its first health check, and is now abandoned — and reconnected — after 15 seconds +- `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout`. This is a behaviour change as well as an API one: a connection previously waited indefinitely for its first health check, and is now abandoned — and reconnected — after 30 seconds, matching the wait the Swift SDK allows for the same handshake - Added `WsSender`, the send capability handed to a `WebSocketAuthenticator` - Added `ConnectUserDetailsRequest.fromUser`, which builds the details a client may send from a `User` - Added `StreamWebSocketClient.dispose`, which closes the connection along with `events` and `connectionState`; the client is `Disposable`, so `isDisposed` reports whether it has been called diff --git a/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart b/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart index f4295085..e2bd5be9 100644 --- a/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart +++ b/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart @@ -42,7 +42,10 @@ class WebSocketOptions { final Duration connectTimeout; /// The [connectTimeout] used when none is given. - static const defaultConnectTimeout = Duration(seconds: 15); + /// + /// Matches the wait the Swift SDK allows for the same handshake; the Android + /// one is stricter at ten seconds. + static const defaultConnectTimeout = Duration(seconds: 30); /// WebSocket sub-protocols to negotiate during the handshake. /// diff --git a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart index 5d53277f..441d5212 100644 --- a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart +++ b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart @@ -442,16 +442,19 @@ void main() { test('does not fire once the connection is established', () { fakeAsync((async) { - final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + // A timeout of its own, well short of the health monitor: the default + // one outlives the monitor's first missed pong, so elapsing past it + // would report an unhealthy connection instead. + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + connectTimeout: const Duration(seconds: 5), + ); client.connect().ignore(); async.flushMicrotasks(); client.onMessage(const _HealthCheckEvent()); expect(client.connectionState.value, isA()); - // Past when the timeout would have fired, but before the health - // monitor's first ping is due. - async.elapse(WebSocketOptions.defaultConnectTimeout + const Duration(seconds: 1)); + async.elapse(const Duration(seconds: 6)); expect(client.connectionState.value, isA()); }); diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index 1ff6bdd9..17a414f4 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -91,7 +91,7 @@ void main() { const options = WebSocketOptions(url: 'wss://example.com'); expect(options.connectTimeout, WebSocketOptions.defaultConnectTimeout); - expect(WebSocketOptions.defaultConnectTimeout, const Duration(seconds: 15)); + expect(WebSocketOptions.defaultConnectTimeout, const Duration(seconds: 30)); }); }); } From 7c772d58c6fb243fcdd5205f02e02a83f1c3c2a4 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 18:23:11 +0200 Subject: [PATCH 09/20] test(llc): keep a connection alive the way production does MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The test proving the connect timeout stops mattering once a connection is established sized its window from `defaultConnectTimeout` — a number it does not care about — and so depended on that default staying under the health monitor's 28 seconds. Raising the default to 30 broke it, not because the timeout fired but because the wider window now reached the monitor's first unanswered ping. Fixing it by giving the test a five-second timeout of its own traded one problem for a worse one: a configuration that ships nowhere. It now runs the default and answers the pings, which is what an established connection actually does and the reason the monitor stays quiet in production. The answer is delivered on a timer rather than inside the send, because a pong crosses the wire — delivered synchronously it would register before the monitor arms the timeout it is meant to cancel, and the connection would be called unhealthy anyway. Verified by deleting the `_cancelConnectTimeout()` call the test exists to cover: it fails. Co-Authored-By: Claude Opus 5 (1M context) --- .../client/stream_web_socket_client_test.dart | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart index 441d5212..e4b05b49 100644 --- a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart +++ b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart @@ -442,19 +442,25 @@ void main() { test('does not fire once the connection is established', () { fakeAsync((async) { - // A timeout of its own, well short of the health monitor: the default - // one outlives the monitor's first missed pong, so elapsing past it - // would report an unhealthy connection instead. - final (:client, :incoming, optionsBuilt: _, sink: _) = _client( - connectTimeout: const Duration(seconds: 5), - ); + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + + // A live connection answers its pings, which is what keeps the health + // monitor quiet across a connection that outlives this timeout. The + // answer arrives over the wire, so it lands after the monitor has armed + // its pong timeout rather than inside the send that triggered it. + when(() => sink.add(any())).thenAnswer((_) { + Timer(const Duration(milliseconds: 50), () { + client.onMessage(const _HealthCheckEvent()); + }); + }); client.connect().ignore(); async.flushMicrotasks(); client.onMessage(const _HealthCheckEvent()); expect(client.connectionState.value, isA()); - async.elapse(const Duration(seconds: 6)); + // Past the timeout, and past a ping cycle with it. + async.elapse(WebSocketOptions.defaultConnectTimeout + const Duration(seconds: 10)); expect(client.connectionState.value, isA()); }); From 68cedda77cc66af94e4702f4cc12b9f9b30cd925 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 18:27:01 +0200 Subject: [PATCH 10/20] test(llc): let the connect-timeout tests reach the state they are about MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The four timeout tests all stopped at `Disconnecting(ConnectTimeout)` with a comment excusing it: `disconnect` awaits the engine's subscription cancel, and a `StreamController`'s cancel completes on the event loop, which `fakeAsync` never drives. So the close stalled halfway and the tests could only ever see the transition, never the outcome — which is the part that matters, since a disconnection's source decides whether it is retried. The harness now hands the client a stream whose subscription cancels without the event loop, so a close under `fakeAsync` finishes as it does in production. All four assert `Disconnected(ConnectTimeout)`, one of them now also pinning that an abandoned attempt is eligible for reconnection, and the test that had to deliver `onClose` by hand to make progress no longer does. Same treatment for the socket-error test, which now closes as the peer would and asserts the whole consequence: a recoverable server error stays recoverable, which is what the source guard is for. Three smaller things: two tests called `onClose` after an awaited `disconnect` had already delivered it, which is a double close production cannot produce; a failed socket close is stubbed as a rejected future rather than a synchronous throw, since that is how a real sink reports one; and the assertion that a disposed client refuses to connect now says that it pins debug behaviour, with the untouched builder count covering release. Verified by removing `_startConnectTimeout`: four tests fail. Co-Authored-By: Claude Opus 5 (1M context) --- .../client/stream_web_socket_client_test.dart | 122 ++++++++++++------ 1 file changed, 83 insertions(+), 39 deletions(-) diff --git a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart index e4b05b49..3b073e3c 100644 --- a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart +++ b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart @@ -43,6 +43,62 @@ final class _PingRequest extends WsRequest { List get props => const []; } +/// A stream whose subscription cancels without going through the event loop. +/// +/// `StreamController`'s cancel completes on the event loop, which `fakeAsync` +/// never drives, so a client awaiting one hangs in a test where it would not in +/// production — leaving the close half finished. +class _CancellableStream extends Stream { + _CancellableStream(this._source); + + final Stream _source; + + @override + StreamSubscription listen( + void Function(T event)? onData, { + Function? onError, + void Function()? onDone, + bool? cancelOnError, + }) { + return _CancellableSubscription( + _source.listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError), + ); + } +} + +class _CancellableSubscription implements StreamSubscription { + _CancellableSubscription(this._delegate); + + final StreamSubscription _delegate; + + @override + Future cancel() { + _delegate.cancel().ignore(); + return Future.value(); + } + + @override + void onData(void Function(T data)? handleData) => _delegate.onData(handleData); + + @override + void onError(Function? handleError) => _delegate.onError(handleError); + + @override + void onDone(void Function()? handleDone) => _delegate.onDone(handleDone); + + @override + void pause([Future? resumeSignal]) => _delegate.pause(resumeSignal); + + @override + void resume() => _delegate.resume(); + + @override + bool get isPaused => _delegate.isPaused; + + @override + Future asFuture([E? futureValue]) => _delegate.asFuture(futureValue); +} + /// Builds a client whose socket opens successfully but sends nothing, so the /// handshake only progresses when a test drives it. ({ @@ -60,7 +116,7 @@ _client({ final channel = _MockWebSocketChannel(); when(() => channel.ready).thenAnswer((_) async {}); - when(() => channel.stream).thenAnswer((_) => incoming.stream); + when(() => channel.stream).thenAnswer((_) => _CancellableStream(incoming.stream)); final sink = _MockWebSocketSink(); when(() => channel.sink).thenReturn(sink); // A socket that closes cleanly; tests that need otherwise re-stub this. @@ -99,7 +155,7 @@ void main() { test('reports the connection closed even when the socket close fails', () async { final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); - when(() => sink.close(any(), any())).thenThrow(Exception('close failed')); + when(() => sink.close(any(), any())).thenAnswer((_) => Future.error(Exception('close failed'))); await client.connect(); await client.disconnect(); @@ -168,7 +224,9 @@ void main() { await client.dispose(); // Asserts rather than throws: a reconnect can come from the recovery - // handler, which does not await it and cannot report an error. + // handler, which does not await it and cannot report an error. In a + // release build the assert is gone and the call is a no-op, which is what + // the untouched builder count pins. await expectLater(client.connect(), throwsA(isA())); expect(optionsBuilt(), 1); }); @@ -192,7 +250,6 @@ void main() { expect(optionsBuilt(), 1); await client.disconnect(); - client.onClose(); await client.connect(); expect(optionsBuilt(), 2); @@ -231,7 +288,6 @@ void main() { expect(calls, 1); await client.disconnect(); - client.onClose(); await client.connect(); await pumpEventQueue(); @@ -332,7 +388,6 @@ void main() { await client.connect(); await pumpEventQueue(); - client.onClose(); final state = client.connectionState.value; expect(state, isA()); @@ -357,14 +412,15 @@ void main() { expect(client.connectionState.value, isA()); async.elapse(const Duration(seconds: 1)); - // Only the source can be observed here: `disconnect` awaits the - // subscription cancel, which `fakeAsync` never completes, so the state - // settles at 'disconnecting'. The reached state is covered by the - // `StreamWebSocketClient.disconnect` group. + final state = client.connectionState.value; expect( - client.connectionState.value, - isA().having((it) => it.source, 'source', isA()), + state, + isA().having((it) => it.source, 'source', isA()), ); + + // An attempt abandoned here is retried: a first health check that never + // arrives is the same failure as one that stops arriving. + expect(state.isAutomaticReconnectionEnabled, isTrue); }); }); @@ -382,13 +438,9 @@ void main() { async.elapse(WebSocketOptions.defaultConnectTimeout); - // Only the source can be observed here: `disconnect` awaits the - // subscription cancel, which `fakeAsync` never completes, so the state - // settles at 'disconnecting'. The reached state is covered by the - // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, - isA().having((it) => it.source, 'source', isA()), + isA().having((it) => it.source, 'source', isA()), ); }); }); @@ -400,20 +452,15 @@ void main() { client.connect().ignore(); async.flushMicrotasks(); async.elapse(WebSocketOptions.defaultConnectTimeout); - client.onClose(); expect(client.connectionState.value, isA()); client.connect().ignore(); async.flushMicrotasks(); async.elapse(WebSocketOptions.defaultConnectTimeout); - // Only the source can be observed here: `disconnect` awaits the - // subscription cancel, which `fakeAsync` never completes, so the state - // settles at 'disconnecting'. The reached state is covered by the - // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, - isA().having((it) => it.source, 'source', isA()), + isA().having((it) => it.source, 'source', isA()), ); }); }); @@ -429,13 +476,9 @@ void main() { async.elapse(const Duration(seconds: 2)); - // Only the source can be observed here: `disconnect` awaits the - // subscription cancel, which `fakeAsync` never completes, so the state - // settles at 'disconnecting'. The reached state is covered by the - // `StreamWebSocketClient.disconnect` group. expect( client.connectionState.value, - isA().having((it) => it.source, 'source', isA()), + isA().having((it) => it.source, 'source', isA()), ); }); }); @@ -479,13 +522,18 @@ void main() { async.elapse(WebSocketOptions.defaultConnectTimeout * 2); - // Replacing this with `ConnectTimeout` would have made a reconnectable - // failure permanent, since `ServerInitiated` is eligible and the - // timeout used not to be. + // The peer closes after sending the error, as the socket protocol has + // it, which is what turns the state into a disconnection. + client.onClose(); + + // Replacing this source with `ConnectTimeout` used to make a + // reconnectable failure permanent. + final state = client.connectionState.value; expect( - client.connectionState.value, - isA().having((it) => it.source, 'source', isA()), + state, + isA().having((it) => it.source, 'source', isA()), ); + expect(state.isAutomaticReconnectionEnabled, isTrue); }); }); @@ -501,14 +549,10 @@ void main() { async.elapse(WebSocketOptions.defaultConnectTimeout * 2); // The timeout would otherwise report this deliberate disconnect as a - // timed-out attempt, which reconnects differently. - // Only the source can be observed here: `disconnect` awaits the - // subscription cancel, which `fakeAsync` never completes, so the state - // settles at 'disconnecting'. The reached state is covered by the - // `StreamWebSocketClient.disconnect` group. + // timed-out attempt, which is retried where this is not. expect( client.connectionState.value, - isA().having((it) => it.source, 'source', isA()), + isA().having((it) => it.source, 'source', isA()), ); }); }); From 300908ce1ee65df5fea9296c814d64998ec2b5ff Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 19:20:30 +0200 Subject: [PATCH 11/20] fix(llc): recover connections that existed, not attempts that never landed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ConnectionRecoveryHandler` retried any reconnect-eligible disconnection, including the very first attempt — while `connect`'s caller was being handed that same failure. Making `ConnectTimeout` retryable turned that from a corner into the common case: on a flaky network the caller is told the connection failed, the handler is already re-establishing it, and if the caller does the documented thing and retries, it gets "connection already in progress" from an attempt it did not start. Reconnection now requires a connection to have existed. The first attempt belongs to whoever called `connect` and awaited the outcome; every drop after that is the handler's, timeouts included, so a slow reconnect still keeps retrying with backoff. This is what the Android SDK does — `hasConnectedBefore && isDisconnected && …` in `StreamConnectionRecoveryEvaluatorImpl`, latched on reaching connected — and what the JS SDK arrives at structurally, by only reconnecting from close and health-check handlers and bailing out of `_reconnect` while a caller's attempt is in flight. Swift avoids it a third way, by never turning its initial-connect timeout into a disconnection source at all. The consequence worth knowing: a first attempt that fails is not retried when the network returns either. That is the caller's to handle, and it is documented on the class. Adds the first tests for this handler: a first attempt that times out is not retried, a connection that stops answering health checks is, and having been connected does not override a deliberate disconnect. Verified by removing the gate — one fails. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 1 + .../connection_recovery_handler.dart | 35 +++- .../connection_recovery_handler_test.dart | 184 ++++++++++++++++++ 3 files changed, 213 insertions(+), 7 deletions(-) create mode 100644 packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index e9d0d8d3..2cbb1cdb 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -36,6 +36,7 @@ - Fixed a failure to close the socket leaving `StreamWebSocketClient` reporting itself as disconnecting for good, since the engine reports such a failure rather than notifying its listener - Fixed a `WebSocketAuthenticator` that throws, rather than returning a failure, escaping as an unhandled error and leaving the connection authenticating until it timed out — losing the cause, which the timeout does not carry. The natural authenticator throws, since loading a token does - Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` +- Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers connections that existed, matching the `hasConnectedBefore` gate in the Android SDK - Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect ### 🔄 Changed diff --git a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart index 55c4a0af..004984aa 100644 --- a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart +++ b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart @@ -18,6 +18,10 @@ import 'retry_strategy.dart'; /// when reconnection should occur, implementing exponential backoff with jitter for optimal /// retry behavior. /// +/// It recovers connections that existed: until [StreamWebSocketClient.connect] has established +/// one, connecting belongs to whoever called it and was handed the failure. A first attempt that +/// fails is therefore not retried here, including when the network returns. +/// /// ## Built-in Policies /// /// The handler automatically includes several reconnection policies: @@ -117,7 +121,19 @@ class ConnectionRecoveryHandler extends Disposable { _reconnectionTimer = null; } - bool _canBeReconnected() => _policies.every((it) => it.canBeReconnected()); + // Set once a connection has been established, and never unset: it is what + // separates a drop this handler recovers from a first attempt it does not. + var _hasConnected = false; + + bool _canBeReconnected() { + // Until a connection has existed, connecting belongs to whoever called + // `connect` and was handed the failure. Retrying here as well would work + // behind a caller already told the attempt failed — and would race the + // retry that caller makes in response. + if (!_hasConnected) return false; + + return _policies.every((it) => it.canBeReconnected()); + } bool _canBeDisconnected() { return switch (_client.connectionState.value) { @@ -145,13 +161,18 @@ class ConnectionRecoveryHandler extends Disposable { } void _onConnectionStateChanged(WebSocketConnectionState state) { - return switch (state) { - Connecting() => _cancelReconnection(), - Connected() => _reconnectStrategy.resetConsecutiveFailures(), - Disconnected() => _scheduleReconnectionIfNeeded(), + switch (state) { + case Connecting(): + _cancelReconnection(); + case Connected(): + _hasConnected = true; + _reconnectStrategy.resetConsecutiveFailures(); + case Disconnected(): + _scheduleReconnectionIfNeeded(); // These states do not require any action. - Initialized() || Authenticating() || Disconnecting() => () {}, - }; + case Initialized() || Authenticating() || Disconnecting(): + break; + } } @override diff --git a/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart b/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart new file mode 100644 index 00000000..a81ae371 --- /dev/null +++ b/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart @@ -0,0 +1,184 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; +import 'package:web_socket_channel/web_socket_channel.dart'; + +class _MockWebSocketChannel extends Mock implements WebSocketChannel {} + +class _MockWebSocketSink extends Mock implements WebSocketSink {} + +/// A stream whose subscription cancels without going through the event loop, so +/// a close under `fakeAsync` runs to completion as it does in production. +class _CancellableStream extends Stream { + _CancellableStream(this._source); + + final Stream _source; + + @override + StreamSubscription listen( + void Function(T event)? onData, { + Function? onError, + void Function()? onDone, + bool? cancelOnError, + }) { + return _CancellableSubscription( + _source.listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError), + ); + } +} + +class _CancellableSubscription implements StreamSubscription { + _CancellableSubscription(this._delegate); + + final StreamSubscription _delegate; + + @override + Future cancel() { + _delegate.cancel().ignore(); + return Future.value(); + } + + @override + void onData(void Function(T data)? handleData) => _delegate.onData(handleData); + + @override + void onError(Function? handleError) => _delegate.onError(handleError); + + @override + void onDone(void Function()? handleDone) => _delegate.onDone(handleDone); + + @override + void pause([Future? resumeSignal]) => _delegate.pause(resumeSignal); + + @override + void resume() => _delegate.resume(); + + @override + bool get isPaused => _delegate.isPaused; + + @override + Future asFuture([E? futureValue]) => _delegate.asFuture(futureValue); +} + +class _NoopCodec implements WebSocketMessageCodec { + const _NoopCodec(); + + @override + Object encode(WsRequest message) => ''; + + @override + WsEvent decode(Object message) => const _HealthCheckEvent(); +} + +final class _HealthCheckEvent extends WsEvent { + const _HealthCheckEvent(); + + @override + HealthCheckInfo? get healthCheckInfo => const HealthCheckInfo(connectionId: 'connection-id'); +} + +final class _PingRequest extends WsRequest { + const _PingRequest(); + + @override + Map toJson() => const {}; + + @override + List get props => const []; +} + +/// A client whose socket opens but answers nothing, with a handler attached and +/// a count of the attempts it has made. +({StreamWebSocketClient client, int Function() attempts}) _client() { + final incoming = StreamController.broadcast(); + addTearDown(incoming.close); + + final channel = _MockWebSocketChannel(); + when(() => channel.ready).thenAnswer((_) async {}); + when(() => channel.stream).thenAnswer((_) => _CancellableStream(incoming.stream)); + final sink = _MockWebSocketSink(); + when(() => channel.sink).thenReturn(sink); + when(() => sink.close(any(), any())).thenAnswer((_) async {}); + + var attempts = 0; + final client = StreamWebSocketClient( + optionsBuilder: () { + attempts++; + return const WebSocketOptions(url: 'wss://example.com'); + }, + wsProvider: (_) => channel, + pingRequestBuilder: ([_]) => const _PingRequest(), + messageCodec: const _NoopCodec(), + ); + + final handler = ConnectionRecoveryHandler(client: client); + addTearDown(handler.dispose); + + return (client: client, attempts: () => attempts); +} + +void main() { + group('ConnectionRecoveryHandler', () { + test('does not retry a first attempt that never connected', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + + // The socket opened but the server never answers, so the attempt is + // abandoned — the failure the caller of `connect` is handed. + async.elapse(WebSocketOptions.defaultConnectTimeout); + async.flushMicrotasks(); + expect(client.connectionState.value, isA()); + + // Retrying here would work behind a caller already told it failed, and + // would race the retry that caller makes in response. + async.elapse(const Duration(minutes: 1)); + expect(attempts(), 1); + }); + }); + + test('retries a connection that dropped after being established', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + + // The connection stops answering health checks — a drop rather than a + // failure to connect, so recovering it is this handler's job. The first + // retry carries no delay, so it is under way by the time this returns. + async.elapse(const Duration(seconds: 29)); + async.flushMicrotasks(); + + expect(attempts(), 2); + expect(client.connectionState.value, isA()); + }); + }); + + test('does not retry a disconnect the caller asked for', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + + client.disconnect().ignore(); + async.flushMicrotasks(); + expect(client.connectionState.value, isA()); + + // Having been connected is not enough on its own: the source still says + // this was deliberate. + async.elapse(const Duration(minutes: 1)); + expect(attempts(), 1); + }); + }); + }); +} From 073772686e049c58290810ef72d8516f2f9cc6db Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 19:24:07 +0200 Subject: [PATCH 12/20] fix(llc): hand connecting back to the caller after a deliberate disconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gate added in 300908c asked whether a connection had *ever* existed, which leaks across sessions: connect, disconnect, connect again, and a first attempt that fails is retried by the handler because a connection existed before the caller disconnected — the same double ownership, one session later. The question is whether a connection has existed since the caller last asked for one, so a user-initiated disconnection clears it. A system-initiated one does not: backgrounding and network loss are exactly what this handler exists to come back from, and the source already distinguishes the two. Also corrects what 300908c claimed about the Android SDK. The `hasConnectedBefore` latch is borrowed from it, but the retry model around it is not: Android's evaluator connects only on `networkBecameAvailable` or a return to the foreground, and has no failure-driven retry at all — a socket dropping on a healthy foreground network reconnects nothing there. Ours retries failures with a backoff as well, which makes the latch load-bearing here in a way it is not there. (Android does not reset it either, for that reason.) Co-Authored-By: Claude Opus 5 (1M context) --- .../connection_recovery_handler.dart | 13 ++++++++--- .../connection_recovery_handler_test.dart | 23 +++++++++++++++++++ 2 files changed, 33 insertions(+), 3 deletions(-) diff --git a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart index 004984aa..2aadbeed 100644 --- a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart +++ b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart @@ -121,8 +121,9 @@ class ConnectionRecoveryHandler extends Disposable { _reconnectionTimer = null; } - // Set once a connection has been established, and never unset: it is what - // separates a drop this handler recovers from a first attempt it does not. + // Whether a connection has been established since the caller last asked for + // one: it is what separates a drop this handler recovers from an attempt whose + // outcome the caller is waiting on. var _hasConnected = false; bool _canBeReconnected() { @@ -167,7 +168,13 @@ class ConnectionRecoveryHandler extends Disposable { case Connected(): _hasConnected = true; _reconnectStrategy.resetConsecutiveFailures(); - case Disconnected(): + case Disconnected(:final source): + // A disconnect the caller asked for hands connecting back to them, so + // the next `connect` is a fresh attempt they await rather than a drop to + // recover. A system-initiated one is the opposite: backgrounding and + // network loss are exactly what this handler exists to come back from. + if (source is UserInitiated) _hasConnected = false; + _scheduleReconnectionIfNeeded(); // These states do not require any action. case Initialized() || Authenticating() || Disconnecting(): diff --git a/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart b/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart index a81ae371..234b4846 100644 --- a/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart +++ b/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart @@ -162,6 +162,29 @@ void main() { }); }); + test('hands connecting back after the caller disconnected', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + client.disconnect().ignore(); + async.flushMicrotasks(); + + // A fresh attempt, awaited by whoever made it, that never connects. + client.connect().ignore(); + async.flushMicrotasks(); + async.elapse(WebSocketOptions.defaultConnectTimeout); + async.flushMicrotasks(); + + // Having connected in the previous session does not make this failure + // the handler's to retry. + async.elapse(const Duration(minutes: 1)); + expect(attempts(), 2); + }); + }); + test('does not retry a disconnect the caller asked for', () { fakeAsync((async) { final (:client, :attempts) = _client(); From 2ad554c7c784d6d1d63ec99b5bf6ae8911cfb16c Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 19:24:30 +0200 Subject: [PATCH 13/20] docs(llc): describe the recovery gate as it ended up The entry still said the handler recovers connections that existed and cited Android's `hasConnectedBefore` as the match. Both were left behind by 0737726: the gate is per-session, cleared when the caller disconnects, and the Android comparison holds for the latch but not for the retry model around it. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 2cbb1cdb..2e78397c 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -36,7 +36,7 @@ - Fixed a failure to close the socket leaving `StreamWebSocketClient` reporting itself as disconnecting for good, since the engine reports such a failure rather than notifying its listener - Fixed a `WebSocketAuthenticator` that throws, rather than returning a failure, escaping as an unhandled error and leaving the connection authenticating until it timed out — losing the cause, which the timeout does not carry. The natural authenticator throws, since loading a token does - Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` -- Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers connections that existed, matching the `hasConnectedBefore` gate in the Android SDK +- Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers only connections that have existed since the caller last asked for one, so a deliberate `disconnect` hands connecting back and the next `connect` is the caller's attempt again - Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect ### 🔄 Changed From f4a8f73f3145e98f0bb282ec2b31821fbcaaae06 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 19:27:10 +0200 Subject: [PATCH 14/20] refactor(llc): make the connection-state switch a dispatch, not a body MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two of the four cases had grown bodies — one doing bookkeeping and scheduling together, with the reasoning for both wedged inside the case — so the switch no longer read as what it is: a map from state to response. Each case now names what happened, and the reasoning lives with the method that acts on it. `_hasConnected` moves up with the handler's other state instead of sitting beside one of its three readers. Co-Authored-By: Claude Opus 5 (1M context) --- .../connection_recovery_handler.dart | 40 ++++++++++++------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart index 2aadbeed..fdd4297f 100644 --- a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart +++ b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart @@ -83,6 +83,11 @@ class ConnectionRecoveryHandler extends Disposable { late final _subscriptions = CompositeSubscription(); + // Whether a connection has been established since the caller last asked for + // one: it is what separates a drop this handler recovers from an attempt whose + // outcome the caller is waiting on. + var _hasConnected = false; + /// Attempts reconnection if policies allow it. /// /// Evaluates all configured policies and initiates reconnection when conditions are met. @@ -121,11 +126,6 @@ class ConnectionRecoveryHandler extends Disposable { _reconnectionTimer = null; } - // Whether a connection has been established since the caller last asked for - // one: it is what separates a drop this handler recovers from an attempt whose - // outcome the caller is waiting on. - var _hasConnected = false; - bool _canBeReconnected() { // Until a connection has existed, connecting belongs to whoever called // `connect` and was handed the failure. Retrying here as well would work @@ -166,22 +166,32 @@ class ConnectionRecoveryHandler extends Disposable { case Connecting(): _cancelReconnection(); case Connected(): - _hasConnected = true; - _reconnectStrategy.resetConsecutiveFailures(); + _onConnectionEstablished(); case Disconnected(:final source): - // A disconnect the caller asked for hands connecting back to them, so - // the next `connect` is a fresh attempt they await rather than a drop to - // recover. A system-initiated one is the opposite: backgrounding and - // network loss are exactly what this handler exists to come back from. - if (source is UserInitiated) _hasConnected = false; - - _scheduleReconnectionIfNeeded(); - // These states do not require any action. + _onConnectionLost(source); + // An attempt on its way up or down decides nothing on its own. case Initialized() || Authenticating() || Disconnecting(): break; } } + // A connection exists, so keeping it is this handler's job from here, and the + // failures the backoff had accumulated are behind us. + void _onConnectionEstablished() { + _hasConnected = true; + _reconnectStrategy.resetConsecutiveFailures(); + } + + // A disconnect the caller asked for hands connecting back to them, so the next + // `connect` is a fresh attempt they await rather than a drop to recover. A + // system-initiated one is the opposite: backgrounding and network loss are what + // this handler exists to come back from. + void _onConnectionLost(DisconnectionSource source) { + if (source is UserInitiated) _hasConnected = false; + + _scheduleReconnectionIfNeeded(); + } + @override Future dispose() async { _cancelReconnection(); From 72b14477f9d88a398b48e3a24d8ef0c2672d2bb6 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 20:52:03 +0200 Subject: [PATCH 15/20] feat(llc): reconnect an expired token only when another one exists MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A connection the server closed for an expired token was never reconnected, so a token expiring mid-session ended the session: nothing retried, and even the network returning could not revive it, since the policy reads the current state. That is now eligible for reconnection. On its own that would be worse than the disease. Whether a retry is worth making depends on something the connection state does not know — whether the provider can produce a different token — and a static one cannot, so a guest or a fixed JWT would have reconnected with exactly what was refused, forever, at the backoff's ceiling. `TokenRefreshReconnectionPolicy` answers that half: the state says the failure is worth retrying, the policy says whether the credential can change. Both halves are needed, and a product still has to expire the cached token between them; `stream-feeds-flutter` does that from its own connection-state listener, since only it knows there is a token manager at all. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 2 + .../automatic_reconnection_policy.dart | 39 ++++++++++ .../connection_recovery_handler.dart | 23 +++--- .../client/web_socket_connection_state.dart | 5 +- .../automatic_reconnection_policy_test.dart | 76 +++++++++++++++++++ .../web_socket_connection_state_test.dart | 6 +- 6 files changed, 133 insertions(+), 18 deletions(-) create mode 100644 packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 2e78397c..82126a53 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -19,6 +19,7 @@ - Added `User.anonymousUserId`, the id every anonymous user has - Added `TokenManager.unconfigured`, for a client that exists before its user does - Added `TokenManager.reset`, which drops the configured identity and its cached token +- Added `TokenRefreshReconnectionPolicy`, which stops a reconnection that would present a token the server has already refused. Whether another token exists is a property of the `TokenProvider`, not of the connection, so the connection state cannot decide it alone - Added `DisconnectionSource.connectTimeout`, reported when a connection attempt is abandoned before the connection is established; it is eligible for automatic reconnection, since a handshake that did not complete in time is the same failure as a connection that stops answering health checks - Added `DisconnectionSource.authenticationFailed`, reported with its cause when a connection opens but cannot be authenticated - `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout`. This is a behaviour change as well as an API one: a connection previously waited indefinitely for its first health check, and is now abandoned — and reconnected — after 30 seconds, matching the wait the Swift SDK allows for the same handshake @@ -36,6 +37,7 @@ - Fixed a failure to close the socket leaving `StreamWebSocketClient` reporting itself as disconnecting for good, since the engine reports such a failure rather than notifying its listener - Fixed a `WebSocketAuthenticator` that throws, rather than returning a failure, escaping as an unhandled error and leaving the connection authenticating until it timed out — losing the cause, which the timeout does not carry. The natural authenticator throws, since loading a token does - Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` +- A connection the server closed because the token expired is now eligible for automatic reconnection, so a token expiring mid-session recovers instead of ending the session. Pair it with `TokenRefreshReconnectionPolicy` and something that expires the cached token, or the retry presents the same one - Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers only connections that have existed since the caller last asked for one, so a deliberate `disconnect` hands connecting back and the next `connect` is the caller's attempt again - Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect diff --git a/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart b/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart index e26ee529..74ad603d 100644 --- a/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart +++ b/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart @@ -1,3 +1,5 @@ +import '../../../errors.dart'; +import '../../../user/token_manager.dart'; import '../../../utils.dart'; import '../web_socket_connection_state.dart'; @@ -95,3 +97,40 @@ class CompositeReconnectionPolicy implements AutomaticReconnectionPolicy { }; } } + +/// A policy that only reconnects when the credential can be replaced. +/// +/// A connection the server closed because the token expired is worth retrying, +/// but only with a different token — and whether one can be obtained is a +/// property of the [TokenProvider] rather than of the connection. A provider +/// that always returns the same token has nothing else to offer, so reconnecting +/// would present exactly what was just refused. +/// +/// Pair it with whatever expires the cached token, so the attempt this permits +/// loads a fresh one. +class TokenRefreshReconnectionPolicy implements AutomaticReconnectionPolicy { + /// Creates a [TokenRefreshReconnectionPolicy]. + const TokenRefreshReconnectionPolicy({ + required this.connectionState, + required this.tokenManager, + }); + + /// The connection state to read the last disconnection from. + final ConnectionStateEmitter connectionState; + + /// The manager whose provider decides whether another token is available. + final TokenManager tokenManager; + + @override + bool canBeReconnected() { + final refusedTheToken = switch (connectionState.value) { + Disconnected(source: ServerInitiated(:final error)) => error?.apiError?.isTokenExpiredError ?? false, + _ => false, + }; + + // Every other disconnection is somebody else's call. + if (!refusedTheToken) return true; + + return !tokenManager.usesStaticProvider; + } +} diff --git a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart index fdd4297f..48df9c97 100644 --- a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart +++ b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart @@ -162,24 +162,20 @@ class ConnectionRecoveryHandler extends Disposable { } void _onConnectionStateChanged(WebSocketConnectionState state) { - switch (state) { - case Connecting(): - _cancelReconnection(); - case Connected(): - _onConnectionEstablished(); - case Disconnected(:final source): - _onConnectionLost(source); - // An attempt on its way up or down decides nothing on its own. - case Initialized() || Authenticating() || Disconnecting(): - break; - } + return switch (state) { + Connecting() => _cancelReconnection(), + Connected() => _onConnectionEstablished(), + Disconnected(:final source) => _onConnectionLost(source), + // These states do not require any action. + Initialized() || Authenticating() || Disconnecting() => () {}, + }; } // A connection exists, so keeping it is this handler's job from here, and the // failures the backoff had accumulated are behind us. void _onConnectionEstablished() { _hasConnected = true; - _reconnectStrategy.resetConsecutiveFailures(); + return _reconnectStrategy.resetConsecutiveFailures(); } // A disconnect the caller asked for hands connecting back to them, so the next @@ -188,8 +184,7 @@ class ConnectionRecoveryHandler extends Disposable { // this handler exists to come back from. void _onConnectionLost(DisconnectionSource source) { if (source is UserInitiated) _hasConnected = false; - - _scheduleReconnectionIfNeeded(); + return _scheduleReconnectionIfNeeded(); } @override diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index d0624095..db463000 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -108,7 +108,10 @@ sealed class WebSocketConnectionState extends Equatable { Disconnected(:final source) => switch (source) { ServerInitiated() => switch (source.error?.apiError) { final error? when error.code == 1000 => false, - final error? when error.isTokenExpiredError => false, + // Worth retrying, but only once the credential has been replaced — + // which is the product's to do, since the token is theirs. A + // reconnect that presents the same one is refused the same way. + final error? when error.isTokenExpiredError => true, final error? when error.isClientError => false, _ => true, // Reconnect on other server initiated disconnections }, diff --git a/packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart b/packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart new file mode 100644 index 00000000..c8a59c51 --- /dev/null +++ b/packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart @@ -0,0 +1,76 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../../../helpers/user_token.dart'; + +StreamApiError _apiError(int code) => StreamApiError( + code: code, + details: const [], + duration: '0ms', + message: 'error $code', + moreInfo: '', + statusCode: 401, +); + +ConnectionStateEmitter _stateOf(WebSocketConnectionState state) { + return MutableConnectionStateEmitter(state); +} + +WebSocketConnectionState _refused(StreamApiError apiError) { + return WebSocketConnectionState.disconnected( + source: DisconnectionSource.serverInitiated( + error: WebSocketEngineException(error: apiError), + ), + ); +} + +void main() { + group('TokenRefreshReconnectionPolicy', () { + test('refuses to reconnect when the provider has only one token', () { + final policy = TokenRefreshReconnectionPolicy( + // Token-invalid error codes are 40..42; 40 = token expired. + connectionState: _stateOf(_refused(_apiError(40))), + tokenManager: TokenManager( + userId: 'user-1', + tokenProvider: TokenProvider.static(generateTestUserToken('user-1')), + ), + ); + + // Reconnecting would present the token the server just refused, over and + // over — the loop this policy exists to prevent. + expect(policy.canBeReconnected(), isFalse); + }); + + test('reconnects when the provider can issue another token', () { + final policy = TokenRefreshReconnectionPolicy( + connectionState: _stateOf(_refused(_apiError(40))), + tokenManager: TokenManager( + userId: 'user-1', + tokenProvider: TokenProvider.dynamic( + (userId) async => generateTestUserToken(userId), + ), + ), + ); + + expect(policy.canBeReconnected(), isTrue); + }); + + test('leaves every other disconnection to the other policies', () { + final policy = TokenRefreshReconnectionPolicy( + // A static provider, so this passes only because the disconnection has + // nothing to do with the token. + connectionState: _stateOf( + const WebSocketConnectionState.disconnected( + source: DisconnectionSource.unHealthyConnection(), + ), + ), + tokenManager: TokenManager( + userId: 'user-1', + tokenProvider: TokenProvider.static(generateTestUserToken('user-1')), + ), + ); + + expect(policy.canBeReconnected(), isTrue); + }); + }); +} diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index 17a414f4..76c9e743 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -23,13 +23,13 @@ Disconnected _serverDisconnect(StreamApiError apiError) => Disconnected( void main() { group('WebSocketConnectionState.isAutomaticReconnectionEnabled', () { test( - 'is disabled when the server closes with a token-expired error, so an ' - 'expired (e.g. guest) token does not trigger a silent reconnect loop', + 'is enabled when the server closes with a token-expired error, since the ' + 'product replaces the credential before the attempt is made', () { // Token-invalid error codes are 40..42; 40 = token expired. final state = _serverDisconnect(_apiError(40)); - expect(state.isAutomaticReconnectionEnabled, isFalse); + expect(state.isAutomaticReconnectionEnabled, isTrue); }, ); From 1949cbc2093163d2e9f0f8e891728cbb4da5c2ae Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 21:08:01 +0200 Subject: [PATCH 16/20] fix(llc): classify token errors the way the iOS SDK does MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two rules in `isAutomaticReconnectionEnabled` could never fire, and one meant something broader than its name. The port compared the API error's `code` against 1000 — a WebSocket close code, which `WebSocketEngineException` declares as `stopErrorCode` and carries itself — so a socket the server closed deliberately was reconnected. It also compared `code` against 400..499, a range Stream's error codes never occupy: the backend numbers them -1 through ~102 and puts the HTTP status in a separate field, so every client error was treated as retryable. iOS reads `statusCode` for that, which is why its rule works. `isTokenExpiredError` covered 40..42, so it answered "the token is invalid" while being named for one particular reason. That is a distinction worth having: an expired token (40) is replaced by asking the provider for another, whereas a signature signed with the wrong secret (43), a clock the token is not valid against yet (41, 42), or a wrong API key (2) are configuration problems a fresh token presents again. Those are now `isInvalidTokenError`, and reconnection refuses them while allowing an expired one — the carve-out iOS spells out as "Expired tokens return 401, so it is considered client error". Verified each code against the backend: `monolith/errors/errors.go` defines accessKeyError=2, expiredToken=40, tokenNotValidYet=41, tokenUsedBeforeIAT=42, invalidTokenSignature=43, and returns all four token errors with a 401. The rules move out of a nested switch into a function, since three of them are about one source and read better as prose than as guards. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 3 + .../lib/src/errors/stream_api_error.dart | 28 ++++++-- .../client/web_socket_connection_state.dart | 45 +++++++++---- .../web_socket_connection_state_test.dart | 65 +++++++++++++------ 4 files changed, 102 insertions(+), 39 deletions(-) diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 82126a53..ce46fa29 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -9,6 +9,8 @@ - `StreamWebSocketClient.onAuthenticate` is now a `WebSocketAuthenticator`: it is handed a `WsSender` and returns a `Result`, so a failure to authenticate can be observed - `TokenManager.userId` is now nullable, and is `null` until an identity is configured - `User` now requires a user of type `UserType.anonymous` to carry `User.anonymousUserId` as its id. The constructor is `const`, so a mismatch in a const context fails to compile rather than throwing in debug mode +- `StreamApiError.isTokenExpiredError` now means the token expired (code 40) rather than any invalid-token code. The rest — not yet valid, used before issued, wrong signature — are `isInvalidTokenError`, along with a wrong API key, since another token does not fix any of them +- `StreamApiError.isClientError` compares the HTTP `statusCode` against 400..499 rather than the Stream error `code`, which never falls in that range and so never matched - `Result.getOrElse`, `getOrDefault`, `recover` and `recoverCatching` no longer take a type parameter of their own and return the result's own type. They previously cast the value to the callback's type, which failed on a successful result — most visibly for a callback that only throws. Kotlin's equivalents widen through a `` bound that Dart cannot express; to widen here, name the wider type on the result (`Result widened = intResult`), which works because `Result` is covariant, or use `fold` ### ✨ Features @@ -37,6 +39,7 @@ - Fixed a failure to close the socket leaving `StreamWebSocketClient` reporting itself as disconnecting for good, since the engine reports such a failure rather than notifying its listener - Fixed a `WebSocketAuthenticator` that throws, rather than returning a failure, escaping as an unhandled error and leaving the connection authenticating until it timed out — losing the cause, which the timeout does not carry. The natural authenticator throws, since loading a token does - Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` +- Fixed `isAutomaticReconnectionEnabled` neither refusing a deliberate server close nor refusing client errors: it compared the API error's code against the close code 1000, and against a 400..499 range that Stream codes never occupy, so both rules were dead. It now mirrors the iOS SDK — close code 1000, invalid tokens and 4xx all refuse, an expired token does not - A connection the server closed because the token expired is now eligible for automatic reconnection, so a token expiring mid-session recovers instead of ending the session. Pair it with `TokenRefreshReconnectionPolicy` and something that expires the cached token, or the retry presents the same one - Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers only connections that have existed since the caller last asked for one, so a deliberate `disconnect` hands connecting back and the next `connect` is the caller's attempt again - Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect diff --git a/packages/stream_core/lib/src/errors/stream_api_error.dart b/packages/stream_core/lib/src/errors/stream_api_error.dart index 19c2fe05..14eeccca 100644 --- a/packages/stream_core/lib/src/errors/stream_api_error.dart +++ b/packages/stream_core/lib/src/errors/stream_api_error.dart @@ -69,16 +69,34 @@ class StreamApiError extends Equatable { ]; } -final _tokenInvalidErrorCodes = _range(40, 42); -final _clientErrorCodes = _range(400, 499); +// The token this was issued for has expired; another one is accepted. +const _expiredTokenCode = 40; + +// The token cannot be accepted for a reason another token does not fix: not +// valid yet, used before it was issued, or signed with the wrong secret. +final _invalidTokenCodes = _range(41, 43); + +// The API key itself is wrong, which no token repairs either. +const _accessKeyErrorCode = 2; + +final _clientErrorStatusCodes = _range(400, 499); /// Extension methods for [StreamApiError] to provide convenient error type checks. extension StreamApiErrorExtension on StreamApiError { - /// Whether this error indicates an expired or invalid token. - bool get isTokenExpiredError => _tokenInvalidErrorCodes.contains(code); + /// Whether the token has expired. + /// + /// Distinct from [isInvalidTokenError]: an expired token is replaced by asking + /// the provider for another, where an invalid one is a configuration problem + /// that a fresh token presents again. + bool get isTokenExpiredError => code == _expiredTokenCode; + + /// Whether the token, or the key it was signed with, cannot be accepted. + bool get isInvalidTokenError { + return _invalidTokenCodes.contains(code) || code == _accessKeyErrorCode; + } /// Whether this error is a client-side error (4xx status codes). - bool get isClientError => _clientErrorCodes.contains(code); + bool get isClientError => _clientErrorStatusCodes.contains(statusCode); /// Whether this error indicates rate limiting (429 status code). bool get isRateLimitError => statusCode == 429; diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index db463000..3c15bbbe 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -98,28 +98,23 @@ sealed class WebSocketConnectionState extends Equatable { /// /// ## Reconnection is disabled for: /// - User-initiated disconnections (explicit disconnect calls) - /// - Server errors with code 1000 (normal closure) - /// - Token expired/invalid errors - /// - Client errors (4xx status codes) + /// - A socket the server closed deliberately (close code 1000) + /// - Tokens another token would not fix, and a wrong API key + /// - Client errors (4xx status codes), other than an expired token + /// - A failure to load or send credentials /// /// Returns `true` if automatic reconnection should be attempted. bool get isAutomaticReconnectionEnabled { return switch (this) { Disconnected(:final source) => switch (source) { - ServerInitiated() => switch (source.error?.apiError) { - final error? when error.code == 1000 => false, - // Worth retrying, but only once the credential has been replaced — - // which is the product's to do, since the token is theirs. A - // reconnect that presents the same one is refused the same way. - final error? when error.isTokenExpiredError => true, - final error? when error.isClientError => false, - _ => true, // Reconnect on other server initiated disconnections - }, + ServerInitiated(:final error) => _canReconnectAfter(error), UnHealthyConnection() => true, SystemInitiated() => true, UserInitiated() => false, - // A handshake that did not complete in time is the same failure as a - // connection that stops answering health checks, at an earlier moment. + // A handshake that did not complete in time is the same failure as one + // that stops answering health checks. A caller's own first attempt is + // kept out of this by the recovery handler, which recovers connections + // that existed rather than attempts that never landed. ConnectTimeout() => true, // Not the server refusing the credentials, which arrives as an error // frame: this is the client failing to load or send them, and it will @@ -368,3 +363,25 @@ final class AuthenticationFailed extends DisconnectionSource { @override List get props => [error]; } + +// Whether a connection the server closed is worth opening again. +// +// Mirrors the rules the iOS SDK applies, which the ported version had drifted +// from: it compared the API error's code against the close code 1000 and against +// a 400..499 range that Stream's codes never occupy, so neither rule could fire. +bool _canReconnectAfter(WebSocketEngineException? error) { + // A deliberate stop rather than a failure to recover from. + if (error?.code == WebSocketEngineException.stopErrorCode) return false; + + final apiError = error?.apiError; + if (apiError == null) return true; + + // Another token is refused for the same reason, so asking for one is futile. + if (apiError.isInvalidTokenError) return false; + + // Whatever else the client got wrong is the caller's to fix — except an + // expired token, which is replaced rather than corrected. + if (apiError.isClientError && !apiError.isTokenExpiredError) return false; + + return true; +} diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index 76c9e743..927270a5 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -1,13 +1,13 @@ import 'package:stream_core/stream_core.dart'; import 'package:test/test.dart'; -StreamApiError _apiError(int code) => StreamApiError( +StreamApiError _apiError(int code, {int statusCode = 401}) => StreamApiError( code: code, details: const [], duration: '0ms', message: 'error $code', moreInfo: '', - statusCode: 401, + statusCode: statusCode, ); Disconnected _serverDisconnect(StreamApiError apiError) => Disconnected( @@ -23,33 +23,58 @@ Disconnected _serverDisconnect(StreamApiError apiError) => Disconnected( void main() { group('WebSocketConnectionState.isAutomaticReconnectionEnabled', () { test( - 'is enabled when the server closes with a token-expired error, since the ' - 'product replaces the credential before the attempt is made', + 'is enabled when the token has expired, which another token replaces', () { - // Token-invalid error codes are 40..42; 40 = token expired. - final state = _serverDisconnect(_apiError(40)); - - expect(state.isAutomaticReconnectionEnabled, isTrue); + // 40 = expired; the server returns 401 with it, so the client-error rule + // has to make room for this one. + expect(_serverDisconnect(_apiError(40)).isAutomaticReconnectionEnabled, isTrue); }, ); - test('is enabled for a generic, retryable server-initiated disconnection', () { - // A server error that is neither a normal closure (1000), a token error - // (40..42), nor a client error (400..499) should still reconnect. - final state = _serverDisconnect(_apiError(43)); + test('is disabled when another token would be refused too', () { + // 41 not valid yet, 42 used before issued, 43 signed with the wrong + // secret, 2 wrong API key — none of which a fresh token repairs. + for (final code in [41, 42, 43, 2]) { + expect( + _serverDisconnect(_apiError(code)).isAutomaticReconnectionEnabled, + isFalse, + reason: 'code $code', + ); + } + }); + + test('is disabled for any other client error', () { + // 17 = not allowed. Nothing about retrying changes the answer. + final state = _serverDisconnect(_apiError(17, statusCode: 403)); + + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + + test('is enabled for a server-side failure', () { + // Stream error codes never fall in 400..499, so this is classified by the + // status code alone — which is what the ported rule got wrong. + final state = _serverDisconnect(_apiError(9, statusCode: 500)); expect(state.isAutomaticReconnectionEnabled, isTrue); }); - test( - 'is enabled when a connection attempt timed out, since a handshake that ' - 'did not complete in time is the same failure as one that stopped', - () { - const state = Disconnected(source: ConnectTimeout()); + test('is disabled when the socket was closed deliberately', () { + const state = Disconnected( + source: ServerInitiated( + error: WebSocketEngineException( + code: WebSocketEngineException.stopErrorCode, + ), + ), + ); - expect(state.isAutomaticReconnectionEnabled, isTrue); - }, - ); + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + + test('is enabled when the server closed without saying why', () { + const state = Disconnected(source: ServerInitiated()); + + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); test( 'is disabled when a connection could not be authenticated, since the ' From 0a6d8e6e88448c4131cb22ba7529f6070b91e41f Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 21:13:55 +0200 Subject: [PATCH 17/20] refactor(llc): keep the reconnection rules in the switch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The rules were the same, but split between a switch and a function, so reading what happens after a server-initiated close meant leaving the one place that lists every source. Two `ServerInitiated` cases — one guarded on the close code, one on what the API error says — keep it whole, and the switch stays exhaustive over the sealed source. Co-Authored-By: Claude Opus 5 (1M context) --- .../lib/src/errors/stream_api_error.dart | 4 +-- .../client/web_socket_connection_state.dart | 34 ++++++------------- 2 files changed, 12 insertions(+), 26 deletions(-) diff --git a/packages/stream_core/lib/src/errors/stream_api_error.dart b/packages/stream_core/lib/src/errors/stream_api_error.dart index 14eeccca..f08f5052 100644 --- a/packages/stream_core/lib/src/errors/stream_api_error.dart +++ b/packages/stream_core/lib/src/errors/stream_api_error.dart @@ -91,9 +91,7 @@ extension StreamApiErrorExtension on StreamApiError { bool get isTokenExpiredError => code == _expiredTokenCode; /// Whether the token, or the key it was signed with, cannot be accepted. - bool get isInvalidTokenError { - return _invalidTokenCodes.contains(code) || code == _accessKeyErrorCode; - } + bool get isInvalidTokenError => _invalidTokenCodes.contains(code) || code == _accessKeyErrorCode; /// Whether this error is a client-side error (4xx status codes). bool get isClientError => _clientErrorStatusCodes.contains(statusCode); diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index 3c15bbbe..63064401 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -107,7 +107,17 @@ sealed class WebSocketConnectionState extends Equatable { bool get isAutomaticReconnectionEnabled { return switch (this) { Disconnected(:final source) => switch (source) { - ServerInitiated(:final error) => _canReconnectAfter(error), + // A deliberate stop rather than a failure to recover from. + ServerInitiated(:final error) when error?.code == WebSocketEngineException.stopErrorCode => false, + ServerInitiated(:final error) => switch (error?.apiError) { + // Another token is refused for the same reason, so asking the provider + // for one is futile. + final it? when it.isInvalidTokenError => false, + // Whatever else the client got wrong is the caller's to fix — except + // an expired token, which is replaced rather than corrected. + final it? when it.isClientError && !it.isTokenExpiredError => false, + _ => true, // Reconnect on other server initiated disconnections + }, UnHealthyConnection() => true, SystemInitiated() => true, UserInitiated() => false, @@ -363,25 +373,3 @@ final class AuthenticationFailed extends DisconnectionSource { @override List get props => [error]; } - -// Whether a connection the server closed is worth opening again. -// -// Mirrors the rules the iOS SDK applies, which the ported version had drifted -// from: it compared the API error's code against the close code 1000 and against -// a 400..499 range that Stream's codes never occupy, so neither rule could fire. -bool _canReconnectAfter(WebSocketEngineException? error) { - // A deliberate stop rather than a failure to recover from. - if (error?.code == WebSocketEngineException.stopErrorCode) return false; - - final apiError = error?.apiError; - if (apiError == null) return true; - - // Another token is refused for the same reason, so asking for one is futile. - if (apiError.isInvalidTokenError) return false; - - // Whatever else the client got wrong is the caller's to fix — except an - // expired token, which is replaced rather than corrected. - if (apiError.isClientError && !apiError.isTokenExpiredError) return false; - - return true; -} From e4ac1b467bf4ec79e661d02b78a0de0b1c3ea892 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 21:22:15 +0200 Subject: [PATCH 18/20] refactor(llc): name the close code with the type that models close codes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `WebSocketEngineException.stopErrorCode` was a second name for 1000, kept from the iOS SDK's `WebSocketEngineError.stopErrorCode`, sitting beside a `CloseCode` extension type that already documents the whole range — and describing a normal closure as a "stop error" besides. Two names for one number in the same library is what let the reconnection rule compare it against an API error's code in the first place: a bare int has no home to be wrong about. `CloseCode.normalClosure` is the value `disconnect` and the engine's `close` already pass, and it is a `CloseCode` implementing `int`, so the comparison is unchanged. Its only readers were that rule and its test. A stale reference survives inside the commented-out block in `client_exception.dart`, left alone with the rest of it. Co-Authored-By: Claude Opus 5 (1M context) --- .../stream_core/lib/src/ws/client/engine/web_socket_engine.dart | 2 -- .../lib/src/ws/client/web_socket_connection_state.dart | 2 +- .../test/ws/client/web_socket_connection_state_test.dart | 2 +- 3 files changed, 2 insertions(+), 4 deletions(-) diff --git a/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart b/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart index 7abf7513..d44a0c00 100644 --- a/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart +++ b/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart @@ -195,8 +195,6 @@ class WebSocketEngineException extends Equatable implements Exception { return null; } - static const stopErrorCode = 1000; - @override List get props => [reason, code, error]; } diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index 63064401..9363ab97 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -108,7 +108,7 @@ sealed class WebSocketConnectionState extends Equatable { return switch (this) { Disconnected(:final source) => switch (source) { // A deliberate stop rather than a failure to recover from. - ServerInitiated(:final error) when error?.code == WebSocketEngineException.stopErrorCode => false, + ServerInitiated(:final error) when error?.code == CloseCode.normalClosure => false, ServerInitiated(:final error) => switch (error?.apiError) { // Another token is refused for the same reason, so asking the provider // for one is futile. diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index 927270a5..fd5b76cf 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -62,7 +62,7 @@ void main() { const state = Disconnected( source: ServerInitiated( error: WebSocketEngineException( - code: WebSocketEngineException.stopErrorCode, + code: CloseCode.normalClosure, ), ), ); From dd48bce8a24234cd066684f5b2c81752b226d0ae Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 21:31:34 +0200 Subject: [PATCH 19/20] fix(llc): reconnect after a rate limit, which clears on its own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A rate-limited connect was treated like every other 4xx and never retried, so a client that hit the limit stayed down until the app intervened — for a condition that resolves without anyone doing anything. The backend says as much on the websocket path. It calls `SetHeaders` before the check, so the limit, the remainder and the window's `Reset` are on the upgrade response, then closes with `RateLimitError("Too many requests, check response headers for more information.")` over a one-minute window (`monolith/server/base.go:1707`). Of every 4xx it can reject a connect with, this is the only one carrying a reset: `auth_rejected`, `app_disabled` and `validation_failed` carry none, and `too_many_connections` carries a link to the client-instantiation docs — a pointer at the caller's bug, not a time to retry. Our backoff tops out at 25 seconds per attempt, comfortably inside that window. This also gives `isRateLimitError` its first reader; it had none. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 1 + .../client/web_socket_connection_state.dart | 46 +++++++------------ .../web_socket_connection_state_test.dart | 9 ++++ 3 files changed, 26 insertions(+), 30 deletions(-) diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index ce46fa29..984caa27 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -40,6 +40,7 @@ - Fixed a `WebSocketAuthenticator` that throws, rather than returning a failure, escaping as an unhandled error and leaving the connection authenticating until it timed out — losing the cause, which the timeout does not carry. The natural authenticator throws, since loading a token does - Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` - Fixed `isAutomaticReconnectionEnabled` neither refusing a deliberate server close nor refusing client errors: it compared the API error's code against the close code 1000, and against a 400..499 range that Stream codes never occupy, so both rules were dead. It now mirrors the iOS SDK — close code 1000, invalid tokens and 4xx all refuse, an expired token does not +- A connection the server closed because the request was rate limited is now eligible for automatic reconnection. The backend closes it with the rate-limit window's reset in the response headers and a one-minute window, so the condition clears on its own — unlike every other 4xx it can close a socket with, which carry no reset - A connection the server closed because the token expired is now eligible for automatic reconnection, so a token expiring mid-session recovers instead of ending the session. Pair it with `TokenRefreshReconnectionPolicy` and something that expires the cached token, or the retry presents the same one - Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers only connections that have existed since the caller last asked for one, so a deliberate `disconnect` hands connecting back and the next `connect` is the caller's attempt again - Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index 9363ab97..a46ca53d 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -100,40 +100,26 @@ sealed class WebSocketConnectionState extends Equatable { /// - User-initiated disconnections (explicit disconnect calls) /// - A socket the server closed deliberately (close code 1000) /// - Tokens another token would not fix, and a wrong API key - /// - Client errors (4xx status codes), other than an expired token + /// - Client errors (4xx status codes), other than an expired token or a rate limit /// - A failure to load or send credentials /// /// Returns `true` if automatic reconnection should be attempted. - bool get isAutomaticReconnectionEnabled { - return switch (this) { - Disconnected(:final source) => switch (source) { - // A deliberate stop rather than a failure to recover from. - ServerInitiated(:final error) when error?.code == CloseCode.normalClosure => false, - ServerInitiated(:final error) => switch (error?.apiError) { - // Another token is refused for the same reason, so asking the provider - // for one is futile. - final it? when it.isInvalidTokenError => false, - // Whatever else the client got wrong is the caller's to fix — except - // an expired token, which is replaced rather than corrected. - final it? when it.isClientError && !it.isTokenExpiredError => false, - _ => true, // Reconnect on other server initiated disconnections - }, - UnHealthyConnection() => true, - SystemInitiated() => true, - UserInitiated() => false, - // A handshake that did not complete in time is the same failure as one - // that stops answering health checks. A caller's own first attempt is - // kept out of this by the recovery handler, which recovers connections - // that existed rather than attempts that never landed. - ConnectTimeout() => true, - // Not the server refusing the credentials, which arrives as an error - // frame: this is the client failing to load or send them, and it will - // fail the same way on a retry. - AuthenticationFailed() => false, + bool get isAutomaticReconnectionEnabled => switch (this) { + Disconnected(:final source) => switch (source) { + ServerInitiated(:final error) when error?.code == CloseCode.normalClosure => false, + ServerInitiated(:final error) => switch (error?.apiError) { + final error? when error.isInvalidTokenError => false, + final error? when error.isClientError && !error.isTokenExpiredError && !error.isRateLimitError => false, + _ => true, // Reconnect on other server initiated disconnections }, - _ => false, // No automatic reconnection for other states - }; - } + UnHealthyConnection() => true, + SystemInitiated() => true, + ConnectTimeout() => true, + UserInitiated() => false, + AuthenticationFailed() => false, + }, + _ => false, // No automatic reconnection for other states + }; @override List get props => []; diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index fd5b76cf..1b854860 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -43,6 +43,15 @@ void main() { } }); + test('is enabled when the request was rate limited', () { + // 9 = rate limited, sent as 429. The server closes with the window's reset + // in the response headers, so the condition clears without the caller + // doing anything. + final state = _serverDisconnect(_apiError(9, statusCode: 429)); + + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); + test('is disabled for any other client error', () { // 17 = not allowed. Nothing about retrying changes the answer. final state = _serverDisconnect(_apiError(17, statusCode: 403)); From 0e4b96a9cbac0cdbbc1ed29628568318ce48078e Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Thu, 20 Aug 2026 21:39:56 +0200 Subject: [PATCH 20/20] fix(llc): let the caller replace a refused token, and stop trying to help MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Supersedes the arrangement built up over 72b1447 and dd48bce: a token-expired close was made reconnect-eligible, a `TokenRefreshReconnectionPolicy` vetoed the cases where that would loop, and `stream-feeds-flutter` expired the cached token from a connection-state listener so the retry would present a new one. Three pieces, in two packages, splitting one decision — and the ordering only worked because the recovery handler schedules a timer rather than connecting inline. The Android SDK does it in one place instead, inside the connect operation: socketSession.connect(data).onTokenError { error, code -> tokenManager.invalidate() tokenManager.refresh().flatMap { newToken -> socketSession.connect(data.copy(token = newToken.rawValue)) } } Invalidate, obtain another, attempt once more, and if that fails, fail the connect. Nothing consults reconnection eligibility, because the connect itself resolved it. `StreamFeedsClient.connect` already awaits the whole outcome, so it can do exactly that, and does. So automatic reconnection refuses a token-expired close again — correct for the reason it always was: the recovery handler cannot replace a credential, so a retry it makes presents the same one. `isExpiredTokenDisconnection` stays, as the way a caller that *can* replace it is told to. The cost, stated plainly: a token expiring on a live connection is no longer recovered automatically. The connection closes and waits for the app to connect again, which then refreshes as above. Android accepts the same — its evaluator reconnects only on network and lifecycle transitions — and iOS calls `connect()` explicitly rather than relying on recovery. Co-Authored-By: Claude Opus 5 (1M context) --- packages/stream_core/CHANGELOG.md | 3 +- .../automatic_reconnection_policy.dart | 39 ---------- .../client/web_socket_connection_state.dart | 16 +++- .../automatic_reconnection_policy_test.dart | 76 ------------------- .../web_socket_connection_state_test.dart | 12 ++- 5 files changed, 22 insertions(+), 124 deletions(-) delete mode 100644 packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 984caa27..e0d30d7d 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -21,11 +21,11 @@ - Added `User.anonymousUserId`, the id every anonymous user has - Added `TokenManager.unconfigured`, for a client that exists before its user does - Added `TokenManager.reset`, which drops the configured identity and its cached token -- Added `TokenRefreshReconnectionPolicy`, which stops a reconnection that would present a token the server has already refused. Whether another token exists is a property of the `TokenProvider`, not of the connection, so the connection state cannot decide it alone - Added `DisconnectionSource.connectTimeout`, reported when a connection attempt is abandoned before the connection is established; it is eligible for automatic reconnection, since a handshake that did not complete in time is the same failure as a connection that stops answering health checks - Added `DisconnectionSource.authenticationFailed`, reported with its cause when a connection opens but cannot be authenticated - `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout`. This is a behaviour change as well as an API one: a connection previously waited indefinitely for its first health check, and is now abandoned — and reconnected — after 30 seconds, matching the wait the Swift SDK allows for the same handshake - Added `WsSender`, the send capability handed to a `WebSocketAuthenticator` +- Added `WebSocketConnectionState.isExpiredTokenDisconnection`, so a caller that can replace the token knows when to. Automatic reconnection deliberately refuses this case: whoever retries it would present the token the server just refused, and only the caller can obtain another - Added `ConnectUserDetailsRequest.fromUser`, which builds the details a client may send from a `User` - Added `StreamWebSocketClient.dispose`, which closes the connection along with `events` and `connectionState`; the client is `Disposable`, so `isDisposed` reports whether it has been called - Added `teams` field to `User` class @@ -41,7 +41,6 @@ - Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` - Fixed `isAutomaticReconnectionEnabled` neither refusing a deliberate server close nor refusing client errors: it compared the API error's code against the close code 1000, and against a 400..499 range that Stream codes never occupy, so both rules were dead. It now mirrors the iOS SDK — close code 1000, invalid tokens and 4xx all refuse, an expired token does not - A connection the server closed because the request was rate limited is now eligible for automatic reconnection. The backend closes it with the rate-limit window's reset in the response headers and a one-minute window, so the condition clears on its own — unlike every other 4xx it can close a socket with, which carry no reset -- A connection the server closed because the token expired is now eligible for automatic reconnection, so a token expiring mid-session recovers instead of ending the session. Pair it with `TokenRefreshReconnectionPolicy` and something that expires the cached token, or the retry presents the same one - Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers only connections that have existed since the caller last asked for one, so a deliberate `disconnect` hands connecting back and the next `connect` is the caller's attempt again - Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect diff --git a/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart b/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart index 74ad603d..e26ee529 100644 --- a/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart +++ b/packages/stream_core/lib/src/ws/client/reconnect/automatic_reconnection_policy.dart @@ -1,5 +1,3 @@ -import '../../../errors.dart'; -import '../../../user/token_manager.dart'; import '../../../utils.dart'; import '../web_socket_connection_state.dart'; @@ -97,40 +95,3 @@ class CompositeReconnectionPolicy implements AutomaticReconnectionPolicy { }; } } - -/// A policy that only reconnects when the credential can be replaced. -/// -/// A connection the server closed because the token expired is worth retrying, -/// but only with a different token — and whether one can be obtained is a -/// property of the [TokenProvider] rather than of the connection. A provider -/// that always returns the same token has nothing else to offer, so reconnecting -/// would present exactly what was just refused. -/// -/// Pair it with whatever expires the cached token, so the attempt this permits -/// loads a fresh one. -class TokenRefreshReconnectionPolicy implements AutomaticReconnectionPolicy { - /// Creates a [TokenRefreshReconnectionPolicy]. - const TokenRefreshReconnectionPolicy({ - required this.connectionState, - required this.tokenManager, - }); - - /// The connection state to read the last disconnection from. - final ConnectionStateEmitter connectionState; - - /// The manager whose provider decides whether another token is available. - final TokenManager tokenManager; - - @override - bool canBeReconnected() { - final refusedTheToken = switch (connectionState.value) { - Disconnected(source: ServerInitiated(:final error)) => error?.apiError?.isTokenExpiredError ?? false, - _ => false, - }; - - // Every other disconnection is somebody else's call. - if (!refusedTheToken) return true; - - return !tokenManager.usesStaticProvider; - } -} diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index a46ca53d..d18e8cf9 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -100,16 +100,26 @@ sealed class WebSocketConnectionState extends Equatable { /// - User-initiated disconnections (explicit disconnect calls) /// - A socket the server closed deliberately (close code 1000) /// - Tokens another token would not fix, and a wrong API key - /// - Client errors (4xx status codes), other than an expired token or a rate limit + /// - Client errors (4xx status codes), other than a rate limit /// - A failure to load or send credentials /// + /// Whether the server closed this connection because the token had expired. + /// + /// Reported so a caller that can replace the token knows to, since a + /// reconnection cannot: [isAutomaticReconnectionEnabled] refuses this, because + /// whoever retries it here would present the token that was just refused. + bool get isExpiredTokenDisconnection => switch (this) { + Disconnected(source: ServerInitiated(:final error)) => error?.apiError?.isTokenExpiredError ?? false, + _ => false, + }; + /// Returns `true` if automatic reconnection should be attempted. bool get isAutomaticReconnectionEnabled => switch (this) { Disconnected(:final source) => switch (source) { ServerInitiated(:final error) when error?.code == CloseCode.normalClosure => false, ServerInitiated(:final error) => switch (error?.apiError) { - final error? when error.isInvalidTokenError => false, - final error? when error.isClientError && !error.isTokenExpiredError && !error.isRateLimitError => false, + final it? when it.isInvalidTokenError => false, + final it? when it.isClientError && !it.isRateLimitError => false, _ => true, // Reconnect on other server initiated disconnections }, UnHealthyConnection() => true, diff --git a/packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart b/packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart deleted file mode 100644 index c8a59c51..00000000 --- a/packages/stream_core/test/ws/client/reconnect/automatic_reconnection_policy_test.dart +++ /dev/null @@ -1,76 +0,0 @@ -import 'package:stream_core/stream_core.dart'; -import 'package:test/test.dart'; - -import '../../../helpers/user_token.dart'; - -StreamApiError _apiError(int code) => StreamApiError( - code: code, - details: const [], - duration: '0ms', - message: 'error $code', - moreInfo: '', - statusCode: 401, -); - -ConnectionStateEmitter _stateOf(WebSocketConnectionState state) { - return MutableConnectionStateEmitter(state); -} - -WebSocketConnectionState _refused(StreamApiError apiError) { - return WebSocketConnectionState.disconnected( - source: DisconnectionSource.serverInitiated( - error: WebSocketEngineException(error: apiError), - ), - ); -} - -void main() { - group('TokenRefreshReconnectionPolicy', () { - test('refuses to reconnect when the provider has only one token', () { - final policy = TokenRefreshReconnectionPolicy( - // Token-invalid error codes are 40..42; 40 = token expired. - connectionState: _stateOf(_refused(_apiError(40))), - tokenManager: TokenManager( - userId: 'user-1', - tokenProvider: TokenProvider.static(generateTestUserToken('user-1')), - ), - ); - - // Reconnecting would present the token the server just refused, over and - // over — the loop this policy exists to prevent. - expect(policy.canBeReconnected(), isFalse); - }); - - test('reconnects when the provider can issue another token', () { - final policy = TokenRefreshReconnectionPolicy( - connectionState: _stateOf(_refused(_apiError(40))), - tokenManager: TokenManager( - userId: 'user-1', - tokenProvider: TokenProvider.dynamic( - (userId) async => generateTestUserToken(userId), - ), - ), - ); - - expect(policy.canBeReconnected(), isTrue); - }); - - test('leaves every other disconnection to the other policies', () { - final policy = TokenRefreshReconnectionPolicy( - // A static provider, so this passes only because the disconnection has - // nothing to do with the token. - connectionState: _stateOf( - const WebSocketConnectionState.disconnected( - source: DisconnectionSource.unHealthyConnection(), - ), - ), - tokenManager: TokenManager( - userId: 'user-1', - tokenProvider: TokenProvider.static(generateTestUserToken('user-1')), - ), - ); - - expect(policy.canBeReconnected(), isTrue); - }); - }); -} diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index 1b854860..5219159d 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -23,11 +23,15 @@ Disconnected _serverDisconnect(StreamApiError apiError) => Disconnected( void main() { group('WebSocketConnectionState.isAutomaticReconnectionEnabled', () { test( - 'is enabled when the token has expired, which another token replaces', + 'is disabled when the token has expired, since a retry here would present ' + 'the same one', () { - // 40 = expired; the server returns 401 with it, so the client-error rule - // has to make room for this one. - expect(_serverDisconnect(_apiError(40)).isAutomaticReconnectionEnabled, isTrue); + // 40 = expired. Replacing it is the caller's to do, and it is the caller + // that retries — `isExpiredTokenDisconnection` is how they are told. + final state = _serverDisconnect(_apiError(40)); + + expect(state.isAutomaticReconnectionEnabled, isFalse); + expect(state.isExpiredTokenDisconnection, isTrue); }, );