removing DataSourceState.Off

Todd Anderson · Todd Anderson · commit dbb432c3c8b7 · 2025-04-22T12:03:42.000-05:00
diff --git a/packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts b/packages/shared/common/__tests__/subsystem/DataSystem/CompositeDataSource.test.ts
@@ -230,9 +230,10 @@ it('removes synchronizer that reports unrecoverable error and loops on remaining
           _statusCallback: (status: DataSourceState, err?: any) => void,
         ) => {
           _statusCallback(DataSourceState.Initializing);
-          _statusCallback(DataSourceState.Off, {
+          _statusCallback(DataSourceState.Closed, {
             name: 'Error',
-            message: 'I am an unrecoverable error!', // error will lead to culling
+            message: 'I am an unrecoverable error!', // error will lead to culling,
+            recoverable: false,
           });
         },
       ),
@@ -380,93 +381,99 @@ it('reports error when all initializers fail', async () => {
   );
   expect(statusCallback).toHaveBeenNthCalledWith(4, DataSourceState.Closed, {
     name: 'ExhaustedDataSources',
-    message:
-      'CompositeDataSource has exhausted all configured datasources (2 initializers, 0 synchronizers).',
+    message: 'CompositeDataSource has exhausted all configured initializers and synchronizers.',
   });
   expect(statusCallback).toHaveBeenCalledTimes(4);
 });
 
