From cc3cb85c06a6059f7d5880c27078c920b21334db Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 15 Oct 2026 15:15:17 -0500 Subject: [PATCH 1/4] fix: Prevent the data source status from changing after the client closes --- .../data_source_status_manager.dart | 15 ++++++++++++- .../data_source_status_manager_test.dart | 21 +++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/packages/common_client/lib/src/data_sources/data_source_status_manager.dart b/packages/common_client/lib/src/data_sources/data_source_status_manager.dart index 6284f7c0..aa92572b 100644 --- a/packages/common_client/lib/src/data_sources/data_source_status_manager.dart +++ b/packages/common_client/lib/src/data_sources/data_source_status_manager.dart @@ -11,6 +11,7 @@ final class DataSourceStatusManager { DataSourceStatusErrorInfo? _errorInfo; DateTime _stateSince; final DateTime Function() _stamper; + bool _stopped = false; DataSourceStatus get status { return DataSourceStatus( @@ -33,6 +34,9 @@ final class DataSourceStatusManager { _stamper = stamper; void _updateState(DataSourceState requestedState, {bool isError = false}) { + if (_stopped) { + return; + } // While initializing the state remains initializing if the desired // transition is interrupted. final newState = requestedState == DataSourceState.interrupted && @@ -85,6 +89,9 @@ final class DataSourceStatusManager { /// [DataSourceState.interrupted]. void setErrorResponse(num statusCode, String message, {bool shutdown = false}) { + if (_stopped) { + return; + } _errorInfo = DataSourceStatusErrorInfo( kind: ErrorKind.errorResponse, statusCode: statusCode, @@ -101,6 +108,9 @@ final class DataSourceStatusManager { /// [DataSourceState.shutdown] otherwise it will be /// [DataSourceState.interrupted]. void setErrorByKind(ErrorKind kind, String message, {bool shutdown = false}) { + if (_stopped) { + return; + } _errorInfo = DataSourceStatusErrorInfo( kind: kind, statusCode: null, message: message, time: _stamper()); _updateState( @@ -108,8 +118,11 @@ final class DataSourceStatusManager { isError: true); } - /// Shutdown the manager closing the changes stream. + /// Shutdown the manager closing the changes stream. The manager ignores + /// every update after this, because a data source can still finish work + /// that started before the close. void stop() { + _stopped = true; _controller.close(); } } diff --git a/packages/common_client/test/data_sources/data_source_status_manager_test.dart b/packages/common_client/test/data_sources/data_source_status_manager_test.dart index c2442282..732c64d1 100644 --- a/packages/common_client/test/data_sources/data_source_status_manager_test.dart +++ b/packages/common_client/test/data_sources/data_source_status_manager_test.dart @@ -125,4 +125,25 @@ void main() { manager!.setErrorByKind(ErrorKind.networkError, 'bad network'); }); }); + + group('after stop', () { + test('it ignores state changes', () { + final manager = DataSourceStatusManager(stamper: () => DateTime(1)); + manager.stop(); + manager.setValid(); + manager.setOffline(); + manager.setNetworkUnavailable(); + manager.setBackgroundDisabled(); + expect(manager.status.state, DataSourceState.initializing); + }); + + test('it ignores errors', () { + final manager = DataSourceStatusManager(stamper: () => DateTime(1)); + manager.stop(); + manager.setErrorResponse(401, 'unauthorized'); + manager.setErrorByKind(ErrorKind.networkError, 'bad network'); + expect(manager.status.state, DataSourceState.initializing); + expect(manager.status.lastError, isNull); + }); + }); } From f0f5c30e8d649ccc19e7502035e8f679812bbb65 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 15 Oct 2026 17:53:43 -0500 Subject: [PATCH 2/4] fix: Guard data source status updates where they are sent, not in the status manager --- .../data_source_event_handler.dart | 38 +++-- .../src/data_sources/data_source_manager.dart | 12 ++ .../data_source_status_manager.dart | 15 +- .../data_source_manager_test.dart | 149 +++++++++++++++++- .../data_source_status_manager_test.dart | 21 --- 5 files changed, 182 insertions(+), 53 deletions(-) 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/lib/src/data_sources/data_source_status_manager.dart b/packages/common_client/lib/src/data_sources/data_source_status_manager.dart index aa92572b..6284f7c0 100644 --- a/packages/common_client/lib/src/data_sources/data_source_status_manager.dart +++ b/packages/common_client/lib/src/data_sources/data_source_status_manager.dart @@ -11,7 +11,6 @@ final class DataSourceStatusManager { DataSourceStatusErrorInfo? _errorInfo; DateTime _stateSince; final DateTime Function() _stamper; - bool _stopped = false; DataSourceStatus get status { return DataSourceStatus( @@ -34,9 +33,6 @@ final class DataSourceStatusManager { _stamper = stamper; void _updateState(DataSourceState requestedState, {bool isError = false}) { - if (_stopped) { - return; - } // While initializing the state remains initializing if the desired // transition is interrupted. final newState = requestedState == DataSourceState.interrupted && @@ -89,9 +85,6 @@ final class DataSourceStatusManager { /// [DataSourceState.interrupted]. void setErrorResponse(num statusCode, String message, {bool shutdown = false}) { - if (_stopped) { - return; - } _errorInfo = DataSourceStatusErrorInfo( kind: ErrorKind.errorResponse, statusCode: statusCode, @@ -108,9 +101,6 @@ final class DataSourceStatusManager { /// [DataSourceState.shutdown] otherwise it will be /// [DataSourceState.interrupted]. void setErrorByKind(ErrorKind kind, String message, {bool shutdown = false}) { - if (_stopped) { - return; - } _errorInfo = DataSourceStatusErrorInfo( kind: kind, statusCode: null, message: message, time: _stamper()); _updateState( @@ -118,11 +108,8 @@ final class DataSourceStatusManager { isError: true); } - /// Shutdown the manager closing the changes stream. The manager ignores - /// every update after this, because a data source can still finish work - /// that started before the close. + /// Shutdown the manager closing the changes stream. void stop() { - _stopped = true; _controller.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 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); + }); + }); } diff --git a/packages/common_client/test/data_sources/data_source_status_manager_test.dart b/packages/common_client/test/data_sources/data_source_status_manager_test.dart index 732c64d1..c2442282 100644 --- a/packages/common_client/test/data_sources/data_source_status_manager_test.dart +++ b/packages/common_client/test/data_sources/data_source_status_manager_test.dart @@ -125,25 +125,4 @@ void main() { manager!.setErrorByKind(ErrorKind.networkError, 'bad network'); }); }); - - group('after stop', () { - test('it ignores state changes', () { - final manager = DataSourceStatusManager(stamper: () => DateTime(1)); - manager.stop(); - manager.setValid(); - manager.setOffline(); - manager.setNetworkUnavailable(); - manager.setBackgroundDisabled(); - expect(manager.status.state, DataSourceState.initializing); - }); - - test('it ignores errors', () { - final manager = DataSourceStatusManager(stamper: () => DateTime(1)); - manager.stop(); - manager.setErrorResponse(401, 'unauthorized'); - manager.setErrorByKind(ErrorKind.networkError, 'bad network'); - expect(manager.status.state, DataSourceState.initializing); - expect(manager.status.lastError, isNull); - }); - }); } From d70278ac59a527525a85423a3cf8bc6cd3f44f12 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 15 Oct 2026 22:08:15 -0500 Subject: [PATCH 3/4] fix: Complete a pending identify or start when the client closes --- .../lib/src/ld_common_client.dart | 32 ++++--- .../test/ld_dart_client_test.dart | 83 +++++++++++++++++++ .../flutter_client_sdk/lib/src/ld_client.dart | 8 +- 3 files changed, 109 insertions(+), 14 deletions(-) diff --git a/packages/common_client/lib/src/ld_common_client.dart b/packages/common_client/lib/src/ld_common_client.dart index bd8c496c..24776351 100644 --- a/packages/common_client/lib/src/ld_common_client.dart +++ b/packages/common_client/lib/src/ld_common_client.dart @@ -67,7 +67,8 @@ final class IdentifyComplete implements IdentifyResult { } /// The identify has been superseded. Multiple identify calls were outstanding -/// and this one has been cancelled. +/// and this one has been cancelled, or the client closed before the identify +/// completed. final class IdentifySuperseded implements IdentifyResult {} /// The identify operation encountered an error and will not complete. @@ -240,6 +241,7 @@ final class LDCommonClient { bool _networkAvailable = true; Completer? _startCompleter; + final Completer _closed = Completer(); Future? _startFuture; Stream get dataSourceStatusChanges { @@ -358,13 +360,13 @@ final class LDCommonClient { // The setup of modifiers and other items must be done in the identify // queue to ensure that identifies cannot be done without those items // having been set resulting in a crash. - _identifyQueue.execute(() async { + _untilClosed(_identifyQueue.execute(() async { await _startInternal(); await _identifyInternal(_initialUndecoratedContext, (cb) { afterIdentify = cb; }, waitForNetworkResults: waitForNetworkResults); - }).then((res) { + })).then((res) { final identifyResult = _mapIdentifyResult(res); afterIdentify?.call(identifyResult); _startCompleter!.complete(identifyResult); @@ -377,15 +379,9 @@ final class LDCommonClient { switch (result) { case IdentifyComplete(): return true; + // The queue always runs the start identify, so it is superseded only + // when the client closes before it completes. case IdentifySuperseded(): - // This case does not happen because of the queue configuration. First - // item in the queue will always be the start identify and it will - // always be executed. - _logger.error( - 'Identify was superseded, this represents a logic error in the SDK ' - 'implementation. Please file a bug report.'); - continue error; // Simulate fallthrough. - error: case IdentifyError(): return false; } @@ -537,16 +533,20 @@ final class LDCommonClient { return IdentifyError(Exception(message)); } Function(IdentifyResult)? afterRunner; - final res = await _identifyQueue.execute(() async { + final res = await _untilClosed(_identifyQueue.execute(() async { await _identifyInternal(context, (cb) { afterRunner = cb; }, waitForNetworkResults: waitForNetworkResults); - }); + })); final identifyResult = _mapIdentifyResult(res); afterRunner?.call(identifyResult); return identifyResult; } + /// Completes with [TaskShed] if the client closes before [task] completes. + Future> _untilClosed(Future> task) => + Future.any([task, _closed.future.then((_) => TaskShed())]); + IdentifyResult _mapIdentifyResult(TaskResult res) { switch (res) { case TaskComplete(): @@ -957,7 +957,13 @@ final class LDCommonClient { /// Permanently shuts down the client. /// /// It's not normally necessary to explicitly shut down the client. + /// + /// A pending [start] completes with false, and a pending [identify] + /// completes with [IdentifySuperseded]. Future close() async { + if (!_closed.isCompleted) { + _closed.complete(); + } await _eventProcessor?.flush(); _eventProcessor?.stop(); _dataSourceManager.stop(); diff --git a/packages/common_client/test/ld_dart_client_test.dart b/packages/common_client/test/ld_dart_client_test.dart index 6782735f..3c49b80c 100644 --- a/packages/common_client/test/ld_dart_client_test.dart +++ b/packages/common_client/test/ld_dart_client_test.dart @@ -51,6 +51,25 @@ final class TestDataSource implements DataSource { } } +/// A data source that never sends events. +final class SilentDataSource implements DataSource { + final StreamController _eventController = StreamController(); + + @override + Stream get events => _eventController.stream; + + @override + void restart() {} + + @override + void start() {} + + @override + void stop() { + _eventController.close(); + } +} + void main() { group('given an offline client', () { late LDCommonClient client; @@ -294,6 +313,70 @@ void main() { }); }); + group('given a data source that sends data only for bob', () { + late LDCommonClient client; + + setUp(() { + client = LDCommonClient( + TestConfig('the-sdk-key', AutoEnvAttributes.disabled), + CommonPlatform(persistence: MockPersistence()), + LDContextBuilder().kind('user', 'bob').build(), + DiagnosticSdkData(name: '', version: ''), dataSourceFactories: + (LDCommonConfig config, LDLogger logger, + HttpProperties properties) { + DataSource factory(LDContext context) => context.canonicalKey == 'bob' + ? TestDataSource() + : SilentDataSource(); + return { + ConnectionMode.streaming: factory, + ConnectionMode.polling: factory, + }; + }); + }); + + test('close completes a pending identify as superseded', () async { + await client.start(); + final identify = + client.identify(LDContextBuilder().kind('user', 'joe').build()); + + await client.close(); + + expect(await identify.timeout(const Duration(seconds: 1)), + isA()); + }); + + test('close completes a pending start with false', () async { + final otherClient = LDCommonClient( + TestConfig('the-sdk-key', AutoEnvAttributes.disabled), + CommonPlatform(persistence: MockPersistence()), + LDContextBuilder().kind('user', 'joe').build(), + DiagnosticSdkData(name: '', version: ''), dataSourceFactories: + (LDCommonConfig config, LDLogger logger, + HttpProperties properties) { + return { + ConnectionMode.streaming: (_) => SilentDataSource(), + ConnectionMode.polling: (_) => SilentDataSource(), + }; + }); + final start = otherClient.start(); + + await otherClient.close(); + + expect(await start.timeout(const Duration(seconds: 1)), isFalse); + }); + + test('close does not change a completed identify or start', () async { + expect(await client.start(), isTrue); + expect( + await client.identify(LDContextBuilder().kind('user', 'bob').build()), + isA()); + + await client.close(); + + expect(await client.start(), isTrue); + }); + }); + group('given an invalid context', () { test('start returns false with invalid context', () async { final invalidClient = LDCommonClient( diff --git a/packages/flutter_client_sdk/lib/src/ld_client.dart b/packages/flutter_client_sdk/lib/src/ld_client.dart index 058a70fa..6a0ed4f7 100644 --- a/packages/flutter_client_sdk/lib/src/ld_client.dart +++ b/packages/flutter_client_sdk/lib/src/ld_client.dart @@ -159,6 +159,8 @@ interface class LDClient { /// /// If [waitForNetworkResults] is true, and an error is encountered, then /// false may be returned even if cached values were loaded. + /// + /// If the client closes before start completes, then false is returned. Future start({bool waitForNetworkResults = false}) async { return _client.start(waitForNetworkResults: waitForNetworkResults); } @@ -207,7 +209,8 @@ interface class LDClient { /// without waiting for the previous identify to complete, then you may get /// this result. For instance if you called identify 10 times rapidly, then /// it is likely that 2 total identifies would complete, the first one and the - /// last one. The intermediates would be cancelled for performance. + /// last one. The intermediates would be cancelled for performance. It is + /// also returned when the client closes before the identify completes. /// /// [IdentifyError] this means that the identify has permanently failed. For /// instance the SDK key is no longer valid. @@ -387,6 +390,9 @@ interface class LDClient { /// Permanently shuts down the client. /// /// It's not normally necessary to explicitly shut down the client. + /// + /// A pending [start] completes with false, and a pending [identify] + /// completes with [IdentifySuperseded]. Future close() async { await _client.close(); _connectionManager.dispose(); From a7b481aab91b3f5c4b57d2f07c8b9ea8dc5fa466 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Thu, 15 Oct 2026 22:22:20 -0500 Subject: [PATCH 4/4] chore: Move the pending identify on close change to its own PR --- .../lib/src/ld_common_client.dart | 32 +++---- .../test/ld_dart_client_test.dart | 83 ------------------- .../flutter_client_sdk/lib/src/ld_client.dart | 8 +- 3 files changed, 14 insertions(+), 109 deletions(-) diff --git a/packages/common_client/lib/src/ld_common_client.dart b/packages/common_client/lib/src/ld_common_client.dart index 24776351..bd8c496c 100644 --- a/packages/common_client/lib/src/ld_common_client.dart +++ b/packages/common_client/lib/src/ld_common_client.dart @@ -67,8 +67,7 @@ final class IdentifyComplete implements IdentifyResult { } /// The identify has been superseded. Multiple identify calls were outstanding -/// and this one has been cancelled, or the client closed before the identify -/// completed. +/// and this one has been cancelled. final class IdentifySuperseded implements IdentifyResult {} /// The identify operation encountered an error and will not complete. @@ -241,7 +240,6 @@ final class LDCommonClient { bool _networkAvailable = true; Completer? _startCompleter; - final Completer _closed = Completer(); Future? _startFuture; Stream get dataSourceStatusChanges { @@ -360,13 +358,13 @@ final class LDCommonClient { // The setup of modifiers and other items must be done in the identify // queue to ensure that identifies cannot be done without those items // having been set resulting in a crash. - _untilClosed(_identifyQueue.execute(() async { + _identifyQueue.execute(() async { await _startInternal(); await _identifyInternal(_initialUndecoratedContext, (cb) { afterIdentify = cb; }, waitForNetworkResults: waitForNetworkResults); - })).then((res) { + }).then((res) { final identifyResult = _mapIdentifyResult(res); afterIdentify?.call(identifyResult); _startCompleter!.complete(identifyResult); @@ -379,9 +377,15 @@ final class LDCommonClient { switch (result) { case IdentifyComplete(): return true; - // The queue always runs the start identify, so it is superseded only - // when the client closes before it completes. case IdentifySuperseded(): + // This case does not happen because of the queue configuration. First + // item in the queue will always be the start identify and it will + // always be executed. + _logger.error( + 'Identify was superseded, this represents a logic error in the SDK ' + 'implementation. Please file a bug report.'); + continue error; // Simulate fallthrough. + error: case IdentifyError(): return false; } @@ -533,20 +537,16 @@ final class LDCommonClient { return IdentifyError(Exception(message)); } Function(IdentifyResult)? afterRunner; - final res = await _untilClosed(_identifyQueue.execute(() async { + final res = await _identifyQueue.execute(() async { await _identifyInternal(context, (cb) { afterRunner = cb; }, waitForNetworkResults: waitForNetworkResults); - })); + }); final identifyResult = _mapIdentifyResult(res); afterRunner?.call(identifyResult); return identifyResult; } - /// Completes with [TaskShed] if the client closes before [task] completes. - Future> _untilClosed(Future> task) => - Future.any([task, _closed.future.then((_) => TaskShed())]); - IdentifyResult _mapIdentifyResult(TaskResult res) { switch (res) { case TaskComplete(): @@ -957,13 +957,7 @@ final class LDCommonClient { /// Permanently shuts down the client. /// /// It's not normally necessary to explicitly shut down the client. - /// - /// A pending [start] completes with false, and a pending [identify] - /// completes with [IdentifySuperseded]. Future close() async { - if (!_closed.isCompleted) { - _closed.complete(); - } await _eventProcessor?.flush(); _eventProcessor?.stop(); _dataSourceManager.stop(); diff --git a/packages/common_client/test/ld_dart_client_test.dart b/packages/common_client/test/ld_dart_client_test.dart index 3c49b80c..6782735f 100644 --- a/packages/common_client/test/ld_dart_client_test.dart +++ b/packages/common_client/test/ld_dart_client_test.dart @@ -51,25 +51,6 @@ final class TestDataSource implements DataSource { } } -/// A data source that never sends events. -final class SilentDataSource implements DataSource { - final StreamController _eventController = StreamController(); - - @override - Stream get events => _eventController.stream; - - @override - void restart() {} - - @override - void start() {} - - @override - void stop() { - _eventController.close(); - } -} - void main() { group('given an offline client', () { late LDCommonClient client; @@ -313,70 +294,6 @@ void main() { }); }); - group('given a data source that sends data only for bob', () { - late LDCommonClient client; - - setUp(() { - client = LDCommonClient( - TestConfig('the-sdk-key', AutoEnvAttributes.disabled), - CommonPlatform(persistence: MockPersistence()), - LDContextBuilder().kind('user', 'bob').build(), - DiagnosticSdkData(name: '', version: ''), dataSourceFactories: - (LDCommonConfig config, LDLogger logger, - HttpProperties properties) { - DataSource factory(LDContext context) => context.canonicalKey == 'bob' - ? TestDataSource() - : SilentDataSource(); - return { - ConnectionMode.streaming: factory, - ConnectionMode.polling: factory, - }; - }); - }); - - test('close completes a pending identify as superseded', () async { - await client.start(); - final identify = - client.identify(LDContextBuilder().kind('user', 'joe').build()); - - await client.close(); - - expect(await identify.timeout(const Duration(seconds: 1)), - isA()); - }); - - test('close completes a pending start with false', () async { - final otherClient = LDCommonClient( - TestConfig('the-sdk-key', AutoEnvAttributes.disabled), - CommonPlatform(persistence: MockPersistence()), - LDContextBuilder().kind('user', 'joe').build(), - DiagnosticSdkData(name: '', version: ''), dataSourceFactories: - (LDCommonConfig config, LDLogger logger, - HttpProperties properties) { - return { - ConnectionMode.streaming: (_) => SilentDataSource(), - ConnectionMode.polling: (_) => SilentDataSource(), - }; - }); - final start = otherClient.start(); - - await otherClient.close(); - - expect(await start.timeout(const Duration(seconds: 1)), isFalse); - }); - - test('close does not change a completed identify or start', () async { - expect(await client.start(), isTrue); - expect( - await client.identify(LDContextBuilder().kind('user', 'bob').build()), - isA()); - - await client.close(); - - expect(await client.start(), isTrue); - }); - }); - group('given an invalid context', () { test('start returns false with invalid context', () async { final invalidClient = LDCommonClient( diff --git a/packages/flutter_client_sdk/lib/src/ld_client.dart b/packages/flutter_client_sdk/lib/src/ld_client.dart index 6a0ed4f7..058a70fa 100644 --- a/packages/flutter_client_sdk/lib/src/ld_client.dart +++ b/packages/flutter_client_sdk/lib/src/ld_client.dart @@ -159,8 +159,6 @@ interface class LDClient { /// /// If [waitForNetworkResults] is true, and an error is encountered, then /// false may be returned even if cached values were loaded. - /// - /// If the client closes before start completes, then false is returned. Future start({bool waitForNetworkResults = false}) async { return _client.start(waitForNetworkResults: waitForNetworkResults); } @@ -209,8 +207,7 @@ interface class LDClient { /// without waiting for the previous identify to complete, then you may get /// this result. For instance if you called identify 10 times rapidly, then /// it is likely that 2 total identifies would complete, the first one and the - /// last one. The intermediates would be cancelled for performance. It is - /// also returned when the client closes before the identify completes. + /// last one. The intermediates would be cancelled for performance. /// /// [IdentifyError] this means that the identify has permanently failed. For /// instance the SDK key is no longer valid. @@ -390,9 +387,6 @@ interface class LDClient { /// Permanently shuts down the client. /// /// It's not normally necessary to explicitly shut down the client. - /// - /// A pending [start] completes with false, and a pending [identify] - /// completes with [IdentifySuperseded]. Future close() async { await _client.close(); _connectionManager.dispose();