diff --git a/src/logger/constants.ts b/src/logger/constants.ts index ab711cac..ada74e13 100644 --- a/src/logger/constants.ts +++ b/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; diff --git a/src/logger/messages/info.ts b/src/logger/messages/info.ts index bf30a97c..a8746dec 100644 --- a/src/logger/messages/info.ts +++ b/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.'], diff --git a/src/sdkFactory/index.ts b/src/sdkFactory/index.ts index 61433a5e..5290d5c7 100644 --- a/src/sdkFactory/index.ts +++ b/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), diff --git a/src/storages/inLocalStorage/index.ts b/src/storages/inLocalStorage/index.ts index 0d35341b..68604e30 100644 --- a/src/storages/inLocalStorage/index.ts +++ b/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), diff --git a/src/storages/inMemory/ImpressionsCacheInMemory.ts b/src/storages/inMemory/ImpressionsCacheInMemory.ts index dcbf6b7a..57621c71 100644 --- a/src/storages/inMemory/ImpressionsCacheInMemory.ts +++ b/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(); + } } /** diff --git a/src/storages/inMemory/InMemoryStorage.ts b/src/storages/inMemory/InMemoryStorage.ts index 00d666fc..6f446f20 100644 --- a/src/storages/inMemory/InMemoryStorage.ts +++ b/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), diff --git a/src/storages/inMemory/InMemoryStorageCS.ts b/src/storages/inMemory/InMemoryStorageCS.ts index 7ac34ea8..4fa9b962 100644 --- a/src/storages/inMemory/InMemoryStorageCS.ts +++ b/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), diff --git a/src/storages/inMemory/__tests__/ImpressionsCacheInMemory.spec.ts b/src/storages/inMemory/__tests__/ImpressionsCacheInMemory.spec.ts index e5f7ed1b..4a50eaf8 100644 --- a/src/storages/inMemory/__tests__/ImpressionsCacheInMemory.spec.ts +++ b/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. }); diff --git a/src/storages/pluggable/index.ts b/src/storages/pluggable/index.ts index fef04aa6..e4062132 100644 --- a/src/storages/pluggable/index.ts +++ b/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 diff --git a/src/storages/types.ts b/src/storages/types.ts index 42e80a06..e699561b 100644 --- a/src/storages/types.ts +++ b/src/storages/types.ts @@ -298,7 +298,7 @@ export interface IRecorderCacheProducerSync { // @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 { export interface IImpressionsCacheSync extends IImpressionsCacheBase, IRecorderCacheProducerSync { track(data: ImpressionDTO[]): void + /* Registers callback for full queue */ + setOnFullQueueCb(cb: () => void): void } export interface IEventsCacheSync extends IEventsCacheBase, IRecorderCacheProducerSync { 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) */, diff --git a/src/sync/submitters/eventsSyncTask.ts b/src/sync/submitters/eventsSyncTask.ts index 3a15a6e0..151607ea 100644 --- a/src/sync/submitters/eventsSyncTask.ts +++ b/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(); }); diff --git a/src/sync/submitters/impressionsSyncTask.ts b/src/sync/submitters/impressionsSyncTask.ts index cf6b6068..5947c40c 100644 --- a/src/sync/submitters/impressionsSyncTask.ts +++ b/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; } diff --git a/src/types.ts b/src/types.ts index 46451a60..1e3654a8 100644 --- a/src/types.ts +++ b/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 diff --git a/src/utils/settingsValidation/__tests__/settings.mocks.ts b/src/utils/settingsValidation/__tests__/settings.mocks.ts index 80b0392e..3cb458cb 100644 --- a/src/utils/settingsValidation/__tests__/settings.mocks.ts +++ b/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: { diff --git a/src/utils/settingsValidation/index.ts b/src/utils/settingsValidation/index.ts index 3d535d57..20bcd1bc 100644 --- a/src/utils/settingsValidation/index.ts +++ b/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, },