fix: fallback variables need to be read on stream errors

joker23 · joker23 · commit 46de8d450a5f · 2026-07-17T12:51:40.000-04:00
diff --git 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,11 +649,230 @@ 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');
   if (result.type !== 'status') return;
   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
@@ -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,18 +316,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. 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) {
           return;
         }
 
         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,
+          ),
         );
       };
 
