feature: Coalesce reset options, reset coalesce based on result. (error, success, timeout)

README.md

@@ -459,7 +459,45 @@ The code that is summing the stats samples is here:
 
 ### Coalesce calls
 
-Circuitbreaker offers coalescing your calls. If options.coalesce is set, multiple calls to the circuitbreaker will be handled as one, within the given timeframe (options.coalesceTTL). Performance will improve when rapidly firing the circuitbreaker with the same request, especially on a slower action. This is especially useful if multiple events can trigger the same functions at the same time. Of course, caching has the same function, but will only be effective when the call has been executed once to store the return value. Coalescing and cache can be used at the same time, coalescing calls will always use the internal cache.
+Circuitbreaker offers coalescing your calls. If options.coalesce is set, multiple calls to the circuitbreaker will be handled as one, within the given timeframe (options.coalesceTTL). Performance will improve when rapidly firing the circuitbreaker with the same request, especially on a slower action. This is especially useful if multiple events can trigger the same functions at the same time. Of course, caching has the same function, but will only be effective when the call has been successfully executed once to store the return value. Coalescing and cache can be used at the same time, coalescing calls will always use the internal cache. Accessing cache is done prior to coalescing. When using `capacity` option, coalescing reduces the capacity used for the CircuitBreaker and will allow higher throughput of the rest of the application without actually firing the CircuitBreaker protected function. The `cacheGetKey` option is used for coalescing as well.
+
+#### Finetuning Coalesce behavior
+
+By default, all calls within given timeframe are coalesced, including errors and timeouts. This might be unwanted, as this twarths retry mechanisms etc. To finetune coalesce behavior, use the coalesceResetOn parameter. Some examples:
+
+| coalesceResetOn value | behavior |
+| --------------------- | -------- |
+| `error`, `success`, `timeout` | coalescing is reset after every 'done' status, so only concurrent 'running' calls are coalesced. One could consider this the most essential form of coalescing. |
+| `error`, `timeout` | No error state is being stored for longer than the running call, you might want to start here if you use any retry mechanisms. |
+| `error` | Reset on errors. |
+| `timeout` | Reset on timeouts. |
+| `success` | Reset on success. |
+
+You can use any combination of `error`, `success`, `timeout`.
+
+#### Using CircuitBreaker with Coalescing and fetch.
+
+When using the CircuitBreaker with Coalescing enabled to protect calling external services using the Fetch API, it's important to keep this in mind: The Response interface of the Fetch API does not allow reading the same body multiple times, cloning the Response will not help either as it will delay the reading of the response until the slowest reader is done. To work around this you can either choose to wrap handling of the response (e.g. parsing) in the protected function as well, keep in mind any errors and delays in this process will amount to the error thresholds configured. This might not be suitable for complexer setups. Another option would be to flatten the response and revive it. This might come in handy when working with libraries that expect a Response object. Example below:
+
+```js
+const flattenResponse = async (r) => ({
+  arrayBuffer: await r.arrayBuffer(),
+  init: {
+    headers: r.headers,
+    ok: r.ok,
+    redirected: r.redirected,
+    status: r.status,
+    statusText: r.statusText,
+    type: r.type,
+    url: r.url,
+  },
+});
+
+const reviveResponse = (r) => new Response(r.arrayBuffer, r.init);
+```
+
+Also note, Fetch doesn't fail on HTTP errors (e.g. 50x). If you want to protect your application from calling failing APIs, check the response status and throw errors accordingly.
+
 
 ### Typings
 

lib/cache.js

@@ -44,6 +44,15 @@ class MemoryCache {
     });
   }
 
+  /**
+   * Delete cache key
+   * @param {string} key Cache key
+   * @return {void}
+   */
+  delete (key) {
+    this.cache.delete(key);
+  }
+
   /**
    * Clear cache
    * @returns {void}

lib/circuit.js

@@ -111,6 +111,8 @@ Please use options.errorThresholdPercentage`;
  * in milliseconds. Set 0 for infinity cache. Default: same as options.timeout
  * @param {Number} options.coalesceSize the max amount of entries in the
  * coalescing cache. Default: max size of JS map (2^24).
+ * @param {string[]} options.coalesceResetOn when to reset the coalesce cache.
+ * Options: `error`, `success`, `timeout`. Default: not set, reset using TTL.
  * @param {AbortController} options.abortController this allows Opossum to
  * signal upon timeout and properly abort your on going requests instead of
  * leaving it in the background
@@ -193,6 +195,7 @@ class CircuitBreaker extends EventEmitter {
     this.options.rotateBucketController = options.rotateBucketController;
     this.options.coalesce = !!options.coalesce;
     this.options.coalesceTTL = options.coalesceTTL ?? this.options.timeout;
+    this.options.coalesceResetOn = options.coalesceResetOn?.filter(o => ['error', 'success', 'timeout'].includes(o)) || [];
 
     // Set default cache transport if not provided
     if (this.options.cache) {
@@ -743,6 +746,8 @@ class CircuitBreaker extends EventEmitter {
                */
               this.emit('timeout', error, latency, args);
               handleError(error, this, timeout, args, latency, resolve, reject);
