Merge pull request #341 from splitio/SDKS-8407_memberships_cdn_bypass

EmilianoSanchez · web-flow · commit f2ac28e42843 · 2024-09-09T16:54:00.000-03:00
Large Segments: CDN bypass for `/memberships` requests
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@splitsoftware/splitio-commons",
-  "version": "1.16.1-rc.11",
+  "version": "1.16.1-rc.12",
   "description": "Split JavaScript SDK common components",
   "main": "cjs/index.js",
   "module": "esm/index.js",
diff --git a/src/services/__tests__/splitApi.spec.ts b/src/services/__tests__/splitApi.spec.ts
@@ -30,10 +30,10 @@ describe('splitApi', () => {
     assertHeaders(settings, headers);
     expect(url).toBe('auth/v2/auth?s=1.1&users=key1&users=key2');
 
-    splitApi.fetchMemberships('userKey');
+    splitApi.fetchMemberships('userKey', false, 80);
     [url, { headers }] = fetchMock.mock.calls[1];
     assertHeaders(settings, headers);
-    expect(url).toBe('sdk/memberships/userKey');
+    expect(url).toBe('sdk/memberships/userKey?till=80');
 
     splitApi.fetchSegmentChanges(-1, 'segmentName', false, 90);
     [url, { headers }] = fetchMock.mock.calls[2];
diff --git a/src/services/splitApi.ts b/src/services/splitApi.ts
@@ -67,14 +67,14 @@ export function splitApiFactory(
       return splitHttpClient(url, noCache ? noCacheHeaderOptions : undefined, telemetryTracker.trackHttp(SEGMENT));
     },
 
-    fetchMemberships(userMatchingKey: string, noCache?: boolean) {
+    fetchMemberships(userMatchingKey: string, noCache?: boolean, till?: number) {
       /**
        * URI encoding of user keys in order to:
        *  - avoid 400 responses (due to URI malformed). E.g.: '/api/memberships/%'
        *  - avoid 404 responses. E.g.: '/api/memberships/foo/bar'
        *  - match user keys with special characters. E.g.: 'foo%bar', 'foo/bar'
        */
-      const url = `${urls.sdk}/memberships/${encodeURIComponent(userMatchingKey)}`;
+      const url = `${urls.sdk}/memberships/${encodeURIComponent(userMatchingKey)}${till ? '?till=' + till : ''}`;
       return splitHttpClient(url, noCache ? noCacheHeaderOptions : undefined, telemetryTracker.trackHttp(MEMBERSHIPS));
     },
 
diff --git a/src/services/types.ts b/src/services/types.ts
@@ -39,7 +39,7 @@ export type IFetchSplitChanges = (since: number, noCache?: boolean, till?: numbe
 
 export type IFetchSegmentChanges = (since: number, segmentName: string, noCache?: boolean, till?: number) => Promise<IResponse>
 
-export type IFetchMemberships = (userMatchingKey: string, noCache?: boolean) => Promise<IResponse>
+export type IFetchMemberships = (userMatchingKey: string, noCache?: boolean, till?: number) => Promise<IResponse>
 
 export type IPostEventsBulk = (body: string, headers?: Record<string, string>) => Promise<IResponse>
 
diff --git a/src/storages/inMemory/TelemetryCacheInMemory.ts b/src/storages/inMemory/TelemetryCacheInMemory.ts
@@ -51,7 +51,7 @@ export class TelemetryCacheInMemory implements ITelemetryCacheSync {
       spC: this.splits && this.splits.getSplitNames().length,
       seC: this.segments && this.segments.getRegisteredSegments().length,
       skC: this.segments && this.segments.getKeysCount(),
-      lseC: this.largeSegments && this.largeSegments.getRegisteredSegments().length,
+      lsC: this.largeSegments && this.largeSegments.getRegisteredSegments().length,
       lskC: this.largeSegments && this.largeSegments.getKeysCount(),
       sL: this.getSessionLength(),
       eQ: this.getEventStats(QUEUED),
diff --git a/src/storages/types.ts b/src/storages/types.ts
@@ -270,7 +270,7 @@ export interface ISegmentsCacheSync extends ISegmentsCacheBase {
   getRegisteredSegments(): string[]
   getKeysCount(): number // only used for telemetry
   setChangeNumber(name: string, changeNumber: number): boolean | void
-  getChangeNumber(name: string): number
+  getChangeNumber(name?: string): number
   resetSegments(segmentsData: MySegmentsData | IMySegmentsResponse): boolean // only for Sync Client-Side
   clear(): void
 }
diff --git a/src/sync/polling/fetchers/mySegmentsFetcher.ts b/src/sync/polling/fetchers/mySegmentsFetcher.ts
@@ -11,11 +11,12 @@ export function mySegmentsFetcherFactory(fetchMemberships: IFetchMemberships): I
   return function mySegmentsFetcher(
     userMatchingKey: string,
     noCache?: boolean,
+    till?: number,
     // Optional decorator for `fetchMemberships` promise, such as timeout or time tracker
     decorator?: (promise: Promise<IResponse>) => Promise<IResponse>
   ): Promise<IMembershipsResponse> {
 
-    let mySegmentsPromise = fetchMemberships(userMatchingKey, noCache);
+    let mySegmentsPromise = fetchMemberships(userMatchingKey, noCache, till);
     if (decorator) mySegmentsPromise = decorator(mySegmentsPromise);
 
     return mySegmentsPromise.then(resp => resp.json());
diff --git a/src/sync/polling/fetchers/types.ts b/src/sync/polling/fetchers/types.ts
@@ -19,5 +19,6 @@ export type ISegmentChangesFetcher = (
 export type IMySegmentsFetcher = (
   userMatchingKey: string,
   noCache?: boolean,
+  till?: number,
   decorator?: (promise: Promise<IResponse>) => Promise<IResponse>
 ) => Promise<IMembershipsResponse>
diff --git a/src/sync/polling/types.ts b/src/sync/polling/types.ts
@@ -15,7 +15,7 @@ export type MySegmentsData = {
   removed: string[]
 }
 
-export interface IMySegmentsSyncTask extends ISyncTask<[segmentsData?: MySegmentsData, noCache?: boolean], boolean> { }
+export interface IMySegmentsSyncTask extends ISyncTask<[segmentsData?: MySegmentsData, noCache?: boolean, till?: number], boolean> { }
 
 export interface IPollingManager extends ITask {
   syncAll(): Promise<any>
diff --git a/src/sync/polling/updaters/mySegmentsUpdater.ts b/src/sync/polling/updaters/mySegmentsUpdater.ts
@@ -9,7 +9,7 @@ import { MySegmentsData } from '../types';
 import { IMembershipsResponse } from '../../../dtos/types';
 import { MEMBERSHIPS_LS_UPDATE } from '../../streaming/constants';
 
-type IMySegmentsUpdater = (segmentList?: MySegmentsData, noCache?: boolean) => Promise<boolean>
+type IMySegmentsUpdater = (segmentsData?: MySegmentsData, noCache?: boolean, till?: number) => Promise<boolean>
 
 /**
  * factory of MySegments updater, a task that:
@@ -57,12 +57,12 @@ export function mySegmentsUpdaterFactory(
     }
   }
 
-  function _mySegmentsUpdater(retry: number, segmentsData?: MySegmentsData, noCache?: boolean): Promise<boolean> {
+  function _mySegmentsUpdater(retry: number, segmentsData?: MySegmentsData, noCache?: boolean, till?: number): Promise<boolean> {
     const updaterPromise: Promise<boolean> = segmentsData ?
       // If segmentsData is provided, there is no need to fetch mySegments
       new Promise((res) => { updateSegments(segmentsData); res(true); }) :
       // If not provided, fetch mySegments
-      mySegmentsFetcher(matchingKey, noCache, _promiseDecorator).then(segments => {
+      mySegmentsFetcher(matchingKey, noCache, till, _promiseDecorator).then(segments => {
         // Only when we have downloaded segments completely, we should not keep retrying anymore
         startingUp = false;
 
@@ -92,9 +92,10 @@ export function mySegmentsUpdaterFactory(
    *  (2) an object with a segment name and action (true: add, or false: delete) to update the storage,
    *  (3) or `undefined`, for which the updater will fetch mySegments in order to sync the storage.
    * @param {boolean | undefined} noCache true to revalidate data to fetch
+   * @param {boolean | undefined} till query param to bypass CDN requests
    */
-  return function mySegmentsUpdater(segmentsData?: MySegmentsData, noCache?: boolean) {
-    return _mySegmentsUpdater(0, segmentsData, noCache);
+  return function mySegmentsUpdater(segmentsData?: MySegmentsData, noCache?: boolean, till?: number) {
+    return _mySegmentsUpdater(0, segmentsData, noCache, till);
   };
 
 }
diff --git a/src/sync/streaming/UpdateWorkers/MySegmentsUpdateWorker.ts b/src/sync/streaming/UpdateWorkers/MySegmentsUpdateWorker.ts
@@ -3,81 +3,131 @@ import { Backoff } from '../../../utils/Backoff';
 import { IUpdateWorker } from './types';
 import { ITelemetryTracker } from '../../../trackers/types';
 import { MEMBERSHIPS } from '../../../utils/constants';
+import { ISegmentsCacheSync, IStorageSync } from '../../../storages/types';
+import { ILogger } from '../../../logger/types';
+import { FETCH_BACKOFF_MAX_RETRIES } from './constants';
+import { MEMBERSHIPS_LS_UPDATE, MEMBERSHIPS_MS_UPDATE } from '../constants';
 
 /**
  * MySegmentsUpdateWorker factory
  */
-export function MySegmentsUpdateWorker(mySegmentsSyncTask: IMySegmentsSyncTask, telemetryTracker: ITelemetryTracker): IUpdateWorker<[mySegmentsData?: Pick<MySegmentsData, 'type' | 'cn'>, payload?: Pick<MySegmentsData, 'added' | 'removed'>, delay?: number]> {
+export function MySegmentsUpdateWorker(log: ILogger, storage: Pick<IStorageSync, 'segments' | 'largeSegments'>, mySegmentsSyncTask: IMySegmentsSyncTask, telemetryTracker: ITelemetryTracker): IUpdateWorker<[mySegmentsData?: Pick<MySegmentsData, 'type' | 'cn'>, payload?: Pick<MySegmentsData, 'added' | 'removed'>, delay?: number]> {
 
-  let maxChangeNumber = 0; // keeps the maximum changeNumber among queued events
-  let currentChangeNumber = -1;
-  let handleNewEvent = false;
-  let isHandlingEvent: boolean;
-  let _segmentsData: MySegmentsData | undefined; // keeps the segmentsData (if included in notification payload) from the queued event with maximum changeNumber
   let _delay: undefined | number;
   let _delayTimeoutID: any;
-  const backoff = new Backoff(__handleMySegmentsUpdateCall);
-
-  function __handleMySegmentsUpdateCall() {
-    isHandlingEvent = true;
-    if (maxChangeNumber > currentChangeNumber) {
-      handleNewEvent = false;
-      const currentMaxChangeNumber = maxChangeNumber;
-
-      // fetch mySegments revalidating data if cached
-      const syncTask = _delay ?
-        new Promise(res => {
-          _delayTimeoutID = setTimeout(() => {
-            _delay = undefined;
-            mySegmentsSyncTask.execute(_segmentsData, true).then(res);
-          }, _delay);
-        }) :
-        mySegmentsSyncTask.execute(_segmentsData, true);
-
-      syncTask.then((result) => {
-        if (!isHandlingEvent) return; // halt if `stop` has been called
-        if (result !== false) { // Unlike `Splits|SegmentsUpdateWorker`, we cannot use `mySegmentsCache.getChangeNumber` since `/mySegments` endpoint doesn't provide this value.
-          if (_segmentsData) telemetryTracker.trackUpdatesFromSSE(MEMBERSHIPS);
-          currentChangeNumber = Math.max(currentChangeNumber, currentMaxChangeNumber); // use `currentMaxChangeNumber`, in case that `maxChangeNumber` was updated during fetch.
-        }
-        if (handleNewEvent) {
-          __handleMySegmentsUpdateCall();
-        } else {
-          backoff.scheduleCall();
-        }
-      });
-    } else {
-      isHandlingEvent = false;
+
+  function createUpdateWorker(mySegmentsCache: ISegmentsCacheSync) {
+
+    let maxChangeNumber = 0; // keeps the maximum changeNumber among queued events
+    let currentChangeNumber = -1;
+    let handleNewEvent = false;
+    let isHandlingEvent: boolean;
+    let cdnBypass: boolean;
+    let _segmentsData: MySegmentsData | undefined; // keeps the segmentsData (if included in notification payload) from the queued event with maximum changeNumber
+    const backoff = new Backoff(__handleMySegmentsUpdateCall);
+
+    function __handleMySegmentsUpdateCall() {
+      isHandlingEvent = true;
+      if (maxChangeNumber > Math.max(currentChangeNumber, mySegmentsCache.getChangeNumber())) {
+        handleNewEvent = false;
+        const currentMaxChangeNumber = maxChangeNumber;
+
+        // fetch mySegments revalidating data if cached
+        const syncTask = _delay ?
+          new Promise(res => {
+            _delayTimeoutID = setTimeout(() => {
+              _delay = undefined;
+              mySegmentsSyncTask.execute(_segmentsData, true, cdnBypass ? maxChangeNumber : undefined).then(res);
+            }, _delay);
+          }) :
+          mySegmentsSyncTask.execute(_segmentsData, true, cdnBypass ? maxChangeNumber : undefined);
+
+        syncTask.then((result) => {
+          if (!isHandlingEvent) return; // halt if `stop` has been called
+          if (result !== false) { // Unlike `Splits|SegmentsUpdateWorker`, `mySegmentsCache.getChangeNumber` can be -1, since `/memberships` change number is optional
+            const storageChangeNumber = mySegmentsCache.getChangeNumber();
+            currentChangeNumber = storageChangeNumber > -1 ?
+              storageChangeNumber :
+              Math.max(currentChangeNumber, currentMaxChangeNumber); // use `currentMaxChangeNumber`, in case that `maxChangeNumber` was updated during fetch.
+          }
+          if (handleNewEvent) {
+            __handleMySegmentsUpdateCall();
+          } else {
+            if (_segmentsData) telemetryTracker.trackUpdatesFromSSE(MEMBERSHIPS);
+
+            const attempts = backoff.attempts + 1;
+
+            if (maxChangeNumber <= currentChangeNumber) {
+              log.debug(`Refresh completed${cdnBypass ? ' bypassing the CDN' : ''} in ${attempts} attempts.`);
+              isHandlingEvent = false;
+              return;
+            }
+
+            if (attempts < FETCH_BACKOFF_MAX_RETRIES) {
+              backoff.scheduleCall();
+              return;
+            }
+
+            if (cdnBypass) {
+              log.debug(`No changes fetched after ${attempts} attempts with CDN bypassed.`);
+              isHandlingEvent = false;
+            } else {
+              backoff.reset();
+              cdnBypass = true;
+              __handleMySegmentsUpdateCall();
+            }
+          }
+        });
+      } else {
+        isHandlingEvent = false;
+      }
     }
+
+    return {
+      /**
+       * Invoked by NotificationProcessor on MY_(LARGE)_SEGMENTS_UPDATE notifications
+       *
+       * @param changeNumber change number of the notification
+       * @param segmentsData data for KeyList or SegmentRemoval instant updates
+       * @param delay optional time to wait for BoundedFetchRequest or BoundedFetchRequest updates
+       */
+      put(mySegmentsData: Pick<MySegmentsData, 'type' | 'cn'>, payload?: Pick<MySegmentsData, 'added' | 'removed'>, delay?: number) {
+        const { type, cn } = mySegmentsData;
+        // Discard event if it is outdated or there is a pending fetch request (_delay is set), but update target change number
+        if (cn <= Math.max(currentChangeNumber, mySegmentsCache.getChangeNumber()) || cn <= maxChangeNumber) return;
+        maxChangeNumber = cn;
+        if (_delay) return;
+
+        handleNewEvent = true;
+        cdnBypass = false;
+        _segmentsData = payload && { type, cn, added: payload.added, removed: payload.removed };
+        _delay = delay;
+
+        if (backoff.timeoutID || !isHandlingEvent) __handleMySegmentsUpdateCall();
+        backoff.reset();
+      },
+
+      stop() {
+        clearTimeout(_delayTimeoutID);
+        _delay = undefined;
+        isHandlingEvent = false;
+        backoff.reset();
+      }
+    };
   }
 
+  const updateWorkers = {
+    [MEMBERSHIPS_MS_UPDATE]: createUpdateWorker(storage.segments),
+    [MEMBERSHIPS_LS_UPDATE]: createUpdateWorker(storage.largeSegments!),
+  };
+
   return {
-    /**
-     * Invoked by NotificationProcessor on MY_(LARGE)_SEGMENTS_UPDATE notifications
-     *
-     * @param changeNumber change number of the notification
-     * @param segmentsData data for KeyList or SegmentRemoval instant updates
-     * @param delay optional time to wait for BoundedFetchRequest or BoundedFetchRequest updates
-     */
     put(mySegmentsData: Pick<MySegmentsData, 'type' | 'cn'>, payload?: Pick<MySegmentsData, 'added' | 'removed'>, delay?: number) {
-      const { type, cn } = mySegmentsData;
-      // Ignore event if it is outdated or if there is a pending fetch request (_delay is set)
-      if (cn <= currentChangeNumber || cn <= maxChangeNumber || _delay) return;
-
-      maxChangeNumber = cn;
-      handleNewEvent = true;
-      _segmentsData = payload && { type, cn, added: payload.added, removed: payload.removed };
-      _delay = delay;
-
-      if (backoff.timeoutID || !isHandlingEvent) __handleMySegmentsUpdateCall();
-      backoff.reset();
+      updateWorkers[mySegmentsData.type].put(mySegmentsData, payload, delay);
     },
-
     stop() {
-      clearTimeout(_delayTimeoutID);
-      _delay = undefined;
-      isHandlingEvent = false;
-      backoff.reset();
+      updateWorkers[MEMBERSHIPS_MS_UPDATE].stop();
+      updateWorkers[MEMBERSHIPS_LS_UPDATE].stop();
     }
   };
 }
diff --git a/src/sync/streaming/UpdateWorkers/__tests__/MySegmentsUpdateWorker.spec.ts b/src/sync/streaming/UpdateWorkers/__tests__/MySegmentsUpdateWorker.spec.ts
diff --git a/src/sync/streaming/__tests__/parseUtils.spec.ts b/src/sync/streaming/__tests__/parseUtils.spec.ts
diff --git a/src/sync/streaming/pushManager.ts b/src/sync/streaming/pushManager.ts
diff --git a/src/sync/submitters/types.ts b/src/sync/submitters/types.ts

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@splitsoftware/splitio-commons",`
`3`		`- "version": "1.16.1-rc.11",`
	`3`	`+ "version": "1.16.1-rc.12",`
`4`	`4`	`"description": "Split JavaScript SDK common components",`
`5`	`5`	`"main": "cjs/index.js",`
`6`	`6`	`"module": "esm/index.js",`
Original file line number	Diff line number	Diff line change
`@@ -270,7 +270,7 @@ export interface ISegmentsCacheSync extends ISegmentsCacheBase {`
`270`	`270`	`getRegisteredSegments(): string[]`
`271`	`271`	`getKeysCount(): number // only used for telemetry`
`272`	`272`	`setChangeNumber(name: string, changeNumber: number): boolean \| void`
`273`		`- getChangeNumber(name: string): number`
	`273`	`+ getChangeNumber(name?: string): number`
`274`	`274`	`resetSegments(segmentsData: MySegmentsData \| IMySegmentsResponse): boolean // only for Sync Client-Side`
`275`	`275`	`clear(): void`
`276`	`276`	`}`
Original file line number	Diff line number	Diff line change
`@@ -15,7 +15,7 @@ export type MySegmentsData = {`
`15`	`15`	`removed: string[]`
`16`	`16`	`}`
`17`	`17`
`18`		`-export interface IMySegmentsSyncTask extends ISyncTask<[segmentsData?: MySegmentsData, noCache?: boolean], boolean> { }`
	`18`	`+export interface IMySegmentsSyncTask extends ISyncTask<[segmentsData?: MySegmentsData, noCache?: boolean, till?: number], boolean> { }`
`19`	`19`
`20`	`20`	`export interface IPollingManager extends ITask {`
`21`	`21`	`syncAll(): Promise<any>`