fix(llc): don't stall when the token provider fails during a refresh

packages/stream_chat/CHANGELOG.md

@@ -7,6 +7,7 @@
 
 🐞 Fixed
 
+- Fixed the connection getting stuck while connecting, and requests never completing, when the token provider throws while refreshing an expired token.
 - Fixed sending, updating and deleting a message not retrying when the server answers with a timeout (408), a rate limit (429) or a server error (5xx).
 - Fixed hard-deleting an unread message not lowering the channel's unread count.
 - Fixed a draft saved or deleted while offline not being kept.

packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart

@@ -58,7 +58,13 @@ class AuthInterceptor extends QueuedInterceptor {
     final error = ErrorResponse.fromJson(data);
     if (error.code == ChatErrorCode.tokenExpired.code) {
       if (_tokenManager.isStatic) return handler.next(exception);
-      await _tokenManager.loadToken(refresh: true);
+      try {
+        await _tokenManager.loadToken(refresh: true);
+      } catch (_) {
+        // Without a new token the request can't be retried, so it fails with
+        // the original error.
+        return handler.next(exception);
+      }
       try {
         final options = exception.requestOptions;
         final response = await _client.fetch(options);

packages/stream_chat/lib/src/core/http/token_manager.dart

@@ -64,8 +64,17 @@ class TokenManager {
       'Please call `setTokenOrProvider` before calling `loadToken`',
     );
     if (refresh || _token == null) {
-      final rawValue = await _tokenProvider!(_userId!);
-      _token = Token.fromRawValue(rawValue);
+      final expiredToken = _token;
+      try {
+        final rawValue = await _tokenProvider!(_userId!);
+        _token = Token.fromRawValue(rawValue);
+      } catch (_) {
+        // The token being refreshed has expired, so the next load fetches a
+        // new one instead of returning it. A token set while the refresh was
+        // pending is kept.
+        if (refresh && identical(_token, expiredToken)) _token = null;
+        rethrow;
+      }
     }
     return _token!;
   }

packages/stream_chat/lib/src/ws/websocket.dart

@@ -322,14 +322,17 @@ class WebSocket with TimerHelper {
         // Reconnect may have been paused while the delay was pending.
         if (!_reconnectEnabled) return;
 
-        final uri = await _buildUri(
-          refreshToken: refreshToken,
-          includeUserDetails: false,
-        );
-        _logger?.info('[reconnect] #ws; uri: $uri');
         try {
+          final uri = await _buildUri(
+            refreshToken: refreshToken,
+            includeUserDetails: false,
+          );
+          // Disconnected while the token was loading.
+          if (_user == null) return;
+          _logger?.info('[reconnect] #ws; uri: $uri');
           _initWebSocketChannel(uri);
         } catch (e, stk) {
+          if (_user == null) return;
           _onConnectionError(e, stk);
         }
       },

packages/stream_chat/test/src/core/http/interceptor/auth_interceptor_test.dart

@@ -213,6 +213,42 @@ void main() {
     },
   );
 
+  test('`onError` rejects with the original error when refreshing the token fails', () async {
+    const path = 'test-request-path';
+    final options = RequestOptions(path: path);
+    const code = ChatErrorCode.tokenExpired;
+    final errorResponse = ErrorResponse()
+      ..code = code.code
+      ..message = code.message;
+    final response = Response(
+      requestOptions: options,
+      data: errorResponse.toJson(),
+    );
+    final err = DioException(requestOptions: options, response: response);
+    final handler = ErrorInterceptorHandler();
+
+    when(() => tokenManager.isStatic).thenReturn(false);
+    when(() => tokenManager.loadToken(refresh: true)).thenThrow(Exception('token provider is down'));
+
+    authInterceptor.onError(err, handler);
+
+    Object? caught;
+    try {
+      await handler.future.timeout(const Duration(seconds: 5));
+    } catch (e) {
+      caught = e;
+    }
+
+    // need to cast it as the type is private in dio
+    final error = (caught as dynamic)?.data;
+    expect(error, isA<DioException>());
+    expect(
+      StreamChatNetworkError.fromDioException(error as DioException).errorCode,
+      code,
+    );
+    verifyNever(() => client.fetch(options));
+  });
+
   test(
     '`onError` should reject with error if error is not a `tokenExpired error`',
     () async {

packages/stream_chat/test/src/core/http/token_manager_test.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:stream_chat/src/core/http/token.dart';
 import 'package:stream_chat/src/core/http/token_manager.dart';
 import 'package:test/test.dart';
@@ -125,6 +127,50 @@ void main() {
     },
   );
 
+  test('`.loadToken` fetches a new token after a refresh fails', () async {
+    const userId = 'test-user-id';
+    final expiredToken = Token.development(userId);
+    final newToken = Token.fromRawValue('${expiredToken.rawValue}-new');
+
+    var providerCalls = 0;
+    Future<String> tokenProvider(String userId) async {
+      providerCalls += 1;
+      return switch (providerCalls) {
+        1 => expiredToken.rawValue,
+        2 => throw Exception('token provider is down'),
+        _ => newToken.rawValue,
+      };
+    }
+
+    await tokenManager.setTokenOrProvider(userId, provider: tokenProvider);
+
+    await expectLater(tokenManager.loadToken(refresh: true), throwsException);
+
+    expect(await tokenManager.loadToken(), newToken);
+  });
+
+  test('`.loadToken` keeps a token set while a failed refresh was pending', () async {
+    final refresh = Completer<String>();
+    var providerCalls = 0;
+    Future<String> tokenProvider(String userId) async {
+      providerCalls += 1;
+      if (providerCalls == 2) return refresh.future;
+      return Token.development(userId).rawValue;
+    }
+
+    await tokenManager.setTokenOrProvider('user-a', provider: tokenProvider);
+    final pendingRefresh = tokenManager.loadToken(refresh: true);
+
+    tokenManager.reset();
+    final userBToken = Token.development('user-b');
+    await tokenManager.setTokenOrProvider('user-b', token: userBToken);
+
+    refresh.completeError(Exception('token provider is down'));
+    await expectLater(pendingRefresh, throwsException);
+
+    expect(await tokenManager.loadToken(), userBToken);
+  });
+
   test('`.reset` should reset the tokenManager', () async {
     const userId = 'test-user-id';
     final token = Token.development(userId);

packages/stream_chat/test/src/ws/websocket_test.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:convert';
 
 import 'package:mocktail/mocktail.dart';
+import 'package:stream_chat/src/core/http/token.dart';
 import 'package:stream_chat/src/core/http/token_manager.dart';
 import 'package:stream_chat/src/ws/websocket.dart';
 import 'package:stream_chat/stream_chat.dart';
@@ -507,6 +508,118 @@ void main() {
     },
   );
 
+  test('a token provider that fails during a refresh does not stall reconnecting', () async {
+    final user = OwnUser(id: 'test-user');
+
+    var providerCalls = 0;
+    final providerTokenManager = TokenManager();
+    await providerTokenManager.setTokenOrProvider(
+      user.id,
+      provider: (userId) async {
+        providerCalls += 1;
+        // The refresh after the token expires fails.
+        if (providerCalls == 2) throw Exception('token provider is down');
+        return Token.development(userId).rawValue;
+      },
+    );
+
+    var channelsOpened = 0;
+    final providerWebSocket = WebSocket(
+      apiKey: 'api-key',
+      baseUrl: 'ws://<local-ip>:8800',
+      tokenManager: providerTokenManager,
+      webSocketChannelProvider: (_, {protocols}) {
+        channelsOpened += 1;
+        return webSocketChannel;
+      },
+    );
+    addTearDown(providerWebSocket.disconnect);
+
+    final timer = Timer(const Duration(milliseconds: 300), () {
+      final event = Event(type: EventType.healthCheck, connectionId: 'connection-id', me: user);
+      webSocketSink.add(json.encode(event));
+    });
+    addTearDown(timer.cancel);
+
+    await providerWebSocket.connect(user);
+    expect(channelsOpened, 1);
+
+    final tokenExpired = ErrorResponse()
+      ..code = ChatErrorCode.tokenExpired.code
+      ..message = 'Token expired';
+    webSocketSink.add(json.encode({'error': tokenExpired}));
+
+    // The failed refresh still leads to another connection attempt, which
+    // fetches a new token instead of reusing the expired one.
+    await Future.doWhile(() async {
+      await Future<void>.delayed(const Duration(milliseconds: 100));
+      return channelsOpened < 2;
+    }).timeout(const Duration(seconds: 15));
+
+    expect(providerCalls, 3);
+  });
+
+  for (final refreshFails in [true, false]) {
+    final outcome = refreshFails ? 'fails' : 'succeeds';
+    test('a token refresh that $outcome after `disconnect` does not reconnect', () async {
+      final user = OwnUser(id: 'test-user');
+
+      final refresh = Completer<String>();
+      var providerCalls = 0;
+      var channelsOpened = 0;
+      final providerTokenManager = TokenManager();
+      await providerTokenManager.setTokenOrProvider(
+        user.id,
+        provider: (userId) async {
+          providerCalls += 1;
+          if (providerCalls == 2) return refresh.future;
+          return Token.development(userId).rawValue;
+        },
+      );
+
+      final providerWebSocket = WebSocket(
+        apiKey: 'api-key',
+        baseUrl: 'ws://<local-ip>:8800',
+        tokenManager: providerTokenManager,
+        webSocketChannelProvider: (_, {protocols}) {
+          channelsOpened += 1;
+          return webSocketChannel;
+        },
+      );
+      addTearDown(providerWebSocket.disconnect);
+
+      final timer = Timer(const Duration(milliseconds: 300), () {
+        final event = Event(type: EventType.healthCheck, connectionId: 'connection-id', me: user);
+        webSocketSink.add(json.encode(event));
+      });
+      addTearDown(timer.cancel);
+
+      await providerWebSocket.connect(user);
+
+      final tokenExpired = ErrorResponse()
+        ..code = ChatErrorCode.tokenExpired.code
+        ..message = 'Token expired';
+      webSocketSink.add(json.encode({'error': tokenExpired}));
+
+      // Wait for the reconnect to start refreshing the token.
+      await Future.doWhile(() async {
+        await Future<void>.delayed(const Duration(milliseconds: 100));
+        return providerCalls < 2;
+      }).timeout(const Duration(seconds: 15));
+
+      providerWebSocket.disconnect();
+      if (refreshFails) {
+        refresh.completeError(Exception('token provider is down'));
+      } else {
+        refresh.complete(Token.development(user.id).rawValue);
+      }
+      await Future<void>.delayed(const Duration(milliseconds: 100));
+
+      expect(providerWebSocket.connectionStatus, ConnectionStatus.disconnected);
+      expect(channelsOpened, 1);
+    });
+  }
+
   test('`disconnect` successfully disconnects the current user', () async {
     final user = OwnUser(id: 'test-user');
     const connectionId = 'test-connection-id';