-// it('it reports DataSourceState Off when all synchronizers report Off', 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.Off, {
-//             name: 'Error1',
-//             message: 'I am an unrecoverable error!',
-//           });
-//         },
-//       ),
-//     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.Off, {
-//             name: 'Error2',
-//             message: 'I am an unrecoverable error!',
-//           });
-//         },
-//       ),
-//     stop: jest.fn(),
-//   };
-
-//   const underTest = new CompositeDataSource(
-//     [makeDataSourceFactory(mockInitializer1)],
-//     [makeDataSourceFactory(mockSynchronizer1), makeDataSourceFactory(mockSynchronizer2)],
-//     undefined,
-//     makeTestTransitionConditions(),
-//     makeZeroBackoff(),
-//   );
-
-//   let statusCallback;
-//   await new Promise<void>((resolve) => {
-//     statusCallback = jest.fn((state: DataSourceState, err: any) => {
-//       if (err && err.name === 'Error2') {
-//         resolve();
-//       }
-//     });
-
-//     underTest.start(jest.fn(), statusCallback);
-//   });
-
-//   expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
-//   expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1);
-//   expect(mockSynchronizer2.start).toHaveBeenCalledTimes(1);
-//   expect(statusCallback).toHaveBeenNthCalledWith(1, DataSourceState.Initializing, null);
-//   expect(statusCallback).toHaveBeenNthCalledWith(2, DataSourceState.Valid, null);
-//   expect(statusCallback).toHaveBeenNthCalledWith(3, DataSourceState.Interrupted, expect.anything());
-//   expect(statusCallback).toHaveBeenNthCalledWith(4, DataSourceState.Off, null);
-// });
+it('it reports DataSourceState Closed when all synchronizers report Closed with unrecoverable errors', 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, {
+            name: 'Error1',
+            message: 'I am an unrecoverable error!',
+            recoverable: false,
+          });
+        },
+      ),
+    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: 'Error2',
+            message: 'I am an unrecoverable error!',
+            recoverable: false,
+          });
+        },
+      ),
+    stop: jest.fn(),
+  };
+
+  const underTest = new CompositeDataSource(
+    [makeDataSourceFactory(mockInitializer1)],
+    [makeDataSourceFactory(mockSynchronizer1), makeDataSourceFactory(mockSynchronizer2)],
+    undefined,
+    makeTestTransitionConditions(),
+    makeZeroBackoff(),
+  );
+
+  let statusCallback;
+  await new Promise<void>((resolve) => {
+    statusCallback = jest.fn((state: DataSourceState, _: any) => {
+      if (state === DataSourceState.Closed) {
+        resolve();
+      }
+    });
+
+    underTest.start(jest.fn(), statusCallback);
+  });
+
+  expect(mockInitializer1.start).toHaveBeenCalledTimes(1);
+  expect(mockSynchronizer1.start).toHaveBeenCalledTimes(1);
+  expect(mockSynchronizer2.start).toHaveBeenCalledTimes(1);
+  expect(statusCallback).toHaveBeenNthCalledWith(1, DataSourceState.Initializing, undefined);
+  expect(statusCallback).toHaveBeenNthCalledWith(2, DataSourceState.Valid, undefined);
+  expect(statusCallback).toHaveBeenNthCalledWith(3, DataSourceState.Interrupted, undefined); // initializer closes properly
+  expect(statusCallback).toHaveBeenNthCalledWith(4, DataSourceState.Interrupted, expect.anything()); // sync1 closed with unrecoverable error
+  expect(statusCallback).toHaveBeenNthCalledWith(5, DataSourceState.Interrupted, expect.anything()); // sync2 closed with unrecoverable error
+  expect(statusCallback).toHaveBeenNthCalledWith(6, DataSourceState.Closed, {
+    name: 'ExhaustedDataSources',
+    message: `CompositeDataSource has exhausted all configured initializers and synchronizers.`,
+  });
+});
 
 it('can be stopped when in thrashing synchronizer fallback loop', async () => {
   const mockInitializer1 = {
@@ -639,8 +646,7 @@ it('is well behaved with no initializers and no synchronizers configured', async
   expect(statusCallback).toHaveBeenNthCalledWith(1, DataSourceState.Initializing, undefined); // initializer
   expect(statusCallback).toHaveBeenNthCalledWith(2, DataSourceState.Closed, {
     name: 'ExhaustedDataSources',
-    message:
-      'CompositeDataSource has exhausted all configured datasources (0 initializers, 0 synchronizers).',
+    message: 'CompositeDataSource has exhausted all configured initializers and synchronizers.',
   });
 });
 
@@ -730,8 +736,7 @@ it('is well behaved with an initializer and no synchronizers configured', async
   expect(statusCallback).toHaveBeenNthCalledWith(3, DataSourceState.Interrupted, undefined); // initializer got data
   expect(statusCallback).toHaveBeenNthCalledWith(4, DataSourceState.Closed, {
     name: 'ExhaustedDataSources',
-    message:
-      'CompositeDataSource has exhausted all configured datasources (1 initializers, 0 synchronizers).',
+    message: 'CompositeDataSource has exhausted all configured initializers and synchronizers.',
   });
 });
 
diff --git a/packages/shared/common/src/api/subsystem/DataSystem/DataSource.ts b/packages/shared/common/src/api/subsystem/DataSystem/DataSource.ts
@@ -6,10 +6,8 @@ export enum DataSourceState {
   Initializing,
   // Transient issue, automatic retry is expected
   Interrupted,
-  // Data source was closed and will not retry
+  // Data source was closed and will not retry automatically.
   Closed,
-  // This datasource encountered an unrecoverable error and it is not expected to be resolved through trying again in the future
-  Off,
 }
 
 export interface DataSource {
diff --git a/packages/shared/common/src/datasource/CompositeDataSource.ts b/packages/shared/common/src/datasource/CompositeDataSource.ts
@@ -102,6 +102,7 @@ export class CompositeDataSource implements DataSource {
         isPrimary,
         cullDSFactory,
       } = this._pickDataSource(lastTransition);
+
       const internalTransitionPromise = new Promise<TransitionRequest>((transitionResolve) => {
         if (currentDS) {
           // these local variables are used for handling automatic transition related to data source status (ex: recovering to primary after
@@ -129,9 +130,9 @@ export class CompositeDataSource implements DataSource {
               this._logger?.debug(
                 `CompositeDataSource received state ${state} from underlying data source.`,
               );
-              if (err || state === DataSourceState.Closed || state === DataSourceState.Off) {
+              if (err || state === DataSourceState.Closed) {
                 callbackHandler.disable();
-                if (state === DataSourceState.Off) {
+                if (err.recoverable === false) {
                   // don't use this datasource's factory again
                   cullDSFactory?.();
                 }
@@ -173,7 +174,7 @@ export class CompositeDataSource implements DataSource {
             transition: 'stop',
             err: {
               name: 'ExhaustedDataSources',
-              message: `CompositeDataSource has exhausted all configured datasources (${this._initFactories.length()} initializers, ${this._syncFactories.length()} synchronizers).`,
+              message: `CompositeDataSource has exhausted all configured initializers and synchronizers.`,
             },
           });
         }
@@ -340,7 +341,7 @@ export class CompositeDataSource implements DataSource {
    * This wrapper will ensure the following:
    *
    * Don't report DataSourceState.Initializing except as first status callback.
-   * Map underlying DataSourceState.Closed and DataSourceState.Off to interrupted.
+   * Map underlying DataSourceState.Closed to interrupted.
    * Don't report the same status and error twice in a row.
    */
   private _wrapStatusCallbackWithSanitizer(
@@ -353,7 +354,7 @@ export class CompositeDataSource implements DataSource {
     return (status: DataSourceState, err?: any) => {
       let sanitized = status;
       // underlying errors, closed state, or off are masked as interrupted while we transition
-      if (status === DataSourceState.Closed || status === DataSourceState.Off) {
+      if (status === DataSourceState.Closed) {
         sanitized = DataSourceState.Interrupted;
       }
 
diff --git a/packages/shared/sdk-server/__tests__/data_sources/PollingProcessorFDv2.test.ts b/packages/shared/sdk-server/__tests__/data_sources/PollingProcessorFDv2.test.ts
@@ -213,13 +213,14 @@ describe('given a polling processor with a short poll duration', () => {
       expect(mockStatusCallback).toHaveBeenNthCalledWith(1, subsystem.DataSourceState.Initializing);
       expect(mockStatusCallback).toHaveBeenNthCalledWith(
         2,
-        subsystem.DataSourceState.Off,
+        subsystem.DataSourceState.Closed,
         new LDPollingError(
           DataSourceErrorKind.ErrorResponse,
           status === 401
             ? `Received error ${status} (invalid SDK key) for polling request - giving up permanently`
             : `Received error ${status} for polling request - giving up permanently`,
           status as number,
+          false,
         ),
       );
       setTimeout(() => {
diff --git a/packages/shared/sdk-server/__tests__/data_sources/StreamingProcessorFDv2.test.ts b/packages/shared/sdk-server/__tests__/data_sources/StreamingProcessorFDv2.test.ts
@@ -311,8 +311,13 @@ describe('given a stream processor with mock event source', () => {
       expect(willRetry).toBeFalsy();
       expect(mockStatusCallback).toHaveBeenNthCalledWith(
         2,
-        subsystem.DataSourceState.Off,
-        new LDStreamingError(DataSourceErrorKind.Unknown, testError.message, testError.status),
+        subsystem.DataSourceState.Closed,
+        new LDStreamingError(
+          DataSourceErrorKind.Unknown,
+          testError.message,
+          testError.status,
+          true,
+        ),
       );
       expect(logger.error).toBeCalledWith(
         expect.stringMatching(new RegExp(`${status}.*permanently`)),
diff --git a/packages/shared/sdk-server/src/data_sources/OneShotInitializerFDv2.ts b/packages/shared/sdk-server/src/data_sources/OneShotInitializerFDv2.ts
@@ -34,7 +34,7 @@ export default class OneShotInitializerFDv2 implements subsystemCommon.DataSourc
         const message = httpErrorMessage(err, 'initializer', 'initializer does not retry');
         this._logger?.error(message);
         statusCallback(
-          subsystemCommon.DataSourceState.Off,
+          subsystemCommon.DataSourceState.Closed,
           new LDPollingError(DataSourceErrorKind.ErrorResponse, message, status),
         );
         return;
diff --git a/packages/shared/sdk-server/src/data_sources/PollingProcessorFDv2.ts b/packages/shared/sdk-server/src/data_sources/PollingProcessorFDv2.ts
@@ -51,8 +51,8 @@ export default class PollingProcessorFDv2 implements subsystemCommon.DataSource
           const message = httpErrorMessage(err, 'polling request');
           this._logger?.error(message);
           statusCallback(
-            subsystemCommon.DataSourceState.Off,
-            new LDPollingError(DataSourceErrorKind.ErrorResponse, message, status),
+            subsystemCommon.DataSourceState.Closed,
+            new LDPollingError(DataSourceErrorKind.ErrorResponse, message, status, false),
           );
           // It is not recoverable, return and do not trigger another poll.
           return;
diff --git a/packages/shared/sdk-server/src/data_sources/StreamingProcessorFDv2.ts b/packages/shared/sdk-server/src/data_sources/StreamingProcessorFDv2.ts
@@ -82,8 +82,8 @@ export default class StreamingProcessorFDv2 implements subsystemCommon.DataSourc
       this._logger?.error(httpErrorMessage(err, 'streaming request'));
       this._logConnectionResult(false);
       statusCallback(
-        subsystemCommon.DataSourceState.Off,
-        new LDStreamingError(DataSourceErrorKind.ErrorResponse, err.message, err.status),
+        subsystemCommon.DataSourceState.Closed,
+        new LDStreamingError(DataSourceErrorKind.ErrorResponse, err.message, err.status, false),
       );
       return false;
     }
