diff --git a/apps/flutter_client_contract_test_service/bin/contract_test_service.dart b/apps/flutter_client_contract_test_service/bin/contract_test_service.dart index 973fb0ab..a914c72d 100644 --- a/apps/flutter_client_contract_test_service/bin/contract_test_service.dart +++ b/apps/flutter_client_contract_test_service/bin/contract_test_service.dart @@ -34,6 +34,7 @@ class TestApiImpl extends SdkTestApi { 'client-event-source-http-errors', 'fdv1-fallback', 'client-use-post', + 'retry-conformance-fdv1-streaming', 'retry-conformance-fdv1-polling', ]; diff --git a/packages/common_client/lib/src/data_sources/data_manager.dart b/packages/common_client/lib/src/data_sources/data_manager.dart index e31d9d73..5162b728 100644 --- a/packages/common_client/lib/src/data_sources/data_manager.dart +++ b/packages/common_client/lib/src/data_sources/data_manager.dart @@ -35,6 +35,9 @@ final class FDv1DataManager implements DataManager { final loadedFromCache = await _flagManager.loadCached(context); _dataSourceManager.identify(context, completer); if (loadedFromCache && !waitForNetworkResults) { + // Nothing awaits the completer here. A later data source failure + // completes it with an error, and the status already reports it. + completer.future.ignore(); return; } return completer.future; diff --git a/packages/common_client/lib/src/data_sources/data_source.dart b/packages/common_client/lib/src/data_sources/data_source.dart index 39b690b2..66735b97 100644 --- a/packages/common_client/lib/src/data_sources/data_source.dart +++ b/packages/common_client/lib/src/data_sources/data_source.dart @@ -8,7 +8,11 @@ final class DataEvent implements DataSourceEvent { final String data; final String? environmentId; - DataEvent(this.type, this.data, {this.environmentId}); + /// The connection that produced this event, for a data source that opens + /// more than one connection. See [DataSource.restart]. + final int? generation; + + DataEvent(this.type, this.data, {this.environmentId, this.generation}); } /// A change set with typed flag descriptors, not the FDv1 JSON string forms. @@ -19,7 +23,11 @@ final class PayloadEvent implements DataSourceEvent { final ChangeSet changeSet; final String? environmentId; - PayloadEvent(this.changeSet, {this.environmentId}); + /// The connection that produced this event, for a data source that opens + /// more than one connection. See [DataSource.restart]. + final int? generation; + + PayloadEvent(this.changeSet, {this.environmentId, this.generation}); } final class StatusEvent implements DataSourceEvent { @@ -51,5 +59,10 @@ abstract interface class DataSource { /// If the data source maintains a persistent connection, then drop that /// connection and re-establish it with any appropriate delays/backoff. - void restart(); + /// + /// The [generation] is the connection that produced the payload that + /// caused the restart. A data source ignores a restart for a connection + /// that it has already replaced. A null [generation] means the current + /// connection. + void restart({int? generation}); } diff --git a/packages/common_client/lib/src/data_sources/data_source_manager.dart b/packages/common_client/lib/src/data_sources/data_source_manager.dart index 99cf05c0..878a4adb 100644 --- a/packages/common_client/lib/src/data_sources/data_source_manager.dart +++ b/packages/common_client/lib/src/data_sources/data_source_manager.dart @@ -45,7 +45,7 @@ final class DataSourceManager { final Map _dataSourceFactories = {}; DataSource? _activeDataSource; - StreamSubscription? _subscription; + StreamSubscription? _subscription; bool _stopped = false; Completer? _identifyCompleter; @@ -165,7 +165,7 @@ final class DataSourceManager { } _activeDataSource = _createDataSource(_activeConnectionMode); - _subscription = _activeDataSource?.events.asyncMap((event) async { + Future handleEvent(DataSourceEvent event) async { // Events that arrive after a stop must not change the status or // complete an identify. if (_stopped) { @@ -231,9 +231,22 @@ final class DataSourceManager { shutdown: event.shutdown); } } - }).listen((status) { - if (status is MessageStatus && status == MessageStatus.invalidMessage) { - _activeDataSource?.restart(); + return null; + } + + _subscription = _activeDataSource?.events + .asyncMap((event) async => (event, await handleEvent(event))) + .listen((handled) { + final (event, status) = handled; + if (status == MessageStatus.invalidMessage) { + // The generation stops a late result from restarting a newer + // connection. + _activeDataSource?.restart( + generation: switch (event) { + DataEvent(:final generation) => generation, + PayloadEvent(:final generation) => generation, + _ => null, + }); } }); _activeDataSource?.start(); diff --git a/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart b/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart index 8d54892e..e68f5069 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart @@ -154,7 +154,7 @@ final class FDv2DataSourceOrchestrator implements DataSource { } @override - void restart() { + void restart({int? generation}) { if (_closed) return; _logger.debug('Restart requested; recycling the active synchronizer.'); _resolveCurrentOutcome?.call(_SynchronizerOutcome.recycle); diff --git a/packages/common_client/lib/src/data_sources/fdv2/streaming_base.dart b/packages/common_client/lib/src/data_sources/fdv2/streaming_base.dart index 74995038..5150a03a 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/streaming_base.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/streaming_base.dart @@ -392,9 +392,10 @@ final class FDv2StreamingBase { // formats as 'ClientException: , uri=', and in GET // mode the URL embeds the base64-encoded context. Only the // category and a synthetic stack header go to the log. - _logger.warn('SSE error (${err.runtimeType}); will retry'); + final description = _describeError(err); + _logger.warn('$description; will retry'); _logger.debug('SSE error stack:\n$stack'); - _emit(FDv2SourceResults.interrupted(message: _describeError(err))); + _emit(FDv2SourceResults.interrupted(message: description)); } /// Categorizes an exception surfaced on the SSE stream into a fixed @@ -403,11 +404,18 @@ final class FDv2StreamingBase { /// echoes a raw http.ClientException -- whose toString carries the /// full request URL. String _describeError(Object err) { - if (err is TimeoutException) { - return 'Streaming request timed out'; - } - if (err is http.ClientException) { - return 'Network error during streaming request'; + switch (err) { + case SseTransportError(:final cause): + return _describeError(cause); + case SseTimeoutError(): + case TimeoutException(): + return 'Streaming request timed out'; + case SseContentTypeError(): + return 'Streaming response had an unexpected content type'; + case SseStreamClosedError(): + return 'Streaming connection closed by the server'; + case http.ClientException(): + return 'Network error during streaming request'; } final type = err.runtimeType.toString(); if (type.contains('Tls') || type.contains('Handshake')) { diff --git a/packages/common_client/lib/src/data_sources/null_data_source.dart b/packages/common_client/lib/src/data_sources/null_data_source.dart index 085dfb34..7e908a80 100644 --- a/packages/common_client/lib/src/data_sources/null_data_source.dart +++ b/packages/common_client/lib/src/data_sources/null_data_source.dart @@ -19,7 +19,7 @@ final class NullDataSource implements DataSource { } @override - void restart() { + void restart({int? generation}) { // No implementation needed for null data source. } } diff --git a/packages/common_client/lib/src/data_sources/polling_data_source.dart b/packages/common_client/lib/src/data_sources/polling_data_source.dart index 11b1b57e..f485b4b7 100644 --- a/packages/common_client/lib/src/data_sources/polling_data_source.dart +++ b/packages/common_client/lib/src/data_sources/polling_data_source.dart @@ -163,8 +163,11 @@ final class PollingDataSource implements DataSource { /// Records that the SDK could not apply the last payload. The data source /// fetches a full payload again after the retry delay. + /// + /// Polling makes one request at a time, and a restart applies only while + /// the last payload is pending, so the [generation] is not needed. @override - void restart() { + void restart({int? generation}) { if (_stopped || !_payloadPending) { return; } diff --git a/packages/common_client/lib/src/data_sources/streaming_data_source.dart b/packages/common_client/lib/src/data_sources/streaming_data_source.dart index f3e5cb43..edab625b 100644 --- a/packages/common_client/lib/src/data_sources/streaming_data_source.dart +++ b/packages/common_client/lib/src/data_sources/streaming_data_source.dart @@ -8,6 +8,7 @@ import '../config/data_source_config.dart'; import '../config/defaults/default_config.dart'; import 'data_source.dart'; import 'data_source_status.dart'; +import 'fdv2/payload.dart'; import 'get_environment_id.dart'; import 'requestor.dart'; @@ -27,17 +28,25 @@ typedef SseClientFactory = SSEClient Function( SSEClient _defaultClientFactory(Uri uri, HttpProperties httpProperties, String? body, SseHttpMethod? method, EventSourceLogger? logger) { + // The data source owns the retry, so the client must not reconnect by + // itself. return SSEClient(uri, {_putEvent, _patchEvent, _delete, _pingEvent}, headers: httpProperties.baseHeaders, body: body, httpMethod: method ?? SseHttpMethod.get, - logger: logger); + logger: logger, + autoReconnect: false); } HttpClient _defaultHttpClientFactory(HttpProperties httpProperties) { return HttpClient(httpProperties: httpProperties); } +/// A data source that receives flag updates over a streaming connection. +/// +/// The data source makes a new SSE client for each connection attempt. +/// After a failure it closes the client, waits for the delay from its +/// [RetryState], and then connects again. No failure stops it. final class StreamingDataSource implements DataSource { final LDLogger _logger; @@ -66,18 +75,25 @@ final class StreamingDataSource implements DataSource { late final Requestor _requestor; + final RetryState _retryState; + + Timer? _reconnectTimer; + + /// Counts the connections. Events carry it, so that a restart caused by an + /// older connection does not close a newer one. + int _generation = 0; + @override Stream get events => _dataController.stream; - /// Used to track if there has been an unrecoverable error. - bool _permanentShutdown = false; - /// The [clientFactory] parameter is primarily intended for testing, but it also /// could be used for customized SSE clients which support functionality /// our default client support does not, or for alternative implementations - /// which are not based on SSE. + /// which are not based on SSE. A client from the factory must not + /// reconnect by itself. /// The [httpClientFactory] parameter is primarily intended for testing the /// requestor used for ping events. + /// The [retryState] parameter is for testing. StreamingDataSource( {required String credential, required LDContext context, @@ -87,12 +103,14 @@ final class StreamingDataSource implements DataSource { required PollingDataSourceConfig pollingDataSourceConfig, required HttpProperties httpProperties, SseClientFactory clientFactory = _defaultClientFactory, - HttpClientFactory? httpClientFactory}) + HttpClientFactory? httpClientFactory, + RetryState? retryState}) : _logger = logger.subLogger('StreamingDataSource'), _dataSourceConfig = dataSourceConfig, _clientFactory = clientFactory, _httpProperties = httpProperties, - _credential = credential { + _credential = credential, + _retryState = retryState ?? RetryState.streaming() { final plainContextString = jsonEncode(LDContextSerialization.toJson(context, isEvent: false)); @@ -134,25 +152,36 @@ final class StreamingDataSource implements DataSource { @override void start() { - if (_subscription != null || _permanentShutdown || _stopped) { + if (_subscription != null || _reconnectTimer != null || _stopped) { return; } - _stopped = false; + _connect(); + } + + void _connect() { + _reconnectTimer = null; + if (_stopped) { + return; + } + _generation++; + final generation = _generation; _logger.debug('Establishing new streaming connection, uri: $_uri'); - _client = _clientFactory( + final client = _clientFactory( _uri, _httpProperties, _useReport ? _contextString : null, _useReport ? SseHttpMethod.report : SseHttpMethod.get, LDLoggerToEventSourceAdapter(_logger)); + _client = client; - _subscription = _client!.stream.listen((event) async { + _subscription = client.stream.listen((event) async { if (_stopped) { return; } switch (event) { case MessageEvent(): + _retryState.recordSuccess(); if (event.type == _pingEvent) { final res = await _requestor.requestAllFlags(); if (_stopped) { @@ -160,18 +189,16 @@ final class StreamingDataSource implements DataSource { } switch (res) { case null: - // No update, so things stay the same. - return; + // No update. A change set of type none restores the valid + // status. + _dataController.sink.add(PayloadEvent( + const ChangeSet(type: PayloadType.none, updates: {}), + generation: generation)); case DataEvent(): - _dataController.sink.add(res); + _dataController.sink.add(DataEvent(res.type, res.data, + environmentId: res.environmentId, generation: generation)); case StatusEvent(): - final message = res.kind == ErrorKind.errorResponse - ? 'received unexpected status code when polling in response to a ping event' - : 'encountered error with polling request in response to a ping event'; - final argument = res.kind == ErrorKind.errorResponse - ? res.statusCode - : res.message; - _logger.error('$message: $argument'); + _logPingPollFailure(res); _dataController.sink.add(res); case PayloadEvent(): case InitializedEvent(): @@ -182,7 +209,7 @@ final class StreamingDataSource implements DataSource { } else { _logger.debug('Received message event, data: ${event.data}'); _dataController.sink.add(DataEvent(event.type, event.data, - environmentId: _environmentId)); + environmentId: _environmentId, generation: generation)); } case OpenEvent(): _logger.debug('Received connect event, data: ${event.headers}'); @@ -194,39 +221,114 @@ final class StreamingDataSource implements DataSource { } } }) - ..onError((err) { - if (_permanentShutdown) { - return; - } - // The SSE client retries recoverable responses on its own (and - // logs them), so this source is not shutting down -- let the retry - // run. - if (err is SseHttpError && err.recoverable) { - return; - } - _permanentShutdown = true; - _logger.error( - 'Encountered an unrecoverable error: "$err", Shutting down.'); - // The event must be added before calling stop, because stop closes - // the controller and adding an event after that throws a StateError. - _dataController.sink.add(StatusEvent(ErrorKind.unknown, null, - 'Encountered unrecoverable error streaming')); - stop(); - }); + ..onError(_handleError); } - @override - void restart() { - _client?.restart(); + void _logPingPollFailure(StatusEvent status) { + if (status.statusCode case final statusCode?) { + switch (classifyHttpStatus(statusCode)) { + case FailureKind.unexpected: + _logger.error('Received unexpected status code $statusCode when ' + 'polling in response to a ping event.' + '${credentialHint(statusCode)}'); + case FailureKind.normal: + _logger.warn('Received status code $statusCode when polling in ' + 'response to a ping event.'); + } + return; + } + _logger.warn('Encountered error with polling request in response to a ' + 'ping event: ${status.message}'); } - @override - void stop() { + void _handleError(Object err) { + if (_stopped) { + return; + } + final StatusEvent status; + final Duration delay; + if (err is SseHttpError) { + final kind = classifyHttpStatus(err.statusCode); + delay = _retryState.recordFailure(kind); + status = StatusEvent(ErrorKind.errorResponse, err.statusCode, + 'Received unexpected status code: ${err.statusCode}'); + switch (kind) { + case FailureKind.unexpected: + _logger.error('Received unexpected status code ${err.statusCode} ' + 'for the streaming connection.${credentialHint(err.statusCode)} ' + 'Will retry in ${delay.inMilliseconds} ms.'); + case FailureKind.normal: + _logger.warn('Received status code ${err.statusCode} for the ' + 'streaming connection. Will retry in ${delay.inMilliseconds} ' + 'ms.'); + } + } else { + delay = _retryState.recordFailure(FailureKind.normal); + final message = _describeError(err); + status = StatusEvent(ErrorKind.networkError, null, message); + _logger.warn('$message. Will retry in ${delay.inMilliseconds} ms.'); + } + _closeConnection(); + _dataController.sink.add(status); + _scheduleReconnect(delay); + } + + /// Categorizes an error from the SSE client into a fixed message. The + /// string form of a transport error can contain the request URL, which + /// holds the context in GET mode. + String _describeError(Object err) => switch (err) { + SseTimeoutError(phase: SseTimeoutPhase.connect) => + 'The streaming connection timed out while connecting', + SseTimeoutError(phase: SseTimeoutPhase.read) => + 'The streaming connection was silent for too long', + SseContentTypeError() => + 'The streaming connection received an unexpected content type', + SseStreamClosedError() => 'The streaming connection was closed', + _ => 'The streaming connection failed', + }; + + void _closeConnection() { // Cancel is async, but it should only be for the cleanup portion, according // to the method documentation. _subscription?.cancel(); _subscription = null; + _client?.close(); + _client = null; + } + + void _scheduleReconnect(Duration delay) { + _reconnectTimer?.cancel(); + _reconnectTimer = Timer(delay, _connect); + } + + @override + void restart({int? generation}) { + // The payload that caused the restart can come from a ping poll, also + // from a replaced connection. Forget its ETag, so that the next ping poll + // fetches a full payload. + _requestor.clearEtag(); + // A restart during a wait has no effect, because the data source is + // already going to reconnect. + if (_stopped || _client == null) { + return; + } + if (generation != null && generation != _generation) { + _logger.debug('Ignoring a restart for a replaced connection.'); + return; + } + final delay = _retryState.recordFailure(FailureKind.normal); + _logger.warn('Restarting the streaming connection. Will retry in ' + '${delay.inMilliseconds} ms.'); + _closeConnection(); + _scheduleReconnect(delay); + } + + @override + void stop() { _stopped = true; + _reconnectTimer?.cancel(); + _reconnectTimer = null; + _closeConnection(); _dataController.close(); } } diff --git a/packages/common_client/test/data_sources/data_source_manager_test.dart b/packages/common_client/test/data_sources/data_source_manager_test.dart index 35aa0eee..7632baae 100644 --- a/packages/common_client/test/data_sources/data_source_manager_test.dart +++ b/packages/common_client/test/data_sources/data_source_manager_test.dart @@ -42,7 +42,7 @@ final class MockDataSource implements DataSource { } @override - void restart() { + void restart({int? generation}) { restartCalled = true; Future.delayed(Duration(milliseconds: 10)) .then((_) => controller.sink.add(DataEvent('put', '{}'))); diff --git a/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart b/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart index 55b5b996..9faadc40 100644 --- a/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart +++ b/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart @@ -446,6 +446,73 @@ void main() { await sub.cancel(); }); + for (final (error, message) in [ + ( + const SseTimeoutError(SseTimeoutPhase.connect), + 'Streaming request timed out' + ), + ( + const SseTimeoutError(SseTimeoutPhase.read), + 'Streaming request timed out' + ), + ( + SseTransportError(http.ClientException('reset', Uri.parse('/x'))), + 'Network error during streaming request' + ), + ( + const SseContentTypeError(), + 'Streaming response had an unexpected content type' + ), + ( + const SseStreamClosedError(), + 'Streaming connection closed by the server' + ), + ]) { + test('$error is reported as interrupted and keeps the source open', + () async { + final sse = makeSse(); + final base = makeBase(sse); + final emissions = []; + final sub = base.results.listen(emissions.add); + await Future.delayed(Duration.zero); + + sse.emitError(error: error); + await Future.delayed(Duration.zero); + + final status = emissions.single as StatusResult; + expect(status.state, equals(SourceState.interrupted)); + expect(status.message, equals(message)); + expect(sse.isClosed, isFalse); + + await sub.cancel(); + }); + + test('$error is logged as a warning that names the failure', () async { + final adapter = MockLogAdapter(); + when(() => adapter.log(any())).thenReturn(null); + final sse = makeSse(); + final base = FDv2StreamingBase( + sseClient: sse, + pingHandler: () async => + FDv2SourceResults.interrupted(message: 'no ping'), + logger: LDLogger(adapter: adapter, level: LDLogLevel.debug), + ); + final sub = base.results.listen((_) {}); + await Future.delayed(Duration.zero); + + sse.emitError(error: error); + await Future.delayed(Duration.zero); + + final warnings = verify(() => adapter.log(captureAny())) + .captured + .cast() + .where((record) => record.level == LDLogLevel.warn); + expect(warnings.map((record) => record.message), [contains(message)]); + + await sub.cancel(); + }); + } + test( 'SSE transport error log records do not echo the request URL ' 'or any other detail of the underlying exception', () async { diff --git a/packages/common_client/test/data_sources/polling_data_source_manager_test.dart b/packages/common_client/test/data_sources/polling_data_source_manager_test.dart index 9bf7d6e9..681c48fa 100644 --- a/packages/common_client/test/data_sources/polling_data_source_manager_test.dart +++ b/packages/common_client/test/data_sources/polling_data_source_manager_test.dart @@ -9,9 +9,11 @@ import 'package:launchdarkly_common_client/src/data_sources/data_source_event_ha import 'package:launchdarkly_common_client/src/data_sources/data_source_manager.dart'; import 'package:launchdarkly_common_client/src/data_sources/data_source_status_manager.dart'; import 'package:launchdarkly_common_client/src/data_sources/polling_data_source.dart'; +import 'package:launchdarkly_common_client/src/data_sources/streaming_data_source.dart'; import 'package:launchdarkly_common_client/src/flag_manager/flag_manager.dart'; import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart' as ld_common; +import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart'; import 'package:test/test.dart'; const _interval = Duration(seconds: 30); @@ -39,6 +41,7 @@ final class _Harness { DataSourceStatusManager(stamper: () => DateTime(1)); final List requests = []; final List requestTimes = []; + final List sseClients = []; late final FlagManager flagManager; late final DataSourceManager manager; final LDContext context = LDContextBuilder().kind('user', 'test').build(); @@ -78,6 +81,20 @@ final class _Harness { httpProperties: HttpProperties(), testingInterval: _interval, httpClientFactory: httpClientFactory), + const FDv2Streaming(): (context) => StreamingDataSource( + credential: 'dummy-key', + context: context, + endpoints: ServiceEndpoints(), + logger: logger, + dataSourceConfig: StreamingDataSourceConfig(), + pollingDataSourceConfig: PollingDataSourceConfig(), + httpProperties: HttpProperties(), + httpClientFactory: httpClientFactory, + clientFactory: (uri, properties, body, method, logger) { + final sse = SSEClient.testClient(uri, {}); + sseClients.add(sse); + return sse; + }), }); manager.identify(context, Completer()..future.ignore()); } @@ -205,4 +222,95 @@ void main() { }); }); }); + + group('ping poll', () { + test('a ping poll after an unusable payload fetches fresh data', () { + fakeAsync((async) { + final harness = + _Harness(async, mode: ConnectionMode.streaming, (index, request) { + if (index == 0) { + return _ok('not a payload', '"e1"'); + } + return _hasEtag(request) + ? http.Response('', 304) + : _ok(_flagPayload, '"e2"'); + }); + async.flushMicrotasks(); + harness.sseClients.last.emitEvent(MessageEvent('ping', '', null)); + async.elapse(const Duration(seconds: 5)); + + // The unusable payload restarts the stream. Ping the new stream. + harness.sseClients.last.emitEvent(MessageEvent('ping', '', null)); + async.flushMicrotasks(); + + expect(harness.requests, hasLength(2)); + expect(_hasEtag(harness.requests[1]), isFalse); + expect(harness.flagValue, isTrue); + harness.manager.stop(); + }); + }); + + test( + 'a rejected ping poll payload from a replaced connection clears the ' + 'ETag', () { + fakeAsync((async) { + final firstResponse = Completer(); + final harness = + _Harness(async, mode: ConnectionMode.streaming, (index, request) { + if (index == 0) { + return firstResponse.future; + } + return _hasEtag(request) + ? http.Response('', 304) + : _ok(_flagPayload, '"e2"'); + }); + async.flushMicrotasks(); + + // The ping poll is in flight when the stream fails. + harness.sseClients[0].emitEvent(MessageEvent('ping', '', null)); + async.flushMicrotasks(); + harness.sseClients[0].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + + // The poll returns an unusable payload with an ETag, and the SDK + // rejects it. + firstResponse.complete(_ok('not a payload', '"e1"')); + async.flushMicrotasks(); + + async.elapse(const Duration(seconds: 1)); + expect(harness.sseClients, hasLength(2)); + harness.sseClients[1].emitEvent(MessageEvent('ping', '', null)); + async.flushMicrotasks(); + + expect(harness.requests, hasLength(2)); + expect(_hasEtag(harness.requests[1]), isFalse); + harness.manager.stop(); + }); + }); + + test('a 304 for a ping poll after a failure restores the valid status', () { + fakeAsync((async) { + final harness = + _Harness(async, mode: ConnectionMode.streaming, (index, request) { + return switch (index) { + 0 => _ok(_flagPayload, '"e1"'), + 1 => http.Response('', 503), + _ => http.Response('', 304), + }; + }); + async.flushMicrotasks(); + final sse = harness.sseClients.single; + for (final expected in [ + DataSourceState.valid, + DataSourceState.interrupted, + DataSourceState.valid, + ]) { + sse.emitEvent(MessageEvent('ping', '', null)); + async.flushMicrotasks(); + expect(harness.statusManager.status.state, expected); + } + harness.manager.stop(); + }); + }); + }); } diff --git a/packages/common_client/test/data_sources/streaming_data_source_generation_test.dart b/packages/common_client/test/data_sources/streaming_data_source_generation_test.dart new file mode 100644 index 00000000..326988e8 --- /dev/null +++ b/packages/common_client/test/data_sources/streaming_data_source_generation_test.dart @@ -0,0 +1,141 @@ +import 'dart:async'; +import 'dart:math' as math; + +import 'package:fake_async/fake_async.dart'; +import 'package:launchdarkly_common_client/launchdarkly_common_client.dart'; +import 'package:launchdarkly_common_client/src/config/data_source_config.dart'; +import 'package:launchdarkly_common_client/src/data_sources/data_source_event_handler.dart'; +import 'package:launchdarkly_common_client/src/data_sources/data_source_manager.dart'; +import 'package:launchdarkly_common_client/src/data_sources/data_source_status_manager.dart'; +import 'package:launchdarkly_common_client/src/data_sources/streaming_data_source.dart'; +import 'package:launchdarkly_common_client/src/flag_manager/flag_manager.dart'; +import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart' + show RetryState; +import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart'; +import 'package:test/test.dart'; + +/// A random source that removes no jitter. +final class _NoJitter implements math.Random { + @override + double nextDouble() => 0; + + @override + bool nextBool() => false; + + @override + int nextInt(int max) => 0; +} + +/// A persistence whose writes wait for [gate] and then fail. +final class _GatedFailingPersistence implements Persistence { + final Completer gate = Completer(); + + @override + Future read(String namespace, String key) async => null; + + @override + Future remove(String namespace, String key) async {} + + @override + Future set(String namespace, String key, String data) async { + await gate.future; + throw StateError('write failed'); + } +} + +/// Runs a streaming data source behind a real [DataSourceManager]. +final class _Harness { + final List clients = []; + late final DataSourceManager manager; + + _Harness(FakeAsync async, {Persistence? persistence}) { + final logger = LDLogger(level: LDLogLevel.none); + final statusManager = DataSourceStatusManager(); + final flagManager = FlagManager( + sdkKey: 'dummy-key', + maxCachedContexts: 5, + logger: logger, + persistence: persistence); + manager = DataSourceManager( + statusManager: statusManager, + dataSourceEventHandler: DataSourceEventHandler( + flagManager: flagManager, + statusManager: statusManager, + logger: logger), + logger: logger); + manager.setFactories({ + const FDv2Streaming(): (context) => StreamingDataSource( + credential: 'dummy-key', + context: context, + endpoints: ServiceEndpoints(), + logger: logger, + dataSourceConfig: StreamingDataSourceConfig(), + pollingDataSourceConfig: PollingDataSourceConfig(), + httpProperties: HttpProperties(), + retryState: RetryState.streaming( + random: _NoJitter(), clock: () => async.elapsed), + clientFactory: (uri, properties, body, method, logger) { + final client = SSEClient.testClient(uri, {}); + clients.add(client); + return client; + }), + }); + manager.identify(LDContextBuilder().kind('user', 'test').build(), + Completer()..future.ignore()); + } +} + +void main() { + test('a late restart from a replaced connection is ignored', () { + fakeAsync((async) { + final persistence = _GatedFailingPersistence(); + final harness = _Harness(async, persistence: persistence); + async.flushMicrotasks(); + expect(harness.clients, hasLength(1)); + + // Connection 1 delivers a payload. Storing it waits on the gate. + harness.clients[0].emitEvent(MessageEvent('put', '{}', null)); + async.flushMicrotasks(); + + // Connection 1 fails, and connection 2 opens after 1 second. + harness.clients[0].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 1)); + expect(harness.clients, hasLength(2)); + + // The payload from connection 1 now fails to store, which asks for a + // restart. + persistence.gate.complete(); + async.flushMicrotasks(); + + expect(harness.clients[1].isClosed, isFalse); + expect(harness.clients, hasLength(2)); + + // Only one failure is recorded, so the next failure waits 2 seconds + // and not 4. + harness.clients[1].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + async.elapse(const Duration(milliseconds: 1999)); + expect(harness.clients, hasLength(2)); + async.elapse(const Duration(milliseconds: 1)); + expect(harness.clients, hasLength(3)); + harness.manager.stop(); + }); + }); + + test('a restart from the current connection restarts it', () { + fakeAsync((async) { + final harness = _Harness(async); + async.flushMicrotasks(); + + harness.clients[0].emitEvent(MessageEvent('put', 'not a payload', null)); + async.flushMicrotasks(); + + expect(harness.clients[0].isClosed, isTrue); + async.elapse(const Duration(seconds: 1)); + expect(harness.clients, hasLength(2)); + expect(harness.clients[1].isClosed, isFalse); + harness.manager.stop(); + }); + }); +} diff --git a/packages/common_client/test/data_sources/streaming_data_source_reset_test.dart b/packages/common_client/test/data_sources/streaming_data_source_reset_test.dart new file mode 100644 index 00000000..01e57b85 --- /dev/null +++ b/packages/common_client/test/data_sources/streaming_data_source_reset_test.dart @@ -0,0 +1,108 @@ +import 'dart:math' as math; + +import 'package:fake_async/fake_async.dart'; +import 'package:launchdarkly_common_client/launchdarkly_common_client.dart'; +import 'package:launchdarkly_common_client/src/config/data_source_config.dart'; +import 'package:launchdarkly_common_client/src/data_sources/streaming_data_source.dart'; +import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart' + show RetryState; +import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart'; +import 'package:test/test.dart'; + +/// A random source that removes no jitter. +final class _NoJitter implements math.Random { + @override + double nextDouble() => 0; + + @override + bool nextBool() => false; + + @override + int nextInt(int max) => 0; +} + +(StreamingDataSource, List) _makeDataSource(FakeAsync async) { + final clients = []; + final dataSource = StreamingDataSource( + credential: 'dummy-key', + context: LDContextBuilder().kind('user', 'test').build(), + endpoints: ServiceEndpoints(), + logger: LDLogger(level: LDLogLevel.none), + dataSourceConfig: StreamingDataSourceConfig(), + pollingDataSourceConfig: PollingDataSourceConfig(), + httpProperties: HttpProperties(), + retryState: + RetryState.streaming(random: _NoJitter(), clock: () => async.elapsed), + clientFactory: (uri, properties, body, method, logger) { + final client = SSEClient.testClient(uri, {}); + clients.add(client); + return client; + }); + dataSource.events.listen((_) {}); + return (dataSource, clients); +} + +/// Fails the first stream with a 401, then opens a second stream that +/// receives a payload. +void _failThenReceive(FakeAsync async, List clients) { + clients[0].emitError(error: SseHttpError(401, {}, recoverable: false)); + async.flushMicrotasks(); + async.elapse(const Duration(minutes: 5)); + expect(clients, hasLength(2)); + clients[1].emitEvent(MessageEvent('put', '{}', null)); + async.flushMicrotasks(); +} + +void main() { + test('60 seconds of healthy streaming resets the retry state', () { + fakeAsync((async) { + final (dataSource, clients) = _makeDataSource(async); + dataSource.start(); + async.flushMicrotasks(); + _failThenReceive(async, clients); + + async.elapse(const Duration(seconds: 60)); + clients[1].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + + async.elapse(const Duration(seconds: 1)); + expect(clients, hasLength(3)); + }); + }); + + test('a stream that fails before 60 seconds keeps the retry state', () { + fakeAsync((async) { + final (dataSource, clients) = _makeDataSource(async); + dataSource.start(); + async.flushMicrotasks(); + _failThenReceive(async, clients); + + async.elapse(const Duration(seconds: 59)); + clients[1].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + + async.elapse(const Duration(minutes: 9)); + expect(clients, hasLength(2)); + async.elapse(const Duration(minutes: 1)); + expect(clients, hasLength(3)); + }); + }); + + test('a stream with no payload does not count as healthy', () { + fakeAsync((async) { + final (dataSource, clients) = _makeDataSource(async); + dataSource.start(); + async.flushMicrotasks(); + clients[0].emitError(error: SseHttpError(401, {}, recoverable: false)); + async.flushMicrotasks(); + async.elapse(const Duration(minutes: 5)); + + async.elapse(const Duration(minutes: 2)); + clients[1].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + + async.elapse(const Duration(minutes: 9)); + expect(clients, hasLength(2)); + }); + }); +} diff --git a/packages/common_client/test/data_sources/streaming_data_source_retry_test.dart b/packages/common_client/test/data_sources/streaming_data_source_retry_test.dart new file mode 100644 index 00000000..f5c50f22 --- /dev/null +++ b/packages/common_client/test/data_sources/streaming_data_source_retry_test.dart @@ -0,0 +1,315 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:launchdarkly_common_client/launchdarkly_common_client.dart'; +import 'package:launchdarkly_common_client/src/config/data_source_config.dart'; +import 'package:launchdarkly_common_client/src/data_sources/data_source.dart'; +import 'package:launchdarkly_common_client/src/data_sources/data_source_status.dart'; +import 'package:launchdarkly_common_client/src/data_sources/streaming_data_source.dart'; +import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart' + as ld_common; +import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:test/test.dart'; + +class _MockLogAdapter extends Mock implements LDLogAdapter {} + +const _credentialHint = + 'Check that the SDK credential (mobile key or client-side ID) is correct.'; + +/// Holds a streaming data source and every SSE client that it makes. +final class _Harness { + final StreamingDataSource dataSource; + final List clients; + final List events; + + _Harness(this.dataSource, this.clients, this.events); + + List get statuses => events.whereType().toList(); +} + +_Harness _makeHarness({MockClient? pollingClient, LDLogger? logger}) { + final clients = []; + final events = []; + final dataSource = StreamingDataSource( + credential: 'dummy-key', + context: LDContextBuilder().kind('user', 'test').build(), + endpoints: ServiceEndpoints(), + logger: logger ?? LDLogger(level: LDLogLevel.none), + dataSourceConfig: StreamingDataSourceConfig(), + pollingDataSourceConfig: PollingDataSourceConfig(), + httpProperties: HttpProperties(), + httpClientFactory: pollingClient != null + ? (properties) => ld_common.HttpClient( + client: pollingClient, httpProperties: properties) + : null, + clientFactory: (uri, properties, body, method, logger) { + final client = SSEClient.testClient(uri, {}); + clients.add(client); + return client; + }); + dataSource.events.listen(events.add); + return _Harness(dataSource, clients, events); +} + +void main() { + setUpAll(() { + registerFallbackValue(LDLogRecord( + level: LDLogLevel.debug, + message: '', + time: DateTime.now(), + logTag: '')); + }); + + /// Sends a ping whose poll gets [status], and returns the records logged + /// about it. + List pingPollLogs(int status) { + final adapter = _MockLogAdapter(); + final records = []; + fakeAsync((async) { + final harness = _makeHarness( + pollingClient: + MockClient((request) async => http.Response('', status)), + logger: LDLogger(adapter: adapter, level: LDLogLevel.debug)); + harness.dataSource.start(); + async.flushMicrotasks(); + harness.clients[0].emitEvent(MessageEvent('ping', '', null)); + async.flushMicrotasks(); + harness.dataSource.stop(); + records.addAll(verify(() => adapter.log(captureAny())) + .captured + .cast() + .where((record) => record.message.contains('ping'))); + }); + return records; + } + + test('a normal failure of the poll after a ping logs a warning', () { + final records = pingPollLogs(503); + expect(records.map((record) => record.level), [LDLogLevel.warn]); + }); + + for (final status in [401, 403]) { + test('a $status for the poll after a ping logs an error with the hint', () { + final records = pingPollLogs(status); + expect(records.map((record) => record.level), [LDLogLevel.error]); + expect(records.single.message, contains(_credentialHint)); + }); + } + + test('a 404 for the poll after a ping logs an error without the hint', () { + final records = pingPollLogs(404); + expect(records.map((record) => record.level), [LDLogLevel.error]); + expect(records.single.message, isNot(contains('Check that'))); + }); + + /// Fails the stream with [status], and returns the records logged about + /// it. + List streamFailureLogs(int status) { + final adapter = _MockLogAdapter(); + final records = []; + fakeAsync((async) { + final harness = _makeHarness( + logger: LDLogger(adapter: adapter, level: LDLogLevel.debug)); + harness.dataSource.start(); + async.flushMicrotasks(); + harness.clients[0] + .emitError(error: SseHttpError(status, {}, recoverable: false)); + async.flushMicrotasks(); + harness.dataSource.stop(); + records.addAll(verify(() => adapter.log(captureAny())) + .captured + .cast() + .where((record) => record.message.contains('$status'))); + }); + return records; + } + + for (final status in [401, 403]) { + test('a $status for the stream logs an error with the hint', () { + final records = streamFailureLogs(status); + expect(records.map((record) => record.level), [LDLogLevel.error]); + expect(records.single.message, contains(_credentialHint)); + expect(records.single.message, contains('Will retry in')); + }); + } + + test('a 404 for the stream logs an error without the hint', () { + final records = streamFailureLogs(404); + expect(records.map((record) => record.level), [LDLogLevel.error]); + expect(records.single.message, isNot(contains('Check that'))); + expect(records.single.message, contains('Will retry in')); + }); + + test('a 401 reconnects after the extended initial delay', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + expect(harness.clients, hasLength(1)); + + harness.clients[0] + .emitError(error: SseHttpError(401, {}, recoverable: false)); + async.flushMicrotasks(); + + expect(harness.statuses, hasLength(1)); + expect(harness.statuses[0].statusCode, 401); + expect(harness.statuses[0].shutdown, isFalse); + expect(harness.clients[0].isClosed, isTrue); + + // The extended initial delay is 5 minutes, less up to half as jitter. + async.elapse(const Duration(seconds: 149)); + expect(harness.clients, hasLength(1)); + async.elapse(const Duration(seconds: 151)); + expect(harness.clients, hasLength(2)); + }); + }); + + test('a 503 reconnects after the normal initial delay', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + + harness.clients[0] + .emitError(error: SseHttpError(503, {}, recoverable: false)); + async.flushMicrotasks(); + expect(harness.statuses.single.statusCode, 503); + + async.elapse(const Duration(milliseconds: 499)); + expect(harness.clients, hasLength(1)); + async.elapse(const Duration(milliseconds: 501)); + expect(harness.clients, hasLength(2)); + }); + }); + + test('a transport failure backs off and reports a network error', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + + harness.clients[0].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + expect(harness.statuses.single.kind, ErrorKind.networkError); + expect(harness.statuses.single.shutdown, isFalse); + + async.elapse(const Duration(seconds: 1)); + expect(harness.clients, hasLength(2)); + + // The second normal failure doubles the base delay to 2 seconds. + harness.clients[1].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + async.elapse(const Duration(milliseconds: 999)); + expect(harness.clients, hasLength(2)); + async.elapse(const Duration(milliseconds: 1001)); + expect(harness.clients, hasLength(3)); + }); + }); + + test('a normal failure after a 401 keeps the extended delay', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + + harness.clients[0] + .emitError(error: SseHttpError(401, {}, recoverable: false)); + async.flushMicrotasks(); + async.elapse(const Duration(minutes: 5)); + expect(harness.clients, hasLength(2)); + + // The second extended attempt has a base delay of 10 minutes. + harness.clients[1].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 299)); + expect(harness.clients, hasLength(2)); + async.elapse(const Duration(seconds: 301)); + expect(harness.clients, hasLength(3)); + }); + }); + + test('stop interrupts a pending wait', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + + harness.clients[0] + .emitError(error: SseHttpError(401, {}, recoverable: false)); + async.flushMicrotasks(); + + harness.dataSource.stop(); + expect(async.pendingTimers, isEmpty); + + async.elapse(const Duration(hours: 2)); + expect(harness.clients, hasLength(1)); + expect(harness.statuses, hasLength(1)); + }); + }); + + test('start after stop does not connect', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + harness.dataSource.stop(); + harness.dataSource.start(); + async.elapse(const Duration(hours: 2)); + expect(harness.clients, hasLength(1)); + }); + }); + + test('no status is reported for a ping poll that ends after stop', () { + fakeAsync((async) { + final response = Completer(); + final harness = + _makeHarness(pollingClient: MockClient((request) => response.future)); + harness.dataSource.start(); + async.flushMicrotasks(); + + harness.clients[0].emitEvent(MessageEvent('ping', '', null)); + async.flushMicrotasks(); + harness.dataSource.stop(); + response.complete(http.Response('', 401)); + async.flushMicrotasks(); + + expect(harness.events, isEmpty); + }); + }); + + test('restart reconnects with the normal delay', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + harness.clients[0].emitEvent(MessageEvent('put', '{}', null)); + async.flushMicrotasks(); + + harness.dataSource.restart(); + expect(harness.clients[0].isClosed, isTrue); + + async.elapse(const Duration(seconds: 1)); + expect(harness.clients, hasLength(2)); + expect(harness.statuses, isEmpty); + }); + }); + + test('restart during a wait does not make another connection', () { + fakeAsync((async) { + final harness = _makeHarness(); + harness.dataSource.start(); + async.flushMicrotasks(); + harness.clients[0].emitError(error: Exception('connection reset')); + async.flushMicrotasks(); + + harness.dataSource.restart(); + async.elapse(const Duration(seconds: 1)); + expect(harness.clients, hasLength(2)); + async.elapse(const Duration(minutes: 1)); + expect(harness.clients, hasLength(2)); + }); + }); +} diff --git a/packages/common_client/test/data_sources/streaming_data_source_test.dart b/packages/common_client/test/data_sources/streaming_data_source_test.dart index dc09068b..52abf11c 100644 --- a/packages/common_client/test/data_sources/streaming_data_source_test.dart +++ b/packages/common_client/test/data_sources/streaming_data_source_test.dart @@ -507,7 +507,7 @@ void main() { group('stream error handling', () { test( 'it emits a status event without throwing when the stream reports an ' - 'unrecoverable error', () async { + 'error', () async { final controller = StreamController(); final (dataSource, _, statusManager) = makeDataSourceForTest(controller.stream); @@ -515,14 +515,14 @@ void main() { final statusChange = expectLater( statusManager.changes, emits(predicate((status) => - status.lastError?.kind == ErrorKind.unknown && - status.lastError?.message == - 'Encountered unrecoverable error streaming'))); + status.lastError?.kind == ErrorKind.networkError && + status.lastError?.message == 'The streaming connection failed'))); dataSource.start(); controller.sink.addError(Exception('Unable to make request')); await statusChange; + dataSource.stop(); }); }); diff --git a/packages/common_client/test/hooks/ld_common_client_hooks_test.dart b/packages/common_client/test/hooks/ld_common_client_hooks_test.dart index ce1d0dc6..cf66149f 100644 --- a/packages/common_client/test/hooks/ld_common_client_hooks_test.dart +++ b/packages/common_client/test/hooks/ld_common_client_hooks_test.dart @@ -83,7 +83,7 @@ final class TestDataSourceWithEnvironmentId implements DataSource { Stream get events => _eventController.stream; @override - void restart() {} + void restart({int? generation}) {} @override void start() { diff --git a/packages/common_client/test/identify_wait_test.dart b/packages/common_client/test/identify_wait_test.dart index ff247dc9..d4e78f08 100644 --- a/packages/common_client/test/identify_wait_test.dart +++ b/packages/common_client/test/identify_wait_test.dart @@ -8,8 +8,12 @@ import 'package:launchdarkly_common_client/src/config/data_source_config.dart'; import 'package:launchdarkly_common_client/src/data_sources/data_source.dart'; import 'package:launchdarkly_common_client/src/data_sources/data_source_status.dart'; import 'package:launchdarkly_common_client/src/data_sources/polling_data_source.dart'; +import 'package:launchdarkly_common_client/src/data_sources/streaming_data_source.dart'; +import 'package:launchdarkly_common_client/src/persistence/persistence.dart' + show encodePersistenceKey; import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart' as ld_common; +import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart'; import 'package:test/test.dart'; import 'mock_persistence.dart'; @@ -37,7 +41,8 @@ String _userKeyOf(http.Request request) { /// Makes a client that polls every step. Each context gets the statuses in /// [statusesByKey] in order, and the last status repeats. A 200 response /// carries a flag whose value is the user key. -LDCommonClient _pollingClient(Map> statusesByKey) { +LDCommonClient _pollingClient(Map> statusesByKey, + {Persistence? persistence}) { final requestsByKey = {}; final httpClient = MockClient((request) async { final key = _userKeyOf(request); @@ -54,7 +59,7 @@ LDCommonClient _pollingClient(Map> statusesByKey) { _TestConfig('the-sdk-key', AutoEnvAttributes.disabled, dataSourceConfig: DataSourceConfig(initialConnectionMode: ConnectionMode.polling)), - CommonPlatform(persistence: MockPersistence()), + CommonPlatform(persistence: persistence ?? MockPersistence()), _user('start'), DiagnosticSdkData(name: '', version: ''), dataSourceFactories: (config, logger, properties) { @@ -88,7 +93,7 @@ final class _ScriptedDataSource implements DataSource { Stream get events => _controller.stream; @override - void restart() {} + void restart({int? generation}) {} @override void start() { @@ -179,4 +184,152 @@ void main() { expect(client.dataSourceStatus.state, DataSourceState.valid); await client.close(); }); + + test( + 'a stream failure fails start, and then the reconnected stream updates ' + 'the flags', () async { + final sseClients = []; + final client = LDCommonClient( + _TestConfig('the-sdk-key', AutoEnvAttributes.disabled), + CommonPlatform(persistence: MockPersistence()), + _user('c'), + DiagnosticSdkData(name: '', version: ''), + dataSourceFactories: (config, logger, properties) { + StreamingDataSource streaming(LDContext context) => StreamingDataSource( + credential: 'the-sdk-key', + context: context, + endpoints: ServiceEndpoints(), + logger: logger, + dataSourceConfig: StreamingDataSourceConfig(), + pollingDataSourceConfig: PollingDataSourceConfig(), + httpProperties: properties, + clientFactory: (uri, properties, body, method, logger) { + final sse = SSEClient.testClient(uri, {}); + sseClients.add(sse); + // The first connection fails. The next one sends data. + if (sseClients.length == 1) { + Timer(_step, () => sse.emitError(error: Exception('reset'))); + } else { + Timer( + _step, + () => sse + .emitEvent(MessageEvent('put', _payloadFor('c'), null))); + } + return sse; + }); + return { + ConnectionMode.streaming: streaming, + ConnectionMode.polling: streaming, + }; + }); + expect(await client.start(), isFalse); + + // The data source reconnects after up to 1 second. + await _eventually(() => client.stringVariation('flagA', 'default') == 'c'); + expect(sseClients, hasLength(2)); + expect(client.dataSourceStatus.state, DataSourceState.valid); + await client.close(); + }); + + group('a context with cached flags', () { + /// A persistence that holds cached flags for [key]. The cached value of + /// flagA is 'cached'. + MockPersistence seeded(String key) => MockPersistence() + ..storage['LaunchDarkly_${encodePersistenceKey('the-sdk-key')}'] = { + encodePersistenceKey(key): _payloadFor('cached'), + }; + + /// Runs [body] in a zone, and returns the errors that reach the zone. + Future> uncaughtErrors(Future Function() body) async { + final errors = []; + final done = Completer(); + runZonedGuarded(() { + body().then(done.complete, onError: done.completeError); + }, (error, stack) => errors.add(error)); + await done.future; + return errors; + } + + test( + 'a stream failure after the cache load reaches no zone, and then the ' + 'stream recovers', () async { + final errors = await uncaughtErrors(() async { + final sseClients = []; + final client = LDCommonClient( + _TestConfig('the-sdk-key', AutoEnvAttributes.disabled), + CommonPlatform(persistence: seeded('d')), + _user('d'), + DiagnosticSdkData(name: '', version: ''), + dataSourceFactories: (config, logger, properties) { + StreamingDataSource streaming(LDContext context) => + StreamingDataSource( + credential: 'the-sdk-key', + context: context, + endpoints: ServiceEndpoints(), + logger: logger, + dataSourceConfig: StreamingDataSourceConfig(), + pollingDataSourceConfig: PollingDataSourceConfig(), + httpProperties: properties, + clientFactory: (uri, properties, body, method, logger) { + final sse = SSEClient.testClient(uri, {}); + sseClients.add(sse); + // The first connection fails before any payload. The + // next one sends data. + if (sseClients.length == 1) { + Timer( + _step, + () => sse.emitError( + error: + SseHttpError(503, {}, recoverable: false))); + } else { + Timer( + _step, + () => sse.emitEvent( + MessageEvent('put', _payloadFor('d'), null))); + } + return sse; + }); + return { + ConnectionMode.streaming: streaming, + ConnectionMode.polling: streaming, + }; + }); + + expect(await client.start(), isTrue); + expect(client.stringVariation('flagA', 'default'), 'cached'); + + // A failure during initialization keeps the state at initializing, + // and the status reports the error. + await _eventually( + () => client.dataSourceStatus.lastError?.statusCode == 503); + + await _eventually( + () => client.stringVariation('flagA', 'default') == 'd'); + expect(client.dataSourceStatus.state, DataSourceState.valid); + await client.close(); + }); + expect(errors, isEmpty); + }); + + test( + 'a polling failure after the cache load reaches no zone, and then ' + 'polling recovers', () async { + final errors = await uncaughtErrors(() async { + final client = _pollingClient({ + 'start': [503, 200], + }, persistence: seeded('start')); + + expect(await client.start(), isTrue); + expect(client.stringVariation('flagA', 'default'), 'cached'); + + await _eventually( + () => client.dataSourceStatus.lastError?.statusCode == 503); + await _eventually( + () => client.stringVariation('flagA', 'default') == 'start'); + expect(client.dataSourceStatus.state, DataSourceState.valid); + await client.close(); + }); + expect(errors, isEmpty); + }); + }); } diff --git a/packages/common_client/test/ld_dart_client_test.dart b/packages/common_client/test/ld_dart_client_test.dart index 6782735f..21e249d5 100644 --- a/packages/common_client/test/ld_dart_client_test.dart +++ b/packages/common_client/test/ld_dart_client_test.dart @@ -28,7 +28,7 @@ final class TestDataSource implements DataSource { Stream get events => _eventController.stream; @override - void restart() {} + void restart({int? generation}) {} @override void start() {