feat(SDK-2185): FDv2 streaming base, initializer, and synchronizer

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

@@ -263,7 +263,9 @@ final class FDv2ProtocolHandler {
   }
 
   ProtocolAction _processError(ServerErrorEvent data) {
-    _logger.info('Server error encountered receiving updates: ${data.reason}');
+    _logger.info('An issue was encountered receiving updates for payload '
+        "'${data.payloadId ?? '<unknown>'}' with reason: '${data.reason}'. "
+        'Automatic retry will occur.');
     _resetAfterError();
     return ActionServerError(data.reason, id: data.payloadId);
   }

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

@@ -0,0 +1,310 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:http/http.dart' as http;
+import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart';
+import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart';
+
+import 'flag_eval_mapper.dart';
+import 'protocol_handler.dart';
+import 'protocol_types.dart';
+import 'source.dart';
+import 'source_result.dart';
+
+/// Long-lived streaming data source over SSE.
+///
+/// Wraps an [SSEClient] with FDv2 protocol semantics. Each named SSE
+/// event is parsed as JSON, wrapped in an [FDv2Event], and fed to a
+/// fresh [FDv2ProtocolHandler]. The first emitted [ProtocolAction]
+/// per event is translated into an [FDv2SourceResult]:
+///
+/// - [ActionPayload] --> [ChangeSetResult] with `persist: true`.
+/// - [ActionGoodbye] --> goodbye [StatusResult]; the SSE connection is
+///   closed.
+/// - [ActionServerError] / [ActionError] --> interrupted
+///   [StatusResult]; the SSE client's built-in retry handles the
+///   reconnect.
+/// - [ActionNone] --> no emission (waiting for more events).
+///
+/// Legacy `ping` events are routed to the injected [PingHandler] (which
+/// performs a one-shot poll) and the result is forwarded to the
+/// stream. This is the streaming-to-polling bridge for older servers
+/// that pre-date FDv2.
+///
+/// The `x-ld-fd-fallback` header on the initial connection's response
+/// is detected and produces a terminal-error result with
+/// `fdv1Fallback: true`. The connection is closed.
+///
+/// Lifecycle: a single-subscription stream. [results] starts the SSE
+/// connection on subscribe; cancelling the subscription tears it down
+/// without emitting a shutdown. [close] both stops the source and
+/// emits a shutdown [StatusResult] before closing the stream. Both
+/// paths funnel through a `Completer<void> _stoppedSignal` so async
+/// callbacks short-circuit safely.
+///
+/// `SSEClient.restart` is intentionally not surfaced here. The
+/// orchestrator drives connection lifecycle by tearing down a
+/// streaming source and constructing a fresh one (e.g. on credential
+/// rotation or basis change), not by reconnecting an existing one.
+final class FDv2StreamingBase {
+  final SSEClient _sseClient;
+  final PingHandler _pingHandler;
+  final DateTime Function() _now;
+  final LDLogger _logger;
+
+  late final StreamController<FDv2SourceResult> _controller;
+  final Completer<void> _stoppedSignal = Completer<void>();
+  StreamSubscription<Event>? _sseSubscription;
+  FDv2ProtocolHandler? _handler;
+  String? _environmentId;
+  bool _pingInFlight = false;
+
+  FDv2StreamingBase({
+    required SSEClient sseClient,
+    required PingHandler pingHandler,
+    required LDLogger logger,
+    DateTime Function()? now,
+  })  : _sseClient = sseClient,
+        _pingHandler = pingHandler,
+        _logger = logger.subLogger('FDv2StreamingBase'),
+        _now = now ?? DateTime.now {
+    _controller = StreamController<FDv2SourceResult>(
+      onListen: _onListen,
+      onCancel: _onCancel,
+    );
+  }
+
+  /// Single-subscription stream of results. The SSE connection is
+  /// established lazily on the first [Stream.listen] call.
+  Stream<FDv2SourceResult> get results => _controller.stream;
+
+  /// Stops the source, emits a shutdown [StatusResult], and closes the
+  /// stream. Idempotent.
+  void close() {
+    _terminate(
+        finalResult:
+            FDv2SourceResults.shutdown(message: 'Streaming source closed'));
+  }
+
+  /// Terminal-path helper used by [close] and by the in-stream
+  /// terminal paths (goodbye event, fdv1-fallback header). Completes
+  /// [_stoppedSignal] *first* so any subsequent [close] call -- e.g.
+  /// from inside an `onData` listener reacting to the [finalResult]
+  /// we are about to emit -- short-circuits at its guard instead of
+  /// racing into a closed controller. Idempotent.
+  void _terminate({FDv2SourceResult? finalResult}) {
+    if (_stoppedSignal.isCompleted) return;
+    _stoppedSignal.complete();
+    _tearDownConnection();
+    if (!_controller.isClosed) {
+      if (finalResult != null) {
+        _controller.add(finalResult);
+      }
+      _controller.close();
+    }
+  }
+
+  void _onListen() {
+    _resetHandler();
+    _sseSubscription = _sseClient.stream.listen(
+      _handleEvent,
+      onError: _handleSseError,
+    );
+  }
+
+  /// Builds a fresh [FDv2ProtocolHandler]. Called on initial connect
+  /// and on every subsequent [OpenEvent] (SSE auto-reconnect), so a
+  /// partial transfer from the previous connection cannot bleed into
+  /// the new one regardless of whether the server re-sends
+  /// `server-intent` after a Last-Event-ID resumption. Also called
+  /// after a mid-event throw inside [processEvent] so any
+  /// half-accumulated `_tempUpdates` are discarded.
+  void _resetHandler() {
+    _handler = FDv2ProtocolHandler(
+      objProcessors: {flagEvalKind: processFlagEval},
+      logger: _logger,
+    );
+  }
+
+  Future<void> _onCancel() async {
+    if (_stoppedSignal.isCompleted) return;
+    _stoppedSignal.complete();
+    _tearDownConnection();
+    // No shutdown emission -- the subscriber asked us to stop.
+  }
+
+  void _tearDownConnection() {
+    _sseSubscription?.cancel();
+    _sseSubscription = null;
+    // Best-effort close. The SSE client may already be closed if it
+    // emitted an error; that's fine -- the operation is documented as
+    // safe in any state.
+    _sseClient.close();
+  }
+
+  void _handleEvent(Event event) {
+    if (_stoppedSignal.isCompleted) return;
+    switch (event) {
+      case OpenEvent open:
+        _handleOpen(open);
+      case MessageEvent message:
+        _handleMessage(message);
+    }
+  }
+
+  void _handleOpen(OpenEvent event) {
+    // Every OpenEvent represents a (re)established connection. Rebuild
+    // the protocol handler so a partial transfer from the prior
+    // connection cannot bleed into this one -- the SDK must defend
+    // against this regardless of whether the server respects the
+    // protocol's "re-send server-intent on resume" semantic.
+    _resetHandler();
+
+    final headers = event.headers;
+    if (headers == null) return;
+
+    final envId = headers['x-ld-envid'];
+    if (envId != null) {
+      _environmentId = envId;
+    }
+
+    final fallback = headers['x-ld-fd-fallback']?.toLowerCase() == 'true';
+    if (fallback) {
+      // Server told us to fall back; route through the terminal helper
+      // so a close() from the listener's onData -- a natural reaction
+      // to a fallback signal -- doesn't race with our own close.
+      _terminate(
+          finalResult: FDv2SourceResults.terminalError(
+        message: 'Server requested FDv1 fallback',
+        fdv1Fallback: true,
+      ));
+    }
+  }
+
+  Future<void> _handleMessage(MessageEvent event) async {
+    if (event.type == 'ping') {
+      // Legacy bridge: older servers may still send `ping` instead of
+      // FDv2 events. Defer to the injected handler for a one-shot poll.
+      await _handlePing();
+      return;
+    }
+
+    final ProtocolAction action;
+    try {
+      final decoded = jsonDecode(event.data);
+      if (decoded is! Map<String, dynamic>) {
+        _logger.warn('Ignoring SSE event with non-object data: '
+            'event=${event.type}');
+        _emit(FDv2SourceResults.interrupted(
+            message: 'Streaming event payload was not a JSON object'));
+        return;
+      }
+      // Wrap the protocol-handler dispatch in the same try/catch as the
+      // jsonDecode: the structural casts inside the per-event fromJson
+      // factories (e.g. PayloadIntent, PutObjectEvent) throw TypeError
+      // on shape mismatch and would otherwise become unhandled async
+      // exceptions.
+      action =
+          _handler!.processEvent(FDv2Event(event: event.type, data: decoded));
+    } catch (err) {
+      _logger.warn('Failed to parse or process SSE event (${err.runtimeType})');
+      // Reset the handler -- a mid-event throw can leave it with stale
+      // _tempUpdates from the partially-processed payload.
+      _resetHandler();
+      _emit(FDv2SourceResults.interrupted(
+          message: 'Streaming event payload was malformed'));
+      return;
+    }
+
+    if (_stoppedSignal.isCompleted) return;
+
+    switch (action) {
+      case ActionPayload(:final payload):
+        _emit(ChangeSetResult(
+          payload: payload,
+          environmentId: _environmentId,
+          freshness: _now(),
+          persist: true,
+        ));
+      case ActionGoodbye(:final reason):
+        // Server told us to disconnect; route through the terminal
+        // helper so a close() from the listener's onData -- a natural
+        // reaction to a goodbye -- doesn't race with our own close.
+        _terminate(
+            finalResult: FDv2SourceResults.goodbyeResult(message: reason));
+      case ActionServerError(:final reason):
+        _emit(FDv2SourceResults.interrupted(message: reason));
+      case ActionError(:final message):
+        _emit(FDv2SourceResults.interrupted(message: message));
+      case ActionNone():
+        // No emission; continue accumulating events until the handler
+        // reaches a terminal action.
+        break;
+    }
+  }
+
+  Future<void> _handlePing() async {
+    // The FDv2 ping semantic is "go re-poll". A single in-flight poll
+    // already satisfies any number of pings that arrive while it is
+    // running, so drop excess pings rather than spawning concurrent
+    // polls (which would race on emit-order and amplify load on the
+    // polling endpoint).
+    if (_pingInFlight) return;
+    _pingInFlight = true;
+    try {
+      final FDv2SourceResult result;
+      try {
+        result = await _pingHandler();
+      } catch (err) {
+        _logger.warn('Ping handler threw unexpectedly: ${err.runtimeType}');
+        _emit(FDv2SourceResults.interrupted(
+            message: 'Ping handler raised error unexpectedly'));
+        return;
+      }
+      if (_stoppedSignal.isCompleted) return;
+      _emit(result);
+    } finally {
+      _pingInFlight = false;
+    }
+  }
+
+  void _handleSseError(Object err, StackTrace stack) {
+    if (_stoppedSignal.isCompleted) return;
+    // The SSE client's built-in backoff handles reconnection. Surface
+    // the disruption as interrupted; the orchestrator decides whether
+    // to fall through to a different source after enough time.
+    //
+    // Don't log the raw exception. http.ClientException's toString
+    // 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');
+    _logger.debug('SSE error stack:\n$stack');
+    _emit(FDv2SourceResults.interrupted(message: _describeError(err)));
+  }
+
+  /// Categorizes an exception surfaced on the SSE stream into a fixed
+  /// sanitized message. Mirrors the polling base's helper so neither
+  /// surface (the public StatusResult.message nor the warn log) ever
+  /// 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';
+    }
+    final type = err.runtimeType.toString();
+    if (type.contains('Tls') || type.contains('Handshake')) {
+      return 'TLS error during streaming request';
+    }
+    return 'Streaming connection error';
+  }
+
+  void _emit(FDv2SourceResult result) {
+    if (_stoppedSignal.isCompleted) return;
+    if (_controller.isClosed) return;
+    _controller.add(result);
+  }
+}

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

