From 6b5e7d591efc4bfcb9794d400fdac3e5deea678c Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Thu, 16 Jul 2026 16:30:01 -0400 Subject: [PATCH 01/10] feat: Thread FDv1 fallback TTL through polling and streaming data sources This commit will add in fallback TTL reading support as well as allow RN native event source read headers. --- .../react-native-sse/EventSource.test.ts | 34 ++++- .../react-native-sse/EventSource.ts | 25 ++- .../fromExternal/react-native-sse/types.ts | 1 + .../datasource/fdv2/PollingBase.test.ts | 75 +++++++++ .../datasource/fdv2/StreamingFDv2Base.test.ts | 143 ++++++++++++++++++ .../src/datasource/fdv2/PollingBase.ts | 56 +++---- .../src/datasource/fdv2/StreamingFDv2Base.ts | 93 ++++++++++-- 7 files changed, 379 insertions(+), 48 deletions(-) diff --git a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts index 84fa7788d6..8436d41f3b 100644 --- a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts +++ b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts @@ -38,7 +38,12 @@ describe('EventSource', () => { abort: jest.fn(), }; - jest.spyOn(window, 'XMLHttpRequest').mockImplementation(() => mockXhr as XMLHttpRequest); + const xhrSpy = jest.spyOn(window, 'XMLHttpRequest').mockImplementation(() => mockXhr as XMLHttpRequest); + // Preserve static constants that EventSource reads from the constructor reference. + // @ts-ignore + xhrSpy.LOADING = 3; + // @ts-ignore + xhrSpy.DONE = 4; eventSource = new EventSource(uri, { logger }); eventSource.onclose = jest.fn(); @@ -135,4 +140,31 @@ describe('EventSource', () => { expect(mockXhr.open).toHaveBeenLastCalledWith('GET', `${uri}?basis=initial`, true); }); + + test('calls onopen with parsed response headers', () => { + const onopen = jest.fn(); + eventSource.onopen = onopen; + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = ''; + + jest.runAllTimers(); + + mockXhr.readyState = 4; + mockXhr.status = 200; + mockXhr.onreadystatechange(); + + expect(onopen).toHaveBeenCalledTimes(1); + expect(onopen).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'open', + headers: expect.objectContaining({ + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': '60', + }), + }), + ); + }); }); diff --git a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts index 1d04733303..f61b24613c 100644 --- a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts +++ b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts @@ -194,7 +194,10 @@ export default class EventSource { if (this._status === this.CONNECTING) { this._retryCount = 0; this._status = this.OPEN; - this.dispatch('open', { type: 'open' }); + this.dispatch('open', { + type: 'open', + headers: this._parseResponseHeaders(this._xhr.getAllResponseHeaders()), + }); this._logger?.debug('[EventSource][onreadystatechange][OPEN] Connection opened.'); } @@ -353,12 +356,28 @@ export default class EventSource { } } + private _parseResponseHeaders(raw: string | null): Record { + const result: Record = {}; + if (!raw) { + return result; + } + raw.split('\r\n').forEach((line) => { + const separatorIndex = line.indexOf(': '); + if (separatorIndex > 0) { + const key = line.slice(0, separatorIndex).toLowerCase(); + const value = line.slice(separatorIndex + 2); + result[key] = value; + } + }); + return result; + } + dispatch>(type: T, data: EventSourceEvent) { this._eventHandlers[type]?.forEach((handler: EventSourceListener) => handler(data)); switch (type) { case 'open': - this.onopen(); + this.onopen(data); break; case 'close': this.onclose(); @@ -389,7 +408,7 @@ export default class EventSource { return this._status; } - onopen() {} + onopen(_e: any) {} onclose() {} onerror(_err: any) {} onretrying(_e: any) {} diff --git a/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts b/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts index 465b150fdf..df29111ea0 100644 --- a/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts +++ b/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts @@ -10,6 +10,7 @@ export interface MessageEvent { export interface OpenEvent { type: 'open'; + headers?: Record; } export interface CloseEvent { 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 1b17f380fd..7e07dc54c7 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts @@ -303,6 +303,23 @@ describe('given a goodbye event in the response', () => { expect(result.reason).toBe('server-shutdown'); } }); + + it('emits terminal_error when the response header signals fallback', async () => { + const body = makeFDv2Body([{ event: 'goodbye', data: { reason: 'bye' } }]); + const requestor = makeRequestor({ + status: 200, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '45' }), + body, + }); + + const result = await poll(requestor, undefined, logger); + + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('terminal_error'); + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(45000); + }); }); describe('given a server error event in the response', () => { @@ -468,3 +485,61 @@ describe('given a delete-object event', () => { } }); }); + +describe('given x-ld-fd-fallback-ttl header', () => { + it('reads TTL in seconds and converts to ms on a changeSet', async () => { + const body = makeFullPayloadBody({ flagA: { value: true } }); + const requestor = makeRequestor({ + status: 200, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }), + body, + }); + + const result = await poll(requestor, undefined, logger); + + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(60000); + }); + + it('treats TTL "0" as indefinite (0 ms)', async () => { + const body = makeFullPayloadBody({ flagA: { value: true } }); + const requestor = makeRequestor({ + status: 200, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '0' }), + body, + }); + + const result = await poll(requestor, undefined, logger); + + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(0); + }); + + it('leaves TTL undefined when the ttl header is absent but fallback is true', async () => { + const body = makeFullPayloadBody({ flagA: { value: true } }); + const requestor = makeRequestor({ + status: 200, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true' }), + body, + }); + + const result = await poll(requestor, undefined, logger); + + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBeUndefined(); + }); + + it('stamps TTL on a non-success error response', async () => { + const requestor = makeRequestor({ + status: 503, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '30' }), + body: null, + }); + + const result = await poll(requestor, undefined, logger); + + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(30000); + }); +}); + diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts index cc45e8577a..60eb2965c9 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts @@ -514,3 +514,146 @@ it('calling start twice does not create a second EventSource', () => { base.close(); }); + +it('reads x-ld-fd-fallback-ttl on the error path', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + const willRetry = simulateErrorFilter(mockRequests, { + status: 200, + message: 'fallback', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }, + }); + expect(willRetry).toBe(false); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('terminal_error'); + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(60000); + + base.close(); +}); + +it('defers the open fallback directive to the next changeSet', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '90' }, + }); + + sendFullTransfer(mockEventSource, [{ key: 'flag-a', version: 1, value: 'green' }]); + + const result = await base.takeResult(); + expect(result.type).toBe('changeSet'); + if (result.type !== 'changeSet') return; + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(90000); + + base.close(); +}); + +it('emits terminal_error for a goodbye with protocolFallbackTTL', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + simulateEvent(mockEventSource, 'goodbye', { + reason: 'falling back', + protocolFallbackTTL: 60, + }); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('terminal_error'); + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(60000); + + base.close(); +}); + +it('emits terminal_error for a goodbye when a deferred open directive is pending', async () => { + // When onopen sets pendingFallback but the goodbye event has no protocolFallbackTTL, + // the pending directive triggers the fallback path with the deferred TTL. + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // Open with fallback headers (defers the directive) + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '45' }, + }); + + // Goodbye fires before any payload — the pending directive triggers the fallback + simulateEvent(mockEventSource, 'goodbye', { reason: 'bye' }); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('terminal_error'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(45000); + + base.close(); +}); + +it('clears a pending fallback directive when onopen fires without the fallback header', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // First open carries fallback headers — arms the deferred directive + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }, + }); + + // Reconnect without fallback header — must clear the stale directive + mockEventSource.onopen({ type: 'open', headers: {} }); + + // A payload arriving after the clean reconnect must NOT carry fdv1Fallback=true + sendFullTransfer(mockEventSource, [{ key: 'flag-a', version: 1, value: 'blue' }]); + + const result = await base.takeResult(); + expect(result.type).toBe('changeSet'); + if (result.type !== 'changeSet') return; + expect(result.fdv1Fallback).toBe(false); + expect(result.fdv1FallbackTtlMs).toBeUndefined(); + + base.close(); +}); + +it('close resets the pending fallback directive', async () => { + // Calling close() must clear pendingFallback/pendingFallbackTtlMs so a + // subsequently re-created source does not inherit stale state. + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // Arm the deferred directive + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '30' }, + }); + + // Close before any payload arrives — the shutdown result must NOT carry TTL + base.close(); + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('shutdown'); + expect(result.fdv1FallbackTtlMs).toBeUndefined(); +}); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts b/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts index 31b0b84971..a1ff4c42e3 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts @@ -13,11 +13,7 @@ import { interrupted, terminalError, } from './FDv2SourceResult'; - -function getFallback(headers: { get(name: string): string | null }): boolean { - const value = headers.get('x-ld-fd-fallback'); - return value !== null && value.toLowerCase() === 'true'; -} +import { readFallbackDirective } from './fallbackDirective'; function getEnvironmentId(headers: { get(name: string): string | null }): string | undefined { return headers.get('x-ld-envid') ?? undefined; @@ -27,7 +23,7 @@ function getEnvironmentId(headers: { get(name: string): string | null }): string * Process FDv2 events using the protocol handler directly. * * We use `createProtocolHandler` rather than `PayloadProcessor` because - * the PayloadProcessor does not surface goodbye/serverError actions — + * the PayloadProcessor does not surface goodbye/serverError actions: * it only forwards payloads and actionable errors. For polling results, * we need full control over all protocol action types. */ @@ -35,6 +31,7 @@ function processEvents( events: internal.FDv2Event[], fdv1Fallback: boolean, environmentId: string | undefined, + fdv1FallbackTtlMs: number | undefined, logger?: LDLogger, ): FDv2SourceResult { const handler = internal.createProtocolHandler( @@ -55,31 +52,32 @@ function processEvents( switch (action.type) { case 'payload': - earlyResult = changeSet(action.payload, fdv1Fallback, environmentId); + earlyResult = changeSet(action.payload, fdv1Fallback, environmentId, undefined, fdv1FallbackTtlMs); break; case 'goodbye': - earlyResult = goodbye(action.reason, fdv1Fallback); + if (fdv1Fallback) { + earlyResult = terminalError(errorInfoFromUnknown(action.reason), true, fdv1FallbackTtlMs); + } else { + earlyResult = goodbye(action.reason, fdv1Fallback); + } break; case 'serverError': { const errorInfo = errorInfoFromUnknown(action.reason); logger?.error(`Server error during polling: ${action.reason}`); - earlyResult = interrupted(errorInfo, fdv1Fallback); + earlyResult = interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); break; } case 'error': { - // Actionable protocol errors (MISSING_PAYLOAD, PROTOCOL_ERROR) if (action.kind === 'MISSING_PAYLOAD' || action.kind === 'PROTOCOL_ERROR') { const errorInfo = errorInfoFromInvalidData(action.message); logger?.warn(`Protocol error during polling: ${action.message}`); - earlyResult = interrupted(errorInfo, fdv1Fallback); + earlyResult = interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); } else { - // Non-actionable errors (UNKNOWN_EVENT) are logged but don't stop processing logger?.warn(action.message); } break; } default: - // 'none' — continue processing next event break; } }); @@ -88,10 +86,9 @@ function processEvents( return earlyResult; } - // Events didn't produce a result const errorInfo = errorInfoFromUnknown('Unexpected end of polling response'); logger?.error('Unexpected end of polling response'); - return interrupted(errorInfo, fdv1Fallback); + return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); } /** @@ -109,22 +106,26 @@ export async function poll( logger?: LDLogger, ): Promise { let fdv1Fallback = false; + let fdv1FallbackTtlMs: number | undefined; let environmentId: string | undefined; try { const response = await requestor.poll(basis); - fdv1Fallback = getFallback(response.headers); + const directive = readFallbackDirective(response.headers); + fdv1Fallback = directive.fdv1Fallback; + fdv1FallbackTtlMs = directive.fdv1FallbackTtlMs; environmentId = getEnvironmentId(response.headers); - // 304 Not Modified: treat as server-intent with intentCode 'none' - // (Spec Requirement 10.1.2) + // 304 Not Modified: no payload has changed since the last poll. + // Synthesize a 'none' payload so the orchestrator's changeSet path runs + // normally without touching stored flags. if (response.status === 304) { const nonePayload: internal.Payload = { version: 0, type: 'none', updates: [], }; - return changeSet(nonePayload, fdv1Fallback, environmentId); + return changeSet(nonePayload, fdv1Fallback, environmentId, undefined, fdv1FallbackTtlMs); } // Non-success HTTP status @@ -134,15 +135,15 @@ export async function poll( const recoverable = response.status <= 0 || isHttpRecoverable(response.status); return recoverable - ? interrupted(errorInfo, fdv1Fallback) - : terminalError(errorInfo, fdv1Fallback); + ? interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs) + : terminalError(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); } - // Successful response — process FDv2 events + // Successful response - process FDv2 events if (!response.body) { const errorInfo = errorInfoFromInvalidData('Empty response body'); logger?.error('Polling request received empty response body'); - return interrupted(errorInfo, fdv1Fallback); + return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); } let parsed: internal.FDv2EventsCollection; @@ -151,7 +152,7 @@ export async function poll( } catch { const errorInfo = errorInfoFromInvalidData('Malformed JSON data in polling response'); logger?.error('Polling request received malformed data'); - return interrupted(errorInfo, fdv1Fallback); + return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); } if (!Array.isArray(parsed.events)) { @@ -159,15 +160,14 @@ export async function poll( 'Invalid polling response: missing or invalid events array', ); logger?.error('Polling response does not contain a valid events array'); - return interrupted(errorInfo, fdv1Fallback); + return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); } - return processEvents(parsed.events, fdv1Fallback, environmentId, logger); + return processEvents(parsed.events, fdv1Fallback, environmentId, fdv1FallbackTtlMs, logger); } catch (err: any) { - // Network or other I/O error from the fetch itself const message = err?.message ?? String(err); logger?.error(`Polling request failed with network error: ${message}`); const errorInfo = errorInfoFromNetworkError(message); - return interrupted(errorInfo, fdv1Fallback); + return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); } } diff --git a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts index 8a77fda52b..4748a29402 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts @@ -25,6 +25,7 @@ import { shutdown, terminalError, } from './FDv2SourceResult'; +import { readFallbackDirective, readGoodbyeFallbackDirective } from './fallbackDirective'; /** * Handler invoked when a legacy `"ping"` event is received on the stream. @@ -37,7 +38,7 @@ export interface PingHandler { } /** - * The public surface of the streaming base — used by + * The public surface of the streaming base, used by * {@link createStreamingInitializer} and {@link createStreamingSynchronizer} * to control the EventSource connection and consume results. * @@ -109,6 +110,11 @@ export function createStreamingBase(config: { let eventSource: EventSource | undefined; let connectionAttemptStartTime: number | undefined; let fdv1Fallback = false; + let fdv1FallbackTtlMs: number | undefined; + // Directive deferred from a stream `onopen` carrying x-ld-fd-fallback; applied + // to the NEXT changeSet so the payload is delivered before the SDK falls back. + let pendingFallbackTtlMs: number | undefined; + let pendingFallback = false; let started = false; let stopped = false; @@ -127,26 +133,51 @@ export function createStreamingBase(config: { connectionAttemptStartTime = undefined; } - function handleAction(action: internal.ProtocolAction): void { + function handleAction(action: internal.ProtocolAction, rawData?: unknown): void { switch (action.type) { case 'payload': logConnectionResult(true); - resultQueue.put(changeSet(action.payload, fdv1Fallback)); + if (pendingFallback) { + // A fallback directive was deferred from onopen; apply it now that + // the payload has been delivered. + fdv1Fallback = true; + fdv1FallbackTtlMs = pendingFallbackTtlMs; + resultQueue.put(changeSet(action.payload, true, undefined, undefined, pendingFallbackTtlMs)); + pendingFallback = false; + pendingFallbackTtlMs = undefined; + } else { + resultQueue.put(changeSet(action.payload, fdv1Fallback, undefined, undefined, fdv1FallbackTtlMs)); + } break; - case 'goodbye': - resultQueue.put(goodbye(action.reason, fdv1Fallback)); + case 'goodbye': { + // In-band fallback signal read from the raw goodbye data (used by SDKs + // that cannot read streaming response headers), or a directive deferred + // from onopen (pendingFallback). Surface as a terminal fallback result. + const goodbyeDirective = readGoodbyeFallbackDirective(rawData); + if (goodbyeDirective.fdv1Fallback || pendingFallback) { + const ttlMs = goodbyeDirective.fdv1Fallback + ? goodbyeDirective.fdv1FallbackTtlMs + : pendingFallbackTtlMs; + fdv1Fallback = true; + resultQueue.put(terminalError(errorInfoFromUnknown(action.reason), true, ttlMs)); + pendingFallback = false; + pendingFallbackTtlMs = undefined; + } else { + resultQueue.put(goodbye(action.reason, fdv1Fallback)); + } break; + } case 'serverError': - resultQueue.put(interrupted(errorInfoFromUnknown(action.reason), fdv1Fallback)); + resultQueue.put(interrupted(errorInfoFromUnknown(action.reason), fdv1Fallback, fdv1FallbackTtlMs)); break; case 'error': // Only actionable errors are queued; informational ones (UNKNOWN_EVENT) // are logged by the protocol handler. if (action.kind === 'MISSING_PAYLOAD' || action.kind === 'PROTOCOL_ERROR') { - resultQueue.put(interrupted(errorInfoFromInvalidData(action.message), fdv1Fallback)); + resultQueue.put(interrupted(errorInfoFromInvalidData(action.message), fdv1Fallback, fdv1FallbackTtlMs)); } break; @@ -157,25 +188,32 @@ export function createStreamingBase(config: { } function handleError(err: HttpErrorResponse): boolean { - // Check for FDv1 fallback header. - if (err.headers?.['x-ld-fd-fallback'] === 'true') { + // Check for FDv1 fallback header (with optional TTL). + const errHeaders = err.headers ?? {}; + const directive = readFallbackDirective({ + get: (name: string) => errHeaders[name.toLowerCase()] ?? null, + }); + if (directive.fdv1Fallback) { fdv1Fallback = true; + fdv1FallbackTtlMs = directive.fdv1FallbackTtlMs; logConnectionResult(false); - resultQueue.put(terminalError(errorInfoFromHttpError(err.status ?? 0), true)); + resultQueue.put( + terminalError(errorInfoFromHttpError(err.status ?? 0), true, directive.fdv1FallbackTtlMs), + ); return false; } if (!shouldRetry(err)) { config.logger?.error(httpErrorMessage(err, 'streaming request')); logConnectionResult(false); - resultQueue.put(terminalError(errorInfoFromHttpError(err.status ?? 0), fdv1Fallback)); + resultQueue.put(terminalError(errorInfoFromHttpError(err.status ?? 0), fdv1Fallback, fdv1FallbackTtlMs)); return false; } config.logger?.warn(httpErrorMessage(err, 'streaming request', 'will retry')); logConnectionResult(false); logConnectionAttempt(); - resultQueue.put(interrupted(errorInfoFromHttpError(err.status ?? 0), fdv1Fallback)); + resultQueue.put(interrupted(errorInfoFromHttpError(err.status ?? 0), fdv1Fallback, fdv1FallbackTtlMs)); return true; } @@ -206,13 +244,13 @@ export function createStreamingBase(config: { ); config.logger?.debug(`Data follows: ${event.data}`); resultQueue.put( - interrupted(errorInfoFromInvalidData('Malformed JSON in EventStream'), fdv1Fallback), + interrupted(errorInfoFromInvalidData('Malformed JSON in EventStream'), fdv1Fallback, fdv1FallbackTtlMs), ); return; } const action = protocolHandler.processEvent({ event: eventName, data: parsed }); - handleAction(action); + handleAction(action, parsed); }); }); } @@ -249,6 +287,7 @@ export function createStreamingBase(config: { interrupted( errorInfoFromNetworkError(err?.message ?? 'Error during ping poll'), fdv1Fallback, + fdv1FallbackTtlMs, ), ); } @@ -291,13 +330,33 @@ export function createStreamingBase(config: { return; } resultQueue.put( - interrupted(errorInfoFromNetworkError(err?.message ?? 'IO Error'), fdv1Fallback), + interrupted(errorInfoFromNetworkError(err?.message ?? 'IO Error'), fdv1Fallback, fdv1FallbackTtlMs), ); }; - es.onopen = () => { + es.onopen = (e?: { headers?: { [key: string]: string } }) => { + if (stopped) { + return; + } config.logger?.info('Opened LaunchDarkly stream connection'); protocolHandler.reset(); + + // Reset pending directive on every connection open so a reconnect that + // does not carry the fallback header clears any state from a prior attempt. + pendingFallback = false; + pendingFallbackTtlMs = undefined; + + // SDKs whose EventSource exposes response headers can + // detect FDv1 fallback at connection open. Defer the directive and apply + // it to the next changeSet so the payload is delivered first. + const openHeaders = e?.headers; + if (openHeaders) { + const directive = readFallbackDirective({ + get: (name: string) => openHeaders[name.toLowerCase()] ?? null, + }); + pendingFallback = directive.fdv1Fallback; + pendingFallbackTtlMs = directive.fdv1FallbackTtlMs; + } }; es.onretrying = (e) => { @@ -310,6 +369,8 @@ export function createStreamingBase(config: { return; } stopped = true; + pendingFallback = false; + pendingFallbackTtlMs = undefined; eventSource?.close(); eventSource = undefined; resultQueue.put(shutdown()); From 80e3d414f8151f89ddf1d1fe2d6d0afea6882228 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Thu, 16 Jul 2026 17:00:15 -0400 Subject: [PATCH 02/10] ci: bump client common size limit --- .github/workflows/sdk-client.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/sdk-client.yml b/.github/workflows/sdk-client.yml index 4b366d43cd..0e75de6bfc 100644 --- a/.github/workflows/sdk-client.yml +++ b/.github/workflows/sdk-client.yml @@ -32,4 +32,4 @@ jobs: target_file: 'packages/shared/sdk-client/dist/esm/index.mjs' package_name: '@launchdarkly/js-client-sdk-common' pr_number: ${{ github.event.number }} - size_limit: 39300 + size_limit: 40000 From 3ada250b4bf27d663f0a271fc034bd663c175511 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Thu, 16 Jul 2026 19:06:46 -0400 Subject: [PATCH 03/10] fix: RN event source does not capture headers in a timely manner --- .../react-native-sse/EventSource.test.ts | 56 +++++++++++++++++++ .../react-native-sse/EventSource.ts | 25 +++++++-- 2 files changed, 77 insertions(+), 4 deletions(-) diff --git a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts index 8436d41f3b..96b117bb7e 100644 --- a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts +++ b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts @@ -167,4 +167,60 @@ describe('EventSource', () => { }), ); }); + + test('calls onopen with parsed response headers during streaming (onprogress, pre-DONE)', () => { + const onopen = jest.fn(); + eventSource.onopen = onopen; + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = ''; + + jest.runAllTimers(); + + // Simulate a chunk arriving while the connection is still LOADING (readyState 3). + // This is the real runtime path for a live stream: DONE (readyState 4) only + // fires once the connection closes, so gating 'open' on DONE would mean the + // event, and the fallback headers it carries, never fires during normal streaming. + mockXhr.readyState = 3; + mockXhr.status = 200; + mockXhr.onprogress(); + + expect(onopen).toHaveBeenCalledTimes(1); + expect(onopen).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'open', + headers: expect.objectContaining({ + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': '60', + }), + }), + ); + }); + + test('dispatches open exactly once when onprogress precedes an onreadystatechange DONE', () => { + const onopen = jest.fn(); + eventSource.onopen = onopen; + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = ''; + + jest.runAllTimers(); + + // onprogress observes the connection first (LOADING) and transitions to OPEN. + mockXhr.readyState = 3; + mockXhr.status = 200; + mockXhr.onprogress(); + + // A later onreadystatechange at DONE must not re-dispatch open: the status is + // already OPEN (no longer CONNECTING), so its open-dispatch guard is a no-op. + mockXhr.readyState = 4; + mockXhr.status = 200; + mockXhr.onreadystatechange(); + + expect(onopen).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts index f61b24613c..bbdcdfbf77 100644 --- a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts +++ b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts @@ -7,10 +7,17 @@ * 4. replaced all for of loops with foreach * * Additional changes: - * 1. separated event handling to use onprogress for data changes - * and onreadystatechange for status changes. This is to address - * an issue with Vega OS where they do not fire a readyStatechange - * event when the response is received. + * 1. separated event handling to use onprogress for data changes and + * onreadystatechange for status changes, since some platforms (e.g. Vega OS) + * never fire a readystatechange event while the response is still streaming in. + * 2. onprogress now also performs the CONNECTING -> OPEN transition and dispatches + * open, guarded by the same this._status === CONNECTING check onreadystatechange + * already used. Previously open only fired from onreadystatechange at + * readyState DONE, which for a long-lived stream only happens at connection + * close, so the parsed response headers carried on open were never seen while + * the stream was actually running. onprogress fires as soon as the response + * starts loading, so headers are available as soon as the connection opens + * instead of only at teardown. */ import type { EventSourceEvent, EventSourceListener, EventSourceOptions, EventType } from './types'; @@ -158,6 +165,16 @@ export default class EventSource { ); if (this._xhr.status >= 200 && this._xhr.status < 400) { + if (this._status === this.CONNECTING) { + this._retryCount = 0; + this._status = this.OPEN; + this.dispatch('open', { + type: 'open', + headers: this._parseResponseHeaders(this._xhr.getAllResponseHeaders()), + }); + this._logger?.debug('[EventSource][onprogress][OPEN] Connection opened.'); + } + this._handleEvent(this._xhr.responseText || ''); } else { this._status = this.ERROR; From 7b48590cce47c4dab136be9f3a9d8ac0801716ba Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Fri, 17 Jul 2026 12:51:40 -0400 Subject: [PATCH 04/10] fix: fallback variables need to be read on stream errors --- .github/workflows/sdk-client.yml | 2 +- .../datasource/fdv2/StreamingFDv2Base.test.ts | 227 +++++++++++++++++- .../src/datasource/fdv2/StreamingFDv2Base.ts | 74 +++++- 3 files changed, 289 insertions(+), 14 deletions(-) diff --git a/.github/workflows/sdk-client.yml b/.github/workflows/sdk-client.yml index 0e75de6bfc..45e52acd64 100644 --- a/.github/workflows/sdk-client.yml +++ b/.github/workflows/sdk-client.yml @@ -32,4 +32,4 @@ jobs: target_file: 'packages/shared/sdk-client/dist/esm/index.mjs' package_name: '@launchdarkly/js-client-sdk-common' pr_number: ${{ github.event.number }} - size_limit: 40000 + size_limit: 41500 diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts index 60eb2965c9..e241474083 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts @@ -595,7 +595,7 @@ it('emits terminal_error for a goodbye when a deferred open directive is pending headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '45' }, }); - // Goodbye fires before any payload — the pending directive triggers the fallback + // Goodbye fires before any payload; the pending directive triggers the fallback simulateEvent(mockEventSource, 'goodbye', { reason: 'bye' }); const result = await base.takeResult(); @@ -614,13 +614,13 @@ it('clears a pending fallback directive when onopen fires without the fallback h const base = createBase(mockRequests, logger); base.start(); - // First open carries fallback headers — arms the deferred directive + // First open carries fallback headers, arming the deferred directive mockEventSource.onopen({ type: 'open', headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }, }); - // Reconnect without fallback header — must clear the stale directive + // Reconnect without fallback header; must clear the stale directive mockEventSource.onopen({ type: 'open', headers: {} }); // A payload arriving after the clean reconnect must NOT carry fdv1Fallback=true @@ -649,7 +649,7 @@ it('close resets the pending fallback directive', async () => { headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '30' }, }); - // Close before any payload arrives — the shutdown result must NOT carry TTL + // Close before any payload arrives; the shutdown result must NOT carry TTL base.close(); const result = await base.takeResult(); expect(result.type).toBe('status'); @@ -657,3 +657,222 @@ it('close resets the pending fallback directive', async () => { expect(result.state).toBe('shutdown'); expect(result.fdv1FallbackTtlMs).toBeUndefined(); }); + +it('surfaces a deferred fallback directive on the serverError path', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // Arm the deferred directive at open. + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + // A server-side FDv2 'error' event arrives before any payload/goodbye. + simulateEvent(mockEventSource, 'error', { reason: 'server error', payload_id: 'p1' }); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('interrupted'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + +it('surfaces a deferred fallback directive on the protocol error path', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + // A server-intent with no payloads yields a MISSING_PAYLOAD protocol error + // before any payload/goodbye. + simulateEvent(mockEventSource, 'server-intent', { payloads: [] }); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('interrupted'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + +it('surfaces a deferred fallback directive on the malformed-JSON path', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + // Invoke a registered listener directly with unparseable data. + const { calls } = mockEventSource.addEventListener.mock; + const listener = calls.find((c: any[]) => c[0] === 'server-intent')?.[1]; + listener({ data: 'not-valid-json{{{' }); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('interrupted'); + expect(result.errorInfo?.message).toContain('Malformed JSON'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + +it('surfaces a deferred fallback directive on the network-error path', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + // Network error with no numeric status routes through es.onerror + // (a numeric status would be handled by the error filter instead). + mockEventSource.onerror({ message: 'IO Error' }); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('interrupted'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + +it('merges a deferred fallback directive into a successful ping-triggered poll result', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const pingHandler: PingHandler = { + handlePing: jest.fn().mockResolvedValue({ + type: 'changeSet', + payload: { events: [], selector: undefined }, + fdv1Fallback: false, + }), + }; + const base = createBase(mockRequests, logger, { pingHandler }); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + const { calls } = mockEventSource.addEventListener.mock; + const pingListener = calls.find((c: any[]) => c[0] === 'ping')?.[1]; + await pingListener(); + + const result = await base.takeResult(); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + +it('surfaces a deferred fallback directive when a ping-triggered poll throws', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const pingHandler: PingHandler = { + handlePing: jest.fn().mockRejectedValue(new Error('poll failed')), + }; + const base = createBase(mockRequests, logger, { pingHandler }); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + const { calls } = mockEventSource.addEventListener.mock; + const pingListener = calls.find((c: any[]) => c[0] === 'ping')?.[1]; + await pingListener(); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('interrupted'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + +it('surfaces a deferred fallback directive on a non-retryable errorFilter error', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // Arm a directive at onopen (a prior successful connection observed the header). + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + // A later reconnect attempt fails with a non-retryable status that does NOT + // carry its own fallback header. + const willRetry = simulateErrorFilter(mockRequests, { + status: 401, + message: 'Error 401', + }); + expect(willRetry).toBe(false); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('terminal_error'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + +it('surfaces a deferred fallback directive on a retryable errorFilter error', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // Arm a directive at onopen (a prior successful connection observed the header). + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + // A later reconnect attempt fails with a retryable status that does NOT + // carry its own fallback header. + const willRetry = simulateErrorFilter(mockRequests, { + status: 500, + message: 'Error 500', + }); + expect(willRetry).toBe(true); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('interrupted'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts index 4748a29402..80c75cfff4 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts @@ -133,6 +133,28 @@ export function createStreamingBase(config: { connectionAttemptStartTime = undefined; } + /** + * Promotes a directive deferred from `onopen` (`pendingFallback`) into the + * committed `fdv1Fallback`/`fdv1FallbackTtlMs` state, clears the pending + * pair, and returns the current committed fallback state. + * + * Every path that can produce a result, a stream error, a network failure, + * or a ping-triggered poll (including one that succeeds without its own + * fallback signal), calls this instead of reading the closure variables + * directly, so a directive deferred at onopen surfaces no matter which + * path fires next. Safe to call when nothing is pending; it just returns + * the current state unchanged. + */ + function resolveFallback(): { fdv1Fallback: boolean; fdv1FallbackTtlMs: number | undefined } { + if (pendingFallback) { + fdv1Fallback = true; + fdv1FallbackTtlMs = pendingFallbackTtlMs; + pendingFallback = false; + pendingFallbackTtlMs = undefined; + } + return { fdv1Fallback, fdv1FallbackTtlMs }; + } + function handleAction(action: internal.ProtocolAction, rawData?: unknown): void { switch (action.type) { case 'payload': @@ -169,15 +191,22 @@ export function createStreamingBase(config: { break; } - case 'serverError': - resultQueue.put(interrupted(errorInfoFromUnknown(action.reason), fdv1Fallback, fdv1FallbackTtlMs)); + case 'serverError': { + const fallback = resolveFallback(); + resultQueue.put( + interrupted(errorInfoFromUnknown(action.reason), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), + ); break; + } case 'error': // Only actionable errors are queued; informational ones (UNKNOWN_EVENT) // are logged by the protocol handler. if (action.kind === 'MISSING_PAYLOAD' || action.kind === 'PROTOCOL_ERROR') { - resultQueue.put(interrupted(errorInfoFromInvalidData(action.message), fdv1Fallback, fdv1FallbackTtlMs)); + const fallback = resolveFallback(); + resultQueue.put( + interrupted(errorInfoFromInvalidData(action.message), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), + ); } break; @@ -203,17 +232,23 @@ export function createStreamingBase(config: { return false; } + const fallback = resolveFallback(); + if (!shouldRetry(err)) { config.logger?.error(httpErrorMessage(err, 'streaming request')); logConnectionResult(false); - resultQueue.put(terminalError(errorInfoFromHttpError(err.status ?? 0), fdv1Fallback, fdv1FallbackTtlMs)); + resultQueue.put( + terminalError(errorInfoFromHttpError(err.status ?? 0), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), + ); return false; } config.logger?.warn(httpErrorMessage(err, 'streaming request', 'will retry')); logConnectionResult(false); logConnectionAttempt(); - resultQueue.put(interrupted(errorInfoFromHttpError(err.status ?? 0), fdv1Fallback, fdv1FallbackTtlMs)); + resultQueue.put( + interrupted(errorInfoFromHttpError(err.status ?? 0), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), + ); return true; } @@ -243,8 +278,13 @@ export function createStreamingBase(config: { `Stream received data that was unable to be parsed in "${eventName}" message`, ); config.logger?.debug(`Data follows: ${event.data}`); + const fallback = resolveFallback(); resultQueue.put( - interrupted(errorInfoFromInvalidData('Malformed JSON in EventStream'), fdv1Fallback, fdv1FallbackTtlMs), + interrupted( + errorInfoFromInvalidData('Malformed JSON in EventStream'), + fallback.fdv1Fallback, + fallback.fdv1FallbackTtlMs, + ), ); return; } @@ -276,6 +316,16 @@ export function createStreamingBase(config: { return; } + // Trust the poll's own fallback signal (e.g. from its own HTTP + // response headers) over a directive deferred at onopen. Only + // backfill from the deferred directive when the poll result doesn't + // already indicate fallback, so an onopen directive still surfaces + // even while ping-triggered polls keep succeeding. + const fallback = resolveFallback(); + if (!result.fdv1Fallback && fallback.fdv1Fallback) { + result.fdv1Fallback = true; + result.fdv1FallbackTtlMs = fallback.fdv1FallbackTtlMs; + } resultQueue.put(result); } catch (err: any) { if (stopped) { @@ -283,11 +333,12 @@ export function createStreamingBase(config: { } config.logger?.error(`Error handling ping: ${err?.message ?? err}`); + const fallback = resolveFallback(); resultQueue.put( interrupted( errorInfoFromNetworkError(err?.message ?? 'Error during ping poll'), - fdv1Fallback, - fdv1FallbackTtlMs, + fallback.fdv1Fallback, + fallback.fdv1FallbackTtlMs, ), ); } @@ -329,8 +380,13 @@ export function createStreamingBase(config: { // This condition will be handled by the error filter. return; } + const fallback = resolveFallback(); resultQueue.put( - interrupted(errorInfoFromNetworkError(err?.message ?? 'IO Error'), fdv1Fallback, fdv1FallbackTtlMs), + interrupted( + errorInfoFromNetworkError(err?.message ?? 'IO Error'), + fallback.fdv1Fallback, + fallback.fdv1FallbackTtlMs, + ), ); }; From 17e856ccd01b4b85fd389e67bd3c85e8aaa35159 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Fri, 17 Jul 2026 17:05:42 -0400 Subject: [PATCH 05/10] fix: error responses are not propagated in RN eventsource --- .../react-native-sse/EventSource.test.ts | 145 ++++++++++++++++++ .../react-native-sse/EventSource.ts | 70 +++++++-- .../fromExternal/react-native-sse/types.ts | 9 ++ .../datasource/fdv2/StreamingFDv2Base.test.ts | 36 +++++ .../src/datasource/fdv2/StreamingFDv2Base.ts | 12 +- 5 files changed, 259 insertions(+), 13 deletions(-) diff --git a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts index 96b117bb7e..b2da264aef 100644 --- a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts +++ b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts @@ -223,4 +223,149 @@ describe('EventSource', () => { expect(onopen).toHaveBeenCalledTimes(1); }); + + test('calls retryAndHandleError with parsed response headers on an error response', () => { + const retryAndHandleError = jest.fn(() => false); + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = 'error body'; + + const es = new EventSource(uri, { logger, retryAndHandleError }); + es.onerror = jest.fn(); + + jest.runAllTimers(); + + mockXhr.readyState = 4; + mockXhr.status = 500; + mockXhr.onreadystatechange(); + + expect(retryAndHandleError).toHaveBeenCalledTimes(1); + expect(retryAndHandleError).toHaveBeenCalledWith( + expect.objectContaining({ + status: 500, + message: 'error body', + headers: expect.objectContaining({ + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': '60', + }), + }), + ); + }); + + test('dispatches error with status and headers from onprogress mid-stream', () => { + const onerror = jest.fn(); + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = 'error body'; + + eventSource.onerror = onerror; + + jest.runAllTimers(); + + // readyState 3 (LOADING), not DONE, so this exercises onprogress's + // error branch rather than onreadystatechange's. + mockXhr.readyState = 3; + mockXhr.status = 500; + mockXhr.onprogress(); + + expect(onerror).toHaveBeenCalledWith( + expect.objectContaining({ + status: 500, + headers: expect.objectContaining({ + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': '60', + }), + }), + ); + }); + + test('dispatches error with status and headers from onreadystatechange before retrying', () => { + const onerror = jest.fn(); + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = 'error body'; + + eventSource.onerror = onerror; + + jest.runAllTimers(); + + mockXhr.readyState = 4; + mockXhr.status = 500; + mockXhr.onreadystatechange(); + + expect(onerror).toHaveBeenCalledWith( + expect.objectContaining({ + status: 500, + headers: expect.objectContaining({ + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': '60', + }), + }), + ); + }); + + test('invokes retryAndHandleError from onprogress when the connection never reaches DONE', () => { + const retryAndHandleError = jest.fn(() => false); + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = 'error body'; + + const es = new EventSource(uri, { logger, retryAndHandleError }); + es.onerror = jest.fn(); + + jest.runAllTimers(); + + // readyState never advances to DONE for this attempt, so + // onreadystatechange never fires; onprogress must drive the retry + // on its own. + mockXhr.readyState = 3; + mockXhr.status = 500; + mockXhr.onprogress(); + + expect(retryAndHandleError).toHaveBeenCalledTimes(1); + expect(retryAndHandleError).toHaveBeenCalledWith( + expect.objectContaining({ + status: 500, + message: 'error body', + headers: expect.objectContaining({ + 'x-ld-fd-fallback': 'true', + 'x-ld-fd-fallback-ttl': '60', + }), + }), + ); + }); + + test('does not invoke retryAndHandleError twice when onprogress and onreadystatechange observe the same failed attempt', () => { + const retryAndHandleError = jest.fn(() => false); + + mockXhr.getAllResponseHeaders = jest.fn( + () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', + ); + mockXhr.responseText = 'error body'; + + const es = new EventSource(uri, { logger, retryAndHandleError }); + es.onerror = jest.fn(); + + jest.runAllTimers(); + + // onprogress observes the error first (LOADING), then the same response + // reaches DONE via onreadystatechange. retryAndHandleError must fire + // exactly once for this attempt despite both handlers seeing the error. + mockXhr.readyState = 3; + mockXhr.status = 500; + mockXhr.onprogress(); + + mockXhr.readyState = 4; + mockXhr.onreadystatechange(); + + expect(retryAndHandleError).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts index bbdcdfbf77..46225d34b2 100644 --- a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts +++ b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts @@ -18,6 +18,12 @@ * the stream was actually running. onprogress fires as soon as the response * starts loading, so headers are available as soon as the connection opens * instead of only at teardown. + * 3. onprogress now also drives retryAndHandleError/reconnect on an error status, + * guarded exactly-once against onreadystatechange (see isFirstErrorForThisAttempt + * below). Previously only onreadystatechange reaching DONE could trigger a retry, + * so an error response that never reached DONE - the same never-fires- + * readystatechange-while-streaming platforms note 1 above describes - would never + * recover or have its response headers (e.g. an FDv1 fallback directive) read. */ import type { EventSourceEvent, EventSourceListener, EventSourceOptions, EventType } from './types'; @@ -177,14 +183,40 @@ export default class EventSource { this._handleEvent(this._xhr.responseText || ''); } else { + const isFirstErrorForThisAttempt = this._status !== this.ERROR; this._status = this.ERROR; + const headers = this._parseResponseHeaders(this._xhr.getAllResponseHeaders()); this.dispatch('error', { type: 'error', message: this._xhr.responseText, xhrStatus: this._xhr.status, xhrState: this._xhr.readyState, + status: this._xhr.status, + headers, }); + + // A bad status observed here may never be followed by a DONE + // readystatechange: a streaming error response, or a platform + // that does not fire readystatechange reliably, leaves + // onreadystatechange silent. Retry from onprogress too, guarded + // so a later DONE for the same attempt does not trigger a second + // retry. + if (isFirstErrorForThisAttempt) { + if (!this._retryAndHandleError) { + this._tryConnect(); + } else { + const shouldRetry = this._retryAndHandleError({ + status: this._xhr.status, + message: this._xhr.responseText, + headers, + }); + + if (shouldRetry) { + this._tryConnect(); + } + } + } } }; @@ -226,30 +258,46 @@ export default class EventSource { this._tryConnect(); } } else { + // Mirrors onprogress's error-retry guard: if onprogress already + // retried for this attempt's error, this DONE for the same attempt + // must not trigger a second retry. This also relies on + // readystatechange(DONE) firing before the native onerror handler + // for the same failed request - onerror sets _status = ERROR + // without retrying, so if it ran first this guard would + // incorrectly suppress the retry below. XHR fires + // readystatechange(DONE) before error, and RN follows that + // ordering, so this holds today. + const isFirstErrorForThisAttempt = this._status !== this.ERROR; this._status = this.ERROR; + const headers = this._parseResponseHeaders(this._xhr.getAllResponseHeaders()); this.dispatch('error', { type: 'error', message: this._xhr.responseText, xhrStatus: this._xhr.status, xhrState: this._xhr.readyState, + status: this._xhr.status, + headers, }); if (this._xhr.readyState === XMLHttpRequest.DONE) { this._logger?.debug('[EventSource][onreadystatechange][ERROR] Response status error.'); - if (!this._retryAndHandleError) { - // by default just try and reconnect if there's an error. - this._tryConnect(); - } else { - // custom retry logic taking into account status codes. - const shouldRetry = this._retryAndHandleError({ - status: this._xhr.status, - message: this._xhr.responseText, - }); - - if (shouldRetry) { + if (isFirstErrorForThisAttempt) { + if (!this._retryAndHandleError) { + // by default just try and reconnect if there's an error. this._tryConnect(); + } else { + // custom retry logic taking into account status codes. + const shouldRetry = this._retryAndHandleError({ + status: this._xhr.status, + message: this._xhr.responseText, + headers, + }); + + if (shouldRetry) { + this._tryConnect(); + } } } } diff --git a/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts b/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts index df29111ea0..395d7bdc27 100644 --- a/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts +++ b/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts @@ -31,6 +31,15 @@ export interface ErrorEvent { message: string; xhrState: number; xhrStatus: number; + /** + * Present when the error corresponds to an HTTP response with a numeric + * status, as opposed to a network-level failure with no response at all. + * Consumers' `errorFilter`/`onerror` handlers rely on this to tell an + * HTTP-status error apart from a network error. + */ + status?: number; + /** Parsed response headers, present under the same condition as `status`. */ + headers?: Record; } export interface CustomEvent { diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts index e241474083..bedde98d74 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts @@ -635,6 +635,42 @@ it('clears a pending fallback directive when onopen fires without the fallback h base.close(); }); +it('clears a committed fallback after a clean reconnect on the same source', async () => { + // Once a changeSet has been stamped with fdv1Fallback=true, that state is + // committed (not just pending). A later onopen on the SAME source with no + // fallback header must clear the committed flag too, not only the pending + // one, so a stray reconnect racing ahead of the orchestrator's shutdown + // doesn't keep stamping fresh results as fallback. + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // First open carries the fallback header; applied to the next changeSet. + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }, + }); + sendFullTransfer(mockEventSource, [{ key: 'flag-a', version: 1, value: 'blue' }]); + + const fallbackResult = await base.takeResult(); + expect(fallbackResult.type).toBe('changeSet'); + if (fallbackResult.type !== 'changeSet') return; + expect(fallbackResult.fdv1Fallback).toBe(true); + + // Reconnect cleanly on the same source, with no fallback header this time. + mockEventSource.onopen({ type: 'open', headers: {} }); + sendFullTransfer(mockEventSource, [{ key: 'flag-b', version: 1, value: 'green' }]); + + const cleanResult = await base.takeResult(); + expect(cleanResult.type).toBe('changeSet'); + if (cleanResult.type !== 'changeSet') return; + expect(cleanResult.fdv1Fallback).toBe(false); + expect(cleanResult.fdv1FallbackTtlMs).toBeUndefined(); + + base.close(); +}); + it('close resets the pending fallback directive', async () => { // Calling close() must clear pendingFallback/pendingFallbackTtlMs so a // subsequently re-created source does not inherit stale state. diff --git a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts index 80c75cfff4..12479f1ca6 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts @@ -397,8 +397,16 @@ export function createStreamingBase(config: { config.logger?.info('Opened LaunchDarkly stream connection'); protocolHandler.reset(); - // Reset pending directive on every connection open so a reconnect that - // does not carry the fallback header clears any state from a prior attempt. + // Reset both the committed and pending directive on every connection + // open so a reconnect that does not carry the fallback header clears + // any state from a prior attempt. Resetting the committed flag here + // does not undo an already-emitted fallback: the result queue is + // FIFO, so a `fdv1Fallback: true` result queued before this reconnect + // is always drained before anything this connection queues after it, + // and the orchestrator closes this instance as soon as it sees that + // result - any later result from this reconnect is abandoned unread. + fdv1Fallback = false; + fdv1FallbackTtlMs = undefined; pendingFallback = false; pendingFallbackTtlMs = undefined; From 3a7388a2f50b3b58c1a749cd13295afa7703ecf9 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Mon, 20 Jul 2026 11:15:27 -0400 Subject: [PATCH 06/10] refactor: revert the changes for react-native --- .../react-native-sse/EventSource.test.ts | 235 +----------------- .../react-native-sse/EventSource.ts | 120 ++------- .../fromExternal/react-native-sse/types.ts | 10 - 3 files changed, 19 insertions(+), 346 deletions(-) diff --git a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts index b2da264aef..84fa7788d6 100644 --- a/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts +++ b/packages/sdk/react-native/__tests__/fromExternal/react-native-sse/EventSource.test.ts @@ -38,12 +38,7 @@ describe('EventSource', () => { abort: jest.fn(), }; - const xhrSpy = jest.spyOn(window, 'XMLHttpRequest').mockImplementation(() => mockXhr as XMLHttpRequest); - // Preserve static constants that EventSource reads from the constructor reference. - // @ts-ignore - xhrSpy.LOADING = 3; - // @ts-ignore - xhrSpy.DONE = 4; + jest.spyOn(window, 'XMLHttpRequest').mockImplementation(() => mockXhr as XMLHttpRequest); eventSource = new EventSource(uri, { logger }); eventSource.onclose = jest.fn(); @@ -140,232 +135,4 @@ describe('EventSource', () => { expect(mockXhr.open).toHaveBeenLastCalledWith('GET', `${uri}?basis=initial`, true); }); - - test('calls onopen with parsed response headers', () => { - const onopen = jest.fn(); - eventSource.onopen = onopen; - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = ''; - - jest.runAllTimers(); - - mockXhr.readyState = 4; - mockXhr.status = 200; - mockXhr.onreadystatechange(); - - expect(onopen).toHaveBeenCalledTimes(1); - expect(onopen).toHaveBeenCalledWith( - expect.objectContaining({ - type: 'open', - headers: expect.objectContaining({ - 'x-ld-fd-fallback': 'true', - 'x-ld-fd-fallback-ttl': '60', - }), - }), - ); - }); - - test('calls onopen with parsed response headers during streaming (onprogress, pre-DONE)', () => { - const onopen = jest.fn(); - eventSource.onopen = onopen; - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = ''; - - jest.runAllTimers(); - - // Simulate a chunk arriving while the connection is still LOADING (readyState 3). - // This is the real runtime path for a live stream: DONE (readyState 4) only - // fires once the connection closes, so gating 'open' on DONE would mean the - // event, and the fallback headers it carries, never fires during normal streaming. - mockXhr.readyState = 3; - mockXhr.status = 200; - mockXhr.onprogress(); - - expect(onopen).toHaveBeenCalledTimes(1); - expect(onopen).toHaveBeenCalledWith( - expect.objectContaining({ - type: 'open', - headers: expect.objectContaining({ - 'x-ld-fd-fallback': 'true', - 'x-ld-fd-fallback-ttl': '60', - }), - }), - ); - }); - - test('dispatches open exactly once when onprogress precedes an onreadystatechange DONE', () => { - const onopen = jest.fn(); - eventSource.onopen = onopen; - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = ''; - - jest.runAllTimers(); - - // onprogress observes the connection first (LOADING) and transitions to OPEN. - mockXhr.readyState = 3; - mockXhr.status = 200; - mockXhr.onprogress(); - - // A later onreadystatechange at DONE must not re-dispatch open: the status is - // already OPEN (no longer CONNECTING), so its open-dispatch guard is a no-op. - mockXhr.readyState = 4; - mockXhr.status = 200; - mockXhr.onreadystatechange(); - - expect(onopen).toHaveBeenCalledTimes(1); - }); - - test('calls retryAndHandleError with parsed response headers on an error response', () => { - const retryAndHandleError = jest.fn(() => false); - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = 'error body'; - - const es = new EventSource(uri, { logger, retryAndHandleError }); - es.onerror = jest.fn(); - - jest.runAllTimers(); - - mockXhr.readyState = 4; - mockXhr.status = 500; - mockXhr.onreadystatechange(); - - expect(retryAndHandleError).toHaveBeenCalledTimes(1); - expect(retryAndHandleError).toHaveBeenCalledWith( - expect.objectContaining({ - status: 500, - message: 'error body', - headers: expect.objectContaining({ - 'x-ld-fd-fallback': 'true', - 'x-ld-fd-fallback-ttl': '60', - }), - }), - ); - }); - - test('dispatches error with status and headers from onprogress mid-stream', () => { - const onerror = jest.fn(); - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = 'error body'; - - eventSource.onerror = onerror; - - jest.runAllTimers(); - - // readyState 3 (LOADING), not DONE, so this exercises onprogress's - // error branch rather than onreadystatechange's. - mockXhr.readyState = 3; - mockXhr.status = 500; - mockXhr.onprogress(); - - expect(onerror).toHaveBeenCalledWith( - expect.objectContaining({ - status: 500, - headers: expect.objectContaining({ - 'x-ld-fd-fallback': 'true', - 'x-ld-fd-fallback-ttl': '60', - }), - }), - ); - }); - - test('dispatches error with status and headers from onreadystatechange before retrying', () => { - const onerror = jest.fn(); - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = 'error body'; - - eventSource.onerror = onerror; - - jest.runAllTimers(); - - mockXhr.readyState = 4; - mockXhr.status = 500; - mockXhr.onreadystatechange(); - - expect(onerror).toHaveBeenCalledWith( - expect.objectContaining({ - status: 500, - headers: expect.objectContaining({ - 'x-ld-fd-fallback': 'true', - 'x-ld-fd-fallback-ttl': '60', - }), - }), - ); - }); - - test('invokes retryAndHandleError from onprogress when the connection never reaches DONE', () => { - const retryAndHandleError = jest.fn(() => false); - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = 'error body'; - - const es = new EventSource(uri, { logger, retryAndHandleError }); - es.onerror = jest.fn(); - - jest.runAllTimers(); - - // readyState never advances to DONE for this attempt, so - // onreadystatechange never fires; onprogress must drive the retry - // on its own. - mockXhr.readyState = 3; - mockXhr.status = 500; - mockXhr.onprogress(); - - expect(retryAndHandleError).toHaveBeenCalledTimes(1); - expect(retryAndHandleError).toHaveBeenCalledWith( - expect.objectContaining({ - status: 500, - message: 'error body', - headers: expect.objectContaining({ - 'x-ld-fd-fallback': 'true', - 'x-ld-fd-fallback-ttl': '60', - }), - }), - ); - }); - - test('does not invoke retryAndHandleError twice when onprogress and onreadystatechange observe the same failed attempt', () => { - const retryAndHandleError = jest.fn(() => false); - - mockXhr.getAllResponseHeaders = jest.fn( - () => 'X-Ld-Fd-Fallback: true\r\nX-Ld-Fd-Fallback-Ttl: 60\r\nContent-Type: text/event-stream', - ); - mockXhr.responseText = 'error body'; - - const es = new EventSource(uri, { logger, retryAndHandleError }); - es.onerror = jest.fn(); - - jest.runAllTimers(); - - // onprogress observes the error first (LOADING), then the same response - // reaches DONE via onreadystatechange. retryAndHandleError must fire - // exactly once for this attempt despite both handlers seeing the error. - mockXhr.readyState = 3; - mockXhr.status = 500; - mockXhr.onprogress(); - - mockXhr.readyState = 4; - mockXhr.onreadystatechange(); - - expect(retryAndHandleError).toHaveBeenCalledTimes(1); - }); }); diff --git a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts index 46225d34b2..1d04733303 100644 --- a/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts +++ b/packages/sdk/react-native/src/fromExternal/react-native-sse/EventSource.ts @@ -7,23 +7,10 @@ * 4. replaced all for of loops with foreach * * Additional changes: - * 1. separated event handling to use onprogress for data changes and - * onreadystatechange for status changes, since some platforms (e.g. Vega OS) - * never fire a readystatechange event while the response is still streaming in. - * 2. onprogress now also performs the CONNECTING -> OPEN transition and dispatches - * open, guarded by the same this._status === CONNECTING check onreadystatechange - * already used. Previously open only fired from onreadystatechange at - * readyState DONE, which for a long-lived stream only happens at connection - * close, so the parsed response headers carried on open were never seen while - * the stream was actually running. onprogress fires as soon as the response - * starts loading, so headers are available as soon as the connection opens - * instead of only at teardown. - * 3. onprogress now also drives retryAndHandleError/reconnect on an error status, - * guarded exactly-once against onreadystatechange (see isFirstErrorForThisAttempt - * below). Previously only onreadystatechange reaching DONE could trigger a retry, - * so an error response that never reached DONE - the same never-fires- - * readystatechange-while-streaming platforms note 1 above describes - would never - * recover or have its response headers (e.g. an FDv1 fallback directive) read. + * 1. separated event handling to use onprogress for data changes + * and onreadystatechange for status changes. This is to address + * an issue with Vega OS where they do not fire a readyStatechange + * event when the response is received. */ import type { EventSourceEvent, EventSourceListener, EventSourceOptions, EventType } from './types'; @@ -171,52 +158,16 @@ export default class EventSource { ); if (this._xhr.status >= 200 && this._xhr.status < 400) { - if (this._status === this.CONNECTING) { - this._retryCount = 0; - this._status = this.OPEN; - this.dispatch('open', { - type: 'open', - headers: this._parseResponseHeaders(this._xhr.getAllResponseHeaders()), - }); - this._logger?.debug('[EventSource][onprogress][OPEN] Connection opened.'); - } - this._handleEvent(this._xhr.responseText || ''); } else { - const isFirstErrorForThisAttempt = this._status !== this.ERROR; this._status = this.ERROR; - const headers = this._parseResponseHeaders(this._xhr.getAllResponseHeaders()); this.dispatch('error', { type: 'error', message: this._xhr.responseText, xhrStatus: this._xhr.status, xhrState: this._xhr.readyState, - status: this._xhr.status, - headers, }); - - // A bad status observed here may never be followed by a DONE - // readystatechange: a streaming error response, or a platform - // that does not fire readystatechange reliably, leaves - // onreadystatechange silent. Retry from onprogress too, guarded - // so a later DONE for the same attempt does not trigger a second - // retry. - if (isFirstErrorForThisAttempt) { - if (!this._retryAndHandleError) { - this._tryConnect(); - } else { - const shouldRetry = this._retryAndHandleError({ - status: this._xhr.status, - message: this._xhr.responseText, - headers, - }); - - if (shouldRetry) { - this._tryConnect(); - } - } - } } }; @@ -243,10 +194,7 @@ export default class EventSource { if (this._status === this.CONNECTING) { this._retryCount = 0; this._status = this.OPEN; - this.dispatch('open', { - type: 'open', - headers: this._parseResponseHeaders(this._xhr.getAllResponseHeaders()), - }); + this.dispatch('open', { type: 'open' }); this._logger?.debug('[EventSource][onreadystatechange][OPEN] Connection opened.'); } @@ -258,46 +206,30 @@ export default class EventSource { this._tryConnect(); } } else { - // Mirrors onprogress's error-retry guard: if onprogress already - // retried for this attempt's error, this DONE for the same attempt - // must not trigger a second retry. This also relies on - // readystatechange(DONE) firing before the native onerror handler - // for the same failed request - onerror sets _status = ERROR - // without retrying, so if it ran first this guard would - // incorrectly suppress the retry below. XHR fires - // readystatechange(DONE) before error, and RN follows that - // ordering, so this holds today. - const isFirstErrorForThisAttempt = this._status !== this.ERROR; this._status = this.ERROR; - const headers = this._parseResponseHeaders(this._xhr.getAllResponseHeaders()); this.dispatch('error', { type: 'error', message: this._xhr.responseText, xhrStatus: this._xhr.status, xhrState: this._xhr.readyState, - status: this._xhr.status, - headers, }); if (this._xhr.readyState === XMLHttpRequest.DONE) { this._logger?.debug('[EventSource][onreadystatechange][ERROR] Response status error.'); - if (isFirstErrorForThisAttempt) { - if (!this._retryAndHandleError) { - // by default just try and reconnect if there's an error. + if (!this._retryAndHandleError) { + // by default just try and reconnect if there's an error. + this._tryConnect(); + } else { + // custom retry logic taking into account status codes. + const shouldRetry = this._retryAndHandleError({ + status: this._xhr.status, + message: this._xhr.responseText, + }); + + if (shouldRetry) { this._tryConnect(); - } else { - // custom retry logic taking into account status codes. - const shouldRetry = this._retryAndHandleError({ - status: this._xhr.status, - message: this._xhr.responseText, - headers, - }); - - if (shouldRetry) { - this._tryConnect(); - } } } } @@ -421,28 +353,12 @@ export default class EventSource { } } - private _parseResponseHeaders(raw: string | null): Record { - const result: Record = {}; - if (!raw) { - return result; - } - raw.split('\r\n').forEach((line) => { - const separatorIndex = line.indexOf(': '); - if (separatorIndex > 0) { - const key = line.slice(0, separatorIndex).toLowerCase(); - const value = line.slice(separatorIndex + 2); - result[key] = value; - } - }); - return result; - } - dispatch>(type: T, data: EventSourceEvent) { this._eventHandlers[type]?.forEach((handler: EventSourceListener) => handler(data)); switch (type) { case 'open': - this.onopen(data); + this.onopen(); break; case 'close': this.onclose(); @@ -473,7 +389,7 @@ export default class EventSource { return this._status; } - onopen(_e: any) {} + onopen() {} onclose() {} onerror(_err: any) {} onretrying(_e: any) {} diff --git a/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts b/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts index 395d7bdc27..465b150fdf 100644 --- a/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts +++ b/packages/sdk/react-native/src/fromExternal/react-native-sse/types.ts @@ -10,7 +10,6 @@ export interface MessageEvent { export interface OpenEvent { type: 'open'; - headers?: Record; } export interface CloseEvent { @@ -31,15 +30,6 @@ export interface ErrorEvent { message: string; xhrState: number; xhrStatus: number; - /** - * Present when the error corresponds to an HTTP response with a numeric - * status, as opposed to a network-level failure with no response at all. - * Consumers' `errorFilter`/`onerror` handlers rely on this to tell an - * HTTP-status error apart from a network error. - */ - status?: number; - /** Parsed response headers, present under the same condition as `status`. */ - headers?: Record; } export interface CustomEvent { From ff0a597850d7252eeb9865689bc86c129dc214e4 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Mon, 20 Jul 2026 12:37:48 -0400 Subject: [PATCH 07/10] chore: bot comment --- .../datasource/fdv2/StreamingFDv2Base.test.ts | 29 +++++++++++++++++++ .../src/datasource/fdv2/StreamingFDv2Base.ts | 11 ++++--- 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts index bedde98d74..c50a3208e4 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts @@ -825,6 +825,35 @@ it('merges a deferred fallback directive into a successful ping-triggered poll r base.close(); }); +it('backfills the deferred TTL into a ping-triggered poll result that already signals fallback without one', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const pingHandler: PingHandler = { + handlePing: jest.fn().mockResolvedValue({ + type: 'changeSet', + payload: { events: [], selector: undefined }, + fdv1Fallback: true, + }), + }; + const base = createBase(mockRequests, logger, { pingHandler }); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + const { calls } = mockEventSource.addEventListener.mock; + const pingListener = calls.find((c: any[]) => c[0] === 'ping')?.[1]; + await pingListener(); + + const result = await base.takeResult(); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + + base.close(); +}); + it('surfaces a deferred fallback directive when a ping-triggered poll throws', async () => { const mockEventSource = createMockEventSource(); const mockRequests = createMockRequests(mockEventSource); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts index 12479f1ca6..204fa31711 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts @@ -317,14 +317,17 @@ export function createStreamingBase(config: { } // Trust the poll's own fallback signal (e.g. from its own HTTP - // response headers) over a directive deferred at onopen. Only - // backfill from the deferred directive when the poll result doesn't - // already indicate fallback, so an onopen directive still surfaces - // even while ping-triggered polls keep succeeding. + // response headers) over a directive deferred at onopen, so an + // onopen directive still surfaces even while ping-triggered polls + // keep succeeding. Still backfill the TTL alone when the poll's own + // fallback signal did not carry one, so a deferred TTL is not lost + // just because the poll also happened to indicate fallback. const fallback = resolveFallback(); if (!result.fdv1Fallback && fallback.fdv1Fallback) { result.fdv1Fallback = true; result.fdv1FallbackTtlMs = fallback.fdv1FallbackTtlMs; + } else if (result.fdv1Fallback && result.fdv1FallbackTtlMs === undefined) { + result.fdv1FallbackTtlMs = fallback.fdv1FallbackTtlMs; } resultQueue.put(result); } catch (err: any) { From 9d6f7f9a2f786740ffa009f1f441693c954a08d6 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Tue, 21 Jul 2026 13:49:00 -0400 Subject: [PATCH 08/10] chore: pr comments --- .../datasource/fdv2/Conditions.test.ts | 15 +- .../datasource/fdv2/FDv2DataSource.test.ts | 126 +++++++------- .../datasource/fdv2/FDv2SourceResult.test.ts | 14 +- .../datasource/fdv2/StreamingFDv2Base.test.ts | 65 +++++++ .../src/datasource/fdv2/CacheInitializer.ts | 8 +- .../fdv2/FDv1PollingSynchronizer.ts | 12 +- .../src/datasource/fdv2/FDv2SourceResult.ts | 62 ++++--- .../src/datasource/fdv2/PollingBase.ts | 49 +++--- .../src/datasource/fdv2/PollingInitializer.ts | 4 +- .../src/datasource/fdv2/StreamingFDv2Base.ts | 164 ++++++++---------- 10 files changed, 302 insertions(+), 217 deletions(-) diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts index 7710e20fa9..193f4fa81e 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/Conditions.test.ts @@ -33,12 +33,15 @@ function raceTimeout(promise: Promise, ms: number): Promise { condition.inform( terminalError( { kind: DataSourceErrorKind.ErrorResponse, message: 'unauthorized', time: Date.now() }, - false, + { fdv1Fallback: false }, ), ); expect(await raceTimeout(condition.promise, 50)).toBe(DID_NOT_RESOLVE); @@ -117,14 +120,14 @@ it('fallback condition does not start timer on shutdown', async () => { it('fallback condition does not start timer on goodbye', async () => { const condition = createFallbackCondition(10); - condition.inform(goodbye('server-requested', false)); + condition.inform(goodbye('server-requested', { fdv1Fallback: false })); expect(await raceTimeout(condition.promise, 50)).toBe(DID_NOT_RESOLVE); condition.close(); }); it('fallback condition changeSet without active timer does not cause issues', async () => { const condition = createFallbackCondition(10); - // changeSet without a prior interrupted — should be safe + // changeSet without a prior interrupted, should be safe condition.inform(makeChangeSet()); expect(await raceTimeout(condition.promise, 50)).toBe(DID_NOT_RESOLVE); condition.close(); @@ -165,7 +168,7 @@ it('recovery condition does not fire after close', async () => { it('recovery condition close after timer fires does not cause error', async () => { const condition = createRecoveryCondition(10); expect(await condition.promise).toBe('recovery'); - // Close after timer already fired — should not throw + // Close after timer already fired, should not throw condition.close(); }); diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts index 896d93596b..de42df00cc 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts @@ -33,7 +33,7 @@ it('resolves start() when initializer returns changeSet with selector', async () const payload = makePayload({ state: 'my-selector' }); const ds = createFDv2DataSource({ - initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(payload, false)))], + initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(payload, { fdv1Fallback: false })))], synchronizerSlots: [], dataCallback, statusManager, @@ -56,8 +56,8 @@ it('continues to next initializer when changeSet has no selector', async () => { const ds = createFDv2DataSource({ initializerFactories: [ - makeInitFactory(makeMockInitializer(changeSet(payloadNoSelector, false))), - makeInitFactory(makeMockInitializer(changeSet(payloadWithSelector, false))), + makeInitFactory(makeMockInitializer(changeSet(payloadNoSelector, { fdv1Fallback: false }))), + makeInitFactory(makeMockInitializer(changeSet(payloadWithSelector, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -78,7 +78,7 @@ it('resolves start() when all initializers exhausted but data was received', asy const ds = createFDv2DataSource({ initializerFactories: [ - makeInitFactory(makeMockInitializer(changeSet(payloadNoSelector, false))), + makeInitFactory(makeMockInitializer(changeSet(payloadNoSelector, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -101,8 +101,8 @@ it('skips transfer-none initializer results without calling dataCallback', async const ds = createFDv2DataSource({ initializerFactories: [ - makeInitFactory(makeMockInitializer(changeSet(nonePayload, false))), - makeInitFactory(makeMockInitializer(changeSet(fullPayload, false))), + makeInitFactory(makeMockInitializer(changeSet(nonePayload, { fdv1Fallback: false }))), + makeInitFactory(makeMockInitializer(changeSet(fullPayload, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -123,7 +123,7 @@ it('does not mark data received for transfer-none initializer results', async () const nonePayload = makePayload({ type: 'none' }); const ds = createFDv2DataSource({ - initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(nonePayload, false)))], + initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(nonePayload, { fdv1Fallback: false })))], synchronizerSlots: [], dataCallback, statusManager, @@ -142,7 +142,7 @@ it('resolves start() when only initializer is a cache initializer that returns t const ds = createFDv2DataSource({ initializerFactories: [ - makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, false))), + makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -170,8 +170,8 @@ it('does not overwrite an error status when a later initializer fails after data const ds = createFDv2DataSource({ initializerFactories: [ - makeInitFactory(makeMockInitializer(changeSet(payloadNoSelector, false))), - makeInitFactory(makeMockInitializer(interrupted(makeErrorInfo(), false))), + makeInitFactory(makeMockInitializer(changeSet(payloadNoSelector, { fdv1Fallback: false }))), + makeInitFactory(makeMockInitializer(interrupted(makeErrorInfo(), { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -203,7 +203,7 @@ it('rejects start() when close() is called before cache-only initialization runs const ds = createFDv2DataSource({ initializerFactories: [ - makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, false))), + makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -231,7 +231,7 @@ it('does not overwrite error status when a cache-only initializer reports interr const ds = createFDv2DataSource({ initializerFactories: [ - makeCacheInitFactory(makeMockInitializer(interrupted(makeErrorInfo(), false))), + makeCacheInitFactory(makeMockInitializer(interrupted(makeErrorInfo(), { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -260,8 +260,8 @@ it('rejects when a cache initializer is followed by a non-cache initializer and const ds = createFDv2DataSource({ initializerFactories: [ - makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, false))), - makeInitFactory(makeMockInitializer(changeSet(nonePayload, false))), + makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, { fdv1Fallback: false }))), + makeInitFactory(makeMockInitializer(changeSet(nonePayload, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -279,14 +279,14 @@ it('rejects when a cache initializer returns transfer-none but synchronizers exi const statusManager = makeStatusManager(); const nonePayload = makePayload({ type: 'none' }); - // A synchronizer that produces a terminal error immediately -- no data will - // be delivered, so the orchestrator exhausts all sources and rejects. - const sync = makeMockSynchronizer([terminalError(makeErrorInfo(), false)]); + // A synchronizer that produces a terminal error immediately, so no data is + // delivered and the orchestrator exhausts all sources and rejects. + const sync = makeMockSynchronizer([terminalError(makeErrorInfo(), { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; const ds = createFDv2DataSource({ initializerFactories: [ - makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, false))), + makeCacheInitFactory(makeMockInitializer(changeSet(nonePayload, { fdv1Fallback: false }))), ], synchronizerSlots: slots, dataCallback, @@ -306,8 +306,8 @@ it('continues past initializer errors', async () => { const ds = createFDv2DataSource({ initializerFactories: [ - makeInitFactory(makeMockInitializer(interrupted(makeErrorInfo(), false))), - makeInitFactory(makeMockInitializer(changeSet(payload, false))), + makeInitFactory(makeMockInitializer(interrupted(makeErrorInfo(), { fdv1Fallback: false }))), + makeInitFactory(makeMockInitializer(changeSet(payload, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -331,8 +331,8 @@ it('continues past terminal errors in initializers', async () => { const ds = createFDv2DataSource({ initializerFactories: [ - makeInitFactory(makeMockInitializer(terminalError(makeErrorInfo(), false))), - makeInitFactory(makeMockInitializer(changeSet(payload, false))), + makeInitFactory(makeMockInitializer(terminalError(makeErrorInfo(), { fdv1Fallback: false }))), + makeInitFactory(makeMockInitializer(changeSet(payload, { fdv1Fallback: false }))), ], synchronizerSlots: [], dataCallback, @@ -350,7 +350,7 @@ it('skips to synchronizers when no initializers are configured', async () => { const statusManager = makeStatusManager(); const payload = makePayload({ state: 'selector' }); - const sync = makeMockSynchronizer([changeSet(payload, false)]); + const sync = makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; const ds = createFDv2DataSource({ @@ -374,7 +374,7 @@ it('delivers changeSet from synchronizer to callback', async () => { const statusManager = makeStatusManager(); const payload = makePayload({ state: 'sync-selector' }); - const sync = makeMockSynchronizer([changeSet(payload, false)]); + const sync = makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; const ds = createFDv2DataSource({ @@ -398,8 +398,8 @@ it('blocks synchronizer on terminal error and moves to next', async () => { const logger = makeLogger(); const payload = makePayload({ state: 'selector' }); - const sync1 = makeMockSynchronizer([terminalError(makeErrorInfo(), false)]); - const sync2 = makeMockSynchronizer([changeSet(payload, false)]); + const sync1 = makeMockSynchronizer([terminalError(makeErrorInfo(), { fdv1Fallback: false })]); + const sync2 = makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [ createSynchronizerSlot({ create: () => sync1 }), @@ -429,8 +429,8 @@ it('continues on interrupted results from synchronizer', async () => { const payload = makePayload({ state: 'selector' }); const sync = makeMockSynchronizer([ - interrupted(makeErrorInfo(), false), - changeSet(payload, false), + interrupted(makeErrorInfo(), { fdv1Fallback: false }), + changeSet(payload, { fdv1Fallback: false }), ]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; @@ -454,7 +454,7 @@ it('continues on goodbye results from synchronizer', async () => { const statusManager = makeStatusManager(); const payload = makePayload({ state: 'selector' }); - const sync = makeMockSynchronizer([goodbye('reconnect', false), changeSet(payload, false)]); + const sync = makeMockSynchronizer([goodbye('reconnect', { fdv1Fallback: false }), changeSet(payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; const ds = createFDv2DataSource({ @@ -475,7 +475,7 @@ it('rejects start() when all synchronizers are exhausted without data', async () const dataCallback = jest.fn(); const statusManager = makeStatusManager(); - const sync = makeMockSynchronizer([terminalError(makeErrorInfo(), false)]); + const sync = makeMockSynchronizer([terminalError(makeErrorInfo(), { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; const ds = createFDv2DataSource({ @@ -499,8 +499,8 @@ it('triggers fdv1 fallback when synchronizer changeSet has fdv1Fallback flag', a const fdv2Payload = makePayload({ state: 'selector' }); const fdv1Payload = makePayload({ state: 'fdv1-selector' }); - const fdv2Sync = makeMockSynchronizer([changeSet(fdv2Payload, true)]); - const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, false)]); + const fdv2Sync = makeMockSynchronizer([changeSet(fdv2Payload, { fdv1Fallback: true })]); + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [ createSynchronizerSlot({ create: () => fdv2Sync }), @@ -533,8 +533,8 @@ it('triggers fdv1 fallback on terminal error with fdv1Fallback flag', async () = const fdv1Payload = makePayload({ state: 'fdv1-selector' }); - const fdv2Sync = makeMockSynchronizer([terminalError(makeErrorInfo(), true)]); - const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, false)]); + const fdv2Sync = makeMockSynchronizer([terminalError(makeErrorInfo(), { fdv1Fallback: true })]); + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [ createSynchronizerSlot({ create: () => fdv2Sync }), @@ -571,7 +571,7 @@ it('falls back to next synchronizer when fallback condition fires', async () => const sync1: Synchronizer = { next: jest .fn, []>() - .mockResolvedValueOnce(interrupted(makeErrorInfo(), false)) + .mockResolvedValueOnce(interrupted(makeErrorInfo(), { fdv1Fallback: false })) .mockReturnValue( new Promise((resolve) => { sync1NextResolve = resolve; @@ -582,7 +582,7 @@ it('falls back to next synchronizer when fallback condition fires', async () => }, }; - const sync2 = makeMockSynchronizer([changeSet(payload, false)]); + const sync2 = makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [ createSynchronizerSlot({ create: () => sync1 }), @@ -628,7 +628,7 @@ it('recovers to primary synchronizer when recovery condition fires', async () => return { next: jest .fn, []>() - .mockResolvedValueOnce(interrupted(makeErrorInfo(), false)) + .mockResolvedValueOnce(interrupted(makeErrorInfo(), { fdv1Fallback: false })) .mockReturnValue( new Promise((resolve) => { blockResolve = resolve; @@ -640,7 +640,7 @@ it('recovers to primary synchronizer when recovery condition fires', async () => } as Synchronizer; } // Second invocation (after recovery): sends data - return makeMockSynchronizer([changeSet(payload, false)]); + return makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })]); }; // sync2: blocks immediately (just waits) @@ -672,8 +672,8 @@ it('recovers to primary synchronizer when recovery condition fires', async () => recoveryTimeoutMs: 20, }); - // start() resolves after: sync1 interrupted → fallback (10ms) → sync2 blocks → - // recovery (20ms) → sync1 second invocation delivers changeSet. + // start() resolves once sync1 interrupts, fallback (10ms) moves to sync2, + // recovery (20ms) moves back, and sync1's second invocation delivers the changeSet. await ds.start(); expect(logger.info).toHaveBeenCalledWith(expect.stringContaining('Recovery condition fired')); @@ -728,7 +728,7 @@ it('close during synchronization causes exit', async () => { const sync: Synchronizer = { next: jest .fn, []>() - .mockResolvedValueOnce(changeSet(payload, false)) + .mockResolvedValueOnce(changeSet(payload, { fdv1Fallback: false })) .mockReturnValue( new Promise((resolve) => { pendingResolve = resolve; @@ -767,7 +767,7 @@ it('passes selectorGetter from config through to source factories', async () => const selectorGetter = jest.fn(() => 'test-selector'); const payload = makePayload({ state: 'selector' }); - const syncFactory = jest.fn(() => makeMockSynchronizer([changeSet(payload, false)])); + const syncFactory = jest.fn(() => makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })])); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: syncFactory })]; const ds = createFDv2DataSource({ @@ -811,7 +811,7 @@ it('resolves with initializer data even when no synchronizers exist', async () = const payload = makePayload({ state: 'selector' }); const ds = createFDv2DataSource({ - initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(payload, false)))], + initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(payload, { fdv1Fallback: false })))], synchronizerSlots: [], dataCallback, statusManager, @@ -831,10 +831,10 @@ it('shutdown result from synchronizer exits without moving to next', async () => const statusManager = makeStatusManager(); const payload = makePayload({ state: 'selector' }); - const secondSyncFactory = jest.fn(() => makeMockSynchronizer([changeSet(payload, false)])); + const secondSyncFactory = jest.fn(() => makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })])); const slots: SynchronizerSlot[] = [ createSynchronizerSlot({ - create: () => makeMockSynchronizer([changeSet(payload, false), shutdown()]), + create: () => makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false }), shutdown()]), }), createSynchronizerSlot({ create: secondSyncFactory }), ]; @@ -849,7 +849,7 @@ it('shutdown result from synchronizer exits without moving to next', async () => await ds.start(); - // Wait for the shutdown to be processed — the second synchronizer should not be created. + // Wait for the shutdown to be processed before asserting the second synchronizer was never created. await statusManager.waitForState('VALID', 1); expect(dataCallback).toHaveBeenCalledTimes(1); expect(secondSyncFactory).not.toHaveBeenCalled(); @@ -866,9 +866,9 @@ it('delivers multiple changeSets from synchronizer in order', async () => { const payload3 = makePayload({ state: 'selector-3' }); const sync = makeMockSynchronizer([ - changeSet(payload1, false), - changeSet(payload2, false), - changeSet(payload3, false), + changeSet(payload1, { fdv1Fallback: false }), + changeSet(payload2, { fdv1Fallback: false }), + changeSet(payload3, { fdv1Fallback: false }), ]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; @@ -900,12 +900,12 @@ it('first initializer with selector prevents second initializer from running', a const payload = makePayload({ state: 'good-selector' }); const secondInitCreate = jest.fn(() => - makeMockInitializer(changeSet(makePayload({ state: 'second' }), false)), + makeMockInitializer(changeSet(makePayload({ state: 'second' }), { fdv1Fallback: false })), ); const ds = createFDv2DataSource({ initializerFactories: [ - makeInitFactory(makeMockInitializer(changeSet(payload, false))), + makeInitFactory(makeMockInitializer(changeSet(payload, { fdv1Fallback: false }))), { create: secondInitCreate }, ], synchronizerSlots: [], @@ -930,7 +930,7 @@ it('multiple close calls do not throw', async () => { const payload = makePayload({ state: 'selector' }); const ds = createFDv2DataSource({ - initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(payload, false)))], + initializerFactories: [makeInitFactory(makeMockInitializer(changeSet(payload, { fdv1Fallback: false })))], synchronizerSlots: [], dataCallback, statusManager, @@ -951,13 +951,13 @@ it('close during condition waiting exits cleanly', async () => { const dataCallback = jest.fn(); const statusManager = makeStatusManager(); - // Sync sends changeSet then interrupted, then blocks — condition timer starts + // Sync sends changeSet then interrupted, then blocks; the interrupted result starts the condition timer let pendingResolve: ((r: FDv2SourceResult) => void) | undefined; const sync: Synchronizer = { next: jest .fn, []>() - .mockResolvedValueOnce(changeSet(makePayload({ state: 'selector' }), false)) - .mockResolvedValueOnce(interrupted(makeErrorInfo(), false)) + .mockResolvedValueOnce(changeSet(makePayload({ state: 'selector' }), { fdv1Fallback: false })) + .mockResolvedValueOnce(interrupted(makeErrorInfo(), { fdv1Fallback: false })) .mockReturnValue( new Promise((resolve) => { pendingResolve = resolve; @@ -984,7 +984,7 @@ it('close during condition waiting exits cleanly', async () => { await ds.start(); - // Sync loop is running, condition timer is active — close should not hang. + // Sync loop is running and the condition timer is active; close should not hang. ds.close(); }); @@ -998,7 +998,7 @@ it('fdv1 fallback not triggered when fdv1Fallback flag is absent', async () => { const fdv1Factory = jest.fn(() => makeMockSynchronizer([])); const slots: SynchronizerSlot[] = [ - createSynchronizerSlot({ create: () => makeMockSynchronizer([changeSet(payload, false)]) }), + createSynchronizerSlot({ create: () => makeMockSynchronizer([changeSet(payload, { fdv1Fallback: false })]) }), createSynchronizerSlot({ create: fdv1Factory }, { isFDv1Fallback: true }), ]; @@ -1027,11 +1027,11 @@ it('fdv1 fallback blocks other synchronizers', async () => { const slots: SynchronizerSlot[] = [ createSynchronizerSlot({ - create: () => makeMockSynchronizer([changeSet(fdv2Payload, true)]), + create: () => makeMockSynchronizer([changeSet(fdv2Payload, { fdv1Fallback: true })]), }), createSynchronizerSlot({ create: secondSyncFactory }), createSynchronizerSlot( - { create: () => makeMockSynchronizer([changeSet(fdv1Payload, false)]) }, + { create: () => makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]) }, { isFDv1Fallback: true }, ), ]; @@ -1049,7 +1049,7 @@ it('fdv1 fallback blocks other synchronizers', async () => { // Wait for fdv1 synchronizer to deliver its changeSet (second VALID). await statusManager.waitForState('VALID', 2); - // FDv1 fallback should block non-FDv1 synchronizers — second sync should not be called + // FDv1 fallback blocks non-FDv1 synchronizers: second sync should not be called expect(secondSyncFactory).not.toHaveBeenCalled(); expect(dataCallback).toHaveBeenCalledWith(fdv1Payload); ds.close(); @@ -1061,7 +1061,7 @@ it('fdv1 fallback ignored when no FDv1 synchronizer is configured', async () => const payload = makePayload({ state: 'selector' }); // Synchronizer sends changeSet with fdv1Fallback flag but no FDv1 slot exists - const sync = makeMockSynchronizer([changeSet(payload, true)]); + const sync = makeMockSynchronizer([changeSet(payload, { fdv1Fallback: true })]); const slots: SynchronizerSlot[] = [createSynchronizerSlot({ create: () => sync })]; const ds = createFDv2DataSource({ @@ -1084,8 +1084,8 @@ it('fdv1 fallback triggered on interrupted result with fdv1Fallback flag', async const statusManager = makeStatusManager(); const fdv1Payload = makePayload({ state: 'fdv1-selector' }); - const fdv2Sync = makeMockSynchronizer([interrupted(makeErrorInfo(), true)]); - const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, false)]); + const fdv2Sync = makeMockSynchronizer([interrupted(makeErrorInfo(), { fdv1Fallback: true })]); + const fdv1Sync = makeMockSynchronizer([changeSet(fdv1Payload, { fdv1Fallback: false })]); const slots: SynchronizerSlot[] = [ createSynchronizerSlot({ create: () => fdv2Sync }), 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 3d828c9046..0ec1648fe5 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2SourceResult.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2SourceResult.test.ts @@ -20,7 +20,7 @@ it('creates a changeSet result with a payload', () => { type: 'full' as const, updates: [], }; - const result = changeSet(payload, false, 'env-123'); + const result = changeSet(payload, { fdv1Fallback: false }, 'env-123'); expect(result.type).toBe('changeSet'); expect(result).toEqual({ @@ -33,7 +33,7 @@ it('creates a changeSet result with a payload', () => { it('creates a changeSet result with fdv1Fallback flag', () => { const payload = { version: 1, type: 'full' as const, updates: [] }; - const result = changeSet(payload, true); + const result = changeSet(payload, { fdv1Fallback: true }); expect(result.type).toBe('changeSet'); if (result.type === 'changeSet') { @@ -47,7 +47,7 @@ it('creates an interrupted status result', () => { message: 'connection reset', time: 1000, }; - const result = interrupted(errorInfo, false); + const result = interrupted(errorInfo, { fdv1Fallback: false }); expect(result).toEqual({ type: 'status', @@ -74,7 +74,7 @@ it('creates a terminal error status result', () => { statusCode: 401, time: 2000, }; - const result = terminalError(errorInfo, true); + const result = terminalError(errorInfo, { fdv1Fallback: true }); expect(result).toEqual({ type: 'status', @@ -85,7 +85,7 @@ it('creates a terminal error status result', () => { }); it('creates a goodbye status result', () => { - const result = goodbye('server-shutdown', false); + const result = goodbye('server-shutdown', { fdv1Fallback: false }); expect(result).toEqual({ type: 'status', @@ -96,7 +96,7 @@ it('creates a goodbye status result', () => { }); it('creates a goodbye status result with fdv1Fallback and a TTL', () => { - const result = goodbye('server-shutdown', true, 5000); + const result = goodbye('server-shutdown', { fdv1Fallback: true, fdv1FallbackTtlMs: 5000 }); expect(result).toEqual({ type: 'status', @@ -108,7 +108,7 @@ it('creates a goodbye status result with fdv1Fallback and a TTL', () => { }); it('creates a goodbye status result with TTL 0 (indefinite fallback)', () => { - const result = goodbye('server-shutdown', true, 0); + const result = goodbye('server-shutdown', { fdv1Fallback: true, fdv1FallbackTtlMs: 0 }); expect(result).toEqual({ type: 'status', diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts index c50a3208e4..3f1b17028b 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/StreamingFDv2Base.test.ts @@ -608,6 +608,37 @@ it('emits terminal_error for a goodbye when a deferred open directive is pending base.close(); }); +it('prefers the in-band goodbye directive TTL over a pending onopen-deferred TTL', async () => { + // When onopen defers a fallback directive AND the goodbye event itself carries + // its own protocolFallbackTTL, the in-band directive wins: it uses its own TTL + // (not the pending one) and still clears the pending state. + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const base = createBase(mockRequests, logger); + base.start(); + + // Open with fallback headers (defers a directive with TTL 45s) + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '45' }, + }); + + // Goodbye fires with its own, different TTL (60s) before any payload + simulateEvent(mockEventSource, 'goodbye', { + reason: 'falling back', + protocolFallbackTTL: 60, + }); + + const result = await base.takeResult(); + expect(result.type).toBe('status'); + if (result.type !== 'status') return; + expect(result.state).toBe('terminal_error'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(60000); + + base.close(); +}); + it('clears a pending fallback directive when onopen fires without the fallback header', async () => { const mockEventSource = createMockEventSource(); const mockRequests = createMockRequests(mockEventSource); @@ -854,6 +885,40 @@ it('backfills the deferred TTL into a ping-triggered poll result that already si base.close(); }); +it('does not mutate the ping handler result object when applying a deferred fallback', async () => { + const mockEventSource = createMockEventSource(); + const mockRequests = createMockRequests(mockEventSource); + const pingResult = { + type: 'changeSet' as const, + payload: { events: [], selector: undefined }, + fdv1Fallback: false, + }; + const pingHandler: PingHandler = { + handlePing: jest.fn().mockResolvedValue(pingResult), + }; + const base = createBase(mockRequests, logger, { pingHandler }); + base.start(); + + mockEventSource.onopen({ + type: 'open', + headers: { 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '75' }, + }); + + const { calls } = mockEventSource.addEventListener.mock; + const pingListener = calls.find((c: any[]) => c[0] === 'ping')?.[1]; + await pingListener(); + + const result = await base.takeResult(); + // The queued result carries the deferred fallback... + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(75000); + // ...but the object handed back by the ping handler is left untouched. + expect(pingResult.fdv1Fallback).toBe(false); + expect((pingResult as any).fdv1FallbackTtlMs).toBeUndefined(); + + base.close(); +}); + it('surfaces a deferred fallback directive when a ping-triggered poll throws', async () => { const mockEventSource = createMockEventSource(); const mockRequests = createMockRequests(mockEventSource); diff --git a/packages/shared/sdk-client/src/datasource/fdv2/CacheInitializer.ts b/packages/shared/sdk-client/src/datasource/fdv2/CacheInitializer.ts index caea520464..631ea5b224 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/CacheInitializer.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/CacheInitializer.ts @@ -43,13 +43,13 @@ async function loadFromCache(config: CacheInitializerConfig): Promise string | undefined): Initializer { let shutdownResolve: ((result: FDv2SourceResult) => void) | undefined; const shutdownPromise = new Promise((resolve) => { diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv1PollingSynchronizer.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv1PollingSynchronizer.ts index ec9fdd4cd5..0b0affc808 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv1PollingSynchronizer.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv1PollingSynchronizer.ts @@ -24,6 +24,7 @@ function flagsToPayload(flags: Flags): internal.Payload { const updates: internal.Update[] = Object.entries(flags).map(([key, flag]) => ({ kind: 'flag-eval', key, + // The envelope requires a numeric version; default to 1 if FDv1 omits it version: flag.version ?? 1, object: flag, })); @@ -69,6 +70,8 @@ export function createFDv1PollingSynchronizer( function scheduleNextPoll(startTime: number): void { if (!stopped) { + // Subtract time already spent on the request so the poll cadence stays + // fixed to pollIntervalMs instead of drifting later with each slow response. const elapsed = Date.now() - startTime; const sleepFor = Math.min(Math.max(pollIntervalMs - elapsed, 0), pollIntervalMs); // eslint-disable-next-line @typescript-eslint/no-use-before-define @@ -84,6 +87,9 @@ export function createFDv1PollingSynchronizer( logger?.debug('Polling FDv1 endpoint for feature flag updates'); const startTime = Date.now(); + // Results below always carry fdv1Fallback: false; this synchronizer only + // runs once already on the FDv1 fallback path, so there's no further + // fallback for it to signal. try { const body = await requestor.requestPayload(); @@ -107,7 +113,7 @@ export function createFDv1PollingSynchronizer( return; } - resultQueue.put(changeSet(payload, false)); + resultQueue.put(changeSet(payload, { fdv1Fallback: false })); } catch (err) { if (stopped) { return; @@ -118,7 +124,9 @@ export function createFDv1PollingSynchronizer( if (!isHttpRecoverable(requestError.status)) { logger?.error(httpErrorMessage(err as HttpErrorResponse, 'FDv1 polling request')); stopped = true; - shutdownResolve?.(terminalError(errorInfoFromHttpError(requestError.status), false)); + shutdownResolve?.( + terminalError(errorInfoFromHttpError(requestError.status), { fdv1Fallback: false }), + ); shutdownResolve = undefined; return; } diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts index b742897591..b5b61c2955 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2SourceResult.ts @@ -1,6 +1,7 @@ import { DataSourceErrorKind, internal } from '@launchdarkly/js-sdk-common'; import DataSourceStatusErrorInfo from '../DataSourceStatusErrorInfo'; +import { FallbackDirective } from './fallbackDirective'; /** * Possible states for a status result from an FDv2 data source. @@ -8,7 +9,9 @@ import DataSourceStatusErrorInfo from '../DataSourceStatusErrorInfo'; * - `interrupted`: Transient error; synchronizer will retry automatically. * - `shutdown`: Graceful shutdown; no further results will be produced. * - `terminal_error`: Unrecoverable error; no further results will be produced. - * - `goodbye`: Server-initiated disconnect; no further results will be produced. + * - `goodbye`: Server-initiated disconnect. Synchronizers reconnect internally + * and may still produce results; initializers are single-shot, so the + * orchestrator just moves on to the next one. */ export type SourceState = 'interrupted' | 'shutdown' | 'terminal_error' | 'goodbye'; @@ -61,12 +64,18 @@ export type FDv2SourceResult = ChangeSetResult | StatusResult; */ export function changeSet( payload: internal.Payload, - fdv1Fallback: boolean, + fallback: FallbackDirective, environmentId?: string, freshness?: number, - fdv1FallbackTtlMs?: number, ): FDv2SourceResult { - return { type: 'changeSet', payload, fdv1Fallback, environmentId, freshness, fdv1FallbackTtlMs }; + return { + type: 'changeSet', + payload, + fdv1Fallback: fallback.fdv1Fallback, + environmentId, + freshness, + fdv1FallbackTtlMs: fallback.fdv1FallbackTtlMs, + }; } /** @@ -75,10 +84,15 @@ export function changeSet( */ export function interrupted( errorInfo: DataSourceStatusErrorInfo, - fdv1Fallback: boolean, - fdv1FallbackTtlMs?: number, + fallback: FallbackDirective, ): FDv2SourceResult { - return { type: 'status', state: 'interrupted', errorInfo, fdv1Fallback, fdv1FallbackTtlMs }; + return { + type: 'status', + state: 'interrupted', + errorInfo, + fdv1Fallback: fallback.fdv1Fallback, + fdv1FallbackTtlMs: fallback.fdv1FallbackTtlMs, + }; } /** @@ -94,10 +108,15 @@ export function shutdown(): FDv2SourceResult { */ export function terminalError( errorInfo: DataSourceStatusErrorInfo, - fdv1Fallback: boolean, - fdv1FallbackTtlMs?: number, + fallback: FallbackDirective, ): FDv2SourceResult { - return { type: 'status', state: 'terminal_error', errorInfo, fdv1Fallback, fdv1FallbackTtlMs }; + return { + type: 'status', + state: 'terminal_error', + errorInfo, + fdv1Fallback: fallback.fdv1Fallback, + fdv1FallbackTtlMs: fallback.fdv1FallbackTtlMs, + }; } /** @@ -105,17 +124,20 @@ export function terminalError( * synchronizer will reconnect; the orchestrator does not block this source. * * @param reason Human-readable description of why the server closed the stream. - * @param fdv1Fallback Whether the server directed the client to fall back to FDv1. - * @param fdv1FallbackTtlMs How long (ms) to remain on FDv1 before attempting FDv2 - * recovery. Omit to use the caller's default; pass `0` for indefinite fallback. - * Same semantics as {@link StatusResult.fdv1FallbackTtlMs}. + * @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}. */ -export function goodbye( - reason: string, - fdv1Fallback: boolean, - fdv1FallbackTtlMs?: number, -): FDv2SourceResult { - return { type: 'status', state: 'goodbye', reason, fdv1Fallback, fdv1FallbackTtlMs }; +export function goodbye(reason: string, fallback: FallbackDirective): FDv2SourceResult { + return { + type: 'status', + state: 'goodbye', + reason, + fdv1Fallback: fallback.fdv1Fallback, + fdv1FallbackTtlMs: fallback.fdv1FallbackTtlMs, + }; } /** Builds {@link DataSourceStatusErrorInfo} for an unexpected HTTP status. */ diff --git a/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts b/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts index a1ff4c42e3..cfc86b6873 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/PollingBase.ts @@ -13,7 +13,7 @@ import { interrupted, terminalError, } from './FDv2SourceResult'; -import { readFallbackDirective } from './fallbackDirective'; +import { FallbackDirective, readFallbackDirective } from './fallbackDirective'; function getEnvironmentId(headers: { get(name: string): string | null }): string | undefined { return headers.get('x-ld-envid') ?? undefined; @@ -29,9 +29,8 @@ function getEnvironmentId(headers: { get(name: string): string | null }): string */ function processEvents( events: internal.FDv2Event[], - fdv1Fallback: boolean, + fallback: FallbackDirective, environmentId: string | undefined, - fdv1FallbackTtlMs: number | undefined, logger?: LDLogger, ): FDv2SourceResult { const handler = internal.createProtocolHandler( @@ -52,26 +51,30 @@ function processEvents( switch (action.type) { case 'payload': - earlyResult = changeSet(action.payload, fdv1Fallback, environmentId, undefined, fdv1FallbackTtlMs); + earlyResult = changeSet(action.payload, fallback, environmentId); break; case 'goodbye': - if (fdv1Fallback) { - earlyResult = terminalError(errorInfoFromUnknown(action.reason), true, fdv1FallbackTtlMs); + // A plain goodbye is a status the orchestrator doesn't treat as an + // error. When it's paired with a fallback directive, report it as a + // terminal error instead so the orchestrator actually records the + // error and moves off this source rather than quietly continuing. + if (fallback.fdv1Fallback) { + earlyResult = terminalError(errorInfoFromUnknown(action.reason), fallback); } else { - earlyResult = goodbye(action.reason, fdv1Fallback); + earlyResult = goodbye(action.reason, fallback); } break; case 'serverError': { const errorInfo = errorInfoFromUnknown(action.reason); logger?.error(`Server error during polling: ${action.reason}`); - earlyResult = interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + earlyResult = interrupted(errorInfo, fallback); break; } case 'error': { if (action.kind === 'MISSING_PAYLOAD' || action.kind === 'PROTOCOL_ERROR') { const errorInfo = errorInfoFromInvalidData(action.message); logger?.warn(`Protocol error during polling: ${action.message}`); - earlyResult = interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + earlyResult = interrupted(errorInfo, fallback); } else { logger?.warn(action.message); } @@ -88,7 +91,7 @@ function processEvents( const errorInfo = errorInfoFromUnknown('Unexpected end of polling response'); logger?.error('Unexpected end of polling response'); - return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + return interrupted(errorInfo, fallback); } /** @@ -105,15 +108,12 @@ export async function poll( basis: string | undefined, logger?: LDLogger, ): Promise { - let fdv1Fallback = false; - let fdv1FallbackTtlMs: number | undefined; + let directive: FallbackDirective = { fdv1Fallback: false }; let environmentId: string | undefined; try { const response = await requestor.poll(basis); - const directive = readFallbackDirective(response.headers); - fdv1Fallback = directive.fdv1Fallback; - fdv1FallbackTtlMs = directive.fdv1FallbackTtlMs; + directive = readFallbackDirective(response.headers); environmentId = getEnvironmentId(response.headers); // 304 Not Modified: no payload has changed since the last poll. @@ -125,7 +125,7 @@ export async function poll( type: 'none', updates: [], }; - return changeSet(nonePayload, fdv1Fallback, environmentId, undefined, fdv1FallbackTtlMs); + return changeSet(nonePayload, directive, environmentId); } // Non-success HTTP status @@ -133,17 +133,18 @@ export async function poll( const errorInfo = errorInfoFromHttpError(response.status); logger?.error(`Polling request failed with HTTP error: ${response.status}`); + // status <= 0 means the request never got an HTTP response (e.g. a + // network failure surfaced without a status code); treat that as + // recoverable the same as a retryable HTTP status. const recoverable = response.status <= 0 || isHttpRecoverable(response.status); - return recoverable - ? interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs) - : terminalError(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + return recoverable ? interrupted(errorInfo, directive) : terminalError(errorInfo, directive); } // Successful response - process FDv2 events if (!response.body) { const errorInfo = errorInfoFromInvalidData('Empty response body'); logger?.error('Polling request received empty response body'); - return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + return interrupted(errorInfo, directive); } let parsed: internal.FDv2EventsCollection; @@ -152,7 +153,7 @@ export async function poll( } catch { const errorInfo = errorInfoFromInvalidData('Malformed JSON data in polling response'); logger?.error('Polling request received malformed data'); - return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + return interrupted(errorInfo, directive); } if (!Array.isArray(parsed.events)) { @@ -160,14 +161,14 @@ export async function poll( 'Invalid polling response: missing or invalid events array', ); logger?.error('Polling response does not contain a valid events array'); - return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + return interrupted(errorInfo, directive); } - return processEvents(parsed.events, fdv1Fallback, environmentId, fdv1FallbackTtlMs, logger); + return processEvents(parsed.events, directive, environmentId, logger); } catch (err: any) { const message = err?.message ?? String(err); logger?.error(`Polling request failed with network error: ${message}`); const errorInfo = errorInfoFromNetworkError(message); - return interrupted(errorInfo, fdv1Fallback, fdv1FallbackTtlMs); + return interrupted(errorInfo, directive); } } diff --git a/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts b/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts index bddc56afa5..c4a6340364 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts @@ -55,7 +55,7 @@ export function createPollingInitializer( return result; } - // Recoverable error — save and potentially retry + // Recoverable error: save it and retry if attempts remain lastResult = result; if (attempt < maxRetries) { @@ -72,7 +72,7 @@ export function createPollingInitializer( // Convert final interrupted -> terminal_error const status = lastResult as StatusResult; - return terminalError(status.errorInfo!, status.fdv1Fallback); + return terminalError(status.errorInfo!, { fdv1Fallback: status.fdv1Fallback }); }, close(): void { diff --git a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts index 204fa31711..ad7592b84f 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts @@ -25,7 +25,11 @@ import { shutdown, terminalError, } from './FDv2SourceResult'; -import { readFallbackDirective, readGoodbyeFallbackDirective } from './fallbackDirective'; +import { + FallbackDirective, + readFallbackDirective, + readGoodbyeFallbackDirective, +} from './fallbackDirective'; /** * Handler invoked when a legacy `"ping"` event is received on the stream. @@ -111,8 +115,10 @@ export function createStreamingBase(config: { let connectionAttemptStartTime: number | undefined; let fdv1Fallback = false; let fdv1FallbackTtlMs: number | undefined; - // Directive deferred from a stream `onopen` carrying x-ld-fd-fallback; applied - // to the NEXT changeSet so the payload is delivered before the SDK falls back. + // Directive deferred from a stream `onopen` carrying x-ld-fd-fallback. Promoted + // into the committed state by resolveFallback() the next time a result is + // queued, rather than applied immediately, so a payload already in flight on + // this connection still gets delivered. let pendingFallbackTtlMs: number | undefined; let pendingFallback = false; let started = false; @@ -138,14 +144,16 @@ export function createStreamingBase(config: { * committed `fdv1Fallback`/`fdv1FallbackTtlMs` state, clears the pending * pair, and returns the current committed fallback state. * - * Every path that can produce a result, a stream error, a network failure, - * or a ping-triggered poll (including one that succeeds without its own - * fallback signal), calls this instead of reading the closure variables - * directly, so a directive deferred at onopen surfaces no matter which - * path fires next. Safe to call when nothing is pending; it just returns - * the current state unchanged. + * Called via putWithFallback() by every path that can queue a payload + * result, a stream error, a network failure, or a ping-triggered poll + * result (including one that succeeds without its own fallback signal), + * so a directive deferred at onopen surfaces no matter which path fires + * next. The one exception is the plain goodbye path, which only runs + * when nothing is pending and reads the committed flag directly instead. + * Safe to call when nothing is pending; it just returns the current state + * unchanged. */ - function resolveFallback(): { fdv1Fallback: boolean; fdv1FallbackTtlMs: number | undefined } { + function resolveFallback(): FallbackDirective { if (pendingFallback) { fdv1Fallback = true; fdv1FallbackTtlMs = pendingFallbackTtlMs; @@ -155,58 +163,54 @@ export function createStreamingBase(config: { return { fdv1Fallback, fdv1FallbackTtlMs }; } + /** + * The single place a result derived from the committed/pending stream state + * is enqueued. It resolves the current fallback directive once (promoting any + * directive deferred at onopen) and hands it to `build`, which stamps it onto + * the result. Call sites that carry their own directive source - an in-band + * goodbye directive, an error-response header, or a plain no-fallback goodbye + * or shutdown - enqueue directly instead. + */ + function putWithFallback(build: (fallback: FallbackDirective) => FDv2SourceResult): void { + resultQueue.put(build(resolveFallback())); + } + function handleAction(action: internal.ProtocolAction, rawData?: unknown): void { switch (action.type) { case 'payload': logConnectionResult(true); - if (pendingFallback) { - // A fallback directive was deferred from onopen; apply it now that - // the payload has been delivered. - fdv1Fallback = true; - fdv1FallbackTtlMs = pendingFallbackTtlMs; - resultQueue.put(changeSet(action.payload, true, undefined, undefined, pendingFallbackTtlMs)); - pendingFallback = false; - pendingFallbackTtlMs = undefined; - } else { - resultQueue.put(changeSet(action.payload, fdv1Fallback, undefined, undefined, fdv1FallbackTtlMs)); - } + putWithFallback((fallback) => changeSet(action.payload, fallback)); break; case 'goodbye': { - // In-band fallback signal read from the raw goodbye data (used by SDKs - // that cannot read streaming response headers), or a directive deferred - // from onopen (pendingFallback). Surface as a terminal fallback result. const goodbyeDirective = readGoodbyeFallbackDirective(rawData); - if (goodbyeDirective.fdv1Fallback || pendingFallback) { - const ttlMs = goodbyeDirective.fdv1Fallback - ? goodbyeDirective.fdv1FallbackTtlMs - : pendingFallbackTtlMs; + if (goodbyeDirective.fdv1Fallback) { + // An in-band fallback signal in the goodbye data overrides any pending + // or committed state and carries its own TTL, so it does not go through + // putWithFallback(). Clear the pending pair since it is superseded. fdv1Fallback = true; - resultQueue.put(terminalError(errorInfoFromUnknown(action.reason), true, ttlMs)); pendingFallback = false; pendingFallbackTtlMs = undefined; + resultQueue.put(terminalError(errorInfoFromUnknown(action.reason), goodbyeDirective)); + } else if (pendingFallback) { + // No in-band signal, but a directive was deferred at onopen: let + // putWithFallback() consume the pending fallback and clear the values. + putWithFallback((fallback) => terminalError(errorInfoFromUnknown(action.reason), fallback)); } else { - resultQueue.put(goodbye(action.reason, fdv1Fallback)); + resultQueue.put(goodbye(action.reason, { fdv1Fallback })); } break; } - case 'serverError': { - const fallback = resolveFallback(); - resultQueue.put( - interrupted(errorInfoFromUnknown(action.reason), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), - ); + case 'serverError': + putWithFallback((fallback) => interrupted(errorInfoFromUnknown(action.reason), fallback)); break; - } case 'error': // Only actionable errors are queued; informational ones (UNKNOWN_EVENT) // are logged by the protocol handler. if (action.kind === 'MISSING_PAYLOAD' || action.kind === 'PROTOCOL_ERROR') { - const fallback = resolveFallback(); - resultQueue.put( - interrupted(errorInfoFromInvalidData(action.message), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), - ); + putWithFallback((fallback) => interrupted(errorInfoFromInvalidData(action.message), fallback)); } break; @@ -223,32 +227,27 @@ export function createStreamingBase(config: { get: (name: string) => errHeaders[name.toLowerCase()] ?? null, }); if (directive.fdv1Fallback) { + // A fallback directive overrides normal retry handling, even for an + // otherwise-recoverable HTTP status: the server is telling us to stop + // trying FDv2 now rather than keep retrying this connection. fdv1Fallback = true; fdv1FallbackTtlMs = directive.fdv1FallbackTtlMs; logConnectionResult(false); - resultQueue.put( - terminalError(errorInfoFromHttpError(err.status ?? 0), true, directive.fdv1FallbackTtlMs), - ); + resultQueue.put(terminalError(errorInfoFromHttpError(err.status ?? 0), directive)); return false; } - const fallback = resolveFallback(); - if (!shouldRetry(err)) { config.logger?.error(httpErrorMessage(err, 'streaming request')); logConnectionResult(false); - resultQueue.put( - terminalError(errorInfoFromHttpError(err.status ?? 0), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), - ); + putWithFallback((fallback) => terminalError(errorInfoFromHttpError(err.status ?? 0), fallback)); return false; } config.logger?.warn(httpErrorMessage(err, 'streaming request', 'will retry')); logConnectionResult(false); logConnectionAttempt(); - resultQueue.put( - interrupted(errorInfoFromHttpError(err.status ?? 0), fallback.fdv1Fallback, fallback.fdv1FallbackTtlMs), - ); + putWithFallback((fallback) => interrupted(errorInfoFromHttpError(err.status ?? 0), fallback)); return true; } @@ -278,13 +277,8 @@ export function createStreamingBase(config: { `Stream received data that was unable to be parsed in "${eventName}" message`, ); config.logger?.debug(`Data follows: ${event.data}`); - const fallback = resolveFallback(); - resultQueue.put( - interrupted( - errorInfoFromInvalidData('Malformed JSON in EventStream'), - fallback.fdv1Fallback, - fallback.fdv1FallbackTtlMs, - ), + putWithFallback((fallback) => + interrupted(errorInfoFromInvalidData('Malformed JSON in EventStream'), fallback), ); return; } @@ -316,33 +310,29 @@ export function createStreamingBase(config: { return; } - // Trust the poll's own fallback signal (e.g. from its own HTTP - // response headers) over a directive deferred at onopen, so an - // onopen directive still surfaces even while ping-triggered polls - // keep succeeding. Still backfill the TTL alone when the poll's own - // fallback signal did not carry one, so a deferred TTL is not lost - // just because the poll also happened to indicate fallback. - const fallback = resolveFallback(); - if (!result.fdv1Fallback && fallback.fdv1Fallback) { - result.fdv1Fallback = true; - result.fdv1FallbackTtlMs = fallback.fdv1FallbackTtlMs; - } else if (result.fdv1Fallback && result.fdv1FallbackTtlMs === undefined) { - result.fdv1FallbackTtlMs = fallback.fdv1FallbackTtlMs; - } - resultQueue.put(result); + // A poll can succeed with no fallback signal of its own even though + // a directive was deferred at onopen; override the result in that + // case so the deferred directive still surfaces. When the poll's own + // result already signals fallback, keep its flag but backfill the + // TTL from the deferred directive if the poll didn't carry one. + // Build a new object rather than mutating the poll's result in place. + putWithFallback((fallback) => { + if (!result.fdv1Fallback && fallback.fdv1Fallback) { + return { ...result, fdv1Fallback: true, fdv1FallbackTtlMs: fallback.fdv1FallbackTtlMs }; + } + if (result.fdv1Fallback && result.fdv1FallbackTtlMs === undefined) { + return { ...result, fdv1FallbackTtlMs: fallback.fdv1FallbackTtlMs }; + } + return result; + }); } catch (err: any) { if (stopped) { return; } config.logger?.error(`Error handling ping: ${err?.message ?? err}`); - const fallback = resolveFallback(); - resultQueue.put( - interrupted( - errorInfoFromNetworkError(err?.message ?? 'Error during ping poll'), - fallback.fdv1Fallback, - fallback.fdv1FallbackTtlMs, - ), + putWithFallback((fallback) => + interrupted(errorInfoFromNetworkError(err?.message ?? 'Error during ping poll'), fallback), ); } }); @@ -383,13 +373,8 @@ export function createStreamingBase(config: { // This condition will be handled by the error filter. return; } - const fallback = resolveFallback(); - resultQueue.put( - interrupted( - errorInfoFromNetworkError(err?.message ?? 'IO Error'), - fallback.fdv1Fallback, - fallback.fdv1FallbackTtlMs, - ), + putWithFallback((fallback) => + interrupted(errorInfoFromNetworkError(err?.message ?? 'IO Error'), fallback), ); }; @@ -413,9 +398,10 @@ export function createStreamingBase(config: { pendingFallback = false; pendingFallbackTtlMs = undefined; - // SDKs whose EventSource exposes response headers can - // detect FDv1 fallback at connection open. Defer the directive and apply - // it to the next changeSet so the payload is delivered first. + // EventSource implementations that expose response headers can detect + // FDv1 fallback at connection open. Defer the directive; resolveFallback() + // applies it to whatever result is queued next, so a payload already + // in flight is delivered first. const openHeaders = e?.headers; if (openHeaders) { const directive = readFallbackDirective({ From adfcb4d7b11896734950c255f0efb7e449c7bd2c Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Tue, 21 Jul 2026 15:49:27 -0400 Subject: [PATCH 09/10] chore: extending resolveFallback to take in incoming directives This will allow us to resolve directives that should override the fallback behavior such as goodbye or error --- .../fdv2/PollingInitializer.test.ts | 21 +++++ .../src/datasource/fdv2/PollingInitializer.ts | 5 +- .../src/datasource/fdv2/StreamingFDv2Base.ts | 80 ++++++++++--------- 3 files changed, 66 insertions(+), 40 deletions(-) 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 b17510630c..325d916b20 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingInitializer.test.ts @@ -148,6 +148,27 @@ it('exhausts retries on recoverable error then returns terminal error', async () expect(sleep).toHaveBeenCalledTimes(3); }); +it('preserves the fallback directive and TTL when retries exhaust', async () => { + const requestor: FDv2Requestor = { + poll: jest.fn().mockResolvedValue({ + status: 500, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '300' }), + body: null, + }), + }; + + const initializer = createPollingInitializer(requestor, logger, () => undefined); + const result = await initializer.run(); + + expect(result.type).toBe('status'); + if (result.type === 'status') { + expect(result.state).toBe('terminal_error'); + expect(result.fdv1Fallback).toBe(true); + expect(result.fdv1FallbackTtlMs).toBe(300000); + } + expect(requestor.poll).toHaveBeenCalledTimes(4); +}); + it('exhausts retries on network error then returns terminal error', async () => { const requestor: FDv2Requestor = { poll: jest.fn().mockRejectedValue(new Error('network failure')), diff --git a/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts b/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts index c4a6340364..9bd63341bf 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/PollingInitializer.ts @@ -72,7 +72,10 @@ export function createPollingInitializer( // Convert final interrupted -> terminal_error const status = lastResult as StatusResult; - return terminalError(status.errorInfo!, { fdv1Fallback: status.fdv1Fallback }); + return terminalError(status.errorInfo!, { + fdv1Fallback: status.fdv1Fallback, + fdv1FallbackTtlMs: status.fdv1FallbackTtlMs, + }); }, close(): void { diff --git a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts index ad7592b84f..c73d56f37b 100644 --- a/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts +++ b/packages/shared/sdk-client/src/datasource/fdv2/StreamingFDv2Base.ts @@ -140,20 +140,24 @@ export function createStreamingBase(config: { } /** - * Promotes a directive deferred from `onopen` (`pendingFallback`) into the - * committed `fdv1Fallback`/`fdv1FallbackTtlMs` state, clears the pending - * pair, and returns the current committed fallback state. - * - * Called via putWithFallback() by every path that can queue a payload - * result, a stream error, a network failure, or a ping-triggered poll - * result (including one that succeeds without its own fallback signal), - * so a directive deferred at onopen surfaces no matter which path fires - * next. The one exception is the plain goodbye path, which only runs - * when nothing is pending and reads the committed flag directly instead. - * Safe to call when nothing is pending; it just returns the current state - * unchanged. + * Resolves the current fallback directive, optionally overridden by an + * `incoming` directive from a different source (an in-band goodbye payload + * or an error-response header). When `incoming` signals fallback, it wins + * outright - it carries its own TTL, which takes precedence over anything + * deferred at `onopen`, and the pending pair is cleared since it is now + * superseded. Otherwise, a directive deferred at `onopen` (`pendingFallback`) + * is promoted into the committed `fdv1Fallback`/`fdv1FallbackTtlMs` state and + * the pending pair is cleared. Safe to call with neither; it just returns the + * current committed state unchanged. */ - function resolveFallback(): FallbackDirective { + function resolveFallback(incoming?: FallbackDirective): FallbackDirective { + if (incoming?.fdv1Fallback) { + fdv1Fallback = true; + fdv1FallbackTtlMs = incoming.fdv1FallbackTtlMs; + pendingFallback = false; + pendingFallbackTtlMs = undefined; + return { fdv1Fallback, fdv1FallbackTtlMs }; + } if (pendingFallback) { fdv1Fallback = true; fdv1FallbackTtlMs = pendingFallbackTtlMs; @@ -164,15 +168,18 @@ export function createStreamingBase(config: { } /** - * The single place a result derived from the committed/pending stream state - * is enqueued. It resolves the current fallback directive once (promoting any - * directive deferred at onopen) and hands it to `build`, which stamps it onto - * the result. Call sites that carry their own directive source - an in-band - * goodbye directive, an error-response header, or a plain no-fallback goodbye - * or shutdown - enqueue directly instead. + * The single place a result derived from the fallback directive state is + * enqueued. It resolves the current directive once via `resolveFallback()` + * - passing `incoming` through when the call site has its own directive + * source - and hands the result to `build`, which stamps it onto the + * result. `shutdown()` is the only path with no fallback state to resolve, + * so it still enqueues directly. */ - function putWithFallback(build: (fallback: FallbackDirective) => FDv2SourceResult): void { - resultQueue.put(build(resolveFallback())); + function putWithFallback( + build: (fallback: FallbackDirective) => FDv2SourceResult, + incoming?: FallbackDirective, + ): void { + resultQueue.put(build(resolveFallback(incoming))); } function handleAction(action: internal.ProtocolAction, rawData?: unknown): void { @@ -183,22 +190,16 @@ export function createStreamingBase(config: { break; case 'goodbye': { + // An in-band fallback signal in the goodbye data (its own TTL) takes + // precedence over a directive deferred at onopen; putWithFallback() + // passes it through to resolveFallback() as the incoming override. const goodbyeDirective = readGoodbyeFallbackDirective(rawData); - if (goodbyeDirective.fdv1Fallback) { - // An in-band fallback signal in the goodbye data overrides any pending - // or committed state and carries its own TTL, so it does not go through - // putWithFallback(). Clear the pending pair since it is superseded. - fdv1Fallback = true; - pendingFallback = false; - pendingFallbackTtlMs = undefined; - resultQueue.put(terminalError(errorInfoFromUnknown(action.reason), goodbyeDirective)); - } else if (pendingFallback) { - // No in-band signal, but a directive was deferred at onopen: let - // putWithFallback() consume the pending fallback and clear the values. - putWithFallback((fallback) => terminalError(errorInfoFromUnknown(action.reason), fallback)); - } else { - resultQueue.put(goodbye(action.reason, { fdv1Fallback })); - } + putWithFallback( + (fallback) => (fallback.fdv1Fallback + ? terminalError(errorInfoFromUnknown(action.reason), fallback) + : goodbye(action.reason, fallback)), + goodbyeDirective, + ); break; } @@ -230,10 +231,11 @@ export function createStreamingBase(config: { // A fallback directive overrides normal retry handling, even for an // otherwise-recoverable HTTP status: the server is telling us to stop // trying FDv2 now rather than keep retrying this connection. - fdv1Fallback = true; - fdv1FallbackTtlMs = directive.fdv1FallbackTtlMs; logConnectionResult(false); - resultQueue.put(terminalError(errorInfoFromHttpError(err.status ?? 0), directive)); + putWithFallback( + (fallback) => terminalError(errorInfoFromHttpError(err.status ?? 0), fallback), + directive, + ); return false; } From 20081c86c77e0b0873f79b7f570d85ac4f6316c9 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 22 Jul 2026 13:14:47 -0400 Subject: [PATCH 10/10] chore: bot comments --- .../datasource/fdv2/PollingBase.test.ts | 82 +++++++++---------- 1 file changed, 40 insertions(+), 42 deletions(-) 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 7e07dc54c7..c9c25f2114 100644 --- a/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts +++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/PollingBase.test.ts @@ -486,60 +486,58 @@ describe('given a delete-object event', () => { }); }); -describe('given x-ld-fd-fallback-ttl header', () => { - it('reads TTL in seconds and converts to ms on a changeSet', async () => { - const body = makeFullPayloadBody({ flagA: { value: true } }); - const requestor = makeRequestor({ - status: 200, - headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }), - body, - }); - - const result = await poll(requestor, undefined, logger); - - expect(result.fdv1Fallback).toBe(true); - expect((result as any).fdv1FallbackTtlMs).toBe(60000); +it('reads TTL in seconds and converts to ms on a changeSet', async () => { + const body = makeFullPayloadBody({ flagA: { value: true } }); + const requestor = makeRequestor({ + status: 200, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '60' }), + body, }); - it('treats TTL "0" as indefinite (0 ms)', async () => { - const body = makeFullPayloadBody({ flagA: { value: true } }); - const requestor = makeRequestor({ - status: 200, - headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '0' }), - body, - }); + const result = await poll(requestor, undefined, logger); - const result = await poll(requestor, undefined, logger); + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(60000); +}); - expect(result.fdv1Fallback).toBe(true); - expect((result as any).fdv1FallbackTtlMs).toBe(0); +it('treats TTL "0" as indefinite (0 ms)', async () => { + const body = makeFullPayloadBody({ flagA: { value: true } }); + const requestor = makeRequestor({ + status: 200, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '0' }), + body, }); - it('leaves TTL undefined when the ttl header is absent but fallback is true', async () => { - const body = makeFullPayloadBody({ flagA: { value: true } }); - const requestor = makeRequestor({ - status: 200, - headers: makeHeaders({ 'x-ld-fd-fallback': 'true' }), - body, - }); + const result = await poll(requestor, undefined, logger); - const result = await poll(requestor, undefined, logger); + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(0); +}); - expect(result.fdv1Fallback).toBe(true); - expect((result as any).fdv1FallbackTtlMs).toBeUndefined(); +it('leaves TTL undefined when the ttl header is absent but fallback is true', async () => { + const body = makeFullPayloadBody({ flagA: { value: true } }); + const requestor = makeRequestor({ + status: 200, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true' }), + body, }); - it('stamps TTL on a non-success error response', async () => { - const requestor = makeRequestor({ - status: 503, - headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '30' }), - body: null, - }); + const result = await poll(requestor, undefined, logger); - const result = await poll(requestor, undefined, logger); + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBeUndefined(); +}); - expect(result.fdv1Fallback).toBe(true); - expect((result as any).fdv1FallbackTtlMs).toBe(30000); +it('stamps TTL on a non-success error response', async () => { + const requestor = makeRequestor({ + status: 503, + headers: makeHeaders({ 'x-ld-fd-fallback': 'true', 'x-ld-fd-fallback-ttl': '30' }), + body: null, }); + + const result = await poll(requestor, undefined, logger); + + expect(result.fdv1Fallback).toBe(true); + expect((result as any).fdv1FallbackTtlMs).toBe(30000); });