Fb/better change handling (#65)

* wip 1

* wip 2

* wip 3

* wip 4

* wip 5

* wip 6

* wip 7

* wip 8

* wip 9

* wip 10

Author: rlindner81

src/featureToggles.js

@@ -23,10 +23,9 @@ const { REDIS_INTEGRATION_MODE } = redis;
 const cfEnv = require("./shared/env");
 const { Logger } = require("./shared/logger");
 const { HandlerCollection } = require("./shared/handlerCollection");
-const { promiseAllDone } = require("./shared/promiseAllDone");
 const { LimitedLazyCache } = require("./shared/cache");
 const { Semaphore } = require("./shared/semaphore");
-const { ENV, isObject, tryRequire, tryPathReadable } = require("./shared/static");
+const { ENV, isObject, tryRequire, tryPathReadable, tryJsonParse } = require("./shared/static");
 
 const ENV_UNIQUE_NAME = process.env[ENV.UNIQUE_NAME];
 const DEFAULT_REDIS_CHANNEL = process.env[ENV.REDIS_CHANNEL] || "features";
@@ -1165,7 +1164,7 @@ class FeatureToggles {
    * @returns {Array<ChangeEntry>}
    */
   static _deserializeChangesFromRefreshMessage(message) {
-    return JSON.parse(message);
+    return tryJsonParse(message);
   }
 
   /**
@@ -1229,15 +1228,44 @@ class FeatureToggles {
   }
 
   /**
-   * Handler for refresh message.
+   * Handler for message with change entries.
    */
   async _messageHandler(message) {
-    let featureKey, newValue, scopeMap, options;
-    try {
-      const changeEntries = FeatureToggles._deserializeChangesFromRefreshMessage(message);
-      await promiseAllDone(
-        changeEntries.map(async (changeEntry) => {
-          ({ featureKey, newValue, scopeMap, options } = changeEntry);
+    const changeEntries = FeatureToggles._deserializeChangesFromRefreshMessage(message);
+    if (!Array.isArray(changeEntries)) {
+      logger.error(
+        new VError(
+          {
+            name: VERROR_CLUSTER_NAME,
+            info: {
+              channel: this.__redisChannel,
+              message,
+            },
+          },
+          "error during message deserialization"
+        )
+      );
+      return;
+    }
+
+    await Promise.all(
+      changeEntries.map(async (changeEntry) => {
+        try {
+          if (!isObject(changeEntry) || changeEntry.featureKey === undefined || changeEntry.newValue === undefined) {
+            logger.warning(
+              new VError(
+                {
+                  name: VERROR_CLUSTER_NAME,
+                  info: {
+                    changeEntry: JSON.stringify(changeEntry),
+                  },
+                },
+                "received and ignored change entry"
+              )
+            );
+            return;
+          }
+          const { featureKey, newValue, scopeMap, options } = changeEntry;
 
           const scopeKey = FeatureToggles.getScopeKey(scopeMap);
           const oldValue = FeatureToggles._getFeatureValueForScopeAndStateAndFallback(
@@ -1276,25 +1304,23 @@ class FeatureToggles {
             scopeKey,
             options
           );
-        })
-      );
-    } catch (err) {
-      logger.error(
-        new VError(
-          {
-            name: VERROR_CLUSTER_NAME,
-            cause: err,
-            info: {
-              channel: this.__redisChannel,
-              message,
-              ...(featureKey && { featureKey }),
-              ...(scopeMap && { scopeMap: JSON.stringify(scopeMap) }),
-            },
-          },
-          "error during message handling"
-        )
-      );
-    }
+        } catch (err) {
+          logger.error(
+            new VError(
+              {
+                name: VERROR_CLUSTER_NAME,
+                cause: err,
+                info: {
+                  channel: this.__redisChannel,
+                  changeEntry: JSON.stringify(changeEntry),
+                },
+              },
+              "error during message handling"
+            )
+          );
+        }
+      })
+    );
   }
 
   async _changeRemoteFeatureValue(featureKey, newValue, scopeMap, options) {

src/shared/static.js

@@ -28,6 +28,12 @@ const tryRequire = (module) => {
   } catch (err) {} // eslint-disable-line no-empty
 };
 
+const tryJsonParse = (...args) => {
+  try {
+    return JSON.parse(...args);
+  } catch (err) {} // eslint-disable-line no-empty
+};
+
 const pathReadable = async (path) => (await accessAsync(path, fs.constants.R_OK)) ?? true;
 
 const tryPathReadable = async (path) => {
@@ -43,6 +49,7 @@ module.exports = {
   isNull,
   isObject,
   tryRequire,
+  tryJsonParse,
   pathReadable,
   tryPathReadable,
 };

test/__snapshots__/featureToggles.test.js.snap

@@ -218,9 +218,3 @@ exports[`feature toggles test basic apis initializeFeatureToggles warns for inva
   },
 }
 `;
-
-exports[`feature toggles test basic apis initializeFeatureToggles warns for invalid values 3`] = `
-{
-  "validationErrors": "[{"featureKey":"test/feature_b","errorMessage":"value null is not allowed"},{"featureKey":"test/feature_c","errorMessage":"value \\"{0}\\" has invalid type {1}, must be {2}","errorMessageValues":["1","string","number"]}]",
-}
-`;

test/featureToggles.test.js

@@ -1,5 +1,7 @@
 "use strict";
 
+const util = require("util");
+
 const VError = require("verror");
 const yaml = require("yaml");
 const featureTogglesModule = require("../src/featureToggles");
@@ -25,6 +27,9 @@ jest.mock("../src/shared/env", () => require("./__mocks__/env"));
 const redisWrapperMock = require("../src/redisWrapper");
 jest.mock("../src/redisWrapper", () => require("./__mocks__/redisWrapper"));
 
+const outputFromErrorLogger = (calls) =>
+  calls.map((args) => util.format("%s\n%O", args[0], VError.info(args[0]))).join("\n");
+
 const configToFallbackValues = (config) =>
   Object.fromEntries(Object.entries(config).map(([key, { fallbackValue }]) => [key, fallbackValue]));
 const configToActiveKeys = (config) =>
@@ -131,16 +136,12 @@ describe("feature toggles test", () => {
         `[]`
       );
       expect(loggerSpy.error).toHaveBeenCalledTimes(1);
-      expect(loggerSpy.error.mock.calls[0]).toMatchInlineSnapshot(`
-        [
-          [FeatureTogglesError: scope exceeds allowed number of keys],
-        ]
-      `);
-      expect(VError.info(loggerSpy.error.mock.calls[0][0])).toMatchInlineSnapshot(`
+      expect(outputFromErrorLogger(loggerSpy.error.mock.calls)).toMatchInlineSnapshot(`
+        "FeatureTogglesError: scope exceeds allowed number of keys
         {
-          "maxKeys": 4,
-          "scopeMap": "{"tenantId":"t1","label":"l1","bla":"b1","naa":"n1","xxx":"x1"}",
-        }
+          scopeMap: '{"tenantId":"t1","label":"l1","bla":"b1","naa":"n1","xxx":"x1"}',
+          maxKeys: 4
+        }"
       `);
     });
 
@@ -288,11 +289,12 @@ describe("feature toggles test", () => {
 
       expect(loggerSpy.warning).toHaveBeenCalledTimes(1);
       expect(loggerSpy.warning).toHaveBeenCalledWith(expect.any(VError));
-      const [loggedErr] = loggerSpy.warning.mock.calls[0];
-      expect(loggedErr).toMatchInlineSnapshot(
-        `[FeatureTogglesError: found invalid fallback values during initialization]`
-      );
-      expect(VError.info(loggedErr)).toMatchSnapshot();
+      expect(outputFromErrorLogger(loggerSpy.warning.mock.calls)).toMatchInlineSnapshot(`
+        "FeatureTogglesError: found invalid fallback values during initialization
+        {
+          validationErrors: '[{"featureKey":"test/feature_b","errorMessage":"value null is not allowed"},{"featureKey":"test/feature_c","errorMessage":"value \\\\"{0}\\\\" has invalid type {1}, must be {2}","errorMessageValues":["1","string","number"]}]'
+        }"
+      `);
       expect(loggerSpy.error).not.toHaveBeenCalled();
     });
 
@@ -1151,4 +1153,74 @@ describe("feature toggles test", () => {
       expect(loggerSpy.error).not.toHaveBeenCalled();
     });
   });
+
+  describe("message handling", () => {
+    beforeEach(async () => {
+      await featureToggles.initializeFeatures({ config: mockConfig });
+    });
+
+    it("empty array should be handled fine", async () => {
+      await redisWrapperMock.publishMessage(featureToggles.__redisChannel, "[]");
+      expect(loggerSpy.warning).toHaveBeenCalledTimes(0);
+      expect(loggerSpy.error).toHaveBeenCalledTimes(0);
+    });
+
+    it("error in case message is not a serialized array", async () => {
+      await redisWrapperMock.publishMessage(featureToggles.__redisChannel, "hello world!");
+      expect(loggerSpy.warning).toHaveBeenCalledTimes(0);
+      expect(loggerSpy.error).toHaveBeenCalledTimes(1);
+      expect(outputFromErrorLogger(loggerSpy.error.mock.calls)).toMatchInlineSnapshot(`
+        "FeatureTogglesError: error during message deserialization
+        { channel: 'feature-channel', message: 'hello world!' }"
+      `);
+    });
+
+    it("invalid change entries are ignored but logged", async () => {
+      const changeEntries = [{ featureKey: FEATURE.C, newValue: 10 }];
+
+      expect(featureToggles.getFeatureValue(FEATURE.C)).toBe(mockFallbackValues[FEATURE.C]);
+
+      await redisWrapperMock.publishMessage(featureToggles.__redisChannel, JSON.stringify(changeEntries));
+
+      expect(loggerSpy.warning).toHaveBeenCalledTimes(1);
+      expect(outputFromErrorLogger(loggerSpy.warning.mock.calls)).toMatchInlineSnapshot(`
+        "FeatureTogglesError: received and ignored invalid value from message
+        {
+          validationErrors: '[{"featureKey":"test/feature_c","scopeKey":"//","errorMessage":"value \\\\"{0}\\\\" has invalid type {1}, must be {2}","errorMessageValues":[10,"number","string"]}]'
+        }"
+      `);
+
+      expect(featureToggles.getFeatureValue(FEATURE.C)).toBe(mockFallbackValues[FEATURE.C]);
+    });
+
+    it("all change entries are processed even if one fails", async () => {
+      const scopeMap = { tenant: "testing" };
+      const changeEntries = [
+        { featureKey: FEATURE.C, newValue: "modified" },
+        {},
+        null,
+        "bla",
+        { featureKey: FEATURE.E, newValue: 9, scopeMap },
+      ];
+
+      expect(featureToggles.getFeatureValue(FEATURE.C)).toBe(mockFallbackValues[FEATURE.C]);
+      expect(featureToggles.getFeatureValue(FEATURE.E, scopeMap)).toBe(mockFallbackValues[FEATURE.E]);
+
+      await redisWrapperMock.publishMessage(featureToggles.__redisChannel, JSON.stringify(changeEntries));
+
+      expect(loggerSpy.warning).toHaveBeenCalledTimes(3);
+      expect(outputFromErrorLogger(loggerSpy.warning.mock.calls)).toMatchInlineSnapshot(`
+        "FeatureTogglesError: received and ignored change entry
+        { changeEntry: '{}' }
+        FeatureTogglesError: received and ignored change entry
+        { changeEntry: 'null' }
+        FeatureTogglesError: received and ignored change entry
+        { changeEntry: '"bla"' }"
+      `);
+      expect(loggerSpy.error).toHaveBeenCalledTimes(0);
+
+      expect(featureToggles.getFeatureValue(FEATURE.C)).toMatchInlineSnapshot(`"modified"`);
+      expect(featureToggles.getFeatureValue(FEATURE.E, scopeMap)).toMatchInlineSnapshot(`9`);
+    });
+  });
 });