feat: Support per-attempt URIs and typed unrecoverable status errors in the SSE client (#296)

## What this adds

- **`uriProvider`**: an optional callback on `SSEClient` that is invoked
before every connection attempt (including automatic reconnects) and
whose result is used in place of the fixed URI. This is core to FDv2:
streaming reconnects must carry the current `basis` selector as a query
parameter, which advances as payloads are received. Without it, a
reconnect re-establishes with a stale basis and the server replays
changes the SDK already has. Supported on all platforms — the html
implementation builds a fresh `EventSource` per attempt (its error
handling restarts with its own backoff rather than relying on native
reconnection), so it consults the provider the same way the HTTP
implementation does.
- **`UnrecoverableStatusError`**: unrecoverable HTTP status codes
(anything other than 200, 400, 408, 429, or 5xx) were previously
reported as a `ClientException` with the status embedded in the message
text. They are now a typed error carrying the status code and response
headers, so consumers can distinguish terminal failures from transient
ones and read service directives delivered on error responses (e.g. the
FDv2 fallback directive). The client's behavior is otherwise unchanged:
it still stops reconnecting and reports the error on the stream.

## Platform limitation, now documented

When running in the browser, the native `EventSource` exposes no HTTP
status codes or response headers, so the html implementation is
**incapable of reporting terminal errors**: every failure is treated as
recoverable and retried with backoff indefinitely, and
`UnrecoverableStatusError` is never produced on web. Consumers that need
to react to unrecoverable statuses (e.g. invalid credentials) must
detect them through a transport that can observe HTTP responses, such as
a polling request. This is now stated on `SSEClient`, `HtmlSseClient`,
and `UnrecoverableStatusError`.

Both API changes are additive; existing consumers are unaffected.

## Testing

New unit tests cover the per-attempt URI resolution (fixed URI without a
provider, provider invoked per attempt) and the typed error's status
code and headers. The html implementation was additionally
compile-checked for the web target.

SDK-2186

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Low Risk**
> Additive API and error-type change in the SSE client package; existing
callers without `uriProvider` behave the same aside from a more specific
stream error type on HTTP.
> 
> **Overview**
> Adds optional **`uriProvider`** on `SSEClient` so each connect and
reconnect uses a freshly resolved URI (via `StateValues.connectUri` on
HTTP and a new `EventSource` on web), enabling query params such as an
advancing FDv2 `basis` selector.
> 
> Non-retryable HTTP failures on the HTTP transport are now surfaced as
exported **`UnrecoverableStatusError`** (status + headers) instead of a
`ClientException` with the code in the message; reconnect behavior is
unchanged. Docs call out that browser `EventSource` cannot observe HTTP
responses, so web never emits this error and keeps retrying with
backoff.
> 
> Unit tests cover fixed vs per-attempt URIs and typed 401 errors with
response headers.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
023ad08. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: kinyoklion

packages/event_source_client/lib/launchdarkly_event_source_client.dart

@@ -12,6 +12,7 @@ import 'src/sse_client_stub.dart'
     if (dart.library.js_interop) 'src/sse_client_html.dart';
 import 'src/test_sse_client.dart';
 
+export 'src/errors.dart' show UnrecoverableStatusError;
 export 'src/events.dart' show Event, MessageEvent, OpenEvent;
 export 'src/test_sse_client.dart' show TestSseClient;
 export 'src/logging.dart'
@@ -67,6 +68,12 @@ enum SseHttpMethod {
 /// In certain cases, unrecoverable errors will be reported on the [stream] at
 /// which point the stream will be done.
 ///
+/// On `html` platforms the client is incapable of reporting unrecoverable
+/// errors: the browser's native `EventSource` exposes no HTTP status codes
+/// or response headers, so every failure is treated as recoverable and
+/// retried with backoff indefinitely, and no error is ever reported on the
+/// [stream].
+///
 /// The [SSEClient] will make best effort to maintain the streaming connection.
 abstract class SSEClient {
   static const defaultHeaders = <String, String>{
@@ -106,8 +113,10 @@ abstract class SSEClient {
   /// Factory constructor to return the platform implementation.
   ///
   /// On all platforms, the [uri] and [eventTypes] arguments are required.
-  /// On majority of platforms, the optional arguments are used.
-  /// On web, the optional arguments are not used.
+  /// On majority of platforms, the optional arguments are used. On web,
+  /// the [headers], [connectTimeout], [readTimeout], [body], and
+  /// [httpMethod] arguments are not used; the standard `EventSource`
+  /// does not support them.
   ///
   /// The [uri] specifies where to connect.  The [eventTypes] determines which
   /// event types will be emitted.  For non-web platforms, pass in [headers] to
@@ -126,19 +135,27 @@ abstract class SSEClient {
   ///
   /// An optional [logger] for controlling logging output from the SSE client.
   /// If not provided, a [NoOpLogger] will be used.
+  ///
+  /// An optional [uriProvider]. When provided, it is invoked before every
+  /// connection attempt -- the first connect and each automatic
+  /// reconnect -- and its result is used for that attempt; [uri] is used
+  /// only when no provider is given. This allows query parameters to
+  /// vary between attempts (e.g. a state selector that advances as data
+  /// is received). Supported on all platforms.
   factory SSEClient(Uri uri, Set<String> eventTypes,
       {Map<String, String> headers = defaultHeaders,
       Duration connectTimeout = defaultConnectTimeout,
       Duration readTimeout = defaultReadTimeout,
       String? body,
       SseHttpMethod httpMethod = SseHttpMethod.get,
-      EventSourceLogger? logger}) {
+      EventSourceLogger? logger,
+      Uri Function()? uriProvider}) {
     // merge headers so consumer gets reasonable defaults
     var mergedHeaders = <String, String>{};
     mergedHeaders.addAll(defaultHeaders);
     mergedHeaders.addAll(headers);
     return getSSEClient(uri, eventTypes, mergedHeaders, connectTimeout,
-        readTimeout, body, httpMethod.toString(), logger);
+        readTimeout, body, httpMethod.toString(), logger, uriProvider);
   }
 
   /// Get an SSE client for use in unit tests.

packages/event_source_client/lib/src/errors.dart

@@ -0,0 +1,23 @@
+/// Reported on the event stream when the server responds with an HTTP
+/// status code that the client will not retry (anything other than 200,
+/// 400, 408, 429, or 5xx). After reporting this error the client stops
+/// reconnecting until connection desire changes.
+///
+/// Only produced by implementations whose transport can observe HTTP
+/// responses. The browser's native `EventSource` cannot, so on `html`
+/// platforms this error is never reported and the client retries every
+/// failure indefinitely.
+final class UnrecoverableStatusError implements Exception {
+  /// The HTTP status code of the response.
+  final int statusCode;
+
+  /// The response headers, when available. May carry service directives
+  /// (e.g. protocol fallback instructions) even on error responses.
+  final Map<String, String> headers;
+
+  const UnrecoverableStatusError(this.statusCode,
+      [this.headers = const <String, String>{}]);
+
+  @override
+  String toString() => 'UnrecoverableStatusError(statusCode: $statusCode)';
+}

packages/event_source_client/lib/src/sse_client_html.dart

@@ -9,6 +9,14 @@ import 'backoff.dart';
 import 'events.dart' as ld_message_event;
 
 /// An [SSEClient] that uses the [web.EventSource] available on most browsers for web platform support.
+///
+/// The native `EventSource` API does not expose HTTP status codes or
+/// response headers, so this implementation is incapable of reporting
+/// terminal errors: every failure is treated as recoverable and retried
+/// with backoff indefinitely, and no error is ever reported on the
+/// [stream]. Consumers that need to react to unrecoverable statuses
+/// (e.g. invalid credentials) must detect them through a transport that
+/// can observe HTTP responses, such as a polling request.
 class HtmlSseClient implements SSEClient {
   /// The underlying eventsource
   web.EventSource? _eventSource;
@@ -21,15 +29,22 @@ class HtmlSseClient implements SSEClient {
   Backoff _backoff = Backoff(math.Random());
 
   final Uri _uri;
+  final Uri Function()? _uriProvider;
   final Set<String> _eventTypes;
 
   int? _activeSince;
   Timer? _retryTimer;
 
-  /// Creates an instance of an SSEClient that will connect in the future
-  /// to the [uri] provided.
-  HtmlSseClient(Uri uri, Set<String> eventTypes, EventSourceLogger? logger)
+  /// Creates an instance of an SSEClient that will connect in the future.
+  ///
+  /// Every connection attempt -- the first connect and each reconnect --
+  /// constructs a fresh `EventSource` from the [uriProvider] result when
+  /// a provider is given. The fixed [uri] is used only when no provider
+  /// is given.
+  HtmlSseClient(Uri uri, Set<String> eventTypes, EventSourceLogger? logger,
+      {Uri Function()? uriProvider})
       : _uri = uri,
+        _uriProvider = uriProvider,
         _eventTypes = eventTypes {
     _logger = logger ?? NoOpLogger();
     _messageEventsController =
@@ -56,7 +71,8 @@ class HtmlSseClient implements SSEClient {
   }
 
   void _setupConnection() {
-    _eventSource = web.EventSource(_uri.toString());
+    final connectUri = _uriProvider?.call() ?? _uri;
+    _eventSource = web.EventSource(connectUri.toString());
 
     for (var eventType in _eventTypes) {
       _eventSource?.addEventListener(eventType, _handleMessageEvent.toJS);
@@ -130,6 +146,7 @@ SSEClient getSSEClient(
         Duration readTimeout,
         String? body,
         String method,
-        EventSourceLogger? logger) =>
+        EventSourceLogger? logger,
+        Uri Function()? uriProvider) =>
     // dropping unsupported configuration options
-    HtmlSseClient(uri, eventTypes, logger);
+    HtmlSseClient(uri, eventTypes, logger, uriProvider: uriProvider);

packages/event_source_client/lib/src/sse_client_http.dart

@@ -39,7 +39,8 @@ class HttpSseClient implements SSEClient {
       Duration readTimeout,
       String? body,
       String httpMethod,
-      EventSourceLogger? logger)
+      EventSourceLogger? logger,
+      {Uri Function()? uriProvider})
       : this.internal(
             uri,
             eventTypes,
@@ -51,7 +52,8 @@ class HttpSseClient implements SSEClient {
             math.Random(),
             body,
             httpMethod,
-            logger);
+            logger,
+            uriProvider: uriProvider);
 
   /// An internal constructor for injecting necessary dependencies for testing.
   HttpSseClient.internal(
@@ -65,7 +67,8 @@ class HttpSseClient implements SSEClient {
       math.Random random,
       String? body,
       String httpMethod,
-      EventSourceLogger? logger) {
+      EventSourceLogger? logger,
+      {Uri Function()? uriProvider}) {
     _logger = logger ?? NoOpLogger();
     _messageEventsController = StreamController<Event>.broadcast(
       // this is triggered when first listener subscribes
@@ -88,7 +91,8 @@ class HttpSseClient implements SSEClient {
         body,
         httpMethod,
         _resetRequest.stream,
-        logger ?? NoOpLogger()));
+        logger ?? NoOpLogger(),
+        uriProvider: uriProvider));
   }
 
   /// Subscribe to this [stream] to receive events and sometimes errors.  The first
@@ -134,6 +138,8 @@ SSEClient getSSEClient(
         Duration readTimeout,
         String? body,
         String method,
-        EventSourceLogger? logger) =>
+        EventSourceLogger? logger,
+        Uri Function()? uriProvider) =>
     HttpSseClient(uri, eventTypes, headers, connectTimeout, readTimeout, body,
-        method, logger);
+        method, logger,
+        uriProvider: uriProvider);

packages/event_source_client/lib/src/sse_client_stub.dart

@@ -9,6 +9,7 @@ SSEClient getSSEClient(
         Duration readTimeout,
         String? body,
         String method,
-        EventSourceLogger? logger) =>
+        EventSourceLogger? logger,
+        Uri Function()? uriProvider) =>
     throw UnsupportedError(
         'LaunchDarkly SSE Client is not supported on this platform.');

packages/event_source_client/lib/src/state_connecting.dart

@@ -3,6 +3,7 @@ import 'dart:async';
 import 'package:http/http.dart' as http;
 
 import 'error_utils.dart';
+import 'errors.dart';
 import 'http_consts.dart';
 import 'state_backoff.dart';
 import 'state_connected.dart';
@@ -35,7 +36,7 @@ class StateConnecting {
   /// to connect.  The returned function will run the next state.
   static Future<Function> _tryGetConnected(
       StateValues svo, http.Client client) async {
-    final request = http.Request(svo.httpMethod, svo.uri);
+    final request = http.Request(svo.httpMethod, svo.connectUri);
     request.headers.addAll(svo.headers);
     if (svo.body != null) {
       request.body = svo.body!;
@@ -50,7 +51,7 @@ class StateConnecting {
     }
 
     try {
-      svo.logger.debug('Sending HTTP request to ${svo.uri}');
+      svo.logger.debug('Sending HTTP request to ${request.url}');
       final response = await client.send(request).timeout(svo.connectTimeout);
 
       // anything besides OK is bad, but some may be recoverable with a retry.
@@ -61,8 +62,8 @@ class StateConnecting {
           // looks like the error wasn't recoverable, go to idle and wait
           // for something to change
           return () => StateIdle.run(svo,
-              errorCause: http.ClientException(
-                  'Got unrecoverable status code ${response.statusCode}'));
+              errorCause: UnrecoverableStatusError(
+                  response.statusCode, response.headers));
         }
 
         // the error is recoverable, backoff then we'll try again

packages/event_source_client/lib/src/state_value_object.dart

@@ -19,6 +19,12 @@ typedef ClientFactory = http.Client Function();
 class StateValues {
   // Non-transient configuration data
   final Uri uri;
+
+  /// When provided, produces the URI for every connection attempt -- the
+  /// first connect and each reconnect -- in place of [uri], so callers
+  /// can vary query parameters between attempts (e.g. a state selector
+  /// that advances as data is received).
+  final Uri Function()? uriProvider;
   final Set<String> eventTypes;
   final Map<String, String> headers;
   final Duration connectTimeout;
@@ -50,6 +56,9 @@ class StateValues {
   /// Headers received from the connection.
   Map<String, String>? connectHeaders;
 
+  /// The URI to use for the next connection attempt.
+  Uri get connectUri => uriProvider?.call() ?? uri;
+
   /// Creates a [_StateValues] instance.  Used by the state machine.
   StateValues(
       this.uri,
@@ -65,6 +74,7 @@ class StateValues {
       this.body,
       this.httpMethod,
       this.resetRequest,
-      this.logger)
+      this.logger,
+      {this.uriProvider})
       : backoff = Backoff(random);
 }

packages/event_source_client/test/test_utils.dart

@@ -20,6 +20,7 @@ class TestUtils {
 
   static StateValues makeMockStateValues(
       {Uri? uri,
+      Uri Function()? uriProvider,
       Set<String>? eventTypes,
       Map<String, String>? headers,
       Duration? connectTimeout,
@@ -45,15 +46,18 @@ class TestUtils {
         null,
         'GET',
         resetStream ?? StreamController<void>.broadcast().stream,
-        logger ?? NoOpLogger());
+        logger ?? NoOpLogger(),
+        uriProvider: uriProvider);
   }
 
   static MockClient makeMockHttpClient(
       {int httpStatusCode = HttpStatusCodes.okStatus,
       Map<String, String> headers = defaultHeaders,
-      bool blocking = false}) {
+      bool blocking = false,
+      void Function(BaseRequest)? onRequest}) {
     return MockClient.streaming((request, bodyStream) async {
       return bodyStream.bytesToString().then((bodyString) async {
+        onRequest?.call(request);
         if (blocking) {
           await Completer().future; // blocks indefinitely
         }

packages/event_source_client/test/uri_provider_test.dart

@@ -0,0 +1,61 @@
+// ignore_for_file: close_sinks
+
+import 'dart:async';
+
+import 'package:http/http.dart';
+import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart';
+import 'package:launchdarkly_event_source_client/src/state_connecting.dart';
+import 'package:test/test.dart';
+
+import 'test_utils.dart';
+
+void main() {
+  test('uses the fixed uri when no provider is given', () async {
+    final requestedPaths = <String>[];
+    final svo = TestUtils.makeMockStateValues(
+        uri: Uri.parse('/fixed'),
+        clientFactory: () => TestUtils.makeMockHttpClient(
+            onRequest: (BaseRequest request) =>
+                requestedPaths.add(request.url.path)));
+
+    await StateConnecting.run(svo);
+
+    expect(requestedPaths, equals(['/fixed']));
+  });
+
+  test('invokes the uriProvider for each connection attempt', () async {
+    final requestedPaths = <String>[];
+    var attempt = 0;
+    final svo = TestUtils.makeMockStateValues(
+        uri: Uri.parse('/unused'),
+        uriProvider: () => Uri.parse('/attempt-${++attempt}'),
+        clientFactory: () => TestUtils.makeMockHttpClient(
+            onRequest: (BaseRequest request) =>
+                requestedPaths.add(request.url.path)));
+
+    await StateConnecting.run(svo);
+    await StateConnecting.run(svo);
+
+    expect(requestedPaths, equals(['/attempt-1', '/attempt-2']));
+  });
+
+  test(
+      'reports UnrecoverableStatusError with the status code and headers '
+      'for non-retryable status codes', () async {
+    final eventsController = StreamController<Event>.broadcast();
+    final svo = TestUtils.makeMockStateValues(
+        eventSink: eventsController,
+        clientFactory: () => TestUtils.makeMockHttpClient(
+            httpStatusCode: 401, headers: {'x-ld-fd-fallback': 'true'}));
+
+    final expectation = expectLater(
+        eventsController.stream,
+        emitsError(isA<UnrecoverableStatusError>()
+            .having((error) => error.statusCode, 'statusCode', 401)
+            .having((error) => error.headers['x-ld-fd-fallback'],
+                'fallback header', 'true')));
+
+    await StateConnecting.run(svo);
+    await expectation;
+  });
+}