Addressing review comments

Author: tanderson-ld

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

@@ -1,6 +1,5 @@
 import {
   CompositeDataSource,
-  Transition,
   TransitionConditions,
 } from '../../../src/api/subsystem/DataSystem/CompositeDataSource';
 import {
@@ -57,7 +56,7 @@ function makeZeroBackoff(): Backoff {
   };
 }
 
-it('initializer gets basis, switch to syncrhonizer', async () => {
+it('handles initializer getting basis, switching to syncrhonizer', async () => {
   const mockInitializer1 = {
     run: jest
       .fn()
@@ -112,7 +111,7 @@ it('initializer gets basis, switch to syncrhonizer', async () => {
   expect(callback).toHaveBeenNthCalledWith(2, false, { key: 'sync1' });
 });
 
-it('initializer gets basis, switch to synchronizer 1, fallback to synchronizer 2, recover to synchronizer 1', async () => {
+it('handles initializer getting basis, switches to synchronizer 1, falls back to synchronizer 2, recovers to synchronizer 1', async () => {
   const mockInitializer1: DataSystemInitializer = {
     run: jest
       .fn()
@@ -194,7 +193,7 @@ it('initializer gets basis, switch to synchronizer 1, fallback to synchronizer 2
   expect(callback).toHaveBeenNthCalledWith(3, false, { key: 'sync1' }); // sync2 recovers back to sync1
 });
 
-it('it reports error when all initializers fail', async () => {
+it('reports error when all initializers fail', async () => {
   const mockInitializer1Error = {
     name: 'Error',
     message: 'I am initializer1 error!',
@@ -271,7 +270,7 @@ it('it reports error when all initializers fail', async () => {
   expect(statusCallback).toHaveBeenCalledTimes(3);
 });
 
-it('it can be stopped when in thrashing synchronizer fallback loop', async () => {
+it('can be stopped when in thrashing synchronizer fallback loop', async () => {
   const mockInitializer1 = {
     run: jest
       .fn()
@@ -338,7 +337,7 @@ it('it can be stopped when in thrashing synchronizer fallback loop', async () =>
   expect(statusCallback).toHaveBeenNthCalledWith(2, DataSourceState.Closed, undefined);
 });
 
-it('it can be stopped and restarted', async () => {
+it('can be stopped and restarted', async () => {
   const mockInitializer1Data = { key: 'init1' };
   const mockInitializer1 = {
     run: jest
@@ -415,7 +414,7 @@ it('it can be stopped and restarted', async () => {
   expect(callback2).toHaveBeenCalledTimes(2);
 });
 
-it('it is well behaved with no initializers and no synchronizers configured', async () => {
+it('is well behaved with no initializers and no synchronizers configured', async () => {
   const underTest = new CompositeDataSource(
     [],
     [],
@@ -439,7 +438,7 @@ it('it is well behaved with no initializers and no synchronizers configured', as
   });
 });
 
-it('it is well behaved with an initializer and no synchronizers configured', async () => {
+it('is well behaved with an initializer and no synchronizers configured', async () => {
   const mockInitializer1 = {
     run: jest
       .fn()

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

@@ -17,7 +17,7 @@ export class CallbackHandler {
     this._disabled = true;
   }
 
-  async dataHanlder(basis: boolean, data: Data) {
+  async dataHandler(basis: boolean, data: Data) {
     if (this._disabled) {
       return;
     }

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

@@ -1,5 +1,6 @@
 /* eslint-disable no-await-in-loop */
 import { Backoff } from '../../../datasource/Backoff';
+import { LDLogger } from '../../logging';
 import { CallbackHandler } from './CallbackHandler';
 import {
   Data,
@@ -46,6 +47,7 @@ export class CompositeDataSource implements DataSource {
   private _stopped: boolean = true;
   private _externalTransitionPromise: Promise<TransitionRequest>;
   private _externalTransitionResolve?: (value: TransitionRequest) => void;
+  private _cancelTokens: (() => void)[] = [];
 
   /**
    * @param _initializers factories to create {@link DataSystemInitializer}s, in priority order.
@@ -56,6 +58,7 @@ export class CompositeDataSource implements DataSource {
     private readonly _synchronizers: SynchronizerFactory[],
     private readonly _transitionConditions: TransitionConditions,
     private readonly _backoff: Backoff,
+    private readonly _logger?: LDLogger,
   ) {
     this._externalTransitionPromise = new Promise<TransitionRequest>((resolveTransition) => {
       this._externalTransitionResolve = resolveTransition;
@@ -70,6 +73,7 @@ export class CompositeDataSource implements DataSource {
   ): Promise<void> {
     if (!this._stopped) {
       // don't allow multiple simultaneous runs
+      this._logger?.info('CompositeDataSource already running. Ignoring call to start.');
       return;
     }
     this._stopped = false;
@@ -116,6 +120,7 @@ export class CompositeDataSource implements DataSource {
                   const condition = this._lookupTransitionCondition(state, excludeRecovery);
                   if (condition) {
                     const { promise, cancel } = this._cancellableDelay(condition.durationMS);
+                    this._cancelTokens.push(cancel);
                     cancelScheduledTransition = cancel;
                     promise.then(() => {
                       callbackHandler.disable();
@@ -129,7 +134,7 @@ export class CompositeDataSource implements DataSource {
             },
           );
           currentDS.run(
-            (basis, data) => callbackHandler.dataHanlder(basis, data),
+            (basis, data) => callbackHandler.dataHandler(basis, data),
             (status, err) => callbackHandler.statusHandler(status, err),
           );
         } else {
@@ -156,9 +161,9 @@ export class CompositeDataSource implements DataSource {
       if (transitionRequest.err && transitionRequest.transition !== 'stop') {
         // if the transition was due to an error, throttle the transition
         const delay = this._backoff.fail();
-        const delayedTransition = new Promise((resolve) => {
-          setTimeout(resolve, delay);
-        }).then(() => transitionRequest);
+        const { promise, cancel } = this._cancellableDelay(delay);
+        this._cancelTokens.push(cancel);
+        const delayedTransition = promise.then(() => transitionRequest);
 
         // race the delayed transition and external transition requests to be responsive
         transitionRequest = await Promise.race([
@@ -170,6 +175,7 @@ export class CompositeDataSource implements DataSource {
       if (transitionRequest.transition === 'stop') {
         // exit the loop
         statusCallback(DataSourceState.Closed, transitionRequest.err);
+        lastTransition = transitionRequest.transition;
         break;
       }
 
@@ -181,6 +187,7 @@ export class CompositeDataSource implements DataSource {
   }
 
   async stop() {
+    this._cancelTokens.forEach((cancel) => cancel());
     this._externalTransitionResolve?.({ transition: 'stop' });
   }
 
@@ -254,19 +261,15 @@ export class CompositeDataSource implements DataSource {
 
   private _cancellableDelay = (delayMS: number) => {
     let timeout: ReturnType<typeof setTimeout> | undefined;
-    let reject: ((reason?: any) => void) | undefined;
-    const promise = new Promise((res, rej) => {
+    const promise = new Promise((res, _) => {
       timeout = setTimeout(res, delayMS);
-      reject = rej;
     });
     return {
       promise,
       cancel() {
         if (timeout) {
           clearTimeout(timeout);
-          reject?.();
           timeout = undefined;
-          reject = undefined;
         }
       },
     };