feat: Add FDv2 synchronizer fallback and recovery conditions (#297)

## What this adds

Timed conditions that the FDv2 data source orchestrator (a later PR)
observes alongside the active synchronizer's results to drive tier
transitions:

- **Fallback condition**: starts its timer when the synchronizer reports
an interrupted status and cancels it when a change set arrives. If it
fires (120 seconds by default), the orchestrator moves to the next
available synchronizer. Terminal statuses (shutdown, terminal error,
goodbye) do not arm the timer — the orchestrator reacts to those
immediately rather than waiting out a fallback period — and repeated
interruptions do not extend the deadline; the period counts from the
first interruption.
- **Recovery condition**: starts when observed and ignores results. If
it fires (300 seconds by default), the orchestrator returns to the
primary synchronizer.

`getConditions` selects which conditions apply: none when only one
synchronizer is available (nowhere to fall back to), fallback only for
the primary, and both for a non-primary synchronizer. `ConditionGroup`
merges its members and emits the first to fire.

The timeout defaults match the other client-side FDv2 implementations.

## Why streams instead of futures

The Java and C++ implementations model conditions as one-shot futures,
and both had the same leak (fixed in java-core by `c27bf26` /
launchdarkly/java-core#163): a future's listeners can never be detached,
only released by completion, so an orchestration loop that races a
long-lived pending future per result accumulates an irremovable listener
garland per change set — unbounded on a healthy primary. Dart futures
have the identical property (measured: ~559 B retained per race against
a pending future), but Dart also has the primitive those platforms lack:
cancellable stream subscriptions.

Each condition therefore exposes a single-subscription
`Stream<ConditionType>` that emits **at most once** and then closes
(closing without emitting if the condition is closed first). Lifetimes
are scoped to the subscription: self-starting timers begin when the
stream is listened to, and cancelling the subscription closes the
condition and releases its timers. One bounded edge remains by design —
informing a never-listened group can arm a fallback timer until its
timeout elapses or `close()` is called — and is documented on the API.
The orchestrator PR consumes these with one subscription per
synchronizer run, closes in a `finally`, and includes a soak test
asserting bounded memory across a sustained stream of results.

## Testing

Tests run inside a `package:fake_async` zone, advancing time with
`elapse` and asserting timer state through `pendingTimers` — no timer
injection in the production code. They cover timer start/cancel behavior
for both condition kinds, terminal statuses not arming the fallback
timer, the fallback deadline not extending on repeated interruptions,
at-most-once emission including when two member timers contend in the
same instant, close semantics, subscription cancellation releasing
condition and group timers, group merging and inform broadcast, and the
`getConditions` selection rules.

SDK-2186

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> New timing logic will drive synchronizer failover when wired into the
orchestrator; behavior is isolated in new modules with thorough tests
and no production integration in this PR.
> 
> **Overview**
> Adds **FDv2 synchronizer tier-transition conditions** for a future
orchestrator: timed signals that emit **fallback** (switch to the next
synchronizer after sustained interruption) or **recovery** (return to
primary after running on a backup).
> 
> **Fallback** arms on `interrupted`, cancels on a change set, ignores
terminal statuses, and does not reset the deadline on repeated
interruptions (defaults: 120s). **Recovery** starts when the condition
is observed and ignores results (default 300s). Conditions expose
**single-subscription streams** (at-most-one emit) so subscriptions can
be cancelled without the listener-retention issues of racing futures.
> 
> `ConditionGroup` merges members (first fire wins), broadcasts `inform`
to all members, and `getConditions` picks an empty group, fallback-only
for primary, or fallback+recovery for non-primary when multiple
synchronizers exist.
> 
> Adds **`fake_async`** tests for timer behavior, close/cancel
semantics, group contention, and `getConditions` rules.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
4fc80f6. 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/conditions.dart

@@ -0,0 +1,256 @@
+import 'dart:async';
+
+import 'source_result.dart';
+
+/// Default time a synchronizer may remain interrupted before the
+/// orchestrator falls back to the next available synchronizer.
+const Duration defaultFallbackTimeout = Duration(seconds: 120);
+
+/// Default time a non-primary synchronizer runs before the orchestrator
+/// attempts to recover back to the primary synchronizer.
+const Duration defaultRecoveryTimeout = Duration(seconds: 300);
+
+/// The kind of condition that fired, determining the orchestrator's
+/// response.
+enum ConditionType {
+  /// Move to the next available synchronizer.
+  fallback,
+
+  /// Reset to the primary synchronizer.
+  recovery,
+}
+
+/// A timed condition observed alongside the active synchronizer's
+/// results. When the condition fires, its [events] stream emits a
+/// [ConditionType] that the orchestration loop uses to decide what to
+/// do.
+///
+/// Conditions are streams rather than futures so a consumer can detach:
+/// cancelling a stream subscription releases the consumer's listener,
+/// whereas a listener on a never-completing future can never be
+/// removed and would be retained for the condition's whole lifetime.
+abstract interface class Condition {
+  /// Single-subscription stream that emits at most one [ConditionType]
+  /// when the condition fires and then closes. Closes without emitting
+  /// if the condition is closed first.
+  ///
+  /// The condition's lifetime is scoped to the subscription: timers that
+  /// start on their own start when the stream is listened to, and
+  /// cancelling the subscription closes the condition. A condition that
+  /// is informed but never listened to can hold a timer until its
+  /// timeout elapses; call [close] to release it early.
+  Stream<ConditionType> get events;
+
+  /// Inform the condition about a synchronizer result. Some conditions
+  /// use this to start or cancel their timers.
+  void inform(FDv2SourceResult result);
+
+  /// Cancel any pending timers and close [events]. Idempotent.
+  void close();
+}
+
+final class _TimedCondition implements Condition {
+  final Duration _timeout;
+  final ConditionType _type;
+  final void Function(FDv2SourceResult result,
+      {required void Function() start,
+      required void Function() cancel})? _informHandler;
+
+  late final StreamController<ConditionType> _controller =
+      StreamController<ConditionType>(
+    onListen: _onListen,
+    onCancel: close,
+  );
+  Timer? _timer;
+  bool _closed = false;
+
+  _TimedCondition({
+    required Duration timeout,
+    required ConditionType type,
+    void Function(FDv2SourceResult result,
+            {required void Function() start, required void Function() cancel})?
+        informHandler,
+  })  : assert(timeout > Duration.zero),
+        _timeout = timeout,
+        _type = type,
+        _informHandler = informHandler;
+
+  void _onListen() {
+    // Without an inform handler the timer starts as soon as the
+    // condition is observed (recovery behavior). With one, the handler
+    // decides when to start it.
+    if (_informHandler == null) {
+      _startTimer();
+    }
+  }
+
+  void _startTimer() {
+    if (_timer != null || _closed) return;
+    _timer = Timer(_timeout, () {
+      if (_closed) return;
+      _closed = true;
+      // The controller buffers the event if the subscription has not
+      // started yet, so firing before listen is not lost.
+      _controller.add(_type);
+      _controller.close();
+    });
+  }
+
+  void _cancelTimer() {
+    _timer?.cancel();
+    _timer = null;
+  }
+
+  @override
+  Stream<ConditionType> get events => _controller.stream;
+
+  @override
+  void inform(FDv2SourceResult result) {
+    if (_closed) return;
+    _informHandler?.call(result, start: _startTimer, cancel: _cancelTimer);
+  }
+
+  @override
+  void close() {
+    if (_closed) return;
+    _closed = true;
+    _cancelTimer();
+    _controller.close();
+  }
+}
+
+/// Creates a fallback condition. The condition starts its timer when an
+/// interrupted status is received and cancels it when a change set is
+/// received. If the timer fires, the condition emits
+/// [ConditionType.fallback].
+///
+/// Terminal statuses (shutdown, terminal error, goodbye) do not arm the
+/// timer: the orchestrator reacts to those immediately, out of band,
+/// rather than waiting out a fallback period. The timer is also not
+/// re-armed by repeated interruptions; the fallback period counts from
+/// the first interruption.
+Condition createFallbackCondition(Duration timeout) {
+  return _TimedCondition(
+    timeout: timeout,
+    type: ConditionType.fallback,
+    informHandler: (result, {required start, required cancel}) {
+      switch (result) {
+        case ChangeSetResult():
+          cancel();
+        case StatusResult(state: SourceState.interrupted):
+          start();
+        case StatusResult():
+          break;
+      }
+    },
+  );
+}
+
+/// Creates a recovery condition. The timer starts immediately and the
+/// condition emits [ConditionType.recovery] when it fires. Results do
+/// not affect it.
+Condition createRecoveryCondition(Duration timeout) {
+  return _TimedCondition(
+    timeout: timeout,
+    type: ConditionType.recovery,
+  );
+}
+
+/// A group of conditions managed together. The group merges the member
+/// streams and broadcasts results to all of them.
+final class ConditionGroup {
+  final List<Condition> _conditions;
+  final List<StreamSubscription<ConditionType>> _subscriptions = [];
+
+  late final StreamController<ConditionType> _controller =
+      StreamController<ConditionType>(
+    onListen: _subscribe,
+    onCancel: close,
+  );
+  bool _fired = false;
+  bool _closed = false;
+
+  ConditionGroup(List<Condition> conditions) : _conditions = conditions;
+
+  /// Single-subscription stream that emits at most one [ConditionType]
+  /// (the first member condition to fire) and then closes. Closes
+  /// without emitting if the group is closed first; a group with no
+  /// member conditions never emits.
+  ///
+  /// The group's lifetime is scoped to the subscription: member timers
+  /// that start on their own start when the stream is listened to, and
+  /// cancelling the subscription closes the group and its members. A
+  /// group that is informed but never listened to can hold a timer
+  /// until its timeout elapses; call [close] to release it early.
+  Stream<ConditionType> get events => _controller.stream;
+
+  void _subscribe() {
+    if (_closed) return;
+    for (final condition in _conditions) {
+      _subscriptions.add(condition.events.listen((type) {
+        // First member to fire wins; member timers firing in the same
+        // event-loop turn cannot produce a second emission.
+        if (_fired || _closed) return;
+        _fired = true;
+        _controller.add(type);
+        _finish();
+      }));
+    }
+  }
+
+  /// Broadcast a result to all conditions.
+  void inform(FDv2SourceResult result) {
+    if (_closed) return;
+    for (final condition in _conditions) {
+      condition.inform(result);
+    }
+  }
+
+  /// Close all conditions and the merged stream. Idempotent.
+  void close() {
+    if (_closed) return;
+    _finish();
+  }
+
+  void _finish() {
+    _closed = true;
+    for (final subscription in _subscriptions) {
+      subscription.cancel();
+    }
+    _subscriptions.clear();
+    for (final condition in _conditions) {
+      condition.close();
+    }
+    if (!_controller.isClosed) {
+      _controller.close();
+    }
+  }
+}
+
+/// Determines which conditions apply to the active synchronizer.
+///
+/// - With at most one available synchronizer there is nowhere to fall
+///   back to, so no conditions are created.
+/// - The primary (first available) synchronizer gets only a fallback
+///   condition.
+/// - A non-primary synchronizer gets both fallback and recovery
+///   conditions.
+ConditionGroup getConditions({
+  required int availableSynchronizerCount,
+  required bool isPrimary,
+  Duration fallbackTimeout = defaultFallbackTimeout,
+  Duration recoveryTimeout = defaultRecoveryTimeout,
+}) {
+  if (availableSynchronizerCount <= 1) {
+    return ConditionGroup(const []);
+  }
+
+  if (isPrimary) {
+    return ConditionGroup([createFallbackCondition(fallbackTimeout)]);
+  }
+
+  return ConditionGroup([
+    createFallbackCondition(fallbackTimeout),
+    createRecoveryCondition(recoveryTimeout),
+  ]);
+}