Streaming updates: refactor settings validation and polishing

src/readiness/__tests__/readinessManager.spec.ts

@@ -1,6 +1,7 @@
-import { readinessManagerFactory, SDK_READY, SDK_READY_FROM_CACHE, SDK_UPDATE, SDK_READY_TIMED_OUT, SDK_SPLITS_CACHE_LOADED, SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED } from '../readinessManager';
+import { readinessManagerFactory } from '../readinessManager';
 import EventEmitter from '../../utils/MinEvents';
 import { IReadinessManager } from '../types';
+import { SDK_READY, SDK_UPDATE, SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED, SDK_READY_FROM_CACHE, SDK_SPLITS_CACHE_LOADED, SDK_READY_TIMED_OUT } from '../constants';
 
 const timeoutMs = 100;
 const statusFlagsCount = 5;

src/readiness/__tests__/sdkReadinessManager.spec.ts

@@ -1,7 +1,7 @@
 // @ts-nocheck
 import { loggerMock, mockClear } from '../../logger/__tests__/sdkLogger.mock';
 import { IEventEmitter } from '../../types';
-import { SDK_READY, SDK_READY_FROM_CACHE, SDK_READY_TIMED_OUT, SDK_UPDATE } from '../readinessManager';
+import { SDK_READY, SDK_READY_FROM_CACHE, SDK_READY_TIMED_OUT, SDK_UPDATE } from '../constants';
 import sdkReadinessManagerFactory from '../sdkReadinessManager';
 import { IReadinessManager } from '../types';
 
