Add impressions queue size limit functionality

src/logger/constants.ts

@@ -61,7 +61,7 @@ export const STREAMING_RECONNECT = 111;
 export const STREAMING_CONNECTING = 112;
 export const STREAMING_DISABLED = 113;
 export const STREAMING_DISCONNECTING = 114;
-export const SUBMITTERS_PUSH_FULL_EVENTS_QUEUE = 115;
+export const SUBMITTERS_PUSH_FULL_QUEUE = 115;
 export const SUBMITTERS_PUSH = 116;
 export const SYNC_START_POLLING = 117;
 export const SYNC_CONTINUE_POLLING = 118;

src/logger/messages/info.ts

@@ -20,7 +20,7 @@ export const codesInfo: [number, string][] = codesWarn.concat([
   [c.POLLING_START, c.LOG_PREFIX_SYNC_POLLING + 'Starting polling'],
   [c.POLLING_STOP, c.LOG_PREFIX_SYNC_POLLING + 'Stopping polling'],
   [c.SYNC_SPLITS_FETCH_RETRY, c.LOG_PREFIX_SYNC_SPLITS + 'Retrying download of splits #%s. Reason: %s'],
-  [c.SUBMITTERS_PUSH_FULL_EVENTS_QUEUE, c.LOG_PREFIX_SYNC_SUBMITTERS + 'Flushing full events queue and reseting timer.'],
+  [c.SUBMITTERS_PUSH_FULL_QUEUE, c.LOG_PREFIX_SYNC_SUBMITTERS + 'Flushing full %s queue and reseting timer.'],
   [c.SUBMITTERS_PUSH, c.LOG_PREFIX_SYNC_SUBMITTERS + 'Pushing %s %s.'],
   [c.STREAMING_REFRESH_TOKEN, c.LOG_PREFIX_SYNC_STREAMING + 'Refreshing streaming token in %s seconds, and connecting streaming in %s seconds.'],
   [c.STREAMING_RECONNECT, c.LOG_PREFIX_SYNC_STREAMING + 'Attempting to reconnect streaming in %s seconds.'],

src/sdkFactory/index.ts

@@ -33,6 +33,7 @@ export function sdkFactory(params: ISdkFactoryParams): SplitIO.ICsSDK | SplitIO.
 
   // @TODO consider passing the settings object, so that each storage access only what it needs
   const storageFactoryParams: IStorageFactoryParams = {
+    impressionsQueueSize: settings.scheduler.impressionsQueueSize,
     eventsQueueSize: settings.scheduler.eventsQueueSize,
     optimize: shouldBeOptimized(settings),
 

src/storages/inLocalStorage/index.ts

@@ -40,7 +40,7 @@ export function InLocalStorage(options: InLocalStorageOptions = {}): IStorageSyn
     return {
       splits: new SplitsCacheInLocal(log, keys, expirationTimestamp, params.splitFiltersValidation),
       segments: new MySegmentsCacheInLocal(log, keys),
-      impressions: new ImpressionsCacheInMemory(),
+      impressions: new ImpressionsCacheInMemory(params.impressionsQueueSize),
       impressionCounts: params.optimize ? new ImpressionCountsCacheInMemory() : undefined,
       events: new EventsCacheInMemory(params.eventsQueueSize),
 

src/storages/inMemory/ImpressionsCacheInMemory.ts

@@ -3,13 +3,34 @@ import { ImpressionDTO } from '../../types';
 
 export class ImpressionsCacheInMemory implements IImpressionsCacheSync {
 
-  private queue: ImpressionDTO[] = [];
+  private onFullQueue?: () => void;
+  private readonly maxQueue: number;
+  private queue: ImpressionDTO[];
+
+  /**
+   *
+   * @param impressionsQueueSize number of queued impressions to call onFullQueueCb.
+   * Default value is 0, that means no maximum value, in case we want to avoid this being triggered.
+   */
+  constructor(impressionsQueueSize: number = 0) {
+    this.maxQueue = impressionsQueueSize;
+    this.queue = [];
+  }
+
+  setOnFullQueueCb(cb: () => void) {
+    this.onFullQueue = cb;
+  }
 
   /**
    * Store impressions in sequential order
    */
   track(data: ImpressionDTO[]) {
     this.queue.push(...data);
+
+    // Check if the cache queue is full and we need to flush it.
+    if (this.maxQueue > 0 && this.queue.length >= this.maxQueue && this.onFullQueue) {
+      this.onFullQueue();
+    }
   }
 
   /**

src/storages/inMemory/InMemoryStorage.ts

@@ -16,7 +16,7 @@ export function InMemoryStorageFactory(params: IStorageFactoryParams): IStorageS
   return {
     splits: new SplitsCacheInMemory(),
     segments: new SegmentsCacheInMemory(),
-    impressions: new ImpressionsCacheInMemory(),
+    impressions: new ImpressionsCacheInMemory(params.impressionsQueueSize),
     impressionCounts: params.optimize ? new ImpressionCountsCacheInMemory() : undefined,
     events: new EventsCacheInMemory(params.eventsQueueSize),
 

src/storages/inMemory/InMemoryStorageCS.ts

@@ -16,7 +16,7 @@ export function InMemoryStorageCSFactory(params: IStorageFactoryParams): IStorag
   return {
     splits: new SplitsCacheInMemory(),
     segments: new MySegmentsCacheInMemory(),
-    impressions: new ImpressionsCacheInMemory(),
+    impressions: new ImpressionsCacheInMemory(params.impressionsQueueSize),
     impressionCounts: params.optimize ? new ImpressionCountsCacheInMemory() : undefined,
     events: new EventsCacheInMemory(params.eventsQueueSize),
 

src/storages/inMemory/__tests__/ImpressionsCacheInMemory.spec.ts

@@ -1,13 +1,56 @@
-
+// @ts-nocheck
 import { ImpressionsCacheInMemory } from '../ImpressionsCacheInMemory';
 
-test('IMPRESSIONS CACHE IN MEMORY / should incrementally store values', () => {
+test('IMPRESSIONS CACHE IN MEMORY / should incrementally store values, clear the queue, and tell if it is empty', () => {
   const c = new ImpressionsCacheInMemory();
 
-  // @ts-expect-error
-  c.track([0]); // @ts-expect-error
-  c.track([1, 2]); // @ts-expect-error
+  // queue is initially empty
+  expect(c.state()).toEqual([]);
+  expect(c.isEmpty()).toBe(true);
+
+
+  c.track([0]);
+  c.track([1, 2]);
   c.track([3]);
 
   expect(c.state()).toEqual([0, 1, 2, 3]); // all the items should be stored in sequential order
+  expect(c.isEmpty()).toBe(false);
+
+  // should empty the queue
+  c.clear();
+  expect(c.state()).toEqual([]);
+  expect(c.isEmpty()).toBe(true);
+});
+
+test('IMPRESSIONS CACHE IN MEMORY / Should call "onFullQueueCb" when the queue is full.', () => {
+  let cbCalled = 0;
+  const cache = new ImpressionsCacheInMemory(3); // small impressionsQueueSize to be reached
+  cache.setOnFullQueueCb(() => { cbCalled++; cache.clear(); });
+
+  cache.track([0]);
+  cache.track([0]);
+  expect(cbCalled).toBe(0); // if the queue is not full, it will not run the callback.
+  cache.track([0]);
+  expect(cbCalled).toBe(1); // if we had the queue full, it should flush events.
+  cache.track([1]);
+  expect(cbCalled).toBe(1); // After that, while the queue is below max size, it should not try to flush it.
+  cache.track([2]);
+  expect(cbCalled).toBe(1); // After that, while the queue is below max size, it should not try to flush it.
+  cache.track([3]);
+  expect(cbCalled).toBe(2); // Once we get to the max size, it should try to flush it.
+  cache.track([4]);
+  expect(cbCalled).toBe(2); // And it should not flush again,
+  cache.track([5]);
+  expect(cbCalled).toBe(2); // And it should not flush again,
+  cache.track([6]);
+  expect(cbCalled).toBe(3); // Until the queue is filled with events again.
+});
+
+test('IMPRESSIONS CACHE IN MEMORY / Should not throw if the "onFullQueueCb" callback was not provided.', () => {
+  const cache = new ImpressionsCacheInMemory(3); // small eventsQueueSize to be reached
+
+  cache.track([0]);
+  cache.track([1]); // Cache still not full,
+  expect(cache.track.bind(cache, [2])).not.toThrow(); // but when it is full, as 'onFullQueueCb' was not provided, nothing happens but no exceptions are thrown.
+  expect(cache.track.bind(cache, [3])).not.toThrow(); // but when it is full, as 'onFullQueueCb' was not provided, nothing happens but no exceptions are thrown.
 });

src/storages/pluggable/index.ts

@@ -63,7 +63,7 @@ export function PluggableStorage(options: PluggableStorageOptions): IStorageAsyn
 
   const prefix = validatePrefix(options.prefix);
 
-  function PluggableStorageFactory({ log, metadata, onReadyCb, mode, eventsQueueSize, optimize }: IStorageFactoryParams): IStorageAsync {
+  function PluggableStorageFactory({ log, metadata, onReadyCb, mode, eventsQueueSize, impressionsQueueSize, optimize }: IStorageFactoryParams): IStorageAsync {
     const keys = new KeyBuilderSS(prefix, metadata);
     const wrapper = wrapperAdapter(log, options.wrapper);
     const isPartialConsumer = mode === CONSUMER_PARTIAL_MODE;
@@ -74,7 +74,7 @@ export function PluggableStorage(options: PluggableStorageOptions): IStorageAsyn
     return {
       splits: new SplitsCachePluggable(log, keys, wrapper),
       segments: new SegmentsCachePluggable(log, keys, wrapper),
-      impressions: isPartialConsumer ? new ImpressionsCacheInMemory() : new ImpressionsCachePluggable(log, keys.buildImpressionsKey(), wrapper, metadata),
+      impressions: isPartialConsumer ? new ImpressionsCacheInMemory(impressionsQueueSize) : new ImpressionsCachePluggable(log, keys.buildImpressionsKey(), wrapper, metadata),
       impressionCounts: optimize ? new ImpressionCountsCacheInMemory() : undefined,
       events: isPartialConsumer ? promisifyEventsTrack(new EventsCacheInMemory(eventsQueueSize)) : new EventsCachePluggable(log, keys.buildEventsKey(), wrapper, metadata),
       // @TODO add telemetry cache when required

src/storages/types.ts

@@ -298,7 +298,7 @@ export interface IRecorderCacheProducerSync<T> {
   // @TODO names are inconsistent with spec
   /* Checks if cache is empty. Returns true if the cache was just created or cleared */
   isEmpty(): boolean
-  /* clears cache data */
+  /* Clears cache data */
   clear(): void
   /* Gets cache data */
   state(): T
@@ -307,10 +307,13 @@ export interface IRecorderCacheProducerSync<T> {
 
 export interface IImpressionsCacheSync extends IImpressionsCacheBase, IRecorderCacheProducerSync<ImpressionDTO[]> {
   track(data: ImpressionDTO[]): void
+  /* Registers callback for full queue */
+  setOnFullQueueCb(cb: () => void): void
 }
 
 export interface IEventsCacheSync extends IEventsCacheBase, IRecorderCacheProducerSync<SplitIO.EventData[]> {
   track(data: SplitIO.EventData, size?: number): boolean
+  /* Registers callback for full queue */
   setOnFullQueueCb(cb: () => void): void
 }
 
@@ -423,6 +426,7 @@ export type DataLoader = (storage: IStorageSync, matchingKey: string) => void
 
 export interface IStorageFactoryParams {
   log: ILogger,
+  impressionsQueueSize?: number,
   eventsQueueSize?: number,
   optimize?: boolean /* whether create the `impressionCounts` cache (OPTIMIZED impression mode) or not (DEBUG impression mode) */,
 

src/sync/submitters/eventsSyncTask.ts

@@ -3,7 +3,9 @@ import { IPostEventsBulk } from '../../services/types';
 import { ISyncTask, ITimeTracker } from '../types';
 import { submitterSyncTaskFactory } from './submitterSyncTask';
 import { ILogger } from '../../logger/types';
-import { SUBMITTERS_PUSH_FULL_EVENTS_QUEUE } from '../../logger/constants';
+import { SUBMITTERS_PUSH_FULL_QUEUE } from '../../logger/constants';
+
+const DATA_NAME = 'events';
 
 /**
  * Sync task that periodically posts tracked events
@@ -18,7 +20,7 @@ export function eventsSyncTaskFactory(
 ): ISyncTask {
 
   // don't retry events.
-  const syncTask = submitterSyncTaskFactory(log, postEventsBulk, eventsCache, eventsPushRate, 'queued events', latencyTracker);
+  const syncTask = submitterSyncTaskFactory(log, postEventsBulk, eventsCache, eventsPushRate, DATA_NAME, latencyTracker);
 
   // Set a timer for the first push of events,
   if (eventsFirstPushWindow > 0) {
@@ -34,9 +36,9 @@ export function eventsSyncTaskFactory(
     };
   }
 
-  // register eventsSubmitter to be executed when events cache is full
+  // register events submitter to be executed when events cache is full
   eventsCache.setOnFullQueueCb(() => {
-    log.info(SUBMITTERS_PUSH_FULL_EVENTS_QUEUE);
+    log.info(SUBMITTERS_PUSH_FULL_QUEUE, [DATA_NAME]);
     syncTask.execute();
   });
 

src/sync/submitters/impressionsSyncTask.ts

@@ -6,6 +6,9 @@ import { ImpressionDTO } from '../../types';
 import { submitterSyncTaskFactory } from './submitterSyncTask';
 import { ImpressionsPayload } from './types';
 import { ILogger } from '../../logger/types';
+import { SUBMITTERS_PUSH_FULL_QUEUE } from '../../logger/constants';
+
+const DATA_NAME = 'impressions';
 
 /**
  * Converts `impressions` data from cache into request payload.
@@ -50,5 +53,13 @@ export function impressionsSyncTaskFactory(
 ): ISyncTask {
 
   // retry impressions only once.
-  return submitterSyncTaskFactory(log, postTestImpressionsBulk, impressionsCache, impressionsRefreshRate, 'impressions', latencyTracker, fromImpressionsCollector.bind(undefined, sendLabels), 1);
+  const syncTask = submitterSyncTaskFactory(log, postTestImpressionsBulk, impressionsCache, impressionsRefreshRate, DATA_NAME, latencyTracker, fromImpressionsCollector.bind(undefined, sendLabels), 1);
+
+  // register impressions submitter to be executed when impressions cache is full
+  impressionsCache.setOnFullQueueCb(() => {
+    log.info(SUBMITTERS_PUSH_FULL_QUEUE, [DATA_NAME]);
+    syncTask.execute();
+  });
+
+  return syncTask;
 }

src/types.ts

@@ -73,6 +73,7 @@ export interface ISettings {
   readonly scheduler: {
     featuresRefreshRate: number,
     impressionsRefreshRate: number,
+    impressionsQueueSize: number,
     metricsRefreshRate: number,
     segmentsRefreshRate: number,
     offlineRefreshRate: number,
@@ -255,6 +256,13 @@ interface INodeBasicSettings extends ISharedSettings {
      * @default 300
      */
     impressionsRefreshRate?: number,
+    /**
+     * The maximum number of impression items we want to queue. If we queue more values, it will trigger a flush and reset the timer.
+     * If you use a 0 here, the queue will have no maximum size.
+     * @property {number} impressionsQueueSize
+     * @default 30000
+     */
+    impressionsQueueSize?: number,
     /**
      * The SDK sends diagnostic metrics to Split servers. This parameters controls this metric flush period in seconds.
      * @property {number} metricsRefreshRate
@@ -769,6 +777,13 @@ export namespace SplitIO {
        * @default 60
        */
       impressionsRefreshRate?: number,
+      /**
+       * The maximum number of impression items we want to queue. If we queue more values, it will trigger a flush and reset the timer.
+       * If you use a 0 here, the queue will have no maximum size.
+       * @property {number} impressionsQueueSize
+       * @default 30000
+       */
+      impressionsQueueSize?: number,
       /**
        * The SDK sends diagnostic metrics to Split servers. This parameters controls this metric flush period in seconds.
        * @property {number} metricsRefreshRate

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

@@ -52,6 +52,7 @@ export const fullSettings: ISettings = {
     offlineRefreshRate: 1,
     eventsPushRate: 1,
     eventsQueueSize: 1,
+    impressionsQueueSize: 1,
     pushRetryBackoffBase: 1
   },
   startup: {

src/utils/settingsValidation/index.ts

@@ -38,6 +38,8 @@ const base = {
     eventsPushRate: 60,
     // how many events will be queued before flushing
     eventsQueueSize: 500,
+    // how many impressions will be queued before flushing
+    impressionsQueueSize: 30000,
     // backoff base seconds to wait before re attempting to connect to push notifications
     pushRetryBackoffBase: 1,
   },