chore: now utilizes FDv2 basis param and supports FDv1 fallback

packages/sdk/server-node/package.json

@@ -47,7 +47,7 @@
   "dependencies": {
     "@launchdarkly/js-server-sdk-common": "2.15.1",
     "https-proxy-agent": "^5.0.1",
-    "launchdarkly-eventsource": "2.1.0"
+    "launchdarkly-eventsource": "2.2.0"
   },
   "devDependencies": {
     "@trivago/prettier-plugin-sort-imports": "^4.1.1",

packages/shared/common/__tests__/internal/fdv2/PayloadStreamReader.test.ts

@@ -25,7 +25,7 @@ it('it sets basis to true when intent code is xfer-full', () => {
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('payload-transferred', {
     data: '{"state": "mockState", "version": 1}',
@@ -47,7 +47,7 @@ it('it sets basis to false when intent code is xfer-changes', () => {
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-changes", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-changes", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('payload-transferred', {
     data: '{"state": "mockState", "version": 1}',
@@ -69,7 +69,7 @@ it('it sets basis to false and emits empty payload when intent code is none', ()
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "none", "id": "mockId", "target": 42}]}',
+    data: '{"payloads": [{"intentCode": "none", "id": "mockId", "target": 42}]}',
   });
   expect(receivedPayloads.length).toEqual(1);
   expect(receivedPayloads[0].id).toEqual('mockId');
@@ -88,7 +88,7 @@ it('it handles xfer-full then xfer-changes', () => {
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
@@ -130,7 +130,7 @@ it('it includes multiple types of updates in payload', () => {
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
@@ -171,7 +171,7 @@ it('it does not include messages thats are not between server-intent and payload
     data: '{"kind": "mockKind", "key": "flagShouldIgnore", "version": 123, "object": {"objectFieldShouldIgnore": "objectValueShouldIgnore"}}',
   });
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
@@ -237,7 +237,7 @@ it('logs prescribed message when error event is encountered', () => {
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
@@ -279,7 +279,7 @@ it('discards partially transferred data when an error is encountered', () => {
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
@@ -294,7 +294,7 @@ it('discards partially transferred data when an error is encountered', () => {
     data: '{"state": "mockState", "version": 1}',
   });
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId2"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId2"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagX", "version": 123, "object": {"objectFieldX": "objectValueX"}}',
@@ -338,7 +338,7 @@ it('silently ignores unrecognized kinds', () => {
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',
@@ -368,7 +368,7 @@ it('ignores additional payloads beyond the first payload in the server-intent me
   });
 
   mockStream.simulateEvent('server-intent', {
-    data: '{"payloads": [{"code": "xfer-full", "id": "mockId"},{"code": "IShouldBeIgnored", "id": "IShouldBeIgnored"}]}',
+    data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"},{"intentCode": "IShouldBeIgnored", "id": "IShouldBeIgnored"}]}',
   });
   mockStream.simulateEvent('put-object', {
     data: '{"kind": "mockKind", "key": "flagA", "version": 123, "object": {"objectFieldA": "objectValueA"}}',

packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts

@@ -8,6 +8,8 @@ import {
   CompositeDataSource,
   TransitionConditions,
 } from '../../../src/datasource/CompositeDataSource';
+import { DataSourceErrorKind } from '../../../src/datasource/DataSourceErrorKinds';
+import { LDFlagDeliveryFallbackError } from '../../../src/datasource/errors';
 
 function makeDataSourceFactory(internal: DataSource): LDDataSourceFactory {
   return () => internal;
@@ -75,6 +77,7 @@ it('handles initializer getting basis, switching to synchronizer', async () => {
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [makeDataSourceFactory(mockSynchronizer1)],
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -169,6 +172,7 @@ it('handles initializer getting basis, switches to synchronizer 1, falls back to
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [makeDataSourceFactory(mockSynchronizer1), makeDataSourceFactory(mockSynchronizer2)],
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -270,6 +274,7 @@ it('removes synchronizer that reports unrecoverable error and loops on remaining
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [makeDataSourceFactory(mockSynchronizer1), makeDataSourceFactory(mockSynchronizer2)],
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -306,6 +311,117 @@ it('removes synchronizer that reports unrecoverable error and loops on remaining
   expect(statusCallback).toHaveBeenNthCalledWith(10, DataSourceState.Valid, undefined); // sync1 valid
 });
 
+it('falls back to FDv1 synchronizers when FDv1 fallback error is reported', async () => {
+  const mockInitializer1: DataSource = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Valid);
+          _dataCallback(true, { key: 'init1' });
+          _statusCallback(DataSourceState.Closed);
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const mockSynchronizer1 = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(
+            DataSourceState.Closed,
+            new LDFlagDeliveryFallbackError(
+              DataSourceErrorKind.ErrorResponse,
+              `Response header indicates to fallback to FDv1`,
+              403,
+            ),
+          );
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const mockSynchronizer2 = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Closed, {
+            name: 'Error',
+            message: 'I should NOT be called due to FDv1 Fallback',
+          });
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const mockFDv1Data = { key: 'FDv1Data' };
+  const mockFDv1Synchronizer = {
+    start: jest
+      .fn()
+      .mockImplementation(
+        (
+          _dataCallback: (basis: boolean, data: any) => void,
+          _statusCallback: (status: DataSourceState, err?: any) => void,
+        ) => {
+          _statusCallback(DataSourceState.Initializing);
+          _statusCallback(DataSourceState.Valid, null); // this should lead to recovery
+          _dataCallback(false, mockFDv1Data);
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const underTest = new CompositeDataSource(
+    [makeDataSourceFactory(mockInitializer1)],
+    [makeDataSourceFactory(mockSynchronizer1), makeDataSourceFactory(mockSynchronizer2)],
+    [makeDataSourceFactory(mockFDv1Synchronizer)],
+    undefined,
+    makeTestTransitionConditions(),
+    makeZeroBackoff(),
+  );
+
+  let dataCallback;
+  const statusCallback = jest.fn();
+  await new Promise<void>((resolve) => {
+    dataCallback = jest.fn((_: boolean, data: any) => {
+      if (data === mockFDv1Data) {
+        resolve();
+      }
+    });
+
+    underTest.start(dataCallback, statusCallback);
+  });
+
+  expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
+  expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1);
+  expect(mockSynchronizer2.start).toHaveBeenCalledTimes(0); // this synchronizer should not be called because we fall back to FDv1 synchronizers instead
+  expect(mockFDv1Synchronizer.start).toHaveBeenCalledTimes(1);
+  expect(dataCallback).toHaveBeenCalledTimes(2);
+  expect(dataCallback).toHaveBeenNthCalledWith(1, true, { key: 'init1' });
+  expect(dataCallback).toHaveBeenNthCalledWith(2, false, { key: 'FDv1Data' });
+  expect(statusCallback).toHaveBeenCalledTimes(5);
+  expect(statusCallback).toHaveBeenNthCalledWith(1, DataSourceState.Initializing, undefined);
+  expect(statusCallback).toHaveBeenNthCalledWith(2, DataSourceState.Valid, undefined); // initializer got data
+  expect(statusCallback).toHaveBeenNthCalledWith(3, DataSourceState.Interrupted, undefined); // initializer closed
+  expect(statusCallback).toHaveBeenNthCalledWith(4, DataSourceState.Interrupted, expect.anything()); // sync1 fdv1 fallback error
+  expect(statusCallback).toHaveBeenNthCalledWith(5, DataSourceState.Valid, undefined); // sync1 valid
+});
+
 it('reports error when all initializers fail', async () => {
   const mockInitializer1Error = {
     name: 'Error',
@@ -348,6 +464,7 @@ it('reports error when all initializers fail', async () => {
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1), makeDataSourceFactory(mockInitializer2)],
     [], // no synchronizers for this test
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -445,6 +562,7 @@ it('it reports DataSourceState Closed when all synchronizers report Closed with
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [makeDataSourceFactory(mockSynchronizer1), makeDataSourceFactory(mockSynchronizer2)],
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -508,6 +626,7 @@ it('can be stopped when in thrashing synchronizer fallback loop', async () => {
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [makeDataSourceFactory(mockSynchronizer1)], // will continuously fallback onto itself
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -579,6 +698,7 @@ it('can be stopped and restarted', async () => {
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [makeDataSourceFactory(mockSynchronizer1)],
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -625,6 +745,7 @@ it('can be stopped and restarted', async () => {
 
 it('is well behaved with no initializers and no synchronizers configured', async () => {
   const underTest = new CompositeDataSource(
+    [],
     [],
     [],
     undefined,
@@ -671,6 +792,7 @@ it('is well behaved with no initializer and synchronizer configured', async () =
   const underTest = new CompositeDataSource(
     [],
     [makeDataSourceFactory(mockSynchronizer1)],
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -712,6 +834,7 @@ it('is well behaved with an initializer and no synchronizers configured', async
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [],
+    [],
     undefined,
     makeTestTransitionConditions(),
     makeZeroBackoff(),
@@ -774,6 +897,7 @@ it('consumes cancellation tokens correctly', async () => {
   const underTest = new CompositeDataSource(
     [makeDataSourceFactory(mockInitializer1)],
     [makeDataSourceFactory(mockSynchronizer1)],
+    [],
     undefined,
     {
       // pass in transition condition so that it will thrash, generating cancellation tokens repeatedly

packages/shared/common/src/api/platform/Requests.ts

@@ -134,4 +134,5 @@ export interface Requests {
 export interface HttpErrorResponse {
   message: string;
   status?: number;
+  headers?: Record<string, string>;
 }

packages/shared/common/src/api/subsystem/DataSystem/DataSource.ts

@@ -16,10 +16,12 @@ export interface DataSource {
    * @param dataCallback that will be called when data arrives, may be called multiple times.
    * @param statusCallback that will be called when data source state changes or an unrecoverable error
    * has been encountered.
+   * @param selectorGetter that can be invoked to provide the FDv2 selector/basis if one exists
    */
   start(
     dataCallback: (basis: boolean, data: any) => void,
     statusCallback: (status: DataSourceState, err?: any) => void,
+    selectorGetter?: () => string | undefined,
   ): void;
 
   /**