fix: Keep retrying the streaming connection after an unexpected HTTP error response

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',
   ];
 

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;

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

packages/common_client/lib/src/data_sources/data_source_manager.dart

@@ -45,7 +45,7 @@ final class DataSourceManager {
   final Map<FDv2ConnectionMode, DataSourceFactory> _dataSourceFactories = {};
 
   DataSource? _activeDataSource;
-  StreamSubscription<MessageStatus?>? _subscription;
+  StreamSubscription<void>? _subscription;
   bool _stopped = false;
 
   Completer<void>? _identifyCompleter;
@@ -165,7 +165,7 @@ final class DataSourceManager {
     }
 
     _activeDataSource = _createDataSource(_activeConnectionMode);
-    _subscription = _activeDataSource?.events.asyncMap((event) async {
+    Future<MessageStatus?> 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();

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

packages/common_client/lib/src/data_sources/fdv2/streaming_base.dart

@@ -392,9 +392,10 @@ final class FDv2StreamingBase {
     // formats as 'ClientException: <msg>, uri=<full-url>', 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')) {

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.
   }
 }

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