feat: circuit breaker for usage reporting

Author: n1ru4l

packages/libraries/core/package.json

@@ -50,6 +50,7 @@
     "async-retry": "^1.3.3",
     "js-md5": "0.8.3",
     "lodash.sortby": "^4.7.0",
+    "opossum": "^9.0.0",
     "tiny-lru": "^8.0.2"
   },
   "devDependencies": {
@@ -58,6 +59,7 @@
     "@types/async-retry": "1.4.8",
     "@types/js-md5": "0.8.0",
     "@types/lodash.sortby": "4.7.9",
+    "@types/opossum": "8.1.9",
     "graphql": "16.9.0",
     "nock": "14.0.10",
     "tslib": "2.8.1",

packages/libraries/core/src/client/agent.ts

@@ -1,9 +1,33 @@
+import CircuitBreaker from 'opossum';
+import { fetch as defaultFetch } from '@whatwg-node/fetch';
 import { version } from '../version.js';
 import { http } from './http-client.js';
 import type { Logger } from './types.js';
 
 type ReadOnlyResponse = Pick<Response, 'status' | 'text' | 'json' | 'statusText'>;
 
+export type AgentCircuitBreakerConfiguration = {
+  /** after which time a request should be treated as a timeout in milleseconds */
+  timeout: number;
+  /** percentage after what the circuit breaker should kick in. */
+  errorThresholdPercentage: number;
+  /** count of requests before starting evaluating. */
+  volumeThreshold: number;
+  /** after what time the circuit breaker is resetted in milliseconds */
+  resetTimeout: number;
+};
+
+const defaultCircuitBreakerConfiguration: AgentCircuitBreakerConfiguration = {
+  // if call takes > 5s, count as a failure
+  timeout: 5000,
+  // trip if 50% of calls fail
+  errorThresholdPercentage: 50,
+  // need at least 5 calls before evaluating
+  volumeThreshold: 5,
+  // after 30s, try half-open state
+  resetTimeout: 30000,
+};
+
 export interface AgentOptions {
   enabled?: boolean;
   name?: string;
@@ -47,7 +71,11 @@ export interface AgentOptions {
    * WHATWG Compatible fetch implementation
    * used by the agent to send reports
    */
-  fetch?: typeof fetch;
+  fetch?: typeof defaultFetch;
+  /**
+   * Circuit Breaker Configuration
+   */
+  circuitBreaker?: AgentCircuitBreakerConfiguration;
 }
 
 export function createAgent<TEvent>(
@@ -76,12 +104,13 @@ export function createAgent<TEvent>(
     maxSize: 25,
     logger: console,
     name: 'hive-client',
+    circuitBreaker: defaultCircuitBreakerConfiguration,
     ...pluginOptions,
   };
 
   const enabled = options.enabled !== false;
 
-  let timeoutID: any = null;
+  let timeoutID: ReturnType<typeof setTimeout> | null = null;
 
   function schedule() {
     if (timeoutID) {
@@ -131,20 +160,22 @@ export function createAgent<TEvent>(
 
     if (data.size() >= options.maxSize) {
       debugLog('Sending immediately');
-      setImmediate(() => send({ throwOnError: false, skipSchedule: true }));
+      setImmediate(() => breaker.fire({ throwOnError: false, skipSchedule: true }));
     }
   }
 
   function sendImmediately(event: TEvent): Promise<ReadOnlyResponse | null> {
     data.set(event);
     debugLog('Sending immediately');
-    return send({ throwOnError: true, skipSchedule: true });
+    return breaker.fire({ throwOnError: true, skipSchedule: true });
   }
 
   async function send(sendOptions?: {
     throwOnError?: boolean;
     skipSchedule: boolean;
   }): Promise<ReadOnlyResponse | null> {
+    const signal: AbortSignal = breaker.getSignal();
+
     if (!data.size() || !enabled) {
       if (!sendOptions?.skipSchedule) {
         schedule();
@@ -174,6 +205,7 @@ export function createAgent<TEvent>(
         },
         logger: options.logger,
         fetchImplementation: pluginOptions.fetch,
+        signal,
       })
       .then(res => {
         debugLog(`Report sent!`);
@@ -207,12 +239,21 @@ export function createAgent<TEvent>(
       await Promise.all(inProgressCaptures);
     }
 
-    await send({
+    await breaker.fire({
       skipSchedule: true,
       throwOnError: false,
     });
   }
 
+  const breaker = new CircuitBreaker(send, {
+    ...options.circuitBreaker,
+    autoRenewAbortController: true,
+  });
+
+  breaker.on('open', () => errorLog('circuit opened - backend unreachable'));
+  breaker.on('halfOpen', () => debugLog('testing backend connectivity'));
+  breaker.on('close', () => debugLog('backend recovered - circuit closed'));
+
   return {
     capture,
     sendImmediately,

packages/libraries/core/src/client/http-client.ts

@@ -21,6 +21,8 @@ interface SharedConfig {
    * @default {response => response.ok}
    **/
   isRequestOk?: ResponseAssertFunction;
+  /** Optional abort signal */
+  signal?: AbortSignal;
 }
 
 /**
@@ -78,6 +80,8 @@ export async function makeFetchCall(
      * @default {response => response.ok}
      **/
     isRequestOk?: ResponseAssertFunction;
+    /** Optional abort signal */
+    signal?: AbortSignal;
   },
 ): Promise<Response> {
   const logger = config.logger;
@@ -104,7 +108,10 @@ export async function makeFetchCall(
       );
 
       const getDuration = measureTime();
-      const signal = AbortSignal.timeout(config.timeout ?? 20_000);
+      const timeoutSignal = AbortSignal.timeout(config.timeout ?? 20_000);
+      const signal = config.signal
+        ? AbortSignal.any([config.signal, timeoutSignal])
+        : timeoutSignal;
 
       const response = await (config.fetchImplementation ?? fetch)(endpoint, {
         method: config.method,
@@ -135,6 +142,12 @@ export async function makeFetchCall(
         throw new Error(`Unexpected HTTP error. (x-request-id=${requestId})`, { cause: error });
       });
 
+      if (config.signal?.aborted === true) {
+        // TODO: maybe log some message?
+        bail(new Error('Request aborted.'));
+        return;
+      }
+
       if (isRequestOk(response)) {
         logger?.info(
           `${config.method} ${endpoint} (x-request-id=${requestId}) succeeded with status ${response.status} ${getDuration()}.`,