feat: honor the FDv1 fallback directive on success, error, and goodbye (#312)

Stacked on #311 (the `SseHttpError` surfacing this depends on).

## What

Honors the FDv1 fallback directive across every way the server can
deliver it:

- **Successful connection:** the directive is emitted with the basis
change set, so the streamed payload is applied *before* the SDK falls
back — previously the basis was dropped the moment the header was seen.
- **Any error response carrying the header** (`SseHttpError`,
recoverable or not): the streaming source closes the connection — which
stops the client's own retry — and routes to the fallback tier.
- **`goodbye` event with `protocolFallbackTTL`:** treated as an in-band
fallback directive, for transports that cannot read response headers.

A single helper parses the directive (presence + TTL) from response
headers, used for both the successful and error paths; the goodbye path
reads its TTL in-band.

The streaming source maps `SseHttpError` by its `recoverable` flag:
recoverable → interrupted (the client retries), unrecoverable →
terminal. The FDv1 streaming source ignores recoverable errors so a
transient 5xx no longer shuts it down.

When a fallback tier is configured the orchestrator engages it. When
none is configured, the SDK stays interrupted and retries FDv2 after the
directive's TTL (default 1 hour; a TTL of `0` means remain paused with
no retry) rather than halting or reconnecting immediately. Source
results carry the fallback TTL.

## Tests

- Orchestrator: apply-then-engage from a directive-bearing change set,
and the three no-fallback cases (finite TTL retries, absent TTL defers,
zero TTL pauses).
- `streaming_base`: defer-on-success, the TTL header, the goodbye/TTL
directive, and the `SseHttpError` paths (recoverable → interrupted/stays
open, unrecoverable → terminal/closes, directive → terminal+TTL
regardless of recoverability). `protocol_handler` covers
`protocolFallbackTTL` parsing.
- v3 contract harness `fdv1-fallback` suite passes end-to-end (816
total, 792 ran, exit 0), with no regression.

The contract-test-service capability + `fdv1Fallback` config wiring that
exercises this in the harness lives on the e2e branch and will land with
the v3 contract-tests PR.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes core FDv2 connection lifecycle, tier selection, and flag
delivery ordering when servers send fallback signals; behavior shifts
are broad but covered by new unit/contract tests.
> 
> **Overview**
> Implements end-to-end handling when the server asks the SDK to fall
back from FDv2 to FDv1, including **TTL-aware retry** when no FDv1 tier
is configured.
> 
> **Directive parsing and propagation:** Adds shared
`readFallbackDirective` for `x-ld-fd-fallback` / `x-ld-fd-fallback-ttl`,
threads `fdv1FallbackTtl` through `FDv2SourceResult`, and parses in-band
`protocolFallbackTTL` on goodbye events.
> 
> **Streaming / polling behavior:** On a **successful** stream open, the
directive is **deferred** and emitted with the next change set so the
basis payload is applied before fallback (replacing immediate terminal
error on connect). **HTTP errors** with the header, and **goodbye** with
TTL or a pending header, surface as terminal fallback results so the
orchestrator does not recycle past the directive. Polling mirrors
goodbye/header TTL stamping. FDv2 streaming maps `SseHttpError` by
recoverability; legacy FDv1 streaming **ignores recoverable** SSE HTTP
errors so transient 5xx no longer shuts the source down.
> 
> **Orchestrator:** Classifies directives into engage FDv1 tier, defer
retry (no tier: wait TTL—default 1h, zero = pause indefinitely), or
none; schedules recycle via `_pendingRetryDelay` and interruptible
`_delay` on stop.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
74e7c4b. 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/common_client/lib/src/data_sources/fdv2/fallback_directive.dart

