Merge pull request #283 from splitio/async_storage_clear

EmilianoSanchez · web-flow · commit 64f017cc1c69 · 2023-12-11T16:35:43.000-03:00
Update PluggableStorage to clear storage if SDK key or filter criteria change
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -1,3 +1,7 @@
+1.12.1 (December 12, 2023)
+ - Updated PluggableStorage for producer mode, to clear the storage if it was previously synchronized with a different SDK key (i.e., a different environment) or different Split Filter criteria.
+ - Bugfixing - Fixed an issue when tracking telemetry latencies for the new `getTreatmentsByFlagSet(s)` methods in Redis and Pluggable storages, which was causing the SDK to not track those stats.
+
 1.12.0 (December 4, 2023)
  - Added support for Flag Sets in "consumer" and "partial consumer" modes for Pluggable and Redis storages.
  - Updated evaluation flow to log a warning when using flag sets that don't contain cached feature flags.
diff --git a/src/storages/KeyBuilder.ts b/src/storages/KeyBuilder.ts
@@ -1,4 +1,6 @@
+import { ISettings } from '../types';
 import { startsWith } from '../utils/lang';
+import { hash } from '../utils/murmur3/murmur3';
 
 const everythingAtTheEnd = /[^.]+$/;
 
@@ -10,7 +12,7 @@ export function validatePrefix(prefix: unknown) {
 
 export class KeyBuilder {
 
-  protected readonly prefix: string;
+  readonly prefix: string;
 
   constructor(prefix: string = DEFAULT_PREFIX) {
     this.prefix = prefix;
@@ -73,4 +75,15 @@ export class KeyBuilder {
     }
   }
 
+  buildHashKey() {
+    return `${this.prefix}.hash`;
+  }
+}
+
+/**
+ * Generates a murmur32 hash based on the authorization key and the feature flags filter query.
+ * The hash is in hexadecimal format (8 characters max, 32 bits).
+ */
+export function getStorageHash(settings: ISettings) {
+  return hash(`${settings.core.authorizationKey}::${settings.sync.__splitFiltersValidation.queryString}`).toString(16);
 }
diff --git a/src/storages/KeyBuilderCS.ts b/src/storages/KeyBuilderCS.ts
@@ -9,7 +9,7 @@ export class KeyBuilderCS extends KeyBuilder {
   constructor(prefix: string, matchingKey: string) {
     super(prefix);
     this.matchingKey = matchingKey;
-    this.regexSplitsCacheKey = new RegExp(`^${prefix}\\.(splits?|trafficType)\\.`);
+    this.regexSplitsCacheKey = new RegExp(`^${prefix}\\.(splits?|trafficType|flagSet)\\.`);
   }
 
   /**
diff --git a/src/storages/KeyBuilderSS.ts b/src/storages/KeyBuilderSS.ts
@@ -16,10 +16,10 @@ export const METHOD_NAMES: Record<Method, string> = {
 
 export class KeyBuilderSS extends KeyBuilder {
 
-  latencyPrefix: string;
-  exceptionPrefix: string;
-  initPrefix: string;
-  private versionablePrefix: string;
+  readonly latencyPrefix: string;
+  readonly exceptionPrefix: string;
+  readonly initPrefix: string;
+  private readonly versionablePrefix: string;
 
   constructor(prefix: string, metadata: IMetadata) {
     super(prefix);
diff --git a/src/storages/__tests__/KeyBuilder.spec.ts b/src/storages/__tests__/KeyBuilder.spec.ts
@@ -1,4 +1,5 @@
-import { KeyBuilder } from '../KeyBuilder';
+import { ISettings } from '../../types';
+import { KeyBuilder, getStorageHash } from '../KeyBuilder';
 import { KeyBuilderSS } from '../KeyBuilderSS';
 
 test('KEYS / splits keys', () => {
@@ -111,3 +112,20 @@ test('KEYS / latency and exception keys (telemetry)', () => {
   const expectedExceptionKey = `${prefix}.telemetry.exceptions::${metadata.s}/${metadata.n}/${metadata.i}/treatment`;
   expect(builder.buildExceptionKey(methodName)).toBe(expectedExceptionKey);
 });
+
+test('getStorageHash', () => {
+  expect(getStorageHash({
+    core: { authorizationKey: '<fake-token-rfc>' },
+    sync: { __splitFiltersValidation: { queryString: '&names=p1__split,p2__split' } }
+  } as ISettings)).toBe('6a596ded');
+
+  expect(getStorageHash({
+    core: { authorizationKey: '<fake-token-rfc>' },
+    sync: { __splitFiltersValidation: { queryString: '&names=p2__split,p3__split' } }
+  } as ISettings)).toBe('18ad36f3');
+
+  expect(getStorageHash({
+    core: { authorizationKey: '<fake-token-rfc>' },
+    sync: { __splitFiltersValidation: { queryString: null } }
+  } as ISettings)).toBe('9507ef4');
+});
diff --git a/src/storages/inLocalStorage/__tests__/SplitsCacheInLocal.spec.ts b/src/storages/inLocalStorage/__tests__/SplitsCacheInLocal.spec.ts
@@ -220,4 +220,10 @@ test('SPLIT CACHE / LocalStorage / flag set cache tests without filters', () =>
   expect(cacheWithoutFilters.getNamesByFlagSets(['t'])).toEqual([new _Set(['ff_two', 'ff_three'])]);
   expect(cacheWithoutFilters.getNamesByFlagSets(['y'])).toEqual([emptySet]);
   expect(cacheWithoutFilters.getNamesByFlagSets(['o', 'n', 'e'])).toEqual([new _Set(['ff_one', 'ff_two']), new _Set(['ff_one']), new _Set(['ff_one', 'ff_three'])]);
+
+  // Validate that the cache is cleared when calling `clear` method
+  localStorage.setItem('something', 'something');
+  cacheWithoutFilters.clear();
+  expect(localStorage.length).toBe(1); // only 'something' should remain in localStorage
+  expect(localStorage.getItem(localStorage.key(0)!)).toBe('something');
 });
diff --git a/src/storages/inRedis/TelemetryCacheInRedis.ts b/src/storages/inRedis/TelemetryCacheInRedis.ts
@@ -69,15 +69,10 @@ export class TelemetryCacheInRedis implements ITelemetryCacheAsync {
           return;
         }
 
-        if (!result.has(metadata)) result.set(metadata, {
-          t: newBuckets(),
-          ts: newBuckets(),
-          tc: newBuckets(),
-          tcs: newBuckets(),
-          tr: newBuckets(),
-        });
-
-        result.get(metadata)![method]![bucket] = count;
+        const methodLatencies = result.get(metadata) || {};
+        methodLatencies[method] = methodLatencies[method] || newBuckets();
+        methodLatencies[method]![bucket] = count;
+        result.set(metadata, methodLatencies);
       });
 
       return this.redis.del(this.keys.latencyPrefix).then(() => result);
@@ -109,14 +104,7 @@ export class TelemetryCacheInRedis implements ITelemetryCacheAsync {
 
         const [metadata, method] = parsedField;
 
-        if (!result.has(metadata)) result.set(metadata, {
-          t: 0,
-          ts: 0,
-          tc: 0,
-          tcs: 0,
-          tr: 0,
-        });
-
+        if (!result.has(metadata)) result.set(metadata, {});
         result.get(metadata)![method] = count;
       });
 
diff --git a/src/storages/inRedis/__tests__/TelemetryCacheInRedis.spec.ts b/src/storages/inRedis/__tests__/TelemetryCacheInRedis.spec.ts
@@ -1,7 +1,6 @@
 import { loggerMock } from '../../../logger/__tests__/sdkLogger.mock';
 import { KeyBuilderSS } from '../../KeyBuilderSS';
 import { TelemetryCacheInRedis } from '../TelemetryCacheInRedis';
-import { newBuckets } from '../../inMemory/TelemetryCacheInMemory';
 import { metadata } from '../../__tests__/KeyBuilder.spec';
 import { RedisAdapter } from '../RedisAdapter';
 
@@ -20,16 +19,20 @@ test('TELEMETRY CACHE IN REDIS', async () => {
   // recordException
   expect(await cache.recordException('tr')).toBe(1);
   expect(await cache.recordException('tr')).toBe(2);
+  expect(await cache.recordException('tcfs')).toBe(1);
 
   expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/track')).toBe('2');
   expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatment')).toBe(null);
+  expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatmentsWithConfigByFlagSets')).toBe('1');
 
   // recordLatency
   expect(await cache.recordLatency('tr', 1.6)).toBe(1);
   expect(await cache.recordLatency('tr', 1.6)).toBe(2);
+  expect(await cache.recordLatency('tfs', 1.6)).toBe(1);
 
   expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/track/2')).toBe('2');
   expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatment/2')).toBe(null);
+  expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatmentsByFlagSets/2')).toBe('1');
 
   // recordConfig
   expect(await cache.recordConfig()).toBe(1);
@@ -45,10 +48,7 @@ test('TELEMETRY CACHE IN REDIS', async () => {
   latencies.forEach((latency, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(latency).toEqual({
-      t: newBuckets(),
-      ts: newBuckets(),
-      tc: newBuckets(),
-      tcs: newBuckets(),
+      tfs: [0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
       tr: [0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
     });
   });
@@ -59,10 +59,7 @@ test('TELEMETRY CACHE IN REDIS', async () => {
   exceptions.forEach((exception, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(exception).toEqual({
-      t: 0,
-      ts: 0,
-      tc: 0,
-      tcs: 0,
+      tcfs: 1,
       tr: 2,
     });
   });
diff --git a/src/storages/pluggable/TelemetryCachePluggable.ts b/src/storages/pluggable/TelemetryCachePluggable.ts
@@ -68,15 +68,10 @@ export class TelemetryCachePluggable implements ITelemetryCacheAsync {
               continue;
             }
 
-            if (!result.has(metadata)) result.set(metadata, {
-              t: newBuckets(),
-              ts: newBuckets(),
-              tc: newBuckets(),
-              tcs: newBuckets(),
-              tr: newBuckets(),
-            });
-
-            result.get(metadata)![method]![bucket] = count;
+            const methodLatencies = result.get(metadata) || {};
+            methodLatencies[method] = methodLatencies[method] || newBuckets();
+            methodLatencies[method]![bucket] = count;
+            result.set(metadata, methodLatencies);
           }
 
           return Promise.all(latencyKeys.map((latencyKey) => this.wrapper.del(latencyKey))).then(() => result);
@@ -115,14 +110,7 @@ export class TelemetryCachePluggable implements ITelemetryCacheAsync {
 
             const [metadata, method] = parsedField;
 
-            if (!result.has(metadata)) result.set(metadata, {
-              t: 0,
-              ts: 0,
-              tc: 0,
-              tcs: 0,
-              tr: 0,
-            });
-
+            if (!result.has(metadata)) result.set(metadata, {});
             result.get(metadata)![method] = count;
           }
 
diff --git a/src/storages/pluggable/__tests__/TelemetryCachePluggable.spec.ts b/src/storages/pluggable/__tests__/TelemetryCachePluggable.spec.ts
@@ -19,16 +19,20 @@ test('TELEMETRY CACHE PLUGGABLE', async () => {
   // recordException
   expect(await cache.recordException('tr')).toBe(1);
   expect(await cache.recordException('tr')).toBe(2);
+  expect(await cache.recordException('tcfs')).toBe(1);
 
   expect(await wrapper.get(exceptionKey + '::' + fieldVersionablePrefix + '/track')).toBe('2');
   expect(await wrapper.get(exceptionKey + '::' + fieldVersionablePrefix + '/treatment')).toBe(null);
+  expect(await wrapper.get(exceptionKey + '::' + fieldVersionablePrefix + '/treatmentsWithConfigByFlagSets')).toBe('1');
 
   // recordLatency
   expect(await cache.recordLatency('tr', 1.6)).toBe(1);
   expect(await cache.recordLatency('tr', 1.6)).toBe(2);
+  expect(await cache.recordLatency('tfs', 1.6)).toBe(1);
 
   expect(await wrapper.get(latencyKey + '::' + fieldVersionablePrefix + '/track/2')).toBe('2');
   expect(await wrapper.get(latencyKey + '::' + fieldVersionablePrefix + '/treatment/2')).toBe(null);
+  expect(await wrapper.get(latencyKey + '::' + fieldVersionablePrefix + '/treatmentsByFlagSets/2')).toBe('1');
 
   // recordConfig
   await cache.recordConfig();
@@ -44,13 +48,15 @@ test('TELEMETRY CACHE PLUGGABLE', async () => {
   latencies.forEach((latency, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(latency.tr![2]).toBe(2);
+    expect(latency.tfs![2]).toBe(1);
   });
 
   // popExceptions
   const exceptions = await cache.popExceptions();
   exceptions.forEach((exception, m) => {
     expect(JSON.parse(m)).toEqual(metadata);
     expect(exception.tr).toBe(2);
+    expect(exception.tcfs).toBe(1);
   });
 
   // popConfigs
diff --git a/src/storages/pluggable/__tests__/index.spec.ts b/src/storages/pluggable/__tests__/index.spec.ts
@@ -12,6 +12,7 @@ const prefix = 'some_prefix';
 import { PluggableStorage } from '../index';
 import { assertStorageInterface, assertSyncRecorderCacheInterface } from '../../__tests__/testUtils';
 import { CONSUMER_PARTIAL_MODE } from '../../../utils/constants';
+import { getStorageHash } from '../../KeyBuilder';
 
 describe('PLUGGABLE STORAGE', () => {
 
@@ -92,14 +93,39 @@ describe('PLUGGABLE STORAGE', () => {
   });
 
   test('creates a storage instance for the synchronizer', async () => {
+    wrapperMock._cache[`${prefix}.SPLITIO.splits.till`] = '100'; // Simulate that the cache is already populated
+
+    const storageHash = getStorageHash(fullSettings);
     const storageFactory = PluggableStorage({ prefix, wrapper: wrapperMock });
-    const storage = storageFactory({ ...internalSdkParams, settings: { ...internalSdkParams.settings, mode: undefined } });
+    let storage;
 
-    assertStorageInterface(storage);
+    // Create storage instance. Wrapper is pollute but doesn't have filter query key, so it should clear the cache
+    await new Promise(resolve => {
+      storage = storageFactory({ onReadyCb: resolve, settings: { ...fullSettings, mode: undefined } });
+    });
 
-    // All caches are present
+    // Assert that expected caches are present
+    assertStorageInterface(storage);
     expect(storage.telemetry).toBeDefined();
     expect(storage.impressionCounts).toBeDefined();
     expect(storage.uniqueKeys).toBeDefined();
+
+    // Assert that cache was cleared
+    expect(wrapperMock.get.mock.calls).toEqual([[`${prefix}.SPLITIO.hash`]]);
+    expect(wrapperMock.set.mock.calls).toEqual([[`${prefix}.SPLITIO.hash`, storageHash]]);
+    expect(wrapperMock.del.mock.calls).toEqual([[`${prefix}.SPLITIO.splits.till`]]);
+    expect(wrapperMock._cache).toEqual({ [`${prefix}.SPLITIO.hash`]: storageHash });
+
+    wrapperMock.mockClear();
+
+    // Create storage instance. This time the wrapper has the current filter query key, so it should not clear the cache
+    await new Promise(resolve => {
+      storage = storageFactory({ onReadyCb: resolve, settings: { ...fullSettings, mode: undefined } });
+    });
+
+    // Assert that cache was not cleared
+    expect(wrapperMock.get).toBeCalledTimes(1);
+    expect(wrapperMock.set).toBeCalledTimes(0);
+    expect(wrapperMock.del).toBeCalledTimes(0);
   });
 });
diff --git a/src/storages/pluggable/__tests__/wrapper.mock.ts b/src/storages/pluggable/__tests__/wrapper.mock.ts
@@ -108,7 +108,6 @@ export function wrapperMockFactory() {
     disconnect: jest.fn(() => Promise.resolve()),
 
     mockClear() {
-      this._cache = {};
       this.get.mockClear();
       this.set.mockClear();
       this.del.mockClear();
diff --git a/src/storages/pluggable/index.ts b/src/storages/pluggable/index.ts
@@ -7,7 +7,7 @@ import { ImpressionsCachePluggable } from './ImpressionsCachePluggable';
 import { EventsCachePluggable } from './EventsCachePluggable';
 import { wrapperAdapter, METHODS_TO_PROMISE_WRAP } from './wrapperAdapter';
 import { isObject } from '../../utils/lang';
-import { validatePrefix } from '../KeyBuilder';
+import { getStorageHash, validatePrefix } from '../KeyBuilder';
 import { CONSUMER_PARTIAL_MODE, DEBUG, NONE, STORAGE_PLUGGABLE } from '../../utils/constants';
 import { ImpressionsCacheInMemory } from '../inMemory/ImpressionsCacheInMemory';
 import { EventsCacheInMemory } from '../inMemory/EventsCacheInMemory';
@@ -19,6 +19,7 @@ import { UniqueKeysCachePluggable } from './UniqueKeysCachePluggable';
 import { UniqueKeysCacheInMemory } from '../inMemory/UniqueKeysCacheInMemory';
 import { UniqueKeysCacheInMemoryCS } from '../inMemory/UniqueKeysCacheInMemoryCS';
 import { metadataBuilder } from '../utils';
+import { LOG_PREFIX } from '../pluggable/constants';
 
 const NO_VALID_WRAPPER = 'Expecting pluggable storage `wrapper` in options, but no valid wrapper instance was provided.';
 const NO_VALID_WRAPPER_INTERFACE = 'The provided wrapper instance doesn’t follow the expected interface. Check our docs.';
@@ -90,13 +91,26 @@ export function PluggableStorage(options: PluggableStorageOptions): IStorageAsyn
 
     // Connects to wrapper and emits SDK_READY event on main client
     const connectPromise = wrapper.connect().then(() => {
-      onReadyCb();
-
-      // Start periodic flush of async storages if not running synchronizer (producer mode)
-      if (!isSyncronizer) {
+      if (isSyncronizer) {
+        // In standalone or producer mode, clear storage if SDK key or feature flag filter has changed
+        return wrapper.get(keys.buildHashKey()).then((hash) => {
+          const currentHash = getStorageHash(settings);
+          if (hash !== currentHash) {
+            log.info(LOG_PREFIX + 'Storage HASH has changed (SDK key or feature flag filter criteria was modified). Clearing cache');
+            return wrapper.getKeysByPrefix(`${keys.prefix}.`).then(storageKeys => {
+              return Promise.all(storageKeys.map(storageKey => wrapper.del(storageKey)));
+            }).then(() => wrapper.set(keys.buildHashKey(), currentHash));
+          }
+        }).then(() => {
+          onReadyCb();
+        });
+      } else {
+        // Start periodic flush of async storages if not running synchronizer (producer mode)
         if (impressionCountsCache && (impressionCountsCache as ImpressionCountsCachePluggable).start) (impressionCountsCache as ImpressionCountsCachePluggable).start();
         if (uniqueKeysCache && (uniqueKeysCache as UniqueKeysCachePluggable).start) (uniqueKeysCache as UniqueKeysCachePluggable).start();
         if (telemetry && (telemetry as ITelemetryCacheAsync).recordConfig) (telemetry as ITelemetryCacheAsync).recordConfig();
+
+        onReadyCb();
       }
     }).catch((e) => {
       e = e || new Error('Error connecting wrapper');
diff --git a/src/storages/types.ts b/src/storages/types.ts
@@ -457,7 +457,7 @@ export interface IStorageBase<
   TEventsCache extends IEventsCacheBase,
   TTelemetryCache extends ITelemetryCacheSync | ITelemetryCacheAsync,
   TUniqueKeysCache extends IUniqueKeysCacheBase
-  > {
+> {
   splits: TSplitsCache,
   segments: TSegmentsCache,
   impressions: TImpressionsCache,
@@ -477,7 +477,7 @@ export interface IStorageSync extends IStorageBase<
   IEventsCacheSync,
   ITelemetryCacheSync,
   IUniqueKeysCacheSync
-  > { }
+> { }
 
 export interface IStorageAsync extends IStorageBase<
   ISplitsCacheAsync,
@@ -487,7 +487,7 @@ export interface IStorageAsync extends IStorageBase<
   IEventsCacheAsync | IEventsCacheSync,
   ITelemetryCacheAsync | ITelemetryCacheSync,
   IUniqueKeysCacheBase
-  > { }
+> { }
 
 /** StorageFactory */
 
diff --git a/src/utils/murmur3/murmur3.ts b/src/utils/murmur3/murmur3.ts

Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,7 @@ export class KeyBuilderCS extends KeyBuilder {`
`9`	`9`	`constructor(prefix: string, matchingKey: string) {`
`10`	`10`	`super(prefix);`
`11`	`11`	`this.matchingKey = matchingKey;`
`12`		- this.regexSplitsCacheKey = new RegExp(`^${prefix}\\.(splits?\|trafficType)\\.`);
	`12`	+ this.regexSplitsCacheKey = new RegExp(`^${prefix}\\.(splits?\|trafficType\|flagSet)\\.`);
`13`	`13`	`}`
`14`	`14`
`15`	`15`	`/**`