fix: Prevent the data source status from changing after the client closes

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<MessageStatus> 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;
     }
   }

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();
   }
 }

packages/common_client/test/data_sources/data_source_manager_test.dart

@@ -70,13 +70,42 @@ Map<FDv2ConnectionMode, DataSourceFactory> defaultFactories(
   };
 }
 
+/// Persistence whose operations wait on [gate] when it is set, then throw
+/// when [fail] is true.
+final class GatedPersistence implements Persistence {
+  Completer<void>? gate;
+  bool fail = false;
+
+  Future<void> _wait() async {
+    await gate?.future;
+    if (fail) {
+      throw Exception('persistence failed');
+    }
+  }
+
+  @override
+  Future<String?> read(String namespace, String key) async {
+    await _wait();
+    return null;
+  }
+
+  @override
+  Future<void> remove(String namespace, String key) => _wait();
+
+  @override
+  Future<void> set(String namespace, String key, String data) => _wait();
+}
+
 DataSourceManager makeManager(
     LDContext context, Map<FDv2ConnectionMode, DataSourceFactory> 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<void> identifyCompleter;
+
+    setUp(() {
+      statusManager = DataSourceStatusManager();
+      persistence = GatedPersistence();
+      dataSource = MockDataSource(startEvents: []);
+      manager = makeManager(context, {const FDv2Streaming(): (_) => dataSource},
+          inStatusManager: statusManager, persistence: persistence);
+      identifyCompleter = Completer<void>();
+      manager.identify(context, identifyCompleter);
+    });
+
+    /// Holds persistence, sends [events], stops the manager, and then
+    /// releases persistence so the in-flight work completes.
+    Future<void> sendAndStop(List<DataSourceEvent> events,
+        {bool fail = false}) async {
+      final gate = Completer<void>();
+      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);
+    });
+  });
 }