From 7528fd53577c26698342fcfe5833943572475ae2 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Wed, 10 Jun 2026 15:59:26 -0700 Subject: [PATCH 1/2] feat: add FDv2 synchronizer fallback and recovery conditions Timed conditions the FDv2 orchestrator will observe alongside the active synchronizer: a fallback condition that fires after a synchronizer has been interrupted too long (120 seconds by default), and a recovery condition that periodically returns to the primary synchronizer (300 seconds by default). Conditions are streams rather than futures so a consumer can detach: cancelling a stream subscription releases the consumer, whereas a listener on a never-completing future can never be removed and would be retained for the condition's whole lifetime. Each condition emits at most once and then closes, lifetimes are scoped to the subscription (timers start on listen and cancellation closes the condition), and nothing consumes these yet. --- .../lib/src/data_sources/fdv2/conditions.dart | 260 +++++++++++++++ .../data_sources/fdv2/conditions_test.dart | 309 ++++++++++++++++++ 2 files changed, 569 insertions(+) create mode 100644 packages/common_client/lib/src/data_sources/fdv2/conditions.dart create mode 100644 packages/common_client/test/data_sources/fdv2/conditions_test.dart diff --git a/packages/common_client/lib/src/data_sources/fdv2/conditions.dart b/packages/common_client/lib/src/data_sources/fdv2/conditions.dart new file mode 100644 index 00000000..2c2b049a --- /dev/null +++ b/packages/common_client/lib/src/data_sources/fdv2/conditions.dart @@ -0,0 +1,260 @@ +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 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 _controller = + StreamController( + 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, () { + _timer = null; + 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 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 _conditions; + final List> _subscriptions = []; + + late final StreamController _controller = + StreamController( + onListen: _subscribe, + onCancel: close, + ); + bool _fired = false; + bool _closed = false; + + ConditionGroup(List 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 empty or closed first. + /// + /// 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 get events => _controller.stream; + + void _subscribe() { + if (_closed) return; + if (_conditions.isEmpty) { + _controller.close(); + 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), + ]); +} diff --git a/packages/common_client/test/data_sources/fdv2/conditions_test.dart b/packages/common_client/test/data_sources/fdv2/conditions_test.dart new file mode 100644 index 00000000..39957ffe --- /dev/null +++ b/packages/common_client/test/data_sources/fdv2/conditions_test.dart @@ -0,0 +1,309 @@ +import 'package:fake_async/fake_async.dart'; +import 'package:launchdarkly_common_client/src/data_sources/fdv2/conditions.dart'; +import 'package:launchdarkly_common_client/src/data_sources/fdv2/payload.dart'; +import 'package:launchdarkly_common_client/src/data_sources/fdv2/source_result.dart'; +import 'package:test/test.dart'; + +ChangeSetResult _changeSet() => const ChangeSetResult( + payload: Payload(type: PayloadType.full, updates: []), + persist: true, + ); + +void main() { + test('fallback condition starts its timer on interrupted and emits', () { + fakeAsync((async) { + final condition = createFallbackCondition(const Duration(seconds: 120)); + final emissions = []; + condition.events.listen(emissions.add); + + expect(async.pendingTimers, isEmpty, + reason: 'the timer does not start until an interruption'); + + condition.inform(FDv2SourceResults.interrupted(message: 'down')); + expect(async.pendingTimers, hasLength(1)); + expect(async.pendingTimers.single.duration, const Duration(seconds: 120)); + + async.elapse(const Duration(seconds: 120)); + expect(emissions, equals([ConditionType.fallback])); + }); + }); + + test('fallback condition cancels its timer when data arrives', () { + fakeAsync((async) { + final condition = createFallbackCondition(const Duration(seconds: 120)); + final emissions = []; + condition.events.listen(emissions.add); + + condition.inform(FDv2SourceResults.interrupted(message: 'down')); + expect(async.pendingTimers, hasLength(1)); + + condition.inform(_changeSet()); + expect(async.pendingTimers, isEmpty); + + async.elapse(const Duration(seconds: 300)); + expect(emissions, isEmpty); + }); + }); + + test('recovery condition starts when listened to and ignores results', () { + fakeAsync((async) { + final condition = createRecoveryCondition(const Duration(seconds: 300)); + final emissions = []; + + expect(async.pendingTimers, isEmpty, + reason: 'the recovery clock starts when the condition is observed'); + condition.events.listen(emissions.add); + + expect(async.pendingTimers, hasLength(1)); + condition.inform(_changeSet()); + expect(async.pendingTimers, hasLength(1), + reason: 'data does not cancel recovery'); + + async.elapse(const Duration(seconds: 300)); + expect(emissions, equals([ConditionType.recovery])); + }); + }); + + test('a fired condition emits exactly once and closes its stream', () { + fakeAsync((async) { + final condition = createFallbackCondition(const Duration(seconds: 120)); + final emissions = []; + var done = false; + condition.events.listen(emissions.add, onDone: () => done = true); + + condition.inform(FDv2SourceResults.interrupted(message: 'down')); + async.elapse(const Duration(seconds: 120)); + + // After firing, further informs cannot re-arm the timer. + condition.inform(FDv2SourceResults.interrupted(message: 'down again')); + expect(async.pendingTimers, isEmpty); + + async.flushMicrotasks(); + expect(emissions, equals([ConditionType.fallback])); + expect(done, isTrue); + }); + }); + + test('a closed condition closes its stream without emitting', () { + fakeAsync((async) { + final condition = createRecoveryCondition(const Duration(seconds: 300)); + final emissions = []; + var done = false; + condition.events.listen(emissions.add, onDone: () => done = true); + + condition.close(); + expect(async.pendingTimers, isEmpty); + + async.elapse(const Duration(seconds: 300)); + expect(emissions, isEmpty); + expect(done, isTrue); + }); + }); + + test('cancelling a subscription closes the condition and its timer', () { + fakeAsync((async) { + final condition = createRecoveryCondition(const Duration(seconds: 300)); + final emissions = []; + + final subscription = condition.events.listen(emissions.add); + expect(async.pendingTimers, hasLength(1)); + + subscription.cancel(); + async.flushMicrotasks(); + expect(async.pendingTimers, isEmpty, + reason: 'the condition lifetime is scoped to the subscription'); + + async.elapse(const Duration(seconds: 300)); + expect(emissions, isEmpty); + }); + }); + + test('terminal statuses do not arm the fallback timer', () { + fakeAsync((async) { + final condition = createFallbackCondition(const Duration(seconds: 120)); + condition.events.listen((_) {}); + + condition.inform(FDv2SourceResults.terminalError(message: 'denied')); + condition.inform(FDv2SourceResults.shutdown(message: 'closed')); + condition.inform(FDv2SourceResults.goodbyeResult(message: 'bye')); + + expect(async.pendingTimers, isEmpty, + reason: 'the orchestrator reacts to terminal statuses ' + 'immediately rather than waiting out a fallback period'); + condition.close(); + }); + }); + + test('a second interruption does not extend the fallback deadline', () { + fakeAsync((async) { + final condition = createFallbackCondition(const Duration(seconds: 120)); + final emissions = []; + condition.events.listen(emissions.add); + + condition.inform(FDv2SourceResults.interrupted(message: 'down')); + async.elapse(const Duration(seconds: 60)); + condition.inform(FDv2SourceResults.interrupted(message: 'still down')); + async.elapse(const Duration(seconds: 60)); + + expect(emissions, equals([ConditionType.fallback]), + reason: 'the fallback period counts from the first interruption'); + }); + }); + + group('ConditionGroup', () { + test('emits the first member condition to fire, then closes', () { + fakeAsync((async) { + final group = ConditionGroup([ + createFallbackCondition(const Duration(seconds: 120)), + createRecoveryCondition(const Duration(seconds: 300)), + ]); + final emissions = []; + var done = false; + group.events.listen(emissions.add, onDone: () => done = true); + + // Only the recovery timer is running; fire it. + async.elapse(const Duration(seconds: 300)); + expect(emissions, equals([ConditionType.recovery])); + expect(done, isTrue); + }); + }); + + test('emits exactly once when two member timers contend', () { + fakeAsync((async) { + final group = ConditionGroup([ + createFallbackCondition(const Duration(seconds: 120)), + createRecoveryCondition(const Duration(seconds: 120)), + ]); + final emissions = []; + var done = false; + group.events.listen(emissions.add, onDone: () => done = true); + + // Arm the fallback timer so both members are due at the same + // instant. + group.inform(FDv2SourceResults.interrupted(message: 'down')); + expect(async.pendingTimers, hasLength(2)); + + async.elapse(const Duration(seconds: 120)); + expect(emissions, hasLength(1), + reason: 'the first member to fire wins; the loser must not ' + 'produce a second emission'); + expect(done, isTrue); + }); + }); + + test('cancelling the subscription closes the group and member timers', () { + fakeAsync((async) { + final group = ConditionGroup([ + createFallbackCondition(const Duration(seconds: 120)), + createRecoveryCondition(const Duration(seconds: 300)), + ]); + final emissions = []; + final subscription = group.events.listen(emissions.add); + group.inform(FDv2SourceResults.interrupted(message: 'down')); + expect(async.pendingTimers, hasLength(2)); + + subscription.cancel(); + async.flushMicrotasks(); + expect(async.pendingTimers, isEmpty, + reason: 'the group lifetime is scoped to the subscription'); + + group.inform(FDv2SourceResults.interrupted(message: 'down again')); + expect(async.pendingTimers, isEmpty, + reason: 'a closed group does not forward informs'); + + async.elapse(const Duration(seconds: 300)); + expect(emissions, isEmpty); + }); + }); + + test('inform reaches every member condition', () { + fakeAsync((async) { + final group = ConditionGroup( + [createFallbackCondition(const Duration(seconds: 120))]); + group.events.listen((_) {}); + + expect(async.pendingTimers, isEmpty); + group.inform(FDv2SourceResults.interrupted(message: 'down')); + expect(async.pendingTimers, hasLength(1)); + group.close(); + }); + }); + + test('an empty group closes without emitting', () { + fakeAsync((async) { + final group = ConditionGroup(const []); + final emissions = []; + var done = false; + group.events.listen(emissions.add, onDone: () => done = true); + + async.flushMicrotasks(); + expect(emissions, isEmpty); + expect(done, isTrue); + }); + }); + + test('close closes the members and the merged stream without emitting', () { + fakeAsync((async) { + final group = ConditionGroup( + [createRecoveryCondition(const Duration(seconds: 300))]); + final emissions = []; + var done = false; + group.events.listen(emissions.add, onDone: () => done = true); + + group.close(); + expect(async.pendingTimers, isEmpty, + reason: 'closing the group cancels member timers'); + + async.elapse(const Duration(seconds: 300)); + expect(emissions, isEmpty); + expect(done, isTrue); + }); + }); + }); + + group('getConditions', () { + test('no conditions with a single available synchronizer', () { + fakeAsync((async) { + final group = getConditions( + availableSynchronizerCount: 1, + isPrimary: true, + ); + var done = false; + group.events.listen((_) {}, onDone: () => done = true); + + async.flushMicrotasks(); + expect(done, isTrue); + }); + }); + + test('primary synchronizer gets only a fallback condition', () { + fakeAsync((async) { + final group = getConditions( + availableSynchronizerCount: 2, + isPrimary: true, + ); + group.events.listen((_) {}); + + expect(async.pendingTimers, isEmpty, + reason: 'a recovery condition would have started a timer'); + group.close(); + }); + }); + + test('non-primary synchronizer also gets a recovery condition', () { + fakeAsync((async) { + final group = getConditions( + availableSynchronizerCount: 2, + isPrimary: false, + ); + final emissions = []; + group.events.listen(emissions.add); + + expect(async.pendingTimers, hasLength(1), + reason: 'the recovery timer starts immediately'); + async.elapse(const Duration(seconds: 300)); + expect(emissions, equals([ConditionType.recovery])); + }); + }); + }); +} From c6616d1a51a25497631c95fcf3ec439d0dde9cfa Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Thu, 11 Jun 2026 11:56:31 -0700 Subject: [PATCH 2/2] refactor: simplify condition timer and empty-group handling --- .../lib/src/data_sources/fdv2/conditions.dart | 8 +--- .../data_sources/fdv2/conditions_test.dart | 42 ++++++++++++++++--- 2 files changed, 39 insertions(+), 11 deletions(-) diff --git a/packages/common_client/lib/src/data_sources/fdv2/conditions.dart b/packages/common_client/lib/src/data_sources/fdv2/conditions.dart index 2c2b049a..080c68ea 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/conditions.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/conditions.dart @@ -87,7 +87,6 @@ final class _TimedCondition implements Condition { void _startTimer() { if (_timer != null || _closed) return; _timer = Timer(_timeout, () { - _timer = null; if (_closed) return; _closed = true; // The controller buffers the event if the subscription has not @@ -175,7 +174,8 @@ final class ConditionGroup { /// 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 empty or closed first. + /// 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 @@ -186,10 +186,6 @@ final class ConditionGroup { void _subscribe() { if (_closed) return; - if (_conditions.isEmpty) { - _controller.close(); - return; - } for (final condition in _conditions) { _subscriptions.add(condition.events.listen((type) { // First member to fire wins; member timers firing in the same diff --git a/packages/common_client/test/data_sources/fdv2/conditions_test.dart b/packages/common_client/test/data_sources/fdv2/conditions_test.dart index 39957ffe..b94cff81 100644 --- a/packages/common_client/test/data_sources/fdv2/conditions_test.dart +++ b/packages/common_client/test/data_sources/fdv2/conditions_test.dart @@ -45,6 +45,25 @@ void main() { }); }); + test('fallback condition re-arms for an interruption after a cancel', () { + fakeAsync((async) { + final condition = createFallbackCondition(const Duration(seconds: 120)); + final emissions = []; + condition.events.listen(emissions.add); + + condition.inform(FDv2SourceResults.interrupted(message: 'down')); + condition.inform(_changeSet()); + expect(async.pendingTimers, isEmpty); + + // A fresh interruption begins a new fallback period. + condition.inform(FDv2SourceResults.interrupted(message: 'down again')); + expect(async.pendingTimers, hasLength(1)); + + async.elapse(const Duration(seconds: 120)); + expect(emissions, [ConditionType.fallback]); + }); + }); + test('recovery condition starts when listened to and ignores results', () { fakeAsync((async) { final condition = createRecoveryCondition(const Duration(seconds: 300)); @@ -229,13 +248,22 @@ void main() { }); }); - test('an empty group closes without emitting', () { + test('an empty group never emits and closes when closed', () { fakeAsync((async) { final group = ConditionGroup(const []); final emissions = []; var done = false; group.events.listen(emissions.add, onDone: () => done = true); + group.inform(FDv2SourceResults.interrupted(message: 'down')); + async.flushMicrotasks(); + expect(async.pendingTimers, isEmpty); + expect(emissions, isEmpty); + expect(done, isFalse, + reason: 'no member conditions exist, so nothing can fire and ' + 'the stream stays open until the group is closed'); + + group.close(); async.flushMicrotasks(); expect(emissions, isEmpty); expect(done, isTrue); @@ -268,11 +296,15 @@ void main() { availableSynchronizerCount: 1, isPrimary: true, ); - var done = false; - group.events.listen((_) {}, onDone: () => done = true); + final emissions = []; + group.events.listen(emissions.add); - async.flushMicrotasks(); - expect(done, isTrue); + group.inform(FDv2SourceResults.interrupted(message: 'down')); + async.elapse(const Duration(hours: 1)); + expect(async.pendingTimers, isEmpty, + reason: 'there is nowhere to fall back to, so no timers arm'); + expect(emissions, isEmpty); + group.close(); }); });