@@ -0,0 +1,20 @@
+/// The FDv1 fallback directive parsed from a connection's response
+/// headers. Its presence means the server asked the SDK to fall back;
+/// [ttl] is how long to remain on the fallback before retrying FDv2
+/// (null when the server gave no TTL; [Duration.zero] means indefinitely).
+final class FallbackDirective {
+  final Duration? ttl;
+  const FallbackDirective(this.ttl);
+}
+
+/// Reads the FDv1 fallback directive from response headers, or null when
+/// the `x-ld-fd-fallback` header is not `"true"`. The single place that
+/// interprets these headers, shared by the streaming and polling sources.
+FallbackDirective? readFallbackDirective(Map<String, String> headers) {
+  if (headers['x-ld-fd-fallback']?.toLowerCase() != 'true') {
+    return null;
+  }
+  final raw = headers['x-ld-fd-fallback-ttl'];
+  final seconds = raw == null ? null : int.tryParse(raw);
+  return FallbackDirective(seconds == null ? null : Duration(seconds: seconds));
+}

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

@@ -34,6 +34,20 @@ enum _SynchronizerOutcome {
   stop,
 }
 
+/// How the orchestrator responds to server-directed actions.
+enum _DirectiveAction {
+  /// No directive present; carry on normally.
+  none,
+
+  /// An FDv1 fallback tier exists and was engaged; advance to it.
+  engageFdv1Fallback,
+
+  /// The directive was received but no FDv1 fallback tier is configured.
+  /// The SDK stays interrupted and retries FDv2 after the directive's TTL
+  /// rather than removing the source.
+  deferRetry,
+}
+
 /// The FDv2 data source orchestrator.
 ///
 /// Runs the initializer chain to bring the SDK to a usable state, then
