From 2b280eb3bc4251948cd03337db81533e99ba03a9 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Fri, 21 Aug 2026 19:16:41 -0400 Subject: [PATCH] feat: Schedule FDv2 recovery after the FDv1 fallback TTL elapses A directed FDv1 fallback now always carries a concrete deadline for returning to FDv2. An absent, unparseable, or out-of-range fallback TTL is replaced with a 1-hour default, jittered down by up to half so a fleet that fell back together does not retry in lockstep; a server-supplied in-range TTL is used as given. Fallback is therefore never indefinite. FDv2RecoveryTimer holds that single deadline. It outlives individual synchronizer runs, so the orchestrator races it alongside the active synchronizer and, when every slot is blocked, hands it to a background continuation that restarts the FDv2 sources once the deadline elapses. --- .../datasource/fdv2/FDv2DataSource.test.ts | 609 ++++++++++++++++++ .../datasource/fdv2/FDv2RecoveryTimer.test.ts | 110 ++++ .../datasource/fdv2/FDv2SourceResult.test.ts | 2 +- .../datasource/fdv2/PollingBase.test.ts | 15 +- .../fdv2/PollingInitializer.test.ts | 4 +- .../datasource/fdv2/fallbackDirective.test.ts | 112 +++- .../src/datasource/fdv2/FDv2DataSource.ts | 130 +++- .../src/datasource/fdv2/FDv2RecoveryTimer.ts | 85 +++ .../src/datasource/fdv2/FDv2SourceResult.ts | 15 +- .../src/datasource/fdv2/fallbackDirective.ts | 91 ++- 10 files changed, 1109 insertions(+), 64 deletions(-) create mode 100644 packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2RecoveryTimer.test.ts create mode 100644 packages/shared/sdk-client/src/datasource/fdv2/FDv2RecoveryTimer.ts 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)) }; }