@@ -0,0 +1,59 @@
+import 'dart:async';
+
+import 'source.dart';
+import 'source_result.dart';
+import 'streaming_base.dart';
+
+/// One-shot streaming initializer.
+///
+/// Subscribes to the underlying [FDv2StreamingBase], returns the first
+/// emitted [FDv2SourceResult], and tears the connection down. Used at
+/// SDK init time to bring the SDK to a usable state from the streaming
+/// path before handing off to the long-lived synchronizer.
+///
+/// Calling [close] before the first emission resolves the pending
+/// [run] future with a [SourceState.shutdown] result.
+final class FDv2StreamingInitializer implements Initializer {
+  final FDv2StreamingBase _base;
+  final Completer<FDv2SourceResult> _completer = Completer<FDv2SourceResult>();
+  StreamSubscription<FDv2SourceResult>? _subscription;
+  bool _closed = false;
+
+  FDv2StreamingInitializer({required FDv2StreamingBase base}) : _base = base;
+
+  @override
+  Future<FDv2SourceResult> run() {
+    if (_closed) {
+      return Future.value(_shutdownResult());
+    }
+    _subscription = _base.results.listen((result) {
+      if (_completer.isCompleted) return;
+      _completer.complete(result);
+      // First emission received; tear down.
+      _subscription?.cancel();
+      _subscription = null;
+      _base.close();
+    }, onDone: () {
+      if (_completer.isCompleted) return;
+      // The base closed before producing a result. Surface as shutdown.
+      _completer.complete(_shutdownResult());
+    });
+    return _completer.future;
+  }
+
+  @override
+  void close() {
+    if (_closed) return;
+    _closed = true;
+    _subscription?.cancel();
+    _subscription = null;
+    _base.close();
+    if (!_completer.isCompleted) {
+      _completer.complete(_shutdownResult());
+    }
+  }
+
+  StatusResult _shutdownResult() => FDv2SourceResults.shutdown(
+        message: 'Streaming initializer closed before first emission',
+      );
+}