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/FDv1FallbackRecovery.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts new file mode 100644 index 0000000000..f925095483 --- /dev/null +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv1FallbackRecovery.test.ts @@ -0,0 +1,179 @@ +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('schedules FDv2 recovery once and honors the TTL (no FDv1 slot configured)', async () => { + const statusManager = makeStatusManager(); + const logger = makeLogger(); + + // No FDv1 fallback slot: the first directive halts and closes this + // synchronizer (every slot is blocked with nothing to unblock), so it + // never gets the chance to tick again before recovery. + const sync = makeTickingSynchronizer( + () => interrupted(makeErrorInfo(), { fdv1Fallback: true, fdv1FallbackTtlMs: 30 }), + 8, // faster than the 30ms TTL, in case the halt ever stopped applying + ); + + 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; + + // Exactly one directive is ever processed, since halting the source on + // the first one leaves it with no chance to tick again before recovery. + expect(scheduleLogs).toBe(1); + // The deadline set by that directive is honored: the SDK returns to FDv2 + // once it elapses. + 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, + }); + + let resolved = false; + const startPromise = ds.start(); + // 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); + }); + + // 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/__tests__/datasource/fdv2/FDv2DataSource.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts index 0ffd59987b..79c539687a 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(); @@ -1076,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({ @@ -1090,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(); }); @@ -1202,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 () => { @@ -1591,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 }), ); @@ -1624,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 }), ]); @@ -1695,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/FDv2DataManagerBase.ts b/packages/shared/sdk-client/src/datasource/FDv2DataManagerBase.ts index dd5a1b4251..daf90089cd 100644 --- a/packages/shared/sdk-client/src/datasource/FDv2DataManagerBase.ts +++ b/packages/shared/sdk-client/src/datasource/FDv2DataManagerBase.ts @@ -420,6 +420,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 || selector !== undefined, }); dataSource diff --git a/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts b/packages/shared/sdk-client/src/datasource/fdv2/Conditions.ts index b7bb9f53d5..249431a776 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,28 @@ 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` 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' && result.payload.type !== 'none') { + cancel(); + } + }, + true, + ); +} + /** * Creates a group of conditions that are managed together. * @@ -193,8 +224,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 +233,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..34311c8c69 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,26 @@ 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; + + /** + * 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; } /** @@ -85,6 +106,8 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour logger, fallbackTimeoutMs = DEFAULT_FALLBACK_TIMEOUT_MS, recoveryTimeoutMs = DEFAULT_RECOVERY_TIMEOUT_MS, + initFallbackTimeoutMs = DEFAULT_INIT_FALLBACK_TIMEOUT_MS, + hasExistingData = false, } = config; let initialized = false; @@ -166,22 +189,36 @@ 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. + // The deadline is armed on every directive. A non-FDv1-origin directive + // always halts the FDv2 source below, so it can never repeat from a + // still-running source; the only other repeat source is the FDv1 + // synchronizer's own traffic, which legitimately supersedes the pending + // deadline with a fresh TTL. 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. + // Guard: if the FDv1 fallback synchronizer itself produces this result, + // do not re-run the fallback machinery - we are already on FDv1. if (sourceManager.isCurrentSynchronizerFDv1Fallback) { return false; } - 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 @@ -330,8 +367,10 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour const conditions: ConditionGroup = getConditions( sourceManager.getAvailableSynchronizerCount(), sourceManager.isPrimeSynchronizer(), + initialized || hasExistingData, fallbackTimeoutMs, recoveryTimeoutMs, + initFallbackTimeoutMs, ); if (conditions.promise) { @@ -390,7 +429,14 @@ 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. + // 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') { 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; },