diff --git a/packages/common_client/lib/src/data_sources/fdv2/source_manager.dart b/packages/common_client/lib/src/data_sources/fdv2/source_manager.dart new file mode 100644 index 00000000..301a1f7b --- /dev/null +++ b/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 _initializerFactories; + final List _synchronizerSlots; + final SelectorGetter _selectorGetter; + + Initializer? _activeInitializer; + Synchronizer? _activeSynchronizer; + int _initializerIndex = -1; + int _synchronizerIndex = -1; + bool _shutdown = false; + + SourceManager({ + required List initializerFactories, + required List 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(); + } +} diff --git a/packages/common_client/test/data_sources/fdv2/source_manager_test.dart b/packages/common_client/test/data_sources/fdv2/source_manager_test.dart new file mode 100644 index 00000000..b57b280d --- /dev/null +++ b/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 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 get results => const Stream.empty(); + + @override + void close() { + closed = true; + } +} + +SynchronizerSlot _slot(int index, List created, + {bool isFdv1Fallback = false}) { + return SynchronizerSlot( + isFdv1Fallback: isFdv1Fallback, + factory: SynchronizerFactory(create: (_) { + final synchronizer = RecordingSynchronizer(index); + created.add(synchronizer); + return synchronizer; + }), + ); +} + +SourceManager _manager( + {List initializers = const [], + List 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 = []; + 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 = []; + 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 = []; + 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 = []; + 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 = []; + 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 = []; + 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 = []; + final fdv1Created = []; + 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 = []; + 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); + }); +}