From 48d8b270d9848242b449f5a62c27ae9b8d5fbc5e Mon Sep 17 00:00:00 2001 From: Emiliano Sanchez Date: Mon, 13 Jun 2022 10:53:21 -0300 Subject: [PATCH 1/2] Update SyncTask to run executions secuentially to avoid race conditions on submitters --- src/sync/__tests__/syncTask.spec.ts | 112 +++++++++++------- .../__tests__/eventsSubmitter.spec.ts | 19 +-- .../__tests__/impressionsSubmitter.spec.ts | 22 ++++ src/sync/syncTask.ts | 30 +++-- 4 files changed, 123 insertions(+), 60 deletions(-) diff --git a/src/sync/__tests__/syncTask.spec.ts b/src/sync/__tests__/syncTask.spec.ts index 49ecf67f..fa3145cb 100644 --- a/src/sync/__tests__/syncTask.spec.ts +++ b/src/sync/__tests__/syncTask.spec.ts @@ -5,7 +5,7 @@ const period = 30; const taskResult = 'taskResult'; const asyncTask = jest.fn(() => Promise.resolve(taskResult)); -test('syncTaskFactory', (done) => { +test('syncTaskFactory / periodic execution', async () => { const syncTask = syncTaskFactory(loggerMock, asyncTask, period); @@ -40,45 +40,75 @@ test('syncTaskFactory', (done) => { expect(syncTask.isExecuting()).toBe(true); // Executing - setTimeout(() => { - expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments - expect(asyncTask).toBeCalledTimes(4); // The task was executed 4 times: twice due to periodic execution and twice due to execute call - - setTimeout(() => { - expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments - expect(asyncTask).toBeCalledTimes(5); // The task was executed 5 times: 3 due to periodic execution and twice due to execute call - - // Calling `stop` stops the periodic execution of the given task - expect(syncTask.isRunning()).toBe(true); // Running periodically - syncTask.stop(); - expect(syncTask.isRunning()).toBe(false); // Stop running periodically - - setTimeout(() => { - expect(asyncTask).toBeCalledTimes(5); // Stopped task should not be called again - - // Stopping and starting - syncTask.stop(); - syncTask.start(); // Inmediatelly call task - syncTask.stop(); - syncTask.start(); // Inmediatelly call task - syncTask.stop(); - expect(asyncTask).toBeCalledTimes(7); - - // Resume periodic execution - syncTask.start(); // Inmediatelly call task - syncTask.start(); // No effect - expect(asyncTask).toBeCalledTimes(8); - - setTimeout(() => { - expect(asyncTask).toBeCalledTimes(9); // Stopped task should not be called again - expect(syncTask.isRunning()).toBe(true); // Running periodically - syncTask.stop(); // Finally stop to finish the test - expect(syncTask.isRunning()).toBe(false); // Stop running periodically - - done(); - }, period + 10); - }, period + 10); - }, period + 10); - }, period + 10); + await new Promise(res => setTimeout(res, period + 10)); + expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments + expect(asyncTask).toBeCalledTimes(4); // The task was executed 4 times: twice due to periodic execution and twice due to execute call + + await new Promise(res => setTimeout(res, period + 10)); + expect(asyncTask).toHaveBeenLastCalledWith(...startArgs); // Periodic call should be done with the initial `start` arguments + expect(asyncTask).toBeCalledTimes(5); // The task was executed 5 times: 3 due to periodic execution and twice due to execute call + + // Calling `stop` stops the periodic execution of the given task + expect(syncTask.isRunning()).toBe(true); // Running periodically + syncTask.stop(); + expect(syncTask.isRunning()).toBe(false); // Stop running periodically + + await new Promise(res => setTimeout(res, period + 10)); + expect(asyncTask).toBeCalledTimes(5); // Stopped task should not be called again + + // Stopping and starting + syncTask.stop(); + syncTask.start(); // Inmediatelly call task + syncTask.stop(); + syncTask.start(); // Doesn't call task since previous one has not been resolved + syncTask.stop(); + expect(asyncTask).toBeCalledTimes(6); + + // Resume periodic execution + syncTask.start(); // Doesn't call task since previous one has not been resolved + syncTask.start(); // No effect + expect(asyncTask).toBeCalledTimes(6); + + await new Promise(res => setTimeout(res, period + 10)); + expect(asyncTask).toBeCalledTimes(9); // Stopped task should not be called again + expect(syncTask.isRunning()).toBe(true); // Running periodically + syncTask.stop(); // Finally stop to finish the test + expect(syncTask.isRunning()).toBe(false); // Stop running periodically + +}); + +test('syncTaskFactory / chaining executions', (done) => { + let executeCount = 0; + let resolveOrder = 0; + const asyncTask = jest.fn((toReturn) => { + executeCount++; + return new Promise(res => setTimeout(() => res(toReturn))); + }); + + const syncTask = syncTaskFactory(loggerMock, asyncTask, period); + + syncTask.execute(1).then(result=> { + // console.log('1'); + resolveOrder++; + expect(resolveOrder).toBe(1); // @TODO should be 1 ? + expect(executeCount).toBe(1); + expect(result).toBe(1); // @TODO should be 1 + }); + syncTask.execute(2).then(result=> { + // console.log('2'); + resolveOrder++; + expect(resolveOrder).toBe(2); + expect(executeCount).toBe(3); // @TODO borrar + expect(result).toBe(2); // @TODO should be 2 + }); + syncTask.execute(3).then(result=> { + // console.log('3'); + resolveOrder++; + expect(resolveOrder).toBe(3); // @TODO should be 3 ? + expect(executeCount).toBe(3); + expect(result).toBe(3); + + done(); + }); }); diff --git a/src/sync/submitters/__tests__/eventsSubmitter.spec.ts b/src/sync/submitters/__tests__/eventsSubmitter.spec.ts index 60de0379..d5f95ece 100644 --- a/src/sync/submitters/__tests__/eventsSubmitter.spec.ts +++ b/src/sync/submitters/__tests__/eventsSubmitter.spec.ts @@ -48,7 +48,7 @@ describe('Events submitter', () => { expect(eventsSubmitter.isRunning()).toEqual(false); }); - test('without eventsFirstPushWindow', async () => { + test('without eventsFirstPushWindow', (done) => { const eventsFirstPushWindow = 0; params.settings.startup.eventsFirstPushWindow = eventsFirstPushWindow; // @ts-ignore const eventsSubmitter = eventsSubmitterFactory(params); @@ -58,14 +58,19 @@ describe('Events submitter', () => { expect(eventsSubmitter.isExecuting()).toEqual(true); // and executes immediatelly if there isn't a push window expect(eventsCacheMock.isEmpty).toBeCalledTimes(1); - // If queue is full, submitter should be executed + // If queue is full, submitter is executed again after current execution is resolved __onFullQueueCb(); - expect(eventsSubmitter.isExecuting()).toEqual(true); - expect(eventsCacheMock.isEmpty).toBeCalledTimes(2); + expect(eventsCacheMock.isEmpty).toBeCalledTimes(1); - expect(eventsSubmitter.isRunning()).toEqual(true); - eventsSubmitter.stop(); - expect(eventsSubmitter.isRunning()).toEqual(false); + setTimeout(()=> { + expect(eventsSubmitter.isExecuting()).toEqual(false); + expect(eventsCacheMock.isEmpty).toBeCalledTimes(2); // 2 executions: 1st due to start and 2nd due to full queue + + expect(eventsSubmitter.isRunning()).toEqual(true); + eventsSubmitter.stop(); + expect(eventsSubmitter.isRunning()).toEqual(false); + done(); + }); }); test('doesn\'t drop items from cache when POST is resolved', (done) => { diff --git a/src/sync/submitters/__tests__/impressionsSubmitter.spec.ts b/src/sync/submitters/__tests__/impressionsSubmitter.spec.ts index d5c74144..0cf95cd4 100644 --- a/src/sync/submitters/__tests__/impressionsSubmitter.spec.ts +++ b/src/sync/submitters/__tests__/impressionsSubmitter.spec.ts @@ -72,4 +72,26 @@ describe('Impressions submitter', () => { }, params.settings.scheduler.impressionsPushRate + 10); }); + test('if it is executed while POST is pending, execution is queued until POST is resolved and not same items are submitted', (done) => { + // Make the POST request fail + params.splitApi.postTestImpressionsBulk.mockImplementation(() => Promise.resolve()); + + impressionsCacheInMemory.track([imp1]); + impressionsSubmitter.start(); + + // Tracking impression and executing submitter while POST is pending + impressionsCacheInMemory.track([{ ...imp1, keyName: 'k2' }]); + impressionsSubmitter.execute().then(() => { + expect(params.splitApi.postTestImpressionsBulk.mock.calls).toEqual([ + // impression for k1 + ['[{"f":"someFeature","i":[{"k":"k1","t":"someTreatment","m":0,"c":123}]}]'], + // impression for k2 + ['[{"f":"someFeature","i":[{"k":"k2","t":"someTreatment","m":0,"c":123}]}]']]); + impressionsSubmitter.stop(); + + done(); + }); + + }); + }); diff --git a/src/sync/syncTask.ts b/src/sync/syncTask.ts index eb3a2c30..9d3a5a4e 100644 --- a/src/sync/syncTask.ts +++ b/src/sync/syncTask.ts @@ -3,9 +3,9 @@ import { ILogger } from '../logger/types'; import { ISyncTask } from './types'; /** - * Creates a syncTask that handles the periodic execution of a given task ("start" and "stop" methods). - * The task can be executed once calling the "execute" method. - * NOTE: Multiple calls to "execute" are not queued. Use "isExecuting" method to handle synchronization. + * Creates a syncTask that handles the periodic execution of a given task (`start` and `stop` methods). + * The task can be also executed calling the `execute` method. Multiple calls to `execute` are chained, so the task runs secuentially to avoid race conditions. + * For example, submitters executed due to SDK destroy or full queues, while periodic execution is pending. * * @param log Logger instance. * @param task Task to execute that returns a promise that NEVER REJECTS. Otherwise, periodic execution can result in Unhandled Promise Rejections. @@ -15,8 +15,8 @@ import { ISyncTask } from './types'; */ export function syncTaskFactory(log: ILogger, task: (...args: Input) => Promise, period: number, taskName = 'task'): ISyncTask { - // Flag that indicates if the task is being executed - let executing = false; + // Task promise while it is pending. Undefined once the promise is resolved + let pendingTask: Promise | undefined; // flag that indicates if the task periodic execution has been started/stopped. let running = false; // Auxiliar counter used to avoid race condition when calling `start` & `stop` intermittently @@ -26,14 +26,21 @@ export function syncTaskFactory(log: ILogger, // Id of the periodic call timeout let timeoutID: any; - function execute(...args: Input) { - executing = true; + function execute(...args: Input): Promise { + // If task is executing, chain the new execution + if (pendingTask) { + return pendingTask.then(() => { + return execute(...args); + }); + } + + // Execute task log.debug(SYNC_TASK_EXECUTE, [taskName]); - return task(...args).then(result => { - executing = false; + pendingTask = task(...args).then(result => { + pendingTask = undefined; return result; }); - // No need to handle promise rejection because it is a pre-condition that provided task never rejects. + return pendingTask; } function periodicExecute(currentRunningId: number) { @@ -46,11 +53,10 @@ export function syncTaskFactory(log: ILogger, } return { - // @TODO check if we need to queued `execute` calls, to avoid possible race conditions on submitters and updaters with streaming. execute, isExecuting() { - return executing; + return pendingTask !== undefined; }, start(...args: Input) { From 25d369af8c7e507c925625126701fdcdcfe86907 Mon Sep 17 00:00:00 2001 From: Emiliano Sanchez Date: Wed, 22 Jun 2022 11:08:11 -0300 Subject: [PATCH 2/2] update changelog --- CHANGES.txt | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGES.txt b/CHANGES.txt index 3d681b60..75c777bf 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,3 +1,7 @@ +1.4.2 (June XX, 2022) + - Updated telemetry to submit data even if user consent is not granted. + - Updated submitters logic, to avoid duplicating the post of impressions to Split cloud when the SDK is destroyed while performing its periodic post of impressions. + 1.4.1 (June 13, 2022) - Bugfixing - Updated submitters logic, to avoid dropping impressions and events that are being tracked while POST request is pending.