diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 7028ead278..0fd26e7240 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/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. diff --git a/packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart b/packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart index 105dba4a70..760575440a 100644 --- a/packages/stream_chat/lib/src/core/http/interceptor/auth_interceptor.dart +++ b/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); diff --git a/packages/stream_chat/lib/src/core/http/token_manager.dart b/packages/stream_chat/lib/src/core/http/token_manager.dart index f3e855dddd..1c6e90e851 100644 --- a/packages/stream_chat/lib/src/core/http/token_manager.dart +++ b/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!; } diff --git a/packages/stream_chat/lib/src/ws/websocket.dart b/packages/stream_chat/lib/src/ws/websocket.dart index 37a29c6deb..0317953355 100644 --- a/packages/stream_chat/lib/src/ws/websocket.dart +++ b/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); } }, diff --git a/packages/stream_chat/test/src/core/http/interceptor/auth_interceptor_test.dart b/packages/stream_chat/test/src/core/http/interceptor/auth_interceptor_test.dart index fda3018af0..6f805bab53 100644 --- a/packages/stream_chat/test/src/core/http/interceptor/auth_interceptor_test.dart +++ b/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()); + 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 { diff --git a/packages/stream_chat/test/src/core/http/token_manager_test.dart b/packages/stream_chat/test/src/core/http/token_manager_test.dart index 5affb7df1a..f1ad91c199 100644 --- a/packages/stream_chat/test/src/core/http/token_manager_test.dart +++ b/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 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(); + var providerCalls = 0; + Future 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); diff --git a/packages/stream_chat/test/src/ws/websocket_test.dart b/packages/stream_chat/test/src/ws/websocket_test.dart index f4c039442a..f09798a43d 100644 --- a/packages/stream_chat/test/src/ws/websocket_test.dart +++ b/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://: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.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(); + 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://: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.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.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';