@@ -64,6 +78,18 @@ final class FDv2DataSourceOrchestrator implements DataSource {
   bool _emittedPayload = false;
   bool _initialized = false;
 
+  /// Default wait before retrying FDv2 after a fallback directive arrives
+  /// with no FDv1 fallback tier configured and no server-provided TTL.
+  static const Duration _defaultFallbackRetryInterval = Duration(hours: 1);
+
+  /// Overrides the next recycle delay when set (consumed once), used to
+  /// honor a fallback directive's retry TTL.
+  Duration? _pendingRetryDelay;
+
+  /// Completes when [stop] is called so a long interruptible wait (e.g. a
+  /// fallback retry delay) wakes promptly on shutdown.
+  final Completer<void> _stopCompleter = Completer<void>();
+
   /// True when the only sources are cache initializers (no synchronizers).
   /// Such a system must still reach a usable state on a cache miss, so an
   /// empty payload is emitted when no data was produced.
@@ -117,6 +143,7 @@ final class FDv2DataSourceOrchestrator implements DataSource {
   void stop() {
     if (_closed) return;
     _closed = true;
+    if (!_stopCompleter.isCompleted) _stopCompleter.complete();
     _sourceManager.close();
     // Wake the active synchronizer run so it can observe the closed
     // state.
@@ -190,22 +217,52 @@ final class FDv2DataSourceOrchestrator implements DataSource {
     _controller.add(InitializedEvent());
   }
 
-  /// True when the source indicated an FDv1 fallback directive and a
-  /// fallback tier exists to engage. The directive is ignored when the
+  /// Classifies a server-directed FDv1 fallback and, for
+  /// [_DirectiveAction.engageFdv1Fallback], makes the source manager switch
+  /// tiers. The directive is ignored ([_DirectiveAction.none]) when the
   /// FDv1 fallback synchronizer is already the active source, so it cannot
-  /// loop (the fallback source also never re-asserts the directive).
-  bool _handleFdv1Fallback(FDv2SourceResult result) {
-    if (result.fdv1Fallback &&
-        _sourceManager.hasFdv1FallbackConfigured &&
-        !_sourceManager.isCurrentSynchronizerFdv1Fallback) {
+  /// loop (the fallback source also never re-asserts the directive). When
+  /// there is no FDv1 fallback tier we defer retrying FDv2
+  /// ([_DirectiveAction.deferRetry]).
+  _DirectiveAction _processDirective(FDv2SourceResult result) {
+    if (!result.fdv1Fallback ||
+        _sourceManager.isCurrentSynchronizerFdv1Fallback) {
+      return _DirectiveAction.none;
+    }
+    if (_sourceManager.hasFdv1FallbackConfigured) {
       _logger.warn('Server directed fallback to FDv1; engaging the FDv1 '
           'fallback synchronizer.');
       _sourceManager.engageFdv1Fallback();
-      return true;
+      return _DirectiveAction.engageFdv1Fallback;
+    }
+    return _DirectiveAction.deferRetry;
+  }
+
+  /// Honors a fallback directive that has no FDv1 tier to engage: the SDK
+  /// stays interrupted and schedules an FDv2 retry after the directive's
+  /// TTL. A TTL of zero means remain indefinitely (no retry); an absent
+  /// TTL uses [_defaultFallbackRetryInterval].
+  void _scheduleFallbackRetry(
+      Duration? ttl, void Function(_SynchronizerOutcome) resolve) {
+    if (ttl == Duration.zero) {
+      _logger.warn('Server directed FDv1 fallback with no fallback tier '
+          'configured and an indefinite TTL; FDv2 updates are paused.');
+      resolve(_SynchronizerOutcome.stop);
+      return;
     }
-    return false;
+    final delay = ttl ?? _defaultFallbackRetryInterval;
+    _logger.warn('Server directed FDv1 fallback but no fallback tier is '
+        'configured; staying interrupted and retrying FDv2 in '
+        '${delay.inSeconds}s.');
+    _pendingRetryDelay = delay;
+    resolve(_SynchronizerOutcome.recycle);
   }
 
+  /// Waits [d], returning early if the orchestrator is stopped so a long
+  /// fallback-retry delay does not pin the run open past shutdown.
+  Future<void> _delay(Duration d) =>
+      Future.any([Future<void>.delayed(d), _stopCompleter.future]);
+
   Future<void> _runInitializers() async {
     var errorDuringInit = false;
     var dataReceived = false;
@@ -225,19 +282,26 @@ final class FDv2DataSourceOrchestrator implements DataSource {
             _emitPayload(result);
             dataReceived = true;
 
-            if (_handleFdv1Fallback(result)) {
-              // Data was received but the server directed FDv1 fallback;
-              // move on to synchronizers where the fallback tier runs and
-              // its first change set completes initialization.
-              return;
-            }
-
-            if (result.changeSet.selector.isNotEmpty) {
-              // A selector means a complete, server-versioned payload:
-              // initialization is done. A selector-less payload (e.g. cache)
-              // is applied, but we keep initializing toward network data.
-              _emitInitialized();
-              return;
+            switch (_processDirective(result)) {
+              case _DirectiveAction.engageFdv1Fallback:
+                // Data received but the server directed FDv1 fallback;
+                // move on to synchronizers where the fallback tier runs
+                // and its first change set completes initialization.
+                return;
+              case _DirectiveAction.deferRetry:
+                // An initializer is one-shot and cannot retry itself;
+                // let the chain exhaust and leave retry timing to the
+                // synchronizer tier.
+                break;
+              case _DirectiveAction.none:
+                if (result.changeSet.selector.isNotEmpty) {
+                  // A selector means a complete, server-versioned payload:
+                  // initialization is done. A selector-less payload (e.g.
+                  // cache) is applied, but we keep initializing toward
+                  // network data.
+                  _emitInitialized();
+                  return;
+                }
             }
           }
         case StatusResult():
@@ -253,7 +317,8 @@ final class FDv2DataSourceOrchestrator implements DataSource {
             case SourceState.goodbye:
               break;
           }
-          if (_handleFdv1Fallback(result)) {
+          if (_processDirective(result) ==
+              _DirectiveAction.engageFdv1Fallback) {
             return;
           }
       }
@@ -298,8 +363,10 @@ final class FDv2DataSourceOrchestrator implements DataSource {
     while (!_closed) {
       if (recycleCurrent) {
         recycleCurrent = false;
-        if (_recycleDelay > Duration.zero) {
-          await Future<void>.delayed(_recycleDelay);
+        final delay = _pendingRetryDelay ?? _recycleDelay;
+        _pendingRetryDelay = null;
+        if (delay > Duration.zero) {
+          await _delay(delay);
           if (_closed) return;
         }
         synchronizer = _sourceManager.recreateCurrentSynchronizer();
@@ -393,12 +460,15 @@ final class FDv2DataSourceOrchestrator implements DataSource {
               _logger.warn('Synchronizer terminal error: '
                   '${result.message ?? 'unknown error'}');
               _reportTransientError(result);
-              if (_handleFdv1Fallback(result)) {
-                resolve(_SynchronizerOutcome.advance);
-                return;
+              switch (_processDirective(result)) {
+                case _DirectiveAction.engageFdv1Fallback:
+                  resolve(_SynchronizerOutcome.advance);
+                case _DirectiveAction.deferRetry:
+                  _scheduleFallbackRetry(result.fdv1FallbackTtl, resolve);
+                case _DirectiveAction.none:
+                  _sourceManager.blockCurrentSynchronizer();
+                  resolve(_SynchronizerOutcome.advance);
               }
-              _sourceManager.blockCurrentSynchronizer();
-              resolve(_SynchronizerOutcome.advance);
               return;
             case SourceState.shutdown:
               // A synchronizer that shuts itself down before the system
@@ -421,8 +491,16 @@ final class FDv2DataSourceOrchestrator implements DataSource {
           }
       }
 
-      if (_handleFdv1Fallback(result)) {
-        resolve(_SynchronizerOutcome.advance);
+      switch (_processDirective(result)) {
+        case _DirectiveAction.engageFdv1Fallback:
+          // A change set carrying the directive (e.g. a successful stream
+          // that delivered its basis then asked us to fall back) has been
+          // applied above; now advance to the fallback tier.
+          resolve(_SynchronizerOutcome.advance);
+        case _DirectiveAction.deferRetry:
+          _scheduleFallbackRetry(result.fdv1FallbackTtl, resolve);
+        case _DirectiveAction.none:
+          break;
       }
     }, onDone: () {
       if (outcome.isCompleted || _closed) return;

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

@@ -4,6 +4,7 @@ import 'dart:convert';
 import 'package:http/http.dart' as http;
 import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart';
 
+import 'fallback_directive.dart';
 import 'flag_eval_mapper.dart';
 import 'payload.dart';
 import 'protocol_handler.dart';
@@ -27,12 +28,14 @@ import 'source_result.dart';
 ///   through an [FDv2ProtocolHandler]. The first emitted action
 ///   determines the result.
 ///
-/// `x-ld-fd-fallback: true` is treated as an annotation on whatever
-/// result the response would otherwise produce: the body is still
-/// parsed and used, the 304 is still treated as no-op, errors are
-/// still classified by status code, and `fdv1Fallback: true` is
-/// stamped on the resulting [FDv2SourceResult]. The orchestrator can
-/// consume the data and transition to FDv1 in the same step.
+/// The `x-ld-fd-fallback` directive (read via [readFallbackDirective],
+/// shared with the streaming source) is treated as an annotation on
+/// whatever result the response would otherwise produce: the body is
+/// still parsed and used, the 304 is still treated as no-op, errors are
+/// still classified by status code, and `fdv1Fallback: true` plus the
+/// fallback TTL are stamped on the resulting [FDv2SourceResult]. The
+/// orchestrator can consume the data and transition to FDv1 in the same
+/// step.
 final class FDv2PollingBase {
   final LDLogger _logger;
   final FDv2Requestor _requestor;
@@ -65,9 +68,9 @@ final class FDv2PollingBase {
   }
 
   FDv2SourceResult _processResponse(RequestorResponse response) {
-    // Match `x-ld-fd-fallback` case-insensitively.
-    final fdv1Fallback =
-        response.headers['x-ld-fd-fallback']?.toLowerCase() == 'true';
+    final directive = readFallbackDirective(response.headers);
+    final fdv1Fallback = directive != null;
+    final fdv1FallbackTtl = directive?.ttl;
     final environmentId = response.headers['x-ld-envid'];
 
     // 304 Not Modified means the SDK's cached data is confirmed current.
@@ -78,6 +81,7 @@ final class FDv2PollingBase {
         freshness: _now(),
         persist: true,
         fdv1Fallback: fdv1Fallback,
+        fdv1FallbackTtl: fdv1FallbackTtl,
       );
     }
 
@@ -89,27 +93,31 @@ final class FDv2PollingBase {
           statusCode: response.status,
           message: message,
           fdv1Fallback: fdv1Fallback,
+          fdv1FallbackTtl: fdv1FallbackTtl,
         );
       }
       _logger.error('$message; will not retry');
       return FDv2SourceResults.terminalError(
         statusCode: response.status,
         message: message,
         fdv1Fallback: fdv1Fallback,
+        fdv1FallbackTtl: fdv1FallbackTtl,
       );
     }
 
     return _parseBody(
       response,
       environmentId: environmentId,
       fdv1Fallback: fdv1Fallback,
+      fdv1FallbackTtl: fdv1FallbackTtl,
     );
   }
 
   FDv2SourceResult _parseBody(
     RequestorResponse response, {
     String? environmentId,
     required bool fdv1Fallback,
+    required Duration? fdv1FallbackTtl,
   }) {
     // The whole parse path is wrapped: jsonDecode plus the structural
     // casts inside FDv2EventsCollection.fromJson and the per-event
@@ -122,6 +130,7 @@ final class FDv2PollingBase {
           statusCode: response.status,
           message: 'Polling response was not a JSON object',
           fdv1Fallback: fdv1Fallback,
+          fdv1FallbackTtl: fdv1FallbackTtl,
         );
       }
 
@@ -145,21 +154,33 @@ final class FDv2PollingBase {
               freshness: _now(),
               persist: true,
               fdv1Fallback: fdv1Fallback,
+              fdv1FallbackTtl: fdv1FallbackTtl,
             );
-          case ActionGoodbye(:final reason):
-            return FDv2SourceResults.goodbyeResult(
-              message: reason,
-              fdv1Fallback: fdv1Fallback,
-            );
+          case ActionGoodbye(:final reason, :final protocolFallbackTtl):
+            // A goodbye can itself carry a fallback directive: in-band via
+            // protocolFallbackTtl, or via the response header. The
+            // orchestrator's goodbye path recycles without consulting the
+            // directive, so surface it as a terminal fallback result (as the
+            // streaming source does) rather than an ordinary goodbye.
+            if (protocolFallbackTtl != null || fdv1Fallback) {
+              return FDv2SourceResults.terminalError(
+                message: reason,
+                fdv1Fallback: true,
+                fdv1FallbackTtl: protocolFallbackTtl ?? fdv1FallbackTtl,
+              );
+            }
+            return FDv2SourceResults.goodbyeResult(message: reason);
           case ActionServerError(:final reason):
             return FDv2SourceResults.interrupted(
               message: reason,
               fdv1Fallback: fdv1Fallback,
+              fdv1FallbackTtl: fdv1FallbackTtl,
             );
           case ActionError(:final message):
             return FDv2SourceResults.interrupted(
               message: message,
               fdv1Fallback: fdv1Fallback,
+              fdv1FallbackTtl: fdv1FallbackTtl,
             );
           case ActionNone():
             // Continue accumulating events until a payload-transferred or
@@ -175,6 +196,7 @@ final class FDv2PollingBase {
         statusCode: response.status,
         message: 'Polling response did not include a complete payload',
         fdv1Fallback: fdv1Fallback,
+        fdv1FallbackTtl: fdv1FallbackTtl,
       );
     } catch (err, stack) {
       // Log only the type at error level (not the message — `jsonDecode`
@@ -187,6 +209,7 @@ final class FDv2PollingBase {
         statusCode: response.status,
         message: 'Polling response body was malformed',
         fdv1Fallback: fdv1Fallback,
+        fdv1FallbackTtl: fdv1FallbackTtl,
       );
     }
   }