diff --git a/packages/shared/common/__tests__/internal/fdv2/fallbackDirective.test.ts b/packages/shared/common/__tests__/internal/fdv2/fallbackDirective.test.ts index d90d4128be..722fc32f26 100644 --- a/packages/shared/common/__tests__/internal/fdv2/fallbackDirective.test.ts +++ b/packages/shared/common/__tests__/internal/fdv2/fallbackDirective.test.ts @@ -1,6 +1,8 @@ import { + DEFAULT_FDV1_FALLBACK_TTL_MS, readFallbackDirective, readGoodbyeFallbackDirective, + resolveFallbackTtlMs, } from '../../../src/internal/fdv2/fallbackDirective'; function makeHeaders(map: Record): { get(name: string): string | null } { @@ -29,10 +31,11 @@ it('matches "true" case-insensitively', () => { expect(result.fdv1Fallback).toBe(true); }); -it('returns fdv1Fallback true with undefined TTL when x-ld-fd-fallback-ttl is absent', () => { +it('applies the jittered default TTL when x-ld-fd-fallback-ttl is absent', () => { const result = readFallbackDirective(makeHeaders({ 'x-ld-fd-fallback': 'true' })); expect(result.fdv1Fallback).toBe(true); - expect(result.fdv1FallbackTtlMs).toBeUndefined(); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); }); it('converts a TTL of "60" seconds to 60000 ms', () => { @@ -43,28 +46,46 @@ it('converts a TTL of "60" seconds to 60000 ms', () => { expect(result.fdv1FallbackTtlMs).toBe(60000); }); -it('converts TTL "0" to 0 ms (indefinite fallback)', () => { +it('applies the default TTL for a TTL of "0"', () => { const result = readFallbackDirective( makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '0' }), ); expect(result.fdv1Fallback).toBe(true); - expect(result.fdv1FallbackTtlMs).toBe(0); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); }); -it('returns undefined TTL for a non-numeric x-ld-fd-fallback-ttl value', () => { +it('applies the default TTL for a non-numeric x-ld-fd-fallback-ttl value', () => { const result = readFallbackDirective( makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': 'soon' }), ); expect(result.fdv1Fallback).toBe(true); - expect(result.fdv1FallbackTtlMs).toBeUndefined(); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); }); -it('clamps negative TTL seconds to 0 ms (treated as indefinite)', () => { +it('applies the default TTL for a negative TTL', () => { const result = readFallbackDirective( makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '-5' }), ); expect(result.fdv1Fallback).toBe(true); - expect(result.fdv1FallbackTtlMs).toBe(0); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('accepts a TTL of exactly one hour', () => { + const result = readFallbackDirective( + makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '3600' }), + ); + expect(result.fdv1FallbackTtlMs).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('applies the default TTL for a TTL longer than one hour', () => { + const result = readFallbackDirective( + makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '7200' }), + ); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); }); it('header lookup is case-insensitive', () => { @@ -101,16 +122,45 @@ it('readGoodbyeFallbackDirective: converts a protocolFallbackTTL of 60 seconds t expect(result.fdv1FallbackTtlMs).toBe(60000); }); -it('readGoodbyeFallbackDirective: converts protocolFallbackTTL 0 to 0 ms (indefinite fallback)', () => { +it('readGoodbyeFallbackDirective: applies the default TTL for a protocolFallbackTTL of 0', () => { const result = readGoodbyeFallbackDirective({ reason: 'falling back', protocolFallbackTTL: 0 }); expect(result.fdv1Fallback).toBe(true); - expect(result.fdv1FallbackTtlMs).toBe(0); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); }); -it('readGoodbyeFallbackDirective: clamps negative protocolFallbackTTL to 0 ms', () => { +it('readGoodbyeFallbackDirective: applies the default TTL for a negative protocolFallbackTTL', () => { const result = readGoodbyeFallbackDirective({ reason: 'falling back', protocolFallbackTTL: -5 }); expect(result.fdv1Fallback).toBe(true); - expect(result.fdv1FallbackTtlMs).toBe(0); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('readGoodbyeFallbackDirective: accepts a protocolFallbackTTL of exactly one hour', () => { + const result = readGoodbyeFallbackDirective({ + reason: 'falling back', + protocolFallbackTTL: 3600, + }); + expect(result.fdv1FallbackTtlMs).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('readGoodbyeFallbackDirective: applies the default TTL for a protocolFallbackTTL over one hour', () => { + const result = readGoodbyeFallbackDirective({ + reason: 'falling back', + protocolFallbackTTL: 7200, + }); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('readGoodbyeFallbackDirective: truncates a fractional protocolFallbackTTL to whole seconds before the range check', () => { + const result = readGoodbyeFallbackDirective({ + reason: 'falling back', + protocolFallbackTTL: 0.001, + }); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); }); it('readGoodbyeFallbackDirective: returns fdv1Fallback false for a non-numeric protocolFallbackTTL', () => { @@ -130,3 +180,44 @@ it('readGoodbyeFallbackDirective: returns fdv1Fallback false for a non-finite pr expect(result.fdv1Fallback).toBe(false); expect(result.fdv1FallbackTtlMs).toBeUndefined(); }); + +it('resolveFallbackTtlMs: converts a whole number of seconds to milliseconds without jitter', () => { + expect(resolveFallbackTtlMs(60, () => 1)).toBe(60000); +}); + +it('resolveFallbackTtlMs: accepts a TTL of exactly one hour unchanged', () => { + expect(resolveFallbackTtlMs(3600, () => 1)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('resolveFallbackTtlMs: uses the default for a TTL greater than one hour', () => { + expect(resolveFallbackTtlMs(3601, () => 0)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('resolveFallbackTtlMs: uses the default for a TTL of zero', () => { + expect(resolveFallbackTtlMs(0, () => 0)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('resolveFallbackTtlMs: uses the default for a negative TTL', () => { + expect(resolveFallbackTtlMs(-5, () => 0)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('resolveFallbackTtlMs: uses the default for an absent TTL', () => { + expect(resolveFallbackTtlMs(undefined, () => 0)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('resolveFallbackTtlMs: uses the default for a TTL that is not a number', () => { + expect(resolveFallbackTtlMs(NaN, () => 0)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS); +}); + +it('resolveFallbackTtlMs: subtracts jitter of up to half the default TTL', () => { + expect(resolveFallbackTtlMs(undefined, () => 0.5)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS * 0.75); + expect(resolveFallbackTtlMs(undefined, () => 1)).toBe(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); +}); + +it('resolveFallbackTtlMs: defaults to Math.random for jitter and stays within bounds', () => { + for (let i = 0; i < 50; i += 1) { + const ttl = resolveFallbackTtlMs(undefined); + expect(ttl).toBeGreaterThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(ttl).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); + } +}); diff --git a/packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts b/packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts index 085b3afe90..072c398109 100644 --- a/packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts +++ b/packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts @@ -595,9 +595,11 @@ it('falls back to FDv1 from initializer phase, skipping remaining initializers a }); // When the FDv1 Fallback Directive arrives but the SDK has not been configured with an -// FDv1 fallback synchronizer, the data source must transition to a terminal Closed state -// instead of looping or trying other FDv2 sources. -it('terminates when FDv1 fallback is requested but no FDv1 fallback synchronizer is configured', async () => { +// FDv1 fallback synchronizer, the data source must wait out the fallback TTL and then +// retry FDv2 rather than terminating. +it('retries FDv2 after the fallback TTL when no FDv1 fallback synchronizer is configured', async () => { + const fdv2Data = { key: 'fdv2-data' }; + let fdv2StartCount = 0; const mockSynchronizer1 = { start: jest .fn() @@ -606,15 +608,22 @@ it('terminates when FDv1 fallback is requested but no FDv1 fallback synchronizer _dataCallback: (basis: boolean, data: any) => void, _statusCallback: (status: DataSourceState, err?: any) => void, ) => { + fdv2StartCount += 1; _statusCallback(DataSourceState.Initializing); - _statusCallback( - DataSourceState.Closed, - new LDFlagDeliveryFallbackError( - DataSourceErrorKind.ErrorResponse, - `Response header indicates to fallback to FDv1`, - 500, - ), - ); + if (fdv2StartCount === 1) { + _statusCallback( + DataSourceState.Closed, + new LDFlagDeliveryFallbackError( + DataSourceErrorKind.ErrorResponse, + `Response header indicates to fallback to FDv1`, + 500, + 1, + ), + ); + } else { + _statusCallback(DataSourceState.Valid); + _dataCallback(true, fdv2Data); + } }, ), stop: jest.fn(), @@ -632,22 +641,84 @@ it('terminates when FDv1 fallback is requested but no FDv1 fallback synchronizer const statusCallback = jest.fn(); const dataCallback = jest.fn(); await new Promise((resolve) => { - statusCallback.mockImplementation((state: DataSourceState) => { - if (state === DataSourceState.Closed) { + dataCallback.mockImplementation((_: boolean, data: any) => { + if (data === fdv2Data) { resolve(); } }); underTest.start(dataCallback, statusCallback); }); - // The underlying synchronizer was started exactly once -- we did not loop trying other - // FDv2 sources after the directive arrived. - expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1); - expect(dataCallback).not.toHaveBeenCalled(); + expect(mockSynchronizer1.start).toHaveBeenCalledTimes(2); + expect(statusCallback).not.toHaveBeenCalledWith(DataSourceState.Closed, expect.anything()); - // The terminal status is Closed (composite signals exhaustion when the FDv1 list is empty). - const lastCall = statusCallback.mock.calls[statusCallback.mock.calls.length - 1]; - expect(lastCall[0]).toBe(DataSourceState.Closed); + await underTest.stop(); +}); + +it('stops the FDv1 synchronizer before restoring FDv2 synchronizers after the fallback TTL', async () => { + const events: string[] = []; + const fdv2Data = { key: 'fdv2-data' }; + let fdv2StartCount = 0; + const mockFDv2Sync = { + start: jest + .fn() + .mockImplementation( + ( + _dataCallback: (basis: boolean, data: any) => void, + _statusCallback: (status: DataSourceState, err?: any) => void, + ) => { + fdv2StartCount += 1; + events.push('fdv2 start'); + _statusCallback(DataSourceState.Initializing); + _statusCallback(DataSourceState.Valid); + if (fdv2StartCount === 1) { + _dataCallback(true, { fallbackToFDv1: true, fdv1FallbackTtlMs: 1 }); + } else { + _dataCallback(true, fdv2Data); + } + }, + ), + stop: jest.fn(() => events.push('fdv2 stop')), + }; + + const mockFDv1Synchronizer = { + start: jest + .fn() + .mockImplementation( + ( + _dataCallback: (basis: boolean, data: any) => void, + _statusCallback: (status: DataSourceState, err?: any) => void, + ) => { + events.push('fdv1 start'); + _statusCallback(DataSourceState.Initializing); + _statusCallback(DataSourceState.Valid); + _dataCallback(true, { key: 'fdv1-data' }); + }, + ), + stop: jest.fn(() => events.push('fdv1 stop')), + }; + + const underTest = new CompositeDataSource( + [], + [makeDataSourceFactory(mockFDv2Sync)], + [makeDataSourceFactory(mockFDv1Synchronizer)], + undefined, + makeTestTransitionConditions(), + makeZeroBackoff(), + ); + + await new Promise((resolve) => { + const dataCallback = jest.fn((_: boolean, data: any) => { + if (data === fdv2Data) { + resolve(); + } + }); + underTest.start(dataCallback, jest.fn()); + }); + + expect(events).toEqual(['fdv2 start', 'fdv2 stop', 'fdv1 start', 'fdv1 stop', 'fdv2 start']); + + await underTest.stop(); }); // When an initializer delivers basis with `data.fallbackToFDv1`, the composite must apply @@ -903,7 +974,10 @@ it('throttles FDv1 retries with backoff after the initial engagement', async () // Two FDv1 factories so the composite cycles through 'fallback' transitions instead // of stopping after the first cull -- each cycle is a chance to (incorrectly) skip // backoff. The tracking Backoff below counts fail() calls as the observable proxy. - const makeFDv1Sync = () => ({ + // The first FDv1 factory fails, so the composite retries with the second -- that retry + // is a chance to (incorrectly) skip backoff. The tracking Backoff below counts fail() + // calls as the observable proxy. + const mockFDv1Sync1 = { start: jest .fn() .mockImplementation( @@ -923,9 +997,22 @@ it('throttles FDv1 retries with backoff after the initial engagement', async () }, ), stop: jest.fn(), - }); - const mockFDv1Sync1 = makeFDv1Sync(); - const mockFDv1Sync2 = makeFDv1Sync(); + }; + + const mockFDv1Sync2 = { + start: jest + .fn() + .mockImplementation( + ( + _dataCallback: (basis: boolean, data: any) => void, + _statusCallback: (status: DataSourceState, err?: any) => void, + ) => { + _statusCallback(DataSourceState.Initializing); + _statusCallback(DataSourceState.Valid); + }, + ), + stop: jest.fn(), + }; let failCalls = 0; let successCalls = 0; @@ -952,7 +1039,7 @@ it('throttles FDv1 retries with backoff after the initial engagement', async () const statusCallback = jest.fn(); await new Promise((resolve) => { statusCallback.mockImplementation((state: DataSourceState) => { - if (state === DataSourceState.Closed) { + if (state === DataSourceState.Valid) { resolve(); } }); @@ -963,8 +1050,10 @@ it('throttles FDv1 retries with backoff after the initial engagement', async () expect(mockFDv1Sync1.start).toHaveBeenCalledTimes(1); expect(mockFDv1Sync2.start).toHaveBeenCalledTimes(1); // One fail() per post-engagement retry; the initial engagement is not counted. - expect(failCalls).toBe(2); + expect(failCalls).toBe(1); expect(successCalls).toBe(0); + + await underTest.stop(); }); it('reports error when all initializers fail', async () => { diff --git a/packages/shared/common/src/datasource/CompositeDataSource.ts b/packages/shared/common/src/datasource/CompositeDataSource.ts index a6f0f5f3d7..6292effc04 100644 --- a/packages/shared/common/src/datasource/CompositeDataSource.ts +++ b/packages/shared/common/src/datasource/CompositeDataSource.ts @@ -5,6 +5,7 @@ import { DataSourceState, LDDataSourceFactory, } from '../api/subsystem/DataSystem/DataSource'; +import { resolveFallbackTtlMs } from '../internal/fdv2/fallbackDirective'; import { Backoff, DefaultBackoff } from './Backoff'; import { DataSourceList } from './dataSourceList'; import { LDFlagDeliveryFallbackError } from './errors'; @@ -27,6 +28,8 @@ export type TransitionConditions = { interface TransitionRequest { transition: Transition; err?: Error; + // Set when the FDv1 fallback TTL elapsed and the FDv2 synchronizers should be restored. + recoverFDv2?: boolean; } /** @@ -39,12 +42,17 @@ export class CompositeDataSource implements DataSource { private _initPhaseActive: boolean; private _initFactories: DataSourceList; private _syncFactories: DataSourceList; + // The FDv2 synchronizers, retained while FDv1 fallback is engaged so they can be restored. + private _fdv2SyncFactories: DataSourceList; private _fdv1Synchronizers: DataSourceList; // Set after the server-directed FDv1 fallback has been engaged. The directive is // one-way: subsequent fallback signals from the FDv1 synchronizer (e.g. if the FDv1 // endpoint were to echo `x-ld-fd-fallback`) are ignored so the data source does not // loop replacing its synchronizer list with itself. private _fdv1FallbackEngaged: boolean = false; + // Resolves when the FDv1 fallback TTL elapses. Fallback is never permanent: once this + // resolves, the FDv2 synchronizers are restored whether or not an FDv1 synchronizer ran. + private _fdv2Recovery?: Promise; private _stopped: boolean = true; private _externalTransitionPromise: Promise; @@ -82,6 +90,7 @@ export class CompositeDataSource implements DataSource { this._initPhaseActive = initializers.length > 0; // init phase if we have initializers this._initFactories = new DataSourceList(false, initializers); this._syncFactories = new DataSourceList(true, synchronizers); + this._fdv2SyncFactories = this._syncFactories; this._fdv1Synchronizers = new DataSourceList(true, fdv1Synchronizers); } @@ -114,6 +123,10 @@ export class CompositeDataSource implements DataSource { cullDSFactory, } = this._pickDataSource(lastTransition); + // Detaches the current data source's callbacks and scheduled transitions so that it + // cannot affect the composite after the loop moves on from it. + let detachCurrentDS = () => {}; + const internalTransitionPromise = new Promise((transitionResolve) => { if (currentDS) { // these local variables are used for handling automatic transition related to data source status (ex: recovering to primary after @@ -122,37 +135,44 @@ export class CompositeDataSource implements DataSource { let cancelScheduledTransition: () => void = () => {}; // Engages the FDv1 Fallback Directive: swaps the synchronizer list to FDv1 (or - // empties it when no FDv1 fallback is configured), reports the in-progress status - // through the sanitizer, cancels any pending scheduled transition, and resolves - // a switchToSync transition. The directive is one-way; the engagement flag is - // checked by the callers so we don't re-enter this branch on repeated signals. - const engageFDv1Fallback = (statusToReport: DataSourceState, err: unknown) => { + // empties it when no FDv1 fallback is configured), schedules the return to FDv2 + // after the fallback TTL, reports the in-progress status through the sanitizer, + // cancels any pending scheduled transition, and resolves a switchToSync + // transition. The engagement flag is checked by the callers so we don't re-enter + // this branch on repeated signals before the TTL elapses. + const engageFDv1Fallback = ( + statusToReport: DataSourceState, + err: unknown, + fdv1FallbackTtlMs: number | undefined, + ) => { + const ttlMs = fdv1FallbackTtlMs ?? resolveFallbackTtlMs(undefined); if (this._fdv1Synchronizers.length() > 0) { - this._logger?.warn(`Falling back to FDv1`); + this._logger?.warn(`Falling back to FDv1; will retry FDv2 in ${ttlMs}ms`); } else { this._logger?.warn( - `FDv1 fallback was requested but no FDv1 fallback synchronizer is configured; data source will terminate`, + `FDv1 fallback was requested but no FDv1 fallback synchronizer is configured; will retry FDv2 in ${ttlMs}ms`, ); } this._fdv1FallbackEngaged = true; this._syncFactories = this._fdv1Synchronizers; + this._scheduleFDv2Recovery(ttlMs); sanitizedStatusCallback(statusToReport, err); this._consumeCancelToken(cancelScheduledTransition); transitionResolve({ transition: 'switchToSync', err: err as Error | undefined }); }; // this callback handler can be disabled and ensures only one transition request occurs - const callbackHandler = new CallbackHandler( + const callbackHandler: CallbackHandler = new CallbackHandler( (basis: boolean, data: any) => { this._backoff.success(); dataCallback(basis, data); // The FDv1 fallback directive can ride along on the same data callback that // delivers a payload, so the swap to FDv1 is atomic with payload application. - // Once engaged, the directive is one-way -- subsequent signals are ignored. + // Once engaged, subsequent signals are ignored until the TTL elapses. if (data?.fallbackToFDv1 && !this._fdv1FallbackEngaged) { callbackHandler.disable(); - engageFDv1Fallback(DataSourceState.Interrupted, undefined); + engageFDv1Fallback(DataSourceState.Interrupted, undefined, data.fdv1FallbackTtlMs); return; } @@ -173,20 +193,22 @@ export class CompositeDataSource implements DataSource { ); if (err || state === DataSourceState.Closed) { callbackHandler.disable(); - if (err?.recoverable === false) { - // don't use this datasource's factory again - this._logger?.debug(`Culling data source due to err ${err}`); - cullDSFactory?.(); - } // Status-callback path for the FDv1 Fallback Directive (used when the // server-directed signal arrives without an accompanying payload, e.g. on - // an HTTP error or a malformed body). + // an HTTP error or a malformed body). The FDv2 source is not culled, so it + // can be restored once the fallback TTL elapses. if (err instanceof LDFlagDeliveryFallbackError && !this._fdv1FallbackEngaged) { - engageFDv1Fallback(state, err); + engageFDv1Fallback(state, err, err.fdv1FallbackTtlMs); return; } + if (err?.recoverable === false) { + // don't use this datasource's factory again + this._logger?.debug(`Culling data source due to err ${err}`); + cullDSFactory?.(); + } + sanitizedStatusCallback(state, err); this._consumeCancelToken(cancelScheduledTransition); transitionResolve({ transition: 'fallback', err }); // unrecoverable error has occurred, so fallback @@ -215,11 +237,18 @@ export class CompositeDataSource implements DataSource { } }, ); + detachCurrentDS = () => { + callbackHandler.disable(); + this._consumeCancelToken(cancelScheduledTransition); + }; currentDS.start( (basis, data) => callbackHandler.dataHandler(basis, data), (status, err) => callbackHandler.statusHandler(status, err), selectorGetter, ); + } else if (this._fdv2Recovery) { + // FDv1 fallback is engaged but no FDv1 synchronizer is available. Wait for the + // fallback TTL to elapse and the FDv2 synchronizers to be restored. } else { // we don't have a data source to use! transitionResolve({ @@ -232,14 +261,18 @@ export class CompositeDataSource implements DataSource { } }); - // await transition triggered by internal data source or an external stop request + // await transition triggered by internal data source, the FDv1 fallback TTL elapsing, + // or an external stop request // eslint-disable-next-line no-await-in-loop let transitionRequest = await Promise.race([ + ...(this._fdv2Recovery ? [this._fdv2Recovery] : []), internalTransitionPromise, this._externalTransitionPromise, ]); - // stop the underlying datasource before transitioning to next state + // stop the underlying datasource before transitioning to next state, so that FDv1 + // and FDv2 sources never deliver data concurrently + detachCurrentDS(); currentDS?.stop(); // Skip backoff only on the initial server-directed FDv1 engagement (a `switchToSync` @@ -276,6 +309,14 @@ export class CompositeDataSource implements DataSource { this._consumeCancelToken(cancelDelay); } + if (transitionRequest.recoverFDv2) { + this._logger?.info('FDv1 fallback TTL elapsed; retrying FDv2'); + this._fdv2Recovery = undefined; + this._fdv1FallbackEngaged = false; + this._syncFactories = this._fdv2SyncFactories; + sanitizedStatusCallback(DataSourceState.Interrupted); + } + lastTransition = transitionRequest.transition; if (transitionRequest.transition === 'stop') { // exit the loop, this is intentionally not the sanitized status callback @@ -298,15 +339,26 @@ export class CompositeDataSource implements DataSource { this._stopped = true; this._initPhaseActive = this._initFactories.length() > 0; // init phase if we have initializers; this._initFactories.reset(); + this._syncFactories = this._fdv2SyncFactories; this._syncFactories.reset(); this._fdv1Synchronizers.reset(); this._fdv1FallbackEngaged = false; + this._fdv2Recovery = undefined; this._externalTransitionPromise = new Promise((tr) => { this._externalTransitionResolve = tr; }); // intentionally not resetting the backoff to avoid a code path that could circumvent throttling } + private _scheduleFDv2Recovery(delayMS: number) { + const { promise, cancel } = this._cancellableDelay(delayMS); + this._cancelTokens.push(cancel); + this._fdv2Recovery = promise.then((): TransitionRequest => { + this._consumeCancelToken(cancel); + return { transition: 'switchToSync', recoverFDv2: true }; + }); + } + /** * Determines the next datasource and returns that datasource as well as a closure to cull the * datasource from the datasource lists. One example where the cull closure is invoked is if the diff --git a/packages/shared/common/src/datasource/errors.ts b/packages/shared/common/src/datasource/errors.ts index 273dc083b9..bf67f54b29 100644 --- a/packages/shared/common/src/datasource/errors.ts +++ b/packages/shared/common/src/datasource/errors.ts @@ -42,11 +42,19 @@ export class LDFlagDeliveryFallbackError extends Error { public readonly kind: DataSourceErrorKind; public readonly code?: number; public readonly recoverable: boolean; - - constructor(kind: DataSourceErrorKind, message: string, code?: number) { + /** How long to remain on FDv1 before retrying FDv2, if the directive carried a TTL. */ + public readonly fdv1FallbackTtlMs?: number; + + constructor( + kind: DataSourceErrorKind, + message: string, + code?: number, + fdv1FallbackTtlMs?: number, + ) { super(message); this.kind = kind; this.code = code; + this.fdv1FallbackTtlMs = fdv1FallbackTtlMs; this.name = 'LDFlagDeliveryFallbackError'; this.recoverable = false; } diff --git a/packages/shared/common/src/internal/fdv2/fallbackDirective.ts b/packages/shared/common/src/internal/fdv2/fallbackDirective.ts index 1a1cc21f40..dd3e430477 100644 --- a/packages/shared/common/src/internal/fdv2/fallbackDirective.ts +++ b/packages/shared/common/src/internal/fdv2/fallbackDirective.ts @@ -1,15 +1,58 @@ /** - * The FDv1 fallback directive parsed from a connection's response headers. - * Its presence (`fdv1Fallback === true`) means the server asked the SDK to - * fall back to FDv1. + * Default time to remain on FDv1 after a fallback directive that carried no + * usable TTL: 1 hour. This is also the upper bound of the range the server may + * ask for; anything larger is replaced with this default. + */ +export const DEFAULT_FDV1_FALLBACK_TTL_MS = 60 * 60 * 1000; + +/** Jitter is subtracted from the default TTL, up to half of it. */ +const DEFAULT_TTL_JITTER_RATIO = 0.5; + +/** + * Normalizes a fallback TTL expressed in whole seconds into milliseconds. + * + * A TTL is only honored when it falls in the range `(0, 1 hour]`. An absent, + * unparseable, zero, negative, or too-large TTL is replaced with the default + * of 1 hour, minus a jitter value drawn uniformly from `[0, half the default]` + * so that a fleet of SDKs that fell back together does not retry FDv2 in + * lockstep. A TTL supplied by the server is already jittered by the server, so + * it is used exactly as given. Fallback is therefore never indefinite. + * + * @param ttlSeconds The TTL carried by the directive, in seconds, or + * `undefined` when the directive carried none. + * @param random Source of randomness for the jitter. Injectable for tests. + */ +export function resolveFallbackTtlMs( + ttlSeconds: number | undefined, + random: () => number = Math.random, +): number { + const ttlMs = ttlSeconds === undefined ? undefined : ttlSeconds * 1000; + if ( + ttlMs === undefined || + !Number.isFinite(ttlMs) || + ttlMs <= 0 || + ttlMs > DEFAULT_FDV1_FALLBACK_TTL_MS + ) { + return ( + DEFAULT_FDV1_FALLBACK_TTL_MS - + Math.trunc(random() * DEFAULT_TTL_JITTER_RATIO * DEFAULT_FDV1_FALLBACK_TTL_MS) + ); + } + return ttlMs; +} + +/** + * The FDv1 fallback directive parsed from a connection's response headers or + * from a `goodbye` message. Its presence (`fdv1Fallback === true`) means the + * server asked the SDK to fall back to FDv1. * - * `fdv1FallbackTtlMs` is how long to remain on FDv1 before retrying FDv2: - * - `undefined`: the server gave no TTL header (caller uses a 1-hour default). - * - `0`: indefinite fallback (no automatic recovery). - * - `> 0`: milliseconds to wait before attempting FDv2 recovery. + * `fdv1FallbackTtlMs` is how long to remain on FDv1 before retrying FDv2. It is + * always set when `fdv1Fallback` is true: a missing, unparseable, or + * out-of-range TTL is replaced with the jittered default, so fallback is never + * indefinite. * - * This is the single place that interprets `x-ld-fd-fallback` and - * `x-ld-fd-fallback-ttl`. + * This is the single place that interprets `x-ld-fd-fallback`, + * `x-ld-fd-fallback-ttl`, and a goodbye message's `protocolFallbackTTL`. */ export interface FallbackDirective { fdv1Fallback: boolean; @@ -33,19 +76,11 @@ export function readFallbackDirective(headers: { } const raw = headers.get('x-ld-fd-fallback-ttl'); - if (raw === null) { - return { fdv1Fallback: true }; - } - - const seconds = parseInt(raw, 10); - if (Number.isNaN(seconds)) { - return { fdv1Fallback: true }; - } + const seconds = raw === null ? undefined : parseInt(raw, 10); - // Clamp negative values to 0 (treated as indefinite, same as TTL=0). - // Prevents a malicious server from sending a large-negative TTL to trigger - // immediate recovery instead of the intended long wait. - return { fdv1Fallback: true, fdv1FallbackTtlMs: Math.max(0, seconds) * 1000 }; + // A missing, unparseable, or out-of-range TTL becomes the jittered default, + // so the directive always carries a concrete deadline for retrying FDv2. + return { fdv1Fallback: true, fdv1FallbackTtlMs: resolveFallbackTtlMs(seconds) }; } /** @@ -53,11 +88,12 @@ export function readFallbackDirective(headers: { * * SDKs that cannot read streaming response headers (e.g. browsers using the * native `EventSource` API) receive the fallback directive in-band via the - * goodbye message's `protocolFallbackTTL` field. - * Presence of a finite numeric `protocolFallbackTTL` signals FDv1 fallback; - * the value carries the same semantics as the `x-ld-fd-fallback-ttl` header - * (`0` indicates indefinite fallback). A missing, non-numeric, or non-finite - * value is not a fallback signal and yields `{ fdv1Fallback: false }`. + * goodbye message's `protocolFallbackTTL` field. Presence of a finite numeric + * `protocolFallbackTTL` signals FDv1 fallback; the value carries the same + * semantics as the `x-ld-fd-fallback-ttl` header, including the replacement of + * an out-of-range value with the jittered default. A missing, non-numeric, or + * non-finite value is not a fallback signal and yields + * `{ fdv1Fallback: false }`. * * @param data The raw, parsed goodbye event data (typed `unknown` because the * caller has not narrowed it). @@ -69,5 +105,5 @@ export function readGoodbyeFallbackDirective(data: unknown): FallbackDirective { return { fdv1Fallback: false }; } - return { fdv1Fallback: true, fdv1FallbackTtlMs: Math.max(0, rawTtl) * 1000 }; + return { fdv1Fallback: true, fdv1FallbackTtlMs: resolveFallbackTtlMs(Math.trunc(rawTtl)) }; } diff --git a/packages/shared/common/src/internal/fdv2/index.ts b/packages/shared/common/src/internal/fdv2/index.ts index b366d44d9e..9664722288 100644 --- a/packages/shared/common/src/internal/fdv2/index.ts +++ b/packages/shared/common/src/internal/fdv2/index.ts @@ -1,4 +1,9 @@ -import { readFallbackDirective, readGoodbyeFallbackDirective } from './fallbackDirective'; +import { + DEFAULT_FDV1_FALLBACK_TTL_MS, + readFallbackDirective, + readGoodbyeFallbackDirective, + resolveFallbackTtlMs, +} from './fallbackDirective'; import type { FallbackDirective } from './fallbackDirective'; import { fdv1PayloadAdaptor as FDv1PayloadAdaptor } from './FDv1PayloadAdaptor'; import { PayloadProcessor } from './payloadProcessor'; @@ -19,11 +24,13 @@ import type { export { createProtocolHandler, + DEFAULT_FDV1_FALLBACK_TTL_MS, FDv1PayloadAdaptor, PayloadProcessor, PayloadStreamReader, readFallbackDirective, readGoodbyeFallbackDirective, + resolveFallbackTtlMs, }; export type { diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts index 80c5114ed7..0ffd59987b 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts @@ -1,3 +1,5 @@ +import { internal } from '@launchdarkly/js-sdk-common'; + import { createFDv2DataSource } from '../../../src/datasource/fdv2/FDv2DataSource'; import { changeSet, @@ -1246,3 +1248,616 @@ it('does not re-trigger fallback when the fdv1 synchronizer itself yields a fall ds.close(); }); + +// -- fdv2 recovery scheduling -- + +it('logs the scheduled fdv2 retry using the directive TTL', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Payload = makePayload({ state: 'fdv2-selector' }); + const fdv1Payload = makePayload({ state: 'fdv1-selector' }); + + const fdv2Sync = makeMockSynchronizer([ + changeSet(fdv2Payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 90000 }), + ]); + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: () => fdv2Sync }), + createSynchronizerSlot({ create: () => fdv1Sync }, { isFDv1Fallback: true }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await ds.start(); + await statusManager.waitForState('VALID', 2); + + expect(logger.info).toHaveBeenCalledWith('FDv2 retry scheduled in 90s.'); + ds.close(); +}); + +it('schedules the jittered default TTL when the directive carries no TTL', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Payload = makePayload({ state: 'fdv2-selector' }); + const fdv1Payload = makePayload({ state: 'fdv1-selector' }); + + const fdv2Sync = makeMockSynchronizer([changeSet(fdv2Payload, { fdv1Fallback: true })]); + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: () => fdv2Sync }), + createSynchronizerSlot({ create: () => fdv1Sync }, { isFDv1Fallback: true }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await ds.start(); + await statusManager.waitForState('VALID', 2); + + const scheduleLog = logger.info.mock.calls.find( + (call: unknown[]) => + typeof call[0] === 'string' && call[0].startsWith('FDv2 retry scheduled in'), + ); + expect(scheduleLog).toBeDefined(); + const seconds = Number(/in (\d+)s/.exec(scheduleLog![0] as string)![1]); + expect(seconds).toBeGreaterThanOrEqual(internal.DEFAULT_FDV1_FALLBACK_TTL_MS / 2000); + expect(seconds).toBeLessThanOrEqual(internal.DEFAULT_FDV1_FALLBACK_TTL_MS / 1000); + + // Cancels the pending default-length deadline so the suite does not hold a timer. + ds.close(); +}); + +it('reschedules without restarting the fdv1 fallback synchronizer when a directive arrives during fallback', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Payload = makePayload({ state: 'fdv2-selector' }); + const fdv1Payload = makePayload({ state: 'fdv1-selector' }); + + const fdv2Sync = makeMockSynchronizer([ + changeSet(fdv2Payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 1000 }), + ]); + + let fdv1Creations = 0; + const fdv1Factory = () => { + fdv1Creations += 1; + // The fallback synchronizer's own response carries a fresh directive. + return makeMockSynchronizer([ + changeSet(fdv1Payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 5000 }), + ]); + }; + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: () => fdv2Sync }), + createSynchronizerSlot({ create: fdv1Factory }, { isFDv1Fallback: true }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await ds.start(); + await statusManager.waitForState('VALID', 2); + await new Promise((resolve) => { + setTimeout(resolve, 20); + }); + + expect(fdv1Creations).toBe(1); + expect(logger.info).toHaveBeenCalledWith('FDv2 retry scheduled in 1s.'); + expect(logger.info).toHaveBeenCalledWith('FDv2 retry scheduled in 5s.'); + ds.close(); +}); + +it('restarts the primary fdv2 synchronizer once the fallback TTL elapses', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Payload = makePayload({ state: 'fdv2-selector' }); + const fdv1Payload = makePayload({ state: 'fdv1-selector' }); + const recoveredPayload = makePayload({ state: 'recovered-selector' }); + + let fdv2Creations = 0; + const fdv2Factory = () => { + fdv2Creations += 1; + if (fdv2Creations === 1) { + return makeMockSynchronizer([ + changeSet(fdv2Payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 20 }), + ]); + } + return makeMockSynchronizer([changeSet(recoveredPayload, { fdv1Fallback: false })]); + }; + + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: fdv2Factory }), + createSynchronizerSlot({ create: () => fdv1Sync }, { isFDv1Fallback: true }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await ds.start(); + // VALID once for the fdv2 changeSet, once for the fdv1 fallback changeSet, + // and once for the changeSet from the restarted fdv2 synchronizer. + await statusManager.waitForState('VALID', 3); + + expect(fdv2Creations).toBe(2); + expect(dataCallback).toHaveBeenCalledWith(recoveredPayload); + expect(logger.info).toHaveBeenCalledWith('Fallback TTL elapsed, restarting FDv2 data sources.'); + ds.close(); +}); + +it('does not restart the fdv2 synchronizers before the fallback TTL elapses', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Payload = makePayload({ state: 'fdv2-selector' }); + const fdv1Payload = makePayload({ state: 'fdv1-selector' }); + + let fdv2Creations = 0; + const fdv2Factory = () => { + fdv2Creations += 1; + return makeMockSynchronizer([ + changeSet(fdv2Payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 10000 }), + ]); + }; + + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: fdv2Factory }), + createSynchronizerSlot({ create: () => fdv1Sync }, { isFDv1Fallback: true }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await ds.start(); + await statusManager.waitForState('VALID', 2); + await new Promise((resolve) => { + setTimeout(resolve, 30); + }); + + expect(fdv2Creations).toBe(1); + expect(logger.info).not.toHaveBeenCalledWith( + 'Fallback TTL elapsed, restarting FDv2 data sources.', + ); + ds.close(); +}); + +it('restarts the fdv2 synchronizers on TTL elapse when no fdv1 fallback synchronizer is configured', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Payload = makePayload({ state: 'fdv2-selector' }); + const recoveredPayload = makePayload({ state: 'recovered-selector' }); + + let fdv2Creations = 0; + const fdv2Factory = () => { + fdv2Creations += 1; + if (fdv2Creations === 1) { + return makeMockSynchronizer([ + changeSet(fdv2Payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 20 }), + ]); + } + return makeMockSynchronizer([changeSet(recoveredPayload, { fdv1Fallback: false })]); + }; + + const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: fdv2Factory })]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await ds.start(); + await statusManager.waitForState('VALID', 2); + + expect(fdv2Creations).toBe(2); + expect(dataCallback).toHaveBeenCalledWith(recoveredPayload); + ds.close(); +}); + +it('cancels the pending recovery deadline when close is called', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Payload = makePayload({ state: 'fdv2-selector' }); + const fdv1Payload = makePayload({ state: 'fdv1-selector' }); + + const fdv2Sync = makeMockSynchronizer([ + changeSet(fdv2Payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 10000 }), + ]); + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: () => fdv2Sync }), + createSynchronizerSlot({ create: () => fdv1Sync }, { isFDv1Fallback: true }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await ds.start(); + await statusManager.waitForState('VALID', 2); + + // Spy installed immediately before close() so the only clearTimeout call it + // can observe is the one that releases the recovery deadline; the condition + // timers are cancelled later, asynchronously, when the loop unwinds. + const clearTimeoutSpy = jest.spyOn(global, 'clearTimeout'); + ds.close(); + expect(clearTimeoutSpy).toHaveBeenCalled(); + clearTimeoutSpy.mockRestore(); +}); + +it('rejects the current attempt but keeps the recovery deadline armed when the synchronizer loop exits', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2RecoveredPayload = makePayload({ state: 'fdv2-recovered' }); + + let fdv2Creations = 0; + const fdv2Factory = () => { + fdv2Creations += 1; + if (fdv2Creations === 1) { + return makeMockSynchronizer([ + terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 20 }), + ]); + } + return makeMockSynchronizer([changeSet(fdv2RecoveredPayload, { fdv1Fallback: false })]); + }; + // A single slot and no FDv1 fallback slot: the terminal error blocks the + // only slot, so the loop exits with a deadline still armed. + const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: fdv2Factory })]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await expect(ds.start()).rejects.toThrow('All data sources exhausted without receiving data.'); + + // The rejection above did not cancel the deadline -- it fires later and + // restarts the synchronizer via sourceManager.fdv2Recovery(). + await statusManager.waitForState('VALID', 1); + expect(fdv2Creations).toBe(2); + expect(dataCallback).toHaveBeenCalledWith(fdv2RecoveredPayload); + expect(logger.info).toHaveBeenCalledWith('Fallback TTL elapsed, restarting FDv2 data sources.'); + + ds.close(); +}); + +it('cancels the pending recovery deadline when an initializer throws after arming it', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + // First initializer carries a directive with no payload data, so it arms + // the recovery deadline without ever calling dataCallback, then the loop + // moves on to the next initializer. There is no fdv1 fallback slot, so + // handleFdv1Fallback does not break the loop early. + const armingInit = makeMockInitializer( + changeSet(makePayload({ type: 'none' }), { fdv1Fallback: true, fdv1FallbackTtlMs: 10000 }), + ); + + // Second initializer throws before runSynchronizers ever starts, exercising + // the uncaught-exception-during-initialization path. + const throwingInit: Initializer = { + run: () => Promise.reject(new Error('initializer failure')), + close: jest.fn(), + }; + + const ds = createFDv2DataSource({ + initializerFactories: [makeInitFactory(armingInit), makeInitFactory(throwingInit)], + synchronizerSlots: [], + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + const clearTimeoutSpy = jest.spyOn(global, 'clearTimeout'); + + await expect(ds.start()).rejects.toThrow('initializer failure'); + + expect(clearTimeoutSpy).toHaveBeenCalled(); + clearTimeoutSpy.mockRestore(); + ds.close(); +}); + +// -- background recovery continuation -- + +it('keeps recovering after an already-elapsed deadline is discovered at exhaustion', async () => { + // The exhaustion branch must handle a deadline that is ALREADY resolved + // when it checks recoveryTimer.promise -- rather than merely arming a + // short TTL and hoping the check lands late. + // + // Two things have to line up for that precondition to occur: + // 1. The deadline has to elapse while nothing is actively racing it. The + // main synchronizer loop always re-reads recoveryTimer.promise fresh on + // each iteration and races it live, so an elapsed deadline is normally + // caught there, not by the exhaustion branch. The one place that never + // races the deadline is initializer processing, so a short TTL armed by + // the first initializer, followed by a second initializer that takes + // real wall-clock time (via an actual setTimeout, comfortably longer + // than the TTL) to resolve, elapses the deadline "off to the side." + // 2. Once synchronizers start, the deadline's own promise and the mock + // synchronizer's (already-resolved) result promise are both settled by + // the time Promise.race is called. Promise.race resolves in favor of + // whichever racer's handler was registered first for already-settled + // inputs, and the code always lists the sync result first -- so the + // terminal error "wins," blocks the only slot, and only THEN does the + // outer loop's exhaustion check discover the stale, already-elapsed + // deadline promise. + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const secondRecoveryPayload = makePayload({ state: 'second-recovery' }); + + // Arms a 5ms recovery deadline, without itself carrying data or blocking + // the initializer chain. + const armingInit = makeMockInitializer( + changeSet(makePayload({ type: 'none', state: '' }), { + fdv1Fallback: true, + fdv1FallbackTtlMs: 5, + }), + ); + // Resolves only after a real 20ms delay -- comfortably longer than the 5ms + // deadline above, so that deadline has genuinely elapsed in wall-clock time + // by the time this initializer (and thus initialization as a whole) + // finishes, well before any synchronizer has started racing it. + const delayingInit: Initializer = { + run: () => + new Promise((resolve) => { + setTimeout(() => { + resolve(changeSet(makePayload({ type: 'none', state: '' }), { fdv1Fallback: false })); + }, 20); + }), + close: jest.fn(), + }; + + let fdv2Creations = 0; + const fdv2Factory = () => { + fdv2Creations += 1; + if (fdv2Creations === 1) { + // Does not carry a directive: the stale deadline armed by armingInit + // is left untouched, so it is what the exhaustion branch discovers. + return makeMockSynchronizer([terminalError(makeErrorInfo(), { fdv1Fallback: false })]); + } + if (fdv2Creations === 2) { + // Recovers, but immediately falls back again with a second short TTL -- + // this forces a second exhaustion-with-pending-recovery pass, proving + // recovery still works after the first cycle. + return makeMockSynchronizer([ + terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 5 }), + ]); + } + return makeMockSynchronizer([changeSet(secondRecoveryPayload, { fdv1Fallback: false })]); + }; + const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: fdv2Factory })]; + + const ds = createFDv2DataSource({ + initializerFactories: [makeInitFactory(armingInit), makeInitFactory(delayingInit)], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await expect(ds.start()).rejects.toThrow('All data sources exhausted without receiving data.'); + + await statusManager.waitForState('VALID', 1); + expect(fdv2Creations).toBe(3); + expect(dataCallback).toHaveBeenCalledWith(secondRecoveryPayload); + + ds.close(); +}); + +it('does not act on a background recovery continuation if close is called first', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const fdv2Sync = makeMockSynchronizer([ + terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 30 }), + ]); + const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => fdv2Sync })]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await expect(ds.start()).rejects.toThrow('All data sources exhausted without receiving data.'); + ds.close(); + + await new Promise((resolve) => { + setTimeout(resolve, 60); + }); + + expect(logger.info).not.toHaveBeenCalledWith( + 'Fallback TTL elapsed, restarting FDv2 data sources.', + ); +}); + +it('does not arm a background recovery continuation with zero synchronizer slots', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const initializer = makeMockInitializer( + changeSet(makePayload({ type: 'none', state: '' }), { + fdv1Fallback: true, + fdv1FallbackTtlMs: 30, + }), + ); + + const ds = createFDv2DataSource({ + initializerFactories: [makeInitFactory(initializer)], + synchronizerSlots: [], + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + const clearTimeoutSpy = jest.spyOn(global, 'clearTimeout'); + await expect(ds.start()).rejects.toThrow('All data sources exhausted without receiving data.'); + expect(clearTimeoutSpy).toHaveBeenCalled(); + clearTimeoutSpy.mockRestore(); + + await new Promise((resolve) => { + setTimeout(resolve, 60); + }); + + expect(logger.info).not.toHaveBeenCalledWith( + 'Fallback TTL elapsed, restarting FDv2 data sources.', + ); + ds.close(); +}); + +it('closes a deadline armed and abandoned by a later recovery generation', async () => { + // A background recovery continuation can hand off to a recursive call that + // itself arms a fresh deadline (a new directive) and then returns normally + // -- via a shutdown result -- without going through the exhaustion branch + // that would otherwise arm its own continuation. Whichever generation + // returns normally without handing off is responsible for closing the + // timer, so the fresh deadline must not survive as a dangling timer. + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + let fdv2Creations = 0; + const fdv2Factory = () => { + fdv2Creations += 1; + if (fdv2Creations === 1) { + // First generation: blocks the primary slot and arms a short deadline, + // switching to the fallback slot. + return makeMockSynchronizer([ + terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 5 }), + ]); + } + // Second generation (after the continuation recovers): blocks the + // primary slot again and arms a fresh deadline -- long enough that it + // could not have elapsed naturally by the time the assertion below + // runs -- before switching back to the fallback slot. + return makeMockSynchronizer([ + terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 500 }), + ]); + }; + + let fallbackCreations = 0; + const fallbackFactory = () => { + fallbackCreations += 1; + if (fallbackCreations === 1) { + // First generation: blocks the fallback slot too, without carrying a + // directive, so all slots end up blocked while the short deadline is + // still pending -- this is what arms the continuation. + return makeMockSynchronizer([terminalError(makeErrorInfo(), { fdv1Fallback: false })]); + } + // Second generation: returns normally via shutdown, without the + // exhaustion branch ever running, even though a fresh deadline is armed. + return makeMockSynchronizer([shutdown()]); + }; + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: fdv2Factory }), + createSynchronizerSlot({ create: fallbackFactory }, { isFDv1Fallback: true }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + }); + + await expect(ds.start()).rejects.toThrow('All data sources exhausted without receiving data.'); + + const clearTimeoutSpy = jest.spyOn(global, 'clearTimeout'); + + // Long enough for the short (5ms) deadline to elapse and for the second + // generation to run to completion (all synchronous/microtask work), but + // far short of the fresh 500ms deadline armed during that generation -- + // so any clearTimeout call observed here can only be the orchestrator + // proactively closing that still-live deadline, not it elapsing on its + // own or ds.close() below cleaning it up after the fact. + await new Promise((resolve) => { + setTimeout(resolve, 30); + }); + + expect(fdv2Creations).toBe(2); + expect(fallbackCreations).toBe(2); + // Exactly one call: the abandoned 500ms deadline being proactively closed. + // Generation 1's own (already-fired) timer never calls clearTimeout, since + // its handle is already cleared by the time anything observes it elapsed. + expect(clearTimeoutSpy).toHaveBeenCalledTimes(1); + clearTimeoutSpy.mockRestore(); + + ds.close(); +}); diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2RecoveryTimer.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2RecoveryTimer.test.ts new file mode 100644 index 0000000000..11b48cac33 --- /dev/null +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2RecoveryTimer.test.ts @@ -0,0 +1,110 @@ +import { createFDv2RecoveryTimer } from '../../../src/datasource/fdv2/FDv2RecoveryTimer'; + +function wait(ms: number): Promise { + return new Promise((resolve) => { + setTimeout(resolve, ms); + }); +} + +it('has no pending promise before anything is scheduled', () => { + const timer = createFDv2RecoveryTimer(); + expect(timer.promise).toBeUndefined(); + timer.close(); +}); + +it('resolves the pending promise once the scheduled ttl elapses', async () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(10); + const pending = timer.promise; + expect(pending).toBeDefined(); + await expect(pending).resolves.toBeUndefined(); + timer.close(); +}); + +it('keeps the resolved promise available until it is cleared', async () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(1); + await timer.promise; + + // A caller that was not waiting when the timer fired can still observe it. + expect(timer.promise).toBeDefined(); + await timer.promise; + + timer.clear(); + expect(timer.promise).toBeUndefined(); +}); + +it('remains usable for a new schedule after clear', async () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(1); + await timer.promise; + timer.clear(); + + timer.schedule(5); + await expect(timer.promise).resolves.toBeUndefined(); + timer.close(); +}); + +it('replaces a pending schedule with the newest ttl', async () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(60000); + const superseded = timer.promise; + timer.schedule(5); + expect(timer.promise).not.toBe(superseded); + await expect(timer.promise).resolves.toBeUndefined(); + timer.close(); +}); + +it('cancels a pending timer on clear', async () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(5); + const captured = timer.promise; + timer.clear(); + expect(timer.promise).toBeUndefined(); + + const outcome = await Promise.race([ + captured!.then(() => 'fired' as const), + wait(30).then(() => 'timeout' as const), + ]); + expect(outcome).toBe('timeout'); + timer.close(); +}); + +it('cancels a pending timer on close', async () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(5); + const captured = timer.promise; + timer.close(); + expect(timer.promise).toBeUndefined(); + + const outcome = await Promise.race([ + captured!.then(() => 'fired' as const), + wait(30).then(() => 'timeout' as const), + ]); + expect(outcome).toBe('timeout'); +}); + +it('ignores schedule after close', () => { + const timer = createFDv2RecoveryTimer(); + timer.close(); + timer.schedule(5); + expect(timer.promise).toBeUndefined(); +}); + +it('clears the underlying timeout on close', () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(60000); + // @ts-ignore + const clearTimeoutSpy = jest.spyOn(global, 'clearTimeout'); + timer.close(); + expect(clearTimeoutSpy).toHaveBeenCalled(); + clearTimeoutSpy.mockRestore(); +}); + +it('close is idempotent', () => { + const timer = createFDv2RecoveryTimer(); + timer.schedule(60000); + timer.close(); + timer.close(); + expect(timer.promise).toBeUndefined(); +}); diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2SourceResult.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2SourceResult.test.ts index 0ec1648fe5..32f2199fe7 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2SourceResult.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2SourceResult.test.ts @@ -107,7 +107,7 @@ it('creates a goodbye status result with fdv1Fallback and a TTL', () => { }); }); -it('creates a goodbye status result with TTL 0 (indefinite fallback)', () => { +it('creates a goodbye status result carrying the TTL it was handed', () => { const result = goodbye('server-shutdown', { fdv1Fallback: true, fdv1FallbackTtlMs: 0 }); expect(result).toEqual({ diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts index d5ba215317..23bac53e14 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts @@ -1,4 +1,4 @@ -import { DataSourceErrorKind } from '@launchdarkly/js-sdk-common'; +import { DataSourceErrorKind, internal } from '@launchdarkly/js-sdk-common'; import { poll } from '../../../src/datasource/fdv2/PollingBase'; import { @@ -500,7 +500,7 @@ it('reads TTL in seconds and converts to ms on a changeSet', async () => { expect((result as any).fdv1FallbackTtlMs).toBe(60000); }); -it('treats TTL "0" as indefinite (0 ms)', async () => { +it('applies the default TTL for a TTL of "0"', async () => { const body = makeFullPayloadBody({ flagA: { value: true } }); const requestor = makeRequestor({ status: 200, @@ -511,10 +511,15 @@ it('treats TTL "0" as indefinite (0 ms)', async () => { const result = await poll(requestor, undefined, logger); expect(result.fdv1Fallback).toBe(true); - expect((result as any).fdv1FallbackTtlMs).toBe(0); + expect((result as any).fdv1FallbackTtlMs).toBeGreaterThanOrEqual( + internal.DEFAULT_FDV1_FALLBACK_TTL_MS / 2, + ); + expect((result as any).fdv1FallbackTtlMs).toBeLessThanOrEqual( + internal.DEFAULT_FDV1_FALLBACK_TTL_MS, + ); }); -it('leaves TTL undefined when the ttl header is absent but fallback is true', async () => { +it('applies the default TTL when the ttl header is absent but fallback is true', async () => { const body = makeFullPayloadBody({ flagA: { value: true } }); const requestor = makeRequestor({ status: 200, @@ -525,7 +530,12 @@ it('leaves TTL undefined when the ttl header is absent but fallback is true', as const result = await poll(requestor, undefined, logger); expect(result.fdv1Fallback).toBe(true); - expect((result as any).fdv1FallbackTtlMs).toBeUndefined(); + expect((result as any).fdv1FallbackTtlMs).toBeGreaterThanOrEqual( + internal.DEFAULT_FDV1_FALLBACK_TTL_MS / 2, + ); + expect((result as any).fdv1FallbackTtlMs).toBeLessThanOrEqual( + internal.DEFAULT_FDV1_FALLBACK_TTL_MS, + ); }); it('stamps TTL on a non-success error response', async () => { diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts index 892c94aa7f..a15b89e04a 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts @@ -1,4 +1,4 @@ -import { sleep } from '@launchdarkly/js-sdk-common'; +import { internal, sleep } from '@launchdarkly/js-sdk-common'; import { FDv2PollResponse, FDv2Requestor } from '../../../src/datasource/fdv2/FDv2Requestor'; import { createPollingInitializer } from '../../../src/datasource/fdv2/PollingInitializer'; @@ -190,7 +190,10 @@ it('returns a terminal error immediately when a fallback directive accompanies a if (result.type === 'status') { expect(result.state).toBe('terminal_error'); expect(result.fdv1Fallback).toBe(true); - expect(result.fdv1FallbackTtlMs).toBeUndefined(); + expect(result.fdv1FallbackTtlMs).toBeGreaterThanOrEqual( + internal.DEFAULT_FDV1_FALLBACK_TTL_MS / 2, + ); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(internal.DEFAULT_FDV1_FALLBACK_TTL_MS); } expect(requestor.poll).toHaveBeenCalledTimes(1); expect(sleep).not.toHaveBeenCalled(); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts index 16662adb83..4f840565fb 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts @@ -8,6 +8,7 @@ import { DEFAULT_RECOVERY_TIMEOUT_MS, getConditions, } from './Conditions'; +import { createFDv2RecoveryTimer } from './FDv2RecoveryTimer'; import { ChangeSetResult, FDv2SourceResult, StatusResult } from './FDv2SourceResult'; import { createSourceManager, InitializerFactory, SynchronizerSlot } from './SourceManager'; @@ -68,7 +69,8 @@ export interface FDv2DataSource { type RaceResult = | { source: 'sync'; value: FDv2SourceResult } - | { source: 'condition'; value: ConditionType }; + | { source: 'condition'; value: ConditionType } + | { source: 'fdv2Recovery' }; /** * Creates an {@link FDv2DataSource} orchestrator. @@ -106,6 +108,11 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour selectorGetter, ); + // Deadline for returning to FDv2 after the server directed a fallback to + // FDv1. It outlives individual synchronizer runs: it is armed while an FDv2 + // source is active and must survive the switch to the fallback synchronizer. + const recoveryTimer = createFDv2RecoveryTimer(); + function markInitialized() { if (!initialized) { initialized = true; @@ -131,13 +138,46 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour } } + /** + * Arms the deadline for returning to FDv2 and reports it. A directive whose + * TTL did not survive parsing falls back to the jittered default, so there is + * always a concrete deadline. + */ + function scheduleFdv2Recovery(result: FDv2SourceResult) { + const ttlMs = result.fdv1FallbackTtlMs ?? internal.resolveFallbackTtlMs(undefined); + recoveryTimer.schedule(ttlMs); + logger?.info(`FDv2 retry scheduled in ${Math.round(ttlMs / 1000)}s.`); + } + + /** + * Clears the recovery deadline and restarts FDv2. Only one caller ever + * observes a given deadline elapse -- the main synchronizer loop and a + * background continuation are never both live for the same deadline at + * once -- so this never double-applies. + */ + function applyFdv2Recovery() { + recoveryTimer.clear(); + logger?.info('Fallback TTL elapsed, restarting FDv2 data sources.'); + sourceManager.fdv2Recovery(); + } + function handleFdv1Fallback(result: FDv2SourceResult): boolean { + if (!result.fdv1Fallback) { + return false; + } + + // The deadline is armed whether or not an FDv1 fallback synchronizer is + // configured, and a newer directive always supersedes the pending one. + scheduleFdv2Recovery(result); + // Guard: if the FDv1 fallback synchronizer itself produces a result flagged - // fdv1Fallback, do not re-run the fallback machinery - we are already on FDv1. + // fdv1Fallback, do not re-run the fallback machinery - we are already on + // FDv1. Only the new deadline applies. if (sourceManager.isCurrentSynchronizerFDv1Fallback) { return false; } - if (result.fdv1Fallback && sourceManager.hasFDv1Fallback()) { + + if (sourceManager.hasFDv1Fallback()) { sourceManager.fdv1Fallback(); return true; } @@ -237,16 +277,54 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour } } - async function runSynchronizers(): Promise { + /** + * @returns `true` if a background continuation has taken over + * responsibility for `recoveryTimer` -- the caller must then leave it + * alone rather than closing it. + */ + async function runSynchronizers(): Promise { while (!closed) { const synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive(); if (synchronizer === undefined) { + // Every slot is currently blocked. If a recovery deadline is still + // armed and there is at least one synchronizer slot for it to + // unblock, don't hold up this attempt on it -- settle it now, and + // arm a background continuation that takes over responsibility for + // `recoveryTimer`, restarting FDv2 on its own once the deadline + // elapses. The returned boolean tells every caller (this call's + // caller, and the continuation's own recursive call below) whether + // it can close the timer itself or whether responsibility has been + // handed off elsewhere. + const pendingDeadline = recoveryTimer.promise; + const handingOff = pendingDeadline !== undefined && synchronizerSlots.length > 0; + if (handingOff) { + pendingDeadline + .then(() => { + if (closed) { + return; + } + applyFdv2Recovery(); + void runSynchronizers() + .then((handedOff) => { + if (!handedOff) { + recoveryTimer.close(); + } + }) + .catch((err) => { + logger?.error(`Orchestration error during recovery: ${err}`); + recoveryTimer.close(); + }); + }) + .catch((err) => { + logger?.error(`Error during background FDv2 recovery: ${err}`); + }); + } if (!initialized) { initReject?.(new Error('All data sources exhausted without receiving data.')); initResolve = undefined; initReject = undefined; } - return; + return handingOff; } const conditions: ConditionGroup = getConditions( @@ -274,14 +352,27 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour conditions.promise.then((value) => ({ source: 'condition' as const, value })), ); } + // Read the deadline fresh each iteration: it is armed part-way + // through the loop, when a directive arrives. + const recoveryPromise = recoveryTimer.promise; + if (recoveryPromise !== undefined) { + racers.push(recoveryPromise.then(() => ({ source: 'fdv2Recovery' as const }))); + } // eslint-disable-next-line no-await-in-loop const winner = await Promise.race(racers); if (closed) { - return; + return false; } - if (winner.source === 'condition') { + if (winner.source === 'fdv2Recovery') { + // Unblocks the FDv2 slots, blocks the FDv1 fallback slot and rewinds + // to the primary. The outer loop then starts the primary FDv2 + // synchronizer, which closes the fallback synchronizer first, so + // only ever one source writes to the store. + applyFdv2Recovery(); + synchronizerRunning = false; + } else if (winner.source === 'condition') { const conditionType = winner.value as ConditionType; if (conditionType === 'fallback') { @@ -319,7 +410,7 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour synchronizerRunning = false; break; case 'shutdown': - return; + return false; case 'goodbye': // The synchronizer will handle reconnection internally. break; @@ -338,6 +429,7 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour conditions.close(); } } + return false; } async function runOrchestration(): Promise { @@ -349,9 +441,22 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour return; } - await runInitializers(); - if (!closed) { - await runSynchronizers(); + let handedOff = false; + try { + await runInitializers(); + if (!closed) { + handedOff = await runSynchronizers(); + } + } finally { + // The deadline can be armed from either phase (an initializer or a + // synchronizer observing a directive), so it is released here once + // orchestration is done for good, whether by normal completion or by + // throw, unless a background continuation (armed when every slot was + // blocked but a deadline was still pending) has taken over + // responsibility for it and will restart FDv2 later. + if (!handedOff) { + recoveryTimer.close(); + } } } @@ -388,6 +493,7 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour close() { closed = true; + recoveryTimer.close(); sourceManager.close(); }, }; diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2RecoveryTimer.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2RecoveryTimer.ts new file mode 100644 index 0000000000..e29d70c31c --- /dev/null +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2RecoveryTimer.ts @@ -0,0 +1,85 @@ +/** + * A single elapsed-time deadline for returning to the FDv2 data sources after + * the server directed the SDK onto its FDv1 fallback synchronizer. + * + * The orchestration loop races {@link FDv2RecoveryTimer.promise} alongside the + * active synchronizer, so the deadline is honored no matter which synchronizer + * is running and no matter how many slots are currently available. Unlike the + * heuristic fallback and recovery conditions, this deadline is unconditional: + * only the elapsed time governs the return to FDv2. + */ +export interface FDv2RecoveryTimer { + /** + * Resolves when the scheduled deadline elapses, and stays resolved until + * {@link FDv2RecoveryTimer.clear} is called, so an elapse that happens while + * nobody is waiting is not lost. `undefined` when no deadline is pending. + * A promise captured before a later {@link clear} or {@link schedule} call + * never settles, so a `Promise.race` consumer must re-read this getter on + * each iteration rather than holding onto an old reference. + */ + readonly promise: Promise | undefined; + + /** + * Schedule the deadline. Supersedes any pending deadline: the newest + * directive's TTL is the one that counts. No-op once closed. + * + * @param ttlMs Time to wait, in milliseconds. + */ + schedule(ttlMs: number): void; + + /** + * Cancel a pending deadline, or discard one that has already fired and been + * acted on. Leaves the timer usable for a later {@link schedule} call. + */ + clear(): void; + + /** Cancel any pending deadline and refuse further scheduling. */ + close(): void; +} + +/** + * Creates an {@link FDv2RecoveryTimer}. + */ +export function createFDv2RecoveryTimer(): FDv2RecoveryTimer { + let handle: ReturnType | undefined; + let pending: Promise | undefined; + let closed = false; + + function cancel() { + if (handle !== undefined) { + clearTimeout(handle); + handle = undefined; + } + pending = undefined; + } + + return { + get promise(): Promise | undefined { + return pending; + }, + + schedule(ttlMs: number): void { + if (closed) { + return; + } + cancel(); + pending = new Promise((resolve) => { + handle = setTimeout(() => { + // The promise is intentionally left in place after firing; the loop + // discards it with clear() once it has acted on the deadline. + handle = undefined; + resolve(); + }, ttlMs); + }); + }, + + clear(): void { + cancel(); + }, + + close(): void { + closed = true; + cancel(); + }, + }; +} diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts index 1892007bf0..ade9eed45f 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts @@ -26,8 +26,9 @@ export interface ChangeSetResult { freshness?: number; /** * When `fdv1Fallback` is true, how long (ms) to remain on FDv1 before - * attempting FDv2 recovery. `undefined` means no TTL was provided (caller - * uses a default); `0` means indefinite (no recovery). + * attempting FDv2 recovery. Always set by the source that read the directive: + * an absent, unparseable, or out-of-range server TTL is replaced with the + * jittered default, so fallback is never indefinite. */ fdv1FallbackTtlMs?: number; } @@ -43,8 +44,9 @@ export interface StatusResult { fdv1Fallback: boolean; /** * When `fdv1Fallback` is true, how long (ms) to remain on FDv1 before - * attempting FDv2 recovery. `undefined` means no TTL was provided (caller - * uses a default); `0` means indefinite (no recovery). + * attempting FDv2 recovery. Always set by the source that read the directive: + * an absent, unparseable, or out-of-range server TTL is replaced with the + * jittered default, so fallback is never indefinite. */ fdv1FallbackTtlMs?: number; } @@ -125,9 +127,8 @@ export function terminalError( * @param reason Human-readable description of why the server closed the stream. * @param fallback The FDv1 fallback directive. `fdv1Fallback === true` means the * server directed the client to fall back to FDv1. `fdv1FallbackTtlMs` is how - * long (ms) to remain on FDv1 before attempting FDv2 recovery (omit for the - * caller's default; `0` for indefinite). Same semantics as - * {@link StatusResult.fdv1FallbackTtlMs}. + * long (ms) to remain on FDv1 before attempting FDv2 recovery. Same semantics + * as {@link StatusResult.fdv1FallbackTtlMs}. */ export function goodbye(reason: string, fallback: internal.FallbackDirective): FDv2SourceResult { return { diff --git a/packages/shared/sdk-server/__tests__/data_sources/Requestor.test.ts b/packages/shared/sdk-server/__tests__/data_sources/Requestor.test.ts index 725c2af469..2e74a9ac7a 100644 --- a/packages/shared/sdk-server/__tests__/data_sources/Requestor.test.ts +++ b/packages/shared/sdk-server/__tests__/data_sources/Requestor.test.ts @@ -215,6 +215,48 @@ describe('given a requestor', () => { expect(res.fallbackToFDv1).toBe(true); }); + it.each([ + ['500', 500], + ['1', 1], + ['3600', 3600], + ])('uses a valid fallback TTL header of %s seconds as-is', async (ttl, seconds) => { + testStatus = 500; + testHeaders = { + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': ttl, + }; + + const res = await promisify<{ err: any; fdv1FallbackTtlMs: number | undefined }>((cb) => { + requestor.requestAllData((err, _body, _headers, _fallbackToFDv1, fdv1FallbackTtlMs) => + cb({ err, fdv1FallbackTtlMs }), + ); + }); + + expect(res.fdv1FallbackTtlMs).toBe(seconds * 1000); + expect(res.err.fdv1FallbackTtlMs).toBe(seconds * 1000); + }); + + it.each([undefined, '0', '-5', '3601', 'abc'])( + 'uses a default fallback TTL between 30 minutes and 1 hour for a TTL header of %s', + async (ttl) => { + testStatus = 200; + testResponse = 'a response'; + testHeaders = { 'x-ld-fd-fallback': 'true' }; + if (ttl !== undefined) { + testHeaders['x-ld-fd-fallback-ttl'] = ttl; + } + + const res = await promisify((cb) => { + requestor.requestAllData((_err, _body, _headers, _fallbackToFDv1, fdv1FallbackTtlMs) => + cb(fdv1FallbackTtlMs), + ); + }); + + expect(res).toBeGreaterThanOrEqual(30 * 60 * 1000); + expect(res).toBeLessThanOrEqual(60 * 60 * 1000); + }, + ); + it('passes the configured timeout (converted to milliseconds) in the request options', async () => { testResponse = 'a response'; const localRequestor = new Requestor(new Configuration({ timeout: 10 }), requests, baseHeaders); diff --git a/packages/shared/sdk-server/__tests__/data_sources/StreamingProcessorFDv2.test.ts b/packages/shared/sdk-server/__tests__/data_sources/StreamingProcessorFDv2.test.ts index 64cc449533..4abe440b31 100644 --- a/packages/shared/sdk-server/__tests__/data_sources/StreamingProcessorFDv2.test.ts +++ b/packages/shared/sdk-server/__tests__/data_sources/StreamingProcessorFDv2.test.ts @@ -404,6 +404,22 @@ describe('given a stream processor with mock event source', () => { expect(mockEventSource.close).toBeCalled(); }); + it('attaches the fallback TTL from the open headers to the data callback', () => { + mockEventSource.onopen({ + headers: { + ...openHeaders, + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': '120', + }, + }); + simulateEvents(); + + expect(mockDataCallback).toHaveBeenCalledWith( + true, + expect.objectContaining({ fallbackToFDv1: true, fdv1FallbackTtlMs: 120000 }), + ); + }); + it('passes error to callback if json data is malformed', async () => { simulateEvents({ 'server-intent': { diff --git a/packages/shared/sdk-server/src/LDClientImpl.ts b/packages/shared/sdk-server/src/LDClientImpl.ts index 38329c97b7..27b96c7689 100644 --- a/packages/shared/sdk-server/src/LDClientImpl.ts +++ b/packages/shared/sdk-server/src/LDClientImpl.ts @@ -525,7 +525,8 @@ function constructFDv2( // Fallback Directive (the `x-ld-fd-fallback: true` response header). When configured // explicitly via `dataSystem.fdv1Fallback`, those values take precedence over the // top-level baseUri/pollInterval. Setting `dataSystem.fdv1Fallback` to `null` leaves - // the SDK without an FDv1 fallback so the data source terminates on the directive. + // the SDK without an FDv1 fallback, so it only waits out the fallback TTL before + // retrying FDv2. const fdv1FallbackConfig = dataSystem.fdv1Fallback; if (fdv1FallbackConfig !== null) { const fdv1FallbackPollInterval = diff --git a/packages/shared/sdk-server/src/api/options/LDDataSystemOptions.ts b/packages/shared/sdk-server/src/api/options/LDDataSystemOptions.ts index 2df869a2c5..dcaf5333d2 100644 --- a/packages/shared/sdk-server/src/api/options/LDDataSystemOptions.ts +++ b/packages/shared/sdk-server/src/api/options/LDDataSystemOptions.ts @@ -73,8 +73,9 @@ export interface LDDataSystemOptions { * of the FDv2 initializer/synchronizer chain configured via {@link dataSource}. * * If omitted, the SDK uses sensible defaults derived from the rest of the configuration. - * If explicitly set to `null`, no FDv1 fallback synchronizer is configured and the SDK - * will transition to a terminal Closed state when the directive is received. + * If explicitly set to `null`, no FDv1 fallback synchronizer is configured. When the + * directive is received, the SDK stops its FDv2 synchronizers and retries them after the + * fallback TTL elapses. */ fdv1Fallback?: FDv1FallbackConfiguration | null; } diff --git a/packages/shared/sdk-server/src/api/subsystems/LDFeatureRequestor.ts b/packages/shared/sdk-server/src/api/subsystems/LDFeatureRequestor.ts index cbd7e9da78..f82c35a8af 100644 --- a/packages/shared/sdk-server/src/api/subsystems/LDFeatureRequestor.ts +++ b/packages/shared/sdk-server/src/api/subsystems/LDFeatureRequestor.ts @@ -5,15 +5,23 @@ * * @param cb The callback invoked with the result of the request. The fourth argument, * `fallbackToFDv1`, is `true` when the response carried the `x-ld-fd-fallback: true` - * header, regardless of whether the response was otherwise successful. Callers must - * apply any accompanying body/headers before honoring the fallback signal. + * header, regardless of whether the response was otherwise successful. The fifth argument, + * `fdv1FallbackTtlMs`, is how long to remain on FDv1 before retrying FDv2 when + * `fallbackToFDv1` is `true`. Callers must apply any accompanying body/headers before + * honoring the fallback signal. * @param queryParams Additional query params that will be included with the request. Values passed into this * function should already be encoded as this function will not perform encoding. * @ignore */ export interface LDFeatureRequestor { requestAllData: ( - cb: (err: any, body: any, headers: any, fallbackToFDv1?: boolean) => void, + cb: ( + err: any, + body: any, + headers: any, + fallbackToFDv1?: boolean, + fdv1FallbackTtlMs?: number, + ) => void, queryParams?: { key: string; value: string }[], ) => void; } diff --git a/packages/shared/sdk-server/src/data_sources/OneShotInitializerFDv2.ts b/packages/shared/sdk-server/src/data_sources/OneShotInitializerFDv2.ts index ff0407dd61..7218c61dad 100644 --- a/packages/shared/sdk-server/src/data_sources/OneShotInitializerFDv2.ts +++ b/packages/shared/sdk-server/src/data_sources/OneShotInitializerFDv2.ts @@ -31,7 +31,7 @@ export default class OneShotInitializerFDv2 implements subsystemCommon.DataSourc statusCallback(subsystemCommon.DataSourceState.Initializing); this._logger?.debug('Performing initialization request to LaunchDarkly for feature flag data.'); - this._requestor.requestAllData((err, body, headers, fallbackToFDv1) => { + this._requestor.requestAllData((err, body, headers, fallbackToFDv1, fdv1FallbackTtlMs) => { if (this._stopped) { return; } @@ -46,7 +46,12 @@ export default class OneShotInitializerFDv2 implements subsystemCommon.DataSourc this._logger?.warn(message); statusCallback( subsystemCommon.DataSourceState.Closed, - new LDFlagDeliveryFallbackError(DataSourceErrorKind.ErrorResponse, message, status), + new LDFlagDeliveryFallbackError( + DataSourceErrorKind.ErrorResponse, + message, + status, + fdv1FallbackTtlMs, + ), ); }; @@ -125,12 +130,18 @@ export default class OneShotInitializerFDv2 implements subsystemCommon.DataSourc // attach a fallbackToFDv1 marker to the data object: CompositeDataSource will swap // its synchronizer list to FDv1 before resolving the switchToSync transition. payloadProcessor.addPayloadListener((payload) => { - const data: { initMetadata: any; payload: any; fallbackToFDv1?: boolean } = { + const data: { + initMetadata: any; + payload: any; + fallbackToFDv1?: boolean; + fdv1FallbackTtlMs?: number; + } = { initMetadata, payload, }; if (fallbackToFDv1) { data.fallbackToFDv1 = true; + data.fdv1FallbackTtlMs = fdv1FallbackTtlMs; this._logger?.warn(`Response header indicates to fallback to FDv1`); } dataCallback(payload.type === 'full', data); diff --git a/packages/shared/sdk-server/src/data_sources/PollingProcessorFDv2.ts b/packages/shared/sdk-server/src/data_sources/PollingProcessorFDv2.ts index 53a4fefb27..c26aa5f189 100644 --- a/packages/shared/sdk-server/src/data_sources/PollingProcessorFDv2.ts +++ b/packages/shared/sdk-server/src/data_sources/PollingProcessorFDv2.ts @@ -73,7 +73,7 @@ export default class PollingProcessorFDv2 implements subsystemCommon.DataSource const startTime = Date.now(); this._logger?.debug('Polling LaunchDarkly for feature flag updates'); - this._requestor.requestAllData((err, body, headers, fallbackToFDv1) => { + this._requestor.requestAllData((err, body, headers, fallbackToFDv1, fdv1FallbackTtlMs) => { if (this._stopped) { return; } @@ -81,7 +81,7 @@ export default class PollingProcessorFDv2 implements subsystemCommon.DataSource const elapsed = Date.now() - startTime; const sleepFor = Math.max(this._pollInterval * 1000 - elapsed, 0); - // Helper to emit the terminal FDv1 fallback signal. Callers must `return` after + // Helper to emit the FDv1 fallback signal. Callers must `return` after // invoking this; no further poll is scheduled. const emitFallback = () => { const fallbackErr = @@ -93,6 +93,7 @@ export default class PollingProcessorFDv2 implements subsystemCommon.DataSource ? httpErrorMessage(err, 'polling request', 'falling back to FDv1') : `Response header indicates to fallback to FDv1`, err?.status, + fdv1FallbackTtlMs, ); this._logger?.warn(fallbackErr.message); statusCallback(subsystemCommon.DataSourceState.Closed, fallbackErr); @@ -173,12 +174,18 @@ export default class PollingProcessorFDv2 implements subsystemCommon.DataSource // object: CompositeDataSource will swap its synchronizer list to FDv1 before // resolving the switchToSync transition. payloadProcessor.addPayloadListener((payload) => { - const data: { initMetadata: any; payload: any; fallbackToFDv1?: boolean } = { + const data: { + initMetadata: any; + payload: any; + fallbackToFDv1?: boolean; + fdv1FallbackTtlMs?: number; + } = { initMetadata, payload, }; if (fallbackToFDv1) { data.fallbackToFDv1 = true; + data.fdv1FallbackTtlMs = fdv1FallbackTtlMs; this._logger?.warn(`Response header indicates to fallback to FDv1`); } dataCallback(payload.type === 'full', data); diff --git a/packages/shared/sdk-server/src/data_sources/Requestor.ts b/packages/shared/sdk-server/src/data_sources/Requestor.ts index 0b30e3567d..1e1378caf8 100644 --- a/packages/shared/sdk-server/src/data_sources/Requestor.ts +++ b/packages/shared/sdk-server/src/data_sources/Requestor.ts @@ -1,6 +1,7 @@ import { DataSourceErrorKind, getPollingUri, + internal, LDFlagDeliveryFallbackError, LDHeaders, LDLogger, @@ -77,7 +78,13 @@ export default class Requestor implements LDFeatureRequestor { } async requestAllData( - cb: (err: any, body: any, headers: any, fallbackToFDv1?: boolean) => void, + cb: ( + err: any, + body: any, + headers: any, + fallbackToFDv1?: boolean, + fdv1FallbackTtlMs?: number, + ) => void, queryParams: { key: string; value: string }[] = [], ) { const options: Options = { @@ -94,10 +101,12 @@ export default class Requestor implements LDFeatureRequestor { this._logger?.debug(`Requestor got (possibly cached) body: ${JSON.stringify(body)}`); // Per the FDv2 spec, x-ld-fd-fallback signals that the SDK should switch to FDv1 - // for the remainder of the lifetime. The signal can ride along on either a successful + // until the fallback TTL elapses. The signal can ride along on either a successful // response or an error response -- in both cases callers must apply any accompanying // payload before honoring the directive. - const fallbackToFDv1 = res.headers.get(`x-ld-fd-fallback`) === `true`; + const { fdv1Fallback: fallbackToFDv1, fdv1FallbackTtlMs } = internal.readFallbackDirective( + res.headers, + ); const responseHeaders = Object.fromEntries(res.headers.entries()); if (res.status !== 200 && res.status !== 304) { @@ -106,15 +115,22 @@ export default class Requestor implements LDFeatureRequestor { DataSourceErrorKind.ErrorResponse, `Response header indicates to fallback to FDv1.`, res.status, + fdv1FallbackTtlMs, ) : new LDPollingError( DataSourceErrorKind.ErrorResponse, `Unexpected status code: ${res.status}`, res.status, ); - return cb(err, undefined, responseHeaders, fallbackToFDv1); + return cb(err, undefined, responseHeaders, fallbackToFDv1, fdv1FallbackTtlMs); } - return cb(undefined, res.status === 304 ? null : body, responseHeaders, fallbackToFDv1); + return cb( + undefined, + res.status === 304 ? null : body, + responseHeaders, + fallbackToFDv1, + fdv1FallbackTtlMs, + ); } catch (err) { return cb(err, undefined, undefined, false); } diff --git a/packages/shared/sdk-server/src/data_sources/StreamingProcessorFDv2.ts b/packages/shared/sdk-server/src/data_sources/StreamingProcessorFDv2.ts index 06c1a8a624..e6067fcd88 100644 --- a/packages/shared/sdk-server/src/data_sources/StreamingProcessorFDv2.ts +++ b/packages/shared/sdk-server/src/data_sources/StreamingProcessorFDv2.ts @@ -20,6 +20,12 @@ import { Flag } from '../evaluation/data/Flag'; import { Segment } from '../evaluation/data/Segment'; import { processFlag, processSegment } from '../store/serialization'; +function readHeaderFallbackDirective(headers?: { + [key: string]: string; +}): internal.FallbackDirective { + return internal.readFallbackDirective({ get: (name) => headers?.[name] ?? null }); +} + export default class StreamingProcessorFDv2 implements subsystemCommon.DataSource { private readonly _serviceEndpoints: ServiceEndpoints; private readonly _headers: { [key: string]: string | string[] }; @@ -84,11 +90,13 @@ export default class StreamingProcessorFDv2 implements subsystemCommon.DataSourc statusCallback: (status: subsystemCommon.DataSourceState, err?: any) => void, ) { // this is a short term error and will be removed once FDv2 adoption is sufficient. - if (err.headers?.[`x-ld-fd-fallback`] === `true`) { + const directive = readHeaderFallbackDirective(err.headers); + if (directive.fdv1Fallback) { const fallbackErr = new LDFlagDeliveryFallbackError( DataSourceErrorKind.ErrorResponse, `Response header indicates to fallback to FDv1`, err.status, + directive.fdv1FallbackTtlMs, ); statusCallback(subsystemCommon.DataSourceState.Closed, fallbackErr); return false; @@ -131,6 +139,7 @@ export default class StreamingProcessorFDv2 implements subsystemCommon.DataSourc // finish applying the next payload before emitting the fallback signal so evaluations // can serve the server-provided data while the FDv1 synchronizer takes over. let fallbackRequested = false; + let fdv1FallbackTtlMs: number | undefined; const eventSource = this._requests.createEventSource(uri, { headers: this._headers, @@ -161,8 +170,10 @@ export default class StreamingProcessorFDv2 implements subsystemCommon.DataSourc this._initMetadata = internal.initMetadataFromHeaders(e.headers); // The fallback signal is captured here from the connection-open response headers and // is honored by the payload listener above once the next payload has been applied. - if (e.headers?.[`x-ld-fd-fallback`] === `true`) { + const directive = readHeaderFallbackDirective(e.headers); + if (directive.fdv1Fallback) { fallbackRequested = true; + fdv1FallbackTtlMs = directive.fdv1FallbackTtlMs; } statusCallback(subsystemCommon.DataSourceState.Valid); }; @@ -193,6 +204,8 @@ export default class StreamingProcessorFDv2 implements subsystemCommon.DataSourc const fallbackErr = new LDFlagDeliveryFallbackError( DataSourceErrorKind.ErrorResponse, `Response header indicates to fallback to FDv1`, + undefined, + fdv1FallbackTtlMs, ); this._logger?.warn(fallbackErr.message); statusCallback(subsystemCommon.DataSourceState.Closed, fallbackErr); @@ -218,12 +231,18 @@ export default class StreamingProcessorFDv2 implements subsystemCommon.DataSourc // switchToSync transition. A separate status callback after the data callback would // be silently dropped because the basis-during-init auto-transition disables the // composite's callback handler. - const data: { initMetadata: any; payload: any; fallbackToFDv1?: boolean } = { + const data: { + initMetadata: any; + payload: any; + fallbackToFDv1?: boolean; + fdv1FallbackTtlMs?: number; + } = { initMetadata: this._initMetadata, payload, }; if (fallbackRequested) { data.fallbackToFDv1 = true; + data.fdv1FallbackTtlMs = fdv1FallbackTtlMs; this._logger?.warn(`Response header indicates to fallback to FDv1`); } dataCallback(payload.type === 'full', data);