feat: Add FDv2 source manager

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

@@ -0,0 +1,177 @@
+import 'entry_factories.dart';
+import 'source.dart';
+
+/// State of a synchronizer slot.
+enum SynchronizerSlotState {
+  /// Can be selected for use.
+  available,
+
+  /// Cannot be selected (terminal error, or an FDv1 fallback slot that has
+  /// not been activated).
+  blocked,
+}
+
+/// A slot in the synchronizer list, wrapping a factory with state.
+final class SynchronizerSlot {
+  final SynchronizerFactory factory;
+
+  /// True when this slot is the FDv1 fallback adapter. FDv1 slots start
+  /// blocked and are only activated by a server fallback directive.
+  final bool isFdv1Fallback;
+
+  SynchronizerSlotState state;
+
+  SynchronizerSlot({
+    required this.factory,
+    this.isFdv1Fallback = false,
+    SynchronizerSlotState? initialState,
+  }) : state = initialState ??
+            (isFdv1Fallback
+                ? SynchronizerSlotState.blocked
+                : SynchronizerSlotState.available);
+}
+
+/// Manages the state of initializers and synchronizers, tracks which
+/// source is active, and handles source transitions for the orchestrator.
+final class SourceManager {
+  final List<InitializerFactory> _initializerFactories;
+  final List<SynchronizerSlot> _synchronizerSlots;
+  final SelectorGetter _selectorGetter;
+
+  Initializer? _activeInitializer;
+  Synchronizer? _activeSynchronizer;
+  int _initializerIndex = -1;
+  int _synchronizerIndex = -1;
+  bool _shutdown = false;
+
+  SourceManager({
+    required List<InitializerFactory> initializerFactories,
+    required List<SynchronizerSlot> synchronizerSlots,
+    required SelectorGetter selectorGetter,
+  })  : _initializerFactories = initializerFactories,
+        _synchronizerSlots = synchronizerSlots,
+        _selectorGetter = selectorGetter;
+
+  bool get isShutdown => _shutdown;
+
+  void _closeActiveSource() {
+    _activeInitializer?.close();
+    _activeInitializer = null;
+    _activeSynchronizer?.close();
+    _activeSynchronizer = null;
+  }
+
+  int _firstAvailableIndex() => _synchronizerSlots
+      .indexWhere((slot) => slot.state == SynchronizerSlotState.available);
+
+  /// Get the next initializer and set it as the active source. Closes the
+  /// previous active source. Returns null when all initializers are
+  /// exhausted.
+  Initializer? nextInitializer() {
+    if (_shutdown) return null;
+
+    _initializerIndex += 1;
+    if (_initializerIndex >= _initializerFactories.length) {
+      return null;
+    }
+
+    _closeActiveSource();
+    final initializer =
+        _initializerFactories[_initializerIndex].create(_selectorGetter);
+    _activeInitializer = initializer;
+    return initializer;
+  }
+
+  /// Get the next available (non-blocked) synchronizer and set it as the
+  /// active source, scanning forward from the current position and
+  /// wrapping around. Closes the previous active source. Returns null
+  /// when no synchronizer is available.
+  Synchronizer? nextAvailableSynchronizer() {
+    if (_shutdown || _synchronizerSlots.isEmpty) return null;
+
+    var visited = 0;
+    while (visited < _synchronizerSlots.length) {
+      _synchronizerIndex += 1;
+      if (_synchronizerIndex >= _synchronizerSlots.length) {
+        _synchronizerIndex = 0;
+      }
+
+      final candidate = _synchronizerSlots[_synchronizerIndex];
+      if (candidate.state == SynchronizerSlotState.available) {
+        _closeActiveSource();
+        final synchronizer = candidate.factory.create(_selectorGetter);
+        _activeSynchronizer = synchronizer;
+        return synchronizer;
+      }
+      visited += 1;
+    }
+
+    return null;
+  }
+
+  /// Close the active synchronizer and create a fresh instance from the
+  /// current slot, without advancing the scan position. Used when a
+  /// source must drop its connection and re-establish it (goodbye,
+  /// invalid data). Returns null if the current slot is not available
+  /// or no synchronizer has been started yet.
+  Synchronizer? recreateCurrentSynchronizer() {
+    if (_shutdown ||
+        _synchronizerIndex < 0 ||
+        _synchronizerIndex >= _synchronizerSlots.length) {
+      return null;
+    }
+    final slot = _synchronizerSlots[_synchronizerIndex];
+    if (slot.state != SynchronizerSlotState.available) {
+      return null;
+    }
+    _closeActiveSource();
+    final synchronizer = slot.factory.create(_selectorGetter);
+    _activeSynchronizer = synchronizer;
+    return synchronizer;
+  }
+
+  /// Mark the current synchronizer as blocked (e.g. after a terminal
+  /// error).
+  void blockCurrentSynchronizer() {
+    if (_synchronizerIndex >= 0 &&
+        _synchronizerIndex < _synchronizerSlots.length) {
+      _synchronizerSlots[_synchronizerIndex].state =
+          SynchronizerSlotState.blocked;
+    }
+  }
+
+  /// Reset the synchronizer scan position so the next call to
+  /// [nextAvailableSynchronizer] starts from the beginning.
+  void resetSynchronizerIndex() {
+    _synchronizerIndex = -1;
+  }
+
+  /// Block all non-FDv1 synchronizers and unblock FDv1 synchronizers.
+  void engageFdv1Fallback() {
+    for (final slot in _synchronizerSlots) {
+      slot.state = slot.isFdv1Fallback
+          ? SynchronizerSlotState.available
+          : SynchronizerSlotState.blocked;
+    }
+    _synchronizerIndex = -1;
+  }
+
+  /// True if the current synchronizer is the first available (primary).
+  bool get isPrimarySynchronizer =>
+      _synchronizerIndex == _firstAvailableIndex();
+
+  /// Count of synchronizers in the available state.
+  int get availableSynchronizerCount => _synchronizerSlots
+      .where((slot) => slot.state == SynchronizerSlotState.available)
+      .length;
+
+  /// True if any synchronizer slot is marked as an FDv1 fallback.
+  bool get hasFdv1Fallback =>
+      _synchronizerSlots.any((slot) => slot.isFdv1Fallback);
+
+  /// Close the active source and mark the manager as shut down.
+  void close() {
+    _shutdown = true;
+    _closeActiveSource();
+  }
+}

