diff --git a/packages/common_client/lib/src/data_sources/fdv2/fallback_directive.dart b/packages/common_client/lib/src/data_sources/fdv2/fallback_directive.dart index 27eb1693..1e104cc4 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/fallback_directive.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/fallback_directive.dart @@ -1,7 +1,14 @@ +import 'dart:math'; + +/// The longest server-supplied fallback TTL that is honored, and the +/// (jittered) default used when the server gives no usable TTL. +const Duration maxFallbackTtl = Duration(hours: 1); + /// 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). +/// [ttl] is the server-supplied time to remain on the fallback before +/// retrying FDv2, unvalidated (null when absent or not an integer). See +/// [resolveFallbackTtl]. final class FallbackDirective { final Duration? ttl; const FallbackDirective(this.ttl); @@ -18,3 +25,18 @@ FallbackDirective? readFallbackDirective(Map headers) { final seconds = raw == null ? null : int.tryParse(raw); return FallbackDirective(seconds == null ? null : Duration(seconds: seconds)); } + +/// Returns how long to remain on the fallback before retrying FDv2. A +/// server-supplied [ttl] in (0, [maxFallbackTtl]] is used as-is. Anything +/// else (absent, zero, negative, or too long) uses [maxFallbackTtl] minus +/// a uniformly random jitter of up to half of it, so SDKs that fell back +/// together do not all retry at once. Fallback is never indefinite. +Duration resolveFallbackTtl(Duration? ttl, [Random? random]) { + if (ttl case final valid? + when valid > Duration.zero && valid <= maxFallbackTtl) { + return valid; + } + final maxJitterSeconds = maxFallbackTtl.inSeconds ~/ 2; + final jitterSeconds = (random ?? Random()).nextInt(maxJitterSeconds + 1); + return maxFallbackTtl - Duration(seconds: jitterSeconds); +} diff --git a/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart b/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart index 8d54892e..1bbf934e 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/orchestrator.dart @@ -8,6 +8,7 @@ import '../data_source_status.dart'; import '../data_source_status_manager.dart'; import 'conditions.dart'; import 'entry_factories.dart'; +import 'fallback_directive.dart'; import 'payload.dart'; import 'selector.dart'; import 'source.dart'; @@ -39,13 +40,11 @@ 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 server directed FDv1 fallback: the FDv2 synchronizers are + /// disabled (and the FDv1 fallback tier, if configured, enabled) until + /// the directive's TTL elapses; advance to the next available + /// synchronizer. + fallBack, } /// The FDv2 data source orchestrator. @@ -68,6 +67,7 @@ final class FDv2DataSourceOrchestrator implements DataSource { final Duration _fallbackTimeout; final Duration _recoveryTimeout; final Duration _recycleDelay; + final Duration Function(Duration? ttl) _fallbackTtlResolver; final LDLogger _logger; final StreamController _controller = @@ -78,13 +78,13 @@ 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); + /// Restores the FDv2 synchronizers once a fallback directive's TTL + /// elapses. + Timer? _fdv2RecoveryTimer; - /// Overrides the next recycle delay when set (consumed once), used to - /// honor a fallback directive's retry TTL. - Duration? _pendingRetryDelay; + /// Completes when [_fdv2RecoveryTimer] fires, so a synchronizer loop + /// with nothing available can wait for FDv2 to be restored. + Completer? _fdv2Recovered; /// Completes when [stop] is called so a long interruptible wait (e.g. a /// fallback retry delay) wakes promptly on shutdown. @@ -110,6 +110,7 @@ final class FDv2DataSourceOrchestrator implements DataSource { Duration fallbackTimeout = defaultFallbackTimeout, Duration recoveryTimeout = defaultRecoveryTimeout, Duration recycleDelay = const Duration(seconds: 1), + Duration Function(Duration? ttl) fallbackTtlResolver = resolveFallbackTtl, }) : _initializerFactories = initializerFactories, _synchronizerSlots = synchronizerSlots, _selectorUpdater = selectorUpdater, @@ -117,6 +118,7 @@ final class FDv2DataSourceOrchestrator implements DataSource { _fallbackTimeout = fallbackTimeout, _recoveryTimeout = recoveryTimeout, _recycleDelay = recycleDelay, + _fallbackTtlResolver = fallbackTtlResolver, _logger = logger.subLogger('FDv2Orchestrator'), _cacheOnlyDataSystem = initializerFactories.isNotEmpty && initializerFactories.every((f) => f.isCache) && @@ -144,6 +146,7 @@ final class FDv2DataSourceOrchestrator implements DataSource { if (_closed) return; _closed = true; if (!_stopCompleter.isCompleted) _stopCompleter.complete(); + _fdv2RecoveryTimer?.cancel(); _sourceManager.close(); // Wake the active synchronizer run so it can observe the closed // state. @@ -217,45 +220,41 @@ final class FDv2DataSourceOrchestrator implements DataSource { _controller.add(InitializedEvent()); } - /// 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). When - /// there is no FDv1 fallback tier we defer retrying FDv2 - /// ([_DirectiveAction.deferRetry]). + /// Handles a server-directed FDv1 fallback: disables the FDv2 + /// synchronizers (enabling the FDv1 fallback tier, if configured) and + /// schedules their recovery after the directive's TTL. 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). _DirectiveAction _processDirective(FDv2SourceResult result) { if (!result.fdv1Fallback || _sourceManager.isCurrentSynchronizerFdv1Fallback) { return _DirectiveAction.none; } + final ttl = _fallbackTtlResolver(result.fdv1FallbackTtl); if (_sourceManager.hasFdv1FallbackConfigured) { _logger.warn('Server directed fallback to FDv1; engaging the FDv1 ' - 'fallback synchronizer.'); - _sourceManager.engageFdv1Fallback(); - return _DirectiveAction.engageFdv1Fallback; + 'fallback synchronizer and retrying FDv2 in ${ttl.inSeconds}s.'); + } else { + _logger.warn('Server directed FDv1 fallback but no fallback tier is ' + 'configured; pausing FDv2 and retrying it in ${ttl.inSeconds}s.'); } - return _DirectiveAction.deferRetry; + _sourceManager.engageFdv1Fallback(); + _scheduleFdv2Recovery(ttl); + return _DirectiveAction.fallBack; } - /// 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; - } - 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); + void _scheduleFdv2Recovery(Duration ttl) { + _fdv2RecoveryTimer?.cancel(); + final recovered = Completer(); + _fdv2Recovered = recovered; + _fdv2RecoveryTimer = Timer(ttl, () { + if (_closed) return; + _logger.info('Fallback TTL elapsed; retrying FDv2.'); + _sourceManager.restoreFdv2Synchronizers(); + recovered.complete(); + _resolveCurrentOutcome?.call(_SynchronizerOutcome.recover); + }); } /// Waits [d], returning early if the orchestrator is stopped so a long @@ -283,16 +282,15 @@ final class FDv2DataSourceOrchestrator implements DataSource { dataReceived = true; switch (_processDirective(result)) { - case _DirectiveAction.engageFdv1Fallback: + case _DirectiveAction.fallBack: // Data received but the server directed FDv1 fallback; - // move on to synchronizers where the fallback tier runs - // and its first change set completes initialization. + // move on to synchronizers. The FDv1 tier's first change + // set completes initialization; without one, this data + // does. + if (!_sourceManager.hasFdv1FallbackConfigured) { + _emitInitialized(); + } 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: @@ -317,8 +315,10 @@ final class FDv2DataSourceOrchestrator implements DataSource { case SourceState.goodbye: break; } - if (_processDirective(result) == - _DirectiveAction.engageFdv1Fallback) { + if (_processDirective(result) == _DirectiveAction.fallBack) { + if (dataReceived && !_sourceManager.hasFdv1FallbackConfigured) { + _emitInitialized(); + } return; } } @@ -363,18 +363,25 @@ final class FDv2DataSourceOrchestrator implements DataSource { while (!_closed) { if (recycleCurrent) { recycleCurrent = false; - final delay = _pendingRetryDelay ?? _recycleDelay; - _pendingRetryDelay = null; - if (delay > Duration.zero) { - await _delay(delay); + if (_recycleDelay > Duration.zero) { + await _delay(_recycleDelay); if (_closed) return; } - synchronizer = _sourceManager.recreateCurrentSynchronizer(); + // The current slot may have been blocked during the delay (e.g. + // FDv2 was restored from the FDv1 fallback); move on instead. + synchronizer = _sourceManager.recreateCurrentSynchronizer() ?? + _sourceManager.nextAvailableSynchronizer(); } else { synchronizer = _sourceManager.nextAvailableSynchronizer(); } if (synchronizer == null) { + // Fallback is never indefinite: with no FDv1 tier (or an unusable + // one), wait for FDv2 to be restored. + if (_fdv2Recovered case final recovered? when !recovered.isCompleted) { + await Future.any([recovered.future, _stopCompleter.future]); + continue; + } if (!_emittedPayload) { _halt('All FDv2 data sources exhausted without receiving data'); } else if (_synchronizerSlots.isNotEmpty) { @@ -461,10 +468,8 @@ final class FDv2DataSourceOrchestrator implements DataSource { '${result.message ?? 'unknown error'}'); _reportTransientError(result); switch (_processDirective(result)) { - case _DirectiveAction.engageFdv1Fallback: + case _DirectiveAction.fallBack: resolve(_SynchronizerOutcome.advance); - case _DirectiveAction.deferRetry: - _scheduleFallbackRetry(result.fdv1FallbackTtl, resolve); case _DirectiveAction.none: _sourceManager.blockCurrentSynchronizer(); resolve(_SynchronizerOutcome.advance); @@ -492,13 +497,11 @@ final class FDv2DataSourceOrchestrator implements DataSource { } switch (_processDirective(result)) { - case _DirectiveAction.engageFdv1Fallback: + case _DirectiveAction.fallBack: // 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; } diff --git a/packages/common_client/lib/src/data_sources/fdv2/protocol_handler.dart b/packages/common_client/lib/src/data_sources/fdv2/protocol_handler.dart index 5e64bf0d..ee0d9aa7 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/protocol_handler.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/protocol_handler.dart @@ -56,9 +56,9 @@ final class ActionGoodbye extends ProtocolAction { final String reason; /// When present, the goodbye carries an in-band protocol-fallback - /// directive (FDv2 to FDv1) with this time-to-live; [Duration.zero] - /// means remain on the fallback indefinitely. `null` means an ordinary - /// disconnect with no fallback directive. + /// directive (FDv2 to FDv1) with this unvalidated time-to-live (see + /// `resolveFallbackTtl`). `null` means an ordinary disconnect with no + /// fallback directive. final Duration? protocolFallbackTtl; const ActionGoodbye(this.reason, {this.protocolFallbackTtl}); diff --git a/packages/common_client/lib/src/data_sources/fdv2/protocol_types.dart b/packages/common_client/lib/src/data_sources/fdv2/protocol_types.dart index e0423b9a..a2d4f5a5 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/protocol_types.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/protocol_types.dart @@ -152,8 +152,8 @@ final class GoodbyeEvent { /// When present, the server is directing a protocol fallback (FDv2 to /// FDv1) in-band -- the same signal as the `x-ld-fd-fallback` response /// header, for transports that cannot read response headers. The value - /// is how long to remain on the fallback before retrying; [Duration.zero] - /// means indefinitely. `null` means no fallback was directed. + /// is how long to remain on the fallback before retrying, unvalidated + /// (see `resolveFallbackTtl`). `null` means no fallback was directed. final Duration? protocolFallbackTtl; const GoodbyeEvent({ 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 index 41dbfa03..5fcd24a3 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/source_manager.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/source_manager.dart @@ -168,16 +168,12 @@ final class SourceManager { _synchronizerIndex = -1; } - /// Block all non-FDv1 synchronizers, unblock the FDv1 fallback, and - /// reset the scan cursor so the next activation selects the fallback. - /// Does nothing when no FDv1 fallback slot is configured, since - /// blocking every slot without unblocking one would leave nothing to - /// activate. The active source keeps running until the next - /// activation closes it. + /// Block all non-FDv1 synchronizers, unblock the FDv1 fallback (if + /// configured), and reset the scan cursor so the next activation + /// selects the fallback. With no FDv1 fallback slot nothing remains to + /// activate until [restoreFdv2Synchronizers]. The active source keeps + /// running until the next activation closes it. void engageFdv1Fallback() { - if (!hasFdv1FallbackConfigured) { - return; - } for (final slot in _synchronizerSlots) { slot.state = slot.isFdv1Fallback ? SynchronizerSlotState.available @@ -186,6 +182,20 @@ final class SourceManager { _synchronizerIndex = -1; } + /// Undo [engageFdv1Fallback] once the fallback TTL elapses: return + /// every slot to its initial state (FDv2 synchronizers available, the + /// FDv1 fallback blocked) and reset the scan cursor so the next + /// activation selects the primary. The active source keeps running + /// until the next activation closes it. + void restoreFdv2Synchronizers() { + for (final slot in _synchronizerSlots) { + slot.state = slot.isFdv1Fallback + ? SynchronizerSlotState.blocked + : SynchronizerSlotState.available; + } + _synchronizerIndex = -1; + } + /// True if the active synchronizer occupies the first available slot /// (the primary). Reads the scan cursor, so it is only meaningful /// while the cursor identifies the active slot (see the class diff --git a/packages/common_client/lib/src/data_sources/fdv2/source_result.dart b/packages/common_client/lib/src/data_sources/fdv2/source_result.dart index 20aaf31c..2649a1ba 100644 --- a/packages/common_client/lib/src/data_sources/fdv2/source_result.dart +++ b/packages/common_client/lib/src/data_sources/fdv2/source_result.dart @@ -21,9 +21,9 @@ sealed class FDv2SourceResult { final bool fdv1Fallback; /// How long to remain on the fallback before retrying FDv2, when - /// [fdv1Fallback] is set. `null` means the server gave no time-to-live - /// (the caller applies its default); [Duration.zero] means remain - /// indefinitely with no automatic retry. + /// [fdv1Fallback] is set, as supplied by the server and unvalidated. + /// `null` means the server gave no usable time-to-live. The orchestrator + /// validates it and applies the default (see `resolveFallbackTtl`). final Duration? fdv1FallbackTtl; const FDv2SourceResult({this.fdv1Fallback = false, this.fdv1FallbackTtl}); diff --git a/packages/common_client/test/data_sources/fdv2/fallback_directive_test.dart b/packages/common_client/test/data_sources/fdv2/fallback_directive_test.dart new file mode 100644 index 00000000..2248c460 --- /dev/null +++ b/packages/common_client/test/data_sources/fdv2/fallback_directive_test.dart @@ -0,0 +1,70 @@ +import 'dart:math'; + +import 'package:launchdarkly_common_client/src/data_sources/fdv2/fallback_directive.dart'; +import 'package:test/test.dart'; + +final class FixedRandom implements Random { + final int Function(int max) _nextInt; + + FixedRandom(this._nextInt); + + @override + int nextInt(int max) => _nextInt(max); + + @override + bool nextBool() => throw UnimplementedError(); + + @override + double nextDouble() => throw UnimplementedError(); +} + +void main() { + group('resolveFallbackTtl', () { + for (final ttl in [ + const Duration(seconds: 1), + const Duration(seconds: 90), + const Duration(hours: 1), + ]) { + test('uses a valid TTL of ${ttl.inSeconds}s as-is', () { + expect(resolveFallbackTtl(ttl), ttl); + }); + } + + for (final ttl in [ + null, + Duration.zero, + const Duration(seconds: -5), + const Duration(seconds: 3601), + ]) { + test('replaces a TTL of ${ttl?.inSeconds}s with the jittered default', + () { + for (var i = 0; i < 100; i++) { + final resolved = resolveFallbackTtl(ttl); + expect(resolved, greaterThanOrEqualTo(const Duration(minutes: 30))); + expect(resolved, lessThanOrEqualTo(const Duration(hours: 1))); + } + }); + } + + test('the jitter spans the full range from 30 minutes to 1 hour', () { + expect(resolveFallbackTtl(null, FixedRandom((max) => 0)), + const Duration(hours: 1)); + expect(resolveFallbackTtl(null, FixedRandom((max) => max - 1)), + const Duration(minutes: 30)); + }); + }); + + group('readFallbackDirective', () { + test('returns null without the fallback header', () { + expect(readFallbackDirective({'x-ld-fd-fallback-ttl': '60'}), isNull); + }); + + test('carries a malformed TTL header as null', () { + final directive = readFallbackDirective( + {'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': 'soon'}); + + expect(directive, isNotNull); + expect(directive!.ttl, isNull); + }); + }); +} diff --git a/packages/common_client/test/data_sources/fdv2/orchestrator_test.dart b/packages/common_client/test/data_sources/fdv2/orchestrator_test.dart index 2c72d124..d3780e2a 100644 --- a/packages/common_client/test/data_sources/fdv2/orchestrator_test.dart +++ b/packages/common_client/test/data_sources/fdv2/orchestrator_test.dart @@ -5,6 +5,7 @@ import 'package:launchdarkly_common_client/src/data_sources/data_source.dart'; import 'package:launchdarkly_common_client/src/data_sources/data_source_status.dart'; import 'package:launchdarkly_common_client/src/data_sources/data_source_status_manager.dart'; import 'package:launchdarkly_common_client/src/data_sources/fdv2/entry_factories.dart'; +import 'package:launchdarkly_common_client/src/data_sources/fdv2/fallback_directive.dart'; import 'package:launchdarkly_common_client/src/data_sources/fdv2/orchestrator.dart'; import 'package:launchdarkly_common_client/src/data_sources/fdv2/payload.dart'; import 'package:launchdarkly_common_client/src/data_sources/fdv2/selector.dart'; @@ -78,11 +79,13 @@ ChangeSetResult changeSet({ Selector selector = Selector.empty, PayloadType type = PayloadType.full, bool fdv1Fallback = false, + Duration? fdv1FallbackTtl, }) { return ChangeSetResult( changeSet: ChangeSet(selector: selector, type: type, updates: const {}), persist: true, fdv1Fallback: fdv1Fallback, + fdv1FallbackTtl: fdv1FallbackTtl, ); } @@ -98,6 +101,7 @@ final class Harness { required List synchronizerSlots, Duration fallbackTimeout = const Duration(seconds: 120), Duration recoveryTimeout = const Duration(seconds: 300), + Duration Function(Duration? ttl) fallbackTtlResolver = resolveFallbackTtl, }) { orchestrator = FDv2DataSourceOrchestrator( initializerFactories: initializerFactories, @@ -112,6 +116,7 @@ final class Harness { fallbackTimeout: fallbackTimeout, recoveryTimeout: recoveryTimeout, recycleDelay: Duration.zero, + fallbackTtlResolver: fallbackTtlResolver, ); subscription = orchestrator.events.listen(events.add); } @@ -121,6 +126,17 @@ final class Harness { await Future.delayed(Duration.zero); } } + + /// Waits, polling every millisecond, until [condition] holds. + Future pumpUntil(bool Function() condition) async { + final deadline = DateTime.now().add(const Duration(seconds: 5)); + while (!condition()) { + if (DateTime.now().isAfter(deadline)) { + fail('condition not met within 5s'); + } + await Future.delayed(const Duration(milliseconds: 1)); + } + } } void main() { @@ -577,55 +593,223 @@ void main() { harness.orchestrator.stop(); }); - test( - 'a directive with no fallback tier and no TTL does not reconnect ' - 'promptly (defaults to a long retry interval)', () async { + /// Resolves a fallback TTL as the SDK does, recording the result, but + /// scales it down by 100000 (an hour becomes 36ms) so tests can wait it + /// out. + Duration Function(Duration?) scaledResolver(List resolved) { + return (ttl) { + final duration = resolveFallbackTtl(ttl); + resolved.add(duration); + return duration ~/ 100000; + }; + } + + for (final ttl in [ + null, + Duration.zero, + const Duration(seconds: -1), + const Duration(hours: 2), + ]) { + test( + 'a directive with no fallback tier and a TTL of ${ttl?.inSeconds}s ' + 'retries FDv2 after the jittered default', () async { + final fdv2Tier = []; + final resolved = []; + + final harness = Harness( + initializerFactories: [], + synchronizerSlots: [synchronizerSlot(fdv2Tier)], + fallbackTtlResolver: scaledResolver(resolved), + ); + + harness.orchestrator.start(); + await harness.pump(); + fdv2Tier.first.controller.add(FDv2SourceResults.terminalError( + message: 'fall back', fdv1Fallback: true, fdv1FallbackTtl: ttl)); + await harness.pumpUntil(() => fdv2Tier.length == 2); + + expect( + resolved.single, greaterThanOrEqualTo(const Duration(minutes: 30))); + expect(resolved.single, lessThanOrEqualTo(const Duration(hours: 1))); + expect(harness.events.whereType(), isEmpty, + reason: 'the directive does not halt the data system'); + + harness.orchestrator.stop(); + }); + } + + test('an FDv1 fallback returns to FDv2 after the server TTL', () async { final fdv2Tier = []; + final fdv1Tier = []; + final harness = Harness( initializerFactories: [], - synchronizerSlots: [synchronizerSlot(fdv2Tier)], + synchronizerSlots: [ + synchronizerSlot(fdv2Tier), + synchronizerSlot(fdv1Tier, isFdv1Fallback: true), + ], ); harness.orchestrator.start(); await harness.pump(); - - fdv2Tier.first.controller.add(FDv2SourceResults.terminalError( - message: 'fall back', fdv1Fallback: true)); - await harness.pump(); - await Future.delayed(const Duration(milliseconds: 80)); + fdv2Tier.single.controller.add(FDv2SourceResults.terminalError( + message: 'fall back', + fdv1Fallback: true, + fdv1FallbackTtl: const Duration(milliseconds: 20))); await harness.pump(); - expect(harness.events.whereType(), isEmpty); - expect(fdv2Tier, hasLength(1), - reason: 'with no TTL the default retry interval is long, so no ' - 'reconnect is observed in the test window'); + expect(fdv2Tier.single.closed, isTrue); + expect(fdv1Tier, hasLength(1)); + + await harness.pumpUntil(() => fdv2Tier.length == 2); + + expect(fdv1Tier.single.closed, isTrue, + reason: 'FDv1 stops before FDv2 resumes'); + expect(fdv2Tier.last.closed, isFalse); harness.orchestrator.stop(); }); - test( - 'a directive with no fallback tier and a zero TTL pauses FDv2 with no ' - 'retry', () async { + test('an FDv1 fallback uses a valid server TTL as-is', () async { final fdv2Tier = []; + final fdv1Tier = []; + final resolved = []; + final harness = Harness( initializerFactories: [], - synchronizerSlots: [synchronizerSlot(fdv2Tier)], + synchronizerSlots: [ + synchronizerSlot(fdv2Tier), + synchronizerSlot(fdv1Tier, isFdv1Fallback: true), + ], + fallbackTtlResolver: scaledResolver(resolved), ); harness.orchestrator.start(); await harness.pump(); + fdv2Tier.single.controller.add(FDv2SourceResults.terminalError( + message: 'fall back', + fdv1Fallback: true, + fdv1FallbackTtl: const Duration(seconds: 90))); + await harness.pumpUntil(() => fdv2Tier.length == 2); - fdv2Tier.first.controller.add(FDv2SourceResults.terminalError( + expect(resolved.single, const Duration(seconds: 90)); + + harness.orchestrator.stop(); + }); + + test('an FDv1 fallback with a zero TTL returns to FDv2 within an hour', + () async { + final fdv2Tier = []; + final fdv1Tier = []; + final resolved = []; + + final harness = Harness( + initializerFactories: [], + synchronizerSlots: [ + synchronizerSlot(fdv2Tier), + synchronizerSlot(fdv1Tier, isFdv1Fallback: true), + ], + fallbackTtlResolver: scaledResolver(resolved), + ); + + harness.orchestrator.start(); + await harness.pump(); + fdv2Tier.single.controller.add(FDv2SourceResults.terminalError( message: 'fall back', fdv1Fallback: true, fdv1FallbackTtl: Duration.zero)); + await harness.pumpUntil(() => fdv2Tier.length == 2); + + expect(resolved.single, greaterThanOrEqualTo(const Duration(minutes: 30))); + expect(resolved.single, lessThanOrEqualTo(const Duration(hours: 1))); + expect(fdv1Tier.single.closed, isTrue); + + harness.orchestrator.stop(); + }); + + test('an FDv1 fallback returns to FDv2 even when the FDv1 tier fails', + () async { + final fdv2Tier = []; + final fdv1Tier = []; + + final harness = Harness( + initializerFactories: [], + synchronizerSlots: [ + synchronizerSlot(fdv2Tier), + synchronizerSlot(fdv1Tier, isFdv1Fallback: true), + ], + ); + + harness.orchestrator.start(); await harness.pump(); - await Future.delayed(const Duration(milliseconds: 60)); + fdv2Tier.single.controller.add(changeSet( + selector: const Selector(state: 'from-stream', version: 1), + fdv1Fallback: true, + fdv1FallbackTtl: const Duration(milliseconds: 20))); await harness.pump(); + fdv1Tier.single.controller + .add(FDv2SourceResults.terminalError(message: 'FDv1 failed')); + await harness.pumpUntil(() => fdv2Tier.length == 2); - expect(harness.events.whereType(), isEmpty, - reason: 'pausing is not a halt'); - expect(fdv2Tier, hasLength(1), reason: 'a zero TTL means no retry'); + expect(harness.events.whereType(), isEmpty); + + harness.orchestrator.stop(); + }); + + test( + 'an initializer directive with no fallback tier delays the ' + 'synchronizers until the TTL elapses', () async { + final fdv2Tier = []; + + final harness = Harness( + initializerFactories: [ + initializerFactory(changeSet( + selector: const Selector(state: 'from-poll', version: 1), + fdv1Fallback: true, + fdv1FallbackTtl: const Duration(milliseconds: 20))), + ], + synchronizerSlots: [synchronizerSlot(fdv2Tier)], + ); + + harness.orchestrator.start(); + await harness.pump(); + + expect(harness.events.whereType(), hasLength(1)); + expect(harness.events.whereType(), hasLength(1)); + expect(fdv2Tier, isEmpty); + + await harness.pumpUntil(() => fdv2Tier.length == 1); + + harness.orchestrator.stop(); + }); + + test( + 'an initializer error directive skips the remaining initializers and ' + 'enables no FDv2 synchronizer until the TTL elapses', () async { + final remaining = []; + final fdv2Tier = []; + + final harness = Harness( + initializerFactories: [ + initializerFactory(FDv2SourceResults.terminalError( + message: 'fall back', + fdv1Fallback: true, + fdv1FallbackTtl: const Duration(milliseconds: 20))), + initializerFactory( + changeSet(selector: const Selector(state: 'state-1', version: 1)), + created: remaining), + ], + synchronizerSlots: [synchronizerSlot(fdv2Tier)], + ); + + harness.orchestrator.start(); + await harness.pump(); + + expect(fdv2Tier, isEmpty); + + await harness.pumpUntil(() => fdv2Tier.length == 1); + + expect(remaining, isEmpty, reason: 'initializers run only once'); harness.orchestrator.stop(); }); diff --git a/packages/common_client/test/data_sources/fdv2/protocol_handler_test.dart b/packages/common_client/test/data_sources/fdv2/protocol_handler_test.dart index cf64c491..ac3695f9 100644 --- a/packages/common_client/test/data_sources/fdv2/protocol_handler_test.dart +++ b/packages/common_client/test/data_sources/fdv2/protocol_handler_test.dart @@ -440,7 +440,7 @@ void main() { equals(const Duration(seconds: 60))); }); - test('a zero protocolFallbackTTL is preserved (indefinite fallback)', () { + test('a zero protocolFallbackTTL still directs fallback', () { final handler = makeHandler(); handler.processEvent(serverIntent('xfer-full')); 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 index 12df6005..82a947b4 100644 --- a/packages/common_client/test/data_sources/fdv2/source_manager_test.dart +++ b/packages/common_client/test/data_sources/fdv2/source_manager_test.dart @@ -182,16 +182,43 @@ void main() { reason: 'the FDv2 tier is disabled after fallback'); }); - test('engaging FDv1 fallback without a configured slot does nothing', () { + test('restoring FDv2 re-enables the FDv2 tier and blocks FDv1', () { + final fdv2Created = []; + final fdv1Created = []; + final manager = _manager(slots: [ + _slot(0, fdv2Created), + _slot(1, fdv1Created, isFdv1Fallback: true), + ]); + + manager.nextAvailableSynchronizer(); + manager.engageFdv1Fallback(); + manager.nextAvailableSynchronizer(); + manager.restoreFdv2Synchronizers(); + + expect(manager.availableSynchronizerCount, 1); + expect(manager.recreateCurrentSynchronizer(), isNull, + reason: 'the cursor is reset, so there is no current slot'); + + manager.nextAvailableSynchronizer(); + + expect(fdv2Created, hasLength(2)); + expect(fdv1Created.single.closed, isTrue); + expect(manager.isPrimarySynchronizer, isTrue); + }); + + test('engaging FDv1 fallback without a configured slot pauses FDv2', () { final created = []; final manager = _manager(slots: [_slot(0, created), _slot(1, created)]); expect(manager.hasFdv1FallbackConfigured, isFalse); manager.engageFdv1Fallback(); - expect(manager.availableSynchronizerCount, 2, - reason: 'blocking every slot without unblocking a fallback would ' - 'leave nothing to activate'); + expect(manager.availableSynchronizerCount, 0); + expect(manager.nextAvailableSynchronizer(), isNull); + + manager.restoreFdv2Synchronizers(); + + expect(manager.availableSynchronizerCount, 2); expect(manager.nextAvailableSynchronizer(), isNotNull); }); diff --git a/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart b/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart index 55b5b996..9c5e3b25 100644 --- a/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart +++ b/packages/common_client/test/data_sources/fdv2/streaming_base_test.dart @@ -583,7 +583,8 @@ void main() { final status = emissions.single as StatusResult; expect(status.state, SourceState.terminalError); expect(status.fdv1Fallback, isTrue); - // A zero TTL means remain on the fallback indefinitely. + // A zero TTL still directs fallback; the orchestrator replaces it with + // the default. expect(status.fdv1FallbackTtl, Duration.zero); expect(sse.isClosed, isTrue); });