24174068770

Committed 09 Apr 2026 05:27AM UTC coverage: 90.103% (+0.2%) from 89.941%

Build # 24174068770

Build Type

push

github

Committed by

web-flow

Commit Message

refactor(circuit-breaker): simplify configuration and remove per-circuit overrides (#2997)

* refactor(circuit-breaker): simplify configuration and remove per-circuit overrides

Unify failureThreshold/successThreshold into a single threshold value,
replace halfOpenMaxRequests with percentage-based halfOpenFailureRateThreshold,
remove per-circuit config overrides, and clean up unused metrics (successCount,
halfOpenRequestCounts). recordFailure now accepts circuit name instead of object.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* style: fix prettier formatting

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(circuit-breaker): lower default threshold and improve error handling

Move canProceedOrFail outside try block, record success for non-server
errors, and reduce default threshold from 20 to 10.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test(circuit-breaker): fix integration test to use dynamic threshold

The test hardcoded 2 failures to trip the circuit, but the default
threshold was lowered to 10. Use defaultThreshold in a loop to match
the pattern of other circuit breaker tests.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(circuit-breaker): address review feedback on HALF_OPEN state and config

- Reset lastFailureTime on HALF_OPEN transition to prevent stale value
  from causing confusing discardOldFailures behavior
- Fix .env.sample.json CIRCUIT_BREAKER_THRESHOLD default (20 -> 10) to
  match configuration.ts
- Add dedicated unit tests for canProceedOrFail public method

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(circuit-breaker): address review feedback on HALF_OPEN state and config

- Skip discardOldFailures in HALF_OPEN state to prevent stale
  lastFailureTime from resetting failureCount
- Fix .env.sample.json CIRCUIT_BREAKER_THRESHOLD default (20 -> 10) to
  match configuration.ts
- Add ... (continued)

Coverage Stats

3236 of 4022 branches covered (80.46%)

Branch coverage included in aggregate %.

48 of 49 new or added lines in 2 files covered. (97.96%)

3 existing lines in 3 files now uncovered.

15227 of 16469 relevant lines covered (92.46%)

610.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.26

/src/datasources/network/network.module.ts

// SPDX-License-Identifier: FSL-1.1-MIT
import { Global, Module } from '@nestjs/common';
import { IConfigurationService } from '@/config/configuration.service.interface';
import { FetchNetworkService } from '@/datasources/network/fetch.network.service';
import { NetworkService } from '@/datasources/network/network.service.interface';
import { NetworkResponse } from '@/datasources/network/entities/network.response.entity';
import {
  NetworkRequestError,
  NetworkResponseError,
} from '@/datasources/network/entities/network.error.entity';
import type { Raw } from '@/validation/entities/raw.entity';
import { ILoggingService, LoggingService } from '@/logging/logging.interface';
import { LogType } from '@/domain/common/entities/log-type.entity';
import { hashSha1 } from '@/domain/common/utils/utils';
import { CircuitBreakerService } from '@/datasources/circuit-breaker/circuit-breaker.service';
import { NetworkRequest } from '@/datasources/network/entities/network.request.entity';
import { setGlobalDispatcher, Agent } from 'undici';
import {
  UndiciAgent,
  UndiciShutdownHook,
} from '@/datasources/network/undici.shutdown.hook';
import { asError } from '@/logging/utils';

export const FetchClientToken = Symbol('FetchClient');

export type FetchClient = <T>(
  url: string,
  options: RequestInit,
  timeout?: number,
  circuitBreaker?: NetworkRequest['circuitBreaker'],
) => Promise<NetworkResponse<T>>;

const cache: Record<string, Promise<NetworkResponse<unknown>>> = {};

/**
 * Use this factory to create a {@link FetchClient} instance
 * that can be used to make HTTP requests.
 */
function fetchClientFactory(
  configurationService: IConfigurationService,
  circuitBreakerService: CircuitBreakerService,
  loggingService: ILoggingService,
): FetchClient {
  const cacheInFlightRequests = configurationService.getOrThrow<boolean>(
    'features.cacheInFlightRequests',
  );
  const defaultTimeout = configurationService.getOrThrow<number>(
    'httpClient.requestTimeout',
  );

  const baseRequest = createRequestFunction(defaultTimeout);
  const circuitBreakerRequest = createCircuitBreakerRequestFunction(
    baseRequest,
    circuitBreakerService,
  );

  if (!cacheInFlightRequests) {
    return circuitBreakerRequest;
  }

  return createCachedRequestFunction(circuitBreakerRequest, loggingService);
}

function createRequestFunction(defaultTimeout: number) {
  return async <T>(
    url: string,
    options: RequestInit,
    customTimeout?: number,
  ): Promise<NetworkResponse<T>> => {
    let urlObject: URL | null = null;
    let response: Response | null = null;

    try {
      urlObject = new URL(url);
      const timeout = customTimeout ?? defaultTimeout;

      response = await fetch(url, {
        ...options,
        signal: AbortSignal.timeout(timeout),
        keepalive: true,
      });
    } catch (error) {
      throw new NetworkRequestError(urlObject, error);
    }

    // We validate data so don't need worry about casting `null` response
    const data = (await response.json().catch(() => null)) as Raw<T>;

    if (!response.ok) {
      throw new NetworkResponseError(urlObject, response, data);
    }

    return {
      status: response.status,
      data,
    };
  };
}

/**
 * Wraps a request function with circuit breaker logic
 *
 * This function intercepts requests and applies circuit breaker protection:
 * - Checks if the circuit is open before allowing the request
 * - Records successes and failures based on response status
 * - Must be explicitly enabled per request via the circuitBreaker parameter
 *
 * @param request - The base request function to wrap
 * @param circuitBreakerService - Service managing circuit breaker state
 *
 * @returns Wrapped request function with circuit breaker logic
 */
function createCircuitBreakerRequestFunction(
  request: <T>(
    url: string,
    options: RequestInit,
    timeout?: number,
  ) => Promise<NetworkResponse<T>>,
  circuitBreakerService: CircuitBreakerService,
) {
  return async <T>(
    url: string,
    options: RequestInit,
    timeout?: number,
    circuitBreaker?: NetworkRequest['circuitBreaker'],
  ): Promise<NetworkResponse<T>> => {
    if (!circuitBreaker || !circuitBreaker.key) {
      return request(url, options, timeout);
    }

    circuitBreakerService.canProceedOrFail(circuitBreaker.key);

    try {
      const response = await request(url, options, timeout);
      circuitBreakerService.recordSuccess(circuitBreaker.key);
      return response;
    } catch (error) {
      const isServerError =
        error instanceof NetworkResponseError && error.response.status >= 500;
      const isNetworkError = error instanceof NetworkRequestError;
      if (isServerError || isNetworkError) {
        circuitBreakerService.getOrRegisterCircuit(circuitBreaker.key);
        circuitBreakerService.recordFailure(circuitBreaker.key);
      } else {
        circuitBreakerService.recordSuccess(circuitBreaker.key);
      }

      throw error;
    }
  };
}

function createCachedRequestFunction(
  request: <T>(
    url: string,
    options: RequestInit,
    timeout?: number,
    circuitBreaker?: NetworkRequest['circuitBreaker'],
  ) => Promise<NetworkResponse<T>>,
  loggingService: ILoggingService,
) {
  return async <T>(
    url: string,
    options: RequestInit,
    timeout?: number,
    circuitBreaker?: NetworkRequest['circuitBreaker'],
  ): Promise<NetworkResponse<T>> => {
    const key = getCacheKey(url, options, timeout, circuitBreaker);
    if (key in cache) {
      loggingService.debug({
        type: LogType.ExternalRequestCacheHit,
        url,
        key,
      });
    } else {
      loggingService.debug({
        type: LogType.ExternalRequestCacheMiss,
        url,
        key,
      });

      cache[key] = request(url, options, timeout, circuitBreaker)
        .catch((err) => {
          loggingService.debug({
            type: LogType.ExternalRequestCacheError,
            url,
            key,
          });
          throw err;
        })
        .finally(() => {
          delete cache[key];
        });
    }

    return cache[key];
  };
}

function getCacheKey(
  url: string,
  requestInit?: RequestInit,
  timeout?: number,
  circuitBreaker?: NetworkRequest['circuitBreaker'],
): string {
  if (!requestInit && timeout === undefined && !circuitBreaker) {
    return url;
  }

  // JSON.stringify does not produce a stable key but initially
  // use a naive implementation for testing the implementation
  // TODO: Revisit this and use a more stable key
  const circuitBreakerKey = circuitBreaker?.key || '';
  const key = JSON.stringify({
    url,
    ...requestInit,
    timeout,
    circuitBreakerKey,
  });
  return hashSha1(key);
}

/**
 * Sets up the global Undici dispatcher with configured connection pooling
 * Returns the Agent instance for graceful shutdown
 */
function setupUndiciDispatcher(
  configurationService: IConfigurationService,
  loggingService: ILoggingService,
): Agent {
  const connections =
    configurationService.getOrThrow<number>('undici.connections');
  const pipelining =
    configurationService.getOrThrow<number>('undici.pipelining');
  const connectTimeout = configurationService.getOrThrow<number>(
    'undici.connectTimeout',
  );
  const keepAliveTimeout = configurationService.getOrThrow<number>(
    'undici.keepAliveTimeout',
  );
  const keepAliveMaxTimeout = configurationService.getOrThrow<number>(
    'undici.keepAliveMaxTimeout',
  );

  try {
    const agent = new Agent({
      connections,
      pipelining,
      connect: {
        timeout: connectTimeout,
        keepAlive: true,
      },
      keepAliveTimeout,
      keepAliveMaxTimeout,
    });

    setGlobalDispatcher(agent);
    return agent;
  } catch (error) {
    loggingService.error(
      `Failed to setup Undici global dispatcher: ${asError(error).message}`,
    );
    throw error;
  }
}

/**
 * A {@link Global} Module which provides HTTP support via {@link NetworkService}
 * Feature Modules don't need to import this module directly in order to inject
 * the {@link NetworkService}.
 *
 * This module should be included in the "root" application module
 */
@Global()
@Module({
  providers: [
    {
      provide: UndiciAgent,
      useFactory: setupUndiciDispatcher,
      inject: [IConfigurationService, LoggingService],
    },
    UndiciShutdownHook,
    {
      provide: FetchClientToken,
      useFactory: fetchClientFactory,
      inject: [IConfigurationService, CircuitBreakerService, LoggingService],
    },
    {
      provide: NetworkService,
      useFactory: (
        client: FetchClient,
        loggingService: ILoggingService,
      ): FetchNetworkService => {
        return new FetchNetworkService(client, loggingService);
      },
      inject: [FetchClientToken, LoggingService],
    },
  ],
  exports: [NetworkService, FetchClientToken],
})
export class NetworkModule {}

1	// SPDX-License-Identifier: FSL-1.1-MIT
2	import { Global, Module } from '@nestjs/common';	170✔
3	import { IConfigurationService } from '@/config/configuration.service.interface';	170✔
4	import { FetchNetworkService } from '@/datasources/network/fetch.network.service';	170✔
5	import { NetworkService } from '@/datasources/network/network.service.interface';	170✔
6	import { NetworkResponse } from '@/datasources/network/entities/network.response.entity';
7	import {	170✔
8	NetworkRequestError,
9	NetworkResponseError,
10	} from '@/datasources/network/entities/network.error.entity';
11	import type { Raw } from '@/validation/entities/raw.entity';
12	import { ILoggingService, LoggingService } from '@/logging/logging.interface';	170✔
13	import { LogType } from '@/domain/common/entities/log-type.entity';	170✔
14	import { hashSha1 } from '@/domain/common/utils/utils';	170✔
15	import { CircuitBreakerService } from '@/datasources/circuit-breaker/circuit-breaker.service';	170✔
16	import { NetworkRequest } from '@/datasources/network/entities/network.request.entity';
17	import { setGlobalDispatcher, Agent } from 'undici';	170✔
18	import {	170✔
19	UndiciAgent,
20	UndiciShutdownHook,
21	} from '@/datasources/network/undici.shutdown.hook';
22	import { asError } from '@/logging/utils';	170✔
23
24	export const FetchClientToken = Symbol('FetchClient');	170✔
25
26	export type FetchClient = <T>(
27	url: string,
28	options: RequestInit,
29	timeout?: number,
30	circuitBreaker?: NetworkRequest['circuitBreaker'],
31	) => Promise<NetworkResponse<T>>;
32
33	const cache: Record<string, Promise<NetworkResponse<unknown>>> = {};	170✔
34
35	/**
36	* Use this factory to create a {@link FetchClient} instance
37	* that can be used to make HTTP requests.
38	*/
39	function fetchClientFactory(
40	configurationService: IConfigurationService,
41	circuitBreakerService: CircuitBreakerService,
42	loggingService: ILoggingService,
43	): FetchClient {
44	const cacheInFlightRequests = configurationService.getOrThrow<boolean>(	12✔
45	'features.cacheInFlightRequests',
46	);
47	const defaultTimeout = configurationService.getOrThrow<number>(	12✔
48	'httpClient.requestTimeout',
49	);
50
51	const baseRequest = createRequestFunction(defaultTimeout);	12✔
52	const circuitBreakerRequest = createCircuitBreakerRequestFunction(	12✔
53	baseRequest,
54	circuitBreakerService,
55	);
56
57	if (!cacheInFlightRequests) {	12✔
58	return circuitBreakerRequest;	8✔
59	}
60
61	return createCachedRequestFunction(circuitBreakerRequest, loggingService);	4✔
62	}
63
64	function createRequestFunction(defaultTimeout: number) {
65	return async <T>(	12✔
66	url: string,
67	options: RequestInit,
68	customTimeout?: number,
69	): Promise<NetworkResponse<T>> => {
70	let urlObject: URL \| null = null;	104✔
71	let response: Response \| null = null;	104✔
72
73	try {	104✔
74	urlObject = new URL(url);	104✔
75	const timeout = customTimeout ?? defaultTimeout;	102✔
76
77	response = await fetch(url, {	102✔
78	...options,
79	signal: AbortSignal.timeout(timeout),
80	keepalive: true,
81	});
82	} catch (error) {
83	throw new NetworkRequestError(urlObject, error);	8✔
84	}
85
86	// We validate data so don't need worry about casting `null` response
87	const data = (await response.json().catch(() => null)) as Raw<T>;	96✔
88
89	if (!response.ok) {	94✔
90	throw new NetworkResponseError(urlObject, response, data);	74✔
91	}
92
93	return {	20✔
94	status: response.status,
95	data,
96	};
97	};
98	}
99
100	/**
101	* Wraps a request function with circuit breaker logic
102	*
103	* This function intercepts requests and applies circuit breaker protection:
104	* - Checks if the circuit is open before allowing the request
105	* - Records successes and failures based on response status
106	* - Must be explicitly enabled per request via the circuitBreaker parameter
107	*
108	* @param request - The base request function to wrap
109	* @param circuitBreakerService - Service managing circuit breaker state
110	*
111	* @returns Wrapped request function with circuit breaker logic
112	*/
113	function createCircuitBreakerRequestFunction(
114	request: <T>(
115	url: string,
116	options: RequestInit,
117	timeout?: number,
118	) => Promise<NetworkResponse<T>>,
119	circuitBreakerService: CircuitBreakerService,
120	) {
121	return async <T>(	12✔
122	url: string,
123	options: RequestInit,
124	timeout?: number,
125	circuitBreaker?: NetworkRequest['circuitBreaker'],
126	): Promise<NetworkResponse<T>> => {
127	if (!circuitBreaker \|\| !circuitBreaker.key) {	108✔
128	return request(url, options, timeout);	38✔
129	}
130
131	circuitBreakerService.canProceedOrFail(circuitBreaker.key);	70✔
132
133	try {	66✔
134	const response = await request(url, options, timeout);	66✔
135	circuitBreakerService.recordSuccess(circuitBreaker.key);	6✔
136	return response;	6✔
137	} catch (error) {
138	const isServerError =
139	error instanceof NetworkResponseError && error.response.status >= 500;	60✔
140	const isNetworkError = error instanceof NetworkRequestError;	60✔
141	if (isServerError \|\| isNetworkError) {	60!
142	circuitBreakerService.getOrRegisterCircuit(circuitBreaker.key);	60✔
143	circuitBreakerService.recordFailure(circuitBreaker.key);	60✔
144	} else {
NEW 145	circuitBreakerService.recordSuccess(circuitBreaker.key);	×
146	}
147
148	throw error;	60✔
149	}
150	};
151	}
152
153	function createCachedRequestFunction(
154	request: <T>(
155	url: string,
156	options: RequestInit,
157	timeout?: number,
158	circuitBreaker?: NetworkRequest['circuitBreaker'],
159	) => Promise<NetworkResponse<T>>,
160	loggingService: ILoggingService,
161	) {
162	return async <T>(	4✔
163	url: string,
164	options: RequestInit,
165	timeout?: number,
166	circuitBreaker?: NetworkRequest['circuitBreaker'],
167	): Promise<NetworkResponse<T>> => {
168	const key = getCacheKey(url, options, timeout, circuitBreaker);	62✔
169	if (key in cache) {	62✔
170	loggingService.debug({	12✔
171	type: LogType.ExternalRequestCacheHit,
172	url,
173	key,
174	});
175	} else {
176	loggingService.debug({	50✔
177	type: LogType.ExternalRequestCacheMiss,
178	url,
179	key,
180	});
181
182	cache[key] = request(url, options, timeout, circuitBreaker)	50✔
183	.catch((err) => {
184	loggingService.debug({	34✔
185	type: LogType.ExternalRequestCacheError,
186	url,
187	key,
188	});
189	throw err;	34✔
190	})
191	.finally(() => {
192	delete cache[key];	50✔
193	});
194	}
195
196	return cache[key];	62✔
197	};
198	}
199
200	function getCacheKey(
201	url: string,
202	requestInit?: RequestInit,
203	timeout?: number,
204	circuitBreaker?: NetworkRequest['circuitBreaker'],
205	): string {
206	if (!requestInit && timeout === undefined && !circuitBreaker) {	62!
207	return url;	×
208	}
209
210	// JSON.stringify does not produce a stable key but initially
211	// use a naive implementation for testing the implementation
212	// TODO: Revisit this and use a more stable key
213	const circuitBreakerKey = circuitBreaker?.key \|\| '';	62✔
214	const key = JSON.stringify({	62✔
215	url,
216	...requestInit,
217	timeout,
218	circuitBreakerKey,
219	});
220	return hashSha1(key);	62✔
221	}
222
223	/**
224	* Sets up the global Undici dispatcher with configured connection pooling
225	* Returns the Agent instance for graceful shutdown
226	*/
227	function setupUndiciDispatcher(
228	configurationService: IConfigurationService,
229	loggingService: ILoggingService,
230	): Agent {
231	const connections =
232	configurationService.getOrThrow<number>('undici.connections');	12✔
233	const pipelining =
234	configurationService.getOrThrow<number>('undici.pipelining');	12✔
235	const connectTimeout = configurationService.getOrThrow<number>(	12✔
236	'undici.connectTimeout',
237	);
238	const keepAliveTimeout = configurationService.getOrThrow<number>(	12✔
239	'undici.keepAliveTimeout',
240	);
241	const keepAliveMaxTimeout = configurationService.getOrThrow<number>(	12✔
242	'undici.keepAliveMaxTimeout',
243	);
244
245	try {	12✔
246	const agent = new Agent({	12✔
247	connections,
248	pipelining,
249	connect: {
250	timeout: connectTimeout,
251	keepAlive: true,
252	},
253	keepAliveTimeout,
254	keepAliveMaxTimeout,
255	});
256
257	setGlobalDispatcher(agent);	12✔
258	return agent;	12✔
259	} catch (error) {
260	loggingService.error(	×
261	`Failed to setup Undici global dispatcher: ${asError(error).message}`,
262	);
263	throw error;	×
264	}
265	}
266
267	/**
268	* A {@link Global} Module which provides HTTP support via {@link NetworkService}
269	* Feature Modules don't need to import this module directly in order to inject
270	* the {@link NetworkService}.
271	*
272	* This module should be included in the "root" application module
273	*/
274	@Global()
275	@Module({
276	providers: [
277	{
278	provide: UndiciAgent,
279	useFactory: setupUndiciDispatcher,
280	inject: [IConfigurationService, LoggingService],
281	},
282	UndiciShutdownHook,
283	{
284	provide: FetchClientToken,
285	useFactory: fetchClientFactory,
286	inject: [IConfigurationService, CircuitBreakerService, LoggingService],
287	},
288	{
289	provide: NetworkService,
290	useFactory: (
291	client: FetchClient,
292	loggingService: ILoggingService,
293	): FetchNetworkService => {
294	return new FetchNetworkService(client, loggingService);	12✔
295	},
296	inject: [FetchClientToken, LoggingService],
297	},
298	],
299	exports: [NetworkService, FetchClientToken],
300	})
301	export class NetworkModule {}	170✔

safe-global / safe-client-gateway / 24174068770

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous