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 285ee0833f..833209c48f 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts @@ -13,6 +13,7 @@ import { SynchronizerSlot, } from '../../../src/datasource/fdv2/SourceManager'; import { Synchronizer } from '../../../src/datasource/fdv2/Synchronizer'; +import { DEFAULT_FDV1_FALLBACK_TTL_MS } from '../../../src/datasource/fdv2/fallbackDirective'; import { makeCacheInitFactory, makeErrorInfo, @@ -1226,3 +1227,611 @@ 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(DEFAULT_FDV1_FALLBACK_TTL_MS / 2000); + expect(seconds).toBeLessThanOrEqual(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 c9c25f2114..0b40bcf1d8 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts @@ -1,6 +1,7 @@ import { DataSourceErrorKind } from '@launchdarkly/js-sdk-common'; import { poll } from '../../../src/datasource/fdv2/PollingBase'; +import { DEFAULT_FDV1_FALLBACK_TTL_MS } from '../../../src/datasource/fdv2/fallbackDirective'; import { makeErrorRequestor, makeFDv2Body, @@ -500,7 +501,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 +512,13 @@ 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( + DEFAULT_FDV1_FALLBACK_TTL_MS / 2, + ); + expect((result as any).fdv1FallbackTtlMs).toBeLessThanOrEqual(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 +529,10 @@ 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( + DEFAULT_FDV1_FALLBACK_TTL_MS / 2, + ); + expect((result as any).fdv1FallbackTtlMs).toBeLessThanOrEqual(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..578217fb6a 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts @@ -2,6 +2,7 @@ import { sleep } from '@launchdarkly/js-sdk-common'; import { FDv2PollResponse, FDv2Requestor } from '../../../src/datasource/fdv2/FDv2Requestor'; import { createPollingInitializer } from '../../../src/datasource/fdv2/PollingInitializer'; +import { DEFAULT_FDV1_FALLBACK_TTL_MS } from '../../../src/datasource/fdv2/fallbackDirective'; import { makeFDv2Body, makeHeaders, makeLogger, makeSuccessResponse } from './testHelpers'; jest.mock('@launchdarkly/js-sdk-common', () => ({ @@ -190,7 +191,8 @@ 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(DEFAULT_FDV1_FALLBACK_TTL_MS / 2); + expect(result.fdv1FallbackTtlMs).toBeLessThanOrEqual(DEFAULT_FDV1_FALLBACK_TTL_MS); } expect(requestor.poll).toHaveBeenCalledTimes(1); expect(sleep).not.toHaveBeenCalled(); diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/fallbackDirective.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/fallbackDirective.test.ts index d8b0aa1097..bb5873599d 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/fallbackDirective.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/fallbackDirective.test.ts @@ -1,6 +1,8 @@ import { + DEFAULT_FDV1_FALLBACK_TTL_MS, readFallbackDirective, readGoodbyeFallbackDirective, + resolveFallbackTtlMs, } from '../../../src/datasource/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,42 @@ 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 +177,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/sdk-client/src/datasource/fdv2/FDv2DataSource.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts index 76fa2fbac0..40367bf23b 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts @@ -8,8 +8,10 @@ import { DEFAULT_RECOVERY_TIMEOUT_MS, getConditions, } from './Conditions'; +import { createFDv2RecoveryTimer } from './FDv2RecoveryTimer'; import { ChangeSetResult, FDv2SourceResult, StatusResult } from './FDv2SourceResult'; import { createSourceManager, InitializerFactory, SynchronizerSlot } from './SourceManager'; +import { resolveFallbackTtlMs } from './fallbackDirective'; /** * Callback invoked when the orchestrator produces a changeSet payload. @@ -68,7 +70,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 +109,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 +139,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 ?? 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 +278,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 +353,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 +411,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,9 +430,9 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour conditions.close(); } } + return false; } - async function runOrchestration(): Promise { // No sources configured at all, so there is nothing to wait for. // Report valid immediately. @@ -350,9 +442,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(); + } } } @@ -389,6 +494,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 b5b61c2955..eedf4fb294 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts @@ -27,8 +27,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; } @@ -44,8 +45,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; } @@ -126,9 +128,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: FallbackDirective): FDv2SourceResult { return { diff --git a/packages/shared/sdk-client/src/datasource/fdv2/fallbackDirective.ts b/packages/shared/sdk-client/src/datasource/fdv2/fallbackDirective.ts index bca2464a27..1d1a0c7813 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/fallbackDirective.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/fallbackDirective.ts @@ -1,15 +1,59 @@ /** - * 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`, shared by the streaming and polling sources. + * This is the single place that interprets `x-ld-fd-fallback`, + * `x-ld-fd-fallback-ttl`, and a goodbye message's `protocolFallbackTTL`, + * shared by the streaming and polling sources. */ export interface FallbackDirective { fdv1Fallback: boolean; @@ -33,19 +77,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 +89,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 +106,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)) }; }