Polishing

EmilianoSanchez · EmilianoSanchez · commit a5e0d3b72808 · 2025-03-18T16:49:11.000-03:00
diff --git a/src/sync/polling/updaters/__tests__/splitChangesUpdater.spec.ts b/src/sync/polling/updaters/__tests__/splitChangesUpdater.spec.ts
@@ -13,6 +13,7 @@ import { loggerMock } from '../../../../logger/__tests__/sdkLogger.mock';
 import { telemetryTrackerFactory } from '../../../../trackers/telemetryTracker';
 import { splitNotifications } from '../../../streaming/__tests__/dataMocks';
 import { RBSegmentsCacheInMemory } from '../../../../storages/inMemory/RBSegmentsCacheInMemory';
+import { RB_SEGMENT_UPDATE, SPLIT_UPDATE } from '../../../streaming/constants';
 
 const ARCHIVED_FF = 'ARCHIVED';
 
@@ -202,7 +203,7 @@ describe('splitChangesUpdater', () => {
       const payload = notification.decoded as Pick<ISplit, 'name' | 'changeNumber' | 'killed' | 'defaultTreatment' | 'trafficTypeName' | 'conditions' | 'status' | 'seed' | 'trafficAllocation' | 'trafficAllocationSeed' | 'configurations'>;
       const changeNumber = payload.changeNumber;
 
-      await expect(splitChangesUpdater(undefined, undefined, { payload, changeNumber: changeNumber })).resolves.toBe(true);
+      await expect(splitChangesUpdater(undefined, undefined, { payload, changeNumber: changeNumber, type: SPLIT_UPDATE })).resolves.toBe(true);
 
       // fetch and RBSegments.update not being called
       expect(fetchSplitChanges).toBeCalledTimes(0);
@@ -226,7 +227,7 @@ describe('splitChangesUpdater', () => {
     const payload = { name: 'rbsegment', status: 'ACTIVE', changeNumber: 1684329854385, conditions: [] } as unknown as IRBSegment;
     const changeNumber = payload.changeNumber;
 
-    await expect(splitChangesUpdater(undefined, undefined, { payload, changeNumber: changeNumber })).resolves.toBe(true);
+    await expect(splitChangesUpdater(undefined, undefined, { payload, changeNumber: changeNumber, type: RB_SEGMENT_UPDATE })).resolves.toBe(true);
 
     // fetch and Splits.update not being called
     expect(fetchSplitChanges).toBeCalledTimes(0);
@@ -256,7 +257,7 @@ describe('splitChangesUpdater', () => {
     let calls = 0;
     // emit always if not configured sets
     for (const setMock of setMocks) {
-      await expect(splitChangesUpdater(undefined, undefined, { payload: { ...payload, sets: setMock.sets, status: 'ACTIVE' }, changeNumber: index })).resolves.toBe(true);
+      await expect(splitChangesUpdater(undefined, undefined, { payload: { ...payload, sets: setMock.sets, status: 'ACTIVE' }, changeNumber: index, type: SPLIT_UPDATE })).resolves.toBe(true);
       expect(splitsEmitSpy.mock.calls[index][0]).toBe('state::splits-arrived');
       index++;
     }
@@ -268,7 +269,7 @@ describe('splitChangesUpdater', () => {
     splitsEmitSpy.mockReset();
     index = 0;
     for (const setMock of setMocks) {
-      await expect(splitChangesUpdater(undefined, undefined, { payload: { ...payload, sets: setMock.sets, status: 'ACTIVE' }, changeNumber: index })).resolves.toBe(true);
+      await expect(splitChangesUpdater(undefined, undefined, { payload: { ...payload, sets: setMock.sets, status: 'ACTIVE' }, changeNumber: index, type: SPLIT_UPDATE })).resolves.toBe(true);
       if (setMock.shouldEmit) calls++;
       expect(splitsEmitSpy.mock.calls.length).toBe(calls);
       index++;
diff --git a/src/sync/polling/updaters/splitChangesUpdater.ts b/src/sync/polling/updaters/splitChangesUpdater.ts
@@ -9,8 +9,10 @@ import { SYNC_SPLITS_FETCH, SYNC_SPLITS_UPDATE, SYNC_RBS_UPDATE, SYNC_SPLITS_FET
 import { startsWith } from '../../../utils/lang';
 import { IN_RULE_BASED_SEGMENT, IN_SEGMENT } from '../../../utils/constants';
 import { setToArray } from '../../../utils/lang/sets';
+import { SPLIT_UPDATE } from '../../streaming/constants';
 
-type ISplitChangesUpdater = (noCache?: boolean, till?: number, splitUpdateNotification?: { payload: ISplit | IRBSegment, changeNumber: number }) => Promise<boolean>
+export type InstantUpdate = { payload: ISplit | IRBSegment, changeNumber: number, type: string };
+type SplitChangesUpdater = (noCache?: boolean, till?: number, instantUpdate?: InstantUpdate) => Promise<boolean>
 
 // Checks that all registered segments have been fetched (changeNumber !== -1 for every segment).
 // Returns a promise that could be rejected.
@@ -27,8 +29,9 @@ function checkAllSegmentsExist(segments: ISegmentsCacheBase): Promise<boolean> {
  * Collect segments from a raw split definition.
  * Exported for testing purposes.
  */
-export function parseSegments({ conditions }: ISplit | IRBSegment, matcherType: typeof IN_SEGMENT | typeof IN_RULE_BASED_SEGMENT = IN_SEGMENT): Set<string> {
-  let segments = new Set<string>();
+export function parseSegments(ruleEntity: ISplit | IRBSegment, matcherType: typeof IN_SEGMENT | typeof IN_RULE_BASED_SEGMENT = IN_SEGMENT): Set<string> {
+  const { conditions, excluded } = ruleEntity as IRBSegment;
+  const segments = new Set<string>(excluded && excluded.segments);
 
   for (let i = 0; i < conditions.length; i++) {
     const matchers = conditions[i].matcherGroup.matchers;
@@ -67,26 +70,22 @@ function matchFilters(featureFlag: ISplit, filters: ISplitFiltersValidation) {
   return matchNames || matchPrefix;
 }
 
-function isFF(ruleBasedEntity: IRBSegment | ISplit): ruleBasedEntity is ISplit {
-  return (ruleBasedEntity as ISplit).defaultTreatment !== undefined;
-}
-
 /**
  * Given the list of splits from /splitChanges endpoint, it returns the mutations,
  * i.e., an object with added splits, removed splits and used segments.
  * Exported for testing purposes.
  */
 export function computeMutation<T extends ISplit | IRBSegment>(rules: Array<T>, segments: Set<string>, filters?: ISplitFiltersValidation): ISplitMutations<T> {
 
-  return rules.reduce((accum, ruleBasedEntity) => {
-    if (ruleBasedEntity.status === 'ACTIVE' && (!filters || matchFilters(ruleBasedEntity as ISplit, filters))) {
-      accum.added.push(ruleBasedEntity);
+  return rules.reduce((accum, ruleEntity) => {
+    if (ruleEntity.status === 'ACTIVE' && (!filters || matchFilters(ruleEntity as ISplit, filters))) {
+      accum.added.push(ruleEntity);
 
-      parseSegments(ruleBasedEntity).forEach((segmentName: string) => {
+      parseSegments(ruleEntity).forEach((segmentName: string) => {
         segments.add(segmentName);
       });
     } else {
-      accum.removed.push(ruleBasedEntity);
+      accum.removed.push(ruleEntity);
     }
 
     return accum;
@@ -116,7 +115,7 @@ export function splitChangesUpdaterFactory(
   requestTimeoutBeforeReady: number = 0,
   retriesOnFailureBeforeReady: number = 0,
   isClientSide?: boolean
-): ISplitChangesUpdater {
+): SplitChangesUpdater {
   const { splits, rbSegments, segments } = storage;
 
   let startingUp = true;
@@ -134,7 +133,7 @@ export function splitChangesUpdaterFactory(
    * @param noCache - true to revalidate data to fetch
    * @param till - query param to bypass CDN requests
    */
-  return function splitChangesUpdater(noCache?: boolean, till?: number, updateNotification?: { payload: ISplit | IRBSegment, changeNumber: number }) {
+  return function splitChangesUpdater(noCache?: boolean, till?: number, instantUpdate?: InstantUpdate) {
 
     /**
      * @param since - current changeNumber at splitsCache
@@ -144,15 +143,15 @@ export function splitChangesUpdaterFactory(
       const [since, rbSince] = sinces;
       log.debug(SYNC_SPLITS_FETCH, sinces);
       const fetcherPromise = Promise.resolve(
-        updateNotification ?
-          isFF(updateNotification.payload) ?
+        instantUpdate ?
+          instantUpdate.type === SPLIT_UPDATE ?
             // IFFU edge case: a change to a flag that adds an IN_RULE_BASED_SEGMENT matcher that is not present yet
-            Promise.resolve(rbSegments.contains(parseSegments(updateNotification.payload, IN_RULE_BASED_SEGMENT))).then((contains) => {
+            Promise.resolve(rbSegments.contains(parseSegments(instantUpdate.payload, IN_RULE_BASED_SEGMENT))).then((contains) => {
               return contains ?
-                { ff: { d: [updateNotification.payload as ISplit], t: updateNotification.changeNumber } } :
+                { ff: { d: [instantUpdate.payload as ISplit], t: instantUpdate.changeNumber } } :
                 splitChangesFetcher(since, noCache, till, rbSince, _promiseDecorator);
             }) :
-            { rbs: { d: [updateNotification.payload as IRBSegment], t: updateNotification.changeNumber } } :
+            { rbs: { d: [instantUpdate.payload as IRBSegment], t: instantUpdate.changeNumber } } :
           splitChangesFetcher(since, noCache, till, rbSince, _promiseDecorator)
       )
         .then((splitChanges: ISplitChangesResponse) => {
@@ -180,14 +179,8 @@ export function splitChangesUpdaterFactory(
           ]).then(([ffChanged, rbsChanged]) => {
             if (splitsEventEmitter) {
               // To emit SDK_SPLITS_ARRIVED for server-side SDK, we must check that all registered segments have been fetched
-              return Promise.resolve(!splitsEventEmitter.splitsArrived ||
-                (
-                  (!splitChanges.ff || since !== splitChanges.ff.t) &&
-                  (!splitChanges.rbs || rbSince !== splitChanges.rbs.t) &&
-                  (ffChanged || rbsChanged) &&
-                  (isClientSide || checkAllSegmentsExist(segments))
-                )
-              )
+              return Promise.resolve(!splitsEventEmitter.splitsArrived || ((ffChanged || rbsChanged) && (isClientSide || checkAllSegmentsExist(segments))))
+                .catch(() => false /** noop. just to handle a possible `checkAllSegmentsExist` rejection, before emitting SDK event */)
                 .then(emitSplitsArrivedEvent => {
                   // emit SDK events
                   if (emitSplitsArrivedEvent) splitsEventEmitter.emit(SDK_SPLITS_ARRIVED);
diff --git a/src/sync/streaming/UpdateWorkers/SplitsUpdateWorker.ts b/src/sync/streaming/UpdateWorkers/SplitsUpdateWorker.ts
@@ -8,6 +8,7 @@ import { ITelemetryTracker } from '../../../trackers/types';
 import { Backoff } from '../../../utils/Backoff';
 import { SPLITS } from '../../../utils/constants';
 import { ISegmentsSyncTask, ISplitsSyncTask } from '../../polling/types';
+import { InstantUpdate } from '../../polling/updaters/splitChangesUpdater';
 import { RB_SEGMENT_UPDATE } from '../constants';
 import { parseFFUpdatePayload } from '../parseUtils';
 import { ISplitKillData, ISplitUpdateData } from '../SSEHandler/types';
@@ -24,26 +25,26 @@ export function SplitsUpdateWorker(log: ILogger, storage: IStorageSync, splitsSy
     let handleNewEvent = false;
     let isHandlingEvent: boolean;
     let cdnBypass: boolean;
-    let payload: ISplit | IRBSegment | undefined;
+    let instantUpdate: InstantUpdate | undefined;
     const backoff = new Backoff(__handleSplitUpdateCall, FETCH_BACKOFF_BASE, FETCH_BACKOFF_MAX_WAIT);
 
     function __handleSplitUpdateCall() {
       isHandlingEvent = true;
       if (maxChangeNumber > cache.getChangeNumber()) {
         handleNewEvent = false;
-        const splitUpdateNotification = payload ? { payload, changeNumber: maxChangeNumber } : undefined;
         // fetch splits revalidating data if cached
-        splitsSyncTask.execute(true, cdnBypass ? maxChangeNumber : undefined, splitUpdateNotification).then(() => {
+        splitsSyncTask.execute(true, cdnBypass ? maxChangeNumber : undefined, instantUpdate).then(() => {
           if (!isHandlingEvent) return; // halt if `stop` has been called
           if (handleNewEvent) {
             __handleSplitUpdateCall();
           } else {
-            if (splitUpdateNotification) telemetryTracker.trackUpdatesFromSSE(SPLITS);
+            if (instantUpdate) telemetryTracker.trackUpdatesFromSSE(SPLITS);
             // fetch new registered segments for server-side API. Not retrying on error
             if (segmentsSyncTask) segmentsSyncTask.execute(true);
 
             const attempts = backoff.attempts + 1;
 
+            // @TODO and with RBS and FF
             if (maxChangeNumber <= cache.getChangeNumber()) {
               log.debug(`Refresh completed${cdnBypass ? ' bypassing the CDN' : ''} in ${attempts} attempts.`);
               isHandlingEvent = false;
@@ -76,18 +77,18 @@ export function SplitsUpdateWorker(log: ILogger, storage: IStorageSync, splitsSy
        *
        * @param changeNumber - change number of the notification
        */
-      put({ changeNumber, pcn }: ISplitUpdateData, _payload?: ISplit | IRBSegment) {
+      put({ changeNumber, pcn, type }: ISplitUpdateData, payload?: ISplit | IRBSegment) {
         const currentChangeNumber = cache.getChangeNumber();
 
         if (changeNumber <= currentChangeNumber || changeNumber <= maxChangeNumber) return;
 
         maxChangeNumber = changeNumber;
         handleNewEvent = true;
         cdnBypass = false;
-        payload = undefined;
+        instantUpdate = undefined;
 
-        if (_payload && currentChangeNumber === pcn) {
-          payload = _payload;
+        if (payload && currentChangeNumber === pcn) {
+          instantUpdate = { payload, changeNumber, type };
         }
 
         if (backoff.timeoutID || !isHandlingEvent) __handleSplitUpdateCall();