@@ -97,6 +97,9 @@ describe('SDK Readiness Manager - Event emitter', () => {
     expect(loggerMock.warn.mock.calls.length).toBe(1); // If the SDK_READY event fires and we have no callbacks for it (neither event nor ready promise) we get a warning.
     expect(loggerMock.warn.mock.calls[0]).toEqual(['No listeners for SDK Readiness detected. Incorrect control treatments could have been logged if you called getTreatment/s while the SDK was not yet ready.']); // Telling us there were no listeners and evaluations before this point may have been incorrect.
 
+    expect(loggerMock.info.mock.calls.length).toBe(1); // If the SDK_READY event fires, we get a info message.
+    expect(loggerMock.info.mock.calls[0]).toEqual(['Split SDK is ready.']); // Telling us the SDK is ready.
+
     // Now it's marked as ready.
     addListenerCB('this event we do not care');
     expect(loggerMock.error.mock.calls.length).toBe(0); // Now if we add a listener to an event unrelated with readiness, we get no errors logged.
@@ -122,6 +125,9 @@ describe('SDK Readiness Manager - Event emitter', () => {
     emitReadyEvent(sdkReadinessManager.readinessManager);
     expect(loggerMock.warn.mock.calls.length).toBe(0); // As we had at least one listener, we get no warnings.
     expect(loggerMock.error.mock.calls.length).toBe(0); // As we had at least one listener, we get no errors.
+
+    expect(loggerMock.info.mock.calls.length).toBe(1); // If the SDK_READY event fires, we get a info message.
+    expect(loggerMock.info.mock.calls[0]).toEqual(['Split SDK is ready.']); // Telling us the SDK is ready.
   });
 
   test('The event callbacks should work as expected - If we end up removing the listeners for SDK_READY, it behaves as if it had none', () => {

src/readiness/constants.ts

@@ -0,0 +1,12 @@
+// Splits events:
+export const SDK_SPLITS_ARRIVED = 'SDK_SPLITS_ARRIVED';
+export const SDK_SPLITS_CACHE_LOADED = 'SDK_SPLITS_CACHE_LOADED';
+
+// Segments events:
+export const SDK_SEGMENTS_ARRIVED = 'SDK_SEGMENTS_ARRIVED';
+
+// Readiness events:
+export const SDK_READY_TIMED_OUT = 'init::timeout';
+export const SDK_READY = 'init::ready';
+export const SDK_READY_FROM_CACHE = 'init::cache-ready';
+export const SDK_UPDATE = 'state::update';

src/readiness/readinessManager.ts

@@ -1,27 +1,18 @@
 import objectAssign from 'object-assign';
 import { IEventEmitter } from '../types';
+import { SDK_SPLITS_ARRIVED, SDK_SPLITS_CACHE_LOADED, SDK_SEGMENTS_ARRIVED, SDK_READY_TIMED_OUT, SDK_READY_FROM_CACHE, SDK_UPDATE, SDK_READY } from './constants';
 import { IReadinessEventEmitter, IReadinessManager, ISegmentsEventEmitter, ISplitsEventEmitter } from './types';
 
-// Splits events:
-export const SDK_SPLITS_ARRIVED = 'SDK_SPLITS_ARRIVED';
-export const SDK_SPLITS_CACHE_LOADED = 'SDK_SPLITS_CACHE_LOADED';
-
-// Segments events:
-export const SDK_SEGMENTS_ARRIVED = 'SDK_SEGMENTS_ARRIVED';
-
-// Readiness events:
-export const SDK_READY_TIMED_OUT = 'init::timeout';
-export const SDK_READY = 'init::ready';
-export const SDK_READY_FROM_CACHE = 'init::cache-ready';
-export const SDK_UPDATE = 'state::update';
-
 function splitsEventEmitterFactory(EventEmitter: new () => IEventEmitter): ISplitsEventEmitter {
   const splitsEventEmitter = objectAssign(new EventEmitter(), {
     splitsArrived: false,
     splitsCacheLoaded: false,
   });
 
-  splitsEventEmitter.once(SDK_SPLITS_ARRIVED, () => { splitsEventEmitter.splitsArrived = true; });
+  // `isSplitKill` condition avoids an edge-case of wrongly emitting SDK_READY if:
+  // - `/mySegments` fetch and SPLIT_KILL occurs before `/splitChanges` fetch, and
+  // - storage has cached splits (for which case `splitsStorage.killLocally` can return true)
+  splitsEventEmitter.on(SDK_SPLITS_ARRIVED, (isSplitKill) => { if (!isSplitKill) splitsEventEmitter.splitsArrived = true; });
   splitsEventEmitter.once(SDK_SPLITS_CACHE_LOADED, () => { splitsEventEmitter.splitsCacheLoaded = true; });
 
   return splitsEventEmitter;

src/readiness/sdkReadinessManager.ts

@@ -1,15 +1,10 @@
 import objectAssign from 'object-assign';
 import promiseWrapper from '../utils/promise/wrapper';
-import {
-  readinessManagerFactory,
-  SDK_READY,
-  SDK_READY_FROM_CACHE,
-  SDK_UPDATE,
-  SDK_READY_TIMED_OUT
-} from './readinessManager';
+import { readinessManagerFactory } from './readinessManager';
 import { ISdkReadinessManager } from './types';
 import { IEventEmitter } from '../types';
 import { logFactory } from '../logger/sdkLogger';
+import { SDK_READY, SDK_READY_TIMED_OUT, SDK_READY_FROM_CACHE, SDK_UPDATE } from './constants';
 const log = logFactory('');
 
 const NEW_LISTENER_EVENT = 'newListener';
@@ -61,6 +56,8 @@ export default function sdkReadinessManagerFactory(
   function generateReadyPromise() {
     const promise = promiseWrapper(new Promise<void>((resolve, reject) => {
       readinessManager.gate.once(SDK_READY, () => {
+        log.info('Split SDK is ready.');
+
         if (readyCbCount === internalReadyCbCount && !promise.hasOnFulfilled()) log.warn('No listeners for SDK Readiness detected. Incorrect control treatments could have been logged if you called getTreatment/s while the SDK was not yet ready.');
         resolve();
       });

src/storages/AbstractSplitsCacheSync.ts

@@ -66,6 +66,16 @@ export default abstract class AbstractSplitsCacheSync implements ISplitsCacheSyn
     return this.getChangeNumber() > -1;
   }
 
+  /**
+   * Kill `name` split and set `defaultTreatment` and `changeNumber`.
+   * Used for SPLIT_KILL push notifications.
+   *
+   * @param {string} name
+   * @param {string} defaultTreatment
+   * @param {number} changeNumber
+   * @returns {Promise} a promise that is resolved once the split kill is performed. The fulfillment value is a boolean: `true` if the kill success updating the split or `false` if no split is updated,
+   * for instance, if the `changeNumber` is old, or if the split is not found (e.g., `/splitchanges` hasn't been fetched yet), or if the storage fails to apply the update.
+   */
   killLocally(name: string, defaultTreatment: string, changeNumber: number): boolean {
     const split = this.getSplit(name);
 

src/storages/inRedis/index.ts

@@ -9,6 +9,7 @@ import EventsCacheInRedis from './EventsCacheInRedis';
 import LatenciesCacheInRedis from './LatenciesCacheInRedis';
 import CountsCacheInRedis from './CountsCacheInRedis';
 import { UNKNOWN } from '../../utils/constants';
+import { SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED } from '../../readiness/constants';
 
 export interface InRedisStorageOptions {
   prefix?: string
@@ -41,8 +42,8 @@ export function InRedisStorage(options: InRedisStorageOptions = {}) {
     // subscription to Redis connect event in order to emit SDK_READY event
     // @TODO pass a callback to simplify custom storages
     redisClient.on('connect', () => {
-      params.readinessManager.splits.emit('SDK_SPLITS_ARRIVED');
-      params.readinessManager.segments.emit('SDK_SEGMENTS_ARRIVED');
+      params.readinessManager.splits.emit(SDK_SPLITS_ARRIVED);
+      params.readinessManager.segments.emit(SDK_SEGMENTS_ARRIVED);
     });
 
     return {

src/sync/offline/syncTasks/fromObjectSyncTask.ts

@@ -8,6 +8,7 @@ import syncTaskFactory from '../../syncTask';
 import { ISyncTask } from '../../types';
 import { ISettings } from '../../../types';
 import { CONTROL } from '../../../utils/constants';
+import { SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED } from '../../../readiness/constants';
 const log = logFactory('splitio-producer:offline');
 
 /**
@@ -55,8 +56,8 @@ export function fromObjectUpdaterFactory(
         storage.splits.clear(),
         storage.splits.addSplits(splits)
       ]).then(() => {
-        readiness.splits.emit('SDK_SPLITS_ARRIVED');
-        readiness.segments.emit('SDK_SEGMENTS_ARRIVED');
+        readiness.splits.emit(SDK_SPLITS_ARRIVED);
+        readiness.segments.emit(SDK_SEGMENTS_ARRIVED);
         return true;
       });
     } else {

src/sync/polling/pollingManagerCS.ts

@@ -8,6 +8,7 @@ import mySegmentsSyncTaskFactory from './syncTasks/mySegmentsSyncTask';
 import splitsSyncTaskFactory from './syncTasks/splitsSyncTask';
 import { ISettings } from '../../types';
 import { getMatching } from '../../utils/key';
+import { SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED } from '../../readiness/constants';
 const log = logFactory('splitio-sync:polling-manager');
 
 /**
@@ -42,7 +43,7 @@ export default function pollingManagerCSFactory(
   }
 
   // smart pausing
-  readiness.splits.on('SDK_SPLITS_ARRIVED', () => {
+  readiness.splits.on(SDK_SPLITS_ARRIVED, () => {
     if (!splitsSyncTask.isRunning()) return; // noop if not doing polling
     const splitsHaveSegments = storage.splits.usesSegments();
     if (splitsHaveSegments !== mySegmentsSyncTask.isRunning()) {
@@ -60,10 +61,10 @@ export default function pollingManagerCSFactory(
 
     // smart ready
     function smartReady() {
-      if (!readiness.isReady() && !storage.splits.usesSegments()) readiness.segments.emit('SDK_SEGMENTS_ARRIVED');
+      if (!readiness.isReady() && !storage.splits.usesSegments()) readiness.segments.emit(SDK_SEGMENTS_ARRIVED);
     }
     if (!storage.splits.usesSegments()) setTimeout(smartReady, 0);
-    else readiness.splits.once('SDK_SPLITS_ARRIVED', smartReady);
+    else readiness.splits.once(SDK_SPLITS_ARRIVED, smartReady);
 
     mySegmentsSyncTasks[matchingKey] = mySegmentsSyncTask;
     return mySegmentsSyncTask;

src/sync/polling/types.ts

@@ -4,9 +4,9 @@ import { IStorageSync } from '../../storages/types';
 import { ISettings } from '../../types';
 import { ITask, ISyncTask } from '../types';
 
-export interface ISplitsSyncTask extends ISyncTask<[], boolean> { }
+export interface ISplitsSyncTask extends ISyncTask<[noCache?: boolean], boolean> { }
 
-export interface ISegmentsSyncTask extends ISyncTask<[string[]?], boolean> { }
+export interface ISegmentsSyncTask extends ISyncTask<[segmentNames?: string[], noCache?: boolean, fetchOnlyNew?: boolean], boolean> { }
 
 export interface IPollingManager extends ITask {
   syncAll(): Promise<any>

src/sync/syncManagerOffline.ts

@@ -3,6 +3,7 @@ import fromObjectSyncTaskFactory from './offline/syncTasks/fromObjectSyncTask';
 import objectAssign from 'object-assign';
 import { ISplitsParser } from './offline/splitsParser/types';
 import { IReadinessManager } from '../readiness/types';
+import { SDK_SEGMENTS_ARRIVED } from '../readiness/constants';
 
 function flush() {
   return Promise.resolve();
@@ -40,7 +41,7 @@ export function syncManagerOfflineFactory(
               // In LOCALHOST mode, shared clients are ready in the next event cycle than created
               // SDK_READY cannot be emitted directly because this will not update the readiness status
               setTimeout(() => {
-                readinessManager.segments.emit('SDK_SEGMENTS_ARRIVED'); // SDK_SPLITS_ARRIVED emitted by main SyncManager
+                readinessManager.segments.emit(SDK_SEGMENTS_ARRIVED); // SDK_SPLITS_ARRIVED emitted by main SyncManager
               }, 0);
             },
             stop() { },

src/types.ts

@@ -72,8 +72,7 @@ export interface ISettings {
     offlineRefreshRate: number,
     eventsPushRate: number,
     eventsQueueSize: number,
-    authRetryBackoffBase: number,
-    streamingReconnectBackoffBase: number
+    pushRetryBackoffBase: number
   },
   readonly startup: {
     readyTimeout: number,
@@ -279,19 +278,12 @@ interface INodeBasicSettings extends ISharedSettings {
      */
     offlineRefreshRate?: number
     /**
-     * When using streaming mode, seconds to wait before re attempting to authenticate for push notifications.
+     * When using streaming mode, seconds to wait before re attempting to connect for push notifications.
      * Next attempts follow intervals in power of two: base seconds, base x 2 seconds, base x 4 seconds, ...
-     * @property {number} authRetryBackoffBase
+     * @property {number} pushRetryBackoffBase
      * @default 1
      */
-    authRetryBackoffBase?: number,
-    /**
-     * When using streaming mode, seconds to wait before re attempting to connect to streaming.
-     * Next attempts follow intervals in power of two: base seconds, base x 2 seconds, base x 4 seconds, ...
-     * @property {number} streamingReconnectBackoffBase
-     * @default 1
-     */
-    streamingReconnectBackoffBase?: number,
+    pushRetryBackoffBase?: number,
   },
   /**
    * SDK Core settings for NodeJS.
@@ -349,7 +341,6 @@ export interface IStatusInterface extends IEventEmitter {
   /**
    * Returns a promise that will be resolved once the SDK has finished loading.
    * @function ready
-   * @deprecated Use on(sdk.Event.SDK_READY, callback: () => void) instead.
    * @returns {Promise<void>}
    */
   ready(): Promise<void>
@@ -776,19 +767,12 @@ export namespace SplitIO {
        */
       offlineRefreshRate?: number
       /**
-       * When using streaming mode, seconds to wait before re attempting to authenticate for push notifications.
-       * Next attempts follow intervals in power of two: base seconds, base x 2 seconds, base x 4 seconds, ...
-       * @property {number} authRetryBackoffBase
-       * @default 1
-       */
-      authRetryBackoffBase?: number,
-      /**
-       * When using streaming mode, seconds to wait before re attempting to connect to streaming.
+       * When using streaming mode, seconds to wait before re attempting to connect for push notifications.
        * Next attempts follow intervals in power of two: base seconds, base x 2 seconds, base x 4 seconds, ...
-       * @property {number} streamingReconnectBackoffBase
+       * @property {number} pushRetryBackoffBase
        * @default 1
        */
-      streamingReconnectBackoffBase?: number,
+      pushRetryBackoffBase?: number,
     },
     /**
      * SDK Core settings for the browser.

src/utils/settingsValidation/__tests__/settings.mocks.ts

@@ -47,8 +47,7 @@ export const fullSettings: ISettings = {
     offlineRefreshRate: 1,
     eventsPushRate: 1,
     eventsQueueSize: 1,
-    authRetryBackoffBase: 1,
-    streamingReconnectBackoffBase: 1
+    pushRetryBackoffBase: 1
   },
   startup: {
     readyTimeout: 1,

src/utils/settingsValidation/index.ts

@@ -39,10 +39,8 @@ const base = {
     eventsPushRate: 60,
     // how many events will be queued before flushing
     eventsQueueSize: 500,
-    // backoff base seconds to wait before re attempting to authenticate for push notifications
-    authRetryBackoffBase: 1,
-    // backoff base seconds to wait before re attempting to connect to streaming
-    streamingReconnectBackoffBase: 1
+    // backoff base seconds to wait before re attempting to connect to push notifications
+    pushRetryBackoffBase: 1,
   },
 
   urls: {
@@ -152,8 +150,7 @@ export function settingsValidation(config: unknown, validationParams: ISettingsV
     withDefaults.streamingEnabled = true;
     // Backoff bases.
     // We are not checking if bases are positive numbers. Thus, we might be reauthenticating immediately (`setTimeout` with NaN or negative number)
-    withDefaults.scheduler.authRetryBackoffBase = fromSecondsToMillis(withDefaults.scheduler.authRetryBackoffBase);
-    withDefaults.scheduler.streamingReconnectBackoffBase = fromSecondsToMillis(withDefaults.scheduler.streamingReconnectBackoffBase);
+    withDefaults.scheduler.pushRetryBackoffBase = fromSecondsToMillis(withDefaults.scheduler.pushRetryBackoffBase);
   }
 
   // validate the `splitFilters` settings and parse splits query