packages/common_client/test/data_sources/fdv2/source_manager_test.dart

@@ -0,0 +1,179 @@
+import 'dart:async';
+
+import 'package:launchdarkly_common_client/src/data_sources/fdv2/entry_factories.dart';
+import 'package:launchdarkly_common_client/src/data_sources/fdv2/selector.dart';
+import 'package:launchdarkly_common_client/src/data_sources/fdv2/source.dart';
+import 'package:launchdarkly_common_client/src/data_sources/fdv2/source_manager.dart';
+import 'package:launchdarkly_common_client/src/data_sources/fdv2/source_result.dart';
+import 'package:test/test.dart';
+
+final class RecordingInitializer implements Initializer {
+  bool closed = false;
+
+  @override
+  Future<FDv2SourceResult> run() async =>
+      FDv2SourceResults.shutdown(message: 'unused');
+
+  @override
+  void close() {
+    closed = true;
+  }
+}
+
+final class RecordingSynchronizer implements Synchronizer {
+  final int slotIndex;
+  bool closed = false;
+
+  RecordingSynchronizer(this.slotIndex);
+
+  @override
+  Stream<FDv2SourceResult> get results => const Stream.empty();
+
+  @override
+  void close() {
+    closed = true;
+  }
+}
+
+SynchronizerSlot _slot(int index, List<RecordingSynchronizer> created,
+    {bool isFdv1Fallback = false}) {
+  return SynchronizerSlot(
+    isFdv1Fallback: isFdv1Fallback,
+    factory: SynchronizerFactory(create: (_) {
+      final synchronizer = RecordingSynchronizer(index);
+      created.add(synchronizer);
+      return synchronizer;
+    }),
+  );
+}
+
+SourceManager _manager(
+    {List<InitializerFactory> initializers = const [],
+    List<SynchronizerSlot> slots = const []}) {
+  return SourceManager(
+    initializerFactories: initializers,
+    synchronizerSlots: slots,
+    selectorGetter: () => Selector.empty,
+  );
+}
+
+void main() {
+  test('initializers are produced once each, in order, and exhaust', () {
+    final created = <RecordingInitializer>[];
+    InitializerFactory factory() => InitializerFactory(create: (_) {
+          final initializer = RecordingInitializer();
+          created.add(initializer);
+          return initializer;
+        });
+
+    final manager = _manager(initializers: [factory(), factory()]);
+
+    expect(manager.nextInitializer(), isNotNull);
+    expect(manager.nextInitializer(), isNotNull);
+    expect(manager.nextInitializer(), isNull);
+    expect(created, hasLength(2));
+    expect(created.first.closed, isTrue,
+        reason: 'starting the second source closes the first');
+  });
+
+  test('synchronizers cycle through available slots and wrap around', () {
+    final created = <RecordingSynchronizer>[];
+    final manager = _manager(slots: [_slot(0, created), _slot(1, created)]);
+
+    expect(manager.nextAvailableSynchronizer(), isNotNull);
+    expect(created.last.slotIndex, 0);
+    expect(manager.isPrimarySynchronizer, isTrue);
+
+    expect(manager.nextAvailableSynchronizer(), isNotNull);
+    expect(created.last.slotIndex, 1);
+    expect(manager.isPrimarySynchronizer, isFalse);
+
+    expect(manager.nextAvailableSynchronizer(), isNotNull);
+    expect(created.last.slotIndex, 0, reason: 'cycling wraps to the start');
+  });
+
+  test('a blocked slot is skipped', () {
+    final created = <RecordingSynchronizer>[];
+    final manager = _manager(slots: [_slot(0, created), _slot(1, created)]);
+
+    manager.nextAvailableSynchronizer();
+    manager.blockCurrentSynchronizer();
+    expect(manager.availableSynchronizerCount, 1);
+
+    manager.nextAvailableSynchronizer();
+    expect(created.last.slotIndex, 1);
+    manager.nextAvailableSynchronizer();
+    expect(created.last.slotIndex, 1,
+        reason: 'only the unblocked slot remains in rotation');
+  });
+
+  test('all slots blocked yields null', () {
+    final created = <RecordingSynchronizer>[];
+    final manager = _manager(slots: [_slot(0, created)]);
+
+    manager.nextAvailableSynchronizer();
+    manager.blockCurrentSynchronizer();
+    expect(manager.nextAvailableSynchronizer(), isNull);
+  });
+
+  test('resetSynchronizerIndex returns rotation to the first slot', () {
+    final created = <RecordingSynchronizer>[];
+    final manager = _manager(slots: [_slot(0, created), _slot(1, created)]);
+
+    manager.nextAvailableSynchronizer();
+    manager.nextAvailableSynchronizer();
+    manager.resetSynchronizerIndex();
+    manager.nextAvailableSynchronizer();
+    expect(created.last.slotIndex, 0);
+  });
+
+  test(
+      'recreateCurrentSynchronizer produces a fresh instance of the same '
+      'slot', () {
+    final created = <RecordingSynchronizer>[];
+    final manager = _manager(slots: [_slot(0, created), _slot(1, created)]);
+
+    manager.nextAvailableSynchronizer();
+    final recreated = manager.recreateCurrentSynchronizer();
+    expect(recreated, isNotNull);
+    expect(created, hasLength(2));
+    expect(created.last.slotIndex, 0);
+    expect(created.first.closed, isTrue);
+    expect(manager.isPrimarySynchronizer, isTrue);
+  });
+
+  test('FDv1 fallback slots start blocked and engage on fallback', () {
+    final fdv2Created = <RecordingSynchronizer>[];
+    final fdv1Created = <RecordingSynchronizer>[];
+    final manager = _manager(slots: [
+      _slot(0, fdv2Created),
+      _slot(1, fdv1Created, isFdv1Fallback: true),
+    ]);
+
+    expect(manager.hasFdv1Fallback, isTrue);
+    expect(manager.availableSynchronizerCount, 1);
+
+    manager.nextAvailableSynchronizer();
+    expect(fdv2Created, hasLength(1));
+
+    manager.engageFdv1Fallback();
+    expect(manager.availableSynchronizerCount, 1);
+
+    manager.nextAvailableSynchronizer();
+    expect(fdv1Created, hasLength(1));
+    expect(fdv2Created, hasLength(1),
+        reason: 'the FDv2 tier is disabled after fallback');
+  });
+
+  test('close prevents further source creation', () {
+    final created = <RecordingSynchronizer>[];
+    final manager = _manager(slots: [_slot(0, created)]);
+
+    manager.nextAvailableSynchronizer();
+    manager.close();
+    expect(manager.isShutdown, isTrue);
+    expect(created.single.closed, isTrue);
+    expect(manager.nextAvailableSynchronizer(), isNull);
+    expect(manager.recreateCurrentSynchronizer(), isNull);
+  });
+}