From aa4fd7a5ca8f997b71a1b99349710da9803046a1 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 2 Sep 2026 17:04:34 -0400 Subject: [PATCH 1/4] fix: Add the missing "10s uninitialized" leg to the FDv2 fallback condition --- .../datasource/fdv2/Conditions.test.ts | 72 +++++++++++++++++-- .../datasource/fdv2/FDv2DataSource.test.ts | 45 ++++++++++++ .../src/datasource/fdv2/Conditions.ts | 65 ++++++++++++++--- .../src/datasource/fdv2/FDv2DataSource.ts | 11 +++ 4 files changed, 176 insertions(+), 17 deletions(-) diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts index 193f4fa81e..6e034dc11e 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts @@ -3,6 +3,7 @@ import { DataSourceErrorKind } from '@launchdarkly/js-sdk-common'; import { createConditionGroup, createFallbackCondition, + createInitFallbackCondition, createRecoveryCondition, getConditions, } from '../../../src/datasource/fdv2/Conditions'; @@ -180,6 +181,34 @@ it('recovery condition close can be called multiple times', () => { // Should not throw }); +// -- init fallback condition -- + +it('init fallback condition fires after timeout without any inform calls', async () => { + const condition = createInitFallbackCondition(10); + expect(await condition.promise).toBe('fallback'); + condition.close(); +}); + +it('init fallback condition cancels timer when a changeSet is received', async () => { + const condition = createInitFallbackCondition(10); + condition.inform(makeChangeSet()); + expect(await raceTimeout(condition.promise, 50)).toBe(DID_NOT_RESOLVE); + condition.close(); +}); + +it('init fallback condition is unaffected by an interrupted result', async () => { + const condition = createInitFallbackCondition(10); + condition.inform(makeInterrupted()); + expect(await condition.promise).toBe('fallback'); + condition.close(); +}); + +it('init fallback condition does not fire after close', async () => { + const condition = createInitFallbackCondition(10); + condition.close(); + expect(await raceTimeout(condition.promise, 50)).toBe(DID_NOT_RESOLVE); +}); + // -- condition group -- it('condition group returns undefined promise for empty group', () => { @@ -233,17 +262,17 @@ it('condition group close stops all conditions', async () => { // -- getConditions -- it('getConditions returns empty group when only one synchronizer is available', () => { - const group = getConditions(1, true, 10, 10); + const group = getConditions(1, true, true, 10, 10, 10); expect(group.promise).toBeUndefined(); }); it('getConditions returns empty group when zero synchronizers are available', () => { - const group = getConditions(0, true, 10, 10); + const group = getConditions(0, true, true, 10, 10, 10); expect(group.promise).toBeUndefined(); }); -it('getConditions returns fallback only for prime synchronizer', async () => { - const group = getConditions(2, true, 10, 200); +it('getConditions returns fallback only for prime synchronizer once initialized', async () => { + const group = getConditions(2, true, true, 10, 200, 200); group.inform(makeInterrupted()); @@ -251,9 +280,40 @@ it('getConditions returns fallback only for prime synchronizer', async () => { group.close(); }); -it('getConditions returns fallback and recovery for non-prime synchronizer', async () => { - const group = getConditions(2, false, 200, 10); +it('getConditions returns fallback and recovery for non-prime synchronizer once initialized', async () => { + const group = getConditions(2, false, true, 200, 10, 200); expect(await group.promise).toBe('recovery'); group.close(); }); + +it('getConditions adds an init-fallback leg for a prime synchronizer before initialization', async () => { + const group = getConditions(2, true, false, 200, 200, 10); + + // No interrupted result ever arrives, so only the init-fallback leg can fire. + expect(await group.promise).toBe('fallback'); + group.close(); +}); + +it('getConditions adds an init-fallback leg for a non-prime synchronizer before initialization', async () => { + const group = getConditions(2, false, false, 200, 200, 10); + + expect(await group.promise).toBe('fallback'); + group.close(); +}); + +it('getConditions does not fire the init-fallback leg once a changeSet is received', async () => { + const group = getConditions(2, true, false, 200, 200, 10); + + group.inform(makeChangeSet()); + + expect(await raceTimeout(group.promise!, 50)).toBe(DID_NOT_RESOLVE); + group.close(); +}); + +it('getConditions omits the init-fallback leg once the data system is initialized', async () => { + const group = getConditions(2, true, true, 200, 200, 10); + + expect(await raceTimeout(group.promise!, 50)).toBe(DID_NOT_RESOLVE); + group.close(); +}); 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 0ffd59987b..40869197ae 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts @@ -619,6 +619,51 @@ it('falls back to next synchronizer when fallback condition fires', async () => ds.close(); }); +it('falls back to next synchronizer when it never delivers data or a status, before initialization', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + const payload = makePayload({ state: 'selector' }); + + // sync1 never resolves. + let sync1NextResolve: ((r: FDv2SourceResult) => void) | undefined; + const sync1: Synchronizer = { + next: () => + new Promise((resolve) => { + sync1NextResolve = resolve; + }), + close() { + sync1NextResolve?.(shutdown()); + }, + }; + + const sync2 = makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })]); + + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot({ create: () => sync1 }), + createSynchronizerSlot({ create: () => sync2 }), + ]; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: slots, + dataCallback, + statusManager, + selectorGetter: noSelector, + logger, + fallbackTimeoutMs: 100000, + initFallbackTimeoutMs: 10, + }); + + // start() resolves when the init-fallback leg fires (after 10ms) despite + // sync1 never reporting interrupted, moving to sync2 which delivers data. + await ds.start(); + + expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('Fallback condition fired')); + expect(dataCallback).toHaveBeenCalledWith(payload); + ds.close(); +}); + it('recovers to primary synchronizer when recovery condition fires', async () => { const dataCallback = jest.fn(); const statusManager = makeStatusManager(); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts b/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts index b7bb9f53d5..57d060726b 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts @@ -4,6 +4,7 @@ import { FDv2SourceResult } from './FDv2SourceResult'; export const DEFAULT_FALLBACK_TIMEOUT_MS = 2 * 60 * 1000; // 120 seconds export const DEFAULT_RECOVERY_TIMEOUT_MS = 5 * 60 * 1000; // 300 seconds +export const DEFAULT_INIT_FALLBACK_TIMEOUT_MS = 10 * 1000; // 10 seconds /** * A timer that resolves with a {@link ConditionType} when it fires. @@ -93,12 +94,18 @@ type InformHandler = (result: FDv2SourceResult, controls: TimerControls) => void * @param type The {@link ConditionType} to resolve with when the timer fires. * @param informHandler Optional callback invoked on each `inform()` call. * When omitted, the timer starts immediately and `inform()` is a no-op. - * When provided, the timer must be started explicitly via `controls.start()`. + * When provided, the timer must be started explicitly via `controls.start()`, + * unless `autoStart` is set. + * @param autoStart Start the timer immediately at construction, in addition + * to letting `informHandler` observe (and potentially cancel) it. Used by + * conditions that must run from the moment the source becomes active but + * still need to react to results (e.g. cancel on a changeSet). */ function createCondition( timeoutMs: number, type: ConditionType, informHandler?: InformHandler, + autoStart?: boolean, ): Condition { let resolve: ((t: ConditionType) => void) | undefined; let timer: ConditionTimer | undefined; @@ -125,8 +132,10 @@ function createCondition( timer = undefined; } - // No inform handler - start immediately (recovery behavior) - if (!informHandler) { + // No inform handler - start immediately (recovery behavior). An inform + // handler that opts into autoStart also starts immediately, but (unlike + // the no-handler case) can still have its timer cancelled via inform(). + if (!informHandler || autoStart) { startTimer(); } @@ -173,6 +182,26 @@ export function createRecoveryCondition(timeoutMs: number): Condition { return createCondition(timeoutMs, 'recovery'); } +/** + * Creates the "not initialized" leg of the fallback condition. The timer + * starts immediately (unlike {@link createFallbackCondition}, which only + * starts once interrupted) and is cancelled once a `changeSet` is received. + * If it fires first, it resolves with `'fallback'`, same as the + * interrupted-based leg -- the two are alternatives (an OR), not a sequence. + */ +export function createInitFallbackCondition(timeoutMs: number): Condition { + return createCondition( + timeoutMs, + 'fallback', + (result, { cancel }) => { + if (result.type === 'changeSet') { + cancel(); + } + }, + true, + ); +} + /** * Creates a group of conditions that are managed together. * @@ -193,8 +222,8 @@ export function createConditionGroup(conditions: Condition[]): ConditionGroup { } /** - * Determines which conditions to create based on the synchronizer's position - * and availability. + * Determines which conditions to create based on the synchronizer's position, + * availability, and whether the data system has ever been initialized. * * - If there is only one available synchronizer, no conditions are needed * (there is nowhere to fall back to). @@ -202,28 +231,42 @@ export function createConditionGroup(conditions: Condition[]): ConditionGroup { * fallback condition is created. * - If the current synchronizer is non-primary, both fallback and recovery * conditions are created. + * - Whenever the data system has not yet been initialized (regardless of + * primacy), an additional fallback leg is added that fires if no data + * arrives within `initFallbackTimeoutMs` -- this is the "10 seconds + * elapses while data system not initialized with data" half of the + * default Fallback Condition, distinct from the "interrupted" half. Once + * the data system has initialized, this leg never applies again for the + * lifetime of the SDK. * * @param availableSyncCount Number of available (non-blocked) synchronizers. * @param isPrime Whether the current synchronizer is the primary. + * @param initialized Whether the data system has ever received data. * @param fallbackTimeoutMs Fallback condition timeout. * @param recoveryTimeoutMs Recovery condition timeout. + * @param initFallbackTimeoutMs Not-yet-initialized fallback leg timeout. */ export function getConditions( availableSyncCount: number, isPrime: boolean, + initialized: boolean, fallbackTimeoutMs: number = DEFAULT_FALLBACK_TIMEOUT_MS, recoveryTimeoutMs: number = DEFAULT_RECOVERY_TIMEOUT_MS, + initFallbackTimeoutMs: number = DEFAULT_INIT_FALLBACK_TIMEOUT_MS, ): ConditionGroup { if (availableSyncCount <= 1) { return createConditionGroup([]); } - if (isPrime) { - return createConditionGroup([createFallbackCondition(fallbackTimeoutMs)]); + const conditions: Condition[] = []; + if (!initialized) { + conditions.push(createInitFallbackCondition(initFallbackTimeoutMs)); + } + + conditions.push(createFallbackCondition(fallbackTimeoutMs)); + if (!isPrime) { + conditions.push(createRecoveryCondition(recoveryTimeoutMs)); } - return createConditionGroup([ - createFallbackCondition(fallbackTimeoutMs), - createRecoveryCondition(recoveryTimeoutMs), - ]); + return createConditionGroup(conditions); } diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts index 4f840565fb..a747b56540 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts @@ -5,6 +5,7 @@ import { ConditionGroup, ConditionType, DEFAULT_FALLBACK_TIMEOUT_MS, + DEFAULT_INIT_FALLBACK_TIMEOUT_MS, DEFAULT_RECOVERY_TIMEOUT_MS, getConditions, } from './Conditions'; @@ -48,6 +49,13 @@ export interface FDv2DataSourceConfig { /** Recovery condition timeout in ms (default 300s). */ recoveryTimeoutMs?: number; + + /** + * Not-yet-initialized fallback leg timeout in ms (default 10s). Fires if + * the data system has never received data within this long of the current + * synchronizer becoming active. + */ + initFallbackTimeoutMs?: number; } /** @@ -85,6 +93,7 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour logger, fallbackTimeoutMs = DEFAULT_FALLBACK_TIMEOUT_MS, recoveryTimeoutMs = DEFAULT_RECOVERY_TIMEOUT_MS, + initFallbackTimeoutMs = DEFAULT_INIT_FALLBACK_TIMEOUT_MS, } = config; let initialized = false; @@ -330,8 +339,10 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour const conditions: ConditionGroup = getConditions( sourceManager.getAvailableSynchronizerCount(), sourceManager.isPrimeSynchronizer(), + initialized, fallbackTimeoutMs, recoveryTimeoutMs, + initFallbackTimeoutMs, ); if (conditions.promise) { From ed6a74ef4d1e725623a53c9843e4b9ffa5f4e88e Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 2 Sep 2026 17:05:00 -0400 Subject: [PATCH 2/4] fix: Avoid starving FDv2 recovery scheduling and premature init on a none payload --- .../fdv2/FDv1FallbackRecovery.test.ts | 131 ++++++++++++++++++ .../src/datasource/fdv2/Conditions.ts | 10 +- .../src/datasource/fdv2/FDv2DataSource.ts | 23 ++- 3 files changed, 154 insertions(+), 10 deletions(-) create mode 100644 packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts new file mode 100644 index 0000000000..6ffa763e91 --- /dev/null +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts @@ -0,0 +1,131 @@ +import { createFDv2DataSource } from '../../../src/datasource/fdv2/FDv2DataSource'; +import { + changeSet, + FDv2SourceResult, + interrupted, + shutdown, +} from '../../../src/datasource/fdv2/FDv2SourceResult'; +import { createSynchronizerSlot } from '../../../src/datasource/fdv2/SourceManager'; +import { Synchronizer } from '../../../src/datasource/fdv2/Synchronizer'; +import { + makeErrorInfo, + makeLogger, + makeMockSynchronizer, + makePayload, + makeStatusManager, + noSelector, +} from './orchestrationTestHelpers'; + +/** + * A synchronizer that emits a new result every `intervalMs`, forever, until + * closed. Unlike `makeMockSynchronizer` (a fixed sequence that then blocks), + * this models a real endpoint that keeps talking -- e.g. a polling + * synchronizer whose every response carries the fallback directive header. + */ +function makeTickingSynchronizer( + makeResult: () => FDv2SourceResult, + intervalMs: number, +): Synchronizer { + let closed = false; + let timer: ReturnType | undefined; + return { + next(): Promise { + if (closed) { + return Promise.resolve(shutdown()); + } + return new Promise((resolve) => { + timer = setTimeout(() => { + resolve(closed ? shutdown() : makeResult()); + }, intervalMs); + }); + }, + close() { + closed = true; + if (timer !== undefined) { + clearTimeout(timer); + } + }, + }; +} + +it('does not let a directive repeated faster than its TTL starve FDv2 recovery (no FDv1 slot configured)', async () => { + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + // No FDv1 fallback slot: per the pre-existing gap, the active FDv2 + // synchronizer is never stopped on a directive, so it keeps running and + // keeps producing directive-carrying results (e.g. a polling synchronizer + // whose every response carries "X-LD-FD-Fallback: true"). + const sync = makeTickingSynchronizer( + () => interrupted(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 30 }), + 8, // faster than the 30ms TTL -- each result used to re-arm the deadline before it fired + ); + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: [createSynchronizerSlot({ create: () => sync })], + dataCallback: jest.fn(), + statusManager, + logger, + selectorGetter: noSelector, + }); + + const startPromise = ds.start(); + startPromise.catch(() => {}); + + await new Promise((resolve) => { + setTimeout(resolve, 200); + }); + ds.close(); + + const infoCalls = logger.info.mock.calls.map((c) => String(c[0])); + const scheduleLogs = infoCalls.filter((m) => m.includes('retry scheduled')).length; + const recoveredLogs = infoCalls.filter((m) => m.includes('restarting FDv2')).length; + + // handleFdv1Fallback() only re-arms the deadline once per Directed Fallback + // engagement -- a repeated directive from a source that has not + // transitioned off FDv2 does not keep resetting it. + expect(scheduleLogs).toBe(1); + // The deadline set by the first directive is honored: the SDK returns to + // FDv2 once it elapses, per DATASYSTEM v2 Requirement 1.6.4.1. + expect(recoveredLogs).toBeGreaterThan(0); +}); + +it('does not let a data-less "none" changeSet mark the data system initialized or disarm the 10s init-fallback leg', async () => { + const statusManager = makeStatusManager(); + + const nonePayload = makePayload({ type: 'none', state: '' }); + const primary = makeMockSynchronizer([changeSet(nonePayload, { fdv1Fallback: false })]); + const secondaryFactory = { create: jest.fn(() => makeMockSynchronizer([])) }; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: [ + createSynchronizerSlot({ create: () => primary }), + createSynchronizerSlot(secondaryFactory), + ], + dataCallback: jest.fn(), + statusManager, + selectorGetter: noSelector, + initFallbackTimeoutMs: 15, + fallbackTimeoutMs: 1_000_000, + }); + + const startPromise = ds.start(); + startPromise.catch(() => {}); + + // Give the 10s ("15ms" in this test) init-fallback leg a chance to fire + // and move the orchestrator on to the secondary synchronizer. + await new Promise((resolve) => { + setTimeout(resolve, 60); + }); + ds.close(); + + // FDv2DataSource.ts's synchronizer-phase changeSet handler and + // Conditions.ts's createInitFallbackCondition() both require + // `payload.type !== 'none'`, matching the initializer phase. A response + // that carries no real data does not disarm the "10 seconds elapses while + // the data system is not initialized with data" leg (DATASYSTEM v2 spec, + // section 1.5), so the secondary synchronizer is tried. + expect(secondaryFactory.create).toHaveBeenCalled(); +}); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts b/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts index 57d060726b..249431a776 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts @@ -185,16 +185,18 @@ export function createRecoveryCondition(timeoutMs: number): Condition { /** * Creates the "not initialized" leg of the fallback condition. The timer * starts immediately (unlike {@link createFallbackCondition}, which only - * starts once interrupted) and is cancelled once a `changeSet` is received. - * If it fires first, it resolves with `'fallback'`, same as the - * interrupted-based leg -- the two are alternatives (an OR), not a sequence. + * starts once interrupted) and is cancelled once a `changeSet` carrying real + * data is received. A `'none'` payload (e.g. an unchanged poll response) + * carries no data, so it does not cancel the timer. If it fires first, it + * resolves with `'fallback'`, same as the interrupted-based leg -- the two + * are alternatives (an OR), not a sequence. */ export function createInitFallbackCondition(timeoutMs: number): Condition { return createCondition( timeoutMs, 'fallback', (result, { cancel }) => { - if (result.type === 'changeSet') { + if (result.type === 'changeSet' && result.payload.type !== 'none') { cancel(); } }, diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts index a747b56540..8e5d6fb691 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts @@ -175,17 +175,25 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour 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. Only the new deadline applies. + // FDv1. Its own traffic legitimately supersedes the pending deadline with + // the new TTL. if (sourceManager.isCurrentSynchronizerFDv1Fallback) { + scheduleFdv2Recovery(result); return false; } + // A directive observed while a deadline is already pending, but not from + // the FDv1 synchronizer's own traffic, is a repeated signal from a source + // that has not yet transitioned off FDv2 (e.g. because no FDv1 fallback + // synchronizer is configured, so the FDv2 source keeps running). The + // first directive's TTL already governs the return to FDv2; a repeat + // must not keep re-arming the deadline, or it could never elapse. + if (recoveryTimer.promise === undefined) { + scheduleFdv2Recovery(result); + } + if (sourceManager.hasFDv1Fallback()) { sourceManager.fdv1Fallback(); return true; @@ -401,7 +409,10 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour if (syncResult.type === 'changeSet') { applyChangeSet(syncResult); - if (!initialized) { + // A 'none' payload (e.g. an unchanged poll response) carries no + // data, so it must not count toward "initialized with data" -- + // matching the initializer phase's same payload.type check. + if (!initialized && syncResult.payload.type !== 'none') { markInitialized(); } } else if (syncResult.type === 'status') { From 17c463be5aeb80b21eab89499595a30dd4e736fb Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 2 Sep 2026 17:05:15 -0400 Subject: [PATCH 3/4] fix: Avoid post-mode-switch fallback churn using hasExistingData --- .../fdv2/FDv1FallbackRecovery.test.ts | 66 ++++++++++++++++--- .../src/datasource/FDv2DataManagerBase.ts | 4 ++ .../src/datasource/fdv2/FDv2DataSource.ts | 22 ++++++- 3 files changed, 82 insertions(+), 10 deletions(-) diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts index 6ffa763e91..77114320ef 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts @@ -111,21 +111,71 @@ it('does not let a data-less "none" changeSet mark the data system initialized o fallbackTimeoutMs: 1_000_000, }); + let resolved = false; const startPromise = ds.start(); - startPromise.catch(() => {}); + // start() rejects once the orchestrator exhausts its sources (or is + // closed) without ever initializing; that rejection is expected and + // irrelevant to this assertion, so it must not surface as an unhandled + // rejection -- both branches of this single chain handle it. + startPromise.then( + () => { + resolved = true; + }, + () => {}, + ); // Give the 10s ("15ms" in this test) init-fallback leg a chance to fire // and move the orchestrator on to the secondary synchronizer. await new Promise((resolve) => { setTimeout(resolve, 60); }); - ds.close(); - // FDv2DataSource.ts's synchronizer-phase changeSet handler and - // Conditions.ts's createInitFallbackCondition() both require - // `payload.type !== 'none'`, matching the initializer phase. A response - // that carries no real data does not disarm the "10 seconds elapses while - // the data system is not initialized with data" leg (DATASYSTEM v2 spec, - // section 1.5), so the secondary synchronizer is tried. + // Conditions.ts's createInitFallbackCondition() requires + // `payload.type !== 'none'` to cancel the leg, so a data-less response + // does not disarm it and the secondary synchronizer is tried. expect(secondaryFactory.create).toHaveBeenCalled(); + // FDv2DataSource.ts's synchronizer-phase changeSet handler requires the + // same check before calling markInitialized(), matching the initializer + // phase, so start() must not resolve on a response with no real data. + // Checked before close() -- close() itself rejects the still-pending + // start() promise, which must not be mistaken for a resolution here. + expect(resolved).toBe(false); + ds.close(); +}); + +it('does not churn synchronizers on a "none" changeSet when the data system already has existing data', async () => { + const statusManager = makeStatusManager(); + + // Models a connection-mode switch after the SDK was already initialized: + // a new FDv2DataSource instance is created, but the server has nothing + // new to say -- a 'none' response is the correct, healthy answer, not + // evidence of an uninitialized system. + const nonePayload = makePayload({ type: 'none', state: '' }); + const primary = makeMockSynchronizer([changeSet(nonePayload, { fdv1Fallback: false })]); + const secondaryFactory = { create: jest.fn(() => makeMockSynchronizer([])) }; + + const ds = createFDv2DataSource({ + initializerFactories: [], + synchronizerSlots: [ + createSynchronizerSlot({ create: () => primary }), + createSynchronizerSlot(secondaryFactory), + ], + dataCallback: jest.fn(), + statusManager, + selectorGetter: noSelector, + initFallbackTimeoutMs: 15, + fallbackTimeoutMs: 1_000_000, + hasExistingData: true, + }); + + await ds.start(); + + // Give the 10s ("15ms" in this test) init-fallback leg a chance to fire + // -- it must not, because hasExistingData suppresses it entirely. + await new Promise((resolve) => { + setTimeout(resolve, 60); + }); + ds.close(); + + expect(secondaryFactory.create).not.toHaveBeenCalled(); }); diff --git a/packages/shared/sdk-client/src/datasource/FDv2DataManagerBase.ts b/packages/shared/sdk-client/src/datasource/FDv2DataManagerBase.ts index 41c092f10e..37f7b41dc9 100644 --- a/packages/shared/sdk-client/src/datasource/FDv2DataManagerBase.ts +++ b/packages/shared/sdk-client/src/datasource/FDv2DataManagerBase.ts @@ -418,6 +418,10 @@ export function createFDv2DataManagerBase( logger, fallbackTimeoutMs, recoveryTimeoutMs, + // Whether the SDK already has data from a previous data source + // instance, e.g. this call is rebuilding the data source for a + // connection-mode switch rather than the first identify. + hasExistingData: initialized, }); dataSource diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts index 8e5d6fb691..064caa91f6 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts @@ -56,6 +56,19 @@ export interface FDv2DataSourceConfig { * synchronizer becoming active. */ initFallbackTimeoutMs?: number; + + /** + * Whether the data system already has data from a previous data source + * instance -- e.g. this instance was created for a connection-mode switch + * after the SDK was already initialized. A fresh instance otherwise has + * no way to know that, so without this flag a data-less `'none'` response + * (the server's "you're already up to date" answer) looks identical to a + * still-uninitialized system: it would never satisfy the init-fallback + * leg's data check, so that leg would keep firing forever. When true, the + * leg never applies, and a `'none'` response is enough to settle this + * instance's {@link FDv2DataSource.start} promise. + */ + hasExistingData?: boolean; } /** @@ -94,6 +107,7 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour fallbackTimeoutMs = DEFAULT_FALLBACK_TIMEOUT_MS, recoveryTimeoutMs = DEFAULT_RECOVERY_TIMEOUT_MS, initFallbackTimeoutMs = DEFAULT_INIT_FALLBACK_TIMEOUT_MS, + hasExistingData = false, } = config; let initialized = false; @@ -347,7 +361,7 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour const conditions: ConditionGroup = getConditions( sourceManager.getAvailableSynchronizerCount(), sourceManager.isPrimeSynchronizer(), - initialized, + initialized || hasExistingData, fallbackTimeoutMs, recoveryTimeoutMs, initFallbackTimeoutMs, @@ -412,7 +426,11 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour // A 'none' payload (e.g. an unchanged poll response) carries no // data, so it must not count toward "initialized with data" -- // matching the initializer phase's same payload.type check. - if (!initialized && syncResult.payload.type !== 'none') { + // Exception: if the data system already has data from a prior + // instance (hasExistingData), a 'none' response is a genuine + // "you're already up to date" answer, not evidence of an empty + // system, so it still settles this instance's start() promise. + if (!initialized && (syncResult.payload.type !== 'none' || hasExistingData)) { markInitialized(); } } else if (syncResult.type === 'status') { From e6a1d582a08ac2046427ff27eaa34c9809ee9562 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 2 Sep 2026 17:05:32 -0400 Subject: [PATCH 4/4] fix: Close orphaned sources when every synchronizer slot is blocked --- .../datasource/fdv2/FDv2DataSource.test.ts | 143 +++++++++++------- .../datasource/fdv2/SourceManager.test.ts | 22 +++ .../src/datasource/fdv2/FDv2DataSource.ts | 20 ++- .../src/datasource/fdv2/SourceManager.ts | 4 + 4 files changed, 129 insertions(+), 60 deletions(-) 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 40869197ae..79c539687a 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts @@ -1121,13 +1121,20 @@ it('fdv1 fallback blocks other synchronizers', async () => { ds.close(); }); -it('fdv1 fallback ignored when no FDv1 synchronizer is configured', async () => { +it('stops the synchronizer and schedules recovery when no FDv1 synchronizer is configured', async () => { const dataCallback = jest.fn(); const statusManager = makeStatusManager(); + const logger = makeLogger(); const payload = makePayload({ state: 'selector' }); + const laterPayload = makePayload({ state: 'should-not-be-delivered' }); - // Synchronizer sends changeSet with fdv1Fallback flag but no FDv1 slot exists - const sync = makeMockSynchronizer([changeSet(payload, { fdv1Fallback: true })]); + // The directive-carrying changeSet is delivered, then (if the + // synchronizer kept running) a second changeSet would follow -- but with + // no FDv1 slot configured, the SDK must halt the synchronizer instead. + const sync = makeMockSynchronizer([ + changeSet(payload, { fdv1Fallback: true, fdv1FallbackTtlMs: 100_000 }), + changeSet(laterPayload, { fdv1Fallback: false }), + ]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; const ds = createFDv2DataSource({ @@ -1135,13 +1142,22 @@ it('fdv1 fallback ignored when no FDv1 synchronizer is configured', async () => synchronizerSlots: slots, dataCallback, statusManager, + logger, selectorGetter: noSelector, }); await ds.start(); - // Should process the changeSet normally without error + // The directive's own changeSet is still delivered before it halts the + // synchronizer. expect(dataCallback).toHaveBeenCalledWith(payload); + // No FDv1 fallback synchronizer configured: run no synchronizer and + // report interrupted while the recovery deadline is pending. + expect(statusManager.requestStateUpdate).toHaveBeenCalledWith('INTERRUPTED'); + // The synchronizer is halted, not left running: its queued second result + // must never be delivered. + expect(dataCallback).not.toHaveBeenCalledWith(laterPayload); + ds.close(); }); @@ -1247,6 +1263,57 @@ it('stops initializer chain when a transfer-none changeSet triggers fdv1 fallbac ds.close(); }); +it('stops the initializer chain on a directive when synchronizers are configured, even with no FDv1 slot', async () => { + const dataCallback = jest.fn(); + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + const secondInit = makeMockInitializer( + changeSet(makePayload({ state: 'second-selector' }), { fdv1Fallback: false }), + ); + const secondInitRunSpy = jest.spyOn(secondInit, 'run'); + + const recoveredPayload = makePayload({ state: 'recovered-selector' }); + let fdv2Creations = 0; + const fdv2Factory = () => { + fdv2Creations += 1; + return makeMockSynchronizer([changeSet(recoveredPayload, { fdv1Fallback: false })]); + }; + const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: fdv2Factory })]; + + const ds = createFDv2DataSource({ + initializerFactories: [ + makeInitFactory( + makeMockInitializer( + terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 20 }), + ), + ), + makeInitFactory(secondInit), + ], + synchronizerSlots: slots, + dataCallback, + statusManager, + logger, + selectorGetter: noSelector, + }); + + await expect(ds.start()).rejects.toThrow('All data sources exhausted without receiving data.'); + + // The second initializer never runs: a synchronizer is configured to + // recover to, so the directive halts the chain per spec even though no + // FDv1 fallback synchronizer exists to hand off to. + expect(secondInitRunSpy).not.toHaveBeenCalled(); + expect(statusManager.requestStateUpdate).toHaveBeenCalledWith('INTERRUPTED'); + + // The recovery deadline armed by the directive still fires and restarts + // the FDv2 synchronizer. + await statusManager.waitForState('VALID', 1); + expect(fdv2Creations).toBe(1); + expect(dataCallback).toHaveBeenCalledWith(recoveredPayload); + + ds.close(); +}); + // -- FDv1 fallback re-trigger guard (regression: SDK-2617) -- it('does not re-trigger fallback when the fdv1 synchronizer itself yields a fallback-flagged result', async () => { @@ -1636,8 +1703,9 @@ it('cancels the pending recovery deadline when an initializer throws after armin // 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. + // moves on to the next initializer. No synchronizers are configured at + // all, so there is nothing to halt or recover to; handleFdv1Fallback does + // not break the loop early. const armingInit = makeMockInitializer( changeSet(makePayload({ type: 'none' }), { fdv1Fallback: true, fdv1FallbackTtlMs: 10000 }), ); @@ -1669,68 +1737,31 @@ it('cancels the pending recovery deadline when an initializer throws after armin // -- 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. +it('keeps recovering across multiple hand-off cycles when no FDv1 fallback synchronizer is configured', async () => { + // With no FDv1 fallback slot, a directive blocks the only synchronizer + // slot outright (see "stops the synchronizer..." tests below), so the + // exhaustion branch's background-continuation hand-off engages on the + // very first directive. This proves that hand-off survives more than one + // cycle: a second directive on the recovered synchronizer must arm and + // hand off its own continuation rather than dropping the deadline. 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 })]); + return makeMockSynchronizer([ + terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 5 }), + ]); } 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. + // this forces a second hand-off cycle, proving recovery still works + // after the first one. return makeMockSynchronizer([ terminalError(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 5 }), ]); @@ -1740,7 +1771,7 @@ it('keeps recovering after an already-elapsed deadline is discovered at exhausti const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: fdv2Factory })]; const ds = createFDv2DataSource({ - initializerFactories: [makeInitFactory(armingInit), makeInitFactory(delayingInit)], + initializerFactories: [], synchronizerSlots: slots, dataCallback, statusManager, diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/SourceManager.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/SourceManager.test.ts index bb5bfb3247..ae0bba35f8 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/SourceManager.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/SourceManager.test.ts @@ -138,6 +138,28 @@ it('returns undefined when all synchronizers are blocked', () => { expect(manager.getNextAvailableSynchronizerAndSetActive()).toBeUndefined(); }); +it('closes the active source when every synchronizer becomes blocked', () => { + const sync1 = makeMockSynchronizer(); + const sync2 = makeMockSynchronizer(); + const slots: SynchronizerSlot[] = [ + createSynchronizerSlot(makeSyncFactory(sync1)), + createSynchronizerSlot(makeSyncFactory(sync2)), + ]; + + const manager = createSourceManager([], slots, () => undefined); + + expect(manager.getNextAvailableSynchronizerAndSetActive()).toBe(sync1); + + // Both slots become blocked while sync1 is still the active source -- + // nothing will call next() on it again, so it must be closed rather than + // left running. + slots[0].state = 'blocked'; + slots[1].state = 'blocked'; + + expect(manager.getNextAvailableSynchronizerAndSetActive()).toBeUndefined(); + expect(sync1.closed).toBe(true); +}); + it('wraps around to find available synchronizers', () => { const sync1a = makeMockSynchronizer(); const sync1b = makeMockSynchronizer(); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts index 064caa91f6..f44f9e9e2e 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts @@ -208,11 +208,23 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour scheduleFdv2Recovery(result); } - if (sourceManager.hasFDv1Fallback()) { - sourceManager.fdv1Fallback(); - return true; + // With no synchronizers configured at all (e.g. one-shot mode), there + // is nothing to halt or recover to -- keep trying other initializers. + if (synchronizerSlots.length === 0) { + return false; } - return false; + + if (!sourceManager.hasFDv1Fallback()) { + // No FDv1 fallback synchronizer configured: run no synchronizer and + // report interrupted while the recovery deadline is pending. + statusManager.requestStateUpdate('INTERRUPTED'); + } + + // Halts the current run: with an FDv1 fallback configured it becomes + // the sole active source; without one, every FDv2 slot is blocked and + // the orchestrator runs no synchronizer until recovery. + sourceManager.fdv1Fallback(); + return true; } // The orchestration loops intentionally use await-in-loop for sequential diff --git a/packages/shared/sdk-client/src/datasource/fdv2/SourceManager.ts b/packages/shared/sdk-client/src/datasource/fdv2/SourceManager.ts index 14dfd65b75..ce25c85a07 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/SourceManager.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/SourceManager.ts @@ -203,6 +203,10 @@ export function createSourceManager( visited += 1; } + // Every slot is blocked: nothing will read from the previously active + // source again, so close it instead of leaving its transport running + // (and its results piling up unread) for as long as recovery takes. + closeActiveSource(); return undefined; },