diff --git a/packages/common_client/lib/src/data_sources/data_source_event_handler.dart b/packages/common_client/lib/src/data_sources/data_source_event_handler.dart index 01b15256..cc0026fc 100644 --- a/packages/common_client/lib/src/data_sources/data_source_event_handler.dart +++ b/packages/common_client/lib/src/data_sources/data_source_event_handler.dart @@ -49,6 +49,19 @@ final class DataSourceEventHandler { final FlagManager _flagManager; final DataSourceStatusManager _statusManager; final LDLogger _logger; + bool _stopped = false; + + /// Stops all later status updates. Work that started before the stop can + /// complete after it, and must not change the status. + void stop() { + _stopped = true; + } + + void _setInvalidData(String message) { + if (!_stopped) { + _statusManager.setErrorByKind(ErrorKind.invalidData, message); + } + } Future handleMessage( LDContext context, String type, String data, @@ -61,8 +74,7 @@ final class DataSourceEventHandler { return _processPut(context, parsed, environmentId); } catch (err) { _logger.error('put message contained invalid json: $err'); - _statusManager.setErrorByKind( - ErrorKind.invalidData, 'Could not parse PUT message'); + _setInvalidData('Could not parse PUT message'); return MessageStatus.invalidMessage; } } @@ -73,8 +85,7 @@ final class DataSourceEventHandler { return _processPatch(context, parsed); } catch (err) { _logger.error('patch message contained invalid json: $err'); - _statusManager.setErrorByKind( - ErrorKind.invalidData, 'Could not parse PATCH message'); + _setInvalidData('Could not parse PATCH message'); return MessageStatus.invalidMessage; } } @@ -85,8 +96,7 @@ final class DataSourceEventHandler { return _processDelete(context, parsed); } catch (err) { _logger.error('delete message contained invalid json: $err'); - _statusManager.setErrorByKind( - ErrorKind.invalidData, 'Could not parse DELETE message'); + _setInvalidData('Could not parse DELETE message'); return MessageStatus.invalidMessage; } } @@ -111,8 +121,7 @@ final class DataSourceEventHandler { return MessageStatus.messageHandled; } catch (err) { _logger.error('Failed to apply an FDv2 change set: ${err.runtimeType}'); - _statusManager.setErrorByKind( - ErrorKind.invalidData, 'Failed to apply an FDv2 change set'); + _setInvalidData('Failed to apply an FDv2 change set'); return MessageStatus.invalidMessage; } } @@ -124,12 +133,13 @@ final class DataSourceEventHandler { (key, value) => MapEntry( key, ItemDescriptor(version: value.version, flag: value))); await _flagManager.init(context, putData, environmentId: environmentId); - _statusManager.setValid(); + if (!_stopped) { + _statusManager.setValid(); + } return MessageStatus.messageHandled; } catch (err) { _logger.error('put message contained an invalid payload: $err'); - _statusManager.setErrorByKind( - ErrorKind.invalidData, 'PUT message contained invalid data'); + _setInvalidData('PUT message contained invalid data'); return MessageStatus.invalidMessage; } } @@ -145,8 +155,7 @@ final class DataSourceEventHandler { return MessageStatus.messageHandled; } catch (err) { _logger.error('patch message contained an invalid payload: $err'); - _statusManager.setErrorByKind( - ErrorKind.invalidData, 'PATCH message contained invalid data'); + _setInvalidData('PATCH message contained invalid data'); return MessageStatus.invalidMessage; } } @@ -160,8 +169,7 @@ final class DataSourceEventHandler { return MessageStatus.messageHandled; } catch (err) { _logger.error('delete message contained an invalid payload: $err'); - _statusManager.setErrorByKind( - ErrorKind.invalidData, 'DELETE message contained invalid data'); + _setInvalidData('DELETE message contained invalid data'); return MessageStatus.invalidMessage; } } 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 755fc55c..99cf05c0 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 @@ -166,6 +166,11 @@ final class DataSourceManager { _activeDataSource = _createDataSource(_activeConnectionMode); _subscription = _activeDataSource?.events.asyncMap((event) async { + // Events that arrive after a stop must not change the status or + // complete an identify. + if (_stopped) { + return MessageStatus.messageHandled; + } if (_activeContext == null) { _logger.error( 'No active context when handling an event from a data source.' @@ -177,6 +182,9 @@ final class DataSourceManager { var handled = await _dataSourceEventHandler.handleMessage( _activeContext!, event.type, event.data, environmentId: event.environmentId); + if (_stopped) { + return handled; + } if (handled == MessageStatus.messageHandled) { _maybeCompleteIdentify(); } @@ -185,6 +193,9 @@ final class DataSourceManager { var handled = await _dataSourceEventHandler.handlePayload( _activeContext!, event.changeSet, environmentId: event.environmentId); + if (_stopped) { + return handled; + } if (handled == MessageStatus.messageHandled) { // Applying any change set from a live source marks it valid -- // including a no-change response, which restores valid after an @@ -230,6 +241,7 @@ final class DataSourceManager { void stop() { _stopped = true; + _dataSourceEventHandler.stop(); _stopConnection(); } } 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 c9b99c74..ac17f44c 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 @@ -70,13 +70,42 @@ Map defaultFactories( }; } +/// Persistence whose operations wait on [gate] when it is set, then throw +/// when [fail] is true. +final class GatedPersistence implements Persistence { + Completer? gate; + bool fail = false; + + Future _wait() async { + await gate?.future; + if (fail) { + throw Exception('persistence failed'); + } + } + + @override + Future read(String namespace, String key) async { + await _wait(); + return null; + } + + @override + Future remove(String namespace, String key) => _wait(); + + @override + Future set(String namespace, String key, String data) => _wait(); +} + DataSourceManager makeManager( LDContext context, Map factories, - {DataSourceStatusManager? inStatusManager}) { + {DataSourceStatusManager? inStatusManager, Persistence? persistence}) { final statusManager = inStatusManager ?? DataSourceStatusManager(); final logger = LDLogger(); - final flagManager = - FlagManager(sdkKey: 'sdk-key', maxCachedContexts: 5, logger: logger); + final flagManager = FlagManager( + persistence: persistence, + sdkKey: 'sdk-key', + maxCachedContexts: 5, + logger: logger); final dataSourceEventHandler = DataSourceEventHandler( flagManager: flagManager, statusManager: statusManager, logger: logger); final manager = DataSourceManager( @@ -446,4 +475,118 @@ void main() { expect(statusManager.status.state, DataSourceState.setOffline, reason: 'a cache load while offline must not report valid'); }); + + group('given work that completes after stop', () { + final context = LDContextBuilder().kind('user', 'bob').build(); + late DataSourceStatusManager statusManager; + late GatedPersistence persistence; + late MockDataSource dataSource; + late DataSourceManager manager; + late Completer identifyCompleter; + + setUp(() { + statusManager = DataSourceStatusManager(); + persistence = GatedPersistence(); + dataSource = MockDataSource(startEvents: []); + manager = makeManager(context, {const FDv2Streaming(): (_) => dataSource}, + inStatusManager: statusManager, persistence: persistence); + identifyCompleter = Completer(); + manager.identify(context, identifyCompleter); + }); + + /// Holds persistence, sends [events], stops the manager, and then + /// releases persistence so the in-flight work completes. + Future sendAndStop(List events, + {bool fail = false}) async { + final gate = Completer(); + persistence.gate = gate; + persistence.fail = fail; + for (final event in events) { + dataSource.controller.add(event); + } + await pumpEventQueue(); + manager.stop(); + gate.complete(); + await pumpEventQueue(); + } + + test('a put does not set valid or complete identify', () async { + await sendAndStop([DataEvent('put', '{}')]); + + expect(statusManager.status.state, DataSourceState.initializing); + expect(identifyCompleter.isCompleted, isFalse); + }); + + test('a failed put does not set an error', () async { + await sendAndStop([DataEvent('put', '{}')], fail: true); + + expect(statusManager.status.state, DataSourceState.initializing); + expect(statusManager.status.lastError, isNull); + }); + + test('an FDv2 payload does not set valid or complete identify', () async { + await sendAndStop([ + PayloadEvent(const ChangeSet( + selector: Selector(state: 'state-1', version: 1), + type: PayloadType.full, + updates: {})) + ]); + + expect(statusManager.status.state, DataSourceState.initializing); + expect(identifyCompleter.isCompleted, isFalse); + }); + + test('a failed FDv2 payload does not set an error', () async { + await sendAndStop([ + PayloadEvent(const ChangeSet( + selector: Selector(state: 'state-1', version: 1), + type: PayloadType.full, + updates: {})) + ], fail: true); + + expect(statusManager.status.state, DataSourceState.initializing); + expect(statusManager.status.lastError, isNull); + }); + + for (final (verb, data) in [ + ( + 'patch', + '{"key": "flag-a", "version": 2, "value": true, "variation": 1,' + ' "trackEvents": false}' + ), + ('delete', '{"key": "flag-a", "version": 2}'), + ]) { + test('a failed $verb does not set an error', () async { + // A put first makes the context active so the update reaches + // persistence. + dataSource.controller.add(DataEvent('put', '{}')); + await identifyCompleter.future; + final validSince = statusManager.status.stateSince; + + await sendAndStop([DataEvent(verb, data)], fail: true); + + expect(statusManager.status.state, DataSourceState.valid); + expect(statusManager.status.stateSince, validSince); + expect(statusManager.status.lastError, isNull); + }); + } + + test('a status event does not set an error', () async { + dataSource.controller.add(DataEvent('put', '{}')); + await identifyCompleter.future; + final validSince = statusManager.status.stateSince; + + // The patch holds the event stream so the status event is still + // pending when the manager stops. + await sendAndStop([ + DataEvent('patch', + '{"key": "flag-a", "version": 2, "value": true, "variation": 1}'), + StatusEvent(ErrorKind.networkError, 503, 'service unavailable'), + ]); + + expect(statusManager.status.state, DataSourceState.valid); + expect(statusManager.status.stateSince, validSince); + expect(statusManager.status.lastError, isNull); + }); + }); }