+              resetCoalesce(this, cacheKey, 'timeout');
+
               if (this.options.abortController) {
                 this.options.abortController.abort();
               }
@@ -764,6 +769,7 @@ class CircuitBreaker extends EventEmitter {
                * @type {any} the return value from the circuit
                */
               this.emit('success', result, (Date.now() - latencyStartTime));
+              resetCoalesce(this, cacheKey, 'success');
               this.semaphore.release();
               resolve(result);
               if (this.options.cache) {
@@ -783,12 +789,14 @@ class CircuitBreaker extends EventEmitter {
                 const latencyEndTime = Date.now() - latencyStartTime;
                 handleError(
                   error, this, timeout, args, latencyEndTime, resolve, reject);
+                resetCoalesce(this, cacheKey, 'error');
               }
             });
         } catch (error) {
           this.semaphore.release();
           const latency = Date.now() - latencyStartTime;
           handleError(error, this, timeout, args, latency, resolve, reject);
+          resetCoalesce(this, cacheKey, 'error');
         }
       } else {
         const latency = Date.now() - latencyStartTime;
@@ -801,6 +809,7 @@ class CircuitBreaker extends EventEmitter {
          */
         this.emit('semaphoreLocked', err, latency);
         handleError(err, this, timeout, args, latency, resolve, reject);
+        resetCoalesce(this, cacheKey);
       }
     });
 
@@ -826,6 +835,10 @@ class CircuitBreaker extends EventEmitter {
     if (this.options.cache) {
       this.options.cacheTransport.flush();
     }
+
+    if (this.options.coalesceCache) {
+      this.options.coalesceCache.flush();
+    }
   }
 
   /**
@@ -940,6 +953,7 @@ function handleError (error, circuit, timeout, args, latency, resolve, reject) {
     const fb = fallback(circuit, error, args);
     if (fb) return resolve(fb);
   }
+
   // In all other cases, reject
   reject(error);
 }
@@ -983,6 +997,20 @@ function fail (circuit, err, args, latency) {
   }
 }
 
+function resetCoalesce (circuit, cacheKey, event) {
+/**
+ * Reset coalesce cache for this cacheKey, depending on
+ * options.coalesceResetOn set.
+ * @param {@link CircuitBreaker} circuit what circuit is to be cleared
+ * @param {string} cacheKey cache key to clear.
+ * @param {string} event optional, can be `error`, `success`, `timeout`
+ * @returns {void}
+ */
+  if (!event || circuit.options.coalesceResetOn.includes(event)) {
+    circuit.options.coalesceCache?.delete(cacheKey);
+  }
+}
+
 function buildError (msg, code) {
   const error = new Error(msg);
   error.code = code;

test/cache.js

@@ -195,6 +195,64 @@ test('Using coalesce cache only.', t => {
     .catch(t.fail);
 });
 
+// Test coalesce coalesceResetOn.
+(function () {
+  const options = {
+    coalesce: true,
+    timeout: 200,
+    coalesceResetOn: ['error', 'success', 'timeout', 'foobar'],
+    errorThresholdPercentage: 99,
+    allowWarmUp: true
+  };
+
+  test('coalesceResetOn: expect proper parsing of options', t => {
+    t.plan(1);
+    const breaker = new CircuitBreaker(passFail, options);
+    t.same(breaker.options.coalesceResetOn, ['error', 'success', 'timeout']);
+    t.end();
+  });
+
+  test('coalesceResetOn: expect no hit after success', t => {
+    t.plan(1);
+    const breaker = new CircuitBreaker(passFail, options);
+    breaker
+      .fire(1)
+      .then(() => {
+        breaker.fire(1).then(() => {
+          const stats = breaker.status.stats;
+          t.equals(stats.coalesceCacheHits, 0, 'no hits to coalesce cache, it is reset when action succeeded.');
+          t.end();
+        });
+      });
+  });
+
+  test('coalesceResetOn: expect no hit after error', t => {
+    t.plan(1);
+    const breaker = new CircuitBreaker(passFail, options);
+    breaker
+      .fire(-1)
+      .catch(() => {
+        breaker.fire(1).then(() => {
+          const stats = breaker.status.stats;
+          t.equals(stats.coalesceCacheHits, 0, 'no hits to coalesce cache, it is reset when action failed.');
+          t.end();
+        });
+      });
+  });
+
+  test('coalesceResetOn: expect no hit after timeout', t => {
+    t.plan(1);
+    const timedBreaker = new CircuitBreaker(common.timedFunction, options);
+    timedBreaker.fire(1000).catch(() => {
+      timedBreaker.fire(1).then(() => {
+        const stats = timedBreaker.status.stats;
+        t.equals(stats.coalesceCacheHits, 0, 'no hits to coalesce cache, it is reset when action timed out.');
+        t.end();
+      });
+    });
+  });
+})();
+
 test('No coalesce cache.', t => {
   t.plan(5);
   const breaker = new CircuitBreaker(passFail);